Implement protocol-level checksums (xxhash3)

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

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/xxhash.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/xxhash.c ../../json11/json11.cpp
 )
 target_link_libraries(test_cluster_client ${OPENSSL_LIBRARIES})
 target_compile_definitions(test_cluster_client PUBLIC -D__MOCK__)

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 <osd_num>
         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())
         {

src/client/messenger.h

@@ -12,6 +12,7 @@
 #include <deque>
 #include <vector>
 
+#include "xxhash.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<osd_op_t*> received_ops;
@@ -110,6 +116,7 @@ struct osd_client_t
     std::deque<osd_op_t*> send_free_ops;
     std::vector<osd_op_t*> 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<addr_mask_t> osd_cluster_network_masks;
     std::vector<std::string> all_osd_networks;
     std::vector<addr_mask_t> 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);
+    bool 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 op_get_read_buffers(osd_client_t *cl, std::vector<iovec> & 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);

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;

src/client/msgr_encrypt.h

@@ -3,6 +3,7 @@
 
 #include <stdint.h>
 
+#include "xxhash.h"
 // WITH_OPENSSL is left to possibly support other crypto libraries
 #ifdef WITH_OPENSSL
 #include <openssl/conf.h>

src/client/msgr_op.h

@@ -186,6 +186,7 @@ struct __attribute__((visibility("default"))) osd_op_t
     void *rmw_buf = NULL;
     std::shared_ptr<osd_op_enc_t> enc;
     uint8_t *enc_buf = NULL;
+    uint64_t csum = 0; // network layer checksum
     osd_primary_op_data_t* op_data = NULL;
     std::function<void(osd_op_t*)> callback;
 
@@ -195,4 +196,5 @@ struct __attribute__((visibility("default"))) osd_op_t
     void cancel();
 
     bool is_recovery_related();
+    uint64_t calc_data_checksum();
 };

src/client/msgr_rdma.cpp

@@ -744,6 +744,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);
             }
         }

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))
-            {
+                goto out_wakeup;
-                if (set_immediate_ops.size())
-                    ringloop->wakeup();
-                return;
-            }
         }
         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())
+    goto out_wakeup;
-        ringloop->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,7 +226,10 @@ 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;
 }
@@ -320,6 +333,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 +391,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
@@ -403,36 +429,40 @@ size_t osd_messenger_t::op_copy_from(osd_client_t *cl, uint8_t *src, size_t src_
             from -= dst_len;
         return true;
     };
+    if (cl->read_csum_state && !from)
+    {
+        XXH3_64bits_update(cl->read_csum_state, (op->op_type == OSD_OP_IN ? op->req.buf : op->reply.buf), OSD_PACKET_SIZE);
+    }
     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;
+                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);
+    if (cl->proto_csum_status == (MSGR_PEER_CSUM_IN|MSGR_PEER_CSUM_OUT))
-    return done;
+    {
+        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<iovec> & lst)
+void 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;
@@ -520,30 +554,30 @@ size_t osd_messenger_t::op_get_read_buffers(osd_client_t *cl, std::vector<iovec>
             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<iovec>
             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<iovec>
             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<iovec>
                         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;
 }

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)
         {
@@ -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 = false)
     {
         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;
@@ -386,6 +399,13 @@ size_t osd_messenger_t::op_copy_to(osd_client_t *cl, uint8_t *dst, size_t dst_le
             }
         }
     }
+    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<iovec> & 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 = false)
     {
         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<iovec>
             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;
@@ -439,6 +467,13 @@ void osd_messenger_t::op_get_write_buffers(osd_client_t *cl, std::vector<iovec>
             }
         }
     }
+    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;
 }

src/client/msgr_stop.cpp

@@ -5,6 +5,7 @@
 #include <assert.h>
 
 #include "messenger.h"
+#include "xxhash.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;
+    }
 }

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())

src/util/xxhash.h

@@ -3,6 +3,9 @@
 
 #pragma once
 
+#include <stdint.h>
+#include <stddef.h>
+
 #if defined(__cplusplus) && !defined(XXH_NO_EXTERNC_GUARD)
 extern "C" {
 #endif