Implement actual client operation cancel

src/client/cluster_client.cpp

@@ -761,6 +761,7 @@ void cluster_client_t::execute_internal(cluster_op_t *op)
     op->cur_inode = op->inode;
     op->retval = 0;
     op->state = 0;
+    op->canceled = false;
     op->retry_after = 0;
     op->inflight_count = 0;
     op->done_count = 0;
@@ -860,6 +861,113 @@ void cluster_client_t::execute_internal(cluster_op_t *op)
     }
 }
 
+// Try to cancel a previously started operation
+//
+// A canceled operation is completed with -ECANCELED, either immediately (from inside
+// this function) or later, when the messenger finishes canceling its sub-operations.
+//
+// Some operations can't be canceled at all - in that case they just complete normally.
+// The result of a canceled write is undefined - some parts of it may be written and
+// some may be not, so the write-back copy of its data is dropped, too.
+//
+// Returns true if the operation is already completed, i.e. if its callback is already called
+bool cluster_client_t::cancel(cluster_op_t *op)
+{
+    if (op->canceled || (op->flags & OP_FLUSH_BUFFER))
+    {
+        // Already canceled, or an internal write-back flush which is never canceled
+        return false;
+    }
+    // Operations which aren't started yet are just removed from their queues
+    for (auto queue: std::initializer_list<std::vector<cluster_op_t*>*>{ &offline_ops, &key_wait_ops, &wb->writeback_overflow })
+    {
+        for (auto op_it = queue->begin(); op_it != queue->end(); op_it++)
+        {
+            if (*op_it == op)
+            {
+                queue->erase(op_it);
+                op->canceled = true;
+                op->retval = -ECANCELED;
+                auto cb = std::move(op->callback);
+                cb(op);
+                return true;
+            }
+        }
+    }
+    if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && op->version)
+    {
+        // CAS operations aren't in the operation queue and have their own callback
+        if (!op->parts.size() || !(op->parts[0].flags & PART_SENT) ||
+            (op->parts[0].flags & (PART_DONE|PART_ERROR)))
+        {
+            // Not sent yet (waiting for a preliminary sync) => can't cancel it
+            return false;
+        }
+        op->canceled = true;
+        if (!msgr.cancel_out(&op->parts[0].op))
+        {
+            // The callback will be called with -ECANCELED by the messenger
+            return false;
+        }
+        op->retval = -ECANCELED;
+        auto cb = std::move(op->callback);
+        cb(op);
+        return true;
+    }
+    op->canceled = true;
+    op->retry_after = 0;
+    // ERR_PRIO(-ECANCELED) is the highest, so replies to the remaining parts
+    // won't override it and won't schedule a retry in handle_op_part()
+    op->retval = -ECANCELED;
+    // msgr.cancel_out() never calls callbacks itself, so the operation
+    // can not be completed in the middle of this loop
+    for (auto & part: op->parts)
+    {
+        if ((part.flags & PART_SENT) && !(part.flags & (PART_DONE|PART_ERROR)) &&
+            msgr.cancel_out(&part.op))
+        {
+            // Canceled synchronously => the callback won't be called at all
+            part.flags |= PART_ERROR;
+            op->inflight_count--;
+        }
+    }
+    if (op->inflight_count > 0)
+    {
+        // The rest of the parts will be canceled asynchronously
+        return false;
+    }
+    finish_canceled(op);
+    return true;
+}
+
+void cluster_client_t::finish_canceled(cluster_op_t *op)
+{
+    assert(op->canceled && !op->inflight_count);
+    op->retval = -ECANCELED;
+    op->retry_after = 0;
+    if (op->opcode == OSD_OP_SYNC)
+    {
+        if (op->parts.size())
+        {
+            // Syncs were already sent - roll the write-back cache state back and
+            // mark all unsynced OSDs as dirty again so the next sync retries them
+            wb->fsync_error();
+            for (auto & part: op->parts)
+            {
+                if (!(part.flags & PART_DONE))
+                    dirty_osds.insert(part.osd_num);
+            }
+        }
+    }
+    else if (op->flush_id)
+    {
+        // The result of a canceled write is undefined, so drop the write-repeat copy
+        wb->delete_flush(op->inode, op->offset, op->len, op->flush_id);
+        op->flush_id = 0;
+    }
+    erase_op(op);
+}
+
 void cluster_client_t::execute_cas(cluster_op_t *op, bool nosync)
 {
     if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && op->version && op->parts.size() > 1)
