diff --git a/store_handler/data_store_service_client_closure.cpp b/store_handler/data_store_service_client_closure.cpp index a13ef568..738f1798 100644 --- a/store_handler/data_store_service_client_closure.cpp +++ b/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; diff --git a/store_handler/data_store_service_client_closure.h b/store_handler/data_store_service_client_closure.h index 95739480..13444ac7 100644 --- a/store_handler/data_store_service_client_closure.h +++ b/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_) diff --git a/tx_service/include/cc/cc_req_misc.h b/tx_service/include/cc/cc_req_misc.h index 5bd672c5..dc7c7277 100644 --- a/tx_service/include/cc/cc_req_misc.h +++ b/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 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,6 +578,19 @@ 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; @@ -584,33 +598,35 @@ struct FetchRecordCc : public FetchCc 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>> 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; diff --git a/tx_service/include/cc/cc_shard.h b/tx_service/include/cc/cc_shard.h index d499048e..b537309c 100644 --- a/tx_service/include/cc/cc_shard.h +++ b/tx_service/include/cc/cc_shard.h @@ -1351,8 +1351,11 @@ class CcShard std::unordered_map> fetch_reqs_; - // For load record from kvstore asynchronously - std::unordered_map 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 fetch_record_reqs_; + CcRequestPool fetch_record_cc_pool_; // For load snapshot from kvstore asynchronously CcRequestPool fetch_snapshot_cc_pool_; diff --git a/tx_service/src/cc/cc_req_misc.cpp b/tx_service/src/cc/cc_req_misc.cpp index 369ea3bc..cc53909f 100644 --- a/tx_service/src/cc/cc_req_misc.cpp +++ b/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) diff --git a/tx_service/src/cc/cc_shard.cpp b/tx_service/src/cc/cc_shard.cpp index 0cd84ca8..4def7898 100644 --- a/tx_service/src/cc/cc_shard.cpp +++ b/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,