diff --git a/src/client/messenger.h b/src/client/messenger.h index b27af203..17598f62 100644 --- a/src/client/messenger.h +++ b/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 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); diff --git a/src/client/msgr_rdma.cpp b/src/client/msgr_rdma.cpp index 9831a1de..9354bf3a 100644 --- a/src/client/msgr_rdma.cpp +++ b/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); diff --git a/src/client/msgr_receive.cpp b/src/client/msgr_receive.cpp index 9236e20c..20d42cdc 100644 --- a/src/client/msgr_receive.cpp +++ b/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) + { + 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; } diff --git a/src/client/msgr_send.cpp b/src/client/msgr_send.cpp index c26b4d71..bd0b4404 100644 --- a/src/client/msgr_send.cpp +++ b/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 { - // Tag is 16 bytes - if (done >= bufsize) - return false; - if (bufsize-done < 16 || cl->enc_tag_size) + if (bufsize-done >= OSD_PACKET_SIZE+16 && batch_size_ptr && cl->write_ops.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; + // More operations may fit, so don't finish the batch yet + return true; + } + // Tag is 16 bytes + if (bufsize-done < 16) + { + // No space for the tag + return false; } - // 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; } diff --git a/src/client/msgr_stop.cpp b/src/client/msgr_stop.cpp index 39138575..54d061b4 100644 --- a/src/client/msgr_stop.cpp +++ b/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)