WIP Batch encrypted records to reduce EncryptInit/EncryptFinal costs

src/client/messenger.h

@@ -105,9 +105,13 @@ struct osd_client_t
     EVP_CIPHER_CTX *enc_ctx = NULL;
     uint8_t enc_tag[16];
     size_t enc_tag_size = 0;
+    bool enc_batch = false;
     EVP_CIPHER_CTX *dec_ctx = NULL;
     uint8_t dec_tag[16];
     size_t dec_tag_size = 0;
+    uint32_t dec_batch_size = 0;
+    size_t dec_batch_size_size = 0;
+    std::vector<osd_op_t*> unverified_ops;
 #endif
 
     // Read state
@@ -377,6 +381,7 @@ protected:
     bool allocate_reply_buffers(osd_client_t *cl, osd_op_t *op);
     bool op_read_from(osd_client_t *cl, msgr_op_reader_t & rdr);
     bool handle_finished_op(osd_client_t *cl);
+    void execute_verified_op(osd_client_t *cl, osd_op_t *op);
     void handle_immediate_ops();
 
     void op_encrypted_copy_buf(osd_client_t *cl, uint8_t *enc_buf, size_t enc_len, uint8_t *plain, size_t plain_len, size_t & done_plain, size_t & done_enc);

src/client/msgr_rdma.cpp

@@ -590,7 +590,36 @@ void osd_messenger_t::try_send_rdma(osd_client_t *cl)
     while (!rc->send_out_full && copied > 0 && rc->cur_send < rc->max_send)
     {
         dst = (uint8_t*)rc->send_out.buf + rc->send_out_pos;
-        dst_len = (rc->send_out_pos < rc->send_out_size ? rc->send_out_size-rc->send_out_pos : rc->send_done_pos-rc->send_out_pos);
+        if (rc->send_out_pos >= rc->send_done_pos)
+        {
+            dst_len = rc->send_out_size-rc->send_out_pos;
+            if (dst_len < 4096)
+            {
+                // free end of the buffer is too small, skip
+                rc->send_out_pos = 0;
+                if (rc->send_out_pos >= rc->send_done_pos)
+                    rc->send_out_full = true;
+                if (!rc->send_sizes.size())
+                {
+                    rc->send_done_pos += dst_len;
+                    rc->send_out_full = false;
+                    if (rc->send_done_pos == rc->send_out_size)
+                        rc->send_done_pos = 0;
+                }
+                else
+                    rc->send_sizes.back() += dst_len;
+                continue;
+            }
+        }
+        else
+        {
+            dst_len = rc->send_done_pos-rc->send_out_pos;
+            if (dst_len < 4096)
+            {
+                // too small buffer, stop
+                break;
+            }
+        }
         if (dst_len > rc->max_msg)
             dst_len = rc->max_msg;
         copied = copy_ops_to(cl, dst, dst_len);

src/client/msgr_receive.cpp

@@ -287,14 +287,6 @@ public:
     void reset()
     {
         from = cl->read_op_pos;
-        uint8_t iv[12] = { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1 };
-        int r = EVP_DecryptInit_ex(cl->dec_ctx, NULL, NULL, (uint8_t*)msgr->test_osd_aes_key.data(), iv);
-        if (r != 1)
-        {
-            fprintf(stderr, "DecryptInit error: ");
-            ERR_print_errors_fp(stderr);
-            abort();
-        }
     }
 
     bool read(uint8_t *dst, size_t dst_len, int flags) override
@@ -340,6 +332,31 @@ public:
         {
             // Here, dst == NULL is not allowed
             assert(dst != NULL);
+            if (cl->dec_batch_size_size < 4)
+            {
+                size_t n = 4-cl->dec_batch_size_size;
+                if (n > bufsize-done)
+                    n = bufsize-done;
+                memcpy(&cl->dec_batch_size, curbuf+done, n);
+                cl->dec_batch_size_size += n;
+                done += n;
+                if (cl->dec_batch_size_size < 4)
+                    return false;
+                if (!cl->dec_batch_size)
+                {
+                    fprintf(stderr, "Client %ju - empty batch received, disconnecting\n", cl->client_id);
+                    cl->io_error = true;
+                    return false;
+                }
+                uint8_t iv[12] = { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1 };
+                int r = EVP_DecryptInit_ex(cl->dec_ctx, NULL, NULL, (uint8_t*)msgr->test_osd_aes_key.data(), iv);
+                if (r != 1)
+                {
+                    fprintf(stderr, "DecryptInit error: ");
+                    ERR_print_errors_fp(stderr);
+                    abort();
+                }
+            }
             size_t n = dst_len-from;
             if (n > bufsize-done)
                 n = bufsize-done;
@@ -369,6 +386,12 @@ public:
 
     bool finish() override
     {
+        if (cl->dec_batch_size > 1)
+        {
+            // No tag yet
+            cl->dec_batch_size--;
+            return true;
+        }
         if (cl->dec_tag_size+bufsize-done < 16)
         {
             // Buffer part of the tag
@@ -402,6 +425,8 @@ public:
         }
         cl->dec_tag_size = 0;
         assert(len == 0);
+        cl->dec_batch_size = 0;
+        cl->dec_batch_size_size = 0;
         return true;
     }
 
@@ -651,6 +676,8 @@ out_wakeup:
             {
                 goto out_wakeup;
             }
+            execute_verified_op(cl, cl->read_op);
+            cl->read_op = NULL;
         }
     }
     cl->read_msg.msg_iovlen = 0;
