github-actions[bot] commented on code in PR #67759:
URL: https://github.com/apache/doris/pull/67759#discussion_r3985986821
##########
be/src/load/channel/load_channel_mgr.cpp:
##########
@@ -203,41 +233,52 @@ void LoadChannelMgr::_finish_load_channel(const UniqueId
load_id) {
Status LoadChannelMgr::cancel(const PTabletWriterCancelRequest& params) {
UniqueId load_id(params.id());
+ const auto reason = params.has_cancel_reason() &&
!params.cancel_reason().empty()
+ ? params.cancel_reason()
+ : "load channel cancelled";
std::shared_ptr<LoadChannel> cancelled_channel;
{
std::lock_guard<std::mutex> l(_lock);
if (_load_channels.contains(load_id)) {
cancelled_channel = _load_channels[load_id];
+ // Publish while this exact instance is still mapped, then cache
its
+ // effective first failure instead of a later cancel request's
reason.
+ // cancel() only updates the shared status; it takes no writer
locks.
+
RETURN_IF_ERROR(cancelled_channel->cancel(Status::Cancelled(reason)));
_load_channels.erase(load_id);
- }
- // We just need to record the first cancel msg
- auto* existing_handle =
_load_state_channels->lookup(load_id.to_string());
- if (existing_handle == nullptr) {
- if (params.has_cancel_reason() && !params.cancel_reason().empty())
{
- std::unique_ptr<CacheValue> cancel_reason_ptr =
std::make_unique<CacheValue>();
- cancel_reason_ptr->_cancel_reason = params.cancel_reason();
- size_t cache_capacity =
- cancel_reason_ptr->_cancel_reason.capacity() +
sizeof(CacheValue);
- auto* handle = _load_state_channels->insert(
- load_id.to_string(), cancel_reason_ptr.get(), 1,
cache_capacity);
- cancel_reason_ptr.release();
- _load_state_channels->release(handle);
- LOG(INFO) << fmt::format("load_id = {}, record_error reason =
{}",
- print_id(load_id),
params.cancel_reason());
- }
+ _record_cancelled_load_channel(load_id,
cancelled_channel->cancel_status().to_string());
} else {
- _load_state_channels->release(existing_handle);
+ _record_cancelled_load_channel(load_id, reason);
}
}
if (cancelled_channel != nullptr) {
- RETURN_IF_ERROR(cancelled_channel->cancel());
+ // Keep the final owner's destruction outside the manager lock, since
it
+ // may still wait for in-flight work.
LOG(INFO) << "load channel has been cancelled: " << load_id;
}
return Status::OK();
}
+void LoadChannelMgr::_record_cancelled_load_channel(const UniqueId& load_id,
+ const std::string& reason)
{
+ // Keep the first terminal state recorded for this load ID.
+ auto* existing_handle = _load_state_channels->lookup(load_id.to_string());
+ if (existing_handle != nullptr) {
+ _load_state_channels->release(existing_handle);
+ return;
+ }
+ auto value = std::make_unique<CacheValue>();
+ value->_cancel_reason = reason;
+ size_t cache_capacity = value->_cancel_reason.capacity() +
sizeof(CacheValue);
+ auto* handle =
+ _load_state_channels->insert(load_id.to_string(), value.get(), 1,
cache_capacity);
+ value.release();
+ _load_state_channels->release(handle);
Review Comment:
[P1] Keep the cancellation fence until retained owners drain
On the new failed-add path, `_cancel_load_channel()` removes the live
mapping before the retained RPC and final owner drain. Releasing this handle
then makes the only failure fence immediately evictable: this NUMBER cache has
1024 weight units split across 32 shards, so 33 terminal entries in one shard
can evict it while the old channel is still alive. A later `open` can create a
same-ID replacement, and in local mode the old `RowsetBuilder` can subsequently
call `rollback_txn()` by partition/txn/tablet and erase the replacement
writer's newly prepared transaction. Please keep a non-evictable
draining-generation fence until the last retained owner releases, and cover
cache churn plus an initialized replacement in a barrier test.
##########
be/src/load/delta_writer/delta_writer.cpp:
##########
@@ -221,6 +237,7 @@ Status DeltaWriter::close() {
Status BaseDeltaWriter::build_rowset() {
SCOPED_TIMER(_close_wait_timer);
RETURN_IF_ERROR(_memtable_writer->close_wait(_profile));
+ RETURN_IF_ERROR(_get_load_cancel_status());
Review Comment:
[P2] Recheck cancellation between group child builds
This status sample is consumed before entering a composite
`GroupRowsetBuilder`. Both local and cloud implementations build the row-binlog
child and then unconditionally start the data child; the first build can wait
for bitmap work and close file writers. With lock-free publication,
cancellation can now complete during that wait, after which the retained close
still begins the second child's file-close/build work. Pass the shared status
into the builder and consume it after blocking internal waits and between the
two child builds, with a barrier test that cancels while the row-binlog child
is blocked and proves the data child never starts.
##########
be/src/load/channel/tablets_channel.cpp:
##########
@@ -449,12 +466,13 @@ Status TabletsChannel::close(LoadChannel* parent, const
PTabletWriterAddBlockReq
// 5. commit all writers
for (auto* writer : need_wait_writers) {
+ RETURN_IF_ERROR(_check_cancelled());
Review Comment:
[P1] Linearize cancellation before local transaction persistence
This check (and the one at `DeltaWriter::commit_txn()` entry) can become
stale while `RowsetBuilder::commit_txn()` runs the MoW correctness scan or
waits for the transaction lock. `TxnManager::commit_txn()` then persists rowset
metadata and installs a committed `TabletTxnInfo` without another way to
observe the load status. Because cancellation now returns without waiting for
this channel lock, the cancel RPC can complete first while this retained EOS
still commits the cancelled generation and only notices cancellation afterward.
Carry the status to a safe pre-persistence linearization point and add a
barrier test that completes cancellation before `RowsetMetaManager::save()` and
proves no commit is installed.
##########
be/src/load/delta_writer/delta_writer.cpp:
##########
@@ -232,6 +249,7 @@ Status DeltaWriter::build_rowset() {
}
Status BaseDeltaWriter::submit_calc_delete_bitmap_task() {
+ RETURN_IF_ERROR(_get_load_cancel_status());
Review Comment:
[P2] Recheck cancellation after preparing bitmap work
This check is consumed before `BaseRowsetBuilder` opens every rowset segment
and, on the commit-phase path, waits for tablet metadata locks and computes the
current rowset set. If cancellation completes during either blocking
preparation, the retained close request still proceeds to enqueue
between-segment or per-segment delete-bitmap tasks; final-owner teardown then
has to drain work created after the load was already cancelled. Pass the shared
status into this layer and recheck it after `load_segments()` and immediately
before the final token submissions, with a barrier test that cancels while
segment loading is delayed.
##########
be/src/cloud/cloud_tablets_channel.cpp:
##########
@@ -278,6 +281,7 @@ Status CloudTabletsChannel::close(LoadChannel* parent,
const PTabletWriterAddBlo
std::vector<std::function<Status()>> tasks;
tasks.reserve(writers_to_commit.size());
for (auto* writer : writers_to_commit) {
+ RETURN_IF_ERROR(_check_cancelled());
Review Comment:
[P1] Linearize cancellation against normal cloud commit
This check only gates task construction, and each task samples once before
entering the blocking meta-service `COMMIT_ROWSET` RPC. Cancellation can now
publish, remove the load, and return while that RPC is in flight; the retained
task can then persist even a single initialized rowset and submit its warm-up
after the load is recorded cancelled. Group writers can additionally start the
data-rowset RPC after the row-binlog RPC returns. Please establish an explicit
cancel-versus-commit winner/reconciliation protocol rather than another
unsynchronized sample, and add barriers for both a delayed single-rowset commit
and a delayed first group-child commit.
--
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]