@@ -1085,6 +1193,16 @@ void cluster_client_t::execute_raw(osd_num_t osd_num, osd_op_t *op)
 
 int cluster_client_t::continue_rw(cluster_op_t *op)
 {
+    if (op->canceled)
+    {
+        if (op->inflight_count > 0)
+        {
+            // Wait for the remaining parts to be canceled
+            return 0;
+        }
+        finish_canceled(op);
+        return 1;
+    }
     if (op->state == 0)
         goto resume_0;
     else if (op->state == 1)
@@ -1568,6 +1686,16 @@ int cluster_client_t::try_send(cluster_op_t *op, int i, std::function<void(osd_o
 
 int cluster_client_t::continue_sync(cluster_op_t *op)
 {
+    if (op->canceled)
+    {
+        if (op->inflight_count > 0)
+        {
+            // Wait for the remaining parts to be canceled
+            return 0;
+        }
+        finish_canceled(op);
+        return 1;
+    }
     if (op->state == 1)
         goto resume_1;
     if (!dirty_osds.size())
@@ -1676,7 +1804,8 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
         if (ERR_PRIO(part->op.reply.hdr.retval) > ERR_PRIO(op->retval))
             op->retval = part->op.reply.hdr.retval;
         uint64_t stop_client_id = 0;
-        if (op->retval != -EINTR && op->retval != -EIO && op->retval != -ENOSPC && op->retval != -EPERM && op->retval != -EAGAIN)
+        if (op->retval != -EINTR && op->retval != -EIO && op->retval != -ENOSPC &&
+            op->retval != -EPERM && op->retval != -EAGAIN && op->retval != -ECANCELED)
         {
             stop_client_id = part->op.client_id;
             if (op->retval != -EPIPE || log_level > 0)

src/client/cluster_client.h

@@ -65,6 +65,7 @@ protected:
     uint64_t cur_inode; // for snapshot reads
     bool needs_reslice: 1;
     bool deoptimise_snapshot: 1;
+    bool canceled = false;
     int retry_after = 0;
     int inflight_count = 0, done_count = 0;
     timespec wait_up_until = {};
@@ -190,6 +191,7 @@ public:
     cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd, json11::Json config, std::unique_ptr<etcd_state_client_t> st_cli);
     ~cluster_client_t();
     void execute(cluster_op_t *op);
+    bool cancel(cluster_op_t *op);
     void execute_raw(osd_num_t osd_num, osd_op_t *op);
     bool is_ready();
     void on_ready(std::function<void(void)> fn);
@@ -232,6 +234,7 @@ protected:
     int continue_sync(cluster_op_t *op);
     void send_sync(cluster_op_t *op, cluster_op_part_t *part);
     void handle_op_part(cluster_op_part_t *part);
+    void finish_canceled(cluster_op_t *op);
     void copy_part_bitmap(cluster_op_t *op, cluster_op_part_t *part);
     void erase_op(cluster_op_t *op);
     void calc_wait(cluster_op_t *op);

src/client/messenger.h

@@ -57,6 +57,7 @@
 #define MSGR_SENDP_IN 2
 #define MSGR_SENDP_OUT 4
 #define MSGR_SENDP_REF 8
+#define MSGR_SENDP_CANCELED 16
 
 struct osd_client_t;
 
@@ -151,6 +152,7 @@ struct osd_client_t
     // Write state
     std::deque<osd_op_t *> write_ops;
     osd_op_t *write_op = NULL;
+    bool write_op_cancel = false;
     size_t write_op_pos = 0;
     msghdr write_msg = { 0 };
     int write_state = 0;
@@ -312,6 +314,7 @@ public:
     void stop_client(uint64_t client_id, bool force_delete = false);
     void destroy_client(osd_client_t *cl);
     void outbox_push(osd_op_t *cur_op);
+    bool cancel_out(osd_op_t *cur_op);
     std::function<void(osd_op_t*)> exec_op;
     std::function<void(osd_num_t)> repeer_pgs;
     std::function<void(osd_num_t)> break_pg_locks;
@@ -352,6 +355,7 @@ protected:
     void check_peer_config(osd_client_t *cl);
     void cancel_osd_ops(osd_client_t *cl);
     void cancel_op(osd_op_t *op);
+    osd_op_t *copy_out_op_to_ignore(osd_op_t *cur_op);
 
     bool try_send(osd_client_t *cl);
     bool wakeup_send(osd_client_t *cl);

src/client/msgr_op.h

@@ -187,13 +187,17 @@ struct __attribute__((visibility("default"))) osd_op_t
     void *rmw_buf = NULL;
     std::shared_ptr<osd_op_enc_t> enc;
     uint64_t csum = 0; // network layer checksum
+    uint64_t send_free_pos = 0;
+    // Read the reply and throw it away instead of copying it anywhere.
+    // Used for stub operations left in place of canceled ones
+    bool ignore_reply = false;
     osd_primary_op_data_t* op_data = NULL;
     std::function<void(osd_op_t*)> callback;
 
     osd_op_buf_list_t iov;
 
     ~osd_op_t();
-    void cancel();
+    void cancel(int retval);
 
     bool is_recovery_related();
     uint64_t calc_data_checksum();

src/client/msgr_receive.cpp

@@ -179,14 +179,15 @@ public:
         }
         else if (flags & RDR_GCM)
         {
-            // Here, dst == NULL is not allowed
-            assert(dst != NULL);
+            // Here, dst == NULL means "decrypt in place and throw the data away".
+            // The data still has to be decrypted to keep the GCM stream and tag in sync
+            uint8_t *out = (dst ? dst+from : curbuf+done);
 #ifdef WITH_ISAL_CRYPTO
-            int r = isal_aes_gcm_dec_256_update(&cl->peer_key_isal, cl->dec_ctx, dst+from, curbuf+done, n);
+            int r = isal_aes_gcm_dec_256_update(&cl->peer_key_isal, cl->dec_ctx, out, curbuf+done, n);
             assert(!r);
 #else
             int actual_out;
-            if (EVP_DecryptUpdate(cl->dec_ctx, dst+from, &actual_out, curbuf+done, n) != 1)
+            if (EVP_DecryptUpdate(cl->dec_ctx, out, &actual_out, curbuf+done, n) != 1)
             {
                 fprintf(stderr, "DecryptUpdate error: ");
                 ERR_print_errors_fp(stderr);
@@ -196,7 +197,7 @@ public:
 #endif
             if (cl->read_csum_state && !(flags & RDR_NO_CSUM))
             {
-                XXH3_64bits_update(cl->read_csum_state, dst+from, n);
+                XXH3_64bits_update(cl->read_csum_state, out, n);
             }
             done += n;
         }
@@ -307,6 +308,8 @@ public:
     get_op_reader_t(osd_messenger_t* msgr, osd_client_t* cl):
         cl(cl), from(cl->read_op_pos), mpos(false)
     {
+        // Discarded replies are always read through in_buf - this reader needs real buffers
+        assert(!cl->read_op->ignore_reply);
         if (cl->read_op->op_type == OSD_OP_OUT &&
             cl->read_op->reply.hdr.opcode == OSD_OP_READ &&
             cl->read_op->reply.hdr.retval > 0)
@@ -397,7 +400,8 @@ void osd_messenger_t::read_requests()
             continue;
         }
         auto cl = cl_it->second;
-        if (!rxbounce && cl->read_op && cl->read_op_pos >= OSD_PACKET_SIZE && cl->read_op_size-(cl->read_op_pos-OSD_PACKET_SIZE) >= receive_buffer_size)
+        if (!rxbounce && cl->read_op && !cl->read_op->ignore_reply && cl->read_op_pos >= OSD_PACKET_SIZE &&
+            cl->read_op_size-(cl->read_op_pos-OSD_PACKET_SIZE) >= receive_buffer_size)
         {
             get_op_reader_t rdr(this, cl);
             if (!op_read_from(cl, rdr))
@@ -570,11 +574,15 @@ void osd_messenger_t::handle_immediate_ops()
             else
                 delete op;
         }
-        else
+        else if (op->callback)
         {
             // Copy lambda to be unaffected by `delete op`
             std::function<void(osd_op_t*)>(op->callback)(op);
         }
+        else
+        {
+            delete op;
+        }
     }
 }
 
@@ -859,12 +867,12 @@ bool osd_messenger_t::allocate_reply_buffers(osd_client_t *cl, osd_op_t *op)
         }
         if (op->reply.hdr.retval >= 0 && bmp_len > 0)
         {
-            assert(op->bitmap);
+            assert(op->ignore_reply || op->bitmap);
             cl->read_op_size += bmp_len;
         }
         if (op->reply.hdr.retval > 0)
         {
-            assert(op->iov.count > 0);
+            assert(op->ignore_reply || op->iov.count > 0);
             cl->read_op_size += op->reply.hdr.retval;
         }
     }
