Rework msgr send/receive to allow encryption support

src/client/messenger.cpp

@@ -510,7 +510,7 @@ void osd_messenger_t::try_connect_peer_tcp(osd_num_t peer_osd, const char *peer_
     cl->peer_state = PEER_CONNECTING;
     cl->connect_timeout_id = -1;
     cl->osd_num = peer_osd;
-    cl->in_buf = malloc_or_die(receive_buffer_size);
+    cl->in_buf = (uint8_t*)malloc_or_die(receive_buffer_size);
     clients[client_id] = cl;
     clients_by_fd[peer_fd] = cl;
     tfd->set_fd_handler(peer_fd, true, [this](int peer_fd, int epoll_events)
@@ -786,7 +786,7 @@ void osd_messenger_t::accept_connections(int listen_fd)
         cl->peer_port = ntohs(((sockaddr_in*)&addr)->sin_port);
         cl->peer_fd = peer_fd;
         cl->peer_state = PEER_CONNECTED;
-        cl->in_buf = malloc_or_die(receive_buffer_size);
+        cl->in_buf = (uint8_t*)malloc_or_die(receive_buffer_size);
         // Add FD to epoll
         tfd->set_fd_handler(peer_fd, false, [this](int peer_fd, int epoll_events)
         {

src/client/messenger.h

@@ -35,9 +35,6 @@
 
 #define DEFAULT_MIN_ZEROCOPY_SEND_SIZE 32*1024
 
-#define MSGR_SENDP_HDR 1
-#define MSGR_SENDP_FREE 2
-
 struct msgr_sendp_t
 {
     osd_op_t *op;
@@ -65,7 +62,7 @@ struct osd_client_t
     osd_num_t in_osd_num = 0;
     bool is_incoming = false;
 
-    void *in_buf = NULL;
+    uint8_t *in_buf = NULL;
 
 #ifdef WITH_RDMA
     msgr_rdma_connection_t *rdma_conn = NULL;
@@ -74,11 +71,12 @@ struct osd_client_t
     // Read state
     int read_ready = 0;
     osd_op_t *read_op = NULL;
+    size_t read_op_size = 0;
+    size_t read_op_pos = 0;
     iovec read_iov = { 0 };
     msghdr read_msg = { 0 };
-    int read_remaining = 0;
+    std::vector<iovec> recv_list;
-    int read_state = 0;
+    size_t recv_list_size = 0;
-    osd_op_buf_list_t recv_list;
     uint64_t read_op_id = 1;
     bool check_sequencing = false;
     bool enable_pg_locks = false;
@@ -94,10 +92,14 @@ struct osd_client_t
     std::set<pool_pg_num_t> dirty_pgs;
 
     // Write state
+    std::deque<osd_op_t *> write_ops;
+    osd_op_t *write_op = NULL;
+    size_t write_op_pos = 0;
     msghdr write_msg = { 0 };
     int write_state = 0;
-    std::vector<iovec> send_list, next_send_list;
+    std::vector<iovec> send_list;
-    std::vector<msgr_sendp_t> outbox, next_outbox;
+    size_t send_list_size = 0;
+    std::deque<osd_op_t*> send_free_ops;
     std::vector<osd_op_t*> zc_free_list;
 
     ~osd_client_t();
@@ -279,17 +281,22 @@ protected:
 
     bool try_send(osd_client_t *cl);
     void handle_send(int result, bool prev, bool more, osd_client_t *cl);
+    size_t op_copy_to(osd_client_t *cl, uint8_t *dst, size_t dst_len);
+    void op_get_write_buffers(osd_client_t *cl, std::vector<iovec> & lst);
 
-    bool handle_read(int result, osd_client_t *cl);
+    void handle_read(int result, osd_client_t *cl);
-    bool handle_read_buffer(osd_client_t *cl, void *curbuf, int remain);
+    bool handle_read_buffer(osd_client_t *cl, uint8_t *curbuf, size_t bufsize);
-    bool handle_finished_read(osd_client_t *cl);
+    bool handle_hdr(osd_client_t *cl);
-    void handle_op_hdr(osd_client_t *cl);
+    bool allocate_op_buffers(osd_client_t *cl);
-    bool handle_reply_hdr(osd_client_t *cl);
+    bool allocate_reply_buffers(osd_client_t *cl, osd_op_t *op);
-    void handle_reply_ready(osd_op_t *op);
+    size_t op_copy_from(osd_client_t *cl, uint8_t *src, size_t src_len, size_t & done);
+    size_t op_get_read_buffers(osd_client_t *cl, std::vector<iovec> & lst);
+    void handle_finished_op(osd_client_t *cl);
     void handle_immediate_ops();
 
 #ifdef WITH_RDMA
     void try_send_rdma(osd_client_t *cl);
+    int try_send_rdma_copy(osd_client_t *cl, uint8_t *dst, int dst_len);
     bool init_recv_rdma(osd_client_t *cl);
     void handle_rdma_events(msgr_rdma_context_t *rdma_context);
     msgr_rdma_context_t* choose_rdma_context(osd_client_t *cl);

src/client/msgr_rdma.cpp

@@ -568,23 +568,28 @@ static void try_send_rdma_wr(osd_client_t *cl, ibv_sge *sge, int op_sge)
     cl->rdma_conn->cur_send++;
 }
 
-static int try_send_rdma_copy(osd_client_t *cl, uint8_t *dst, int dst_len)
+int osd_messenger_t::try_send_rdma_copy(osd_client_t *cl, uint8_t *dst, int dst_len)
 {
-    auto rc = cl->rdma_conn;
     int total_dst_len = dst_len;
-    while (dst_len > 0 && rc->send_pos < cl->send_list.size())
+    while (dst_len > 0 && cl->write_ops.size())
     {
-        iovec & iov = cl->send_list[rc->send_pos];
+        if (!cl->write_op)
-        uint32_t len = (uint32_t)(iov.iov_len-rc->send_buf_pos < dst_len
-            ? iov.iov_len-rc->send_buf_pos : dst_len);
-        memcpy(dst, (uint8_t*)iov.iov_base+rc->send_buf_pos, len);
-        dst += len;
-        dst_len -= len;
-        rc->send_buf_pos += len;
-        if (rc->send_buf_pos >= iov.iov_len)
         {
-            rc->send_pos++;
+            cl->write_op = cl->write_ops.front();
-            rc->send_buf_pos = 0;
+            cl->write_ops.pop_front();
+        }
+        osd_op_t *op = cl->write_op;
+        size_t copied = op_copy_to(cl, dst, dst_len);
+        if (!copied)
+        {
+            break;
+        }
+        dst += copied;
+        dst_len -= copied;
+        if (!cl->write_op && op->op_type == OSD_OP_IN)
+        {
+            // this is a reply, free the op after sending it
+            cl->send_free_ops.push_back(op);
         }
     }
     return total_dst_len-dst_len;
@@ -631,6 +636,7 @@ void osd_messenger_t::try_send_rdma(osd_client_t *cl)
             };
             try_send_rdma_wr(cl, &sge, 1);
             rc->send_sizes.push_back(copied);
+            cl->send_free_ops.push_back(NULL); // end marker
         }
     }
 }
@@ -741,51 +747,20 @@ void osd_messenger_t::handle_rdma_events(msgr_rdma_context_t *rdma_context)
             else
             {
                 rc->cur_send--;
-                uint64_t sent_size = rc->send_sizes.at(0);
+                // byte_len is not filled for send operations
-                rc->send_sizes.erase(rc->send_sizes.begin(), rc->send_sizes.begin()+1);
+                uint64_t sent_size = rc->send_sizes.front();
+                rc->send_sizes.pop_front();
                 rc->send_done_pos += sent_size;
                 rc->send_out_full = false;
                 if (rc->send_done_pos == rc->send_out_size)
                     rc->send_done_pos = 0;
                 assert(rc->send_done_pos < rc->send_out_size);
-                int send_pos = 0, send_buf_pos = 0;
+                while (cl->send_free_ops.front())
-                while (sent_size > 0)
                 {
-                    if (sent_size >= cl->send_list.at(send_pos).iov_len)
+                    delete cl->send_free_ops.front();
-                    {
+                    cl->send_free_ops.pop_front();
-                        sent_size -= cl->send_list[send_pos].iov_len;
-                        send_pos++;
-                    }
-                    else
-                    {
-                        send_buf_pos = sent_size;
-                        sent_size = 0;
-                    }
-                }
-                assert(rc->send_pos >= send_pos);
-                if (rc->send_pos == send_pos)
-                {
-                    rc->send_buf_pos -= send_buf_pos;
-                }
-                rc->send_pos -= send_pos;
-                for (int i = 0; i < send_pos; i++)
-                {
-                    if (cl->outbox[i].flags & MSGR_SENDP_FREE)
-                    {
-                        // Reply fully sent
-                        delete cl->outbox[i].op;
-                    }
-                }
-                if (send_pos > 0)
-                {
-                    cl->send_list.erase(cl->send_list.begin(), cl->send_list.begin()+send_pos);
-                    cl->outbox.erase(cl->outbox.begin(), cl->outbox.begin()+send_pos);
-                }
-                if (send_buf_pos > 0)
-                {
-                    cl->send_list[0].iov_base = (uint8_t*)cl->send_list[0].iov_base + send_buf_pos;
-                    cl->send_list[0].iov_len -= send_buf_pos;
                 }
+                cl->send_free_ops.pop_front();
                 try_send_rdma(cl);
             }
         }

