refactor flush data worker - #470
Merged
Merged
Conversation
|
Caution Review failedThe pull request is closed. ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (23)
WalkthroughIntroduces coroutine-aware cooperative-yield hooks across datastore, flush and sync layers; refactors CC map key-count tracking; moves heap initialization to background threads and makes it thread-safe; switches RecoveryService inbound connection storage to heap-owned pointers; and adds Boost.Context linking (ASan-aware) to the build. Changes
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related issues
Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ❌ 3❌ Failed checks (2 warnings, 1 inconclusive)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
There was a problem hiding this comment.
Pull request overview
This PR refactors the “flush data worker” pipeline to support coroutine-friendly yield/resume behavior during flush-related datastore operations, and restructures internal buffering/queueing to better isolate work per worker. It also updates several datastore handler interfaces to accept optional coroutine callbacks and adjusts some supporting infrastructure (replay connection storage, delta-size estimation, build linking).
Changes:
- Refactor
LocalCcShardsflush workflow: per-data-sync-worker flush buffers, per-flush-worker queues, coroutine-based yielding/resuming, and updated flush/post-process call signatures. - Extend
DataStoreHandlerAPIs (and implementations) to accept optional yield/resume/sync-yield callbacks; propagate through RocksDB and datastore-service client codepaths. - Simplify hash-partition delta-size estimation by tracking CC-map key/dirty-key stats and returning those directly to the delta-size request.
Reviewed changes
Copilot reviewed 24 out of 24 changed files in this pull request and generated 7 comments.
Show a summary per file
| File | Description |
|---|---|
| tx_service/src/tx_index_operation.cpp | Adds additional logging on cluster-config unlock failure (but needs null-safety). |
| tx_service/src/fault/log_replay_service.cpp | Updates inbound connection storage to unique_ptr and adds safer lookup handling. |
| tx_service/src/checkpointer.cpp | Updates test helpers to new datastore handler signatures. |
| tx_service/src/cc/local_cc_shards.cpp | Main flush-worker refactor: per-worker buffers/queues and coroutine-based yield/resume plumbing. |
| tx_service/include/store/int_mem_store.h | Updates interface overrides for new callback parameters (no-op placeholders). |
| tx_service/include/store/data_store_handler.h | Extends datastore handler interface with optional coroutine callbacks. |
| tx_service/include/range_record.h | Adds an explicit empty destructor to TemplateTableRangeEntry (should be defaulted/removed). |
| tx_service/include/fault/log_replay_service.h | Converts inbound connection map to unique_ptr for pointer stability; adds <memory>. |
| tx_service/include/cc/template_cc_map.h | Replaces scan-based delta-size counting with tracked key/dirty-key stats. |
| tx_service/include/cc/range_slice.h | Adds an explicit empty destructor to TemplateStoreRange (should be defaulted/removed). |
| tx_service/include/cc/range_cc_map.h | Switches to AbortCcRequest() for leader-term failure handling. |
| tx_service/include/cc/local_cc_shards.h | Adds coroutine stack allocator, per-worker queues/buffers, and background heap init thread declarations. |
| tx_service/include/cc/cc_request.h | Changes ScanDeltaSizeCcForHashPartition to use key/dirty-key stats rather than incremental scan state. |
| tx_service/include/cc/cc_req_misc.h | Adds yield/resume-aware wait support for CC waitables and update requests. |
| tx_service/include/cc/catalog_cc_map.h | Standardizes abort/error handling and return semantics for replay execution. |
| store_handler/rocksdb_handler.h | Updates handler signatures for new callback parameters. |
| store_handler/rocksdb_handler.cpp | Adds coroutine-aware offload path for PutAll/PersistKV using a sync callback helper. |
| store_handler/eloq_data_store_service/eloq_store_data_store.cpp | Optimizes request building by reserving and pre-sizing entry buffers. |
| store_handler/data_store_service_client_closure.h | Adds yield/resume support to sync callback/coordinator helpers. |
| store_handler/data_store_service_client_closure.cpp | Implements new wait helpers (Wait, WaitForAll, capacity waits) used by client. |
| store_handler/data_store_service_client.h | Updates client signatures and adds internal *Impl helpers for callback-aware implementations. |
| store_handler/data_store_service_client.cpp | Propagates yield/resume/sync-yield into flush/update paths and refactors concurrency waits. |
| store_handler/bigtable_handler.cpp | Preserves/restore mimalloc override thread ID when using table-ranges heap. |
| build_tx_service.cmake | Adds Boost.Context discovery/linking for the coroutine-based flush worker changes. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+545
to
+547
| ~TemplateTableRangeEntry() | ||
| { | ||
| } |
There was a problem hiding this comment.
This explicitly-defined empty destructor makes TemplateTableRangeEntry a user-declared-destructor type, which suppresses implicit move operations and can change type traits (e.g., triviality) compared to a defaulted destructor. If no custom cleanup is needed, prefer ~TemplateTableRangeEntry() = default; (or omit the destructor entirely).
Suggested change
| ~TemplateTableRangeEntry() | |
| { | |
| } | |
| ~TemplateTableRangeEntry() = default; |
Comment on lines
893
to
896
| ~TemplateStoreRange() | ||
| { | ||
| } | ||
|
|
There was a problem hiding this comment.
TemplateStoreRange previously had a defaulted destructor; switching to an explicitly-defined empty destructor suppresses implicit move operations and may affect type traits/optimizations. If no special teardown is required, prefer ~TemplateStoreRange() = default; (or omit it).
Suggested change
| ~TemplateStoreRange() | |
| { | |
| } |
Comment on lines
+122
to
+124
| ~RecoveryService() | ||
| { | ||
| } |
There was a problem hiding this comment.
~RecoveryService() is changed from = default to an empty user-provided destructor. A user-declared destructor suppresses implicit move operations and can change generated special members. If there’s no custom destruction logic, prefer ~RecoveryService() = default; to preserve the default semantics.
Suggested change
| ~RecoveryService() | |
| { | |
| } | |
| ~RecoveryService() = default; |
Comment on lines
+93
to
+96
| ~CoroCtx() | ||
| { | ||
| } | ||
|
|
There was a problem hiding this comment.
CoroCtx has an explicitly-defined empty destructor. Unless you need custom teardown, prefer = default (or omit it) to avoid suppressing implicit move operations and to keep the type as simple as possible.
Suggested change
| ~CoroCtx() | |
| { | |
| } |
Comment on lines
+475
to
+488
| LOG(INFO) | ||
| << "Alter Table Index transaction unlock cluster config " | ||
| "lock failed, txn: " | ||
| << txm->TxNumber() << ", error code: " | ||
| << (int) unlock_cluster_config_op_.hd_result_.ErrorCode() | ||
| << ", error message: " | ||
| << unlock_cluster_config_op_.hd_result_.ErrorMsg() | ||
| << ", cluster config addr term" | ||
| << cluster_config_addr->Term() | ||
| << ", cluster config addr node group id: " | ||
| << cluster_config_addr->NodeGroupId() | ||
| << ", node group term: " | ||
| << Sharder::Instance().LeaderTerm( | ||
| cluster_config_addr->NodeGroupId()); |
There was a problem hiding this comment.
cluster_config_addr can remain null if the cluster-config entry is not present in MetaDataReadSet(), but it is unconditionally dereferenced in the error log (cluster_config_addr->Term() / NodeGroupId()). This would crash in release builds (asserts are compiled out). Please guard the dereference (or reuse the earlier assert(cluster_config_addr && ...) pattern) and log a fallback when the address can't be found.
Suggested change
| LOG(INFO) | |
| << "Alter Table Index transaction unlock cluster config " | |
| "lock failed, txn: " | |
| << txm->TxNumber() << ", error code: " | |
| << (int) unlock_cluster_config_op_.hd_result_.ErrorCode() | |
| << ", error message: " | |
| << unlock_cluster_config_op_.hd_result_.ErrorMsg() | |
| << ", cluster config addr term" | |
| << cluster_config_addr->Term() | |
| << ", cluster config addr node group id: " | |
| << cluster_config_addr->NodeGroupId() | |
| << ", node group term: " | |
| << Sharder::Instance().LeaderTerm( | |
| cluster_config_addr->NodeGroupId()); | |
| if (cluster_config_addr) | |
| { | |
| LOG(INFO) | |
| << "Alter Table Index transaction unlock cluster " | |
| "config " | |
| "lock failed, txn: " | |
| << txm->TxNumber() | |
| << ", error code: " | |
| << (int) unlock_cluster_config_op_.hd_result_ | |
| .ErrorCode() | |
| << ", error message: " | |
| << unlock_cluster_config_op_.hd_result_.ErrorMsg() | |
| << ", cluster config addr term" | |
| << cluster_config_addr->Term() | |
| << ", cluster config addr node group id: " | |
| << cluster_config_addr->NodeGroupId() | |
| << ", node group term: " | |
| << Sharder::Instance().LeaderTerm( | |
| cluster_config_addr->NodeGroupId()); | |
| } | |
| else | |
| { | |
| LOG(INFO) | |
| << "Alter Table Index transaction unlock cluster " | |
| "config " | |
| "lock failed, txn: " | |
| << txm->TxNumber() | |
| << ", error code: " | |
| << (int) unlock_cluster_config_op_.hd_result_ | |
| .ErrorCode() | |
| << ", error message: " | |
| << unlock_cluster_config_op_.hd_result_.ErrorMsg() | |
| << ", cluster config addr not found in " | |
| "MetaDataReadSet()"; | |
| } |
Comment on lines
5856
to
+5862
| // Could not merge, wait if queue is full | ||
| while (pending_flush_work_.size() >= | ||
| static_cast<size_t>(flush_data_worker_ctx_.worker_num_)) | ||
| /* | ||
| while (pending_flush_work.size() >= 2) | ||
| { | ||
| flush_data_worker_ctx_.cv_.wait(worker_lk); | ||
| } | ||
| */ |
There was a problem hiding this comment.
There is a commented-out backpressure loop for pending_flush_work (the while (pending_flush_work.size() >= 2) block). Leaving disabled flow-control code in-place makes it unclear whether queue growth is intentionally unbounded or an unfinished change. Please either remove this block entirely or reintroduce an explicit, documented queue bound/backpressure strategy.
Comment on lines
+5908
to
+5912
| bool LocalCcShards::ShouldYieldFlushData(size_t worker_idx) | ||
| { | ||
| // Retrieve first pending work and pop it (FIFO). | ||
| std::unique_ptr<FlushDataTask> cur_work = | ||
| std::move(pending_flush_work_.front()); | ||
| pending_flush_work_.pop_front(); | ||
|
|
||
| // Notify any threads waiting for queue space | ||
| flush_data_worker_ctx_.cv_.notify_all(); | ||
|
|
||
| flush_worker_lk.unlock(); | ||
| std::lock_guard<std::mutex> lk(flush_data_worker_ctx_.mux_); | ||
| return !pending_flush_work_[worker_idx].empty(); | ||
| } |
There was a problem hiding this comment.
ShouldYieldFlushData() is added but not used anywhere in this file (or elsewhere). Dead code like this increases maintenance burden and can mislead future readers about intended scheduling/yield behavior. Please either wire it into the coroutine-yield logic or remove it.
lokax
force-pushed
the
yf-pro-ckpt-boost-context
branch
from
March 23, 2026 06:28
cdf9e92 to
5d199a7
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 9
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (5)
tx_service/include/cc/cc_req_misc.h (1)
871-879:⚠️ Potential issue | 🟠 MajorClear the new callback state in
WaitableCc::Reset().
Reset()now leavesyield_fn_,resume_fn_, andwaiting_from the previous run intact. A recycled request can carry a stale coroutine wakeup path into the next blockingWait()and hang once the completion side stops taking the condvar path.♻️ Suggested fix
void Reset(std::function<bool(CcShard &ccs)> task = {}, uint16_t core_cnt = 1) { std::lock_guard<bthread::Mutex> lk(mux_); RunOnTxProcessorCc::Reset(std::move(task)); unfinished_cnt_ = core_cnt; error_code_ = CcErrorCode::NO_ERROR; + yield_fn_ = nullptr; + resume_fn_ = nullptr; + waiting_.store(false, std::memory_order_relaxed); }Also applies to: 997-1001
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tx_service/include/cc/cc_req_misc.h` around lines 871 - 879, The Reset implementation for WaitableCc (the one calling RunOnTxProcessorCc::Reset and setting unfinished_cnt_ and error_code_) must also clear the previous callback/wait state to avoid carrying stale coroutine wakeup paths: explicitly reset yield_fn_ and resume_fn_ to empty/null and set waiting_ to false so any recycled request cannot resume a prior Wait; apply the same clearing changes in the other Reset overload (the similar block around the other Reset that mirrors these fields).store_handler/data_store_service_client_closure.h (1)
101-151:⚠️ Potential issue | 🔴 CriticalThe race fix stops at the base class.
SyncCallbackDatanow has a properfinished_ + waiting_handshake, but the coroutine-enabled subclasses later in this file still overrideWait()/Notify()as a one-shotyield/resumepair (FetchTableCallbackData,FetchDatabaseCallbackData,FetchAllDatabaseCallbackData,UpsertDatabaseCallbackData,DropDatabaseCallbackData, andDiscoverAllTableNamesCallbackData). If one of those callbacks fires before the waiter actually yields, the wakeup is lost and the caller can hang indefinitely. Please migrate those subclasses to the same handshake or drop the overrides.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@store_handler/data_store_service_client_closure.h` around lines 101 - 151, The coroutine-enabled subclasses (FetchTableCallbackData, FetchDatabaseCallbackData, FetchAllDatabaseCallbackData, UpsertDatabaseCallbackData, DropDatabaseCallbackData, DiscoverAllTableNamesCallbackData) still implement one-shot overrides of Wait()/Notify() which lose a wakeup if Notify() happens before the waiter yields; update each subclass to use the same finished_ + waiting_ handshake used by the base SyncCallbackData: remove or refactor their custom Wait()/Notify() overrides to call the base implementations (or replicate the waiting_.store/finished_ checks and memory_order semantics used in the base SetCoroCallbacks/Wait/Notify), ensure SetCoroCallbacks is used to store the yield/resume pointers, and preserve unlocking before invoking resume_fn_ so the resume path follows the base-class pattern to avoid lost wakeups.tx_service/include/cc/local_cc_shards.h (2)
474-494:⚠️ Potential issue | 🟡 MinorKeep the allocator-specific init inside the once-only block.
hash_partition_ckpt_heap_is guarded now, but theWITH_JEMALLOCbranch still runs on every call. A second call will overwritehash_partition_ckpt_arena_id_, so this initializer is only half-idempotent.♻️ Proposed fix
void InitializeHashPartitionCkptHeap() { std::unique_lock<std::mutex> lk(hash_partition_ckpt_heap_mux_); if (!hash_partition_ckpt_heap_) { hash_partition_main_thread_id_ = mi_thread_id(); hash_partition_ckpt_heap_ = mi_heap_new(); - } - -#if defined(WITH_JEMALLOC) - // create hash partition ckpt arena - size_t sz = sizeof(uint32_t); - if (mallctl("arenas.create", - &hash_partition_ckpt_arena_id_, - &sz, - NULL, - 0) != 0) - { - LOG(FATAL) << "Failed to create jemalloc arena for hash part heap"; - } -#endif + `#if` defined(WITH_JEMALLOC) + // create hash partition ckpt arena once, together with the heap + size_t sz = sizeof(uint32_t); + if (mallctl("arenas.create", + &hash_partition_ckpt_arena_id_, + &sz, + NULL, + 0) != 0) + { + LOG(FATAL) << "Failed to create jemalloc arena for hash part heap"; + } + `#endif` + } }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tx_service/include/cc/local_cc_shards.h` around lines 474 - 494, The WITH_JEMALLOC initialization runs unconditionally and can overwrite hash_partition_ckpt_arena_id_ on subsequent calls; move the entire jemalloc branch (mallctl creating the arena and setting hash_partition_ckpt_arena_id_) inside the guarded block that checks/creates hash_partition_ckpt_heap_ within InitializeHashPartitionCkptHeap(), so the jemalloc arena is created only once alongside mi_heap_new and hash_partition_main_thread_id_, referencing symbols InitializeHashPartitionCkptHeap, hash_partition_ckpt_heap_, hash_partition_ckpt_arena_id_, hash_partition_main_thread_id_, mi_heap_new and mallctl to locate the correct code to relocate.
2015-2023:⚠️ Potential issue | 🔴 CriticalVerify the lifetime safety of callback pointers in async data store operations.
The pointers
yield_fptr/resume_fptr/sync_yield_fptrare passed toUpdateRangeSlices()and stored as member variables in pooled objects (SyncPutAllData,SyncConcurrentRequest,SyncCallbackData). These objects store raw pointers to caller-owned stack-localstd::functionobjects and invoke them from completion callbacks. WhilePutAllImplcurrently synchronously waits for all partitions to complete viasync_putall->Wait()before returning, this pattern is fragile: the pooled objects can outlive the caller's scope if callback timing changes or async execution extends beyond the current synchronous boundaries. The safer approach is to either copy the callback functions into the pooled objects or usestd::shared_ptrto guarantee lifetime, eliminating the risk of use-after-free if execution contexts evolve.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tx_service/include/cc/local_cc_shards.h` around lines 2015 - 2023, The callbacks yield_fptr/resume_fptr/sync_yield_fptr passed into PostProcessFlushTaskEntries (and forwarded into UpdateRangeSlices/PutAllImpl) are currently stored as raw pointers inside pooled types SyncPutAllData, SyncConcurrentRequest and SyncCallbackData, creating a lifetime/USE-AFTER-FREE risk; modify those pooled objects and all call sites so they take ownership of the callbacks (either by copying the std::function into a member std::function or by storing a std::shared_ptr<std::function<...>>), update constructors/factory methods that create SyncPutAllData/SyncConcurrentRequest/SyncCallbackData to accept the owning form, and ensure callers to PostProcessFlushTaskEntries/UpdateRangeSlices/PutAllImpl pass owned callbacks (not pointers) so completion callbacks can safely invoke them after async/pooled lifetimes.store_handler/data_store_service_client.cpp (1)
3120-3150:⚠️ Potential issue | 🔴 CriticalIncrement the in-flight counter before dispatching the last archive batch.
The final
BatchWriteRecords()is issued beforeunfinished_request_cnt_is incremented. A fast completion can run the callback first, decrement from the wrong state, and leaveWaitForAll()hanging or observing the wrong completion state. It also lets the last request bypass the normal capacity gate by one. UseWaitForCapacityAndIncrement()here too, then drop the manual increment block.🐛 Proposed fix
// Send out the last batch of this partition if (keys.size() > 0) { + sync_concurrent->WaitForCapacityAndIncrement(); BatchWriteRecords(kv_mvcc_archive_name, partition_id, data_shard_id, std::move(keys), std::move(records), @@ sync_concurrent, SyncConcurrentRequestCallback, parts_cnt_per_key, parts_cnt_per_record); - keys.clear(); - records.clear(); - records_ts.clear(); - records_ttl.clear(); - op_types.clear(); - - keys.reserve(recs_cnt * parts_cnt_per_key); - records.reserve(recs_cnt * parts_cnt_per_record); - records_ts.reserve(recs_cnt); - records_ttl.reserve(recs_cnt); - op_types.reserve(recs_cnt); - write_batch_size = 0; - { - std::unique_lock<bthread::Mutex> lk(sync_concurrent->mux_); - sync_concurrent->unfinished_request_cnt_++; - } }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@store_handler/data_store_service_client.cpp` around lines 3120 - 3150, The final BatchWriteRecords call dispatches before incrementing sync_concurrent->unfinished_request_cnt_, risking a race where the callback decrements a not-yet-incremented counter; replace the manual increment block before that dispatch with the same capacity-guarded increment used elsewhere by calling WaitForCapacityAndIncrement(sync_concurrent) (or the existing wrapper used in this file) so the in-flight counter is incremented before BatchWriteRecords(kv_mvcc_archive_name, ...) is invoked; remove the explicit unique_lock and unfinished_request_cnt_++ block and ensure SyncConcurrentRequestCallback and WaitForAll() semantics remain unchanged.
🧹 Nitpick comments (4)
tx_service/include/cc/template_cc_map.h (1)
8125-8147: Add a cheap invariant for the cached counters.The fast path at Lines 7984-7988 now trusts these fields outright. A local
dirty <= totalassertion will fail fast if any future mutation path misses the bookkeeping.🛡️ Suggested assertion
if (dirty_delta != 0) { assert(dirty_delta >= 0 || dirty_data_key_count_ >= static_cast<size_t>(-dirty_delta)); dirty_data_key_count_ = static_cast<size_t>( static_cast<int64_t>(dirty_data_key_count_) + dirty_delta); } + + assert(dirty_data_key_count_ <= data_key_count_); }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tx_service/include/cc/template_cc_map.h` around lines 8125 - 8147, AdjustDataKeyStats currently updates data_key_count_ and dirty_data_key_count_ but doesn't validate the invariant dirty_data_key_count_ <= data_key_count_; add a cheap assertion at the end of AdjustDataKeyStats (after the existing updates and after the early return for table_name_.IsMeta()) that checks dirty_data_key_count_ is never greater than data_key_count_ (taking care of the size_t vs signed conversions) so future missed bookkeeping fails fast; reference AdjustDataKeyStats, data_key_count_, dirty_data_key_count_, and table_name_.IsMeta() to locate where to insert the assertion.store_handler/rocksdb_handler.cpp (1)
526-653: Pull the RocksDB work into helpers.The coroutine branches now re-copy the synchronous
PutAll/PersistKVbodies. That makes TTL/delete handling, metrics, and error paths easy to fix in one path and miss in the other. Please keep these branches toSubmitWork(...) + Wait(...)and move the actual RocksDB work into shared internal helpers.Also applies to: 764-808
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@store_handler/rocksdb_handler.cpp` around lines 526 - 653, The coroutine branch duplicates the full RocksDB write loop (inside the SubmitWork lambda) causing TTL/delete handling, metrics, and error paths to diverge from the synchronous PutAll/PersistKV code; refactor by extracting the core write logic into a shared helper (e.g., a new method like ExecutePutAllBatch or PersistBatchToRocksDB) that accepts the DB pointer (GetDBPtr()), write options, the batch structure, batch_write_size_, and a callback/result indicator, and then have both the coroutine path (the lambda submitted via query_worker_pool_->SubmitWork and using RocksDBWriteSyncCallback/Wait) and the original synchronous path call that helper so all TTL checks (EncodeToKvKey, SerializeFlushRecord, RecordStatus/TTL logic), metrics::kv_meter collection, error logging (status.ToString()/status.code()), and write_batch management live in one place; ensure the helper returns success/failure so the coroutine lambda invokes callback.Notify(...) accordingly and the synchronous callers propagate the same result.tx_service/include/cc/local_cc_shards.h (1)
2546-2551: Make the worker routing explicit inAddFlushTaskEntry().The current API derives worker identity implicitly from
entry->data_sync_task_->id_, obscuring buffer and queue selection once data-sync and flush worker counts diverge. Pass the source worker id or resolved flush worker id as an explicit parameter to keep the mapping visible and testable at call sites.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tx_service/include/cc/local_cc_shards.h` around lines 2546 - 2551, The AddFlushTaskEntry() call currently infers the target flush worker from entry->data_sync_task_->id_, which hides the mapping when data_sync and flush worker counts differ; change AddFlushTaskEntry() signature to accept an explicit worker identifier (either the source data_sync_worker_id or the resolved flush_worker_id) and update all call sites to pass DataSyncWorkerToFlushDataWorker(data_sync_worker_id) or the original data_sync_worker_id as appropriate; update the implementation of AddFlushTaskEntry(), its callers, and any tests to use the new explicit parameter so buffer/queue selection is clear and testable, and remove implicit reliance on entry->data_sync_task_->id_ inside AddFlushTaskEntry().store_handler/data_store_service_client.cpp (1)
318-320: Fail fast on half-configured coroutine callbacks.These entry points treat
yield_fptr/resume_fptras an all-or-nothing pair, but they currently accept a single non-null pointer and silently fall back to blocking behavior. A small DCHECK/helper at the boundary would make miswired flush-worker integrations much easier to catch.Also applies to: 558-559, 1792-1794, 2950-2951, 3191-3192
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@store_handler/data_store_service_client.cpp` around lines 318 - 320, Add a boundary check that fails fast when coroutine callback pointers are half-configured: ensure yield_fptr and resume_fptr are either both null or both non-null (and similarly handle any sync pair like sync_yield_fptr with its companion where applicable). Implement this as a small helper (e.g., ValidateCoroutineCallbacks or a DCHECK pair check) and call it at the start of the entry points that accept these pointers (the functions taking yield_fptr, resume_fptr, sync_yield_fptr) so miswired integrations immediately trigger an assertion/error; apply the same check to the other listed call sites (lines ~558-559, ~1792-1794, ~2950-2951, ~3191-3192).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@build_tx_service.cmake`:
- Around line 33-43: The ASAN detection branch using CMAKE_CXX_FLAGS and the
conditional find_library(NAMES boost_context-asan) is dead and should be removed
or replaced with an explicit ASAN option; implement ASAN support by adding a
boolean option (e.g., USE_ASAN), then apply
add_compile_options(-fsanitize=address) / add_compile_definitions(...) as done
in build_eloq_store.cmake and conditionally pick the ASAN library when USE_ASAN
is ON; also reorder the logic so find_package(Boost 1.70 REQUIRED) runs before
calling find_library(Boost_CONTEXT_LIBRARY) and pass Boost_LIBRARY_DIRS (or
Boost::boost targets) as hints to find_library, and avoid a single plain
${Boost_CONTEXT_LIBRARY} for multi-config generators (use generator expressions
if conditional linking is needed later).
In `@store_handler/data_store_service_client_closure.cpp`:
- Around line 525-583: The loops in SyncPutAllData::Wait,
SyncConcurrentRequest::WaitForCapacityAndIncrement, and
SyncConcurrentRequest::WaitForAll set waiting_ then unlock and call yield_fn(),
which risks a lost-wakeup if a completion fires resume_fn before the coroutine
actually suspends; fix by adding a handshake atomic (e.g., suspended_) and
change the wait loops to set suspended_.store(true, ...) immediately before
unlocking and calling (*yield_fn_)(), and change the completion/resume logic to
only call resume_fn when it successfully observes-and-clears suspended_ (use
compare_exchange to set suspended_ to false when delivering the wakeup); ensure
waiting_ is still used for diagnostics but the actual resume decision is gated
by suspended_ so resume cannot be lost.
In `@tx_service/include/cc/cc_req_misc.h`:
- Around line 897-914: The coroutine handoff can miss a resume because waiting_
is armed before the coroutine actually yields; fix by introducing an atomic
"resume_pending_" (or wakeup_pending_) flag and update the handshake: have
AbortCcRequest/Execute/SetFinished set resume_pending_.store(true) when they
would call resume_fn_, and in WaitableCc::Wait and UpdateCceCkptTsCc::Wait
check/clear resume_pending_ under lock before attempting to yield and only arm
waiting_ while holding the lock so the resume path cannot race; specifically,
change the loop that currently does waiting_.store(true); lk.unlock();
(*yield_fn)(); lk.lock(); waiting_.store(false); to: under lk check if
resume_pending_ is true (consume it and continue), otherwise set waiting_=true
while holding lk, then unlock and call (*yield_fn)(); on re-lock clear waiting_,
and ensure resume code sets resume_pending_ and invokes resume_fn_ if waiting_
is observed or just sets resume_pending_ otherwise so the waiter will see the
pending resume before yielding.
In `@tx_service/include/cc/cc_request.h`:
- Around line 8979-8987: The multiplication in UpdatedMemory() can overflow
because dirty_key_count_ * memory_usage_ is done in the narrower type; fix by
performing the multiplication in a wider integer type (e.g., uint64_t or
unsigned long long) before dividing and then cast the final result back to
size_t. Update the expression in UpdatedMemory() to cast dirty_key_count_ (or
memory_usage_) to the wider type for the intermediate, perform the arithmetic
and rounding there, and return static_cast<size_t>(...) to avoid silent
underestimation while keeping the function signature and logic the same.
In `@tx_service/include/store/data_store_handler.h`:
- Around line 87-95: Derived overrides (PutAll, PutArchivesAll,
CopyBaseToArchive) in the Bigtable/Dynamo handlers currently use different
parameter types (e.g., vector<FlushRecord>, vector<pair<TxKey,int32_t>>) and
thus don't match the base virtual signatures; update these methods in the
derived classes to exactly match the base declarations:
PutAll(...unordered_map<string_view,
vector<unique_ptr<txservice::FlushTaskEntry>>> &flush_task, const
function<void()> *yield_fptr = nullptr, const function<void()> *resume_fptr =
nullptr, const function<void()> *sync_yield_fptr = nullptr), and make
PutArchivesAll and CopyBaseToArchive take the same parameter types as the base
(replace any FlushRecord/vector<pair<...>> parameters with the unordered_map of
unique_ptr<txservice::FlushTaskEntry>), remove or change any conflicting types
like FlushRecord to use FlushTaskEntry, and add the missing UpdateRangeSlices
overload that accepts vector<UpdateRangeSlicesReq> plus the callback parameters
so the derived classes implement both base overloads.
In `@tx_service/src/cc/local_cc_shards.cpp`:
- Around line 6141-6213: The loop currently prioritizes pending_flush_work over
resume_queue causing resumed coroutines (CoroCtx) to starve; change the dispatch
order in the flush worker loop to check and resume entries from resume_queue
(resume_queue.front(), ctx->coro_.resume(), etc.) before pulling new work from
pending_flush_work, or otherwise drain/round-robin resume_queue at least once
per loop iteration so resumed coroutines get scheduled prior to starting fresh
FlushDataImpl tasks; update references in this block that use
pending_flush_work, resume_queue, ctx->coro_, FlushDataImpl, and CoroCtx
accordingly.
- Around line 4811-4819: partition_ids is computed round-robin
(partition_number_this_core and partition_ids via worker_idx + core_number * i)
but later scans use a contiguous [min,max] filter causing workers to overlap
when partition_number_per_scan > 1; change the scan filter to match the actual
round-robin set instead of a single contiguous range: for the scan code that
consumes partition_ids (the logic that builds the [min,max] filter), replace it
with either an explicit per-partition filter (iterate partition_ids and issue
sub-scans or add multiple non-contiguous ranges) or build filters with stride
core_number (e.g., create ranges or a predicate that accepts id % core_number ==
worker_idx) so that scans only cover the partitions in partition_ids and no
duplicates are exported (ensure this change is applied to the similar block
around partition_number_this_core usage at the other location as well).
- Around line 5161-5213: The code filters out records with rec.cce_ == nullptr
when building data_sync_vec (using scan_cc.DataSyncVec()), but
scan_cc.ArchiveVec() and scan_cc.MoveBaseIdxVec() contain indices into the
original unfiltered DataSyncVec, so those indices become invalid; to fix this
either (A) perform the filtering (build data_sync_vec and mv_base_vec) before
ExportForCkpt generates ArchiveVec()/MoveBaseIdxVec() so the export sees
compacted indices, or (B) if export cannot change, create a stable index-mapping
when you build data_sync_vec (map original DataSyncVec() index -> new
data_sync_vec index) and then update the entries in scan_cc.ArchiveVec() and
scan_cc.MoveBaseIdxVec() to use the remapped indices (or remap keys via that map
when calling rec.SetKey and when emplacing mv_base_vec), ensuring all references
(scan_cc.ArchiveVec(), scan_cc.MoveBaseIdxVec(), rec.GetKeyIndex(),
rec.SetKey(), and places using TxKey key_raw) point to the compacted positions.
In `@tx_service/src/tx_index_operation.cpp`:
- Around line 465-488: The log unconditionally dereferences cluster_config_addr
after re-scanning MetaDataReadSet(), which can be nullptr and crash; update the
unlock failure logging in the block that references cluster_config_addr (and the
loop that sets it) to first check cluster_config_addr != nullptr and only append
Term(), NodeGroupId(), and Sharder::Instance().LeaderTerm(...) when non-null,
otherwise log a safe placeholder (or reuse the validated address stored when
unlock_cluster_config_op_.Reset(...) was built); ensure references to
cluster_config_addr, meta_rset, and unlock_cluster_config_op_.hd_result_ are
guarded to prevent null dereference.
---
Outside diff comments:
In `@store_handler/data_store_service_client_closure.h`:
- Around line 101-151: The coroutine-enabled subclasses (FetchTableCallbackData,
FetchDatabaseCallbackData, FetchAllDatabaseCallbackData,
UpsertDatabaseCallbackData, DropDatabaseCallbackData,
DiscoverAllTableNamesCallbackData) still implement one-shot overrides of
Wait()/Notify() which lose a wakeup if Notify() happens before the waiter
yields; update each subclass to use the same finished_ + waiting_ handshake used
by the base SyncCallbackData: remove or refactor their custom Wait()/Notify()
overrides to call the base implementations (or replicate the
waiting_.store/finished_ checks and memory_order semantics used in the base
SetCoroCallbacks/Wait/Notify), ensure SetCoroCallbacks is used to store the
yield/resume pointers, and preserve unlocking before invoking resume_fn_ so the
resume path follows the base-class pattern to avoid lost wakeups.
In `@store_handler/data_store_service_client.cpp`:
- Around line 3120-3150: The final BatchWriteRecords call dispatches before
incrementing sync_concurrent->unfinished_request_cnt_, risking a race where the
callback decrements a not-yet-incremented counter; replace the manual increment
block before that dispatch with the same capacity-guarded increment used
elsewhere by calling WaitForCapacityAndIncrement(sync_concurrent) (or the
existing wrapper used in this file) so the in-flight counter is incremented
before BatchWriteRecords(kv_mvcc_archive_name, ...) is invoked; remove the
explicit unique_lock and unfinished_request_cnt_++ block and ensure
SyncConcurrentRequestCallback and WaitForAll() semantics remain unchanged.
In `@tx_service/include/cc/cc_req_misc.h`:
- Around line 871-879: The Reset implementation for WaitableCc (the one calling
RunOnTxProcessorCc::Reset and setting unfinished_cnt_ and error_code_) must also
clear the previous callback/wait state to avoid carrying stale coroutine wakeup
paths: explicitly reset yield_fn_ and resume_fn_ to empty/null and set waiting_
to false so any recycled request cannot resume a prior Wait; apply the same
clearing changes in the other Reset overload (the similar block around the other
Reset that mirrors these fields).
In `@tx_service/include/cc/local_cc_shards.h`:
- Around line 474-494: The WITH_JEMALLOC initialization runs unconditionally and
can overwrite hash_partition_ckpt_arena_id_ on subsequent calls; move the entire
jemalloc branch (mallctl creating the arena and setting
hash_partition_ckpt_arena_id_) inside the guarded block that checks/creates
hash_partition_ckpt_heap_ within InitializeHashPartitionCkptHeap(), so the
jemalloc arena is created only once alongside mi_heap_new and
hash_partition_main_thread_id_, referencing symbols
InitializeHashPartitionCkptHeap, hash_partition_ckpt_heap_,
hash_partition_ckpt_arena_id_, hash_partition_main_thread_id_, mi_heap_new and
mallctl to locate the correct code to relocate.
- Around line 2015-2023: The callbacks yield_fptr/resume_fptr/sync_yield_fptr
passed into PostProcessFlushTaskEntries (and forwarded into
UpdateRangeSlices/PutAllImpl) are currently stored as raw pointers inside pooled
types SyncPutAllData, SyncConcurrentRequest and SyncCallbackData, creating a
lifetime/USE-AFTER-FREE risk; modify those pooled objects and all call sites so
they take ownership of the callbacks (either by copying the std::function into a
member std::function or by storing a std::shared_ptr<std::function<...>>),
update constructors/factory methods that create
SyncPutAllData/SyncConcurrentRequest/SyncCallbackData to accept the owning form,
and ensure callers to PostProcessFlushTaskEntries/UpdateRangeSlices/PutAllImpl
pass owned callbacks (not pointers) so completion callbacks can safely invoke
them after async/pooled lifetimes.
---
Nitpick comments:
In `@store_handler/data_store_service_client.cpp`:
- Around line 318-320: Add a boundary check that fails fast when coroutine
callback pointers are half-configured: ensure yield_fptr and resume_fptr are
either both null or both non-null (and similarly handle any sync pair like
sync_yield_fptr with its companion where applicable). Implement this as a small
helper (e.g., ValidateCoroutineCallbacks or a DCHECK pair check) and call it at
the start of the entry points that accept these pointers (the functions taking
yield_fptr, resume_fptr, sync_yield_fptr) so miswired integrations immediately
trigger an assertion/error; apply the same check to the other listed call sites
(lines ~558-559, ~1792-1794, ~2950-2951, ~3191-3192).
In `@store_handler/rocksdb_handler.cpp`:
- Around line 526-653: The coroutine branch duplicates the full RocksDB write
loop (inside the SubmitWork lambda) causing TTL/delete handling, metrics, and
error paths to diverge from the synchronous PutAll/PersistKV code; refactor by
extracting the core write logic into a shared helper (e.g., a new method like
ExecutePutAllBatch or PersistBatchToRocksDB) that accepts the DB pointer
(GetDBPtr()), write options, the batch structure, batch_write_size_, and a
callback/result indicator, and then have both the coroutine path (the lambda
submitted via query_worker_pool_->SubmitWork and using
RocksDBWriteSyncCallback/Wait) and the original synchronous path call that
helper so all TTL checks (EncodeToKvKey, SerializeFlushRecord, RecordStatus/TTL
logic), metrics::kv_meter collection, error logging
(status.ToString()/status.code()), and write_batch management live in one place;
ensure the helper returns success/failure so the coroutine lambda invokes
callback.Notify(...) accordingly and the synchronous callers propagate the same
result.
In `@tx_service/include/cc/local_cc_shards.h`:
- Around line 2546-2551: The AddFlushTaskEntry() call currently infers the
target flush worker from entry->data_sync_task_->id_, which hides the mapping
when data_sync and flush worker counts differ; change AddFlushTaskEntry()
signature to accept an explicit worker identifier (either the source
data_sync_worker_id or the resolved flush_worker_id) and update all call sites
to pass DataSyncWorkerToFlushDataWorker(data_sync_worker_id) or the original
data_sync_worker_id as appropriate; update the implementation of
AddFlushTaskEntry(), its callers, and any tests to use the new explicit
parameter so buffer/queue selection is clear and testable, and remove implicit
reliance on entry->data_sync_task_->id_ inside AddFlushTaskEntry().
In `@tx_service/include/cc/template_cc_map.h`:
- Around line 8125-8147: AdjustDataKeyStats currently updates data_key_count_
and dirty_data_key_count_ but doesn't validate the invariant
dirty_data_key_count_ <= data_key_count_; add a cheap assertion at the end of
AdjustDataKeyStats (after the existing updates and after the early return for
table_name_.IsMeta()) that checks dirty_data_key_count_ is never greater than
data_key_count_ (taking care of the size_t vs signed conversions) so future
missed bookkeeping fails fast; reference AdjustDataKeyStats, data_key_count_,
dirty_data_key_count_, and table_name_.IsMeta() to locate where to insert the
assertion.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: c6e58373-8730-43e2-b81f-e1400b27b474
📒 Files selected for processing (24)
build_tx_service.cmakestore_handler/bigtable_handler.cppstore_handler/data_store_service_client.cppstore_handler/data_store_service_client.hstore_handler/data_store_service_client_closure.cppstore_handler/data_store_service_client_closure.hstore_handler/eloq_data_store_service/eloq_store_data_store.cppstore_handler/rocksdb_handler.cppstore_handler/rocksdb_handler.htx_service/include/cc/catalog_cc_map.htx_service/include/cc/cc_req_misc.htx_service/include/cc/cc_request.htx_service/include/cc/local_cc_shards.htx_service/include/cc/range_cc_map.htx_service/include/cc/range_slice.htx_service/include/cc/template_cc_map.htx_service/include/fault/log_replay_service.htx_service/include/range_record.htx_service/include/store/data_store_handler.htx_service/include/store/int_mem_store.htx_service/src/cc/local_cc_shards.cpptx_service/src/checkpointer.cpptx_service/src/fault/log_replay_service.cpptx_service/src/tx_index_operation.cpp
Comment on lines
+33
to
+43
| # boost_context for FlushDataWorker coroutine refactor (Phase 1) | ||
| if(CMAKE_BUILD_TYPE STREQUAL "Debug" AND CMAKE_CXX_FLAGS MATCHES "fsanitize=address") | ||
| find_library(Boost_CONTEXT_LIBRARY NAMES boost_context-asan) | ||
| else() | ||
| find_library(Boost_CONTEXT_LIBRARY NAMES boost_context) | ||
| endif() | ||
| if(NOT Boost_CONTEXT_LIBRARY) | ||
| message(FATAL_ERROR "libboost_context not found") | ||
| endif() | ||
| find_package(Boost 1.70 REQUIRED) | ||
|
|
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Verify how ASAN is configured in CMake and whether repo tooling expects boost_context-asan.
rg -n --glob 'CMakeLists.txt' --glob '*.cmake' \
'fsanitize=address|CMAKE_CXX_FLAGS_DEBUG|CMAKE_BUILD_TYPE|CMAKE_CONFIGURATION_TYPES|boost_context-asan|find_package\(Boost|find_library\(Boost_CONTEXT_LIBRARY'Repository: eloqdata/tx_service
Length of output: 2752
🏁 Script executed:
# Check the target_link_libraries context mentioned in the comment (line 225-227)
sed -n '220,230p' build_tx_service.cmakeRepository: eloqdata/tx_service
Length of output: 458
🏁 Script executed:
# Search for where ASAN flags might be added to CMAKE_CXX_FLAGS in tx_service context
rg -n 'add_compile_options|add_link_options|CMAKE_CXX_FLAGS.*fsanitize' build_tx_service.cmake tx_service/CMakeLists.txtRepository: eloqdata/tx_service
Length of output: 174
🏁 Script executed:
# Check if boost_context-asan is actually built/provisioned anywhere in the repo
rg -n 'boost_context-asan' --glob '*.cmake' --glob 'CMakeLists.txt'Repository: eloqdata/tx_service
Length of output: 393
🏁 Script executed:
# Check if there are any CI/build scripts that explicitly enable ASAN for tx_service
rg -n 'ASAN|asan|sanitize.*address' --glob '*.sh' --glob '*.yml' --glob '*.yaml' --glob 'Dockerfile*'Repository: eloqdata/tx_service
Length of output: 45
🏁 Script executed:
# Check the context around build_tx_service.cmake to see if ASAN is configured elsewhere
head -50 build_tx_service.cmakeRepository: eloqdata/tx_service
Length of output: 1722
🏁 Script executed:
# Compare how eloq_store enables ASAN vs how tx_service should
sed -n '20,40p' store_handler/eloq_data_store_service/build_eloq_store.cmakeRepository: eloqdata/tx_service
Length of output: 790
Remove dead ASAN detection code or implement proper ASAN support.
Lines 34–37 check CMAKE_BUILD_TYPE and CMAKE_CXX_FLAGS for ASAN, but CMAKE_CXX_FLAGS never receives -fsanitize=address in this build (only -DFAULT_INJECTOR is set in CMAKE_CXX_FLAGS_DEBUG). This means the boost_context-asan branch is unreachable dead code. Either remove it, or properly implement ASAN support using an option with add_compile_options() and add_compile_definitions() like eloq_store does (see build_eloq_store.cmake for the pattern).
Additionally, running find_library() before find_package(Boost) (line 42) risks using libraries from the wrong install if Boost_LIBRARY_DIRS is not yet set. Reorder to call find_package(Boost) first, then use the discovered paths as hints for find_library().
The same single-value ${Boost_CONTEXT_LIBRARY} at line 226 cannot adapt to multi-config generators anyway; if conditional linking becomes necessary in the future, use generator expressions.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@build_tx_service.cmake` around lines 33 - 43, The ASAN detection branch using
CMAKE_CXX_FLAGS and the conditional find_library(NAMES boost_context-asan) is
dead and should be removed or replaced with an explicit ASAN option; implement
ASAN support by adding a boolean option (e.g., USE_ASAN), then apply
add_compile_options(-fsanitize=address) / add_compile_definitions(...) as done
in build_eloq_store.cmake and conditionally pick the ASAN library when USE_ASAN
is ON; also reorder the logic so find_package(Boost 1.70 REQUIRED) runs before
calling find_library(Boost_CONTEXT_LIBRARY) and pass Boost_LIBRARY_DIRS (or
Boost::boost targets) as hints to find_library, and avoid a single plain
${Boost_CONTEXT_LIBRARY} for multi-config generators (use generator expressions
if conditional linking is needed later).
Comment on lines
+525
to
+583
| void SyncPutAllData::Wait(const std::function<void()> *yield_fn, | ||
| const std::function<void()> *resume_fn) | ||
| { | ||
| if (yield_fn == nullptr || resume_fn == nullptr) | ||
| { | ||
| Wait(); | ||
| return; | ||
| } | ||
| std::unique_lock<bthread::Mutex> lk(mux_); | ||
| while (completed_partitions_ < total_partitions_) | ||
| { | ||
| waiting_.store(true, std::memory_order_release); | ||
| lk.unlock(); | ||
| (*yield_fn)(); | ||
| lk.lock(); | ||
| waiting_.store(false, std::memory_order_release); | ||
| } | ||
| } | ||
|
|
||
| void SyncConcurrentRequest::WaitForCapacityAndIncrement() | ||
| { | ||
| std::unique_lock<bthread::Mutex> lk(mux_); | ||
| while (unfinished_request_cnt_ >= max_flying_write_count) | ||
| { | ||
| if (yield_fn_ != nullptr && resume_fn_ != nullptr) | ||
| { | ||
| waiting_.store(true, std::memory_order_release); | ||
| lk.unlock(); | ||
| (*yield_fn_)(); | ||
| lk.lock(); | ||
| waiting_.store(false, std::memory_order_release); | ||
| } | ||
| else | ||
| { | ||
| cv_.wait(lk); | ||
| } | ||
| } | ||
| unfinished_request_cnt_++; | ||
| } | ||
|
|
||
| void SyncConcurrentRequest::WaitForAll() | ||
| { | ||
| std::unique_lock<bthread::Mutex> lk(mux_); | ||
| all_request_started_ = true; | ||
| while (unfinished_request_cnt_ != 0) | ||
| { | ||
| if (yield_fn_ != nullptr && resume_fn_ != nullptr) | ||
| { | ||
| waiting_.store(true, std::memory_order_release); | ||
| lk.unlock(); | ||
| (*yield_fn_)(); | ||
| lk.lock(); | ||
| waiting_.store(false, std::memory_order_release); | ||
| } | ||
| else | ||
| { | ||
| cv_.wait(lk); | ||
| } | ||
| } |
There was a problem hiding this comment.
Avoid the arm-then-yield lost-wakeup race.
These loops set waiting_ = true, drop mux_, and only then call yield_fn. If a completion path sees waiting_ in that gap and fires resume_fn, the resume can happen before the coroutine has actually suspended, and the following yield_fn() can sleep forever with no remaining wakeup.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@store_handler/data_store_service_client_closure.cpp` around lines 525 - 583,
The loops in SyncPutAllData::Wait,
SyncConcurrentRequest::WaitForCapacityAndIncrement, and
SyncConcurrentRequest::WaitForAll set waiting_ then unlock and call yield_fn(),
which risks a lost-wakeup if a completion fires resume_fn before the coroutine
actually suspends; fix by adding a handshake atomic (e.g., suspended_) and
change the wait loops to set suspended_.store(true, ...) immediately before
unlocking and calling (*yield_fn_)(), and change the completion/resume logic to
only call resume_fn when it successfully observes-and-clears suspended_ (use
compare_exchange to set suspended_ to false when delivering the wakeup); ensure
waiting_ is still used for diagnostics but the actual resume decision is gated
by suspended_ so resume cannot be lost.
Comment on lines
+4811
to
+4819
| auto partition_number_this_core = | ||
| partition_number / core_number + | ||
| (worker_idx < partition_number % core_number); | ||
| std::vector<size_t> partition_ids; | ||
| partition_ids.reserve(partition_number_this_core); | ||
| for (size_t i = 0; i < partition_number_this_core; ++i) | ||
| { | ||
| partition_ids.emplace_back(worker_idx + core_number * i); | ||
| } |
There was a problem hiding this comment.
Keep the hash-partition scan filter round-robin.
partition_ids are assigned as worker_idx + core_number * i, but each chunk is filtered as a contiguous [min,max] range. Once partition_number_per_scan > 1, workers overlap on partitions owned by other workers and can export the same records multiple times.
💡 Minimal fix
- size_t min_partition_id_this_scan = partition_ids[i];
- size_t max_partition_id_this_scan =
+ const size_t min_partition_id_this_scan = partition_ids[i];
+ const size_t max_partition_id_this_scan =
partition_ids[std::min(i + partition_number_per_scan,
partition_number_this_core) -
1];
std::function<bool(size_t)> filter_lambda =
- [min_partition_id_this_scan,
- max_partition_id_this_scan,
+ [worker_idx,
+ core_number,
+ min_partition_id_this_scan,
+ max_partition_id_this_scan,
&filter_func =
data_sync_task->filter_lambda_](const size_t hash_code)
{
- return (hash_code % total_hash_partitions) >=
- min_partition_id_this_scan &&
- (hash_code % total_hash_partitions) <=
- max_partition_id_this_scan &&
+ const size_t partition_id = hash_code % total_hash_partitions;
+ return partition_id >= min_partition_id_this_scan &&
+ partition_id <= max_partition_id_this_scan &&
+ (partition_id % core_number) == worker_idx &&
(!filter_func || filter_func(hash_code));
};Also applies to: 4856-4875
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@tx_service/src/cc/local_cc_shards.cpp` around lines 4811 - 4819,
partition_ids is computed round-robin (partition_number_this_core and
partition_ids via worker_idx + core_number * i) but later scans use a contiguous
[min,max] filter causing workers to overlap when partition_number_per_scan > 1;
change the scan filter to match the actual round-robin set instead of a single
contiguous range: for the scan code that consumes partition_ids (the logic that
builds the [min,max] filter), replace it with either an explicit per-partition
filter (iterate partition_ids and issue sub-scans or add multiple non-contiguous
ranges) or build filters with stride core_number (e.g., create ranges or a
predicate that accepts id % core_number == worker_idx) so that scans only cover
the partitions in partition_ids and no duplicates are exported (ensure this
change is applied to the similar block around partition_number_this_core usage
at the other location as well).
Comment on lines
+6141
to
+6213
| if (!pending_flush_work.empty()) | ||
| { | ||
| std::unique_ptr<FlushDataTask> cur_work = | ||
| std::move(pending_flush_work.front()); | ||
| pending_flush_work.pop_front(); | ||
| flush_data_worker_ctx_.cv_.notify_all(); | ||
| flush_worker_lk.unlock(); | ||
|
|
||
| auto ctx = std::make_shared<CoroCtx>(); | ||
| ctx->task_ = std::move(cur_work); | ||
| ctx->coro_ = boost::context::callcc( | ||
| std::allocator_arg, | ||
| flush_coro_stack_allocator_, | ||
| [this, ctx, worker_idx](continuation &&sink) | ||
| { | ||
| ctx->yield_fn = [&sink]() { sink = sink.resume(); }; | ||
| ctx->sync_yield_func = | ||
| [&sink, | ||
| weak_ctx = std::weak_ptr<CoroCtx>(ctx), | ||
| this, | ||
| worker_idx]() | ||
| { | ||
| if (auto c = weak_ctx.lock()) | ||
| { | ||
| { | ||
| std::lock_guard<std::mutex> lk( | ||
| flush_data_worker_ctx_.mux_); | ||
| resume_queue_[worker_idx].push_back( | ||
| std::move(c)); | ||
| flush_data_worker_ctx_.cv_.notify_all(); | ||
| } | ||
| sink = sink.resume(); | ||
| } | ||
| }; | ||
| ctx->resume_fn = [this, | ||
| worker_idx, | ||
| weak_ctx = std::weak_ptr<CoroCtx>(ctx)]() | ||
| { | ||
| if (auto c = weak_ctx.lock()) | ||
| { | ||
| std::lock_guard<std::mutex> lk( | ||
| flush_data_worker_ctx_.mux_); | ||
| resume_queue_[worker_idx].push_back(std::move(c)); | ||
| flush_data_worker_ctx_.cv_.notify_all(); | ||
| } | ||
| else | ||
| { | ||
| LOG(FATAL) | ||
| << "CoroCtx resume_fn: weak_ctx is expired"; | ||
| } | ||
| }; | ||
| FlushDataImpl(ctx->task_.get(), | ||
| worker_idx, | ||
| ctx->sync_yield_func, | ||
| ctx->yield_fn, | ||
| ctx->resume_fn); | ||
| return std::move(sink); | ||
| }); | ||
|
|
||
| flush_worker_lk.lock(); | ||
| continue; | ||
| } | ||
|
|
||
| if (!resume_queue.empty()) | ||
| { | ||
| std::shared_ptr<CoroCtx> ctx = std::move(resume_queue.front()); | ||
| resume_queue.pop_front(); | ||
| flush_worker_lk.unlock(); | ||
|
|
||
| ctx->coro_ = ctx->coro_.resume(); | ||
|
|
||
| flush_worker_lk.lock(); | ||
| continue; |
There was a problem hiding this comment.
Resume ready coroutines before starting fresh flush work.
The loop gives pending_flush_work priority over resume_queue. With a single flush worker and steady incoming batches, resumed coroutines can starve indefinitely, which delays PostProcessFlushTaskEntries() and DeallocateFlushMemQuota() for already-started flushes.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@tx_service/src/cc/local_cc_shards.cpp` around lines 6141 - 6213, The loop
currently prioritizes pending_flush_work over resume_queue causing resumed
coroutines (CoroCtx) to starve; change the dispatch order in the flush worker
loop to check and resume entries from resume_queue (resume_queue.front(),
ctx->coro_.resume(), etc.) before pulling new work from pending_flush_work, or
otherwise drain/round-robin resume_queue at least once per loop iteration so
resumed coroutines get scheduled prior to starting fresh FlushDataImpl tasks;
update references in this block that use pending_flush_work, resume_queue,
ctx->coro_, FlushDataImpl, and CoroCtx accordingly.
lokax
force-pushed
the
yf-pro-ckpt-boost-context
branch
from
March 23, 2026 06:40
5d199a7 to
db88b4b
Compare
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
tx_service/include/cc/cc_req_misc.h (1)
881-898:⚠️ Potential issue | 🔴 CriticalMake callback registration single-source-of-truth.
Wait(...)acceptsyield_fn/resume_fn, but the completion path only consults the storedresume_fn_, andSetCoroCallbacks()writes that state withoutmux_. A caller that only usesWait(yield, resume)can hang, and arming callbacks concurrently with a fast completion is a read/write race onresume_fn_. Please install/clear the callback pair undermux_in one place only;WaitableCc::Reset()should also clear it before reuse.Also applies to: 998-1001, 1048-1086, 1128-1131
♻️ Duplicate comments (4)
tx_service/include/cc/cc_req_misc.h (1)
897-914:⚠️ Potential issue | 🔴 CriticalThe coroutine handoff still loses a pre-yield resume.
waiting_is set before the coroutine actually yields. If the last worker reachesAbortCcRequest(),Execute(), orSetFinished()in that window, it can runresume_fn_immediately and the subsequentyield_fn()has no wakeup left to consume. Please add a pending-resume bit/state undermux_and consume it before yielding.Also applies to: 941-952, 964-975, 1055-1072, 1085-1101
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tx_service/include/cc/cc_req_misc.h` around lines 897 - 914, The Wait(...) coroutine handoff races because waiting_ is set before the coroutine actually yields, allowing a pre-yield resume (from AbortCcRequest, Execute, SetFinished or where resume_fn_/resume logic runs) to be lost; add a protected "pending_resume" boolean/state guarded by mux_ that is set by the producers instead of directly invoking resume_fn_ when the waiter hasn't yet yielded, and modify Wait (and the other Wait variants at the same pattern) to check and clear pending_resume while holding mux_ before actually unlocking and calling (*yield_fn)(), and only call (*yield_fn)() when no pending resume remains (if pending_resume is set, consume it and call (*resume_fn)() instead of yielding). Ensure producers set pending_resume under mutex (or invoke resume_fn_ immediately if waiting_ is true and pending_resume is clear), and clear pending_resume when consumed so no resume is lost.tx_service/src/cc/local_cc_shards.cpp (3)
6141-6213:⚠️ Potential issue | 🟠 MajorResumed coroutines can still starve behind newly queued flush work.
Line 6141 checks
pending_flush_workbeforeresume_queue. Under sustained intake, resumed coroutines may not make progress promptly.💡 Prioritize resumes before starting fresh work
- if (!pending_flush_work.empty()) + if (!resume_queue.empty()) + { + std::shared_ptr<CoroCtx> ctx = std::move(resume_queue.front()); + resume_queue.pop_front(); + flush_worker_lk.unlock(); + ctx->coro_ = ctx->coro_.resume(); + flush_worker_lk.lock(); + continue; + } + + if (!pending_flush_work.empty()) { std::unique_ptr<FlushDataTask> cur_work = std::move(pending_flush_work.front()); pending_flush_work.pop_front(); flush_data_worker_ctx_.cv_.notify_all(); flush_worker_lk.unlock(); @@ - if (!resume_queue.empty()) - { - std::shared_ptr<CoroCtx> ctx = std::move(resume_queue.front()); - resume_queue.pop_front(); - flush_worker_lk.unlock(); - - ctx->coro_ = ctx->coro_.resume(); - - flush_worker_lk.lock(); - continue; - }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tx_service/src/cc/local_cc_shards.cpp` around lines 6141 - 6213, The loop currently favors starting new FlushData tasks (pending_flush_work) before resuming coroutines (resume_queue), causing resumed coroutines to starve; reorder the branch checks so that the resume_queue handling runs before pending_flush_work, i.e., check and pop from resume_queue and call ctx->coro_.resume() (using the existing resume_queue, ctx->coro_.resume(), and flush_worker_lk logic) prior to creating/launching a new coro from pending_flush_work, and ensure the same lock/unlock semantics around flush_worker_lk are preserved when swapping the two blocks so lock handling and notify_all calls remain correct.
4856-4875:⚠️ Potential issue | 🔴 CriticalRound-robin partition assignment is still filtered as a contiguous range.
Line 4859-Line 4873 still treats a round-robin set as
[min,max]. Whenpartition_number_per_scan > 1, this can overlap partitions owned by other workers.💡 Minimal fix
- std::function<bool(size_t)> filter_lambda = - [min_partition_id_this_scan, - max_partition_id_this_scan, - &filter_func = - data_sync_task->filter_lambda_](const size_t hash_code) + std::function<bool(size_t)> filter_lambda = + [worker_idx, + core_number, + min_partition_id_this_scan, + max_partition_id_this_scan, + &filter_func = + data_sync_task->filter_lambda_](const size_t hash_code) { - return (hash_code % total_hash_partitions) >= - min_partition_id_this_scan && - (hash_code % total_hash_partitions) <= - max_partition_id_this_scan && + const size_t partition_id = hash_code % total_hash_partitions; + return partition_id >= min_partition_id_this_scan && + partition_id <= max_partition_id_this_scan && + (partition_id % core_number) == worker_idx && (!filter_func || filter_func(hash_code)); };🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tx_service/src/cc/local_cc_shards.cpp` around lines 4856 - 4875, The filter_lambda incorrectly treats the round-robin selection as a contiguous range using min_partition_id_this_scan and max_partition_id_this_scan; instead, change the predicate in the lambda (filter_lambda) to test membership against the actual partition_ids slice for this scan (from index i to i + partition_number_per_scan - 1) rather than using a range check — e.g., capture the relevant partition_ids sub-range (or build a small local set/vector of those IDs) and check (hash_code % total_hash_partitions) is equal to one of those IDs in addition to honoring data_sync_task->filter_lambda_; update the captures accordingly so partition_ids and partition subset are available inside filter_lambda.
5161-5213:⚠️ Potential issue | 🔴 CriticalArchive/MoveBase indices are dereferenced after compaction without remapping.
Line 5167-Line 5195 skips null-
cce_records, but Line 5203/Line 5208 still index by original scan positions. That can produce wrong keys or out-of-bounds access.💡 Remap original scan indices to compacted indices
+ std::vector<size_t> compact_idx_map( + scan_cc.accumulated_scan_cnt_, std::numeric_limits<size_t>::max()); data_sync_vec->reserve(scan_cc.accumulated_scan_cnt_); for (size_t j = 0; j < scan_cc.accumulated_scan_cnt_; ++j) { auto &rec = scan_cc.DataSyncVec()[j]; // Note. Clone key instead of move key. The memory of // rec.Key() will be reused to avoid memory allocation. if (rec.cce_) { @@ - data_sync_vec->emplace_back( + data_sync_vec->emplace_back( rec.Key().Clone(), rec.ReleaseVersionedPayload(), rec.payload_status_, rec.commit_ts_, rec.cce_, rec.post_flush_size_, part_id); } + compact_idx_map[j] = data_sync_vec->size() - 1; } } for (size_t j = 0; j < scan_cc.ArchiveVec().size(); ++j) { auto &rec = scan_cc.ArchiveVec()[j]; - rec.SetKey((*data_sync_vec)[rec.GetKeyIndex()].Key()); + const size_t old_idx = rec.GetKeyIndex(); + const size_t new_idx = old_idx < compact_idx_map.size() + ? compact_idx_map[old_idx] + : std::numeric_limits<size_t>::max(); + if (new_idx != std::numeric_limits<size_t>::max()) + { + rec.SetKey((*data_sync_vec)[new_idx].Key()); + } } for (size_t j = 0; j < scan_cc.MoveBaseIdxVec().size(); ++j) { - size_t key_idx = scan_cc.MoveBaseIdxVec()[j]; - TxKey key_raw = (*data_sync_vec)[key_idx].Key(); + const size_t old_idx = scan_cc.MoveBaseIdxVec()[j]; + const size_t key_idx = old_idx < compact_idx_map.size() + ? compact_idx_map[old_idx] + : std::numeric_limits<size_t>::max(); + if (key_idx == std::numeric_limits<size_t>::max()) + { + continue; + } + TxKey key_raw = (*data_sync_vec)[key_idx].Key(); int32_t part_id = Sharder::MapKeyHashToHashPartitionId(key_raw.Hash()); mv_base_vec->emplace_back(std::move(key_raw), part_id); }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tx_service/src/cc/local_cc_shards.cpp` around lines 5161 - 5213, The archive/move-base loops use original scan positions but earlier code compacts out records with null cce_, so create a remapping from original scan indices to compaction indices: allocate a vector<int> (size scan_cc.accumulated_scan_cnt_) initialized to -1, and in the loop that emplaces into data_sync_vec (the block using scan_cc.DataSyncVec(), rec.cce_, and data_sync_vec->emplace_back) set map[j] = current_compacted_index (increment as you push); then in the ArchiveVec loop (rec.SetKey((*data_sync_vec)[rec.GetKeyIndex()].Key())) and MoveBaseIdxVec handling (key_idx = scan_cc.MoveBaseIdxVec()[j]; TxKey key_raw = (*data_sync_vec)[key_idx].Key()) replace uses of the original index with the mapped index (map[orig_idx]) and handle map==-1 (skip or assert/error) to avoid wrong keys or OOB access. Ensure you reference scan_cc.DataSyncVec(), scan_cc.ArchiveVec(), scan_cc.MoveBaseIdxVec(), data_sync_vec->emplace_back, and FlushRecord::SetKey when applying the remap.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Duplicate comments:
In `@tx_service/include/cc/cc_req_misc.h`:
- Around line 897-914: The Wait(...) coroutine handoff races because waiting_ is
set before the coroutine actually yields, allowing a pre-yield resume (from
AbortCcRequest, Execute, SetFinished or where resume_fn_/resume logic runs) to
be lost; add a protected "pending_resume" boolean/state guarded by mux_ that is
set by the producers instead of directly invoking resume_fn_ when the waiter
hasn't yet yielded, and modify Wait (and the other Wait variants at the same
pattern) to check and clear pending_resume while holding mux_ before actually
unlocking and calling (*yield_fn)(), and only call (*yield_fn)() when no pending
resume remains (if pending_resume is set, consume it and call (*resume_fn)()
instead of yielding). Ensure producers set pending_resume under mutex (or invoke
resume_fn_ immediately if waiting_ is true and pending_resume is clear), and
clear pending_resume when consumed so no resume is lost.
In `@tx_service/src/cc/local_cc_shards.cpp`:
- Around line 6141-6213: The loop currently favors starting new FlushData tasks
(pending_flush_work) before resuming coroutines (resume_queue), causing resumed
coroutines to starve; reorder the branch checks so that the resume_queue
handling runs before pending_flush_work, i.e., check and pop from resume_queue
and call ctx->coro_.resume() (using the existing resume_queue,
ctx->coro_.resume(), and flush_worker_lk logic) prior to creating/launching a
new coro from pending_flush_work, and ensure the same lock/unlock semantics
around flush_worker_lk are preserved when swapping the two blocks so lock
handling and notify_all calls remain correct.
- Around line 4856-4875: The filter_lambda incorrectly treats the round-robin
selection as a contiguous range using min_partition_id_this_scan and
max_partition_id_this_scan; instead, change the predicate in the lambda
(filter_lambda) to test membership against the actual partition_ids slice for
this scan (from index i to i + partition_number_per_scan - 1) rather than using
a range check — e.g., capture the relevant partition_ids sub-range (or build a
small local set/vector of those IDs) and check (hash_code %
total_hash_partitions) is equal to one of those IDs in addition to honoring
data_sync_task->filter_lambda_; update the captures accordingly so partition_ids
and partition subset are available inside filter_lambda.
- Around line 5161-5213: The archive/move-base loops use original scan positions
but earlier code compacts out records with null cce_, so create a remapping from
original scan indices to compaction indices: allocate a vector<int> (size
scan_cc.accumulated_scan_cnt_) initialized to -1, and in the loop that emplaces
into data_sync_vec (the block using scan_cc.DataSyncVec(), rec.cce_, and
data_sync_vec->emplace_back) set map[j] = current_compacted_index (increment as
you push); then in the ArchiveVec loop
(rec.SetKey((*data_sync_vec)[rec.GetKeyIndex()].Key())) and MoveBaseIdxVec
handling (key_idx = scan_cc.MoveBaseIdxVec()[j]; TxKey key_raw =
(*data_sync_vec)[key_idx].Key()) replace uses of the original index with the
mapped index (map[orig_idx]) and handle map==-1 (skip or assert/error) to avoid
wrong keys or OOB access. Ensure you reference scan_cc.DataSyncVec(),
scan_cc.ArchiveVec(), scan_cc.MoveBaseIdxVec(), data_sync_vec->emplace_back, and
FlushRecord::SetKey when applying the remap.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 0d3537e6-68ca-46cd-b677-e747aaff9161
📒 Files selected for processing (8)
store_handler/data_store_service_client.hstore_handler/data_store_service_client_closure.cpptx_service/include/cc/cc_req_misc.htx_service/include/cc/local_cc_shards.htx_service/include/fault/log_replay_service.htx_service/include/range_record.htx_service/src/cc/local_cc_shards.cpptx_service/src/fault/log_replay_service.cpp
✅ Files skipped from review due to trivial changes (1)
- tx_service/include/range_record.h
🚧 Files skipped from review as they are similar to previous changes (4)
- tx_service/include/fault/log_replay_service.h
- tx_service/src/fault/log_replay_service.cpp
- tx_service/include/cc/local_cc_shards.h
- store_handler/data_store_service_client.h
lokax
force-pushed
the
yf-pro-ckpt-boost-context
branch
from
March 23, 2026 08:26
db88b4b to
305e7af
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (2)
tx_service/include/cc/cc_req_misc.h (2)
881-886:⚠️ Potential issue | 🔴 CriticalComplete the coroutine wait/resume handshake.
Wait(yield_fn, resume_fn)never installs its callbacks into the members thatAbortCcRequest()/Execute()read, andwaiting_is armed before the coroutine has actually yielded. That makes the wakeup easy to lose, and reusedWaitableCcinstances can also keep stale callback pointers because the new state is never cleared on reset.Also applies to: 897-914, 934-975, 998-1001
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tx_service/include/cc/cc_req_misc.h` around lines 881 - 886, Set up the full coroutine wait/resume handshake by ensuring Wait installs the callbacks into the members read by AbortCcRequest()/Execute(): call SetCoroCallbacks(...) (or assign yield_fn_/resume_fn_ directly) before any chance to be woken, only arm waiting_ after the coroutine actually yields (i.e., set waiting_ from inside the yield function or immediately after calling the yield callback), and clear waiting_, yield_fn_ and resume_fn_ when resetting/reusing a WaitableCc (e.g., in Reset()/constructor) so no stale pointers remain; apply the same changes to the other Wait implementations referenced (lines ~897-914, 934-975, 998-1001) to keep behavior consistent.
1048-1053:⚠️ Potential issue | 🔴 Critical
UpdateCceCkptTsCcstill has the same lost-wakeup window.This path repeats the same pattern: the completion side only consults member
resume_fn_, whileWait(yield_fn, resume_fn)does not publish its arguments into that state and setswaiting_before the actual yield. IfSetFinished()wins that race, the flush coroutine can block forever.Also applies to: 1055-1101, 1128-1131
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tx_service/include/cc/cc_req_misc.h` around lines 1048 - 1053, The lost-wakeup race comes from SetCoroCallbacks/Wait using separate local args and member resume_fn_/yield_fn_ and setting waiting_ before the published callbacks, so SetFinished can race and miss the resume; fix by making Wait publish the coroutine callbacks into the shared members (assign yield_fn_ and resume_fn_) before flipping waiting_, or alternatively have SetFinished atomically snapshot and invoke any resume_fn_ even if waiting_ races; update functions UpdateCceCkptTsCc/Wait/SetCoroCallbacks/SetFinished to either (a) assign yield_fn_ and resume_fn_ into the members before setting waiting_ and use appropriate memory ordering, or (b) in SetFinished always load resume_fn_ into a local under the same synchronization and call it if non-null, ensuring no lost-wakeup.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@tx_service/include/cc/local_cc_shards.h`:
- Around line 476-481: The arenas.create call and assignment to
hash_partition_ckpt_arena_id_ must be moved inside the one-time init guard so it
only runs when hash_partition_ckpt_heap_ is null; currently arenas.create runs
on every invocation and overwrites hash_partition_ckpt_arena_id_, leaking prior
arenas. Update the init sequence around hash_partition_ckpt_heap_ and
hash_partition_main_thread_id_ (the if (!hash_partition_ckpt_heap_) block) to
also call arenas.create and set hash_partition_ckpt_arena_id_ there, ensuring
subsequent calls skip arena creation and do not mutate or leak the previous
arena ID.
In `@tx_service/include/fault/log_replay_service.h`:
- Around line 230-232: The map currently stores std::unique_ptr<ConnectionInfo>
in inbound_connections_ which doesn't prevent use-after-free when callbacks
(on_received_messages, on_idle_timeout, on_closed) grab raw pointers under
inbound_mux_ and use them after unlocking; change the ownership to
std::shared_ptr<ConnectionInfo> (i.e., make inbound_connections_ an
unordered_map<brpc::StreamId, std::shared_ptr<ConnectionInfo>>) and update all
places that construct entries to create shared_ptrs; then in each callback
acquire a local std::shared_ptr<ConnectionInfo> under inbound_mux_ (copying the
map value) and use that local shared_ptr after releasing the lock to keep the
object alive.
---
Duplicate comments:
In `@tx_service/include/cc/cc_req_misc.h`:
- Around line 881-886: Set up the full coroutine wait/resume handshake by
ensuring Wait installs the callbacks into the members read by
AbortCcRequest()/Execute(): call SetCoroCallbacks(...) (or assign
yield_fn_/resume_fn_ directly) before any chance to be woken, only arm waiting_
after the coroutine actually yields (i.e., set waiting_ from inside the yield
function or immediately after calling the yield callback), and clear waiting_,
yield_fn_ and resume_fn_ when resetting/reusing a WaitableCc (e.g., in
Reset()/constructor) so no stale pointers remain; apply the same changes to the
other Wait implementations referenced (lines ~897-914, 934-975, 998-1001) to
keep behavior consistent.
- Around line 1048-1053: The lost-wakeup race comes from SetCoroCallbacks/Wait
using separate local args and member resume_fn_/yield_fn_ and setting waiting_
before the published callbacks, so SetFinished can race and miss the resume; fix
by making Wait publish the coroutine callbacks into the shared members (assign
yield_fn_ and resume_fn_) before flipping waiting_, or alternatively have
SetFinished atomically snapshot and invoke any resume_fn_ even if waiting_
races; update functions UpdateCceCkptTsCc/Wait/SetCoroCallbacks/SetFinished to
either (a) assign yield_fn_ and resume_fn_ into the members before setting
waiting_ and use appropriate memory ordering, or (b) in SetFinished always load
resume_fn_ into a local under the same synchronization and call it if non-null,
ensuring no lost-wakeup.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 7cf748d7-4aa2-4afd-bed7-9c32de733e3b
📒 Files selected for processing (12)
store_handler/data_store_service_client.cppstore_handler/data_store_service_client.hstore_handler/data_store_service_client_closure.cppstore_handler/rocksdb_handler.htx_service/include/cc/cc_req_misc.htx_service/include/cc/local_cc_shards.htx_service/include/fault/log_replay_service.htx_service/include/range_record.htx_service/include/store/int_mem_store.htx_service/src/cc/local_cc_shards.cpptx_service/src/fault/log_replay_service.cpptx_service/src/tx_index_operation.cpp
✅ Files skipped from review due to trivial changes (1)
- store_handler/data_store_service_client_closure.cpp
🚧 Files skipped from review as they are similar to previous changes (7)
- tx_service/include/range_record.h
- tx_service/src/fault/log_replay_service.cpp
- tx_service/include/store/int_mem_store.h
- store_handler/rocksdb_handler.h
- store_handler/data_store_service_client.h
- tx_service/src/cc/local_cc_shards.cpp
- store_handler/data_store_service_client.cpp
Comment on lines
+476
to
+481
| std::unique_lock<std::mutex> lk(hash_partition_ckpt_heap_mux_); | ||
| if (!hash_partition_ckpt_heap_) | ||
| { | ||
| hash_partition_main_thread_id_ = mi_thread_id(); | ||
| hash_partition_ckpt_heap_ = mi_heap_new(); | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
cat -n tx_service/include/cc/local_cc_shards.h | sed -n '470,500p'Repository: eloqdata/tx_service
Length of output: 1098
Move jemalloc arena creation inside the one-time init guard.
The arenas.create call (lines 483-494) executes on every invocation, not just during initialization. After the first call, subsequent invocations recreate the arena and overwrite hash_partition_ckpt_arena_id_, leaking the previous arena and splitting accounting across multiple arena IDs. This breaks idempotency.
♻️ Suggested fix
void InitializeHashPartitionCkptHeap()
{
std::unique_lock<std::mutex> lk(hash_partition_ckpt_heap_mux_);
if (!hash_partition_ckpt_heap_)
{
hash_partition_main_thread_id_ = mi_thread_id();
hash_partition_ckpt_heap_ = mi_heap_new();
+
+#if defined(WITH_JEMALLOC)
+ // create hash partition ckpt arena
+ size_t sz = sizeof(uint32_t);
+ if (mallctl("arenas.create",
+ &hash_partition_ckpt_arena_id_,
+ &sz,
+ NULL,
+ 0) != 0)
+ {
+ LOG(FATAL) << "Failed to create jemalloc arena for hash part heap";
+ }
+#endif
}
-
-#if defined(WITH_JEMALLOC)
- // create hash partition ckpt arena
- size_t sz = sizeof(uint32_t);
- if (mallctl("arenas.create",
- &hash_partition_ckpt_arena_id_,
- &sz,
- NULL,
- 0) != 0)
- {
- LOG(FATAL) << "Failed to create jemalloc arena for hash part heap";
- }
-#endif
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| std::unique_lock<std::mutex> lk(hash_partition_ckpt_heap_mux_); | |
| if (!hash_partition_ckpt_heap_) | |
| { | |
| hash_partition_main_thread_id_ = mi_thread_id(); | |
| hash_partition_ckpt_heap_ = mi_heap_new(); | |
| } | |
| void InitializeHashPartitionCkptHeap() | |
| { | |
| std::unique_lock<std::mutex> lk(hash_partition_ckpt_heap_mux_); | |
| if (!hash_partition_ckpt_heap_) | |
| { | |
| hash_partition_main_thread_id_ = mi_thread_id(); | |
| hash_partition_ckpt_heap_ = mi_heap_new(); | |
| `#if` defined(WITH_JEMALLOC) | |
| // create hash partition ckpt arena | |
| size_t sz = sizeof(uint32_t); | |
| if (mallctl("arenas.create", | |
| &hash_partition_ckpt_arena_id_, | |
| &sz, | |
| NULL, | |
| 0) != 0) | |
| { | |
| LOG(FATAL) << "Failed to create jemalloc arena for hash part heap"; | |
| } | |
| `#endif` | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@tx_service/include/cc/local_cc_shards.h` around lines 476 - 481, The
arenas.create call and assignment to hash_partition_ckpt_arena_id_ must be moved
inside the one-time init guard so it only runs when hash_partition_ckpt_heap_ is
null; currently arenas.create runs on every invocation and overwrites
hash_partition_ckpt_arena_id_, leaking prior arenas. Update the init sequence
around hash_partition_ckpt_heap_ and hash_partition_main_thread_id_ (the if
(!hash_partition_ckpt_heap_) block) to also call arenas.create and set
hash_partition_ckpt_arena_id_ there, ensuring subsequent calls skip arena
creation and do not mutate or leak the previous arena ID.
Comment on lines
+230
to
+232
| // pair. unique_ptr ensures pointer stability across map rehashing. | ||
| std::unordered_map<brpc::StreamId, std::unique_ptr<ConnectionInfo>> | ||
| inbound_connections_; |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
file="$(fd -t f 'log_replay_service.cpp' | head -n1)"
echo "Inspecting: ${file}"
# Verify pointer-escape patterns and erase sites in stream callbacks
rg -n -C5 'on_received_messages|on_idle_timeout|on_closed|it->second\.get\(\)|inbound_connections_\.erase\(' "$file"Repository: eloqdata/tx_service
Length of output: 3220
unique_ptr does not prevent use-after-free when raw pointers outlive inbound_mux_ lock release.
Callbacks in on_received_messages, on_idle_timeout, and on_closed extract raw pointers via it->second.get() under inbound_mux_ lock, then use those pointers after releasing the lock. Meanwhile, on_closed can erase the same entry and delete the object (line 850), causing a use-after-free. The comment's claim about rehash stability is unrelated to the actual safety issue.
💡 Suggested ownership fix at this declaration
- // pair. unique_ptr ensures pointer stability across map rehashing.
- std::unordered_map<brpc::StreamId, std::unique_ptr<ConnectionInfo>>
+ // pair. shared_ptr keeps ConnectionInfo alive even after erase().
+ std::unordered_map<brpc::StreamId, std::shared_ptr<ConnectionInfo>>
inbound_connections_;In each callback, capture std::shared_ptr<ConnectionInfo> under lock and use the local shared_ptr after unlock instead of extracting raw pointers.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@tx_service/include/fault/log_replay_service.h` around lines 230 - 232, The
map currently stores std::unique_ptr<ConnectionInfo> in inbound_connections_
which doesn't prevent use-after-free when callbacks (on_received_messages,
on_idle_timeout, on_closed) grab raw pointers under inbound_mux_ and use them
after unlocking; change the ownership to std::shared_ptr<ConnectionInfo> (i.e.,
make inbound_connections_ an unordered_map<brpc::StreamId,
std::shared_ptr<ConnectionInfo>>) and update all places that construct entries
to create shared_ptrs; then in each callback acquire a local
std::shared_ptr<ConnectionInfo> under inbound_mux_ (copying the map value) and
use that local shared_ptr after releasing the lock to keep the object alive.
liunyl
approved these changes
Mar 23, 2026
lokax
force-pushed
the
yf-pro-ckpt-boost-context
branch
from
March 25, 2026 08:12
305e7af to
21b539c
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Here are some reminders before you submit the pull request
fixes eloqdb/tx_service#issue_id./mtr --suite=mono_main,mono_multi,mono_basicSummary by CodeRabbit
Performance & Optimization
Reliability
Developer Experience
Observability