github-actions[bot] commented on code in PR #67778:
URL: https://github.com/apache/doris/pull/67778#discussion_r3980188469
##########
be/src/storage/delete/calc_delete_bitmap_executor.h:
##########
@@ -66,36 +90,41 @@ 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 _submit_func(
+ [this, resource_ctx = std::move(resource_ctx), func =
std::forward<Func>(func)]() {
+ SCOPED_ATTACH_TASK(resource_ctx);
+ // Cancellation can arrive while the owning channel is
blocked in close().
+ auto st = _get_status();
+ if (st.ok()) {
+ st = func();
+ }
+ if (!st.ok()) {
Review Comment:
[P2] Keep the first status when a running callback fails after cancellation
A task can pass the status check above, then
`DeleteBitmapCancellation::cancel()` publishes `CANCELLED` and waits in token
shutdown, and finally `func()` returns a real error here. This stores the later
error in wrapper `_status`; `_get_status()` checks that cell before the
already-failed coordinator, so `wait()` and future submissions return the later
error instead of the load's first-published cancellation. The existing test
covers failure-before-cancel, not this reverse ordering. Please arbitrate
wrapper/coordinator publications as one first terminal status (or publish
cancellation into each wrapper before shutdown) and add a barrier test for
running-task failure after cancellation publication.
##########
be/src/load/channel/load_channel_mgr.cpp:
##########
@@ -183,22 +183,62 @@ Status LoadChannelMgr::add_batch(const
PTabletWriterAddBlockRequest& request,
// 4. handle finish
if (channel->is_finished()) {
- _finish_load_channel(load_id);
+ RETURN_IF_ERROR(_finish_load_channel(load_id, channel));
}
return Status::OK();
}
-void LoadChannelMgr::_finish_load_channel(const UniqueId load_id) {
- VLOG_NOTICE << "removing load channel " << load_id << " because it's
finished";
- {
- std::lock_guard<std::mutex> l(_lock);
- if (_load_channels.contains(load_id)) {
- _load_channels.erase(load_id);
+Status LoadChannelMgr::_finish_load_channel(const UniqueId& load_id,
+ const
std::shared_ptr<LoadChannel>& channel) {
+ std::lock_guard<std::mutex> l(_lock);
+ auto it = _load_channels.find(load_id);
+ if (it != _load_channels.end() && it->second != channel) {
+ return Status::Cancelled("Load channel {} has been replaced",
load_id.to_string());
+ }
+
+ // Cancellation/timeout removes the channel and records its terminal state
under
+ // this same lock, before draining bitmap work. A late EOS must not
overwrite it.
+ auto* existing_handle = _load_state_channels->lookup(load_id.to_string());
Review Comment:
[P1] Do not apply an old terminal record to a reopened channel
After cancel/timeout removes a channel and caches failure, `open()` still
accepts the same load ID because it checks only `_load_channels`. Batches then
run on that replacement, but when its EOS reaches this branch the identity
check has passed, so the old cache entry erases the replacement and returns
stale cancellation after the replacement's close/commit work. A
retried/reordered open should either be rejected while terminal state is live
or start a generation whose state cannot collide with the old one. Please cover
cancel/timeout -> real `open()` -> replacement EOS;
`LateEosCannotFinishReplacementChannel` uses an empty cache and misses this
path.
##########
be/test/load/delta_writer/delta_writer_cancel_test.cpp:
##########
@@ -0,0 +1,512 @@
+// 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 "cpp/sync_point.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.
+ rowset_writer->_rowset_meta = std::make_shared<RowsetMeta>();
+ rowset_writer->_calc_delete_bitmap_token = make_token();
+ builder->_calc_delete_bitmap_token = make_token();
+ builder->_rowset_writer = rowset_writer;
+ _tokens.push_back(rowset_writer->_calc_delete_bitmap_token.get());
+ _tokens.push_back(builder->_calc_delete_bitmap_token.get());
+ }
+ if (is_group()) {
+ auto group_writer = std::make_shared<GroupRowsetWriter>();
+ group_writer->set_data_writer(_builders[0]->rowset_writer());
+ group_writer->set_row_binlog_writer(_builders[1]->rowset_writer());
+ _writer->_rowset_builder->_rowset_writer = std::move(group_writer);
+ }
+ }
+
+ std::unique_ptr<CalcDeleteBitmapToken> make_token() {
+ return std::make_unique<CalcDeleteBitmapToken>(
+ _pool->new_token(ThreadPool::ExecutionMode::CONCURRENT),
+ _load_channel->_delete_bitmap_cancellation);
+ }
+
+ FragmentMgr* _previous_fragment_mgr = nullptr;
+ std::unique_ptr<FragmentMgr> _fragment_mgr;
+ std::shared_ptr<LoadChannel> _load_channel;
+ std::unique_ptr<AttachTask> _attach_task;
+ std::unique_ptr<StorageEngine> _local_engine;
+ std::unique_ptr<CloudStorageEngine> _cloud_engine;
+ std::unique_ptr<ThreadPool> _pool;
+ std::unique_ptr<BaseDeltaWriter> _writer;
+ std::vector<BaseRowsetBuilder*> _builders;
+ std::vector<CalcDeleteBitmapToken*> _tokens;
+ CountDownLatch _worker_started {1};
+ CountDownLatch _release_worker {1};
+ std::atomic<int> _executed {0};
+};
+
+TEST_P(DeltaWriterCancelTest, CancelBeforeInit) {
+ ASSERT_TRUE(_writer->cancel().ok());
+ EXPECT_TRUE(_writer->_is_cancelled);
+ EXPECT_TRUE(_writer->_memtable_writer->_is_cancelled);
+ for (auto* builder : _builders) {
+ EXPECT_TRUE(builder->_is_cancelled);
+ }
+ EXPECT_TRUE(_writer->cancel().ok());
+}
+
+TEST_P(DeltaWriterCancelTest, CancelRemovesBothPhasesBeforeDestruction) {
+ install_tokens();
+ ASSERT_TRUE(_pool->submit_func([this] {
+ _worker_started.count_down();
+ _release_worker.wait();
+ }).ok());
+ ASSERT_TRUE(_worker_started.wait_for(std::chrono::seconds(10)));
+ for (auto* token : _tokens) {
+ ASSERT_TRUE(token->submit_func([this] {
+ ++_executed;
+ return Status::OK();
+ }).ok());
+ }
+ ASSERT_EQ(_pool->get_queue_size(), _tokens.size());
+
+ const auto cancelled = Status::Cancelled("load cancelled by sender");
+ ASSERT_TRUE(_writer->cancel_with_status(cancelled).ok());
+ EXPECT_EQ(_pool->get_queue_size(), 0);
+ EXPECT_EQ(_executed.load(), 0);
+ EXPECT_EQ(_writer->_memtable_writer->_cancel_status, cancelled);
+ for (auto* token : _tokens) {
+ EXPECT_EQ(token->wait(), cancelled);
+ EXPECT_EQ(token->submit_func([] { return Status::OK(); }), cancelled);
+ }
+ ASSERT_TRUE(_writer->cancel_with_status(Status::Cancelled("second
cancel")).ok());
+ for (auto* token : _tokens) {
+ EXPECT_EQ(token->wait(), cancelled);
+ }
+
+ _release_worker.count_down();
+ _pool->wait();
+ EXPECT_EQ(_executed.load(), 0);
+}
+
+TEST_P(DeltaWriterCancelTest, PreserveEarlierCalculationFailure) {
+ install_tokens();
+ const auto failure = Status::InternalError("delete bitmap calculation
failed");
+ ASSERT_TRUE(_tokens.front()->submit_func([failure] { return failure;
}).ok());
+ ASSERT_EQ(_tokens.front()->wait(), failure);
+ ASSERT_TRUE(_load_channel->cancel().ok());
+ ASSERT_TRUE(_writer->cancel().ok());
+ EXPECT_EQ(_tokens.front()->wait(), failure);
+ EXPECT_EQ(_tokens.front()->submit_func([] { return Status::OK(); }),
failure);
+ for (size_t i = 1; i < _tokens.size(); ++i) {
+ EXPECT_TRUE(_tokens[i]->wait().is<ErrorCode::CANCELLED>());
+ }
+}
+
+// Model close holding a tablets-channel/writer/builder lock while waiting for
bitmap.
+// Load cancellation keeps its own lock, drains bitmap, then cancels the
tablets channel.
+TEST_P(DeltaWriterCancelTest,
LoadCancelDrainsBitmapBeforeTabletsChannelCancellation) {
+ install_tokens();
+ ASSERT_TRUE(_pool->submit_func([this] {
+ _worker_started.count_down();
+ _release_worker.wait();
+ }).ok());
+ ASSERT_TRUE(_worker_started.wait_for(std::chrono::seconds(10)));
+ for (auto* token : _tokens) {
+ ASSERT_TRUE(token->submit_func([this] {
+ ++_executed;
+ return Status::OK();
+ }).ok());
+ }
+
+ PUniqueId id;
+ TabletsChannelKey key(id, 1);
+ std::shared_ptr<BaseTabletsChannel> tablets_channel;
+ if (is_cloud()) {
+ tablets_channel =
std::make_shared<CloudTabletsChannel>(*_cloud_engine, key, UniqueId {},
+ false,
nullptr);
+ } else {
+ tablets_channel =
+ std::make_shared<TabletsChannel>(*_local_engine, key, UniqueId
{}, false, nullptr);
+ }
+
tablets_channel->set_delete_bitmap_cancellation(_load_channel->_delete_bitmap_cancellation);
+ _load_channel->_tablets_channels.emplace(1, tablets_channel);
+ LoadChannelMgr manager;
Review Comment:
[P1] Deregister this stack manager's metric hook
`LoadChannelMgr` construction installs the process-global
`load_channel_count` hook with a `[this]` capture, but only `stop()` removes
it; the class has no destructor cleanup. This stack instance leaves scope
without `stop()`, and under `BE_TEST` duplicate hook registration uses
`emplace`, so the first parameter instance leaves a dangling callback that
later instances cannot replace. Subsequent metric sampling can dereference the
destroyed manager or contaminate other tests. Please guarantee `stop()` on
every exit (for example with fixture ownership or a scope guard).
--
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]