@@ -878,7 +886,8 @@ bool osd_messenger_t::allocate_reply_buffers(osd_client_t *cl, osd_op_t *op)
                 cl->client_id, op->reply.hdr.retval);
             return false;
         }
-        op->buf = malloc_or_die(cl->read_op_size);
+        if (!op->ignore_reply)
+            op->buf = malloc_or_die(cl->read_op_size);
     }
     else if (op->reply.hdr.opcode == OSD_OP_SEC_READ_BMP && op->reply.hdr.retval > 0)
     {
@@ -891,8 +900,11 @@ bool osd_messenger_t::allocate_reply_buffers(osd_client_t *cl, osd_op_t *op)
             return false;
         }
         cl->read_op_size = op->reply.hdr.retval;
-        free(op->buf);
-        op->buf = malloc_or_die(cl->read_op_size);
+        if (!op->ignore_reply)
+        {
+            free(op->buf);
+            op->buf = malloc_or_die(cl->read_op_size);
+        }
     }
     else if (op->reply.hdr.opcode == OSD_OP_SHOW_CONFIG && op->reply.hdr.retval > 0)
     {
@@ -903,14 +915,20 @@ bool osd_messenger_t::allocate_reply_buffers(osd_client_t *cl, osd_op_t *op)
                 cl->client_id, op->reply.hdr.retval, MAX_SIMPLE_PAYLOAD_SIZE);
             return false;
         }
-        free(op->buf);
-        op->buf = malloc_or_die(op->reply.hdr.retval);
+        if (!op->ignore_reply)
+        {
+            free(op->buf);
+            op->buf = malloc_or_die(op->reply.hdr.retval);
+        }
     }
     else if (op->reply.hdr.opcode == OSD_OP_DESCRIBE && op->reply.describe.result_bytes > 0)
     {
         cl->read_op_size = op->reply.describe.result_bytes;
-        free(op->buf);
-        op->buf = malloc_or_die(op->reply.describe.result_bytes);
+        if (!op->ignore_reply)
+        {
+            free(op->buf);
+            op->buf = malloc_or_die(op->reply.describe.result_bytes);
+        }
     }
     if (cl->proto_csum_status == MSGR_CSUM_FULL ||
         cl->read_op_size > 0 && cl->proto_csum_status == MSGR_CSUM_PAYLOAD)
@@ -974,7 +992,37 @@ bool osd_messenger_t::op_read_from(osd_client_t *cl, msgr_op_reader_t & rdr)
         if (!rdr.read(op->reply.buf, OSD_PACKET_SIZE, RDR_GCM | (cl->proto_csum_status == MSGR_CSUM_PAYLOAD ? RDR_NO_CSUM : 0)))
             return false;
 switched_type:
