github-actions[bot] commented on code in PR #63258:
URL: https://github.com/apache/doris/pull/63258#discussion_r4254216551
##########
be/src/exec/operator/exchange_sink_buffer.cpp:
##########
@@ -49,6 +49,52 @@
namespace doris {
+void borrow_block(PBlock* src, PBlock* dst, BlockSource source) {
+ if (source == BlockSource::OWNED) {
+ // O(1) pointer swap; the owner is destroyed right after the send and
takes the request's
+ // cleared metas with it.
+ dst->mutable_column_metas()->Swap(src->mutable_column_metas());
+ } else {
+ dst->mutable_column_metas()->CopyFrom(src->column_metas());
+ }
+ // set_allocated_column_values() stores the owner's std::string pointer as
is; the thirdparty
+ // protobuf is built without NDEBUG checks, otherwise
ArenaStringPtr::SetAllocated would move the
+ // string into a fresh copy and delete the owner's one.
+ dst->set_be_exec_version(src->be_exec_version());
+ dst->set_compressed(src->compressed());
+ dst->set_compression_type(src->compression_type());
+ dst->set_uncompressed_size(src->uncompressed_size());
+ dst->set_allocated_column_values(src->mutable_column_values());
Review Comment:
[P2] Keep shared-source lookup free of mutable protobuf access
With `exchange_multi_blocks_byte_size = -1`, an unpartitioned exchange can
queue one `BroadcastPBlockHolder` for two busy destinations. Their previous RPC
completions then call `_send_rpc()` concurrently under different `RpcInstance`
mutexes and both reach this call. In pinned protobuf 21.11, the [mutable string
accessor](https://github.com/protocolbuffers/protobuf/blob/v21.11/src/google/protobuf/compiler/cpp/string_field.cc#L243-L275)
always [writes the non-atomic presence
bits](https://github.com/protocolbuffers/protobuf/blob/v21.11/src/google/protobuf/compiler/cpp/field.cc#L269-L279),
even when the payload is already present, producing a data race on the shared
owner. This is distinct from the earlier HTTP release/restore thread: that
payload mutation is fixed, but the old single-block branch only borrowed the
whole `PBlock`, and this helper adds source mutation even for small ordinary
RPCs. Obtain the already-initialized shared payload through the const
`column_values()` acces
sor and keep mutable protobuf operations on the destination request.
##########
be/src/util/proto_util.h:
##########
@@ -36,26 +42,128 @@ namespace doris {
// 2G: In the default "baidu_std" brpcd, upper limit of the request and
attachment length is 2G.
constexpr size_t MIN_HTTP_BRPC_SIZE = (1ULL << 31);
-// Embed column_values and brpc request serialization string in controller
attachment.
+// Attachment layout shared by every http send path:
+//
+// [int64 request_size][serialized request][int64 payload_size][payload]...
+//
+// The serialized request carries everything but the block payloads
(PBlock::column_values).
+// One payload section follows per payload block of the request, in block
order, see
+// payload_blocks(). A request with a single block therefore keeps the layout
older
+// receivers expect; receivers older than this layout read exactly one payload
into the
+// singular `block`, so a multi-payload attachment (repeated `blocks`) is only
understood by
+// receivers that carry it. Such requests could not be sent at all before
(plain brpc rejected
+// them for exceeding its size limit), so no working mixed-version case is
lost.
+
+// The blocks whose column_values travel as attachment payloads. A generic
request carries one
+// block; PTransmitDataParams carries either the singular `block` or the
repeated `blocks`.
+template <typename Params>
+std::vector<PBlock*> payload_blocks(Params* brpc_request) {
+ return {brpc_request->mutable_block()};
+}
+
+inline std::vector<PBlock*> payload_blocks(PTransmitDataParams* brpc_request) {
+ if (brpc_request->has_block()) {
+ DCHECK(brpc_request->blocks().empty());
+ return {brpc_request->mutable_block()};
+ }
+ std::vector<PBlock*> blocks;
+ blocks.reserve(brpc_request->blocks_size());
+ for (auto& block : *brpc_request->mutable_blocks()) {
+ blocks.push_back(&block);
+ }
+ return blocks;
+}
+
+// Append one `[int64 size][bytes]` section to the attachment.
+inline Status attachment_append_section(butil::IOBuf* attachment, const
std::string& data) {
+ int64_t data_size = data.size();
+ attachment->append(&data_size, sizeof(data_size));
+ try {
+ attachment->append(data);
+ } catch (...) {
+ LOG(WARNING) << "Try to alloc " << data_size
+ << " bytes for append data to attachment failed. ";
+ return Status::MemoryAllocFailed("request embed attachment failed to
memcpy {} bytes",
+ data_size);
+ }
+ return Status::OK();
+}
+
+// Embed `brpc_request` followed by `payloads`, one section each, in the
controller attachment.
+template <typename Params, typename Closure>
+Status request_embed_attachmentv2(Params* brpc_request,
+ const std::vector<const std::string*>&
payloads,
+ std::unique_ptr<Closure>& closure) {
+ std::string req_str;
+ if (!brpc_request->SerializeToString(&req_str)) {
+ return Status::InternalError("failed to serialize the request");
+ }
+ butil::IOBuf attachment;
+ RETURN_IF_ERROR(attachment_append_section(&attachment, req_str));
+ for (const auto* payload : payloads) {
+ RETURN_IF_ERROR(attachment_append_section(&attachment, *payload));
+ }
+ closure->cntl_->request_attachment().swap(attachment);
+ return Status::OK();
+}
+
+template <typename Params, typename Closure>
+Status request_embed_attachmentv2(Params* brpc_request, const std::string&
data,
+ std::unique_ptr<Closure>& closure) {
+ return request_embed_attachmentv2(brpc_request, std::vector<const
std::string*> {&data},
+ closure);
+}
+
+// Embed a request that owns its block: column_values is moved into the
attachment and the
+// request keeps the field present but empty.
template <typename Params, typename Closure>
Status request_embed_attachment_contain_blockv2(Params* brpc_request,
std::unique_ptr<Closure>&
closure) {
- std::string column_values =
std::move(*brpc_request->mutable_block()->mutable_column_values());
- brpc_request->mutable_block()->mutable_column_values()->clear();
+ auto* block = brpc_request->mutable_block();
+ std::string column_values = std::move(*block->mutable_column_values());
+ block->mutable_column_values()->clear();
return request_embed_attachmentv2(brpc_request, column_values, closure);
}
+// Embed a request whose blocks only borrow their column_values (see
+// ExchangeSinkBuffer / borrow_block()). Each payload is detached from the
request while the
+// request is serialized, appended by reference and attached back, so the
owner of the payload
+// is never read through the request's presence bits nor mutated: the same
owner may be
+// serialized by other channels at the same time.
+template <typename Closure>
+Status request_embed_attachment_borrowed_blocksv2(PTransmitDataParams*
brpc_request,
+ std::unique_ptr<Closure>&
closure) {
+ std::vector<PBlock*> blocks = payload_blocks(brpc_request);
+ std::vector<std::string*> payloads;
+ payloads.reserve(blocks.size());
+ for (auto* block : blocks) {
+ payloads.push_back(block->release_column_values());
+ DCHECK(payloads.back() != nullptr);
+ // Keep the field present on the wire: receivers that take the payload
slot before
+ // parsing the request rely on the serialized field to restore its
presence bit.
+ block->set_column_values("");
+ }
+ Defer attach_back([&]() {
Review Comment:
[P3] Install the restoration guard before creating placeholders
`release_column_values()` resets each request slot to its default, so the
following `set_column_values("")` [allocates a new empty string
object](https://github.com/protocolbuffers/protobuf/blob/v21.11/src/google/protobuf/arenastring.cc#L99-L128).
In a multi-block HTTP batch, if the first placeholder succeeds and a later
allocation throws, this `attach_back` guard has not been constructed. The
enclosing `_send_rpc()` guard then calls `release_borrowed_blocks()`, which
releases and discards the earlier request-owned placeholder as though it were a
borrowed owner string. The original payloads remain safe, but those empty
string objects have no remaining owner and leak during unwind. Register
restoration before the loop and restore the payloads already detached if a
later placeholder allocation fails.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
