mrhhsg commented on code in PR #63258:
URL: https://github.com/apache/doris/pull/63258#discussion_r4254638430
##########
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:
Fixed in a8061f1e2c3. The SHARED borrow now takes the payload pointer
through the const accessor (`const_cast<std::string*>(&src->column_values())`)
guarded by `DCHECK(src->has_column_values())`, so nothing on the shared holder
is written any more. The invariant holds because the sink only enqueues a
holder after `Block::serialize` succeeded, which always sets `column_values`;
`ArenaStringPtr::Get()` and `UnsafeMutablePointer()` return the same heap
pointer, so the borrow/release pairing is unchanged. This also removes the
same-value has-bit write that the multi-block broadcast path already had on
master.
##########
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:
Fixed in a8061f1e2c3. `Defer attach_back` is registered before the first
`release_column_values()` and iterates `payloads.size()`, so a placeholder
allocation that throws midway attaches every payload detached so far back to
its block; `set_allocated_column_values()` frees the placeholder it replaces,
so nothing leaks.
--
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]
