Reduce EloqKV cold-read handoff overhead

store_handler/data_store_service_client_closure.cpp

@@ -168,14 +168,12 @@ void FetchRecordCallback(void *data,
     }
     else if (err_code == remote::DataStoreError::NO_ERROR)
     {
-        uint64_t now = txservice::LocalCcShards::ClockTsInMillseconds();
-        uint64_t rec_ttl = read_closure->Ttl();
-        std::string_view val = read_closure->Value();
-
         if (fetch_cc->table_name_.Engine() == txservice::TableEngine::EloqKv)
         {
             // Hash partition
-            if (rec_ttl > 0 && rec_ttl < now)
+            const uint64_t rec_ttl = read_closure->Ttl();
+            if (rec_ttl > 0 &&
+                rec_ttl < txservice::LocalCcShards::ClockTsInMillseconds())
             {
                 // expired record
                 fetch_cc->rec_status_ = txservice::RecordStatus::Deleted;
@@ -186,11 +184,12 @@ void FetchRecordCallback(void *data,
 
             fetch_cc->rec_status_ = txservice::RecordStatus::Normal;
             fetch_cc->rec_ts_ = read_closure->Ts();
-            fetch_cc->rec_str_.assign(val.data(), val.size());
+            fetch_cc->rec_str_ = read_closure->TakeValue();
             fetch_cc->SetFinish(0);
         }
         else
         {
+            const std::string_view val = read_closure->Value();
             // Range partition
             bool is_deleted = false;
             size_t offset = 0;

store_handler/data_store_service_client_closure.h

@@ -1018,6 +1018,18 @@ class ReadClosure : public ::google::protobuf::Closure, public Poolable
         }
     }
 
+    // Transfers the completed read buffer to its consumer. ReadClosure is
+    // returned to its pool after the callback, so retaining a second copy here
+    // only adds allocation and memcpy cost on the cold-read path.
+    std::string TakeValue()
+    {
+        if (is_local_request_)
+        {
+            return std::move(value_);
+        }
+        return std::move(*response_.mutable_value());
+    }
+
     uint64_t Ts() const
     {
         if (is_local_request_)

tx_service/include/cc/cc_req_misc.h

@@ -85,12 +85,13 @@ struct FetchCc : public CcRequestBase
     metrics::TimePoint start_;
 
 protected:
+    FetchCc() = default;
     FetchCc(CcShard &ccs, NodeGroupId cc_ng_id, int64_t cc_ng_term);
 
     std::vector<CcRequestBase *> requesters_;
-    CcShard &ccs_;
-    NodeGroupId cc_ng_id_;
-    int64_t cc_ng_term_;
+    CcShard *ccs_{nullptr};
+    NodeGroupId cc_ng_id_{0};
+    int64_t cc_ng_term_{-1};
 };
 
 struct FetchCatalogCc : public FetchCc
@@ -562,7 +563,7 @@ struct FillStoreSliceCc : public CcRequestBase
 struct FetchRecordCc : public FetchCc
 {
 public:
-    FetchRecordCc() = delete;
+    FetchRecordCc() = default;
     FetchRecordCc(const TableName *tbl_name,
                   const TableSchema *tbl_schema,
                   TxKey tx_key,
@@ -577,40 +578,55 @@ struct FetchRecordCc : public FetchCc
                   bool reopen = false);
     ~FetchRecordCc() = default;
 
+    void Reset(const TableName *tbl_name,
+               const TableSchema *tbl_schema,
+               TxKey tx_key,
+               LruEntry *cce,
+               CcShard &ccs,
+               NodeGroupId cc_ng_id,
+               int64_t cc_ng_term,
+               int32_t partition_id,
+               bool fetch_from_primary = false,
+               uint64_t snapshot_read_ts = 0,
+               bool only_fetch_archives = false,
+               bool reopen = false);
+
     bool ValidTermCheck();
 
     bool Execute(CcShard &ccs) override;
 
     void SetFinish(int err);
 
     // table_name is a string view, cannot access it outside TxProcessor.
-    TableName table_name_;
+    TableName table_name_{
+        std::string(""), TableType::Primary, TableEngine::None};
     const TableSchema *table_schema_{nullptr};
     std::string kv_table_name_;
     TxKey tx_key_;
     LruEntry *cce_{nullptr};
     KeyGapLockAndExtraData *lock_{nullptr};
     uint64_t rec_ts_{0};
-    RecordStatus rec_status_{RecordStatus::Unknown};
-    std::string rec_str_;
-    int error_code_{0};
-    int partition_id_;
-    bool fetch_from_primary_{false};
-
     // If set snapshot_read_ts_ (not equal 0), the snapshot_read_ts_ will be
     // used to fetch record from archives table.
     uint64_t snapshot_read_ts_{0};
-    // If set only_fetch_archives_ (true), don't fetch record from base table.
-    bool only_fetch_archives_{false};
-    bool reopen_{false};
     std::unique_ptr<
         std::vector<std::tuple<uint64_t, RecordStatus, std::string>>>
         archive_records_{nullptr};
 
+    std::string rec_str_;
+
     // These variables only be used in DataStoreHandler
     std::string kv_session_id_;
     std::string kv_start_key_;
     std::string kv_end_key_;
+
+    int error_code_{0};
+    int partition_id_{0};
+    RecordStatus rec_status_{RecordStatus::Unknown};
+    bool fetch_from_primary_{false};
+    // If set only_fetch_archives_ (true), don't fetch record from base table.
+    bool only_fetch_archives_{false};
+    bool reopen_{false};
 };
 
 struct FetchBucketDataCc;

tx_service/include/cc/cc_shard.h

@@ -1351,8 +1351,11 @@ class CcShard
 
     std::unordered_map<TableName, std::unique_ptr<FetchCc>> fetch_reqs_;
 
-    // For load record from kvstore asynchronously
-    std::unordered_map<LruEntry *, FetchRecordCc> fetch_record_reqs_;
+    // FetchRecordCc addresses must remain stable while data-store callbacks are
+    // in flight. The pool owns the requests and the flat map only indexes the
+    // active single-flight fetch for each entry.
+    absl::flat_hash_map<LruEntry *, FetchRecordCc *> fetch_record_reqs_;
+    CcRequestPool<FetchRecordCc> fetch_record_cc_pool_;
 
     // For load snapshot from kvstore asynchronously
     CcRequestPool<FetchSnapshotCc> fetch_snapshot_cc_pool_;

tx_service/src/cc/cc_req_misc.cpp

@@ -49,7 +49,7 @@
 namespace txservice
 {
 FetchCc::FetchCc(CcShard &ccs, NodeGroupId cc_ng_id, int64_t cc_ng_term)
-    : ccs_(ccs), cc_ng_id_(cc_ng_id), cc_ng_term_(cc_ng_term)
+    : ccs_(&ccs), cc_ng_id_(cc_ng_id), cc_ng_term_(cc_ng_term)
 {
 }
 
@@ -187,7 +187,7 @@ void FetchCatalogCc::SetFinish(RecordStatus status, int err)
         commit_ts_ = 0;
         catalog_image_.clear();
     });
-    ccs_.Enqueue(this);
+    ccs_->Enqueue(this);
 }
 
 FetchTableStatisticsCc::FetchTableStatisticsCc(const TableName &table_name,
@@ -256,7 +256,7 @@ void FetchTableStatisticsCc::SetFinish(int err)
         current_version_ = 0;
         sample_pool_map_.clear();
     });
