From ca27b9191986d38bf2417662665bfb578cbb9e59 Mon Sep 17 00:00:00 2001 From: Vitaliy Filippov Date: Sun, 30 Nov 2025 14:49:59 +0300 Subject: [PATCH] Handle CAS writes with built-in SYNC and without retries in the client --- .gitea/workflows/test.yml | 18 +++++++ src/client/cluster_client.cpp | 88 ++++++++++++++++++++++++++++++----- src/client/cluster_client.h | 3 +- src/client/osd_ops.h | 5 +- src/kv/kv_db.cpp | 26 +---------- src/osd/osd_primary_write.cpp | 1 - tests/run_tests.sh | 1 + tests/test_kv_stress.sh | 1 - 8 files changed, 102 insertions(+), 41 deletions(-) diff --git a/.gitea/workflows/test.yml b/.gitea/workflows/test.yml index 4f07d940..dc2a7f58 100644 --- a/.gitea/workflows/test.yml +++ b/.gitea/workflows/test.yml @@ -558,6 +558,24 @@ jobs: echo "" done + test_kv_stress_imm: + runs-on: ubuntu-latest + needs: build + container: ${{env.TEST_IMAGE}}:${{github.sha}} + steps: + - name: Run test + id: test + timeout-minutes: 3 + run: IMMEDIATE_COMMIT=1 /root/vitastor/tests/test_kv_stress.sh + - name: Print logs + if: always() && steps.test.outcome == 'failure' + run: | + for i in /root/vitastor/testdata/*.log /root/vitastor/testdata/*.txt; do + echo "-------- $i --------" + cat $i + echo "" + done + test_splitbrain: runs-on: ubuntu-latest needs: build diff --git a/src/client/cluster_client.cpp b/src/client/cluster_client.cpp index d29b15a2..07bc7a29 100644 --- a/src/client/cluster_client.cpp +++ b/src/client/cluster_client.cpp @@ -765,8 +765,13 @@ void cluster_client_t::execute_internal(cluster_op_t *op) { return; } - if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && enable_writeback && !(op->flags & OP_FLUSH_BUFFER) && - !op->version /* no CAS writeback */) + // CAS writes are simplified: they're not cached, not resliced, not retried, and not part of the regular write queue at all + if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && op->version) + { + execute_cas(op); + return; + } + if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && enable_writeback && !(op->flags & OP_FLUSH_BUFFER)) { if (wb->writebacks_active >= client_max_writeback_iodepth) { @@ -788,7 +793,7 @@ void cluster_client_t::execute_internal(cluster_op_t *op) } if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && !(op->flags & OP_IMMEDIATE_COMMIT)) { - if (!(op->flags & OP_FLUSH_BUFFER) && !op->version /* no CAS write-repeat */) + if (!(op->flags & OP_FLUSH_BUFFER)) { uint64_t flush_id = ++wb->last_flush_id; wb->copy_write(op, CACHE_REPEATING, flush_id); @@ -847,6 +852,72 @@ void cluster_client_t::execute_internal(cluster_op_t *op) } } +void cluster_client_t::execute_cas(cluster_op_t *op) +{ + slice_rw(op); + op->needs_reslice = false; + if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && op->version && op->parts.size() > 1) + { + // Atomic writes to multiple stripes are unsupported + op->retval = -EINVAL; + auto cb = std::move(op->callback); + cb(op); + return; + } + int res = try_send(op, 0, [this, op](osd_op_t *part) + { + int expected = part->req.hdr.opcode == OSD_OP_DELETE ? 0 : part->req.rw.len; + op->retval = part->reply.hdr.retval; + op->retval = op->retval == expected ? 0 : (op->retval >= 0 ? -EIO : op->retval); + op->retval = op->retval == -EPIPE ? -EINTR : op->retval; + auto peer_it = msgr.osd_peer_fds.find(op->parts[0].osd_num); + if (op->retval != 0 || (op->flags & OP_IMMEDIATE_COMMIT)) + { + auto cb = std::move(op->callback); + cb(op); + } + else if (peer_it == msgr.osd_peer_fds.end()) + { + // Care must be taken to make sure that the client doesn't reconnect to the OSD + // before executing the previously completed operation callback (!) + op->retval = -EINTR; + auto cb = std::move(op->callback); + cb(op); + } + else + { + // CAS writes have a built-in sync + auto peer_fd = peer_it->second; + *part = (osd_op_t){ + .op_type = OSD_OP_OUT, + .peer_fd = peer_fd, + .req = { + .hdr = { + .magic = SECONDARY_OSD_OP_MAGIC, + .opcode = OSD_OP_SYNC, + }, + }, + .callback = [this, op](osd_op_t *part) + { + op->retval = part->reply.hdr.retval; + op->retval = op->retval == -EPIPE ? -EINTR : op->retval; + auto cb = std::move(op->callback); + cb(op); + }, + }; + msgr.outbox_push(part); + } + }); + if (res == TRY_SEND_CONNECTING || res == TRY_SEND_OFFLINE) + { + // In theory, CAS writes could wait for the PG to come up, but it's easier to just fail it + op->retval = -EINTR; + auto cb = std::move(op->callback); + cb(op); + return; + } +} + bool cluster_client_t::check_rw(cluster_op_t *op) { if (op->opcode == OSD_OP_SYNC) @@ -956,13 +1027,6 @@ resume_0: // Slice the operation into parts slice_rw(op); op->needs_reslice = false; - if ((op->opcode == OSD_OP_WRITE || op->opcode == OSD_OP_DELETE) && op->version && op->parts.size() > 1) - { - // Atomic writes to multiple stripes are unsupported - op->retval = -EINVAL; - erase_op(op); - return 1; - } resume_1: // Send unsent parts, if they're not subject to change op->state = 2; @@ -1307,7 +1371,7 @@ bool cluster_client_t::affects_osd(uint64_t inode, uint64_t offset, uint64_t len return false; } -int cluster_client_t::try_send(cluster_op_t *op, int i) +int cluster_client_t::try_send(cluster_op_t *op, int i, std::function cb) { if (!msgr_initialized) { @@ -1367,7 +1431,7 @@ int cluster_client_t::try_send(cluster_op_t *op, int i) ? (uint8_t*)op->part_bitmaps + pg_bitmap_size*i : NULL), .bitmap_len = (unsigned)(op->opcode == OSD_OP_READ || op->opcode == OSD_OP_READ_BITMAP || op->opcode == OSD_OP_READ_CHAIN_BITMAP ? pg_bitmap_size : 0), - .callback = [this, part](osd_op_t *op_part) + .callback = cb ? cb : [this, part](osd_op_t *op_part) { handle_op_part(part); }, diff --git a/src/client/cluster_client.h b/src/client/cluster_client.h index 74fce18a..23a8b05c 100644 --- a/src/client/cluster_client.h +++ b/src/client/cluster_client.h @@ -175,12 +175,13 @@ protected: void on_change_node_placement_hook(); void execute_internal(cluster_op_t *op); + void execute_cas(cluster_op_t *op); void unshift_op(cluster_op_t *op); int continue_rw(cluster_op_t *op); bool check_rw(cluster_op_t *op); void slice_rw(cluster_op_t *op); void reset_retry_timer(int new_duration); - int try_send(cluster_op_t *op, int i); + int try_send(cluster_op_t *op, int i, std::function cb = nullptr); 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); diff --git a/src/client/osd_ops.h b/src/client/osd_ops.h index d93b9b27..8840a65b 100644 --- a/src/client/osd_ops.h +++ b/src/client/osd_ops.h @@ -262,7 +262,10 @@ struct __attribute__((__packed__)) osd_reply_del_t uint32_t left_on_dead_count; }; -// sync to the primary OSD +// sync to the primary OSD. semantics: +// 1) any non-synced write may disappear on OSD restart. even if it's a CAS write. +// 2) sync only guarantees to commit completed writes, not in-progress ones. +// 3) sync is a no-op when immediate_commit is active. struct __attribute__((__packed__)) osd_op_sync_t { osd_op_header_t header; diff --git a/src/kv/kv_db.cpp b/src/kv/kv_db.cpp index 959a136e..c2b25c70 100644 --- a/src/kv/kv_db.cpp +++ b/src/kv/kv_db.cpp @@ -139,7 +139,6 @@ struct kv_db_t uint64_t next_free = 0; uint32_t kv_block_size = 0; uint32_t ino_block_size = 0; - bool immediate_commit = false; uint64_t memory_limit = 128*1024*1024; uint64_t evict_unused_age = 1000; uint64_t evict_max_misses = 10; @@ -529,13 +528,6 @@ void kv_db_t::open(inode_t inode_id, json11::Json cfg, std::function return; } this->inode_id = inode_id; - this->immediate_commit = cli->get_immediate_commit(inode_id); - if (!this->immediate_commit) - { - // FIXME: CAS is unusable without immediate_commit at the moment. Fix it - cb(-EINVAL); - return; - } this->ino_block_size = pool_cfg.data_block_size * pg_data_size; this->kv_block_size = kv_block_size; this->next_free = 0; @@ -695,7 +687,6 @@ void kv_db_t::close(std::function cb) next_free = 0; kv_block_size = 0; ino_block_size = 0; - immediate_commit = false; block_cache.clear(); known_versions.clear(); cb(); @@ -1404,22 +1395,7 @@ static void write_block(kv_db_t *db, kv_block_t *blk, std::function c } } delete op; - if (res < 0 || db->immediate_commit) - { - cb(res); - } - else - { - op = new cluster_op_t; - op->opcode = OSD_OP_SYNC; - op->callback = [cb](cluster_op_t *op) - { - auto res = op->retval; - delete op; - cb(res); - }; - db->cli->execute(op); - } + cb(res); }; db->cli->execute(op); } diff --git a/src/osd/osd_primary_write.cpp b/src/osd/osd_primary_write.cpp index d6c0ba8b..b75a2188 100644 --- a/src/osd/osd_primary_write.cpp +++ b/src/osd/osd_primary_write.cpp @@ -165,7 +165,6 @@ resume_3: return; } // Check CAS version - // FIXME: Handle CAS writes as "immediate" in non-immediate_commit pools, otherwise CAS doesn't make sense if (cur_op->req.rw.version && op_data->fact_ver != (cur_op->req.rw.version-1)) { deref_object_state(pg, &op_data->object_state, true); diff --git a/tests/run_tests.sh b/tests/run_tests.sh index 86009d23..d59c7bc0 100755 --- a/tests/run_tests.sh +++ b/tests/run_tests.sh @@ -49,6 +49,7 @@ SCHEME=ec ./test_snapshot_chain.sh SCHEME=ec ./test_snapshot_down.sh ./test_kv_stress.sh +IMMEDIATE_COMMIT=1 ./test_kv_stress.sh ./test_splitbrain.sh diff --git a/tests/test_kv_stress.sh b/tests/test_kv_stress.sh index 8d23afc4..efc06aa8 100755 --- a/tests/test_kv_stress.sh +++ b/tests/test_kv_stress.sh @@ -1,6 +1,5 @@ #!/bin/bash -ex -IMMEDIATE_COMMIT=1 PG_COUNT=16 . `dirname $0`/run_3osds.sh