src/client/msgr_rdma.h

@@ -8,8 +8,11 @@
 #include <infiniband/verbs.h>
 #include <string>
 #include <vector>
+#include <deque>
 #include "addr_util.h"
 
+struct osd_op_t;
+
 struct msgr_rdma_address_t
 {
     ibv_gid gid;
@@ -72,9 +75,9 @@ struct msgr_rdma_connection_t
     int cur_send = 0, cur_recv = 0;
     int send_pos = 0, send_buf_pos = 0;
     int next_recv_buf = 0;
-    std::vector<void*> recv_buffers;
+    std::vector<uint8_t*> recv_buffers;
     msgr_rdma_buf_t recv_buf;
-    std::vector<uint64_t> send_sizes;
+    std::deque<uint64_t> send_sizes;
     msgr_rdma_buf_t send_out;
     int send_out_pos = 0, send_done_pos = 0, send_out_size = 0;
     bool send_out_full = false;

src/client/msgr_rdmacm.cpp

@@ -495,7 +495,7 @@ void osd_messenger_t::rdmacm_established(rdma_cm_event *ev)
     cl->peer_state = PEER_RDMA;
     cl->connect_timeout_id = -1;
     cl->osd_num = peer_osd;
-    cl->in_buf = malloc_or_die(receive_buffer_size);
+    cl->in_buf = (uint8_t*)malloc_or_die(receive_buffer_size);
     cl->rdma_conn = rc;
     clients[conn->client_id] = cl;
     if (conn->timeout_id >= 0)

src/client/msgr_receive.cpp

@@ -1,6 +1,8 @@
 // Copyright (c) Vitaliy Filippov, 2019+
 // License: VNPL-1.1 or GNU GPL-2.0+ (see README.md for details)
 
+#define _XOPEN_SOURCE
+#include <limits.h>
 #include "messenger.h"
 
 void osd_messenger_t::read_requests()
@@ -15,7 +17,11 @@ void osd_messenger_t::read_requests()
             continue;
         }
         auto cl = cl_it->second;
-        if (cl->read_remaining < receive_buffer_size)
+        if (cl->read_op && cl->read_op_size-(cl->read_op_pos-OSD_PACKET_SIZE) >= receive_buffer_size)
+        {
+            op_get_read_buffers(cl, cl->recv_list);
+        }
+        if (!cl->recv_list.size())
         {
             cl->read_iov.iov_base = cl->in_buf;
             cl->read_iov.iov_len = receive_buffer_size;
@@ -25,10 +31,11 @@ void osd_messenger_t::read_requests()
         else
         {
             cl->read_iov.iov_base = 0;
-            cl->read_iov.iov_len = cl->read_remaining;
+            cl->read_iov.iov_len = 0;
-            cl->read_msg.msg_iov = cl->recv_list.get_iovec();
+            cl->read_msg.msg_iov = cl->recv_list.data();
-            cl->read_msg.msg_iovlen = cl->recv_list.get_size();
+            cl->read_msg.msg_iovlen = cl->recv_list.size();
         }
+        assert(!cl->read_op || cl->read_op_pos < OSD_PACKET_SIZE || cl->read_op_size >= (cl->read_op_pos-OSD_PACKET_SIZE));
         cl->refs++;
         if (ringloop && !use_sync_send_recv)
         {
@@ -50,7 +57,7 @@ void osd_messenger_t::read_requests()
             }
             ring_data_t* data = ((ring_data_t*)sqe->user_data);
             data->callback = [this, cl](ring_data_t *data) { handle_read(data->res, cl); };
-            io_uring_prep_recvmsg(sqe, cl->peer_fd, &cl->read_msg, 0);
+            io_uring_prep_recvmsg(sqe, cl->peer_fd, &cl->read_msg, cl->recv_list.size() ? MSG_WAITALL : 0);
             if (iothread)
             {
                 iothread->add_sqe(sqe_local);
@@ -70,14 +77,12 @@ void osd_messenger_t::read_requests()
     read_ready_clients.clear();
 }
 
-bool osd_messenger_t::handle_read(int result, osd_client_t *cl)
+void osd_messenger_t::handle_read(int result, osd_client_t *cl)
 {
-    bool ret = false;
-    cl->read_msg.msg_iovlen = 0;
     cl->refs--;
     if (cl->peer_state == PEER_RDMA)
     {
-        return true;
+        return;
     }
     if (cl->peer_state == PEER_STOPPED)
     {
@@ -85,7 +90,7 @@ bool osd_messenger_t::handle_read(int result, osd_client_t *cl)
         {
             destroy_client(cl);
         }
-        return false;
+        return;
     }
     if (result <= 0 && result != -EAGAIN && result != -EINTR)
     {
@@ -95,9 +100,50 @@ bool osd_messenger_t::handle_read(int result, osd_client_t *cl)
             fprintf(stderr, "Client %ju socket read error: %d (%s). Disconnecting client\n", cl->client_id, -result, strerror(-result));
         }
         stop_client(cl->client_id);
-        return false;
+        return;
     }
-    if (result == -EAGAIN || result == -EINTR || result < cl->read_iov.iov_len)
+    bool full_read = false;
+    if (result > 0)
+    {
+        if (cl->read_iov.iov_base == cl->in_buf)
+        {
+            full_read = result >= cl->read_iov.iov_len;
+            if (!handle_read_buffer(cl, cl->in_buf, result))
+            {
+                handle_immediate_ops();
+                return;
+            }
+        }
+        else
+        {
+            // Reset OSD ping state
+            cl->ping_time_remaining = 0;
+            cl->idle_time_remaining = osd_idle_timeout;
+            // Long data
+            size_t i = 0;
+            while (i < cl->recv_list.size() && result >= cl->recv_list[i].iov_len)
+            {
+                result -= cl->recv_list[i].iov_len;
+                i++;
+            }
+            if (i < cl->recv_list.size())
+            {
+                cl->recv_list[i].iov_base += result;
+                cl->recv_list[i].iov_len -= result;
+            }
+            else
+            {
+                full_read = true;
+            }
+            cl->recv_list.erase(cl->recv_list.begin(), cl->recv_list.begin()+i);
+            if (!cl->recv_list.size())
+            {
+                handle_finished_op(cl);
+            }
+        }
+    }
+    cl->read_msg.msg_iovlen = 0;
+    if (result == -EAGAIN || result == -EINTR || !full_read)
     {
         cl->read_ready--;
         if (cl->read_ready > 0)
@@ -107,37 +153,7 @@ bool osd_messenger_t::handle_read(int result, osd_client_t *cl)
     {
         read_ready_clients.push_back(cl->client_id);
     }
-    if (result > 0)
-    {
-        if (cl->read_iov.iov_base == cl->in_buf)
-        {
-            if (!handle_read_buffer(cl, cl->in_buf, result))
-            {
-                handle_immediate_ops();
-                return false;
-            }
-        }
-        else
-        {
-            // Long data
-            cl->read_remaining -= result;
-            cl->recv_list.eat(result);
-            if (cl->recv_list.done >= cl->recv_list.count)
-            {
-                if (!handle_finished_read(cl))
-                {
-                    handle_immediate_ops();
-                    return false;
-                }
-            }
-        }
-        if (result >= cl->read_iov.iov_len)
-        {
-            ret = true;
-        }
-    }
     handle_immediate_ops();
-    return ret;
 }
 
 void osd_messenger_t::handle_immediate_ops()
@@ -162,113 +178,94 @@ void osd_messenger_t::handle_immediate_ops()
     }
 }
 