-        if (op->reply.hdr.opcode == OSD_OP_SEC_READ)
+        if (op->ignore_reply)
+        {
+            // Reply to a canceled operation - read it into nowhere and throw it away.
+            // Read flags have to exactly match the normal path below, otherwise
+            // the GCM stream and the protocol checksum would desync
+            if (op->reply.hdr.opcode == OSD_OP_SEC_READ || op->reply.hdr.opcode == OSD_OP_READ)
+            {
+                // Conditions mirror allocate_reply_buffers() which defines cl->read_op_size,
+                // so that we always consume exactly the framed length
+                uint32_t bmp_len = (op->reply.hdr.opcode == OSD_OP_SEC_READ
+                    ? op->reply.sec_rw.attr_len : op->reply.rw.bitmap_len);
+                if (op->reply.hdr.retval >= 0 && bmp_len > 0 && !rdr.read(NULL, bmp_len, RDR_GCM))
+                    return false;
+                // Note that RDR_XTS is intentionally not passed - the stub has no keys and
+                // XTS is length-preserving, so both the GCM stream and the checksum match
+                if (op->reply.hdr.retval > 0 && !rdr.read(NULL, op->reply.hdr.retval,
+                    (cl->proto_csum_status == MSGR_CSUM_GCM ? RDR_GCM : 0)))
+                    return false;
+            }
+            else
+            {
+                // All other replies have a single payload region, always read with RDR_GCM
+                size_t payload = cl->read_op_size;
+                if (cl->proto_csum_status == MSGR_CSUM_FULL ||
+                    payload > 0 && cl->proto_csum_status == MSGR_CSUM_PAYLOAD)
+                    payload -= 8;
+                if (payload > 0 && !rdr.read(NULL, payload, RDR_GCM))
+                    return false;
+            }
+        }
+        else if (op->reply.hdr.opcode == OSD_OP_SEC_READ)
         {
             if (op->reply.sec_rw.attr_len > 0)
             {

src/client/msgr_send.cpp

@@ -436,7 +436,7 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
     auto cl_it = clients.find(cur_op->client_id);
     if (cl_it == clients.end() || cl_it->second->peer_state == PEER_STOPPED)
     {
-        cur_op->cancel();
+        cur_op->cancel(-EPIPE);
         return;
     }
     osd_client_t *cl = cl_it->second;
@@ -494,6 +494,81 @@ bool osd_messenger_t::wakeup_send(osd_client_t *cl)
     return true;
 }
 
+// Try to cancel an outbound operation
+// Returns true when canceled synchronously, false if canceling is impossible or will be
+// performed asynchronously and the caller has to wait for the callback to be called
+bool osd_messenger_t::cancel_out(osd_op_t *cur_op)
+{
+    assert(cur_op->client_id);
+    assert(cur_op->op_type == OSD_OP_OUT);
+    auto cl_it = clients.find(cur_op->client_id);
+    if (cl_it == clients.end() || cl_it->second->peer_state == PEER_STOPPED)
+    {
+        // Client is stopped => messenger will call the callback anyway
+        return false;
+    }
+    osd_client_t *cl = cl_it->second;
+    auto sent_it = cl->sent_ops.find(cur_op->req.hdr.id);
+    if (sent_it == cl->sent_ops.end())
+    {
+        // Reply header is already received => callback is pending
+        return false;
+    }
+    if (sent_it->second != cur_op)
+    {
+        // Already canceled
+        return false;
+    }
+    // Make a copy to have some space to place the reply to and forget the original object
+    if (cl->write_op == cur_op)
+    {
+        // Operation is being sent => wait for full send, then cancel it
+        sent_it->second = copy_out_op_to_ignore(cur_op);
+        cl->write_op_cancel = true;
+        return false;
+    }
+    for (auto wr_it = cl->write_ops.begin(); wr_it != cl->write_ops.end(); wr_it++)
+    {
+        if (*wr_it == cur_op)
+        {
+            // Operation is only queued => just remove it from the send queue
+            cl->sent_ops.erase(sent_it);
+            cl->write_ops.erase(wr_it);
+            cur_op->callback = NULL;
+            return true;
+        }
+    }
+    sent_it->second = copy_out_op_to_ignore(cur_op);
+    if (cur_op->send_free_pos >= cl->next_send_free)
+    {
+        // In progress of sending => free it when the send_free reference is cleared
+        assert(cur_op->send_free_pos < cl->next_send_free + cl->send_free.size());
+        auto & sf = cl->send_free[cur_op->send_free_pos - cl->next_send_free];
+        assert((osd_op_t*)sf.buf == cur_op);
+        sf.flags |= MSGR_SENDP_CANCELED;
+        return false;
+    }
+    // Operation is fully sent and not referenced by the send_free queue
+    cur_op->callback = NULL;
+    return true;
+}
+
+// Make a stub operation to swallow the reply to a canceled operation.
+// The reply body is read into nowhere, so no data buffers are needed here -
+// only the request header, which is required to validate the reply
+osd_op_t *osd_messenger_t::copy_out_op_to_ignore(osd_op_t *cur_op)
+{
+    osd_op_t *op_copy = new osd_op_t;
+    op_copy->op_type = OSD_OP_OUT;
+    op_copy->ignore_reply = true;
+    op_copy->tv_begin = cur_op->tv_begin;
+    op_copy->client_id = cur_op->client_id;
+    op_copy->osd_num = cur_op->osd_num;
+    op_copy->bitmap_len = cur_op->bitmap_len;
+    memcpy(op_copy->req.buf, cur_op->req.buf, OSD_PACKET_SIZE);
+    return op_copy;
+}
+
 bool osd_messenger_t::try_send(osd_client_t *cl)
 {
     assert(cl->peer_state != PEER_RDMA);
@@ -686,7 +761,11 @@ size_t osd_messenger_t::copy_ops_to_with(osd_client_t *cl, uint8_t *dst, size_t
             break;
         }
         assert(cl->send_free.size() < UINT64_MAX - cl->next_send_free);
-        cl->send_free.push_back({ .buf = op, .flags = (uint32_t)((op->op_type == OSD_OP_IN ? MSGR_SENDP_IN : MSGR_SENDP_OUT) | MSGR_SENDP_REF) });
+        uint32_t flags = ((op->op_type == OSD_OP_IN ? MSGR_SENDP_IN : MSGR_SENDP_OUT) | MSGR_SENDP_REF);
+        if (cl->write_op_cancel)
+            flags |= MSGR_SENDP_CANCELED;
+        op->send_free_pos = cl->next_send_free + cl->send_free.size();
+        cl->send_free.push_back({ .buf = op, .flags = flags });
     }
     return wr.get_done();
 }