@@ -734,6 +761,28 @@ bool osd_messenger_t::handle_buffer_with(osd_client_t *cl, uint8_t *curbuf, size
             }
             break;
         }
+        if constexpr (std::is_same_v<T, gcm_op_reader_t>)
+        {
+            if (cl->dec_batch_size_size)
+            {
+                // Operation is not verified yet
+                cl->unverified_ops.push_back(cl->read_op);
+            }
+            else
+            {
+                for (auto & op: cl->unverified_ops)
+                {
+                    execute_verified_op(cl, op);
+                }
+                cl->unverified_ops.clear();
+                execute_verified_op(cl, cl->read_op);
+            }
+        }
+        else
+        {
+            execute_verified_op(cl, cl->read_op);
+        }
+        cl->read_op = NULL;
     }
     assert(rdr.get_done() == bufsize);
     return true;
@@ -1039,12 +1088,7 @@ bool osd_messenger_t::handle_finished_op(osd_client_t *cl)
             return false;
         }
     }
-    if (op->op_type == OSD_OP_IN)
-    {
-        // Operation is ready
-        cl->received_ops.push_back(op);
-    }
-    else
+    if (op->op_type == OSD_OP_OUT)
     {
         // Inline decryption
         if (cl->read_op_inline_decrypt_pos != (size_t)-1)
@@ -1052,6 +1096,20 @@ bool osd_messenger_t::handle_finished_op(osd_client_t *cl)
             op_decrypt_inline(cl);
             cl->read_op_inline_decrypt_pos = (size_t)-1;
         }
+    }
+    op_decrypt_free(cl);
+    return true;
+}
+
+void osd_messenger_t::execute_verified_op(osd_client_t *cl, osd_op_t *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);
@@ -1066,8 +1124,5 @@ bool osd_messenger_t::handle_finished_op(osd_client_t *cl)
             (tv_end.tv_nsec - op->tv_begin.tv_nsec)/1000
         );
     }
-    op_decrypt_free(cl);
     set_immediate_ops.push_back(op);
-    cl->read_op = NULL;
-    return true;
 }

src/client/msgr_send.cpp

@@ -236,24 +236,17 @@ class gcm_op_writer_t: public msgr_op_writer_t
     uint8_t *curbuf;
     size_t bufsize;
     size_t done;
+    uint32_t *batch_size_ptr;
 
 public:
     gcm_op_writer_t(osd_messenger_t* msgr, osd_client_t* cl, uint8_t *curbuf, size_t bufsize):
-        msgr(msgr), cl(cl), from(cl->write_op_pos), curbuf(curbuf), bufsize(bufsize), done(0)
+        msgr(msgr), cl(cl), from(cl->write_op_pos), curbuf(curbuf), bufsize(bufsize), done(0), batch_size_ptr(NULL)
     {
     }
 
     void reset()
     {
         from = cl->write_op_pos;
-        uint8_t iv[12] = { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1 };
-        int r = EVP_EncryptInit_ex(cl->enc_ctx, NULL, NULL, (uint8_t*)msgr->test_osd_aes_key.data(), iv);
-        if (r != 1)
-        {
-            fprintf(stderr, "EncryptInit error: ");
-            ERR_print_errors_fp(stderr);
-            abort();
-        }
     }
 
     bool write(uint8_t *src, size_t src_len, int flags) override