-bool osd_messenger_t::handle_read_buffer(osd_client_t *cl, void *curbuf, int remain)
+bool osd_messenger_t::handle_read_buffer(osd_client_t *cl, uint8_t *curbuf, size_t bufsize)
 {
+    // Reset OSD ping state
+    cl->ping_time_remaining = 0;
+    cl->idle_time_remaining = osd_idle_timeout;
     // Compose operation(s) from the buffer
-    while (remain > 0)
+    size_t done = 0;
+    while (done < bufsize)
     {
         if (!cl->read_op)
         {
             cl->read_op = new osd_op_t;
             cl->read_op->client_id = cl->client_id;
             cl->read_op->op_type = OSD_OP_IN;
-            cl->recv_list.push_back(cl->read_op->req.buf, OSD_PACKET_SIZE);
+            cl->read_op_pos = 0;
-            cl->read_remaining = OSD_PACKET_SIZE;
+            cl->read_op_size = 0;
-            cl->read_state = CL_READ_HDR;
         }
-        while (cl->recv_list.done < cl->recv_list.count && remain > 0)
+        if (cl->read_op_pos < OSD_PACKET_SIZE)
         {
-            iovec* cur = cl->recv_list.get_iovec();
+            int len = OSD_PACKET_SIZE - cl->read_op_pos;
-            if (cur->iov_len > remain)
+            if (len > bufsize-done)
-            {
+                len = bufsize-done;
-                memcpy(cur->iov_base, curbuf, remain);
+            memcpy(cl->read_op->req.buf + cl->read_op_pos, curbuf+done, len);
-                cl->read_remaining -= remain;
+            done += len;
-                cur->iov_len -= remain;
+            cl->read_op_pos += len;
-                cur->iov_base = (uint8_t*)cur->iov_base + remain;
+            if (cl->read_op_pos < OSD_PACKET_SIZE)
-                remain = 0;
+                return true;
-            }
+            if (!handle_hdr(cl))
-            else
-            {
-                memcpy(cur->iov_base, curbuf, cur->iov_len);
-                curbuf = (uint8_t*)curbuf + cur->iov_len;
-                cl->read_remaining -= cur->iov_len;
-                remain -= cur->iov_len;
-                cur->iov_len = 0;
-                cl->recv_list.done++;
-            }
-        }
-        if (cl->recv_list.done >= cl->recv_list.count)
-        {
-            if (!handle_finished_read(cl))
             {
+                stop_client(cl->client_id);
                 return false;
             }
         }
+        op_copy_from(cl, curbuf, bufsize, done);
     }
     return true;
 }
 
-bool osd_messenger_t::handle_finished_read(osd_client_t *cl)
+bool osd_messenger_t::handle_hdr(osd_client_t *cl)
 {
-    // Reset OSD ping state
+    if (cl->read_op->req.hdr.magic == SECONDARY_OSD_REPLY_MAGIC)
-    cl->ping_time_remaining = 0;
-    cl->idle_time_remaining = osd_idle_timeout;
-    cl->recv_list.reset();
-    if (cl->read_state == CL_READ_HDR)
     {
-        if (cl->read_op->req.hdr.magic == SECONDARY_OSD_REPLY_MAGIC)
+        auto req_it = cl->sent_ops.find(cl->read_op->req.hdr.id);
-            return handle_reply_hdr(cl);
+        if (req_it == cl->sent_ops.end())
-        else if (cl->read_op->req.hdr.magic == SECONDARY_OSD_OP_MAGIC)
         {
-            if (cl->check_sequencing)
+            // Command out of sync. Drop connection
-            {
+            fprintf(stderr, "Client %ju command out of sync: id %ju\n", cl->client_id, cl->read_op->req.hdr.id);
-                if (cl->read_op->req.hdr.id != cl->read_op_id)
+            return false;
-                {
-                    fprintf(stderr, "Warning: operation sequencing is broken on client %ju: expected num %ju, got %ju, stopping client\n", cl->client_id, cl->read_op_id, cl->read_op->req.hdr.id);
-                    stop_client(cl->client_id);
-                    return false;
-                }
-                cl->read_op_id++;
-            }
-            handle_op_hdr(cl);
         }
-        else
+        osd_op_t *op = req_it->second;
+        memcpy(op->reply.buf, cl->read_op->req.buf, OSD_PACKET_SIZE);
+        if (!allocate_reply_buffers(cl, op))
+        {
+            return false;
+        }
+        cl->sent_ops.erase(req_it);
+        delete cl->read_op;
+        cl->read_op = op;
+    }
+    else if (cl->read_op->req.hdr.magic == SECONDARY_OSD_OP_MAGIC)
+    {
+        if (cl->check_sequencing)
+        {
+            if (cl->read_op->req.hdr.id != cl->read_op_id)
+            {
+                fprintf(stderr, "Warning: operation sequencing is broken on client %d: expected num %ju, got %ju, stopping client\n", cl->peer_fd, cl->read_op_id, cl->read_op->req.hdr.id);
+                return false;
+            }
+            cl->read_op_id++;
+        }
+        if (!allocate_op_buffers(cl))
         {
-            fprintf(stderr, "Received garbage: magic=%jx id=%ju opcode=%jx from client %ju\n", cl->read_op->req.hdr.magic, cl->read_op->req.hdr.id, cl->read_op->req.hdr.opcode, cl->client_id);
-            stop_client(cl->client_id);
             return false;
         }
     }
-    else if (cl->read_state == CL_READ_DATA)
-    {
-        // Operation is ready
-        cl->received_ops.push_back(cl->read_op);
-        set_immediate_ops.push_back(cl->read_op);
-        cl->read_op = NULL;
-        cl->read_state = 0;
-    }
-    else if (cl->read_state == CL_READ_REPLY_DATA)
-    {
-        // Reply is ready
-        handle_reply_ready(cl->read_op);
-        cl->read_op = NULL;
-        cl->read_state = 0;
-    }
     else
     {
-        assert(0);
+        fprintf(stderr, "Received garbage: magic=%jx id=%ju opcode=%jx from client %ju\n", cl->read_op->req.hdr.magic, cl->read_op->req.hdr.id, cl->read_op->req.hdr.opcode, cl->client_id);
+        return false;
     }
     return true;
 }
 
