github-actions[bot] commented on code in PR #67778:
URL: https://github.com/apache/doris/pull/67778#discussion_r3977236993
##########
be/src/storage/delete/calc_delete_bitmap_executor.h:
##########
@@ -66,36 +89,40 @@ class CalcDeleteBitmapToken {
// submit a generic function to the thread pool
template <typename Func>
Status submit_func(Func&& func) {
- {
- std::shared_lock rlock(_lock);
- RETURN_IF_ERROR(_status);
- _resource_ctx = thread_context()->resource_ctx();
- }
- return _thread_token->submit_func([this, func =
std::forward<Func>(func)]() {
- SCOPED_ATTACH_TASK(_resource_ctx);
- auto st = func();
- if (!st.ok()) {
- std::lock_guard wlock(_lock);
- if (_status.ok()) {
- _status = st;
- }
- }
- });
+ RETURN_IF_ERROR(_get_status());
+ auto resource_ctx = thread_context()->resource_ctx();
+ return _thread_token->submit_func(
Review Comment:
[P2] Preserve the published cancellation across the submit race
`_get_status()` and the underlying token submission are not atomic. A close
thread can observe OK here, then `LoadChannel::cancel()` can publish
`CANCELLED` and quiesce this token before `ThreadPool::do_submit()` acquires
the pool lock. `do_submit()` then returns synthesized `SERVICE_UNAVAILABLE`
(`token was shut down`) instead of the coordinator's first cancellation status,
and local/cloud close surfaces that misleading error without remapping it.
Please re-read `_get_status()` after a failed underlying submit and prefer a
now-published wrapper/coordinator status, while retaining a genuine pool error
when no cancellation was published, and cover this exact barrier-driven
interleaving.
##########
be/src/storage/delete/calc_delete_bitmap_executor.cpp:
##########
@@ -29,22 +29,63 @@
namespace doris {
using namespace ErrorCode;
+void DeleteBitmapCancellation::_register_token(const
std::shared_ptr<ThreadPoolToken>& token) {
+ {
+ std::lock_guard lock(_lock);
+ if (_status.ok()) {
+ _tokens.emplace_back(token);
+ return;
+ }
+ }
+ // A token created after cancellation must reject submissions too.
+ token->shutdown();
+}
+
+void DeleteBitmapCancellation::cancel(const Status& reason) {
+ DCHECK(!reason.ok());
+ std::vector<std::shared_ptr<ThreadPoolToken>> tokens;
+ {
+ std::lock_guard lock(_lock);
+ _status.update(reason);
+ for (const auto& weak_token : _tokens) {
+ if (auto token = weak_token.lock()) {
+ tokens.push_back(std::move(token));
+ }
+ }
+ }
+ // Publish to all tasks before waiting. Neither registration nor task
completion
+ // needs to wait for this lock while shutdown waits for running tasks.
+ for (const auto& token : tokens) {
Review Comment:
[P1] Avoid one pool-wide queue scan per tablet token
`ThreadPoolToken::shutdown()` scans the executor's entire shared `_queue` to
erase that token. Calling it serially for every registered writer/builder token
makes cancellation O(T*Q), and quadratic when a large MOW load has queued work
on T tablet tokens - the exact overload case this PR is meant to unblock.
Unrelated loads' entries are scanned once per cancelled token too, and each
scan holds the shared executor mutex, delaying their submissions and dispatch.
Please provide a batched quiesce/unlink mechanism grouped by underlying pool so
each queue is traversed once before waiting for running callbacks, and add a
many-token cancellation test that includes unrelated queued work.
##########
be/test/load/delta_writer/delta_writer_cancel_test.cpp:
##########
@@ -0,0 +1,453 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+
+#include <atomic>
+#include <chrono>
+#include <future>
+#include <memory>
+#include <mutex>
+#include <thread>
+#include <vector>
+
+#include "cloud/cloud_delta_writer.h"
+#include "cloud/cloud_rowset_builder.h"
+#include "cloud/cloud_rowset_writer.h"
+#include "cloud/cloud_storage_engine.h"
+#include "cloud/cloud_tablets_channel.h"
+#include "load/channel/load_channel_mgr.h"
+#include "load/channel/tablets_channel.h"
+#include "load/delta_writer/delta_writer.h"
+#include "runtime/exec_env.h"
+#include "runtime/fragment_mgr.h"
+#include "runtime/memory/mem_tracker_limiter.h"
+#include "runtime/thread_context.h"
+#include "storage/delete/calc_delete_bitmap_executor.h"
+#include "storage/options.h"
+#include "storage/rowset/beta_rowset_writer.h"
+#include "storage/rowset/group_rowset_writer.h"
+#include "storage/rowset_builder.h"
+#include "storage/storage_engine.h"
+#include "util/countdown_latch.h"
+
+namespace doris {
+
+// Exercise both local/cloud writers, with and without row binlog, without
tablet I/O.
+class DeltaWriterCancelTest : public testing::TestWithParam<int> {
+protected:
+ void SetUp() override {
+ auto tracker =
MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::OTHER,
+
"DeltaWriterCancelTest");
+ _attach_task = std::make_unique<AttachTask>(tracker);
+ ASSERT_TRUE(ThreadPoolBuilder("DeltaWriterCancelTest")
+ .set_min_threads(1)
+ .set_max_threads(1)
+ .build(&_pool)
+ .ok());
+ _previous_fragment_mgr = ExecEnv::GetInstance()->_fragment_mgr;
+ _fragment_mgr = std::make_unique<FragmentMgr>(ExecEnv::GetInstance());
+ ExecEnv::GetInstance()->_fragment_mgr = _fragment_mgr.get();
+ _load_channel = std::make_shared<LoadChannel>(UniqueId {}, 60, false,
"", 0, false, -1);
+ WriteRequest data_req;
+ data_req.delete_bitmap_cancellation =
_load_channel->_delete_bitmap_cancellation;
+ WriteRequest group_req = data_req;
+ group_req.write_req_type = WriteRequestType::GROUP;
+ WriteRequest binlog_req = data_req;
+ binlog_req.write_req_type = WriteRequestType::ROW_BINLOG;
+ if (is_cloud()) {
+ _cloud_engine = std::make_unique<CloudStorageEngine>(EngineOptions
{});
+ if (is_group()) {
+ _writer = std::make_unique<CloudDeltaWriter>(*_cloud_engine,
group_req, data_req,
+ binlog_req,
nullptr, UniqueId {});
+ auto* group =
static_cast<CloudGroupRowsetBuilder*>(_writer->_rowset_builder.get());
+ _builders = {group->data_builder(),
group->row_binlog_builder()};
+ } else {
+ _writer = std::make_unique<CloudDeltaWriter>(*_cloud_engine,
data_req, nullptr,
+ UniqueId {});
+ }
+ } else {
+ _local_engine = std::make_unique<StorageEngine>(EngineOptions {});
+ if (is_group()) {
+ _writer = std::make_unique<DeltaWriter>(*_local_engine,
group_req, data_req,
+ binlog_req, nullptr,
UniqueId {});
+ auto* group =
static_cast<GroupRowsetBuilder*>(_writer->_rowset_builder.get());
+ _builders = {group->txn_rowset_builder(),
group->row_binlog_builder()};
+ } else {
+ _writer = std::make_unique<DeltaWriter>(*_local_engine,
data_req, nullptr,
+ UniqueId {});
+ }
+ }
+ if (!is_group()) {
+ _builders = {_writer->_rowset_builder.get()};
+ }
+ }
+
+ void TearDown() override {
+ // Release the unrelated task before destroying writers, even after a
failed assertion.
+ _release_worker.count_down();
+ if (_pool) {
+ _pool->wait();
+ }
+ _writer.reset();
+ _tokens.clear();
+ _pool.reset();
+ _cloud_engine.reset();
+ _local_engine.reset();
+ _load_channel.reset();
+ if (_fragment_mgr) {
+ _fragment_mgr->stop();
+ ExecEnv::GetInstance()->_fragment_mgr = _previous_fragment_mgr;
+ _fragment_mgr.reset();
+ }
+ _attach_task.reset();
+ }
+
+ bool is_cloud() const { return GetParam() & 1; }
+ bool is_group() const { return GetParam() & 2; }
+
+ void install_tokens() {
+ for (auto* builder : _builders) {
+ std::shared_ptr<BaseBetaRowsetWriter> rowset_writer;
+ if (is_cloud()) {
+ rowset_writer =
std::make_shared<CloudRowsetWriter>(*_cloud_engine);
+ } else {
+ rowset_writer =
std::make_shared<BetaRowsetWriter>(*_local_engine);
+ }
+ // Set up only the state needed by cancellation and destruction.
No files are created.
Review Comment:
[P1] Exercise the production wiring and cleanup lifetime
Every parameterized case bypasses the paths changed by this PR:
`install_tokens()` writes private token/writer fields directly, uses a single
fixture pool plus empty metadata, and queues callbacks that only increment a
counter. `TearDown()` also waits the pool before destroying the writer. The
suite therefore remains green if channel -> request -> builder/writer ->
real-executor propagation is removed, or if callback draining moves back after
local file/cloud-cache cleanup. Please keep these interleaving tests, but add a
production-wired local/cloud MOW case that initializes both real phases, blocks
a real writer-capturing callback against an actual temporary segment, and
proves cancellation plus partial-init destruction drains it before file/cache
cleanup.
--
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]