@@ -694,6 +773,7 @@ size_t osd_messenger_t::copy_ops_to_with(osd_client_t *cl, uint8_t *dst, size_t
 void osd_messenger_t::next_write_op(osd_client_t *cl)
 {
     cl->write_op = cl->write_ops.front();
+    cl->write_op_cancel = false;
     cl->write_ops.pop_front();
     if (cl->proto_csum_status == MSGR_CSUM_FULL || cl->proto_csum_status == MSGR_CSUM_PAYLOAD)
     {

src/client/msgr_stop.cpp

@@ -22,24 +22,24 @@ void osd_client_t::cancel_ops()
     sent_ops.clear();
     for (auto op: cancel_ops)
     {
-        op->cancel();
+        op->cancel(-EPIPE);
     }
 }
 
-void osd_op_t::cancel()
+void osd_op_t::cancel(int retval)
 {
     if (op_type == OSD_OP_OUT && callback)
     {
         reply.hdr.magic = SECONDARY_OSD_REPLY_MAGIC;
         reply.hdr.id = req.hdr.id;
         reply.hdr.opcode = req.hdr.opcode;
-        reply.hdr.retval = -EPIPE;
+        reply.hdr.retval = retval;
         // Copy lambda to be unaffected by `delete this`
         (std::function<void(osd_op_t*)>(callback))(this);
     }
     else
     {
-        // This function is only called in stop_client(), so it's fine to destroy the operation
+        // It's fine to just free responses to incoming ops
         delete this;
     }
 }
@@ -57,6 +57,8 @@ void osd_client_t::unref_send_free(uint64_t from, uint64_t to)
         {
             if (it->flags & MSGR_SENDP_IN)
                 delete (osd_op_t*)it->buf;
+            else if (it->flags & MSGR_SENDP_CANCELED)
+                ((osd_op_t*)it->buf)->cancel(-ECANCELED);
             else if (it->flags & MSGR_SENDP_BUF)
                 free(it->buf);
         }
@@ -67,6 +69,8 @@ void osd_client_t::unref_send_free(uint64_t from, uint64_t to)
         {
             if (it->flags & MSGR_SENDP_IN)
                 delete (osd_op_t*)it->buf;
+            else if (it->flags & MSGR_SENDP_CANCELED)
+                ((osd_op_t*)it->buf)->cancel(-ECANCELED);
             else if (it->flags & MSGR_SENDP_BUF)
                 free(it->buf);
             to++;
@@ -204,14 +208,21 @@ osd_client_t::~osd_client_t()
     if (read_op)
     {
         // read_op may be an incoming op or a continued response for an outbound op
-        read_op->cancel();
+        read_op->cancel(-EPIPE);
         read_op = NULL;
     }
     // Delete unsent replies - before cancel_ops() because it may delete outbound operations.
     if (write_op)
     {
         if (write_op->op_type == OSD_OP_IN)
+        {
             delete write_op;
+        }
+        else if (write_op_cancel)
+        {
+            // not in sent_ops anymore - should be canceled here
+            write_op->cancel(-ECANCELED);
+        }
         write_op = NULL;
     }
     for (auto op: write_ops)

src/test/CMakeLists.txt

@@ -27,6 +27,17 @@ target_link_libraries(stub_uring_osd
 	${RDMACM_LIBRARIES}
 )
 
+# test_messenger
+add_executable(test_messenger EXCLUDE_FROM_ALL test_messenger.cpp)
+target_link_libraries(test_messenger
+	vitastor_net
+	${LIBURING_LIBRARIES}
+	${IBVERBS_LIBRARIES}
+	${RDMACM_LIBRARIES}
+)
+add_dependencies(build_tests test_messenger)
+add_test(NAME test_messenger COMMAND test_messenger)
+
 # test_allocator
 add_executable(test_allocator EXCLUDE_FROM_ALL test_allocator.cpp)
 target_link_libraries(test_allocator util)

src/test/mock/messenger.cpp

@@ -25,7 +25,7 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
     auto cl_it = clients.find(cur_op->client_id);
     if (cl_it == clients.end() || cl_it->second->peer_state == PEER_STOPPED)
     {
-        cur_op->cancel();
+        cur_op->cancel(-EPIPE);
         return;
     }
     auto cl = cl_it->second;
@@ -34,6 +34,35 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op)
     cl->sent_ops[cur_op->req.hdr.id] = cur_op;
 }
 
+// Set to true to make cancel_out() pretend that operations are already sent
+// and have to be canceled asynchronously
+bool mock_msgr_async_cancel = false;
+
+bool osd_messenger_t::cancel_out(osd_op_t *cur_op)
+{
+    assert(cur_op->client_id);
+    assert(cur_op->op_type == OSD_OP_OUT);
+    auto cl_it = clients.find(cur_op->client_id);
+    if (cl_it == clients.end() || cl_it->second->peer_state == PEER_STOPPED)
+    {
+        return false;
+    }
+    auto cl = cl_it->second;
+    auto sent_it = cl->sent_ops.find(cur_op->req.hdr.id);
+    if (sent_it == cl->sent_ops.end() || sent_it->second != cur_op)
+    {
+        return false;
+    }
+    if (mock_msgr_async_cancel)
+    {
+        // The caller is expected to "complete" the operation with -ECANCELED itself
+        return false;
+    }
+    cl->sent_ops.erase(sent_it);
+    cur_op->callback = NULL;
+    return true;
+}
+
 void osd_messenger_t::parse_config(const json11::Json & config, bool init)
 {
 }

src/test/test_cluster_client.cpp

@@ -12,6 +12,9 @@
 #include "cluster_client_impl.h"
 #include "msgr_encrypt.h"
 
+// Defined in test/mock/messenger.cpp
+extern bool mock_msgr_async_cancel;
+
 class cluster_client_test_t
 {
 public:
@@ -53,6 +56,9 @@ void configure_single_pg_pool(etcd_state_client_mock_t *mock)
     });
 }
 
+// Last operation posted by test_write()/test_read()/test_sync(), for cancellation tests
+cluster_op_t *last_posted_op = NULL;
+
 int *test_write(cluster_client_t *cli, uint64_t offset, uint64_t len, uint8_t c, std::function<void()> cb = NULL, bool instant = false)
 {
     printf("Post write %jx+%jx\n", offset, len);
@@ -78,6 +84,7 @@ int *test_write(cluster_client_t *cli, uint64_t offset, uint64_t len, uint8_t c,
         if (cb != NULL)
             cb();
     };
+    last_posted_op = op;
     cli->execute(op);
     if (instant)
     {
@@ -113,6 +120,7 @@ int *test_read(cluster_client_t *cli, uint64_t offset, uint64_t len, std::functi
         if (cb != NULL)
             cb();
     };
+    last_posted_op = op;
     cli->execute(op);
     return r;
 }
@@ -129,10 +137,11 @@ int *test_sync(cluster_client_t *cli)
         if (*r == -1000)
             printf("Error: Not allowed to complete yet\n");
         assert(*r != -1000);
-        *r = op->retval == 0 ? 1 : 0;
+        *r = op->retval == 0 ? 1 : op->retval;
         printf("Done sync r=%d\n", op->retval);
         delete op;
     };
+    last_posted_op = op;
     cli->execute(op);
     return r;
 }
@@ -1122,6 +1131,127 @@ void test_snapshot_over_nonexistent_pool()
     printf("[ok] test_snapshot_over_nonexistent_pool\n");
 }
 