-void osd_messenger_t::handle_op_hdr(osd_client_t *cl)
+bool osd_messenger_t::allocate_op_buffers(osd_client_t *cl)
 {
     osd_op_t *cur_op = cl->read_op;
-    if (cur_op->req.hdr.opcode == OSD_OP_SEC_READ)
+    cl->read_op_size = 0;
-    {
+    if (cur_op->req.hdr.opcode == OSD_OP_SEC_WRITE ||
-        cl->read_remaining = 0;
-    }
-    else if (cur_op->req.hdr.opcode == OSD_OP_SEC_WRITE ||
         cur_op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE)
     {
         if (cur_op->req.sec_rw.attr_len > 0)
@@ -277,14 +274,12 @@ void osd_messenger_t::handle_op_hdr(osd_client_t *cl)
                 cur_op->bitmap = cur_op->rmw_buf = malloc_or_die(cur_op->req.sec_rw.attr_len);
             else
                 cur_op->bitmap = &cur_op->bmp_data;
-            cl->recv_list.push_back(cur_op->bitmap, cur_op->req.sec_rw.attr_len);
         }
         if (cur_op->req.sec_rw.len > 0)
         {
             cur_op->buf = memalign_or_die(MEM_ALIGNMENT, cur_op->req.sec_rw.len);
-            cl->recv_list.push_back(cur_op->buf, cur_op->req.sec_rw.len);
         }
-        cl->read_remaining = cur_op->req.sec_rw.len + cur_op->req.sec_rw.attr_len;
+        cl->read_op_size = cur_op->req.sec_rw.len + cur_op->req.sec_rw.attr_len;
     }
     else if (cur_op->req.hdr.opcode == OSD_OP_SEC_STABILIZE ||
         cur_op->req.hdr.opcode == OSD_OP_SEC_ROLLBACK)
@@ -292,27 +287,24 @@ void osd_messenger_t::handle_op_hdr(osd_client_t *cl)
         if (cur_op->req.sec_stab.len > 0)
         {
             cur_op->buf = memalign_or_die(MEM_ALIGNMENT, cur_op->req.sec_stab.len);
-            cl->recv_list.push_back(cur_op->buf, cur_op->req.sec_stab.len);
         }
-        cl->read_remaining = cur_op->req.sec_stab.len;
+        cl->read_op_size = cur_op->req.sec_stab.len;
     }
     else if (cur_op->req.hdr.opcode == OSD_OP_SEC_READ_BMP)
     {
         if (cur_op->req.sec_read_bmp.len > 0)
         {
             cur_op->buf = memalign_or_die(MEM_ALIGNMENT, cur_op->req.sec_read_bmp.len);
-            cl->recv_list.push_back(cur_op->buf, cur_op->req.sec_read_bmp.len);
         }
-        cl->read_remaining = cur_op->req.sec_read_bmp.len;
+        cl->read_op_size = cur_op->req.sec_read_bmp.len;
     }
     else if (cur_op->req.hdr.opcode == OSD_OP_WRITE)
     {
         if (cur_op->req.rw.len > 0)
         {
             cur_op->buf = memalign_or_die(MEM_ALIGNMENT, cur_op->req.rw.len);
-            cl->recv_list.push_back(cur_op->buf, cur_op->req.rw.len);
         }
-        cl->read_remaining = cur_op->req.rw.len;
+        cl->read_op_size = cur_op->req.rw.len;
     }
     else if (cur_op->req.hdr.opcode == OSD_OP_SHOW_CONFIG)
     {
@@ -320,44 +312,15 @@ void osd_messenger_t::handle_op_hdr(osd_client_t *cl)
         {
             cur_op->buf = malloc_or_die(cur_op->req.show_conf.json_len+1);
             ((uint8_t*)cur_op->buf)[cur_op->req.show_conf.json_len] = 0;
-            cl->recv_list.push_back(cur_op->buf, cur_op->req.show_conf.json_len);
         }
-        cl->read_remaining = cur_op->req.show_conf.json_len;
+        cl->read_op_size = cur_op->req.show_conf.json_len;
-    }
-    /*else if (cur_op->req.hdr.opcode == OSD_OP_READ ||
-        cur_op->req.hdr.opcode == OSD_OP_SCRUB ||
-        cur_op->req.hdr.opcode == OSD_OP_DESCRIBE)
-    {
-        cl->read_remaining = 0;
-    }*/
-    if (cl->read_remaining > 0)
-    {
-        // Read data
-        cl->read_state = CL_READ_DATA;
-    }
-    else
-    {
-        // Operation is ready
-        cl->received_ops.push_back(cur_op);
-        set_immediate_ops.push_back(cur_op);
-        cl->read_op = NULL;
-        cl->read_state = 0;
     }
+    return true;
 }
 