-    ccs_.Enqueue(this);
+    ccs_->Enqueue(this);
 }
 
 FetchTableRangesCc::FetchTableRangesCc(const TableName &table_name,
@@ -343,7 +343,7 @@ void FetchTableRangesCc::SetFinish(int err)
         ranges_vec_.clear();
         partition_ranges_vec_.clear();
     });
-    ccs_.Enqueue(this);
+    ccs_->Enqueue(this);
 }
 
 void FetchTableRangesCc::Merge()
@@ -810,14 +810,59 @@ FetchRecordCc::FetchRecordCc(const TableName *tbl_name,
       tx_key_(std::move(tx_key)),
       cce_(cce),
       lock_(cce->GetKeyGapLockAndExtraData()),
+      snapshot_read_ts_(snapshot_read_ts),
       partition_id_(partition_id),
       fetch_from_primary_(fetch_from_primary),
-      snapshot_read_ts_(snapshot_read_ts),
       only_fetch_archives_(only_fetch_archives),
       reopen_(reopen)
 {
 }
 
+void FetchRecordCc::Reset(const TableName *tbl_name,
+                          const TableSchema *tbl_schema,
+                          TxKey tx_key,
+                          LruEntry *cce,
+                          CcShard &ccs,
+                          NodeGroupId cc_ng_id,
+                          int64_t cc_ng_term,
+                          int32_t partition_id,
+                          bool fetch_from_primary,
+                          uint64_t snapshot_read_ts,
+                          bool only_fetch_archives,
+                          bool reopen)
+{
+    // A request never migrates between shards: its pool and active-fetch map
+    // are both owned by the shard captured on first use.
+    assert(ccs_ == nullptr || ccs_ == &ccs);
+    ccs_ = &ccs;
+    cc_ng_id_ = cc_ng_id;
+    cc_ng_term_ = cc_ng_term;
+    requesters_.clear();
+    start_ = metrics::TimePoint{};
+
+    table_name_ =
+        TableName(tbl_name->StringView(), tbl_name->Type(), tbl_name->Engine());
+    table_schema_ = tbl_schema;
+    kv_table_name_ =
+        table_schema_->GetKVCatalogInfo()->GetKvTableName(table_name_);
+    tx_key_ = std::move(tx_key);
+    cce_ = cce;
+    lock_ = cce->GetKeyGapLockAndExtraData();
+    rec_ts_ = 0;
+    snapshot_read_ts_ = snapshot_read_ts;
+    archive_records_.reset();
+    rec_str_.clear();
+    kv_session_id_.clear();
+    kv_start_key_.clear();
+    kv_end_key_.clear();
+    error_code_ = 0;
+    partition_id_ = partition_id;
+    rec_status_ = RecordStatus::Unknown;
+    fetch_from_primary_ = fetch_from_primary;
+    only_fetch_archives_ = only_fetch_archives;
+    reopen_ = reopen;
+}
+
 bool FetchRecordCc::ValidTermCheck()
 {
     if (fetch_from_primary_)
@@ -987,7 +1032,7 @@ bool FetchRecordCc::Execute(CcShard &ccs)
 void FetchRecordCc::SetFinish(int err)
 {
     error_code_ = err;
-    ccs_.Enqueue(this);
+    ccs_->Enqueue(this);
 }
 
 bool RecoverDeadTxCc::Execute(CcShard &ccs)

tx_service/src/cc/cc_shard.cpp

@@ -130,6 +130,8 @@ CcShard::CcShard(
       system_handler_(system_handler),
       active_si_txs_()
 {
+    fetch_record_reqs_.reserve(64);
+
     // Reserve range_slice_memory_limit_percent% for range slice info.
     // We update this to dynamically reserve the configured range slice
     // percentage.
@@ -2184,20 +2186,32 @@ store::DataStoreHandler::DataStoreOpStatus CcShard::FetchRecord(
     bool only_fetch_archives,
     bool reopen)
 {
-    auto tab_it = fetch_record_reqs_.try_emplace(cce,
-                                                 &table_name,
-                                                 tbl_schema,
-                                                 std::move(key),
-                                                 cce,
-                                                 *this,
-                                                 cc_ng_id,
-                                                 cc_ng_term,
-                                                 partition_id,
-                                                 fetch_from_primary,
-                                                 snapshot_read_ts,
-                                                 only_fetch_archives,
-                                                 reopen);
-    FetchRecordCc *fetch_req = &(tab_it.first->second);
+    auto tab_it = fetch_record_reqs_.find(cce);
+    FetchRecordCc *fetch_req = nullptr;
+    if (tab_it == fetch_record_reqs_.end())
+    {
+        fetch_req = fetch_record_cc_pool_.NextRequest();
+        fetch_req->Reset(&table_name,
+                         tbl_schema,
+                         std::move(key),
+                         cce,
+                         *this,
+                         cc_ng_id,
+                         cc_ng_term,
+                         partition_id,
+                         fetch_from_primary,
+                         snapshot_read_ts,
+                         only_fetch_archives,
+                         reopen);
+        const bool inserted =
+            fetch_record_reqs_.try_emplace(cce, fetch_req).second;
+        assert(inserted);
+        (void) inserted;
+    }
+    else
+    {
+        fetch_req = tab_it->second;
+    }
     fetch_req->AddRequester(requester);
 
     CODE_FAULT_INJECTOR("disable_fetch_record", {
@@ -2395,7 +2409,15 @@ store::DataStoreHandler::DataStoreOpStatus CcShard::FetchBucketData(
 
 void CcShard::RemoveFetchRecordRequest(LruEntry *cce)
 {
-    fetch_record_reqs_.erase(cce);
+    auto fetch_it = fetch_record_reqs_.find(cce);
+    assert(fetch_it != fetch_record_reqs_.end());
+    FetchRecordCc *fetch_req = fetch_it->second;
+    fetch_record_reqs_.erase(fetch_it);
+
+    // Free marks the request reusable while its Execute call is unwinding, but
+    // both FetchRecord and resumed requesters run on this shard, so NextRequest
+    // cannot observe it until control returns to the shard loop.
+    fetch_req->Free();
 }
 
 CcMap *CcShard::CreateOrUpdatePkCcMap(const TableName &table_name,