+void test_cancel()
+{
+    json11::Json config;
+    timerfd_manager_t *tfd = new timerfd_manager_t([](int fd, bool wr, std::function<void(int, int)> callback){});
+    etcd_state_client_mock_t *mock = new etcd_state_client_mock_t();
+    mock->pause();
+    cluster_client_t *cli = new cluster_client_t(NULL, tfd, config, std::unique_ptr<etcd_state_client_t>(mock));
+
+    // Cancel an operation posted before the configuration is loaded
+    int *r1 = test_write(cli, 0, 4096, 0x55);
+    cluster_op_t *op1 = last_posted_op;
+    can_complete(r1);
+    assert(cli->cancel(op1));
+    assert(*r1 == -ECANCELED);
+    delete r1;
+
+    configure_single_pg_pool(mock);
+    mock->resume();
+    pretend_connected(cli, 1);
+
+    // Cancel an operation which is queued, but not sent because the OSD is down
+    pretend_disconnected(cli, 1);
+    r1 = test_write(cli, 0, 4096, 0x56);
+    op1 = last_posted_op;
+    can_complete(r1);
+    assert(cli->cancel(op1));
+    assert(*r1 == -ECANCELED);
+    delete r1;
+    // The canceled write must not be repeated when the OSD comes back
+    pretend_connected(cli, 1);
+    check_op_count(cli, 1, 0);
+
+    // Cancel an already sent operation - the messenger cancels it synchronously
+    r1 = test_write(cli, 0, 4096, 0x57);
+    op1 = last_posted_op;
+    check_op_count(cli, 1, 1);
+    can_complete(r1);
+    assert(cli->cancel(op1));
+    assert(*r1 == -ECANCELED);
+    delete r1;
+    check_op_count(cli, 1, 0);
+    // The connection must NOT be dropped because of a cancellation
+    assert(cli->msgr.osd_peers.find(1) != cli->msgr.osd_peers.end());
+
+    // Cancel an already sent operation - the messenger cancels it asynchronously
+    mock_msgr_async_cancel = true;
+    r1 = test_write(cli, 0, 4096, 0x58);
+    op1 = last_posted_op;
+    check_op_count(cli, 1, 1);
+    osd_op_t *part = find_op(cli, 1, OSD_OP_WRITE, 0, 4096);
+    assert(!cli->cancel(op1));
+    can_complete(r1);
+    pretend_op_completed(cli, part, -ECANCELED);
+    assert(*r1 == -ECANCELED);
+    delete r1;
+    assert(cli->msgr.osd_peers.find(1) != cli->msgr.osd_peers.end());
+    mock_msgr_async_cancel = false;
+
+    // Cancel a sync which waits for a previous write
+    r1 = test_write(cli, 0, 4096, 0x59);
+    check_op_count(cli, 1, 1);
+    int *r2 = test_sync(cli);
+    cluster_op_t *op2 = last_posted_op;
+    check_op_count(cli, 1, 1);
+    can_complete(r2);
+    assert(cli->cancel(op2));
+    assert(*r2 == -ECANCELED);
+    delete r2;
+    // The write must complete as usual
+    can_complete(r1);
+    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_WRITE, 0, 4096), 0);
+    check_completed(r1);
+
+    // Cancel an already sent sync - the next sync must still sync the same OSD
+    r2 = test_sync(cli);
+    op2 = last_posted_op;
+    check_op_count(cli, 1, 1);
+    can_complete(r2);
+    assert(cli->cancel(op2));
+    assert(*r2 == -ECANCELED);
+    delete r2;
+    check_op_count(cli, 1, 0);
+    r2 = test_sync(cli);
+    check_op_count(cli, 1, 1);
+    can_complete(r2);
+    pretend_op_completed(cli, find_op(cli, 1, OSD_OP_SYNC, 0, 0), 0);
+    check_completed(r2);
+
+    // Cancel a CAS write
+    int *r3 = new int;
+    *r3 = -1000;
+    cluster_op_t *op3 = new cluster_op_t();
+    op3->opcode = OSD_OP_WRITE;
+    op3->inode = 0x1000000000001;
+    op3->offset = 0;
+    op3->len = 4096;
+    op3->version = 1;
+    op3->iov.push_back(malloc_or_die(4096), 4096);
+    memset(op3->iov.buf[0].iov_base, 0x5a, 4096);
+    op3->callback = [r3](cluster_op_t *op)
+    {
+        assert(*r3 != -1000);
+        *r3 = op->retval;
+        printf("Done CAS write %jx+%jx r=%d\n", op->offset, op->len, op->retval);
+        free(op->iov.buf[0].iov_base);
+        delete op;
+    };
+    printf("Post CAS write 0+1000\n");
+    cli->execute(op3);
+    check_op_count(cli, 1, 1);
+    can_complete(r3);
+    assert(cli->cancel(op3));
+    assert(*r3 == -ECANCELED);
+    delete r3;
+    check_op_count(cli, 1, 0);
+
+    delete cli;
+    delete tfd;
+    printf("[ok] test_cancel\n");
+}
+
 int main(int narg, char *args[])
 {
     test1();
@@ -1135,5 +1265,6 @@ int main(int narg, char *args[])
     cluster_client_test_t::test_vault();
     test_pg_config_disappears();
     test_snapshot_over_nonexistent_pool();
+    test_cancel();
     return 0;
 }

src/test/test_messenger.cpp

