diff --git a/src/blockstore/CMakeLists.txt b/src/blockstore/CMakeLists.txt index ff70a7e2..3113cb32 100644 --- a/src/blockstore/CMakeLists.txt +++ b/src/blockstore/CMakeLists.txt @@ -4,7 +4,7 @@ project(vitastor) # libvitastor_blk.a add_library(vitastor_blk STATIC - ../util/allocator.cpp ../util/crc32c.c ../util/xxhash.c ../util/ringloop.cpp + ../util/allocator.cpp ../util/crc32c.c ../util/ringloop.cpp multilist.cpp blockstore_heap.cpp blockstore_disk.cpp blockstore.cpp blockstore_impl.cpp blockstore_init.cpp blockstore_open.cpp blockstore_flush.cpp blockstore_read.cpp blockstore_stable.cpp blockstore_sync.cpp blockstore_write.cpp diff --git a/src/client/CMakeLists.txt b/src/client/CMakeLists.txt index c0378e2a..15f0080f 100644 --- a/src/client/CMakeLists.txt +++ b/src/client/CMakeLists.txt @@ -12,7 +12,7 @@ if (RDMACM_LIBRARIES) set(MSGR_RDMACM "msgr_rdmacm.cpp") endif (RDMACM_LIBRARIES) add_library(vitastor_common STATIC - ../util/epoll_manager.cpp etcd_state_client.cpp messenger.cpp msgr_iothread.cpp ../util/addr_util.cpp + ../util/epoll_manager.cpp etcd_state_client.cpp messenger.cpp msgr_iothread.cpp ../util/addr_util.cpp ../util/xxh_x86dispatch.c msgr_encrypt.cpp msgr_stop.cpp msgr_op.cpp msgr_send.cpp msgr_receive.cpp ../util/ringloop.cpp ../../json11/json11.cpp http_client.cpp osd_ops.cpp pg_states.cpp ../util/timerfd_manager.cpp ../util/str_util.cpp ../util/json_util.cpp ${MSGR_RDMA} ${MSGR_RDMACM} ) @@ -98,7 +98,7 @@ add_executable(test_cluster_client EXCLUDE_FROM_ALL ../test/test_cluster_client.cpp pg_states.cpp osd_ops.cpp cluster_client.cpp cluster_client_list.cpp cluster_client_wb.cpp cluster_client_icache.cpp msgr_op.cpp ../test/mock/messenger.cpp msgr_stop.cpp msgr_encrypt.cpp - etcd_state_client.cpp ../util/timerfd_manager.cpp ../util/addr_util.cpp ../util/str_util.cpp ../util/json_util.cpp ../../json11/json11.cpp + etcd_state_client.cpp ../util/timerfd_manager.cpp ../util/addr_util.cpp ../util/str_util.cpp ../util/json_util.cpp ../util/xxh_x86dispatch.c ../../json11/json11.cpp ) target_link_libraries(test_cluster_client ${LIBURING_LIBRARIES} ${OPENSSL_LIBRARIES}) target_compile_definitions(test_cluster_client PUBLIC -D__MOCK__) diff --git a/src/client/messenger.cpp b/src/client/messenger.cpp index ac204ca8..24d8482c 100644 --- a/src/client/messenger.cpp +++ b/src/client/messenger.cpp @@ -229,6 +229,7 @@ void osd_messenger_t::parse_config(const json11::Json & config) this->max_aes_xts_pool_size = config["max_aes_xts_pool_size"].uint64_value(); if (!this->max_aes_xts_pool_size) this->max_aes_xts_pool_size = 256; + this->use_proto_checksums = config["use_proto_checksums"].is_null() || config["use_proto_checksums"].bool_value(); if (!osd_num) this->iothread_count = (uint32_t)config["client_iothread_count"].uint64_value(); else @@ -548,7 +549,12 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl) // Inform that we're OSD payload["osd_num"] = osd_num; } - payload["features"] = json11::Json::object{ { "check_sequencing", true } }; + auto features = json11::Json::object{ { "check_sequencing", true } }; + if (use_proto_checksums) + { + features["proto_checksums"] = true; + } + payload["features"] = features; #ifdef WITH_RDMA if (!use_rdmacm && rdma_contexts.size()) { @@ -623,6 +629,10 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl) delete op; return; } + if (use_proto_checksums && config["features"]["proto_checksums"].bool_value()) + { + cl->proto_csum_status = MSGR_PEER_CSUM_IN|MSGR_PEER_CSUM_OUT; + } #ifdef WITH_RDMA if (!use_rdmacm && cl->rdma_conn && config["rdma_address"].is_string()) { diff --git a/src/client/messenger.h b/src/client/messenger.h index 9b4415dc..e4fc40d9 100644 --- a/src/client/messenger.h +++ b/src/client/messenger.h @@ -12,6 +12,7 @@ #include #include +#include "../util/xxh_x86dispatch.h" #include "../util/robin_hood.h" #include "malloc_or_die.h" #include "json11/json11.hpp" @@ -31,6 +32,9 @@ #define PEER_RDMA 4 #define PEER_STOPPED 5 +#define MSGR_PEER_CSUM_IN 1 +#define MSGR_PEER_CSUM_OUT 2 + #define VITASTOR_CONFIG_PATH "/etc/vitastor/vitastor.conf" #define DEFAULT_MIN_ZEROCOPY_SEND_SIZE 32*1024 @@ -88,6 +92,8 @@ struct osd_client_t op_aes_xts_decrypt_t *decrypt_ctx = NULL; size_t read_op_inline_decrypt_pos = 0; size_t read_op_inline_decrypt_in = 0; + int proto_csum_status = 0; + XXH3_state_t* read_csum_state = NULL; // Incoming operations std::vector received_ops; @@ -110,6 +116,7 @@ struct osd_client_t std::deque send_free_ops; std::vector zc_free_list; op_aes_xts_encrypt_t *encrypt_ctx = NULL; + XXH3_state_t* write_csum_state = NULL; ~osd_client_t(); void cancel_ops(); @@ -209,6 +216,7 @@ public: std::vector osd_cluster_network_masks; std::vector all_osd_networks; std::vector all_osd_network_masks; + bool use_proto_checksums = true; // op statistics osd_op_stats_t stats, recovery_stats; @@ -268,10 +276,10 @@ protected: bool handle_hdr(osd_client_t *cl); bool allocate_op_buffers(osd_client_t *cl); bool allocate_reply_buffers(osd_client_t *cl, 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 & lst); + bool op_copy_from(osd_client_t *cl, uint8_t *src, size_t src_len, size_t & done); + void op_get_read_buffers(osd_client_t *cl, std::vector & lst); void op_alloc_temp_buffers(osd_op_t *op, int i); - void handle_finished_op(osd_client_t *cl); + bool handle_finished_op(osd_client_t *cl); void handle_immediate_ops(); bool op_encrypted_copy_data_to(osd_client_t* cl, uint8_t *buf, size_t len, size_t from, size_t & done); diff --git a/src/client/msgr_encrypt.cpp b/src/client/msgr_encrypt.cpp index b71e9253..f6a02677 100644 --- a/src/client/msgr_encrypt.cpp +++ b/src/client/msgr_encrypt.cpp @@ -356,6 +356,8 @@ bool osd_messenger_t::op_encrypted_copy_data_to(osd_client_t* cl, uint8_t *enc_b size_t done_in = 0; size_t done_out = 0; cl->encrypt_ctx->update(plain+from, plain_len-from, enc_buf+done, enc_len-done, done_in, done_out); + if (cl->write_csum_state && done_out > 0) + XXH3_64bits_update(cl->write_csum_state, enc_buf+done, done_out); done += done_out; op_pos += done_in; from += done_in; @@ -390,6 +392,8 @@ bool osd_messenger_t::op_decrypted_copy_data_from(osd_client_t* cl, uint8_t *enc size_t done_out = 0; // plain == NULL means skip output cl->decrypt_ctx->update(enc_buf+done, enc_len-done, plain ? plain+from : NULL, plain_len-from, done_in, done_out); + if (cl->read_csum_state && done_in > 0) + XXH3_64bits_update(cl->read_csum_state, enc_buf+done, done_in); done += done_in; cl->read_op_pos += done_out; cl->read_op_inline_decrypt_in += done_in; diff --git a/src/client/msgr_encrypt.h b/src/client/msgr_encrypt.h index 41bf9fbd..8ea7c9bc 100644 --- a/src/client/msgr_encrypt.h +++ b/src/client/msgr_encrypt.h @@ -3,6 +3,7 @@ #include +#include "../util/xxh_x86dispatch.h" // WITH_OPENSSL is left to possibly support other crypto libraries #ifdef WITH_OPENSSL #include diff --git a/src/client/msgr_op.h b/src/client/msgr_op.h index 6abe54b1..f922081b 100644 --- a/src/client/msgr_op.h +++ b/src/client/msgr_op.h @@ -187,6 +187,7 @@ struct __attribute__((visibility("default"))) osd_op_t void *rmw_buf = NULL; std::shared_ptr enc; uint8_t *enc_buf = NULL; + uint64_t csum = 0; // network layer checksum osd_primary_op_data_t* op_data = NULL; std::function callback; @@ -196,4 +197,5 @@ struct __attribute__((visibility("default"))) osd_op_t void cancel(); bool is_recovery_related(); + uint64_t calc_data_checksum(); }; diff --git a/src/client/msgr_rdma.cpp b/src/client/msgr_rdma.cpp index 25cbe2fb..03de0919 100644 --- a/src/client/msgr_rdma.cpp +++ b/src/client/msgr_rdma.cpp @@ -760,6 +760,11 @@ void osd_messenger_t::handle_rdma_events(msgr_rdma_context_t *rdma_context) cl->send_free_ops.pop_front(); } cl->send_free_ops.pop_front(); + if (cl->proto_csum_status == MSGR_PEER_CSUM_IN && !cl->write_op && !cl->write_ops.size()) + { + // Checksums negotiated, enable + cl->proto_csum_status = MSGR_PEER_CSUM_IN|MSGR_PEER_CSUM_OUT; + } try_send_rdma(cl); } } diff --git a/src/client/msgr_receive.cpp b/src/client/msgr_receive.cpp index 1545d3bf..205a1495 100644 --- a/src/client/msgr_receive.cpp +++ b/src/client/msgr_receive.cpp @@ -102,6 +102,9 @@ void 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); +out_wakeup: + if (set_immediate_ops.size()) + ringloop->wakeup(); return; } bool full_read = false; @@ -111,11 +114,7 @@ void osd_messenger_t::handle_read(int result, osd_client_t *cl) { full_read = result >= cl->read_iov.iov_len; if (!handle_read_buffer(cl, cl->in_buf, result)) - { - if (set_immediate_ops.size()) - ringloop->wakeup(); - return; - } + goto out_wakeup; } else { @@ -126,6 +125,11 @@ void osd_messenger_t::handle_read(int result, osd_client_t *cl) size_t i = 0; while (i < cl->recv_list.size() && result >= cl->recv_list[i].iov_len) { + if (cl->read_csum_state && cl->recv_list[i].iov_len > 0 && + i != cl->recv_list.size()-1) // skip the checksum itself + { + XXH3_64bits_update(cl->read_csum_state, cl->recv_list[i].iov_base, cl->recv_list[i].iov_len); + } result -= cl->recv_list[i].iov_len; i++; } @@ -141,7 +145,8 @@ void osd_messenger_t::handle_read(int result, osd_client_t *cl) cl->recv_list.erase(cl->recv_list.begin(), cl->recv_list.begin()+i); if (!cl->recv_list.size()) { - handle_finished_op(cl); + if (!handle_finished_op(cl)) + goto out_wakeup; } } } @@ -156,8 +161,7 @@ void osd_messenger_t::handle_read(int result, osd_client_t *cl) { read_ready_clients.push_back(cl->client_id); } - if (set_immediate_ops.size()) - ringloop->wakeup(); + goto out_wakeup; } void osd_messenger_t::handle_immediate_ops() @@ -200,6 +204,12 @@ bool osd_messenger_t::handle_read_buffer(osd_client_t *cl, uint8_t *curbuf, size cl->read_op_size = 0; cl->read_op_inline_decrypt_in = 0; cl->read_op_inline_decrypt_pos = (size_t)-1; + if (cl->proto_csum_status == (MSGR_PEER_CSUM_IN|MSGR_PEER_CSUM_OUT)) + { + if (!cl->read_csum_state) + cl->read_csum_state = XXH3_createState(); + XXH3_64bits_reset(cl->read_csum_state); + } } if (cl->read_op_pos < OSD_PACKET_SIZE) { @@ -217,13 +227,20 @@ bool osd_messenger_t::handle_read_buffer(osd_client_t *cl, uint8_t *curbuf, size return false; } } - op_copy_from(cl, curbuf, bufsize, done); + if (!op_copy_from(cl, curbuf, bufsize, done)) + { + return false; + } } return true; } bool osd_messenger_t::handle_hdr(osd_client_t *cl) { + if (cl->read_csum_state) + { + XXH3_64bits_update(cl->read_csum_state, cl->read_op->req.buf, OSD_PACKET_SIZE); + } if (cl->read_op->req.hdr.magic == SECONDARY_OSD_REPLY_MAGIC) { auto req_it = cl->sent_ops.find(cl->read_op->req.hdr.id); @@ -321,6 +338,10 @@ bool osd_messenger_t::allocate_op_buffers(osd_client_t *cl) } cl->read_op_size = cur_op->req.show_conf.json_len; } + if (cl->proto_csum_status == (MSGR_PEER_CSUM_IN|MSGR_PEER_CSUM_OUT)) + { + cl->read_op_size += 8; + } return true; } @@ -375,20 +396,29 @@ bool osd_messenger_t::allocate_reply_buffers(osd_client_t *cl, osd_op_t *op) free(op->buf); op->buf = malloc_or_die(op->reply.describe.result_bytes); } + if (cl->proto_csum_status == (MSGR_PEER_CSUM_IN|MSGR_PEER_CSUM_OUT)) + { + cl->read_op_size += 8; + } return true; } -size_t osd_messenger_t::op_copy_from(osd_client_t *cl, uint8_t *src, size_t src_len, size_t & done) +bool osd_messenger_t::op_copy_from(osd_client_t *cl, uint8_t *src, size_t src_len, size_t & done) { osd_op_t *op = cl->read_op; size_t from = cl->read_op_pos-OSD_PACKET_SIZE; - auto op_read_buf = [&](uint8_t *dst, size_t dst_len) + auto op_read_buf = [&](uint8_t *dst, size_t dst_len, bool skip_csum = false) { if (from < dst_len) { size_t n = dst_len-from; if (n > src_len-done) n = src_len-done; + if (cl->read_csum_state && !skip_csum) + { + // it may be skipped if !dst but checksum is still calculated + XXH3_64bits_update(cl->read_csum_state, src+done, n); + } if (dst) memcpy(dst+from, src+done, n); else @@ -410,30 +440,30 @@ size_t osd_messenger_t::op_copy_from(osd_client_t *cl, uint8_t *src, size_t src_ 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; + return true; if (!op_read_buf((uint8_t*)op->buf, op->req.sec_rw.len)) - return done; + return true; } 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; + return true; } 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; + return true; } else if (op->req.hdr.opcode == OSD_OP_WRITE) { if (!op_read_buf((uint8_t*)op->buf, op->req.rw.len)) - return done; + return true; } 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; + return true; } } else @@ -443,13 +473,13 @@ size_t osd_messenger_t::op_copy_from(osd_client_t *cl, uint8_t *src, size_t src_ if (op->reply.sec_rw.attr_len > 0) { if (!op_read_buf((uint8_t*)op->bitmap, op->reply.sec_rw.attr_len)) - return done; + return true; } 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; + return true; } } else if (op->reply.hdr.opcode == OSD_OP_READ) @@ -457,45 +487,49 @@ size_t osd_messenger_t::op_copy_from(osd_client_t *cl, uint8_t *src, size_t src_ if (op->reply.rw.bitmap_len > 0) { if (!op_read_buf((uint8_t*)op->bitmap, op->reply.rw.bitmap_len)) - return done; + return true; } if (op->reply.hdr.retval > 0) { if (op->enc) { if (!op_decrypted_copy_data_from(cl, src, src_len, from, done)) - return done; + return true; } else { 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; + return true; } } } 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; + return true; } 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; + return true; } 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 true; } } - handle_finished_op(cl); - return done; + if (cl->proto_csum_status == (MSGR_PEER_CSUM_IN|MSGR_PEER_CSUM_OUT)) + { + if (!op_read_buf((uint8_t*)&op->csum, 8, true)) + return true; + } + return handle_finished_op(cl); } -size_t osd_messenger_t::op_get_read_buffers(osd_client_t *cl, std::vector & lst) +void osd_messenger_t::op_get_read_buffers(osd_client_t *cl, std::vector & lst) { osd_op_t *op = cl->read_op; size_t from = cl->read_op_pos-OSD_PACKET_SIZE; @@ -521,30 +555,30 @@ size_t osd_messenger_t::op_get_read_buffers(osd_client_t *cl, std::vector 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; + return; if (!op_read_buf((uint8_t*)op->buf, op->req.sec_rw.len)) - return done; + return; } 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; + return; } 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; + return; } else if (op->req.hdr.opcode == OSD_OP_WRITE) { if (!op_read_buf((uint8_t*)op->buf, op->req.rw.len)) - return done; + return; } 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; + return; } } else @@ -554,13 +588,13 @@ size_t osd_messenger_t::op_get_read_buffers(osd_client_t *cl, std::vector if (op->reply.sec_rw.attr_len > 0) { if (!op_read_buf((uint8_t*)op->bitmap, op->reply.sec_rw.attr_len)) - return done; + return; } 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; + return; } } else if (op->reply.hdr.opcode == OSD_OP_READ) @@ -568,7 +602,7 @@ size_t osd_messenger_t::op_get_read_buffers(osd_client_t *cl, std::vector if (op->reply.rw.bitmap_len > 0) { if (!op_read_buf((uint8_t*)op->bitmap, op->reply.rw.bitmap_len)) - return done; + return; } if (op->reply.hdr.retval > 0) { @@ -588,28 +622,32 @@ size_t osd_messenger_t::op_get_read_buffers(osd_client_t *cl, std::vector op_alloc_temp_buffers(op, i); } if (!op_read_buf((uint8_t*)op->iov.buf[i].iov_base, op->iov.buf[i].iov_len)) - return done; + return; } } } 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; + return; } 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; + return; } 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; } } - return done; + if (cl->proto_csum_status == (MSGR_PEER_CSUM_IN|MSGR_PEER_CSUM_OUT)) + { + if (!op_read_buf((uint8_t*)&op->csum, 8)) + return; + } } void osd_messenger_t::op_alloc_temp_buffers(osd_op_t *op, int i) @@ -636,9 +674,20 @@ void osd_messenger_t::op_alloc_temp_buffers(osd_op_t *op, int i) } } -void osd_messenger_t::handle_finished_op(osd_client_t *cl) +bool osd_messenger_t::handle_finished_op(osd_client_t *cl) { osd_op_t *op = cl->read_op; + if (cl->proto_csum_status == (MSGR_PEER_CSUM_IN|MSGR_PEER_CSUM_OUT)) + { + uint64_t real_csum = XXH3_64bits_digest(cl->read_csum_state); + if (op->csum != real_csum) + { + fprintf(stderr, "Client %ju checksum mismatch for received data: expected %016jx, got %016jx, disconnecting client\n", + cl->client_id, op->csum, real_csum); + stop_client(cl->client_id); + return false; + } + } if (op->op_type == OSD_OP_IN) { // Operation is ready @@ -668,4 +717,5 @@ void osd_messenger_t::handle_finished_op(osd_client_t *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 ad6b89f0..8fb9a5e3 100644 --- a/src/client/msgr_send.cpp +++ b/src/client/msgr_send.cpp @@ -269,6 +269,11 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t cl->zc_free_list.push_back(NULL); // end marker cl->send_free_ops.clear(); cl->write_state = cl->write_op || cl->write_ops.size() ? CL_WRITE_READY : 0; + if (cl->proto_csum_status == MSGR_PEER_CSUM_IN && !cl->write_op && !cl->write_ops.size()) + { + // Checksums negotiated, enable + cl->proto_csum_status = MSGR_PEER_CSUM_IN|MSGR_PEER_CSUM_OUT; + } #ifdef WITH_RDMA if (cl->rdma_conn && !cl->write_op && !cl->write_ops.size() && cl->peer_state == PEER_RDMA_CONNECTING) { @@ -289,36 +294,36 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t } } -static inline bool op_write_headers(osd_op_t *op, std::function op_write_buf) +static inline bool op_write_headers(osd_op_t *op, std::function op_write_buf) { // Header - if (!op_write_buf((op->op_type == OSD_OP_IN ? op->reply.buf : op->req.buf), OSD_PACKET_SIZE)) + if (!op_write_buf((op->op_type == OSD_OP_IN ? op->reply.buf : op->req.buf), OSD_PACKET_SIZE, false)) 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)) + if (!op_write_buf((uint8_t*)op->bitmap, op->reply.sec_rw.attr_len, false)) 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)) + if (!op_write_buf((uint8_t*)op->bitmap, op->req.sec_rw.attr_len, false)) 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)) + if (!op_write_buf((uint8_t*)op->buf, (size_t)op->reply.hdr.retval, false)) 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)) + if (!op_write_buf((uint8_t*)op->buf, (size_t)op->req.sec_read_bmp.len, false)) return false; } } @@ -345,13 +350,15 @@ size_t osd_messenger_t::op_copy_to(osd_client_t *cl, uint8_t *dst, size_t dst_le { size_t done = 0; size_t from = cl->write_op_pos; - auto op_write_buf = [&](uint8_t *src, size_t src_len) + auto op_write_buf = [&](uint8_t *src, size_t src_len, bool skip_csum) { if (from < src_len) { size_t n = src_len-from; if (n > dst_len-done) n = dst_len-done; + if (cl->write_csum_state && !skip_csum) + XXH3_64bits_update(cl->write_csum_state, src+from, n); memcpy(dst+done, src+from, n); done += n; cl->write_op_pos += n; @@ -364,6 +371,12 @@ size_t osd_messenger_t::op_copy_to(osd_client_t *cl, uint8_t *dst, size_t dst_le from -= src_len; return true; }; + if (cl->proto_csum_status == (MSGR_PEER_CSUM_IN|MSGR_PEER_CSUM_OUT) && !from) + { + if (!cl->write_csum_state) + cl->write_csum_state = XXH3_createState(); + XXH3_64bits_reset(cl->write_csum_state); + } if (!op_write_headers(cl->write_op, op_write_buf)) { return done; @@ -382,11 +395,18 @@ size_t osd_messenger_t::op_copy_to(osd_client_t *cl, uint8_t *dst, size_t dst_le { 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)) + if (!op_write_buf((uint8_t*)cl->write_op->iov.buf[i].iov_base, cl->write_op->iov.buf[i].iov_len, false)) return done; } } } + if (cl->write_csum_state) + { + if (!from) + cl->write_op->csum = XXH3_64bits_digest(cl->write_csum_state); + if (!op_write_buf((uint8_t*)&cl->write_op->csum, 8, true)) + return done; + } cl->write_op = NULL; cl->write_op_pos = 0; return done; @@ -395,12 +415,14 @@ size_t osd_messenger_t::op_copy_to(osd_client_t *cl, uint8_t *dst, size_t dst_le void osd_messenger_t::op_get_write_buffers(osd_client_t *cl, std::vector & lst) { size_t from = cl->write_op_pos; - auto op_write_buf = [&](uint8_t *src, size_t src_len) + auto op_write_buf = [&](uint8_t *src, size_t src_len, bool skip_csum) { if (lst.size() >= IOV_MAX) return false; if (from < src_len) { + if (cl->write_csum_state && !skip_csum) + XXH3_64bits_update(cl->write_csum_state, src+from, src_len-from); lst.push_back((iovec){ .iov_base = src+from, .iov_len = src_len-from }); cl->write_op_pos += src_len-from; from = 0; @@ -409,6 +431,12 @@ void osd_messenger_t::op_get_write_buffers(osd_client_t *cl, std::vector from -= src_len; return true; }; + if (cl->proto_csum_status == (MSGR_PEER_CSUM_IN|MSGR_PEER_CSUM_OUT) && !from) + { + if (!cl->write_csum_state) + cl->write_csum_state = XXH3_createState(); + XXH3_64bits_reset(cl->write_csum_state); + } if (!op_write_headers(cl->write_op, op_write_buf)) { return; @@ -435,11 +463,18 @@ void osd_messenger_t::op_get_write_buffers(osd_client_t *cl, std::vector { 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)) + if (!op_write_buf((uint8_t*)cl->write_op->iov.buf[i].iov_base, cl->write_op->iov.buf[i].iov_len, false)) return; } } } + if (cl->write_csum_state) + { + if (!from) + cl->write_op->csum = XXH3_64bits_digest(cl->write_csum_state); + if (!op_write_buf((uint8_t*)&cl->write_op->csum, 8, true)) + return; + } cl->write_op = NULL; cl->write_op_pos = 0; } diff --git a/src/client/msgr_stop.cpp b/src/client/msgr_stop.cpp index 6af4a7cf..753f8c06 100644 --- a/src/client/msgr_stop.cpp +++ b/src/client/msgr_stop.cpp @@ -5,6 +5,7 @@ #include #include "messenger.h" +#include "../util/xxh_x86dispatch.h" void osd_client_t::cancel_ops() { @@ -209,4 +210,14 @@ osd_client_t::~osd_client_t() delete op; } } + if (read_csum_state) + { + XXH3_freeState(read_csum_state); + read_csum_state = NULL; + } + if (write_csum_state) + { + XXH3_freeState(write_csum_state); + write_csum_state = NULL; + } } diff --git a/src/osd/osd_secondary.cpp b/src/osd/osd_secondary.cpp index e91af4e4..3cb0f78d 100644 --- a/src/osd/osd_secondary.cpp +++ b/src/osd/osd_secondary.cpp @@ -344,6 +344,12 @@ void osd_t::exec_show_config(osd_op_t *cur_op) cl->check_sequencing = true; cl->read_op_id = cur_op->req.hdr.id + 1; } + auto features = json11::Json::object{ { "pg_locks", true } }; + if (req_json["features"]["proto_checksums"].bool_value() && msgr.use_proto_checksums) + { + cl->proto_csum_status = MSGR_PEER_CSUM_IN; + features["proto_checksums"] = true; + } // Expose sensitive configuration values so peers can check them json11::Json::object wire_config = json11::Json::object { { "osd_num", osd_num }, @@ -356,7 +362,7 @@ void osd_t::exec_show_config(osd_op_t *cur_op) { "immediate_commit", (immediate_commit == IMMEDIATE_ALL ? "all" : (immediate_commit == IMMEDIATE_SMALL ? "small" : "none")) }, { "lease_timeout", etcd_report_interval+(st_cli.max_etcd_attempts*(2*st_cli.etcd_quick_timeout)+999)/1000 }, - { "features", json11::Json::object{ { "pg_locks", true } } }, + { "features", features }, }; #ifdef WITH_RDMA if (msgr.is_rdma_enabled()) diff --git a/src/util/xxhash.h b/src/util/xxhash.h index d413cc85..66266a40 100644 --- a/src/util/xxhash.h +++ b/src/util/xxhash.h @@ -3,6 +3,9 @@ #pragma once +#include +#include + #if defined(__cplusplus) && !defined(XXH_NO_EXTERNC_GUARD) extern "C" { #endif