-bool osd_messenger_t::handle_reply_hdr(osd_client_t *cl)
+bool osd_messenger_t::allocate_reply_buffers(osd_client_t *cl, osd_op_t *op)
 {
-    auto req_it = cl->sent_ops.find(cl->read_op->req.hdr.id);
+    cl->read_op_size = 0;
-    if (req_it == cl->sent_ops.end())
-    {
-        // Command out of sync. Drop connection
-        fprintf(stderr, "Client %ju command out of sync: id %ju\n", cl->client_id, cl->read_op->req.hdr.id);
-        stop_client(cl->client_id);
-        return false;
-    }
-    osd_op_t *op = req_it->second;
-    memcpy(op->reply.buf, cl->read_op->req.buf, OSD_PACKET_SIZE);
-    cl->sent_ops.erase(req_it);
     if (op->reply.hdr.opcode == OSD_OP_SEC_READ || op->reply.hdr.opcode == OSD_OP_READ)
     {
         // Read data. In this case we assume that the buffer is preallocated by the caller (!)
@@ -368,97 +331,279 @@ bool osd_messenger_t::handle_reply_hdr(osd_client_t *cl)
             // Check reply length to not overflow the buffer
             fprintf(stderr, "Client %ju read reply of different length: expected %u+%u, got %jd+%u\n",
                 cl->client_id, expected_size, op->bitmap_len, op->reply.hdr.retval, bmp_len);
-            cl->sent_ops[op->req.hdr.id] = op;
-            stop_client(cl->client_id);
             return false;
         }
         if (bmp_len > 0)
         {
             assert(op->bitmap);
-            cl->recv_list.push_back(op->bitmap, bmp_len);
+            cl->read_op_size += bmp_len;
-            cl->read_remaining += bmp_len;
         }
         if (op->reply.hdr.retval > 0)
         {
             assert(op->iov.count > 0);
-            cl->recv_list.append(op->iov);
+            cl->read_op_size += op->reply.hdr.retval;
-            cl->read_remaining += op->reply.hdr.retval;
         }
-        if (cl->read_remaining == 0)
-        {
-            goto reuse;
-        }
-        delete cl->read_op;
-        cl->read_op = op;
-        cl->read_state = CL_READ_REPLY_DATA;
     }
     else if (op->reply.hdr.opcode == OSD_OP_SEC_LIST && op->reply.hdr.retval > 0)
     {
         assert(!op->iov.count);
-        delete cl->read_op;
+        cl->read_op_size = sizeof(obj_ver_id) * op->reply.hdr.retval;
-        cl->read_op = op;
+        op->buf = memalign_or_die(MEM_ALIGNMENT, cl->read_op_size);
-        cl->read_state = CL_READ_REPLY_DATA;
-        cl->read_remaining = sizeof(obj_ver_id) * op->reply.hdr.retval;
-        op->buf = memalign_or_die(MEM_ALIGNMENT, cl->read_remaining);
-        cl->recv_list.push_back(op->buf, cl->read_remaining);
     }
     else if (op->reply.hdr.opcode == OSD_OP_SEC_READ_BMP && op->reply.hdr.retval > 0)
     {
         assert(!op->iov.count);
-        delete cl->read_op;
+        cl->read_op_size = op->reply.hdr.retval;
-        cl->read_op = op;
-        cl->read_state = CL_READ_REPLY_DATA;
-        cl->read_remaining = op->reply.hdr.retval;
         free(op->buf);
-        op->buf = memalign_or_die(MEM_ALIGNMENT, cl->read_remaining);
+        op->buf = memalign_or_die(MEM_ALIGNMENT, cl->read_op_size);
-        cl->recv_list.push_back(op->buf, cl->read_remaining);
     }
     else if (op->reply.hdr.opcode == OSD_OP_SHOW_CONFIG && op->reply.hdr.retval > 0)
     {
-        delete cl->read_op;
+        cl->read_op_size = op->reply.hdr.retval;
-        cl->read_op = op;
-        cl->read_state = CL_READ_REPLY_DATA;
-        cl->read_remaining = op->reply.hdr.retval;
         free(op->buf);
         op->buf = malloc_or_die(op->reply.hdr.retval);
-        cl->recv_list.push_back(op->buf, op->reply.hdr.retval);
     }
     else if (op->reply.hdr.opcode == OSD_OP_DESCRIBE && op->reply.describe.result_bytes > 0)
     {
-        delete cl->read_op;
+        cl->read_op_size = op->reply.describe.result_bytes;
-        cl->read_op = op;
-        cl->read_state = CL_READ_REPLY_DATA;
-        cl->read_remaining = op->reply.describe.result_bytes;
         free(op->buf);
         op->buf = malloc_or_die(op->reply.describe.result_bytes);
-        cl->recv_list.push_back(op->buf, op->reply.describe.result_bytes);
-    }
-    else
-    {
-reuse:
-        // It's fine to reuse cl->read_op for the next reply
-        handle_reply_ready(op);
-        cl->recv_list.push_back(cl->read_op->req.buf, OSD_PACKET_SIZE);
-        cl->read_remaining = OSD_PACKET_SIZE;
-        cl->read_state = CL_READ_HDR;
     }
     return true;
 }
 
