From e643c3ee7542262c2065e948d3333d62cccc6fe2 Mon Sep 17 00:00:00 2001 From: liunyl Date: Sun, 30 Aug 2026 16:20:37 +0000 Subject: [PATCH] fix: clear recycled CC message payloads --- docs/06-distribution-and-clustering.md | 2 +- tx_service/include/remote/cc_message_pool.h | 77 +++++++++++ .../include/remote/cc_stream_receiver.h | 7 +- tx_service/include/remote/cc_stream_sender.h | 6 +- tx_service/include/sharder.h | 3 +- tx_service/src/remote/cc_stream_receiver.cpp | 127 ++++++++---------- tx_service/src/remote/cc_stream_sender.cpp | 5 +- tx_service/tests/CMakeLists.txt | 1 + tx_service/tests/CcMessagePool-Test.cpp | 57 ++++++++ 9 files changed, 205 insertions(+), 80 deletions(-) create mode 100644 tx_service/include/remote/cc_message_pool.h create mode 100644 tx_service/tests/CcMessagePool-Test.cpp diff --git a/docs/06-distribution-and-clustering.md b/docs/06-distribution-and-clustering.md index faf6f4cbd..8d08d687d 100644 --- a/docs/06-distribution-and-clustering.md +++ b/docs/06-distribution-and-clustering.md @@ -166,7 +166,7 @@ High-frequency cc traffic rides `CcMessage` (proto: `proto/cc_request.proto:1407 - **`CcStreamSender`** (`remote/cc_stream_sender.h`): per remote node keeps a *regular* stream and a *long-message* stream (for slow-to-deserialize payloads like `ScanSliceResponse`, so they don't head-of-line-block small messages). A dedicated `connect_thd_` (re)establishes streams (`Connect` RPC of `CcStreamService` then brpc stream accept); each stream carries a version number to serialize concurrent reconnects, plus the peer IP so an IP change forces reconnect (`UpdateStreamIP`). `SendMessageToNg` resolves the NG leader via `ng_leader_cache_` first. - **Resend logic**: `SendMessageResult{sent, queued_for_retry, need_reconnect}`. Failed writes are queued per node (`resend_message_list_`, EAGAIN cases in `eagain_resend_*`) and re-driven by `resend_thd_`; a hard failure flags the stream for reconnect. Exception: `SendStandbyMessageToNode` is deliberately best-effort (no retry queue) because standby replication has its own sequence-based resend (see standby doc §4.4). -- **`CcStreamReceiver`** (`remote/cc_stream_receiver.h`): a `brpc::StreamInputHandler` that parses inbound messages from a pooled `msg_pool_` (zero-alloc steady state) and dispatches in `OnReceiveCcMsg`: +- **`CcStreamReceiver`** (`remote/cc_stream_receiver.h`): a `brpc::StreamInputHandler` that acquires reusable protobuf shells from `CcMessagePool`, parses inbound messages, and dispatches them in `OnReceiveCcMsg`. Every recycle clears the protobuf before publishing it back to the concurrent pool, so idle messages do not retain payload allocations that the next `ParseFrom` would discard anyway: - **Requests** (Acquire/Read/Apply/ScanNext/PostWriteAll/...) are wrapped in pooled `Remote*` cc-request objects (`remote/remote_cc_request.h`, e.g. `RemoteAcquire : AcquireCc`) and enqueued to the proper CcShard — from a shard's point of view a remote request is indistinguishable from a local one (see 03-concurrency-control.md). - **Responses** must find their transaction: the message carries `tx_number`, `txm_addr`, `handler_addr`, `command_id`, `tx_term`. The receiver first fences with `CheckLeaderTerm(tx_node_id, tx_term)` (tx node id is encoded in the tx number: `(tx_number >> 32) >> 10`), then dereferences `txm_addr`, takes the txm's **shared forward latch** (`AcquireSharedForwardLatch`), and re-validates `txm->TxNumber() == msg.tx_number && txm->CommandId() == msg.command_id` before touching the `CcHandlerResult` at `handler_addr` (`cc_stream_receiver.cpp:332-400`). The tx_number+command_id check makes stale/duplicate responses for a recycled txm harmless; the latch keeps the txm from being reused mid-update. - Locking-protocol requests that block remotely send an **acknowledgement** (`is_ack=true` responses carry the remote cce lock address) so the sender can later probe blocked requests (`BlockedCcReqCheckRequest`). diff --git a/tx_service/include/remote/cc_message_pool.h b/tx_service/include/remote/cc_message_pool.h new file mode 100644 index 000000000..654e7ed28 --- /dev/null +++ b/tx_service/include/remote/cc_message_pool.h @@ -0,0 +1,77 @@ +/** + * Copyright (C) 2025 EloqData Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under either of the following two licenses: + * 1. GNU Affero General Public License, version 3, as published by the Free + * Software Foundation. + * 2. GNU General Public License as published by the Free Software + * Foundation; version 2 of the License. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License or GNU General Public License for more + * details. + * + * You should have received a copy of the GNU Affero General Public License + * and GNU General Public License V2 along with this program. If not, see + * . + */ +#pragma once + +#include + +#include +#include + +#include "proto/cc_request.pb.h" + +namespace txservice::remote +{ +/** + * @brief Thread-safe reuse pool for CC protobuf messages. + * + * Recycle clears a message before making it available to another thread. This + * keeps the idle pool from retaining payload allocations that ParseFrom would + * discard before the next use anyway. + */ +class CcMessagePool +{ +public: + CcMessagePool() = default; + CcMessagePool(const CcMessagePool &) = delete; + CcMessagePool &operator=(const CcMessagePool &) = delete; + CcMessagePool(CcMessagePool &&) = delete; + CcMessagePool &operator=(CcMessagePool &&) = delete; + + /** + * @brief Returns an empty pooled message or allocates a new one. + * + * The caller owns the returned message until passing it to Recycle(). + */ + std::unique_ptr Acquire() + { + std::unique_ptr msg; + if (!pool_.try_dequeue(msg)) + { + msg = std::make_unique(); + } + return msg; + } + + /** + * @brief Clears and transfers a message back to the shared pool. + * + * @param msg A non-null message exclusively owned by the caller. + */ + void Recycle(std::unique_ptr msg) + { + msg->Clear(); + pool_.enqueue(std::move(msg)); + } + +private: + moodycamel::ConcurrentQueue> pool_; +}; +} // namespace txservice::remote diff --git a/tx_service/include/remote/cc_stream_receiver.h b/tx_service/include/remote/cc_stream_receiver.h index 1b660bdf9..20ed7cba2 100644 --- a/tx_service/include/remote/cc_stream_receiver.h +++ b/tx_service/include/remote/cc_stream_receiver.h @@ -31,6 +31,7 @@ #include "cc_req_pool.h" #include "proto/cc_request.pb.h" +#include "remote/cc_message_pool.h" #include "tx_record.h" #include "type.h" @@ -43,9 +44,7 @@ namespace remote class CcStreamReceiver : public brpc::StreamInputHandler, public CcStreamService { public: - explicit CcStreamReceiver( - LocalCcShards &local_shards, - moodycamel::ConcurrentQueue> &msg_pool); + CcStreamReceiver(LocalCcShards &local_shards, CcMessagePool &msg_pool); ~CcStreamReceiver() = default; void Shutdown(); @@ -82,7 +81,7 @@ class CcStreamReceiver : public brpc::StreamInputHandler, public CcStreamService // receives a message, de-serializes it and dispatches it to local shards // for processing. The message is put back into the pool after the cc // request is processed. - moodycamel::ConcurrentQueue> &msg_pool_; + CcMessagePool &msg_pool_; moodycamel::ConcurrentQueue> scan_resp_pool_; }; diff --git a/tx_service/include/remote/cc_stream_sender.h b/tx_service/include/remote/cc_stream_sender.h index 5bb2f1d35..41a82a188 100644 --- a/tx_service/include/remote/cc_stream_sender.h +++ b/tx_service/include/remote/cc_stream_sender.h @@ -34,6 +34,7 @@ #include "cc/cc_handler_result.h" #include "proto/cc_request.pb.h" +#include "remote/cc_message_pool.h" #include "sharder.h" namespace txservice @@ -96,8 +97,7 @@ struct SendMessageResult class CcStreamSender { public: - CcStreamSender( - moodycamel::ConcurrentQueue> &msg_pool); + explicit CcStreamSender(CcMessagePool &msg_pool); ~CcStreamSender(); void RecycleCcMsg(std::unique_ptr msg); @@ -169,7 +169,7 @@ class CcStreamSender int ConnectStream(uint32_t node_id, int64_t version); int ConnectLongMsgStream(uint32_t node_id, int64_t version); - moodycamel::ConcurrentQueue> &msg_pool_; + CcMessagePool &msg_pool_; brpc::StreamWriteOptions stream_write_options_; diff --git a/tx_service/include/sharder.h b/tx_service/include/sharder.h index 77f60fdaf..52ae9d2e4 100644 --- a/tx_service/include/sharder.h +++ b/tx_service/include/sharder.h @@ -37,6 +37,7 @@ #include "brpc/server.h" #include "butil/third_party/murmurhash3/murmurhash3.h" #include "proto/cc_request.pb.h" +#include "remote/cc_message_pool.h" #include "tx_serialize.h" #include "txlog.h" #include "type.h" @@ -839,7 +840,7 @@ class Sharder */ std::atomic cc_nodes_init_{false}; - moodycamel::ConcurrentQueue> msg_pool_; + remote::CcMessagePool msg_pool_; // The cc stream sender establishes connections to remote nodes and // sends cc requests and responses to remote nodes via streams. It is diff --git a/tx_service/src/remote/cc_stream_receiver.cpp b/tx_service/src/remote/cc_stream_receiver.cpp index 3858074ae..d7aa97829 100644 --- a/tx_service/src/remote/cc_stream_receiver.cpp +++ b/tx_service/src/remote/cc_stream_receiver.cpp @@ -110,9 +110,8 @@ thread_local CcRequestPool key_obj_standby_forward_pool_; thread_local CcRequestPool parse_standby_forward_pool_; -CcStreamReceiver::CcStreamReceiver( - LocalCcShards &local_shards, - moodycamel::ConcurrentQueue> &msg_pool) +CcStreamReceiver::CcStreamReceiver(LocalCcShards &local_shards, + CcMessagePool &msg_pool) : local_shards_(local_shards), msg_pool_(msg_pool) { } @@ -278,7 +277,7 @@ int CcStreamReceiver::on_received_messages(brpc::StreamId stream_id, remote::CcMessage::MessageType:: CcMessage_MessageType_KeyObjectStandbyForwardRequest) { - msg_pool_.enqueue(std::move(cc_msg)); + msg_pool_.Recycle(std::move(cc_msg)); continue; } OnReceiveCcMsg(std::move(cc_msg)); @@ -326,15 +325,7 @@ void CcStreamReceiver::on_closed(brpc::StreamId stream) std::unique_ptr CcStreamReceiver::GetCcMsg() { - std::unique_ptr msg; - if (msg_pool_.try_dequeue(msg)) - { - return msg; - } - else - { - return std::make_unique(); - } + return msg_pool_.Acquire(); } std::unique_ptr CcStreamReceiver::GetScanSliceResp() @@ -493,7 +484,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term)) { // The tx node has failed. Pointer stability does not hold anymore. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } else @@ -512,7 +503,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) // The original tx has terminated and the tx machine has been // recycled. The response message is directed to an obsolete tx. // Skips setting the cc handler result. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); txm->ReleaseSharedForwardLatch(); break; } @@ -525,7 +516,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) if (acq_res.IsRemoteHdResultSet(std::memory_order_acquire)) { - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); txm->ReleaseSharedForwardLatch(); break; } @@ -583,7 +574,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) txm->ReleaseSharedForwardLatch(); } - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } case CcMessage::MessageType::CcMessage_MessageType_AcquireAllRequest: @@ -606,7 +597,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term)) { // The tx node has failed. Pointer stability does not hold anymore. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } else @@ -623,7 +614,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) // The original tx has terminated and the tx machine has been // recycled. The response message is directed to an obsolete tx. // Skips setting the cc handler result. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); txm->ReleaseSharedForwardLatch(); break; } @@ -685,7 +676,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) txm->ReleaseSharedForwardLatch(); } - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } case CcMessage::MessageType::CcMessage_MessageType_ValidateRequest: @@ -725,7 +716,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) Sharder::Instance().GetCcStreamSender(); cc_stream_sender->SendMessageToNode(req.src_node_id(), return_msg); - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); } else { @@ -749,7 +740,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term)) { // The tx node has failed. Pointer stability does not hold anymore. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } else @@ -767,7 +758,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) // The original tx has terminated and the tx machine has been // recycled. The response message is directed to an obsolete tx. // Skips setting the cc handler result. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); txm->ReleaseSharedForwardLatch(); break; } @@ -791,7 +782,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) txm->ReleaseSharedForwardLatch(); - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } case CcMessage::MessageType::CcMessage_MessageType_PostprocessResponse: @@ -806,7 +797,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term)) { // The tx node has failed. Pointer stability does not hold anymore. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } else @@ -824,7 +815,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) // The original tx has terminated and the tx machine has been // recycled. The response message is directed to an obsolete tx. // Skips setting the cc handler result. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); txm->ReleaseSharedForwardLatch(); break; } @@ -845,7 +836,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) txm->ReleaseSharedForwardLatch(); - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } case CcMessage::MessageType::CcMessage_MessageType_ReadRequest: @@ -880,7 +871,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term)) { // The tx node has failed. Pointer stability does not hold anymore. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } else @@ -900,7 +891,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) // The original tx has terminated and the tx machine has been // recycled. The response message is directed to an obsolete tx. // Skips setting the cc handler result. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); txm->ReleaseSharedForwardLatch(); break; } @@ -950,7 +941,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) txm->ReleaseSharedForwardLatch(); } - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } case CcMessage::MessageType::CcMessage_MessageType_PostCommitRequest: @@ -990,7 +981,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) Sharder::Instance().GetCcStreamSender(); cc_stream_sender->SendMessageToNode(post_commit.src_node_id(), return_msg); - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); } else { @@ -1026,7 +1017,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) Sharder::Instance().GetCcStreamSender(); cc_stream_sender->SendMessageToNode(post_commit.src_node_id(), return_msg); - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); } else { @@ -1050,7 +1041,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) { assert(false && "Unimplemented"); LOG(ERROR) << "ScanOpenRequest is unsupported; replying error."; - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } case CcMessage::MessageType::CcMessage_MessageType_ScanOpenResponse: @@ -1064,7 +1055,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) assert(false && "Unimplemented"); LOG(ERROR) << "ScanOpenResponse is unsupported; replying error."; - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } case CcMessage::MessageType::CcMessage_MessageType_ScanNextRequest: @@ -1098,7 +1089,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term)) { // The tx node has failed. Pointer stability does not hold anymore. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } else @@ -1116,7 +1107,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) // The original tx has terminated and the tx machine has been // recycled. The response message is directed to an obsolete tx. // Skips setting the cc handler result. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); txm->ReleaseSharedForwardLatch(); break; } @@ -1137,7 +1128,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) // The handler result has been reset while this response is in // transit. Drops this stale response to avoid dereferencing a // recycled scanner pointer. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); txm->ReleaseSharedForwardLatch(); break; } @@ -1278,7 +1269,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) txm->ReleaseSharedForwardLatch(); - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } case CcMessage::MessageType::CcMessage_MessageType_ScanSliceRequest: @@ -1310,7 +1301,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) int64_t tx_term = msg->tx_term(); if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term)) { - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } else @@ -1328,7 +1319,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) // The original tx has terminated and the tx machine has been // recycled. The response message is directed to an obsolete tx. // Skips setting the cc handler result. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); txm->ReleaseSharedForwardLatch(); break; } @@ -1347,7 +1338,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) } txm->ReleaseSharedForwardLatch(); - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } case CcMessage::MessageType::CcMessage_MessageType_FaultInjectRequest: @@ -1370,7 +1361,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term)) { // The tx node has failed. Pointer stability does not hold anymore. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } CcHandlerResult *hd_res = @@ -1387,7 +1378,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) // The original tx has terminated and the tx machine has been // recycled. The response message is directed to an obsolete tx. // Skips setting the cc handler result. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); txm->ReleaseSharedForwardLatch(); break; } @@ -1406,7 +1397,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) } txm->ReleaseSharedForwardLatch(); - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } case CcMessage::MessageType::CcMessage_MessageType_AnalyzeTableAllRequest: @@ -1429,7 +1420,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) int64_t tx_term = msg->tx_term(); if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term)) { - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } else @@ -1447,7 +1438,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) // The original tx has terminated and the tx machine has been // recycled. The response message is directed to an obsolete tx. // Skips setting the cc handler result. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); txm->ReleaseSharedForwardLatch(); break; } @@ -1467,7 +1458,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) } txm->ReleaseSharedForwardLatch(); - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } case CcMessage::MessageType:: @@ -1492,7 +1483,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term)) { // The tx node has failed. Pointer stability does not hold anymore. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } CcHandlerResult *hd_res = @@ -1511,7 +1502,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) hd_res->SetFinished(); } - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } case CcMessage::MessageType::CcMessage_MessageType_RecoverStateCheckRequest: @@ -1545,7 +1536,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) Sharder::Instance().GetCcStreamSender()->SendMessageToNode( req.src_node_id(), send_msg); - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } case CcMessage::MessageType:: @@ -1565,7 +1556,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) << ", error_code:" << resp.error_code(); Sharder::Instance().UpdateLeader(resp.node_group_id()); } - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } case CcMessage::MessageType::CcMessage_MessageType_DeadLockRequest: @@ -1584,7 +1575,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) { const DeadLockResponse &rsp = msg->dead_lock_response(); DeadLockCheck::MergeRemoteWaitingLockInfo(&rsp); - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } case CcMessage::MessageType:: @@ -1612,7 +1603,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) } case CcMessage::MessageType::CcMessage_MessageType_AbortTransactionResponse: { - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } case CcMessage::MessageType::CcMessage_MessageType_BlockedCcReqCheckRequest: @@ -1640,7 +1631,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term)) { // The tx node has failed. Pointer stability does not hold anymore. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } @@ -1664,7 +1655,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) // The original tx has terminated and the tx machine has been // recycled. The response message is directed to an obsolete tx. // Skips setting the cc handler result. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); txm->ReleaseSharedForwardLatch(); break; } @@ -1674,7 +1665,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) if (acq_key_result_vec.at(resp.acq_key_result_vec_idx()) .IsRemoteHdResultSet(std::memory_order_acquire)) { - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); txm->ReleaseSharedForwardLatch(); break; } @@ -1716,7 +1707,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) // The original tx has terminated and the tx machine has been // recycled. The response message is directed to an obsolete tx. // Skips setting the cc handler result. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); txm->ReleaseSharedForwardLatch(); break; } @@ -1745,7 +1736,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) txm->ReleaseSharedForwardLatch(); } - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } case CcMessage::MessageType::CcMessage_MessageType_KickoutDataRequest: @@ -1787,7 +1778,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) { // The tx coordinator node has failed. Pointer stability // does not hold anymore. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); LOG(ERROR) << "Receive remote kickoutccentry response, but tx" " coordinator has failed."; break; @@ -1807,7 +1798,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) // The original tx has terminated and the tx state machine has // been recycled. The response message is directed to an // obsolete tx. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); txm->ReleaseSharedForwardLatch(); break; } @@ -1830,7 +1821,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) txm->ReleaseSharedForwardLatch(); // Recycle the cc message - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } case CcMessage::MessageType::CcMessage_MessageType_ApplyRequest: @@ -1853,7 +1844,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term)) { // The tx node has failed. Pointer stability does not hold anymore. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } else @@ -1871,7 +1862,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) // The original tx has terminated and the tx machine has been // recycled. The response message is directed to an obsolete tx. // Skips setting the cc handler result. - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); txm->ReleaseSharedForwardLatch(); break; } @@ -1905,7 +1896,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) } } txm->ReleaseSharedForwardLatch(); - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } case CcMessage::MessageType::CcMessage_MessageType_PublishRequest: @@ -1953,7 +1944,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) } dbcc->AddRemoteObjSize(resp.dbsize_term(), total_obj_sizes); - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } case CcMessage::MessageType:: @@ -1992,7 +1983,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) int64_t tx_term = msg->tx_term(); if (!Sharder::Instance().CheckLeaderTerm(tx_node_id, tx_term)) { - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } else @@ -2007,7 +1998,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) if (txm->TxNumber() != msg->tx_number() || txm->CommandId() != msg->command_id()) { - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); txm->ReleaseSharedForwardLatch(); break; } @@ -2027,7 +2018,7 @@ void CcStreamReceiver::OnReceiveCcMsg(std::unique_ptr msg) } txm->ReleaseSharedForwardLatch(); - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); break; } default: diff --git a/tx_service/src/remote/cc_stream_sender.cpp b/tx_service/src/remote/cc_stream_sender.cpp index c28237c96..b52dc4ec9 100644 --- a/tx_service/src/remote/cc_stream_sender.cpp +++ b/tx_service/src/remote/cc_stream_sender.cpp @@ -52,8 +52,7 @@ CcStreamSender::~CcStreamSender() resend_thd_.join(); } -CcStreamSender::CcStreamSender( - moodycamel::ConcurrentQueue> &msg_pool) +CcStreamSender::CcStreamSender(CcMessagePool &msg_pool) : msg_pool_(msg_pool), terminate_(false), to_connect_flag_(false) { stream_write_options_.write_in_background = true; @@ -65,7 +64,7 @@ CcStreamSender::CcStreamSender( void CcStreamSender::RecycleCcMsg(std::unique_ptr msg) { - msg_pool_.enqueue(std::move(msg)); + msg_pool_.Recycle(std::move(msg)); } void CcStreamSender::ReConnectStream(uint32_t dest_node_id) diff --git a/tx_service/tests/CMakeLists.txt b/tx_service/tests/CMakeLists.txt index cec3ae450..66898a888 100644 --- a/tx_service/tests/CMakeLists.txt +++ b/tx_service/tests/CMakeLists.txt @@ -59,6 +59,7 @@ set_property(TARGET txnode PROPERTY CXX_STANDARD 20) # files). They link Catch2::Catch2WithMain. set(CATCH_MAIN_TESTS CcEntry-Test + CcMessagePool-Test CcPage-Test LargeObjLRU-Test CcRequestWait-Test diff --git a/tx_service/tests/CcMessagePool-Test.cpp b/tx_service/tests/CcMessagePool-Test.cpp new file mode 100644 index 000000000..74cd8b779 --- /dev/null +++ b/tx_service/tests/CcMessagePool-Test.cpp @@ -0,0 +1,57 @@ +/** + * Copyright (C) 2025 EloqData Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under either of the following two licenses: + * 1. GNU Affero General Public License, version 3, as published by the Free + * Software Foundation. + * 2. GNU General Public License as published by the Free Software + * Foundation; version 2 of the License. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License or GNU General Public License for more + * details. + * + * You should have received a copy of the GNU Affero General Public License + * and GNU General Public License V2 along with this program. If not, see + * . + */ +#include +#include +#include +#include + +/** Let Catch provide main(). */ +#include + +#include "remote/cc_message_pool.h" + +namespace txservice::remote +{ +TEST_CASE("CcMessagePool clears payloads before recycling", "[cc-message-pool]") +{ + constexpr size_t payload_size = 64 * 1024; + + CcMessagePool pool; + std::unique_ptr msg = pool.Acquire(); + CcMessage *original = msg.get(); + msg->set_type(CcMessage::KeyObjectStandbyForwardRequest); + msg->set_tx_number(42); + KeyObjectStandbyForwardRequest *request = + msg->mutable_key_obj_standby_forward_req(); + request->set_key(std::string(payload_size, 'k')); + request->add_cmd_list(std::string(payload_size, 'c')); + const size_t populated_space = msg->SpaceUsedLong(); + + pool.Recycle(std::move(msg)); + std::unique_ptr recycled = pool.Acquire(); + + REQUIRE(recycled.get() == original); + REQUIRE(recycled->content_case() == CcMessage::CONTENT_NOT_SET); + REQUIRE(recycled->type() == CcMessage::AcquireRequest); + REQUIRE(recycled->tx_number() == 0); + REQUIRE(recycled->SpaceUsedLong() < populated_space / 2); +} +} // namespace txservice::remote