@@ -0,0 +1,302 @@
+// Copyright (c) Vitaliy Filippov, 2019+
+// License: VNPL-1.1 or GNU GPL-2.0+ (see README.md for details)
+//
+// Tests for the messenger receive path, in particular for reading and
+// discarding replies to canceled operations (osd_op_t::ignore_reply)
+
+#undef NDEBUG
+#include <assert.h>
+#include <stdio.h>
+#include <vector>
+
+#include "messenger.h"
+#include "util/xxh_x86dispatch.h"
+
+#define BMP_LEN 8
+#define DATA_LEN 4096
+
+// The receive path is protected, so reach it through a subclass
+struct test_msgr_t: osd_messenger_t
+{
+    bool feed(osd_client_t *cl, uint8_t *buf, size_t len) { return handle_read_buffer(cl, buf, len); }
+    void drain() { handle_immediate_ops(); }
+    osd_op_t *make_stub(osd_op_t *op) { return copy_out_op_to_ignore(op); }
+    size_t serialize(osd_client_t *cl, uint8_t *dst, size_t len) { return copy_ops_to(cl, dst, len); }
+    size_t immediate_count() { return set_immediate_ops.size(); }
+};
+
+static void append_csum(std::vector<uint8_t> & buf, int csum_status, size_t payload)
+{
+    if (csum_status == MSGR_CSUM_FULL)
+    {
+        // The header is covered by the checksum too
+        uint64_t csum = XXH3_64bits(buf.data(), buf.size());
+        buf.resize(buf.size()+8);
+        memcpy(buf.data()+buf.size()-8, &csum, 8);
+    }
+    else if (csum_status == MSGR_CSUM_PAYLOAD)
+    {
+        uint64_t csum = XXH3_64bits(buf.data()+OSD_PACKET_SIZE, payload);
+        buf.resize(buf.size()+8);
+        memcpy(buf.data()+buf.size()-8, &csum, 8);
+    }
+}
+
+// SEC_READ reply: header + bitmap + data
+static std::vector<uint8_t> make_read_reply(uint64_t id, uint8_t fill, int csum_status = 0)
+{
+    std::vector<uint8_t> buf(OSD_PACKET_SIZE + BMP_LEN + DATA_LEN);
+    osd_any_reply_t *r = (osd_any_reply_t*)buf.data();
+    memset(buf.data(), 0, OSD_PACKET_SIZE);
+    r->hdr.magic = SECONDARY_OSD_REPLY_MAGIC;
+    r->hdr.id = id;
+    r->hdr.opcode = OSD_OP_SEC_READ;
+    r->hdr.retval = DATA_LEN;
+    r->sec_rw.attr_len = BMP_LEN;
+    memset(buf.data()+OSD_PACKET_SIZE, 0xB0|fill, BMP_LEN);
+    memset(buf.data()+OSD_PACKET_SIZE+BMP_LEN, fill, DATA_LEN);
+    append_csum(buf, csum_status, BMP_LEN+DATA_LEN);
+    return buf;
+}
+
+// SEC_LIST reply - a single payload region, unlike reads
+static std::vector<uint8_t> make_list_reply(uint64_t id, uint64_t count, int csum_status = 0)
+{
+    size_t payload = sizeof(obj_ver_id)*count;
+    std::vector<uint8_t> buf(OSD_PACKET_SIZE + payload);
+    osd_any_reply_t *r = (osd_any_reply_t*)buf.data();
+    memset(buf.data(), 0, OSD_PACKET_SIZE);
+    r->hdr.magic = SECONDARY_OSD_REPLY_MAGIC;
+    r->hdr.id = id;
+    r->hdr.opcode = OSD_OP_SEC_LIST;
+    r->hdr.retval = count;
+    memset(buf.data()+OSD_PACKET_SIZE, 0x5A, payload);
+    append_csum(buf, csum_status, payload);
+    return buf;
+}
+
+static osd_op_t *make_read_op(uint64_t id)
+{
+    osd_op_t *op = new osd_op_t;
+    op->op_type = OSD_OP_OUT;
+    op->req.hdr.magic = SECONDARY_OSD_REPLY_MAGIC;
+    op->req.hdr.id = id;
+    op->req.hdr.opcode = OSD_OP_SEC_READ;
+    op->req.sec_rw.len = DATA_LEN;
+    op->bitmap_len = BMP_LEN;
+    return op;
+}
+
+// A read operation with preallocated buffers, like cluster_client and the OSD make them
+static osd_op_t *make_real_read_op(uint64_t id, int *done, uint8_t **out_data)
+{
+    osd_op_t *op = make_read_op(id);
+    op->buf = malloc_or_die(BMP_LEN + DATA_LEN);
+    op->iov.push_back(op->buf, DATA_LEN);
+    op->bitmap = (uint8_t*)op->buf + DATA_LEN;
+    *out_data = (uint8_t*)op->buf;
+    op->callback = [done](osd_op_t *op) { *done = op->reply.hdr.retval; };
+    return op;
+}
+
+// Put a stub in place of a canceled operation, like cancel_out() does
+static void cancel_op(test_msgr_t *msgr, osd_client_t *cl, osd_op_t *op)
+{
+    osd_op_t *stub = msgr->make_stub(op);
+    assert(stub->ignore_reply);
+    assert(stub->op_type == OSD_OP_OUT);
+    // No data buffers are allocated for a discarded reply
+    assert(!stub->buf && !stub->iov.count && !stub->bitmap);
+    cl->sent_ops[op->req.hdr.id] = stub;
+    delete op;
+}
+
+static osd_client_t *add_client(test_msgr_t *msgr, int csum_status)
+{
+    auto cl = new osd_client_t();
+    cl->client_id = msgr->next_client_id++;
+    cl->peer_fd = -1;
+    cl->peer_state = PEER_CONNECTED;
+    cl->proto_csum_status = csum_status;
+    msgr->clients[cl->client_id] = cl;
+    return cl;
+}
+
+// A canceled reply is read and thrown away, and the stub is freed
+static void test_discard_one_chunk(test_msgr_t *msgr)
+{
+    auto cl = add_client(msgr, 0);
+    cancel_op(msgr, cl, make_read_op(1));
+    auto buf = make_read_reply(1, 0x11);
+    assert(msgr->feed(cl, buf.data(), buf.size()));
+    assert(cl->sent_ops.size() == 0);
+    assert(msgr->immediate_count() == 1);
+    msgr->drain();
+    msgr->stop_client(cl->client_id);
+    printf("[ok] discarded reply, single chunk\n");
+}
+
+// The same, but split into small buffers - checks that the skip resumes correctly
+static void test_discard_chunked(test_msgr_t *msgr)
+{
+    auto cl = add_client(msgr, 0);
+    cancel_op(msgr, cl, make_read_op(1));
+    auto buf = make_read_reply(1, 0x22);
+    size_t chunk = 100;
+    for (size_t pos = 0; pos < buf.size(); pos += chunk)
+    {
+        size_t n = buf.size()-pos < chunk ? buf.size()-pos : chunk;
+        assert(msgr->feed(cl, buf.data()+pos, n));
+    }
+    assert(cl->sent_ops.size() == 0);
+    msgr->drain();
+    msgr->stop_client(cl->client_id);
+    printf("[ok] discarded reply, %zu-byte chunks\n", chunk);
+}
+
+// The important one: if the discard consumes the wrong number of bytes,
+// the next reply in the same buffer is garbage
+static void test_stream_stays_in_sync(test_msgr_t *msgr, int csum_status)
+{
+    auto cl = add_client(msgr, csum_status);
+    cancel_op(msgr, cl, make_read_op(1));
+    int done = 0;
+    uint8_t *data = NULL;
+    osd_op_t *real = make_real_read_op(2, &done, &data);
+    cl->sent_ops[2] = real;
+    auto b1 = make_read_reply(1, 0x33, csum_status);
+    auto b2 = make_read_reply(2, 0x44, csum_status);
+    std::vector<uint8_t> both(b1);
+    both.insert(both.end(), b2.begin(), b2.end());
+    assert(msgr->feed(cl, both.data(), both.size()));
+    msgr->drain();
+    assert(!cl->io_error);
+    assert(cl->sent_ops.size() == 0);
+    assert(done == DATA_LEN);
+    for (int i = 0; i < DATA_LEN; i++)
+        assert(data[i] == 0x44);
+    assert(((uint8_t*)real->bitmap)[0] == (0xB0|0x44));
+    delete real;
+    msgr->stop_client(cl->client_id);
+    printf("[ok] stream stays in sync after a discarded reply, csum_status=%d\n", csum_status);
+}
+
+// SEC_LIST goes through the single-payload-region branch of the discard path
+static void test_discard_list(test_msgr_t *msgr, int csum_status)
+{
+    auto cl = add_client(msgr, csum_status);
+    osd_op_t *op = new osd_op_t;
+    op->op_type = OSD_OP_OUT;
+    op->req.hdr.id = 1;
+    op->req.hdr.opcode = OSD_OP_SEC_LIST;
+    cancel_op(msgr, cl, op);
+    int done = 0;
+    uint8_t *data = NULL;
+    osd_op_t *real = make_real_read_op(2, &done, &data);
+    cl->sent_ops[2] = real;
+    auto b1 = make_list_reply(1, 37, csum_status);
+    auto b2 = make_read_reply(2, 0x77, csum_status);
+    std::vector<uint8_t> both(b1);
+    both.insert(both.end(), b2.begin(), b2.end());
+    assert(msgr->feed(cl, both.data(), both.size()));
+    msgr->drain();
+    assert(!cl->io_error);
+    assert(cl->sent_ops.size() == 0);
+    assert(done == DATA_LEN);
+    for (int i = 0; i < DATA_LEN; i++)
+        assert(data[i] == 0x77);
+    delete real;
+    msgr->stop_client(cl->client_id);
+    printf("[ok] discarded SEC_LIST + real reply, csum_status=%d\n", csum_status);
+}
+
+// The data is thrown away, but it's still checksummed on the way
+static void test_discard_checks_csum(test_msgr_t *msgr, int csum_status)
+{
+    auto cl = add_client(msgr, csum_status);
+    uint64_t client_id = cl->client_id;
+    cancel_op(msgr, cl, make_read_op(1));
+    auto buf = make_read_reply(1, 0x88, csum_status);
+    buf[OSD_PACKET_SIZE+BMP_LEN+10] ^= 0xff;
+    printf("-- a checksum mismatch message is expected below:\n");
+    msgr->feed(cl, buf.data(), buf.size());
+    // A checksum mismatch stops the client, so it has to be gone by now
+    assert(msgr->clients.find(client_id) == msgr->clients.end());
+    printf("[ok] corruption inside a discarded reply is still detected, csum_status=%d\n", csum_status);
+}
+
+// Build a SEC_READ reply the way an OSD does, for the real send path
+static osd_op_t *make_reply_op(uint64_t id, uint8_t fill)
+{
+    osd_op_t *op = new osd_op_t;
+    op->op_type = OSD_OP_IN;
+    op->req.hdr.id = id;
+    op->req.hdr.opcode = OSD_OP_SEC_READ;
+    op->req.sec_rw.len = DATA_LEN;
+    op->reply.hdr.magic = SECONDARY_OSD_REPLY_MAGIC;
+    op->reply.hdr.id = id;
+    op->reply.hdr.opcode = OSD_OP_SEC_READ;
+    op->reply.hdr.retval = DATA_LEN;
+    op->reply.sec_rw.attr_len = BMP_LEN;
+    op->buf = malloc_or_die(BMP_LEN + DATA_LEN);
+    memset(op->buf, fill, DATA_LEN);
+    op->bitmap = (uint8_t*)op->buf + DATA_LEN;
+    memset(op->bitmap, 0xB0|fill, BMP_LEN);
+    op->iov.push_back(op->buf, DATA_LEN);
+    return op;
+}
+
+// Serialize two replies through the real send path and feed them into the real receive
+// path, discarding the first one. Unlike the tests above, the checksum is produced by
+// the messenger itself, so this also verifies that the sender and the receiver agree
+static void test_roundtrip(test_msgr_t *msgr, int csum_status)
+{
+    auto scl = add_client(msgr, csum_status);
+    scl->write_ops.push_back(make_reply_op(1, 0x11));
+    scl->write_ops.push_back(make_reply_op(2, 0x22));
+    std::vector<uint8_t> wire(2*(OSD_PACKET_SIZE + BMP_LEN + DATA_LEN + 8));
+    size_t n = msgr->serialize(scl, wire.data(), wire.size());
+    assert(!scl->write_op && !scl->write_ops.size());
+    wire.resize(n);
+
+    auto rcl = add_client(msgr, csum_status);
+    cancel_op(msgr, rcl, make_read_op(1));
+    int done = 0;
+    uint8_t *data = NULL;
+    osd_op_t *real = make_real_read_op(2, &done, &data);
+    rcl->sent_ops[2] = real;
+    assert(msgr->feed(rcl, wire.data(), wire.size()));
+    msgr->drain();
+    assert(!rcl->io_error);
+    assert(rcl->sent_ops.size() == 0);
+    assert(done == DATA_LEN);
+    for (int i = 0; i < DATA_LEN; i++)
+        assert(data[i] == 0x22);
+    assert(((uint8_t*)real->bitmap)[0] == (0xB0|0x22));
+    delete real;
+    msgr->stop_client(rcl->client_id);
+    msgr->stop_client(scl->client_id);
+    printf("[ok] send->receive round trip with a discarded reply, csum_status=%d\n", csum_status);
+}
+
+int main()
+{
+    setvbuf(stdout, NULL, _IONBF, 0);
+    auto msgr = new test_msgr_t();
+    test_discard_one_chunk(msgr);
+    test_discard_chunked(msgr);
+    test_stream_stays_in_sync(msgr, 0);
+    test_stream_stays_in_sync(msgr, MSGR_CSUM_PAYLOAD);
+    test_stream_stays_in_sync(msgr, MSGR_CSUM_FULL);
+    test_discard_list(msgr, 0);
+    test_discard_list(msgr, MSGR_CSUM_PAYLOAD);
+    test_discard_list(msgr, MSGR_CSUM_FULL);
+    test_discard_checks_csum(msgr, MSGR_CSUM_PAYLOAD);
+    test_discard_checks_csum(msgr, MSGR_CSUM_FULL);
+    test_roundtrip(msgr, 0);
+    test_roundtrip(msgr, MSGR_CSUM_PAYLOAD);
+    test_roundtrip(msgr, MSGR_CSUM_FULL);
+    delete msgr;
+    printf("All messenger tests passed\n");
+    return 0;
+}