github-actions[bot] commented on code in PR #67759:
URL: https://github.com/apache/doris/pull/67759#discussion_r3980257147
##########
be/src/load/channel/load_channel_mgr.cpp:
##########
@@ -177,7 +186,12 @@ Status LoadChannelMgr::add_batch(const
PTabletWriterAddBlockRequest& request,
// this case will be handled in load channel's add batch method.
Status st = channel->add_batch(request, response);
if (UNLIKELY(!st.ok())) {
- RETURN_IF_ERROR(channel->cancel());
+ // Release the manager's ownership too. The last in-flight request
+ // releases the writers with the channel.
+ PTabletWriterCancelRequest cancel_request;
+ *cancel_request.mutable_id() = request.id();
+ cancel_request.set_cancel_reason(st.to_string());
+ RETURN_IF_ERROR(cancel(cancel_request));
Review Comment:
[P1] Bind the failure transition to the retained channel instance
At this point this RPC has already observed `st`, but it re-enters the
manager by load ID without first publishing `st` to `channel`. A concurrent
final EOS can take `_lock`, still see that same channel as uncancelled, and
install a success tombstone; the later cancel then finds no live entry and
preserves that success. Conversely, if timeout cleanup removed X and an open
created Y with the same ID, this stale RPC cancels Y instead of X. Both
orderings are reachable from independent heavy-work-pool RPCs. Please publish
the error on the captured channel and make the manager transition compare the
current mapping with that exact instance (or generation) before erasing/caching
it. A barrier test should cover failure versus final EOS and replacement open.
##########
be/src/cloud/cloud_delta_writer.cpp:
##########
@@ -154,6 +157,7 @@ void CloudDeltaWriter::update_tablet_stats() {
Status CloudDeltaWriter::commit_rowset() {
g_cloud_commit_rowset_count << 1;
std::lock_guard<bthread::Mutex> lock(_mtx);
+ RETURN_IF_ERROR(_get_load_cancel_status());
Review Comment:
[P2] Recheck cancellation before the empty-rowset commit
This is the only check before `_commit_empty_rowset()`, but
`CloudRowsetBuilder::init()` ends with a separate, potentially blocking
`prepare_rowset` RPC. If cancellation is published while prepare is in flight,
the helper returns from init, builds the rowset, and starts the later metadata
`commit_rowset` RPC without another status load. That creates avoidable
post-cancel metadata work and can extend close/final-owner latency while
transaction abort is already under way. Please recheck after init/build and
before the distinct commit RPC; a delayed-prepare test should assert that
cancellation prevents commit.
##########
be/src/load/channel/load_channel.cpp:
##########
@@ -304,13 +310,18 @@ bool LoadChannel::is_finished() {
return _tablets_channels.empty();
}
-Status LoadChannel::cancel() {
- _cancelled.store(true);
- std::lock_guard<std::mutex> l(_lock);
- for (auto& it : _tablets_channels) {
- static_cast<void>(it.second->cancel());
- }
+Status LoadChannel::cancel(const Status& reason) {
+ DCHECK(!reason.ok());
+ _cancel_status->update(reason);
Review Comment:
[P2] Make pressure flushing observe the shared load status
This publisher no longer traverses writers, so
`MemTableWriter::_is_cancelled` stays false until final-owner destruction.
Meanwhile an unrelated request hitting global memory pressure can promote this
load's registered writer and call `flush_async()`, which checks only that local
flag; even after manager cancellation has returned it can submit/reset the
active memtable and start new local/cloud segment work. Please make
`flush_async()` inspect its copied `WriteRequest::load_cancel_status` under the
writer lock before submission/reset (without counting cancellation as flushed
memory or invoking a blocking error path under the limiter lock), and cover the
retained-cancelled-writer pressure ordering with a deterministic test.
##########
be/src/load/delta_writer/delta_writer.cpp:
##########
@@ -182,6 +196,7 @@ Status DeltaWriter::write(const Block* block, const
TabletAddRowsPayload& rows,
config::memtable_flush_running_count_limit *
(_req.write_req_type == WriteRequestType::GROUP ? 2 : 1);
while (_memtable_writer->flush_running_count() >=
effective_flush_running_count_limit) {
+ RETURN_IF_ERROR(_get_load_cancel_status());
Review Comment:
[P2] Recheck cancellation after the blocking writer phases
The new checks do not cover return from lazy `init()` or exit from the final
backpressure sleep. If cancellation is published while either phase blocks and
the next flush-count condition is false, this path reaches
`MemTableWriter::write()` without another status load; the empty local
`close()` has the same post-`init()` gap. `CloudDeltaWriter::write()` has the
parallel post-loop omission, including the S3-queue condition. Because
cancellation no longer waits on writer locks, these paths can append/close and
acknowledge work after cancellation is already observable. Please add
post-`init()` and post-loop gates immediately before the next memtable
operation in both implementations, with a barrier test for the final-sleep exit.
--
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]