@@ -286,6 +279,32 @@ public:
         }
         else
         {
+            if (!cl->write_op_pos)
+            {
+                if (batch_size_ptr)
+                {
+                    if (bufsize-done < 1)
+                        return false;
+                    (*batch_size_ptr)++;
+                }
+                else
+                {
+                    if (bufsize-done < 5)
+                        return false;
+                    batch_size_ptr = (uint32_t*)(curbuf+done);
+                    *batch_size_ptr = 1; // FIXME like header, but now for tests
+                    done += 4;
+                    cl->enc_batch = true;
+                    uint8_t iv[12] = { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1 };
+                    int r = EVP_EncryptInit_ex(cl->enc_ctx, NULL, NULL, (uint8_t*)msgr->test_osd_aes_key.data(), iv);
+                    if (r != 1)
+                    {
+                        fprintf(stderr, "EncryptInit error: ");
+                        ERR_print_errors_fp(stderr);
+                        abort();
+                    }
+                }
+            }
             size_t n = src_len-from;
             if (n > bufsize-done)
                 n = bufsize-done;
@@ -328,28 +347,22 @@ public:
 
     bool finish() override
     {
+        if (bufsize-done >= OSD_PACKET_SIZE+16 && batch_size_ptr && cl->write_ops.size())
+        {
+            // More operations may fit, so don't finish the batch yet
+            return true;
+        }
         // Tag is 16 bytes
-        if (done >= bufsize)
+        if (bufsize-done < 16)
+        {
+            // No space for the tag
             return false;
-        if (bufsize-done < 16 || cl->enc_tag_size)
-        {
-            // No space for the full tag, but msgr_rdma expects us to always fill the whole buffer
-            if (!cl->enc_tag_size)
-            {
-                write_tag_to(cl, cl->enc_tag);
-                cl->enc_tag_size = 16;
         }
-            size_t n = bufsize-done;
-            if (n > cl->enc_tag_size)
-                n = cl->enc_tag_size;
-            memcpy(curbuf+done, cl->enc_tag+16-cl->enc_tag_size, n);
-            done += n;
-            cl->enc_tag_size -= n;
-            return !cl->enc_tag_size;
-        }
-        // The whole tag fits at once
         write_tag_to(cl, curbuf+done);
         done += 16;
+        // Batch is completed
+        cl->enc_batch = false;
+        batch_size_ptr = NULL;
         return true;
     }
 
@@ -368,6 +381,10 @@ class get_op_writer_t: public msgr_op_writer_t
     size_t enc_size;
     size_t done_enc;
 
+    bool have_batch;
+    size_t batch_size_offset;
+    size_t batch_bytes;
+
     void ssl_extend_buf(size_t more = 0)
     {
         size_t min_cap = cl->ssl_out_buf_size*2;
@@ -407,7 +424,8 @@ class get_op_writer_t: public msgr_op_writer_t
 
 public:
     get_op_writer_t(osd_messenger_t* msgr, osd_client_t* cl):
-        msgr(msgr), cl(cl), from(cl->write_op_pos), enc_size(0), done_enc(0)
+        msgr(msgr), cl(cl), from(cl->write_op_pos), enc_size(0), done_enc(0),
+        have_batch(false), batch_size_offset(0), batch_bytes(0)
     {
     }
 
@@ -416,17 +434,6 @@ public:
         from = cl->write_op_pos;
         enc_size = 0;
         done_enc = 0;
-        if (cl->enc_ctx)
-        {
-            uint8_t iv[12] = { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1 };
-            int r = EVP_EncryptInit_ex(cl->enc_ctx, NULL, NULL, (uint8_t*)msgr->test_osd_aes_key.data(), iv);
-            if (r != 1)
-            {
-                fprintf(stderr, "EncryptInit error: ");
-                ERR_print_errors_fp(stderr);
-                abort();
-            }
-        }
     }
 
     void flush_ssl()
@@ -443,6 +450,22 @@ public:
         copy_ssl();
     }
 
+    void send_out_buf(size_t n)
+    {
+        if (cl->send_list.size() > 0)
+        {
+            iovec& last = cl->send_list.back();
+            if (last.iov_base+last.iov_len == cl->ssl_out_buf+cl->ssl_out_buf_size)
+            {
+                last.iov_len += n;
+                cl->ssl_out_buf_size += n;
+                return;
+            }
+        }
+        cl->send_list.push_back((iovec){ .iov_base = cl->ssl_out_buf+cl->ssl_out_buf_size, .iov_len = n });
+        cl->ssl_out_buf_size += n;
+    }
+
     bool write(uint8_t *src, size_t src_len, int flags) override
     {
         if (from >= src_len)
@@ -479,6 +502,30 @@ public:
             else if (cl->enc_ctx)
             {
                 // Encrypt data to client's temporary output buffer (all at once)
+                if (!cl->write_op_pos)
+                {
+                    if (have_batch)
+                    {
+                        (*(uint32_t*)(cl->ssl_out_buf+batch_size_offset))++;
+                    }
+                    else
+                    {
+                        ssl_extend_buf(4);
+                        have_batch = true;
+                        batch_size_offset = cl->ssl_out_buf_size;
+                        (*(uint32_t*)(cl->ssl_out_buf+batch_size_offset)) = 1;
+                        send_out_buf(4);
+                        cl->enc_batch = true;
+                        uint8_t iv[12] = { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1 };
+                        int r = EVP_EncryptInit_ex(cl->enc_ctx, NULL, NULL, (uint8_t*)msgr->test_osd_aes_key.data(), iv);
+                        if (r != 1)
+                        {
+                            fprintf(stderr, "EncryptInit error: ");
+                            ERR_print_errors_fp(stderr);
+                            abort();
+                        }
+                    }
+                }
                 size_t n = src_len-from;
                 ssl_extend_buf(n);
                 int actual_out;
@@ -491,8 +538,8 @@ public:
                 assert(actual_out == n);
                 if (cl->write_csum_state && !(flags & WR_NO_CSUM))
                     XXH3_64bits_update(cl->write_csum_state, src+from, n);
-                cl->send_list.push_back((iovec){ .iov_base = cl->ssl_out_buf+cl->ssl_out_buf_size, .iov_len = n });
-                cl->ssl_out_buf_size += n;
+                send_out_buf(n);
+                batch_bytes += n;
                 cl->write_op_pos += n;
                 from += n;
                 if (from < src_len)
@@ -518,8 +565,10 @@ public:
                 cl->send_list.push_back((iovec){ .iov_base = cl->write_op->enc_buf, .iov_len = enc_size });
             }
             assert(enc_size > 0);
+            size_t old_from = from;
             msgr->op_encrypted_copy_buf(cl, cl->write_op->enc_buf, enc_size, src, src_len, from, done_enc);
             assert(from == src_len);
+            batch_bytes += src_len-old_from;
         }
         else
         {
@@ -527,6 +576,7 @@ public:
                 XXH3_64bits_update(cl->write_csum_state, src+from, src_len-from);
             cl->send_list.push_back((iovec){ src+from, src_len-from });
             cl->write_op_pos += src_len-from;
+            batch_bytes += src_len-from;
         }
         from = 0;
         return true;
@@ -544,12 +594,18 @@ public:
         {
             if (cl->send_list.size() >= IOV_MAX)
                 return false;
+            if (batch_bytes < 131072 && have_batch && cl->write_ops.size())
+            {
+                // More operations may fit, so don't finish the batch yet
+                return true;
+            }
             // Tag is 16 bytes
             ssl_extend_buf(16);
             gcm_op_writer_t::write_tag_to(cl, cl->ssl_out_buf+cl->ssl_out_buf_size);
-            // FIXME coalesce entries in ssl_out_buf
-            cl->send_list.push_back((iovec){ .iov_base = cl->ssl_out_buf+cl->ssl_out_buf_size, .iov_len = 16 });
-            cl->ssl_out_buf_size += 16;
+            send_out_buf(16);
+            cl->enc_batch = false;
+            have_batch = false;
+            batch_bytes = 0;
         }
         return true;
     }

src/client/msgr_stop.cpp

@@ -210,6 +210,12 @@ osd_client_t::~osd_client_t()
         read_op->cancel();
         read_op = NULL;
     }
+    while (unverified_ops.size())
+    {
+        auto op = unverified_ops.back();
+        unverified_ops.pop_back();
+        op->cancel();
+    }
     // Cancel outbound ops
     cancel_ops();
     for (osd_op_t *op: send_free_ops)