This is an automated email from the ASF dual-hosted git repository.
gavinchou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new ec0fe00cce2 [fix](load) Avoid heavy work pool starvation during
incremental channel close (#68628)
ec0fe00cce2 is described below
commit ec0fe00cce244b1cca51400db3fd0f0f44835829
Author: hui lai <[email protected]>
AuthorDate: Mon Oct 12 12:51:29 2026 +0800
[fix](load) Avoid heavy work pool starvation during incremental channel
close (#68628)
### What problem does this PR solve?
When `VTabletWriter` opens new node channels during execution, it closes
channels in two stages. Initial channels receive `hang_wait` EOS
requests; senders wait for this first-stage barrier before closing their
incrementally opened channels.
The receiver currently polls `is_finished()` inside the heavy work pool.
Early EOS tasks can occupy every worker while waiting for the remaining
senders, whose EOS tasks are queued in the same pool. For example, 32
waiting EOS tasks can prevent the remaining 68 requests in a 100-sender
load from running. Normal progress then depends on external cancellation
or timeout cleanup.
The first-stage barrier protects against closing an incremental channel
before another sender has finished incrementally opening it. Its success
condition is that all senders have reached EOS. The existing `kFinished`
transition occurs before final flush/commit, allowing earlier senders to
proceed with incremental-channel close while the final sender completes
storage work.
Replace polling with a one-shot sender-arrival completion shared by each
tablets channel:
- Early EOS tasks release their heavy workers while their RPCs remain
pending. Existing sender deduplication/counting still elects the final
closer in both local and cloud modes.
- When the final sender arrives, set `kFinished` under the channel lock,
release the lock, and notify earlier EOS waiters. Reacquire the lock for
writer finalization. This preserves the existing arrival barrier and
overlap between initial- and incremental-channel close.
- The final EOS RPC still waits for its own flush/build/commit and
synchronous response writes. Finalization errors and final
tablet/row-count results are returned on that RPC; errors cannot revoke
success already delivered to earlier senders.
- Closed channels explicitly skip block writes, including block-bearing
EOS retries during arrival notification. A retry of the final sender
that holds the active channel waits for finalization and receives its
status and copied tablet results; earlier sender retries retain the
arrival barrier.
- Register RPC callbacks only after synchronous response/profile/timing
writes finish. Transfer the BRPC closure before registration, which may
complete inline. Callbacks run outside channel/completion locks.
- Cancellation or channel expiration releases pending waiters with
`Cancelled` if cancellation wins before arrival success is published.
The first terminal outcome wins; subsequent cancellation does not
overwrite published success.
- No protobuf or sender protocol changes.
This shared path is used by:
- Auto-partition loads that open new node channels while creating
partitions.
- `INSERT OVERWRITE ... PARTITION(*)`, which creates replacement
temporary partitions at runtime and reuses the partition-creation
callback to incrementally open channels. This can also affect ordinary
partitioned tables.
- Whole-table `INSERT OVERWRITE` with automatic partition creation
enabled, when new node channels are opened during execution.
The trigger is runtime creation of new node channels followed by the
two-stage EOS barrier in `VTabletWriter`. Incrementally adding tablets
to existing node channels alone does not set
`has_incremental_node_channel()`.
Based on master `d2cfcc1dcd4019fbae0d6ffbce16a7563bd9ecf4`.
### Release note
Fix a possible heavy work pool starvation deadlock during incremental
channel close in auto-partition loads and INSERT OVERWRITE operations
that open new node channels. Preserve the existing sender-arrival
barrier: earlier EOS RPCs may return once all senders reach EOS, before
final flush/commit; the final RPC still waits for finalization. Pending
EOS waiters now receive `Cancelled` when cancellation/expiration wins
before barrier success, rather than potentially receiving OK after
cancellation.
---
be/src/cloud/cloud_tablets_channel.cpp | 22 +-
be/src/cloud/cloud_tablets_channel.h | 1 +
be/src/load/channel/eos_completion.cpp | 57 +++
be/src/load/channel/eos_completion.h | 53 +++
be/src/load/channel/load_channel.cpp | 50 ++-
be/src/load/channel/load_channel.h | 9 +-
be/src/load/channel/load_channel_mgr.cpp | 5 +-
be/src/load/channel/load_channel_mgr.h | 6 +-
be/src/load/channel/tablets_channel.cpp | 69 ++-
be/src/load/channel/tablets_channel.h | 30 +-
be/src/service/internal_service.cpp | 15 +-
be/test/load/channel/eos_completion_test.cpp | 151 +++++++
be/test/load/channel/load_channel_eos_test.cpp | 561 +++++++++++++++++++++++++
13 files changed, 974 insertions(+), 55 deletions(-)
diff --git a/be/src/cloud/cloud_tablets_channel.cpp
b/be/src/cloud/cloud_tablets_channel.cpp
index 9a5f25c0439..28caf74f79d 100644
--- a/be/src/cloud/cloud_tablets_channel.cpp
+++ b/be/src/cloud/cloud_tablets_channel.cpp
@@ -23,6 +23,7 @@
#include "cloud/cloud_meta_mgr.h"
#include "cloud/cloud_storage_engine.h"
#include "cloud/config.h"
+#include "cpp/sync_point.h"
#include "load/channel/tablets_channel.h"
#include "load/delta_writer/delta_writer.h"
#include "storage/tablet_info.h"
@@ -83,17 +84,12 @@ Status CloudTabletsChannel::add_batch(const
PTabletWriterAddBlockRequest& reques
_add_batch_number_counter->update(1);
}
- auto status = _get_current_seq(cur_seq, request);
- if (UNLIKELY(!status.ok())) {
+ bool should_write = false;
+ auto status = _get_current_seq(cur_seq, request, should_write);
+ if (UNLIKELY(!status.ok()) || !should_write) {
return status;
}
- if (request.packet_seq() < cur_seq) {
- LOG(INFO) << "packet has already recept before, expect_seq=" << cur_seq
- << ", recept_seq=" << request.packet_seq();
- return Status::OK();
- }
-
if (request.is_adaptive_random_bucket()) {
std::unordered_map<int64_t, DorisVector<uint32_t>>
partition_to_rowidxs;
RETURN_IF_ERROR(_build_partition_to_rowidxs_for_adaptive_random_bucket(
@@ -166,9 +162,9 @@ Status CloudTabletsChannel::_init_writers_by_partition_ids(
Status CloudTabletsChannel::close(LoadChannel* parent, const
PTabletWriterAddBlockRequest& req,
PTabletWriterAddBlockResult* res, bool*
finished) {
// FIXME(plat1ko): Too many duplicate code with `TabletsChannel`
- std::lock_guard l(_lock);
+ std::unique_lock l(_lock);
if (_state == kFinished) {
- return _close_status;
+ return _get_close_result(l, req.sender_id(), res);
}
auto sender_id = req.sender_id();
@@ -194,9 +190,13 @@ Status CloudTabletsChannel::close(LoadChannel* parent,
const PTabletWriterAddBlo
return Status::OK();
}
+ return _close_and_notify(l, sender_id, res);
+}
+
+Status CloudTabletsChannel::_close_writers(PTabletWriterAddBlockResult* res) {
auto* tablet_errors = res->mutable_tablet_errors();
auto* tablet_vec = res->mutable_tablet_vec();
- _state = kFinished;
+
TEST_SYNC_POINT_RETURN_WITH_VALUE("CloudTabletsChannel::close.before_flush",
Status::OK(), res);
// All senders are closed
// 1. close all delta writers. under _lock.
diff --git a/be/src/cloud/cloud_tablets_channel.h
b/be/src/cloud/cloud_tablets_channel.h
index 0f9704bb4b4..c6592ff17d2 100644
--- a/be/src/cloud/cloud_tablets_channel.h
+++ b/be/src/cloud/cloud_tablets_channel.h
@@ -41,6 +41,7 @@ public:
PTabletWriterAddBlockResult* res, bool* finished) override;
private:
+ Status _close_writers(PTabletWriterAddBlockResult* response) override;
Status _prepare_adaptive_random_bucket_writer(BaseDeltaWriter* writer)
override;
Status _init_writers_by_partition_ids(const std::unordered_set<int64_t>&
partition_ids);
diff --git a/be/src/load/channel/eos_completion.cpp
b/be/src/load/channel/eos_completion.cpp
new file mode 100644
index 00000000000..c9d1f5b3cd9
--- /dev/null
+++ b/be/src/load/channel/eos_completion.cpp
@@ -0,0 +1,57 @@
+// 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 "load/channel/eos_completion.h"
+
+#include <utility>
+
+namespace doris {
+
+EosCompletion::~EosCompletion() {
+ complete(Status::Cancelled("Load channel destroyed before all EOS senders
arrived"));
+}
+
+void EosCompletion::add_waiter(Callback callback) {
+ Status status;
+ {
+ std::lock_guard lock(_mutex);
+ if (!_status.has_value()) {
+ _waiters.push_back(std::move(callback));
+ return;
+ }
+ status = *_status;
+ }
+ callback(status);
+}
+
+void EosCompletion::complete(const Status& status) {
+ std::vector<Callback> waiters;
+ {
+ std::lock_guard lock(_mutex);
+ if (_status.has_value()) {
+ // Sender arrival can race cancellation or load-channel expiration.
+ return;
+ }
+ _status = status;
+ waiters.swap(_waiters);
+ }
+ for (auto& waiter : waiters) {
+ waiter(status);
+ }
+}
+
+} // namespace doris
diff --git a/be/src/load/channel/eos_completion.h
b/be/src/load/channel/eos_completion.h
new file mode 100644
index 00000000000..ab1cec1657f
--- /dev/null
+++ b/be/src/load/channel/eos_completion.h
@@ -0,0 +1,53 @@
+// 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.
+
+#pragma once
+
+#include <functional>
+#include <mutex>
+#include <optional>
+#include <vector>
+
+#include "common/status.h"
+
+namespace doris {
+
+// One-shot arrival barrier for all senders of one tablets channel. Success
means
+// all senders reached EOS, not that the final flush/commit completed. Sender
+// accounting and final close remain in TabletsChannel/CloudTabletsChannel.
This
+// object only owns RPC callbacks, never a channel or request, so pending RPCs
+// cannot keep a cancelled/expired load channel alive.
+class EosCompletion {
+public:
+ using Callback = std::function<void(const Status&)>;
+
+ ~EosCompletion();
+
+ // Register only after all synchronous access to the RPC objects has ended.
+ // May invoke callback inline if all senders arrived or cancellation
already won.
+ void add_waiter(Callback callback);
+
+ // The first terminal result wins. Invoke callbacks outside the mutex.
+ void complete(const Status& status);
+
+private:
+ std::mutex _mutex;
+ std::optional<Status> _status;
+ std::vector<Callback> _waiters;
+};
+
+} // namespace doris
diff --git a/be/src/load/channel/load_channel.cpp
b/be/src/load/channel/load_channel.cpp
index 4b483824d36..e1066347c87 100644
--- a/be/src/load/channel/load_channel.cpp
+++ b/be/src/load/channel/load_channel.cpp
@@ -20,9 +20,12 @@
#include <gen_cpp/internal_service.pb.h>
#include <glog/logging.h>
+#include <vector>
+
#include "cloud/cloud_tablets_channel.h"
#include "cloud/config.h"
#include "common/logging.h"
+#include "load/channel/eos_completion.h"
#include "load/channel/tablets_channel.h"
#include "runtime/exec_env.h"
#include "runtime/fragment_mgr.h"
@@ -179,7 +182,8 @@ Status
LoadChannel::_get_tablets_channel(std::shared_ptr<BaseTabletsChannel>& ch
}
Status LoadChannel::add_batch(const PTabletWriterAddBlockRequest& request,
- PTabletWriterAddBlockResult* response) {
+ PTabletWriterAddBlockResult* response,
+ std::shared_ptr<EosCompletion>* eos_completion) {
DBUG_EXECUTE_IF("LoadChannel.add_batch.failed",
{ return Status::InternalError("fault injection"); });
SCOPED_TIMER(_add_batch_timer);
@@ -205,8 +209,12 @@ Status LoadChannel::add_batch(const
PTabletWriterAddBlockRequest& request,
}
// 3. handle eos
- // if channel is incremental, maybe hang on close until all close request
arrived.
+ // Keep the RPC pending without occupying a heavy worker while senders
arrive.
if (request.has_eos() && request.eos()) {
+ if (request.hang_wait()) {
+ DCHECK(!channel->is_incremental_channel());
+ *eos_completion = channel->eos_completion();
+ }
st = _handle_eos(channel.get(), request, response);
_report_profile(response);
if (!st.ok()) {
@@ -228,25 +236,10 @@ Status LoadChannel::_handle_eos(BaseTabletsChannel*
channel,
bool finished = false;
auto index_id = request.index_id();
+ // close() publishes sender arrival before flushing. Final flush/commit
+ // errors belong to this RPC and cannot revoke earlier EOS responses.
RETURN_IF_ERROR(channel->close(this, request, response, &finished));
- // for init node, we close waiting(hang on) all close request and let them
return together.
- if (request.has_hang_wait() && request.hang_wait()) {
- DCHECK(!channel->is_incremental_channel());
- VLOG_DEBUG << fmt::format("txn {}: reciever index {} close waiting by
sender {}", _txn_id,
- request.index_id(), request.sender_id());
- int count = 0;
- while (!channel->is_finished()) {
- bthread_usleep(1000);
- count++;
- }
- // now maybe finished or cancelled.
- VLOG_TRACE << "reciever close wait finished!" << request.sender_id();
- if (count >= 1000 * _timeout_s) { // maybe
config::streaming_load_rpc_max_alive_time_sec
- return Status::InternalError("Tablets channel didn't wait all
close");
- }
- }
-
if (finished) {
std::lock_guard<std::mutex> l(_lock);
{
@@ -306,9 +299,22 @@ bool LoadChannel::is_finished() {
Status LoadChannel::cancel() {
_cancelled.store(true);
- std::lock_guard<std::mutex> l(_lock);
- for (auto& it : _tablets_channels) {
- static_cast<void>(it.second->cancel());
+ std::vector<std::shared_ptr<BaseTabletsChannel>> channels;
+ {
+ std::lock_guard<std::mutex> l(_lock);
+ channels.reserve(_tablets_channels.size());
+ for (auto& [id, channel] : _tablets_channels) {
+ channels.push_back(channel);
+ }
+ }
+ // Complete pending RPCs before waiting for writer cancellation. No
callback
+ // may run under _lock, including when the timeout cleaner calls cancel().
+ for (auto& channel : channels) {
+ channel->eos_completion()->complete(
+ Status::Cancelled("Load channel {} cancelled",
_load_id.to_string()));
+ }
+ for (auto& channel : channels) {
+ static_cast<void>(channel->cancel());
}
return Status::OK();
}
diff --git a/be/src/load/channel/load_channel.h
b/be/src/load/channel/load_channel.h
index 77550eb3317..108abeedac4 100644
--- a/be/src/load/channel/load_channel.h
+++ b/be/src/load/channel/load_channel.h
@@ -35,6 +35,7 @@
namespace doris {
+class EosCompletion;
class PTabletWriterOpenRequest;
class PTabletWriterAddBlockRequest;
class PTabletWriterAddBlockResult;
@@ -52,9 +53,13 @@ public:
// open a new load channel if not exist
Status open(const PTabletWriterOpenRequest& request);
- // this batch must belong to a index in one transaction
+ // This batch must belong to one index in one transaction. For hang_wait
EOS,
+ // return the sender-arrival barrier in eos_completion. Final flush/commit
+ // stays synchronous on the last sender. The caller must register its RPC
+ // completion only after all synchronous request/response access has ended.
Status add_batch(const PTabletWriterAddBlockRequest& request,
- PTabletWriterAddBlockResult* response);
+ PTabletWriterAddBlockResult* response,
+ std::shared_ptr<EosCompletion>* eos_completion);
// return true if this load channel has been opened and all tablets
channels are closed then.
bool is_finished();
diff --git a/be/src/load/channel/load_channel_mgr.cpp
b/be/src/load/channel/load_channel_mgr.cpp
index a5f655b5fb6..5682edeb99e 100644
--- a/be/src/load/channel/load_channel_mgr.cpp
+++ b/be/src/load/channel/load_channel_mgr.cpp
@@ -149,7 +149,8 @@ Status
LoadChannelMgr::_get_load_channel(std::shared_ptr<LoadChannel>& channel,
}
Status LoadChannelMgr::add_batch(const PTabletWriterAddBlockRequest& request,
- PTabletWriterAddBlockResult* response) {
+ PTabletWriterAddBlockResult* response,
+ std::shared_ptr<EosCompletion>*
eos_completion) {
UniqueId load_id(request.id());
// 1. get load channel
std::shared_ptr<LoadChannel> channel;
@@ -175,7 +176,7 @@ Status LoadChannelMgr::add_batch(const
PTabletWriterAddBlockRequest& request,
// 3. add batch to load channel
// batch may not exist in request(eg: eos request without batch),
// this case will be handled in load channel's add batch method.
- Status st = channel->add_batch(request, response);
+ Status st = channel->add_batch(request, response, eos_completion);
if (UNLIKELY(!st.ok())) {
RETURN_IF_ERROR(channel->cancel());
return st;
diff --git a/be/src/load/channel/load_channel_mgr.h
b/be/src/load/channel/load_channel_mgr.h
index fdddbee0400..255f8b1c223 100644
--- a/be/src/load/channel/load_channel_mgr.h
+++ b/be/src/load/channel/load_channel_mgr.h
@@ -42,6 +42,7 @@
namespace doris {
class PTabletWriterCancelRequest;
+class EosCompletion;
class PTabletWriterOpenRequest;
class Thread;
@@ -56,8 +57,11 @@ public:
// open a new load channel if not exist
Status open(const PTabletWriterOpenRequest& request);
+ // For hang_wait EOS, return a barrier handle to the service, which
transfers
+ // the RPC completion after finishing synchronous response writes.
Status add_batch(const PTabletWriterAddBlockRequest& request,
- PTabletWriterAddBlockResult* response);
+ PTabletWriterAddBlockResult* response,
+ std::shared_ptr<EosCompletion>* eos_completion);
// cancel all tablet stream for 'load_id' load
Status cancel(const PTabletWriterCancelRequest& request);
diff --git a/be/src/load/channel/tablets_channel.cpp
b/be/src/load/channel/tablets_channel.cpp
index a589b8ba13e..ab540459727 100644
--- a/be/src/load/channel/tablets_channel.cpp
+++ b/be/src/load/channel/tablets_channel.cpp
@@ -42,6 +42,8 @@
#include "common/metrics/doris_metrics.h"
#include "common/metrics/metrics.h"
#include "core/block/block.h"
+#include "cpp/sync_point.h"
+#include "load/channel/eos_completion.h"
#include "load/channel/load_channel.h"
#include "load/delta_writer/delta_writer.h"
#include "storage/storage_engine.h"
@@ -63,6 +65,7 @@ std::atomic<uint64_t>
BaseTabletsChannel::_s_tablet_writer_count;
BaseTabletsChannel::BaseTabletsChannel(const TabletsChannelKey& key, const
UniqueId& load_id,
bool is_high_priority, RuntimeProfile*
profile)
: _key(key),
+ _eos_completion(std::make_shared<EosCompletion>()),
_state(kInitialized),
_load_id(load_id),
_closed_senders(64),
@@ -88,8 +91,10 @@ BaseTabletsChannel::~BaseTabletsChannel() {
TabletsChannel::~TabletsChannel() = default;
Status BaseTabletsChannel::_get_current_seq(int64_t& cur_seq,
- const
PTabletWriterAddBlockRequest& request) {
+ const
PTabletWriterAddBlockRequest& request,
+ bool& should_write) {
std::lock_guard<std::mutex> l(_lock);
+ should_write = false;
if (_state != kOpened) {
return _state == kFinished ? _close_status
: Status::InternalError("TabletsChannel {}
state: {}",
@@ -102,6 +107,11 @@ Status BaseTabletsChannel::_get_current_seq(int64_t&
cur_seq,
<< ", recept_seq=" << request.packet_seq();
return Status::InternalError("lost data packet");
}
+ should_write = request.packet_seq() == cur_seq;
+ if (!should_write) {
+ LOG(INFO) << "packet has already recept before, expect_seq=" << cur_seq
+ << ", recept_seq=" << request.packet_seq();
+ }
return Status::OK();
}
@@ -338,15 +348,47 @@ std::unique_ptr<BaseDeltaWriter>
TabletsChannel::create_delta_writer(const Write
_profile, _load_id);
}
+Status BaseTabletsChannel::_close_and_notify(std::unique_lock<std::mutex>&
lock, int sender_id,
+ PTabletWriterAddBlockResult*
response) {
+ DCHECK_EQ(_num_remaining_senders, 0);
+ // Stop accepting data before releasing the lock. The arrival barrier
allows
+ // earlier senders to close incremental channels while this sender flushes.
+ // A retry of this final sender must instead wait for the final close
result.
+ _state = kFinished;
+ _final_sender_id = sender_id;
+ _final_close_in_progress = true;
+ lock.unlock();
+ _eos_completion->complete(Status::OK());
+ lock.lock();
+
+ _close_status = _close_writers(response);
+ _final_close_result =
std::make_unique<PTabletWriterAddBlockResult>(*response);
+ _final_close_in_progress = false;
+ _final_close_cv.notify_all();
+ return _close_status;
+}
+
+Status BaseTabletsChannel::_get_close_result(std::unique_lock<std::mutex>&
lock, int sender_id,
+ PTabletWriterAddBlockResult*
response) {
+ if (sender_id == _final_sender_id) {
+ // The final closer is already running, so waiting here cannot prevent
+ // another sender's EOS from reaching the arrival barrier.
+ TEST_SYNC_POINT("BaseTabletsChannel::close.wait_for_final_result");
+ _final_close_cv.wait(lock, [this] { return !_final_close_in_progress;
});
+ DCHECK(_final_close_result != nullptr);
+ response->CopyFrom(*_final_close_result);
+ }
+ return _close_status;
+}
+
Status TabletsChannel::close(LoadChannel* parent, const
PTabletWriterAddBlockRequest& req,
PTabletWriterAddBlockResult* res, bool* finished)
{
int sender_id = req.sender_id();
int64_t backend_id = req.backend_id();
const auto& partition_ids = req.partition_ids();
- auto* tablet_errors = res->mutable_tablet_errors();
- std::lock_guard<std::mutex> l(_lock);
+ std::unique_lock<std::mutex> l(_lock);
if (_state == kFinished) {
- return _close_status;
+ return _get_close_result(l, sender_id, res);
}
if (_closed_senders.Get(sender_id)) {
// Double close from one sender, just return OK
@@ -370,7 +412,13 @@ Status TabletsChannel::close(LoadChannel* parent, const
PTabletWriterAddBlockReq
return Status::OK();
}
- _state = kFinished;
+ return _close_and_notify(l, sender_id, res);
+}
+
+Status TabletsChannel::_close_writers(PTabletWriterAddBlockResult* res) {
+ auto* tablet_errors = res->mutable_tablet_errors();
+ TEST_SYNC_POINT_RETURN_WITH_VALUE("TabletsChannel::close.before_flush",
Status::OK(), res);
+
// All senders are closed
// 1. close all delta writers
std::set<DeltaWriter*> need_wait_writers;
@@ -876,17 +924,12 @@ Status TabletsChannel::add_batch(const
PTabletWriterAddBlockRequest& request,
_add_batch_number_counter->update(1);
}
- auto status = _get_current_seq(cur_seq, request);
- if (UNLIKELY(!status.ok())) {
+ bool should_write = false;
+ auto status = _get_current_seq(cur_seq, request, should_write);
+ if (UNLIKELY(!status.ok()) || !should_write) {
return status;
}
- if (request.packet_seq() < cur_seq) {
- LOG(INFO) << "packet has already recept before, expect_seq=" << cur_seq
- << ", recept_seq=" << request.packet_seq();
- return Status::OK();
- }
-
// Adaptive random bucket add-block RPCs carry partition ids. The receiver
maps rows to
// the current tablet and advances its selected-tablet state after
memtable flush.
if (request.is_adaptive_random_bucket()) {
diff --git a/be/src/load/channel/tablets_channel.h
b/be/src/load/channel/tablets_channel.h
index 2cc42403ba9..51a1de5c7db 100644
--- a/be/src/load/channel/tablets_channel.h
+++ b/be/src/load/channel/tablets_channel.h
@@ -20,6 +20,7 @@
#include <glog/logging.h>
#include <atomic>
+#include <condition_variable>
#include <cstdint>
#include <mutex>
#include <ostream>
@@ -80,6 +81,7 @@ class BaseDeltaWriter;
class MemTableWriter;
class OlapTableSchemaParam;
class LoadChannel;
+class EosCompletion;
struct WriteRequest;
// Write channel for a particular (load, index).
@@ -103,6 +105,7 @@ public:
// Mark sender with 'sender_id' as closed.
// If all senders are closed, close this channel, set '*finished' to true,
update 'tablet_vec'
// to include all tablets written in this channel.
+ // Release the EOS arrival barrier before final flushing/committing,
outside _lock.
// no-op when this channel has been closed or cancelled
virtual Status close(LoadChannel* parent, const
PTabletWriterAddBlockRequest& req,
PTabletWriterAddBlockResult* res, bool* finished) = 0;
@@ -110,6 +113,8 @@ public:
// no-op when this channel has been closed or cancelled
virtual Status cancel();
+ const std::shared_ptr<EosCompletion>& eos_completion() const { return
_eos_completion; }
+
void refresh_profile();
size_t total_received_rows() const { return _total_received_rows; }
@@ -122,6 +127,15 @@ public:
bool is_finished() const { return _state == kFinished; }
protected:
+ // Called by the final sender under _lock. Publish arrival before flushing,
+ // releasing the channel lock while invoking RPC callbacks.
+ Status _close_and_notify(std::unique_lock<std::mutex>& lock, int sender_id,
+ PTabletWriterAddBlockResult* response);
+ Status _get_close_result(std::unique_lock<std::mutex>& lock, int sender_id,
+ PTabletWriterAddBlockResult* response);
+ // Called exactly once by the final sender, with _lock held.
+ virtual Status _close_writers(PTabletWriterAddBlockResult* response) = 0;
+
Status _init_adaptive_random_bucket_state(const PTabletWriterOpenRequest&
request);
Status _write_block_data(const PTabletWriterAddBlockRequest& request,
int64_t cur_seq,
std::unordered_map<int64_t,
TabletAddRowsPayload>& tablet_to_rows,
@@ -136,7 +150,9 @@ protected:
std::unordered_map<int64_t, DorisVector<uint32_t>>*
partition_to_rowidxs);
std::shared_ptr<std::mutex> _get_partition_route_lock(int64_t
partition_id);
- Status _get_current_seq(int64_t& cur_seq, const
PTabletWriterAddBlockRequest& request);
+ // OK does not imply a write: duplicates and closed channels are no-ops.
+ Status _get_current_seq(int64_t& cur_seq, const
PTabletWriterAddBlockRequest& request,
+ bool& should_write);
// open all writer
Status _open_all_writers(const PTabletWriterOpenRequest& request);
@@ -153,6 +169,7 @@ protected:
// id of this load channel
TabletsChannelKey _key;
+ std::shared_ptr<EosCompletion> _eos_completion;
// protect _state change. open and close. when add_batch finished, lock to
change _next_seqs also
std::mutex _lock;
@@ -177,9 +194,14 @@ protected:
int _num_remaining_senders = 0;
std::vector<int64_t> _next_seqs;
Bitmap _closed_senders;
- // status to return when operate on an already closed/cancelled channel
- // currently it's OK.
+ // Arrival releases earlier senders; only the final sender's retries must
+ // wait for flush/commit and receive its status and tablet results. All
these
+ // fields are protected by _lock, including while callbacks run without it.
+ int _final_sender_id = -1;
+ bool _final_close_in_progress = false;
+ std::condition_variable _final_close_cv;
Status _close_status;
+ std::unique_ptr<PTabletWriterAddBlockResult> _final_close_result;
// tablet_id -> TabletChannel. it will only be changed in open() or
inc_open()
std::unordered_map<int64_t, std::unique_ptr<BaseDeltaWriter>>
_tablet_writers;
@@ -241,6 +263,8 @@ public:
PTabletWriterAddBlockResult* res, bool* finished) override;
private:
+ Status _close_writers(PTabletWriterAddBlockResult* response) override;
+
// deal with DeltaWriter commit_txn(), add tablet to list for return.
void _commit_txn(DeltaWriter* writer, PTabletWriterAddBlockResult* res);
diff --git a/be/src/service/internal_service.cpp
b/be/src/service/internal_service.cpp
index eeed9f230d1..8858f5eceb8 100644
--- a/be/src/service/internal_service.cpp
+++ b/be/src/service/internal_service.cpp
@@ -80,6 +80,7 @@
#include "io/fs/local_file_system.h"
#include "io/fs/stream_load_pipe.h"
#include "io/io_common.h"
+#include "load/channel/eos_completion.h"
#include "load/channel/load_channel_mgr.h"
#include "load/channel/load_stream_mgr.h"
#include "load/delta_writer/delta_writer.h"
@@ -540,10 +541,11 @@ void
PInternalService::tablet_writer_add_block(google::protobuf::RpcController*
int64_t wait_execution_time_ns = MonotonicNanos() -
submit_task_time_ns;
brpc::ClosureGuard closure_guard(done);
int64_t execution_time_ns = 0;
+ std::shared_ptr<EosCompletion> eos_completion;
{
SCOPED_RAW_TIMER(&execution_time_ns);
signal::SignalTaskIdKeeper keeper(request->id());
- auto st = _exec_env->load_channel_mgr()->add_batch(*request,
response);
+ auto st = _exec_env->load_channel_mgr()->add_batch(*request,
response, &eos_completion);
if (!st.ok()) {
LOG(WARNING) << "tablet writer add block failed, message=" <<
st
<< ", id=" << request->id() << ", index_id=" <<
request->index_id()
@@ -554,6 +556,17 @@ void
PInternalService::tablet_writer_add_block(google::protobuf::RpcController*
}
response->set_execution_time_us(execution_time_ns / NANOS_PER_MICRO);
response->set_wait_execution_time_us(wait_execution_time_ns /
NANOS_PER_MICRO);
+ if (eos_completion && response->status().status_code() ==
TStatusCode::OK) {
+ // Transfer done before publishing the callback: a completed
barrier
+ // can invoke it inline. All request/response access above must end
+ // first, including profiling, timers and SignalTaskIdKeeper.
+ auto* deferred_done = closure_guard.release();
+ eos_completion->add_waiter([response, deferred_done](const Status&
status) {
+ status.to_protobuf(response->mutable_status());
+ deferred_done->Run();
+ // The RPC objects (including HTTP's request wrapper) are gone.
+ });
+ }
});
if (!ret) {
offer_failed(response, done, _heavy_work_pool);
diff --git a/be/test/load/channel/eos_completion_test.cpp
b/be/test/load/channel/eos_completion_test.cpp
new file mode 100644
index 00000000000..80c3df32215
--- /dev/null
+++ b/be/test/load/channel/eos_completion_test.cpp
@@ -0,0 +1,151 @@
+// 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 "load/channel/eos_completion.h"
+
+#include <gtest/gtest.h>
+
+#include <atomic>
+#include <string>
+#include <thread>
+#include <vector>
+
+#include "util/countdown_latch.h"
+
+namespace doris {
+
+TEST(EosCompletionTest, MoreSendersThanWorkers) {
+ EosCompletion completion;
+ int replies = 0;
+ // A single worker can process all 100 EOS tasks: registering a waiter must
+ // return without waiting for any other sender or sending a successful RPC.
+ for (int sender = 0; sender < 100; ++sender) {
+ completion.add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ ++replies;
+ });
+ EXPECT_EQ(replies, 0);
+ }
+ completion.complete(Status::OK());
+ EXPECT_EQ(replies, 100);
+}
+
+TEST(EosCompletionTest, FailureAndLateRegistration) {
+ EosCompletion completion;
+ int replies = 0;
+ auto callback = [&](const Status& status) {
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("flush failed"), std::string::npos);
+ ++replies;
+ };
+ completion.add_waiter(callback);
+ completion.complete(Status::InternalError("flush failed"));
+ // Models an EOS task finishing its response/profile writes after failure.
+ completion.add_waiter(callback);
+ completion.complete(Status::OK());
+ EXPECT_EQ(replies, 2);
+}
+
+TEST(EosCompletionTest, ArrivalSuccessSurvivesLaterCloseFailure) {
+ EosCompletion completion;
+ int replies = 0;
+ auto reply = [&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ ++replies;
+ };
+ completion.add_waiter(reply);
+ completion.complete(Status::OK());
+ completion.complete(Status::InternalError("final flush failed"));
+ completion.complete(Status::Cancelled("late cancellation"));
+ completion.add_waiter(reply);
+ EXPECT_EQ(replies, 2);
+}
+
+TEST(EosCompletionTest, CancellationWinsBeforeRegistration) {
+ EosCompletion completion;
+ completion.complete(Status::Cancelled("load expired"));
+ completion.complete(Status::OK());
+ int replies = 0;
+ completion.add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.is<ErrorCode::CANCELLED>());
+ ++replies;
+ });
+ EXPECT_EQ(replies, 1);
+}
+
+TEST(EosCompletionTest, DestroyingUnfinishedChannelReleasesWaiters) {
+ int replies = 0;
+ {
+ EosCompletion completion;
+ completion.add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.is<ErrorCode::CANCELLED>());
+ ++replies;
+ });
+ }
+ EXPECT_EQ(replies, 1);
+}
+
+TEST(EosCompletionTest, CallbacksRunOutsideLock) {
+ EosCompletion completion;
+ int replies = 0;
+ completion.add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ ++replies;
+ completion.add_waiter([&](const Status& result) {
+ EXPECT_TRUE(result.ok());
+ ++replies;
+ });
+ completion.complete(Status::Cancelled("late cancellation"));
+ });
+ completion.complete(Status::OK());
+ EXPECT_EQ(replies, 2);
+}
+
+TEST(EosCompletionTest,
ConcurrentRegistrationArrivalAndCancellationCompleteOnce) {
+ EosCompletion completion;
+ CountDownLatch start(1);
+ std::atomic<int> replies {0};
+ std::atomic<int> successes {0};
+ std::vector<std::thread> tasks;
+ for (int i = 0; i < 100; ++i) {
+ tasks.emplace_back([&] {
+ start.wait();
+ completion.add_waiter([&](const Status& status) {
+ if (status.ok()) {
+ ++successes;
+ }
+ ++replies;
+ });
+ });
+ }
+ tasks.emplace_back([&] {
+ start.wait();
+ completion.complete(Status::OK());
+ });
+ tasks.emplace_back([&] {
+ start.wait();
+ completion.complete(Status::Cancelled("load expired"));
+ });
+ start.count_down();
+ for (auto& task : tasks) {
+ task.join();
+ }
+ EXPECT_EQ(replies.load(), 100);
+ EXPECT_TRUE(successes.load() == 0 || successes.load() == 100);
+}
+
+} // namespace doris
diff --git a/be/test/load/channel/load_channel_eos_test.cpp
b/be/test/load/channel/load_channel_eos_test.cpp
new file mode 100644
index 00000000000..ca76dac5c83
--- /dev/null
+++ b/be/test/load/channel/load_channel_eos_test.cpp
@@ -0,0 +1,561 @@
+// 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 <gen_cpp/internal_service.pb.h>
+#include <gtest/gtest.h>
+
+#include <atomic>
+#include <chrono>
+#include <ctime>
+#include <memory>
+#include <string>
+#include <thread>
+#include <utility>
+
+#include "agent/be_exec_version_manager.h"
+#include "cloud/cloud_storage_engine.h"
+#include "cloud/cloud_tablets_channel.h"
+#include "core/block/block.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_number.h"
+#include "cpp/sync_point.h"
+#include "load/channel/eos_completion.h"
+#include "load/channel/load_channel.h"
+#include "load/channel/load_channel_mgr.h"
+#include "load/channel/tablets_channel.h"
+#include "runtime/exec_env.h"
+#include "runtime/fragment_mgr.h"
+#include "storage/storage_engine.h"
+#include "util/countdown_latch.h"
+#include "util/defer_op.h"
+
+namespace doris {
+
+class LoadChannelEosTest : public testing::TestWithParam<bool> {
+protected:
+ void SetUp() override {
+ auto* env = ExecEnv::GetInstance();
+ _previous_fragment_mgr = env->_fragment_mgr;
+ _fragment_mgr = std::make_unique<FragmentMgr>(env);
+ env->_fragment_mgr = _fragment_mgr.get();
+ _load = std::make_shared<LoadChannel>(UniqueId(1, 2), 60, true, "", 0,
false, 0);
+ }
+
+ void TearDown() override {
+ _load.reset();
+ _fragment_mgr->stop();
+ ExecEnv::GetInstance()->_fragment_mgr = _previous_fragment_mgr;
+ }
+
+ std::shared_ptr<BaseTabletsChannel> make_channel(int senders) {
+ PUniqueId id;
+ id.set_hi(1);
+ id.set_lo(2);
+ std::shared_ptr<BaseTabletsChannel> channel;
+ if (GetParam()) {
+ _cloud_engine = std::make_unique<CloudStorageEngine>(EngineOptions
{});
+ channel = std::make_shared<CloudTabletsChannel>(
+ *_cloud_engine, TabletsChannelKey(id, 10), UniqueId(1, 2),
true, nullptr);
+ } else {
+ _engine = std::make_unique<StorageEngine>(EngineOptions {});
+ channel = std::make_shared<TabletsChannel>(*_engine,
TabletsChannelKey(id, 10),
+ UniqueId(1, 2), true,
nullptr);
+ }
+ // No writers are needed to exercise real sender accounting and close.
+ channel->_state = BaseTabletsChannel::kOpened;
+ channel->_num_remaining_senders = senders;
+ channel->_closed_senders.Reset(senders);
+ channel->_next_seqs.resize(senders);
+ _load->_tablets_channels.emplace(10, channel);
+ _load->_opened = true;
+ return channel;
+ }
+
+ static PTabletWriterAddBlockRequest eos(int sender, bool hang_wait = true)
{
+ PTabletWriterAddBlockRequest request;
+ request.mutable_id()->set_hi(1);
+ request.mutable_id()->set_lo(2);
+ request.set_index_id(10);
+ request.set_sender_id(sender);
+ request.set_backend_id(1);
+ request.set_eos(true);
+ request.set_hang_wait(hang_wait);
+ return request;
+ }
+
+ static PTabletWriterAddBlockRequest eos_with_block(int sender) {
+ auto request = eos(sender);
+ request.set_packet_seq(7);
+ request.add_tablet_ids(123);
+ auto column = ColumnInt32::create();
+ column->insert_value(42);
+ Block block {{std::move(column), std::make_shared<DataTypeInt32>(),
"value"}};
+ size_t uncompressed_bytes = 0;
+ size_t compressed_bytes = 0;
+ int64_t compress_time = 0;
+ EXPECT_TRUE(block.serialize(BeExecVersionManager::get_newest_version(),
+ request.mutable_block(),
&uncompressed_bytes, &compressed_bytes,
+ &compress_time,
segment_v2::CompressionTypePB::SNAPPY)
+ .ok());
+ return request;
+ }
+
+ Status retry_on_rpc_thread(const PTabletWriterAddBlockRequest& request,
+ PTabletWriterAddBlockResult* response,
+ std::shared_ptr<EosCompletion>* completion) {
+ // Arrival callbacks run inside the final closer's attached load task.
+ // A real retry runs on another RPC worker: calling add_batch inline
+ // would illegally nest SCOPED_ATTACH_TASK on the callback thread.
+ Status status;
+ std::thread rpc([&] { status = _load->add_batch(request, response,
completion); });
+ rpc.join();
+ return status;
+ }
+
+ void check_final_sender_retry(bool with_block, bool close_fails) {
+ auto channel = make_channel(2);
+ auto request = with_block ? eos_with_block(1) : eos(1);
+ // Model the packet already accepted by add_batch before entering EOS
+ // accounting. No writer is installed: reaching the writer lookup on a
+ // retry would fail with "unknown tablet" instead of succeeding
silently.
+ channel->_next_seqs[1] = 8;
+ PTabletWriterAddBlockResult first_response;
+ std::shared_ptr<EosCompletion> first;
+ ASSERT_TRUE(_load->add_batch(eos(0), &first_response, &first).ok());
+
+ CountDownLatch arrival_callback(1);
+ CountDownLatch finish_callback(1);
+ CountDownLatch retry_waiting(1);
+ first->add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ arrival_callback.count_down();
+ finish_callback.wait();
+ });
+ auto* sp = SyncPoint::get_instance();
+ SyncPoint::CallbackGuard wait_guard;
+ SyncPoint::CallbackGuard close_guard;
+ sp->set_call_back(
+ "BaseTabletsChannel::close.wait_for_final_result",
+ [&](auto&&) { retry_waiting.count_down(); }, &wait_guard);
+ int final_closes = 0;
+ sp->set_call_back(
+ before_flush_sync_point(),
+ [&](auto&& args) {
+ ++final_closes;
+ auto* response =
try_any_cast<PTabletWriterAddBlockResult*>(args[0]);
+ auto* tablet = response->add_tablet_vec();
+ tablet->set_tablet_id(123);
+ tablet->set_schema_hash(0);
+ tablet->set_received_rows(1);
+ if (close_fails) {
+ auto* ret = try_any_cast_ret<Status>(args);
+ ret->first = Status::InternalError("final commit
failed");
+ ret->second = true;
+ }
+ },
+ &close_guard);
+ sp->enable_processing();
+ Defer disable_sync_points {[&] { sp->disable_processing(); }};
+
+ Status final_status;
+ PTabletWriterAddBlockResult final_response;
+ std::thread finalizer([&] {
+ std::shared_ptr<EosCompletion> completion;
+ final_status = _load->add_batch(request, &final_response,
&completion);
+ });
+ bool arrived = arrival_callback.wait_for(std::chrono::seconds(10));
+ EXPECT_TRUE(arrived);
+ Status retry_status;
+ PTabletWriterAddBlockResult retry_response;
+ std::atomic<bool> retry_returned {false};
+ std::thread retry([&] {
+ std::shared_ptr<EosCompletion> completion;
+ retry_status = _load->add_batch(request, &retry_response,
&completion);
+ retry_returned = true;
+ });
+ // The finalizer is still inside an arrival callback, with _lock
released.
+ // Require the retry to actually enter this window before allowing
close.
+ EXPECT_TRUE(retry_waiting.wait_for(std::chrono::seconds(10)));
+ EXPECT_FALSE(retry_returned.load());
+ finish_callback.count_down();
+ finalizer.join();
+ retry.join();
+ EXPECT_TRUE(retry_returned.load());
+ EXPECT_EQ(final_closes, 1);
+ EXPECT_EQ(channel->_num_remaining_senders, 0);
+ EXPECT_EQ(channel->_next_seqs[1], 8);
+ EXPECT_EQ(final_status.ok(), !close_fails);
+ EXPECT_EQ(retry_status.to_string(), final_status.to_string());
+ EXPECT_EQ(retry_response.tablet_errors_size(), 0);
+ ASSERT_EQ(retry_response.tablet_vec_size(), 1);
+ EXPECT_EQ(retry_response.tablet_vec(0).tablet_id(), 123);
+ EXPECT_EQ(retry_response.tablet_vec(0).received_rows(), 1);
+ if (close_fails) {
+ EXPECT_NE(retry_status.to_string().find("final commit failed"),
std::string::npos);
+ }
+ }
+
+ const char* before_flush_sync_point() const {
+ return GetParam() ? "CloudTabletsChannel::close.before_flush"
+ : "TabletsChannel::close.before_flush";
+ }
+
+ FragmentMgr* _previous_fragment_mgr = nullptr;
+ std::unique_ptr<FragmentMgr> _fragment_mgr;
+ std::unique_ptr<StorageEngine> _engine;
+ std::unique_ptr<CloudStorageEngine> _cloud_engine;
+ std::shared_ptr<LoadChannel> _load;
+};
+
+TEST_P(LoadChannelEosTest,
SingleWorkerProcessesOneHundredSendersAndDuplicateEos) {
+ auto channel = make_channel(100);
+ std::atomic<int> replies {0};
+ CountDownLatch processed(1);
+ std::thread worker([&] {
+ // Retry sender zero before any other sender. It must not decrement the
+ // remaining count twice, and both RPCs must wait for all senders to
arrive.
+ for (int i = 0; i <= 100; ++i) {
+ int sender = i == 0 ? 0 : i - 1;
+ PTabletWriterAddBlockResult response;
+ std::shared_ptr<EosCompletion> completion;
+ EXPECT_TRUE(_load->add_batch(eos(sender), &response,
&completion).ok());
+ EXPECT_NE(completion, nullptr);
+ completion->add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ EXPECT_EQ(channel->_num_remaining_senders, 0);
+ ++replies;
+ });
+ if (sender < 99) {
+ EXPECT_EQ(replies.load(), 0);
+ }
+ }
+ processed.count_down();
+ });
+ bool finished = processed.wait_for(std::chrono::seconds(10));
+ if (!finished) {
+ // Also makes the old blocking implementation fail instead of hanging
UT.
+ EXPECT_TRUE(_load->cancel().ok());
+ }
+ worker.join();
+ EXPECT_TRUE(finished);
+ EXPECT_EQ(replies.load(), 101);
+ EXPECT_TRUE(_load->is_finished());
+ EXPECT_EQ(channel->_num_remaining_senders, 0);
+}
+
+TEST_P(LoadChannelEosTest, NonWaitingLastSenderReleasesEarlierWaiter) {
+ auto channel = make_channel(2);
+ PTabletWriterAddBlockResult response;
+ std::shared_ptr<EosCompletion> first;
+ ASSERT_TRUE(_load->add_batch(eos(0), &response, &first).ok());
+ int replies = 0;
+ first->add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ ++replies;
+ });
+ EXPECT_EQ(replies, 0);
+ bool reached_close = false;
+ auto* sp = SyncPoint::get_instance();
+ SyncPoint::CallbackGuard guard;
+ sp->set_call_back(
+ before_flush_sync_point(),
+ [&](auto&&) {
+ reached_close = true;
+ EXPECT_EQ(replies, 1);
+ },
+ &guard);
+ sp->enable_processing();
+ Defer disable_sync_points {[&] { sp->disable_processing(); }};
+ std::shared_ptr<EosCompletion> last;
+ EXPECT_TRUE(_load->add_batch(eos(1, false), &response, &last).ok());
+ EXPECT_TRUE(reached_close);
+ EXPECT_EQ(last, nullptr);
+ EXPECT_EQ(replies, 1);
+}
+
+TEST_P(LoadChannelEosTest, CancelReleasesRegisteredAndNotYetRegisteredRpcs) {
+ auto channel = make_channel(3);
+ PTabletWriterAddBlockResult response;
+ std::shared_ptr<EosCompletion> first;
+ std::shared_ptr<EosCompletion> second;
+ ASSERT_TRUE(_load->add_batch(eos(0), &response, &first).ok());
+ ASSERT_TRUE(_load->add_batch(eos(1), &response, &second).ok());
+ int replies = 0;
+ auto callback = [&](const Status& status) {
+ EXPECT_TRUE(status.is<ErrorCode::CANCELLED>());
+ // Re-enter the load lock; cancellation must invoke callbacks outside
it.
+ EXPECT_FALSE(_load->is_finished());
+ ++replies;
+ };
+ first->add_waiter(callback);
+ EXPECT_TRUE(_load->cancel().ok());
+ second->add_waiter(callback);
+ EXPECT_TRUE(_load->cancel().ok());
+ EXPECT_EQ(replies, 2);
+}
+
+TEST_P(LoadChannelEosTest, TimeoutCleanerReleasesPendingRpc) {
+ auto channel = make_channel(2);
+ PTabletWriterAddBlockResult response;
+ std::shared_ptr<EosCompletion> completion;
+ ASSERT_TRUE(_load->add_batch(eos(0), &response, &completion).ok());
+ LoadChannelMgr manager;
+ manager._load_channels.emplace(_load->load_id(), _load);
+ _load->_last_updated_time.store(time(nullptr) - 61);
+ int replies = 0;
+ completion->add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.is<ErrorCode::CANCELLED>());
+ EXPECT_TRUE(manager.get_all_load_channel_ids().empty());
+ ++replies;
+ });
+ EXPECT_TRUE(manager._start_load_channels_clean().ok());
+ EXPECT_EQ(replies, 1);
+ manager.stop();
+}
+
+// Exercise the real local/cloud close paths. Hold the final closer immediately
+// before writer flushing, after sender accounting and barrier publication.
+TEST_P(LoadChannelEosTest, ArrivalReleasesWaitersBeforeFinalCloseFailure) {
+ auto channel = make_channel(3);
+ CountDownLatch closing(1);
+ CountDownLatch finish_close(1);
+ auto* sp = SyncPoint::get_instance();
+ SyncPoint::CallbackGuard guard;
+ sp->set_call_back(
+ before_flush_sync_point(),
+ [&](auto&& args) {
+ closing.count_down();
+ finish_close.wait();
+ auto* ret = try_any_cast_ret<Status>(args);
+ ret->first = Status::InternalError("final flush failed");
+ ret->second = true;
+ },
+ &guard);
+ sp->enable_processing();
+ Defer disable_sync_points {[&] { sp->disable_processing(); }};
+
+ PTabletWriterAddBlockResult first_response;
+ PTabletWriterAddBlockResult second_response;
+ std::shared_ptr<EosCompletion> first;
+ std::shared_ptr<EosCompletion> second;
+ ASSERT_TRUE(_load->add_batch(eos(0), &first_response, &first).ok());
+ ASSERT_TRUE(_load->add_batch(eos(1), &second_response, &second).ok());
+ std::atomic<int> replies {0};
+ auto reply = [&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ ++replies;
+ };
+ first->add_waiter(reply);
+ std::atomic<bool> final_rpc_returned {false};
+ std::thread finalizer([&] {
+ PTabletWriterAddBlockResult response;
+ std::shared_ptr<EosCompletion> last;
+ auto status = _load->add_batch(eos(2), &response, &last);
+ EXPECT_FALSE(status.ok());
+ EXPECT_NE(status.to_string().find("final flush failed"),
std::string::npos);
+ // The service returns this error directly instead of registering the
+ // final RPC on the already successful arrival barrier.
+ final_rpc_returned = true;
+ });
+ bool reached_close = closing.wait_for(std::chrono::seconds(10));
+ EXPECT_TRUE(reached_close);
+ if (reached_close) {
+ EXPECT_TRUE(channel->is_finished());
+ EXPECT_FALSE(_load->is_finished());
+ EXPECT_FALSE(final_rpc_returned.load());
+ EXPECT_EQ(replies.load(), 1);
+ // A sender may finish its response writes after the last EOS arrived.
+ second->add_waiter(reply);
+ EXPECT_EQ(replies.load(), 2);
+ }
+ finish_close.count_down();
+ finalizer.join();
+ EXPECT_TRUE(final_rpc_returned.load());
+ EXPECT_TRUE(_load->cancel().ok());
+ // Neither the later close failure nor cancellation can revoke arrival.
+ first->add_waiter(reply);
+ EXPECT_EQ(replies.load(), 3);
+}
+
+TEST_P(LoadChannelEosTest, FinalRpcKeepsItsResponseUntilCloseReturns) {
+ auto channel = make_channel(2);
+ auto first_response = std::make_unique<PTabletWriterAddBlockResult>();
+ std::shared_ptr<EosCompletion> first;
+ ASSERT_TRUE(_load->add_batch(eos(0), first_response.get(), &first).ok());
+ int replies = 0;
+ first->add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ first_response.reset();
+ ++replies;
+ });
+
+ bool reached_close = false;
+ auto* sp = SyncPoint::get_instance();
+ SyncPoint::CallbackGuard guard;
+ sp->set_call_back(
+ before_flush_sync_point(),
+ [&](auto&& args) {
+ reached_close = true;
+ EXPECT_EQ(replies, 1);
+ EXPECT_EQ(first_response, nullptr);
+ EXPECT_FALSE(_load->is_finished());
+ // Model tablet results written by the final closer after all
+ // earlier RPCs have already been released.
+ auto* response =
try_any_cast<PTabletWriterAddBlockResult*>(args[0]);
+ auto* tablet = response->add_tablet_vec();
+ tablet->set_tablet_id(123);
+ tablet->set_schema_hash(0);
+ tablet->set_received_rows(456);
+ tablet->set_num_rows_filtered(7);
+ },
+ &guard);
+ sp->enable_processing();
+ Defer disable_sync_points {[&] { sp->disable_processing(); }};
+
+ auto response = std::make_unique<PTabletWriterAddBlockResult>();
+ std::shared_ptr<EosCompletion> last;
+ ASSERT_TRUE(_load->add_batch(eos(1), response.get(), &last).ok());
+ EXPECT_TRUE(reached_close);
+ EXPECT_TRUE(_load->is_finished());
+ // Same ownership boundary as the service: finish synchronous writes before
+ // registering the final callback, which can destroy the response inline.
+ response->set_execution_time_us(123);
+ last->add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ EXPECT_EQ(response->execution_time_us(), 123);
+ ASSERT_EQ(response->tablet_vec_size(), 1);
+ EXPECT_EQ(response->tablet_vec(0).tablet_id(), 123);
+ EXPECT_EQ(response->tablet_vec(0).received_rows(), 456);
+ EXPECT_EQ(response->tablet_vec(0).num_rows_filtered(), 7);
+ response.reset();
+ ++replies;
+ });
+ EXPECT_EQ(response, nullptr);
+ EXPECT_EQ(replies, 2);
+}
+
+TEST_P(LoadChannelEosTest, ArrivalCallbacksAllowConcurrentRetryAndCancel) {
+ auto channel = make_channel(2);
+ PTabletWriterAddBlockResult response;
+ std::shared_ptr<EosCompletion> first;
+ ASSERT_TRUE(_load->add_batch(eos(0), &response, &first).ok());
+ int replies = 0;
+ first->add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ EXPECT_FALSE(_load->is_finished());
+ ++replies;
+ // The independent RPC worker must acquire the load/channel locks and
+ // finish while this callback is running. Joining it also checks that
+ // the final closer invokes callbacks without holding either lock.
+ PTabletWriterAddBlockResult duplicate_response;
+ std::shared_ptr<EosCompletion> duplicate;
+ EXPECT_TRUE(retry_on_rpc_thread(eos(0), &duplicate_response,
&duplicate).ok());
+ ASSERT_NE(duplicate, nullptr);
+ EXPECT_EQ(channel->_num_remaining_senders, 0);
+ duplicate->add_waiter([&](const Status& duplicate_status) {
+ EXPECT_TRUE(duplicate_status.ok());
+ ++replies;
+ });
+ EXPECT_TRUE(_load->cancel().ok());
+ });
+ int final_closes = 0;
+ auto* sp = SyncPoint::get_instance();
+ SyncPoint::CallbackGuard guard;
+ sp->set_call_back(
+ before_flush_sync_point(), [&](auto&&) { ++final_closes; },
&guard);
+ sp->enable_processing();
+ Defer disable_sync_points {[&] { sp->disable_processing(); }};
+
+ std::shared_ptr<EosCompletion> last;
+ EXPECT_TRUE(_load->add_batch(eos(1), &response, &last).ok());
+ last->add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ ++replies;
+ });
+ EXPECT_EQ(replies, 3);
+ EXPECT_EQ(final_closes, 1);
+ EXPECT_TRUE(_load->is_cancelled());
+ EXPECT_TRUE(_load->is_finished());
+}
+
+TEST_P(LoadChannelEosTest, FinalSenderRetryWaitsForCommitFailure) {
+ check_final_sender_retry(false, true);
+}
+
+TEST_P(LoadChannelEosTest, FinalSenderRetryReceivesTabletResults) {
+ check_final_sender_retry(false, false);
+}
+
+TEST_P(LoadChannelEosTest, FinalSenderBlockRetryWaitsForCommitFailure) {
+ check_final_sender_retry(true, true);
+}
+
+TEST_P(LoadChannelEosTest, FinalSenderBlockRetryReceivesTabletResults) {
+ check_final_sender_retry(true, false);
+}
+
+TEST_P(LoadChannelEosTest,
EarlierSenderBlockRetryDoesNotWriteDuringArrivalCallback) {
+ auto channel = make_channel(2);
+ channel->_next_seqs[0] = 8;
+ auto request = eos_with_block(0);
+ PTabletWriterAddBlockResult response;
+ std::shared_ptr<EosCompletion> first;
+ ASSERT_TRUE(_load->add_batch(request, &response, &first).ok());
+ int replies = 0;
+ first->add_waiter([&](const Status& status) {
+ EXPECT_TRUE(status.ok());
+ PTabletWriterAddBlockResult retry_response;
+ std::shared_ptr<EosCompletion> retry;
+ auto retry_status = retry_on_rpc_thread(request, &retry_response,
&retry);
+ EXPECT_TRUE(retry_status.ok()) << retry_status;
+ EXPECT_EQ(retry_response.tablet_errors_size(), 0);
+ EXPECT_EQ(channel->_next_seqs[0], 8);
+ ASSERT_NE(retry, nullptr);
+ retry->add_waiter([&](const Status& st) {
+ EXPECT_TRUE(st.ok());
+ ++replies;
+ });
+ });
+ std::shared_ptr<EosCompletion> last;
+ EXPECT_TRUE(_load->add_batch(eos(1), &response, &last).ok());
+ EXPECT_EQ(replies, 1);
+}
+
+TEST_P(LoadChannelEosTest, PacketSequenceAdmission) {
+ auto channel = make_channel(1);
+ channel->_next_seqs[0] = 8;
+ auto request = eos_with_block(0);
+ int64_t current_seq = 0;
+ bool should_write = true;
+ ASSERT_TRUE(channel->_get_current_seq(current_seq, request,
should_write).ok());
+ EXPECT_FALSE(should_write);
+ EXPECT_EQ(current_seq, 8);
+ request.set_packet_seq(8);
+ ASSERT_TRUE(channel->_get_current_seq(current_seq, request,
should_write).ok());
+ EXPECT_TRUE(should_write);
+ request.set_packet_seq(9);
+ EXPECT_FALSE(channel->_get_current_seq(current_seq, request,
should_write).ok());
+ EXPECT_FALSE(should_write);
+ EXPECT_TRUE(channel->cancel().ok());
+ request.set_packet_seq(8);
+ ASSERT_TRUE(channel->_get_current_seq(current_seq, request,
should_write).ok());
+ EXPECT_FALSE(should_write);
+}
+
+INSTANTIATE_TEST_SUITE_P(LocalAndCloud, LoadChannelEosTest, testing::Bool());
+
+} // namespace doris
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