-void osd_messenger_t::handle_reply_ready(osd_op_t *op)
+size_t osd_messenger_t::op_copy_from(osd_client_t *cl, uint8_t *src, size_t src_len, size_t & done)
 {
-    // Measure subop latency
+    osd_op_t *op = cl->read_op;
-    timespec tv_end;
+    size_t from = cl->read_op_pos-OSD_PACKET_SIZE;
-    clock_gettime(CLOCK_REALTIME, &tv_end);
+    auto op_read_buf = [&](uint8_t *dst, size_t dst_len)
-    stats.subop_stat_count[op->req.hdr.opcode]++;
-    if (!stats.subop_stat_count[op->req.hdr.opcode])
     {
-        stats.subop_stat_count[op->req.hdr.opcode]++;
+        if (from < dst_len)
-        stats.subop_stat_sum[op->req.hdr.opcode] = 0;
+        {
+            size_t n = dst_len-from;
+            if (n > src_len-done)
+                n = src_len-done;
+            memcpy(dst+from, src+done, n);
+            done += n;
+            cl->read_op_pos += n;
+            from += n;
+            if (from < dst_len)
+                return false;
+            from = 0;
+        }
+        else
+            from -= dst_len;
+        return true;
+    };
+    if (op->op_type == OSD_OP_IN)
+    {
+        if (op->req.hdr.opcode == OSD_OP_SEC_WRITE ||
+            op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE)
+        {
+            if (!op_read_buf((uint8_t*)op->bitmap, op->req.sec_rw.attr_len))
+                return done;
+            if (!op_read_buf((uint8_t*)op->buf, op->req.sec_rw.len))
+                return done;
+        }
+        else if (op->req.hdr.opcode == OSD_OP_SEC_STABILIZE ||
+            op->req.hdr.opcode == OSD_OP_SEC_ROLLBACK)
+        {
+            if (!op_read_buf((uint8_t*)op->buf, op->req.sec_stab.len))
+                return done;
+        }
+        else if (op->req.hdr.opcode == OSD_OP_SEC_READ_BMP)
+        {
+            if (!op_read_buf((uint8_t*)op->buf, op->req.sec_read_bmp.len))
+                return done;
+        }
+        else if (op->req.hdr.opcode == OSD_OP_WRITE)
+        {
+            if (!op_read_buf((uint8_t*)op->buf, op->req.rw.len))
+                return done;
+        }
+        else if (op->req.hdr.opcode == OSD_OP_SHOW_CONFIG)
+        {
+            if (!op_read_buf((uint8_t*)op->buf, op->req.show_conf.json_len))
+                return done;
+        }
     }
-    stats.subop_stat_sum[op->req.hdr.opcode] += (
+    else
-        (tv_end.tv_sec - op->tv_begin.tv_sec)*1000000 +
+    {
-        (tv_end.tv_nsec - op->tv_begin.tv_nsec)/1000
+        if (op->reply.hdr.opcode == OSD_OP_SEC_READ)
-    );
+        {
-    set_immediate_ops.push_back(op);
+            if (op->reply.sec_rw.attr_len > 0)
+            {
+                if (!op_read_buf((uint8_t*)op->bitmap, op->reply.sec_rw.attr_len))
+                    return done;
+            }
+            if (op->reply.hdr.retval > 0)
+            {
+                for (int i = 0; i < op->iov.count; i++)
+                    if (!op_read_buf((uint8_t*)op->iov.buf[i].iov_base, op->iov.buf[i].iov_len))
+                        return done;
+            }
+        }
+        else if (op->reply.hdr.opcode == OSD_OP_READ)
+        {
+            if (op->reply.rw.bitmap_len > 0)
+            {
+                if (!op_read_buf((uint8_t*)op->bitmap, op->reply.rw.bitmap_len))
+                    return done;
+            }
+            if (op->reply.hdr.retval > 0)
+            {
+                for (int i = 0; i < op->iov.count; i++)
+                    if (!op_read_buf((uint8_t*)op->iov.buf[i].iov_base, op->iov.buf[i].iov_len))
+                        return done;
+            }
+        }
+        else if (op->reply.hdr.opcode == OSD_OP_SEC_LIST && op->reply.hdr.retval > 0)
+        {
+            if (!op_read_buf((uint8_t*)op->buf, sizeof(obj_ver_id) * op->reply.hdr.retval))
+                return done;
+        }
+        else if ((op->reply.hdr.opcode == OSD_OP_SEC_READ_BMP ||
+            op->reply.hdr.opcode == OSD_OP_SHOW_CONFIG) && op->reply.hdr.retval > 0)
+        {
+            if (!op_read_buf((uint8_t*)op->buf, op->reply.hdr.retval))
+                return done;
+        }
+        else if (op->reply.hdr.opcode == OSD_OP_DESCRIBE && op->reply.describe.result_bytes > 0)
+        {
+            if (!op_read_buf((uint8_t*)op->buf, op->reply.describe.result_bytes))
+                return done;
+        }
+    }
+    handle_finished_op(cl);
+    return done;
+}
+
+size_t osd_messenger_t::op_get_read_buffers(osd_client_t *cl, std::vector<iovec> & lst)
+{
+    osd_op_t *op = cl->read_op;
+    size_t from = cl->read_op_pos-OSD_PACKET_SIZE;
+    size_t done = 0;
+    auto op_read_buf = [&](uint8_t *dst, size_t dst_len)
+    {
+        if (lst.size() >= IOV_MAX)
+            return false;
+        if (from < dst_len)
+        {
+            lst.push_back((iovec){ .iov_base = dst+from, .iov_len = dst_len-from });
+            cl->read_op_pos += dst_len-from;
+            done += dst_len-from;
+            from = 0;
+        }
+        else
+            from -= dst_len;
+        return true;
+    };
+    if (op->op_type == OSD_OP_IN)
+    {
+        if (op->req.hdr.opcode == OSD_OP_SEC_WRITE ||
+            op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE)
+        {
+            if (!op_read_buf((uint8_t*)op->bitmap, op->req.sec_rw.attr_len))
+                return done;
+            if (!op_read_buf((uint8_t*)op->buf, op->req.sec_rw.len))
+                return done;
+        }
+        else if (op->req.hdr.opcode == OSD_OP_SEC_STABILIZE ||
+            op->req.hdr.opcode == OSD_OP_SEC_ROLLBACK)
+        {
+            if (!op_read_buf((uint8_t*)op->buf, op->req.sec_stab.len))
+                return done;
+        }
+        else if (op->req.hdr.opcode == OSD_OP_SEC_READ_BMP)
+        {
+            if (!op_read_buf((uint8_t*)op->buf, op->req.sec_read_bmp.len))
+                return done;
+        }
+        else if (op->req.hdr.opcode == OSD_OP_WRITE)
+        {
+            if (!op_read_buf((uint8_t*)op->buf, op->req.rw.len))
+                return done;
+        }
+        else if (op->req.hdr.opcode == OSD_OP_SHOW_CONFIG)
+        {
+            if (!op_read_buf((uint8_t*)op->buf, op->req.show_conf.json_len))
+                return done;
+        }
+    }
+    else
+    {
+        if (op->reply.hdr.opcode == OSD_OP_SEC_READ)
+        {
+            if (op->reply.sec_rw.attr_len > 0)
+            {
+                if (!op_read_buf((uint8_t*)op->bitmap, op->reply.sec_rw.attr_len))
+                    return done;
+            }
+            if (op->reply.hdr.retval > 0)
+            {
+                for (int i = 0; i < op->iov.count; i++)
+                    if (!op_read_buf((uint8_t*)op->iov.buf[i].iov_base, op->iov.buf[i].iov_len))
+                        return done;
+            }
+        }
+        else if (op->reply.hdr.opcode == OSD_OP_READ)
+        {
+            if (op->reply.rw.bitmap_len > 0)
+            {
+                if (!op_read_buf((uint8_t*)op->bitmap, op->reply.rw.bitmap_len))
+                    return done;
+            }
+            if (op->reply.hdr.retval > 0)
+            {
+                for (int i = 0; i < op->iov.count; i++)
+                    if (!op_read_buf((uint8_t*)op->iov.buf[i].iov_base, op->iov.buf[i].iov_len))
+                        return done;
+            }
+        }
+        else if (op->reply.hdr.opcode == OSD_OP_SEC_LIST && op->reply.hdr.retval > 0)
+        {
+            if (!op_read_buf((uint8_t*)op->buf, sizeof(obj_ver_id) * op->reply.hdr.retval))
+                return done;
+        }
+        else if ((op->reply.hdr.opcode == OSD_OP_SEC_READ_BMP ||
+            op->reply.hdr.opcode == OSD_OP_SHOW_CONFIG) && op->reply.hdr.retval > 0)
+        {
+            if (!op_read_buf((uint8_t*)op->buf, op->reply.hdr.retval))
+                return done;
+        }
+        else if (op->reply.hdr.opcode == OSD_OP_DESCRIBE && op->reply.describe.result_bytes > 0)
+        {
+            if (!op_read_buf((uint8_t*)op->buf, op->reply.describe.result_bytes))
+                return done;
+        }
+    }
+    return done;
+}
+
+void osd_messenger_t::handle_finished_op(osd_client_t *cl)
+{
+    osd_op_t *op = cl->read_op;
+    if (op->op_type == OSD_OP_IN)
+    {
+        // Operation is ready
+        cl->received_ops.push_back(op);
+    }
+    else
+    {
+        // Measure subop (outbound op) latency
+        timespec tv_end;
+        clock_gettime(CLOCK_REALTIME, &tv_end);
+        stats.subop_stat_count[op->req.hdr.opcode]++;
+        if (!stats.subop_stat_count[op->req.hdr.opcode])
+        {
+            stats.subop_stat_count[op->req.hdr.opcode]++;
+            stats.subop_stat_sum[op->req.hdr.opcode] = 0;
+        }
+        stats.subop_stat_sum[op->req.hdr.opcode] += (
+            (tv_end.tv_sec - op->tv_begin.tv_sec)*1000000 +
+            (tv_end.tv_nsec - op->tv_begin.tv_nsec)/1000
+        );
+    }
+    set_immediate_ops.push_back(op);
+    cl->read_op = NULL;
 }

src/client/msgr_send.cpp

@@ -5,6 +5,12 @@
 #include <limits.h>
 #include <sys/epoll.h>
 
+#ifdef WITH_OPENSSL
+#include <openssl/conf.h>
+#include <openssl/evp.h>
+#include <openssl/err.h>
+#endif
+
 #include "messenger.h"
 
 void osd_messenger_t::outbox_push(osd_op_t *cur_op)
@@ -21,6 +27,7 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
     {
         clock_gettime(CLOCK_REALTIME, &cur_op->tv_begin);
         cur_op->req.hdr.id = ++cl->send_op_id;
+        cl->sent_ops[cur_op->req.hdr.id] = cur_op;
     }
     else
     {
@@ -37,77 +44,9 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
         }
         // Can't be not found because client IDs are unique
         assert(found);
