Support canceling blockstore v1 operations (v2 doesn't need it)

src/blockstore/blockstore.h

@@ -172,6 +172,7 @@ struct __attribute__ ((visibility("default"))) blockstore_op_t
     uint8_t *buf = NULL;
     uint8_t *bitmap = NULL;
     int64_t retval = 0;
+    bool canceled = false;
 
     uint8_t private_data[BS_OP_PRIVATE_DATA_SIZE];
 };

src/blockstore/v1/read.cpp

@@ -447,6 +447,12 @@ bool blockstore_impl_t::read_checksum_block(blockstore_op_t *op, int rv_pos, uin
 
 int blockstore_impl_t::dequeue_read(blockstore_op_t *read_op)
 {
+    if (read_op->canceled)
+    {
+        read_op->retval = -EPIPE;
+        FINISH_OP(read_op);
+        return 2;
+    }
     auto & clean_db = clean_db_shard(read_op->oid);
     auto clean_it = clean_db.find(read_op->oid);
     auto dirty_it = dirty_db.upper_bound((obj_ver_id){

src/blockstore/v1/rollback.cpp

@@ -12,6 +12,12 @@ int blockstore_impl_t::dequeue_rollback(blockstore_op_t *op)
     {
         return continue_rollback(op);
     }
+    if (op->canceled)
+    {
+        op->retval = -EPIPE;
+        FINISH_OP(op);
+        return 2;
+    }
     int r = split_stab_op(op, [this](obj_ver_id ov)
     {
         // Check that there are some versions greater than v->version (which may be zero),

src/blockstore/v1/stable.cpp

@@ -301,6 +301,12 @@ int blockstore_impl_t::dequeue_stable(blockstore_op_t *op)
     {
         return continue_stable(op);
     }
+    if (op->canceled)
+    {
+        op->retval = -EPIPE;
+        FINISH_OP(op);
+        return 2;
+    }
     int r = split_stab_op(op, [this](obj_ver_id ov)
     {
         auto dirty_it = dirty_db.find(ov);

src/blockstore/v1/write.cpp

@@ -329,6 +329,11 @@ void blockstore_impl_t::unshift_stable_flush(obj_ver_id ov)
     }
 }
 
+// Why cancel all writes:
+// 1) enqueue_write forms dirty_db entries based on one another.
+//    So if we remove an entry, newer entries become possibly incorrect.
+// 2) small writes may wait for the previous big write. if we cancel it,
+//    we also have to unblock them, but it's easier to cancel them too.
 void blockstore_impl_t::cancel_all_writes(blockstore_op_t *op, blockstore_dirty_db_t::iterator dirty_it, int retval)
 {
     while (dirty_it != dirty_db.end() && dirty_it->first.oid == op->oid)
@@ -351,9 +356,7 @@ void blockstore_impl_t::cancel_all_writes(blockstore_op_t *op, blockstore_dirty_
         else if (found && other_op->oid == op->oid &&
             (other_op->opcode == BS_OP_WRITE || other_op->opcode == BS_OP_WRITE_STABLE))
         {
-            // Mark operations to cancel them
-            PRIV(other_op)->real_version = UINT64_MAX;
-            other_op->retval = retval;
+            other_op->canceled = true;
         }
     }
     op->retval = retval;
@@ -372,6 +375,11 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
         .version = op->version,
     });
     assert(dirty_it != dirty_db.end());
+    if (op->canceled)
+    {
+        cancel_all_writes(op, dirty_it, -EPIPE);
+        return 2;
+    }
     if ((dirty_it->second.state & BS_ST_WORKFLOW_MASK) < BS_ST_IN_FLIGHT)
     {
         // Don't dequeue
@@ -379,12 +387,6 @@ int blockstore_impl_t::dequeue_write(blockstore_op_t *op)
     }
     if (PRIV(op)->real_version != 0)
     {
-        if (PRIV(op)->real_version == UINT64_MAX)
-        {
-            // This is the flag value used to cancel operations
-            FINISH_OP(op);
-            return 2;
-        }
         // Restore original low version number for unblocked operations
 #ifdef BLOCKSTORE_DEBUG
         printf("Restoring %jx:%jx version: v%ju -> v%ju\n", op->oid.inode, op->oid.stripe, op->version, PRIV(op)->real_version);
@@ -683,6 +685,16 @@ resume_2:
             .version = op->version,
         });
         assert(dirty_it != dirty_db.end());
+        if (op->canceled)
+        {
+            auto uo_it = used_clean_objects.find(dirty_it->second.location);
+            if (uo_it != used_clean_objects.end())
+                uo_it->second.was_freed = true;
+            else
+                data_alloc->set(dirty_it->second.location / dsk.data_block_size, false);
+            cancel_all_writes(op, dirty_it, -EPIPE);
+            return 2;
+        }
         uint64_t dyn_size = dsk.dirty_dyn_size(op->offset, op->len);
         blockstore_journal_check_t space_check(this);
         if (!space_check.check_available(op, 1, sizeof(journal_entry_big_write) + dyn_size,
@@ -920,6 +932,11 @@ int blockstore_impl_t::dequeue_del(blockstore_op_t *op)
         .version = op->version,
     });
     assert(dirty_it != dirty_db.end());
+    if (op->canceled)
+    {
+        cancel_all_writes(op, dirty_it, -EPIPE);
+        return 2;
+    }
     if (!forget_unstable_before_delete(op, dirty_it))
     {
         // A preceding write is still running, retry when it finishes

src/client/messenger.h

@@ -303,6 +303,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;
@@ -343,6 +344,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

@@ -188,6 +188,7 @@ struct __attribute__((visibility("default"))) osd_op_t
     std::shared_ptr<osd_op_enc_t> enc;
     uint64_t csum = 0; // network layer checksum
     osd_primary_op_data_t* op_data = NULL;
+    bool canceled = false;
     std::function<void(osd_op_t*)> callback;
 
     osd_op_buf_list_t iov;

src/client/msgr_receive.cpp

@@ -570,11 +570,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;
+        }
     }
 }
 

src/client/msgr_send.cpp

@@ -494,6 +494,107 @@ bool osd_messenger_t::wakeup_send(osd_client_t *cl)
     return true;
 }
 
+// Try to cancel an outbound operation
+// Returns true when successfully canceled, returns false if the reply is already received
+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;
+    }
+    for (auto op: set_immediate_ops)
+    {
+        if (op == cur_op)
+        {
+            // Reply is already received, callback is pending
+            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 just queued and not sent and can be canceled easily
+            cl->sent_ops.erase(sent_it);
+            cl->write_ops.erase(wr_it);
+            cur_op->callback = NULL;
+            return true;
+        }
+    }
+    // Not in queue: sent or in progress of sending.
+    bool in_progress = (cl->write_op == cur_op);
+    if (!in_progress)
+    {
+        for (osd_op_t *op: cl->send_free_ops)
+        {
+            if (op == cur_op)
+            {
+                in_progress = true;
+                break;
+            }
+        }
+    }
+    if (!in_progress)
+    {
+        for (osd_op_t *op: cl->zc_free_list)
+        {
+            if (op == cur_op)
+            {
+                in_progress = true;
+                break;
+            }
+        }
+    }
+    sent_it->second = copy_out_op_to_ignore(cur_op);
+    cur_op->callback = NULL;
+    if (in_progress)
+    {
+        // We CAN cancel it, but only when the reference from send_free_ops/zc_free_list is cleared
+        cur_op->canceled = true;
+        return false;
+    }
+    // Make a copy to have some space to place the reply and forget the original object.
+    return true;
+}
+
+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->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);
+    if (op_copy->req.hdr.opcode == OSD_OP_SEC_READ)
+    {
+        // will be freed
+        op_copy->buf = (uint8_t*)malloc_or_die(op_copy->bitmap_len + op_copy->req.sec_rw.len);
+        // won't be freed
+        op_copy->iov.push_back(op_copy->buf, op_copy->req.sec_rw.len);
+        op_copy->bitmap = op_copy->buf + op_copy->req.sec_rw.len;
+    }
+    else if (op_copy->req.hdr.opcode == OSD_OP_READ)
+    {
+        // will be freed
+        op_copy->buf = (uint8_t*)malloc_or_die(op_copy->bitmap_len + op_copy->req.rw.len);
+        // won't be freed
+        op_copy->iov.push_back(op_copy->buf, op_copy->req.rw.len);
+        op_copy->bitmap = op_copy->buf + op_copy->req.rw.len;
+    }
+    return op_copy;
+}
+
 bool osd_messenger_t::try_send(osd_client_t *cl)
 {
     assert(cl->peer_state != PEER_RDMA);
@@ -674,9 +775,9 @@ size_t osd_messenger_t::copy_ops_to_with(osd_client_t *cl, uint8_t *dst, size_t
                 return 0;
             break;
         }
-        if (!cl->write_op && op->op_type == OSD_OP_IN)
+        if (!cl->write_op)
         {
-            // this is a reply, free the op after sending it
+            // remember the operation for the current send-list
             cl->send_free_ops.push_back(op);
         }
     }

src/client/msgr_stop.cpp

@@ -49,7 +49,15 @@ void osd_messenger_t::post_send_free(osd_op_t *op)
     if (!op)
         return;
     else if (!((size_t)op & 7))
-        delete op;
+    {
+        if (op->op_type == OSD_OP_IN)
+        {
+            // this is a reply, free the op after sending it
+            delete op;
+        }
+        else if (op->canceled)
+            op->cancel();
+    }
     else
         free((void*)((size_t)op & ~(size_t)7));
 }