Check operation sequencing and stop clients when it breaks

src/client/cluster_client.cpp

@@ -1244,7 +1244,6 @@ int cluster_client_t::try_send(cluster_op_t *op, int i)
                 .req = { .rw = {
                     .header = {
                         .magic = SECONDARY_OSD_OP_MAGIC,
-                        .id = next_op_id(),
                         .opcode = op->opcode == OSD_OP_READ_BITMAP || op->opcode == OSD_OP_READ_CHAIN_BITMAP ? OSD_OP_READ : op->opcode,
                     },
                     .inode = op->cur_inode,
@@ -1353,7 +1352,6 @@ void cluster_client_t::send_sync(cluster_op_t *op, cluster_op_part_t *part)
         .req = {
             .hdr = {
                 .magic = SECONDARY_OSD_OP_MAGIC,
-                .id = next_op_id(),
                 .opcode = OSD_OP_SYNC,
             },
         },
@@ -1498,8 +1496,3 @@ void cluster_client_t::copy_part_bitmap(cluster_op_t *op, cluster_op_part_t *par
         part_len--;
     }
 }
-
-uint64_t cluster_client_t::next_op_id()
-{
-    return msgr.next_subop_id++;
-}

src/client/cluster_client.h

@@ -152,7 +152,6 @@ public:
 
     //inline uint32_t get_bs_bitmap_granularity() { return st_cli.global_bitmap_granularity; }
     //inline uint64_t get_bs_block_size() { return st_cli.global_block_size; }
-    uint64_t next_op_id();
 
 #ifndef __MOCK__
 protected:

src/client/cluster_client_list.cpp

@@ -342,7 +342,6 @@ void cluster_client_t::send_list(inode_list_osd_t *cur_list)
         .sec_list = {
             .header = {
                 .magic = SECONDARY_OSD_OP_MAGIC,
-                .id = next_op_id(),
                 .opcode = OSD_OP_SEC_LIST,
             },
             .list_pg = cur_list->pg->pg_num,

src/client/messenger.cpp

@@ -217,7 +217,6 @@ void osd_messenger_t::init()
                     op->req = (osd_any_op_t){
                         .hdr = {
                             .magic = SECONDARY_OSD_OP_MAGIC,
-                            .id = this->next_subop_id++,
                             .opcode = OSD_OP_PING,
                         },
                     };
@@ -629,11 +628,17 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl)
         .show_conf = {
             .header = {
                 .magic = SECONDARY_OSD_OP_MAGIC,
-                .id = this->next_subop_id++,
                 .opcode = OSD_OP_SHOW_CONFIG,
             },
         },
     };
+    json11::Json::object payload;
+    if (osd_num)
+    {
+        // Inform that we're OSD <osd_num>
+        payload["osd_num"] = osd_num;
+    }
+    payload["check_sequencing"] = true;
 #ifdef WITH_RDMA
     if (!use_rdmacm && rdma_contexts.size())
     {
@@ -649,19 +654,20 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl)
             cl->rdma_conn = msgr_rdma_connection_t::create(selected_ctx, rdma_max_send, rdma_max_recv, rdma_max_sge, rdma_max_msg);
             if (cl->rdma_conn)
             {
-                json11::Json payload = json11::Json::object {
-                    { "connect_rdma", cl->rdma_conn->addr.to_string() },
-                    { "rdma_max_msg", cl->rdma_conn->max_msg },
-                };
-                std::string payload_str = payload.dump();
-                op->req.show_conf.json_len = payload_str.size();
-                op->buf = malloc_or_die(payload_str.size());
-                op->iov.push_back(op->buf, payload_str.size());
-                memcpy(op->buf, payload_str.c_str(), payload_str.size());
+                payload["connect_rdma"] = cl->rdma_conn->addr.to_string();
+                payload["rdma_max_msg"] = cl->rdma_conn->max_msg;
             }
         }
     }
 #endif
+    if (payload.size())
+    {
+        std::string payload_str = json11::Json(payload).dump();
+        op->req.show_conf.json_len = payload_str.size();
+        op->buf = malloc_or_die(payload_str.size());
+        op->iov.push_back(op->buf, payload_str.size());
+        memcpy(op->buf, payload_str.c_str(), payload_str.size());
+    }
     op->callback = [this, cl](osd_op_t *op)
     {
         std::string json_err;

src/client/messenger.h

@@ -75,12 +75,15 @@ struct osd_client_t
     int read_remaining = 0;
     int read_state = 0;
     osd_op_buf_list_t recv_list;
+    uint64_t read_op_id = 1;
+    bool check_sequencing = false;
 
     // Incoming operations
     std::vector<osd_op_t*> received_ops;
 
     // Outbound operations
     std::map<uint64_t, osd_op_t*> sent_ops;
+    uint64_t send_op_id = 0;
 
     // PGs dirtied by this client's primary-writes
     std::set<pool_pg_num_t> dirty_pgs;
@@ -211,7 +214,6 @@ public:
     bool has_sendmsg_zc = false;
     // osd_num_t is only for logging and asserts
     osd_num_t osd_num;
-    uint64_t next_subop_id = 1;
     std::map<int, osd_client_t*> clients;
     std::map<osd_num_t, osd_wanted_peer_t> wanted_peers;
     std::map<uint64_t, int> osd_peer_fds;

src/client/msgr_receive.cpp

@@ -223,7 +223,19 @@ bool osd_messenger_t::handle_finished_read(osd_client_t *cl)
         if (cl->read_op->req.hdr.magic == SECONDARY_OSD_REPLY_MAGIC)
             return handle_reply_hdr(cl);
         else if (cl->read_op->req.hdr.magic == SECONDARY_OSD_OP_MAGIC)
+        {
+            if (cl->check_sequencing)
+            {
+                if (cl->read_op->req.hdr.id != cl->read_op_id)
+                {
+                    fprintf(stderr, "Warning: operation sequencing is broken on client %d, stopping client\n", cl->peer_fd);
+                    stop_client(cl->peer_fd);
+                    return false;
+                }
+                cl->read_op_id++;
+            }
             handle_op_hdr(cl);
+        }
         else
         {
             fprintf(stderr, "Received garbage: magic=%jx id=%ju opcode=%jx from %d\n", cl->read_op->req.hdr.magic, cl->read_op->req.hdr.id, cl->read_op->req.hdr.opcode, cl->peer_fd);

src/client/msgr_send.cpp

@@ -14,6 +14,7 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
     if (cur_op->op_type == OSD_OP_OUT)
     {
         clock_gettime(CLOCK_REALTIME, &cur_op->tv_begin);
+        cur_op->req.hdr.id = ++cl->send_op_id;
     }
     else
     {