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 063ac373..3fc08889 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 ../util/addr_util.cpp + ../util/epoll_manager.cpp etcd_state_client.cpp messenger.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} ) @@ -101,7 +101,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 ${OPENSSL_LIBRARIES}) target_compile_definitions(test_cluster_client PUBLIC -D__MOCK__) diff --git a/src/client/messenger.cpp b/src/client/messenger.cpp index 537bb033..93267be5 100644 --- a/src/client/messenger.cpp +++ b/src/client/messenger.cpp @@ -340,6 +340,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 @@ -659,7 +660,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()) { @@ -734,6 +740,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 8999869f..97810b21 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(); @@ -245,6 +252,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; @@ -302,10 +310,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 26d8714f..d331f4f2 100644 --- a/src/client/msgr_op.h +++ b/src/client/msgr_op.h @@ -186,6 +186,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; @@ -195,4 +196,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 98fab127..3e1bc0b0 100644 --- a/src/client/msgr_rdma.cpp +++ b/src/client/msgr_rdma.cpp @@ -758,6 +758,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 ccc349d9..0ef8b11c 100644 --- a/src/client/msgr_receive.cpp +++ b/src/client/msgr_receive.cpp @@ -101,6 +101,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; @@ -110,11 +113,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 { @@ -125,6 +124,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++; } @@ -140,7 +144,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; } } } @@ -155,8 +160,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() @@ -199,6 +203,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) { @@ -216,13 +226,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); @@ -320,6 +337,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; } @@ -374,20 +395,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 @@ -409,30 +439,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 @@ -442,13 +472,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) @@ -456,45 +486,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; @@ -520,30 +554,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 @@ -553,13 +587,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) @@ -567,7 +601,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) { @@ -587,28 +621,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) @@ -635,9 +673,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 @@ -667,4 +716,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 a39fd5ed..945bfeb4 100644 --- a/src/client/msgr_send.cpp +++ b/src/client/msgr_send.cpp @@ -268,6 +268,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) { @@ -288,36 +293,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; } } @@ -344,13 +349,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; @@ -363,6 +370,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; @@ -381,11 +394,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; @@ -394,12 +414,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; @@ -408,6 +430,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; @@ -434,11 +462,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 79c4691d..bd7f6d07 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" #ifdef WITH_RDMA #include "msgr_rdma.h" #endif @@ -228,4 +229,14 @@ osd_client_t::~osd_client_t() } #endif #endif + 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 d553a128..937536cd 100644 --- a/src/osd/osd_secondary.cpp +++ b/src/osd/osd_secondary.cpp @@ -347,6 +347,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 }, @@ -359,7 +365,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