better handling of framed sockets in fnet

also minor tweaking of crypto codec adapter

3 files changed, 19 insertions, 10 deletions

diff --git a/fnet/src/vespa/fnet/connection.cpp b/fnet/src/vespa/fnet/connection.cpp
index f2864d1dd58..e028afe5deb 100644
--- a/fnet/src/vespa/fnet/connection.cpp
+++ b/fnet/src/vespa/fnet/connection.cpp
@@ -225,6 +225,7 @@ FNET_Connection::handshake()
     case vespalib::CryptoSocket::HandshakeResult::DONE: {
         EnableReadEvent(true);
         EnableWriteEvent(writePendingAfterConnect());
+        _flags._framed = (_socket->min_read_buffer_size() > 1);
         size_t chunk_size = std::max(size_t(FNET_READ_SIZE), _socket->min_read_buffer_size());
         ssize_t res = 0;
         do { // drain input pipeline
@@ -287,7 +288,7 @@ FNET_Connection::Read()
         _input.FreeToData((uint32_t)res);
         broken = !handle_packets();
         _input.resetIfEmpty();
-        if (broken || (_input.GetFreeLen() > 0) || (readCnt >= FNET_READ_REDO)) {
+        if (broken || ((_input.GetFreeLen() > 0) && !_flags._framed) || (readCnt >= FNET_READ_REDO)) {
             goto done_read;
         }
         _input.EnsureFree(chunk_size);
@@ -302,7 +303,6 @@ done_read:
         _input.EnsureFree(chunk_size);
         res = _socket->drain(_input.GetFree(), _input.GetFreeLen());
         my_errno = errno;
-        readCnt++;
         if (res > 0) {
             _input.FreeToData((uint32_t)res);
             broken = !handle_packets();
@@ -340,6 +340,7 @@ done_read:
 bool
 FNET_Connection::Write()
 {
+    size_t   chunk_size     = std::max(size_t(FNET_WRITE_SIZE), _socket->min_read_buffer_size());
     uint32_t my_write_work  = 0;
     int      writeCnt       = 0;     // write count
     bool     broken         = false; // is this conn broken ?
@@ -353,7 +354,7 @@ FNET_Connection::Write()
 
         // fill output buffer
 
-        while (_output.GetDataLen() < FNET_WRITE_SIZE) {
+        while (_output.GetDataLen() < chunk_size) {
             if (_myQueue.IsEmpty_NoLock())
                 break;
 
diff --git a/fnet/src/vespa/fnet/connection.h b/fnet/src/vespa/fnet/connection.h
index 8e275d68b18..44ad3fea97d 100644
--- a/fnet/src/vespa/fnet/connection.h
+++ b/fnet/src/vespa/fnet/connection.h
@@ -70,12 +70,14 @@ private:
             _gotheader(false),
             _inCallback(false),
             _callbackWait(false),
-            _discarding(false)
+            _discarding(false),
+            _framed(false)
         { }
         bool _gotheader;
         bool _inCallback;
         bool _callbackWait;
         bool _discarding;
+        bool _framed;
     };
     struct ResolveHandler : public vespalib::AsyncResolver::ResultHandler {
         FNET_Connection *connection;
diff --git a/vespalib/src/vespa/vespalib/net/tls/crypto_codec_adapter.cpp b/vespalib/src/vespa/vespalib/net/tls/crypto_codec_adapter.cpp
index 435f16cc340..f8fa4ab2b53 100644
--- a/vespalib/src/vespa/vespalib/net/tls/crypto_codec_adapter.cpp
+++ b/vespalib/src/vespa/vespalib/net/tls/crypto_codec_adapter.cpp
@@ -34,7 +34,7 @@ CryptoCodecAdapter::hs_try_fill()
 ssize_t
 CryptoCodecAdapter::fill_input()
 {
-    if (_input.get().size < _codec->min_encode_buffer_size()) {
+    if (_input.obtain().size < _codec->min_encode_buffer_size()) {
         auto dst = _input.reserve(_codec->min_encode_buffer_size());
         ssize_t res = _socket.read(dst.data, dst.size);
         if (res > 0) {
@@ -89,9 +89,9 @@ CryptoCodecAdapter::read(char *buf, size_t len)
     if (fill_res <= 0) {
         return fill_res;
     }
-    ssize_t res = drain(buf, len);
-    if (res != 0) {
-        return res;
+    auto drain_res = drain(buf, len);
+    if (drain_res != 0) {
+        return drain_res;
     }
     errno = EWOULDBLOCK;
     return -1;
@@ -113,8 +113,14 @@ CryptoCodecAdapter::drain(char *buf, size_t len)
 ssize_t
 CryptoCodecAdapter::write(const char *buf, size_t len)
 {
-    if (flush_all() < 0) {
-        return -1;
+    if (_output.obtain().size >= _codec->min_encode_buffer_size()) {
+        if (flush() < 0) {
+            return -1;
+        }
+        if (_output.obtain().size > 0) {
+            errno = EWOULDBLOCK;
+            return -1;
+        }
     }
     auto dst = _output.reserve(_codec->min_encode_buffer_size());
     auto res = _codec->encode(buf, len, dst.data, dst.size);