-    }
-    auto & to_send_list = cl->write_msg.msg_iovlen ? cl->next_send_list : cl->send_list;
-    auto & to_outbox = cl->write_msg.msg_iovlen ? cl->next_outbox : cl->outbox;
-    if (cur_op->op_type == OSD_OP_IN)
-    {
         measure_exec(cur_op);
-        to_send_list.push_back((iovec){ .iov_base = cur_op->reply.buf, .iov_len = OSD_PACKET_SIZE });
-    }
-    else
-    {
-        to_send_list.push_back((iovec){ .iov_base = cur_op->req.buf, .iov_len = OSD_PACKET_SIZE });
-        cl->sent_ops[cur_op->req.hdr.id] = cur_op;
-    }
-    to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = MSGR_SENDP_HDR });
-    // Bitmap
-    if (cur_op->op_type == OSD_OP_IN &&
-        cur_op->req.hdr.opcode == OSD_OP_SEC_READ &&
-        cur_op->reply.sec_rw.attr_len > 0)
-    {
-        to_send_list.push_back((iovec){
-            .iov_base = cur_op->bitmap,
-            .iov_len = cur_op->reply.sec_rw.attr_len,
-        });
-        to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = 0 });
-    }
-    else if (cur_op->op_type == OSD_OP_OUT &&
-        (cur_op->req.hdr.opcode == OSD_OP_SEC_WRITE || cur_op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE) &&
-        cur_op->req.sec_rw.attr_len > 0)
-    {
-        to_send_list.push_back((iovec){
-            .iov_base = cur_op->bitmap,
-            .iov_len = cur_op->req.sec_rw.attr_len,
-        });
-        to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = 0 });
-    }
-    // Operation data
-    if ((cur_op->op_type == OSD_OP_IN
-        ? (cur_op->req.hdr.opcode == OSD_OP_READ ||
-        cur_op->req.hdr.opcode == OSD_OP_SEC_READ ||
-        cur_op->req.hdr.opcode == OSD_OP_SEC_LIST ||
-        cur_op->req.hdr.opcode == OSD_OP_SHOW_CONFIG ||
-        cur_op->req.hdr.opcode == OSD_OP_DESCRIBE)
-        : (cur_op->req.hdr.opcode == OSD_OP_WRITE ||
-        cur_op->req.hdr.opcode == OSD_OP_SEC_WRITE ||
-        cur_op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE ||
-        cur_op->req.hdr.opcode == OSD_OP_SEC_STABILIZE ||
-        cur_op->req.hdr.opcode == OSD_OP_SEC_ROLLBACK ||
-        cur_op->req.hdr.opcode == OSD_OP_SHOW_CONFIG)) && cur_op->iov.count > 0)
-    {
-        for (int i = 0; i < cur_op->iov.count; i++)
-        {
-            if (cur_op->iov.buf[i].iov_len > 0)
-            {
-                assert(cur_op->iov.buf[i].iov_base);
-                to_send_list.push_back(cur_op->iov.buf[i]);
-                to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = 0 });
-            }
-        }
-    }
-    if (cur_op->req.hdr.opcode == OSD_OP_SEC_READ_BMP)
-    {
-        if (cur_op->op_type == OSD_OP_IN && cur_op->reply.hdr.retval > 0)
-            to_send_list.push_back((iovec){ .iov_base = cur_op->buf, .iov_len = (size_t)cur_op->reply.hdr.retval });
-        else if (cur_op->op_type == OSD_OP_OUT && cur_op->req.sec_read_bmp.len > 0)
-            to_send_list.push_back((iovec){ .iov_base = cur_op->buf, .iov_len = (size_t)cur_op->req.sec_read_bmp.len });
-        to_outbox.push_back((msgr_sendp_t){ .op = cur_op, .flags = 0 });
-    }
-    if (cur_op->op_type == OSD_OP_IN)
-    {
-        to_outbox[to_outbox.size()-1].flags |= MSGR_SENDP_FREE;
     }
+    cl->write_ops.push_back(cur_op);
 #ifdef WITH_RDMA
     if (cl->peer_state == PEER_RDMA)
     {
@@ -118,7 +57,7 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
     if (!ringloop)
     {
         // FIXME: It's worse because it doesn't allow batching
-        while (cl->outbox.size())
+        while (cl->write_ops.size())
         {
             try_send(cl);
         }
@@ -185,11 +124,25 @@ void osd_messenger_t::measure_exec(osd_op_t *cur_op)
 
 bool osd_messenger_t::try_send(osd_client_t *cl)
 {
-    if (!cl->send_list.size() || cl->write_msg.msg_iovlen > 0 || cl->peer_state == PEER_STOPPED || cl->peer_fd < 0)
+    if (!cl->write_op && !cl->write_ops.size() || cl->write_msg.msg_iovlen > 0 || cl->peer_state == PEER_STOPPED || cl->peer_fd < 0)
     {
         return true;
     }
     assert(cl->peer_state != PEER_RDMA);
+    while ((cl->write_op || cl->write_ops.size()) && cl->send_list.size() < IOV_MAX)
+    {
+        if (!cl->write_op)
+        {
+            cl->write_op = cl->write_ops.front();
+            cl->write_ops.pop_front();
+        }
+        osd_op_t *op = cl->write_op;
+        op_get_write_buffers(cl, cl->send_list);
+        if (!cl->write_op && op->op_type == OSD_OP_IN)
+        {
+            cl->send_free_ops.push_back(op);
+        }
+    }
     if (ringloop && !use_sync_send_recv)
     {
         auto iothread = iothreads.size() ? iothreads[cl->peer_fd % iothreads.size()] : NULL;
@@ -202,20 +155,24 @@ bool osd_messenger_t::try_send(osd_client_t *cl)
             data_local = {};
         }
         if (!sqe)
+        {
             return false;
+        }
+        cl->send_list_size = 0;
+        for (auto & iov: cl->send_list)
+        {
+            cl->send_list_size += iov.iov_len;
+        }
         cl->write_msg.msg_iov = cl->send_list.data();
         cl->write_msg.msg_iovlen = cl->send_list.size() < IOV_MAX ? cl->send_list.size() : IOV_MAX;
         cl->refs++;
         ring_data_t* data = ((ring_data_t*)sqe->user_data);
         data->callback = [this, cl](ring_data_t *data) { handle_send(data->res, data->prev, data->more, cl); };
         bool use_zc = has_sendmsg_zc && min_zerocopy_send_size >= 0;
-        if (use_zc && min_zerocopy_send_size > 0)
+        if (use_zc && min_zerocopy_send_size > 0 &&
+            cl->send_list_size/cl->write_msg.msg_iovlen < min_zerocopy_send_size)
         {
-            size_t avg_size = 0;
+            use_zc = false;
-            for (size_t i = 0; i < cl->write_msg.msg_iovlen; i++)
-                avg_size += cl->write_msg.msg_iov[i].iov_len;
-            if (avg_size/cl->write_msg.msg_iovlen < min_zerocopy_send_size)
-                use_zc = false;
         }
         if (use_zc)
         {
@@ -266,6 +223,7 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
     if (!prev)
     {
         cl->write_msg.msg_iovlen = 0;
+        cl->send_list.clear();
     }
     if (!more)
     {
@@ -298,57 +256,26 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
                 cl->zc_free_list.erase(cl->zc_free_list.begin(), cl->zc_free_list.begin()+i+1);
             return;
         }
-        int done = 0;
+        if (cl->send_list_size > result)
-        while (result > 0 && done < cl->send_list.size())
         {
-            iovec & iov = cl->send_list[done];
+            fprintf(stderr, "Client %ju socket write error: expected to send "
-            if (iov.iov_len <= result)
+                "%zu bytes with MSG_WAITALL but sent %u. Disconnecting client\n", cl->client_id, cl->send_list_size, result);
-            {
+            stop_client(cl->peer_fd);
-                if (cl->outbox[done].flags & MSGR_SENDP_FREE)
+            return;
-                {
+        }
-                    // Reply fully sent
+        for (auto op: cl->send_free_ops)
-                    if (more)
+        {
-                        cl->zc_free_list.push_back(cl->outbox[done].op);
+            if (more)
-                    else
+                cl->zc_free_list.push_back(op);
-                        delete cl->outbox[done].op;
-                }
-                result -= iov.iov_len;
-                done++;
-            }
             else
-            {
+                delete op;
-                iov.iov_len -= result;
-                iov.iov_base = (uint8_t*)iov.iov_base + result;
-                break;
-            }
         }
         if (more)
-        {
-            int expected = cl->send_list.size() < IOV_MAX ? cl->send_list.size() : IOV_MAX;
-            if (done != expected)
-            {
-                fprintf(stderr, "Client %ju socket write error: expected to send "
-                    "%d iovecs with MSG_WAITALL but sent %d. Disconnecting client\n", cl->client_id, expected, done);
-                stop_client(cl->client_id);
-                return;
-            }
             cl->zc_free_list.push_back(NULL); // end marker
-        }
+        cl->send_free_ops.clear();
-        if (done > 0)
+        cl->write_state = cl->write_op || cl->write_ops.size() ? CL_WRITE_READY : 0;
-        {
-            cl->send_list.erase(cl->send_list.begin(), cl->send_list.begin()+done);
-            cl->outbox.erase(cl->outbox.begin(), cl->outbox.begin()+done);
-        }
-        if (cl->next_send_list.size())
-        {
-            cl->send_list.insert(cl->send_list.end(), cl->next_send_list.begin(), cl->next_send_list.end());
-            cl->outbox.insert(cl->outbox.end(), cl->next_outbox.begin(), cl->next_outbox.end());
-            cl->next_send_list.clear();
-            cl->next_outbox.clear();
-        }
-        cl->write_state = cl->outbox.size() > 0 ? CL_WRITE_READY : 0;
 #ifdef WITH_RDMA
-        if (cl->rdma_conn && !cl->outbox.size() && cl->peer_state == PEER_RDMA_CONNECTING)
+        if (cl->rdma_conn && !cl->write_op && !cl->write_ops.size() && cl->peer_state == PEER_RDMA_CONNECTING)
         {
             // FIXME: Ignore pings during RDMA state transition
             if (log_level > 0)
@@ -366,3 +293,130 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t
         write_ready_clients.push_back(cl->client_id);
     }
 }
+
+static inline bool op_write_headers(osd_op_t *op, std::function<bool(uint8_t*, size_t)> op_write_buf)
+{
+    // Header
+    if (!op_write_buf((op->op_type == OSD_OP_IN ? op->reply.buf : op->req.buf), OSD_PACKET_SIZE))
+        return false;
+    // Bitmap
+    if (op->op_type == OSD_OP_IN &&
+        op->req.hdr.opcode == OSD_OP_SEC_READ &&
+        op->reply.sec_rw.attr_len > 0)
+    {
+        if (!op_write_buf((uint8_t*)op->bitmap, op->reply.sec_rw.attr_len))
+            return false;
+    }
+    else if (op->op_type == OSD_OP_OUT &&
+        (op->req.hdr.opcode == OSD_OP_SEC_WRITE || op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE) &&
+        op->req.sec_rw.attr_len > 0)
+    {
+        if (!op_write_buf((uint8_t*)op->bitmap, op->req.sec_rw.attr_len))
+            return false;
+    }
+    if (op->req.hdr.opcode == OSD_OP_SEC_READ_BMP)
+    {
+        if (op->op_type == OSD_OP_IN && op->reply.hdr.retval > 0)
+        {
+            if (!op_write_buf((uint8_t*)op->buf, (size_t)op->reply.hdr.retval))
+                return false;
+        }
+        else if (op->op_type == OSD_OP_OUT && op->req.sec_read_bmp.len > 0)
+        {
+            if (!op_write_buf((uint8_t*)op->buf, (size_t)op->req.sec_read_bmp.len))
+                return false;
+        }
+    }
+    return true;
+}
+
+static inline bool op_has_data(osd_op_t *op)
+{
+    return (op->op_type == OSD_OP_IN
+        ? (op->req.hdr.opcode == OSD_OP_READ ||
+        op->req.hdr.opcode == OSD_OP_SEC_READ ||
+        op->req.hdr.opcode == OSD_OP_SEC_LIST ||
+        op->req.hdr.opcode == OSD_OP_SHOW_CONFIG ||
+        op->req.hdr.opcode == OSD_OP_DESCRIBE)
+        : (op->req.hdr.opcode == OSD_OP_WRITE ||
+        op->req.hdr.opcode == OSD_OP_SEC_WRITE ||
+        op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE ||
+        op->req.hdr.opcode == OSD_OP_SEC_STABILIZE ||
+        op->req.hdr.opcode == OSD_OP_SEC_ROLLBACK ||
+        op->req.hdr.opcode == OSD_OP_SHOW_CONFIG)) && op->iov.count > 0;
+}
+
+size_t osd_messenger_t::op_copy_to(osd_client_t *cl, uint8_t *dst, size_t dst_len)
+{
+    size_t done = 0;
+    size_t from = cl->write_op_pos;
+    auto op_write_buf = [&](uint8_t *src, size_t src_len)
+    {
+        if (from < src_len)
+        {
+            size_t n = src_len-from;
+            if (n > dst_len-done)
+                n = dst_len-done;
+            memcpy(dst+done, src+from, n);
+            done += n;
+            cl->write_op_pos += n;
+            from += n;
+            if (from < src_len)
+                return false;
+            from = 0;
+        }
+        else
+            from -= src_len;
+        return true;
+    };
+    if (!op_write_headers(cl->write_op, op_write_buf))
+    {
+        return done;
+    }
+    // Operation data
+    if (op_has_data(cl->write_op))
+    {
+        for (int i = 0; i < cl->write_op->iov.count; i++)
+        {
+            if (!op_write_buf((uint8_t*)cl->write_op->iov.buf[i].iov_base, cl->write_op->iov.buf[i].iov_len))
+                return done;
+        }
+    }
+    cl->write_op = NULL;
+    cl->write_op_pos = 0;
+    return done;
+}
+
+void osd_messenger_t::op_get_write_buffers(osd_client_t *cl, std::vector<iovec> & lst)
+{
+    size_t from = cl->write_op_pos;
+    auto op_write_buf = [&](uint8_t *src, size_t src_len)
+    {
+        if (lst.size() >= IOV_MAX)
+            return false;
+        if (from < src_len)
+        {
+            lst.push_back((iovec){ .iov_base = src+from, .iov_len = src_len-from });
+            cl->write_op_pos += src_len-from;
+            from = 0;
+        }
+        else
+            from -= src_len;
+        return true;
+    };
+    if (!op_write_headers(cl->write_op, op_write_buf))
+    {
+        return;
+    }
+    // Operation data
+    if (op_has_data(cl->write_op))
+    {
+        for (int i = 0; i < cl->write_op->iov.count; i++)
+        {
+            if (!op_write_buf((uint8_t*)cl->write_op->iov.buf[i].iov_base, cl->write_op->iov.buf[i].iov_len))
+                return;
+        }
+    }
+    cl->write_op = NULL;
+    cl->write_op_pos = 0;
+}

src/client/msgr_stop.cpp

@@ -189,6 +189,13 @@ osd_client_t::~osd_client_t()
     }
     // Cancel outbound ops
     cancel_ops();
+    for (osd_op_t *op: send_free_ops)
+    {
+        if (op)
+        {
+            delete op;
+        }
+    }
     for (osd_op_t *op: zc_free_list)
     {
         if (op)