This is an automated email from the ASF dual-hosted git repository.
morningman 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 71a63249bbd [fix](be) Own the allocated LSN map in SharedMemtable to
fix UAF on flush task teardown (#67442)
71a63249bbd is described below
commit 71a63249bbd578a9aa18129d0cc20dfec86c5d48
Author: Nelson Boss <[email protected]>
AuthorDate: Thu Sep 3 11:57:09 2026 +0800
[fix](be) Own the allocated LSN map in SharedMemtable to fix UAF on flush
task teardown (#67442)
### What problem does this PR solve?
Issue Number: close #67428
Related PR: #66889 (introduced the UAF)
Problem Summary:
ASAN heap-use-after-free in `SharedMemtable::~SharedMemtable()` during
group-commit memtable flush task teardown, introduced by #66889.
`PartOfGroupMemtableFlushTask` holds its `SharedMemtable` by
`shared_ptr` but only a `weak_ptr` to the `FlushToken`. When `run()`'s
local `shared_ptr<FlushToken>` drops the last reference at the end of
`run()`, destruction cascades `FlushToken` -> `RowsetWriter` ->
`RowsetWriterContext`; the thread pool then destroys the task object,
and `~SharedMemtable()` dereferences the dangling raw
`RowsetWriterContext* rowset_ctx` in `remove_segment_allocated_lsns()`.
Reproduced by the nonConcurrent pipeline on two unrelated PRs (#67404,
#67402).
### What changed?
- `SharedMemtable` now owns `std::shared_ptr<SegmentAllocatedLsnMap>`
captured from the group writer context at submission; insert/remove go
through the owned map. A null map stands for "no LSN allocation"
(equivalent to `need_allocated_lsn()`, since `GroupRowsetWriter::init()`
creates the map exactly when needed). This keeps the precise cleanup
dependency alive without extending the whole `RowsetWriter` lifetime —
the approach recommended in the #67428 triage.
- Add `SegmentAllocatedLsnMap::contains_segment()` for test assertions.
- Regression tests covering: (a) the last token/writer owner released
while a group flush task finishes, (b) a queued subtask running after
its weak token expired, (c) cancellation, plus LSN-entry cleanup
assertions on the flush-error path.
### Verification
ASAN BE UT, both directions: **without** the fix the new tests abort
with the exact reported UAF (`SUMMARY: AddressSanitizer:
heap-use-after-free rowset_writer_context.h:202:9 in
doris::RowsetWriterContext::remove_segment_allocated_lsns`); **with**
the fix the `MemTableFlushExecutor*` tests pass 8/8 (two runs).
---
be/src/load/memtable/memtable_flush_executor.cpp | 17 +-
be/src/load/memtable/memtable_flush_executor.h | 8 +-
be/src/storage/binlog.h | 5 +
.../load/memtable/memtable_flush_executor_test.cpp | 171 ++++++++++++++++++++-
4 files changed, 189 insertions(+), 12 deletions(-)
diff --git a/be/src/load/memtable/memtable_flush_executor.cpp
b/be/src/load/memtable/memtable_flush_executor.cpp
index 1970d074a5e..2ad61661d4f 100644
--- a/be/src/load/memtable/memtable_flush_executor.cpp
+++ b/be/src/load/memtable/memtable_flush_executor.cpp
@@ -120,8 +120,8 @@ SharedMemtable::~SharedMemtable() {
return;
}
if (has_allocated_lsns) {
- DCHECK(rowset_ctx != nullptr);
- rowset_ctx->remove_segment_allocated_lsns(segment_id);
+ DCHECK(allocated_lsn_map != nullptr);
+ allocated_lsn_map->remove_segment(segment_id);
}
DCHECK(memtable != nullptr);
SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(
@@ -187,8 +187,7 @@ Status FlushToken::submit(std::shared_ptr<MemTable>
mem_table) {
shared_memtable = std::make_shared<SharedMemtable>();
shared_memtable->memtable = mem_table;
- shared_memtable->rowset_ctx =
-
const_cast<RowsetWriterContext*>(&group_rowset_writer->context());
+ shared_memtable->allocated_lsn_map =
group_rowset_writer->context().allocated_lsn_map;
// Keep data/binlog segment_id allocators in sync.
auto segment_id = DORIS_TRY(data_writer->allocate_segment_id());
auto binlog_segment_id =
DORIS_TRY(binlog_writer->allocate_segment_id());
@@ -314,13 +313,13 @@ Status FlushToken::_memtable2block(MemTable* memtable,
SharedMemtable* shared_me
shared_memtable->block_status = memtable->to_block(&block);
if (shared_memtable->block_status.ok()) {
shared_memtable->block.reset(block.release());
- auto* rowset_ctx = shared_memtable->rowset_ctx;
- DCHECK(rowset_ctx != nullptr);
- if (rowset_ctx->need_allocated_lsn() &&
shared_memtable->block->rows() > 0) {
+ // A non-null map is equivalent to
RowsetWriterContext::need_allocated_lsn():
+ // GroupRowsetWriter::init() creates the map exactly when LSN
allocation is needed.
+ const auto& lsn_map = shared_memtable->allocated_lsn_map;
+ if (lsn_map != nullptr && shared_memtable->block->rows() > 0) {
auto memtable_lsns = memtable->allocated_lsns();
DCHECK_EQ(memtable_lsns->size(),
shared_memtable->block->rows());
-
rowset_ctx->insert_segment_allocated_lsns(shared_memtable->segment_id,
- memtable_lsns);
+
lsn_map->insert_segment_allocated_lsns(shared_memtable->segment_id,
memtable_lsns);
shared_memtable->has_allocated_lsns = true;
}
}
diff --git a/be/src/load/memtable/memtable_flush_executor.h
b/be/src/load/memtable/memtable_flush_executor.h
index c66051cc5f7..c3053d483e5 100644
--- a/be/src/load/memtable/memtable_flush_executor.h
+++ b/be/src/load/memtable/memtable_flush_executor.h
@@ -28,6 +28,7 @@
#include "common/status.h"
#include "load/delta_writer/delta_writer_context.h"
#include "load/memtable/memtable.h"
+#include "storage/binlog.h"
#include "util/threadpool.h"
namespace doris {
@@ -64,7 +65,12 @@ struct SharedMemtable {
std::once_flag block_once;
Status block_status;
std::shared_ptr<Block> block;
- RowsetWriterContext* rowset_ctx = nullptr;
+ // Owns the segment LSN map so cleanup in ~SharedMemtable stays valid even
when
+ // the FlushToken's last reference drops inside
PartOfGroupMemtableFlushTask::run()
+ // before the thread pool destroys the task (and with it this
SharedMemtable).
+ // A null map means this group writes no per-row LSNs (see
GroupRowsetWriter::init,
+ // which creates the map exactly when LSN allocation is needed).
+ std::shared_ptr<segment_v2::SegmentAllocatedLsnMap> allocated_lsn_map;
bool has_allocated_lsns = false;
std::atomic<int> finished_sub_task_count {0};
diff --git a/be/src/storage/binlog.h b/be/src/storage/binlog.h
index b59eded10ce..c4e227e387a 100644
--- a/be/src/storage/binlog.h
+++ b/be/src/storage/binlog.h
@@ -181,6 +181,11 @@ public:
_seg_id_to_lsn_ids.erase(seg_id);
}
+ bool contains_segment(int64_t seg_id) const {
+ std::lock_guard<std::mutex> l(_mutex);
+ return _seg_id_to_lsn_ids.count(seg_id) > 0;
+ }
+
ConstAllocatedLsnVectorSharedPtr get_segment_allocated_lsns(int64_t
seg_id) const {
std::lock_guard<std::mutex> l(_mutex);
auto it = _seg_id_to_lsn_ids.find(seg_id);
diff --git a/be/test/load/memtable/memtable_flush_executor_test.cpp
b/be/test/load/memtable/memtable_flush_executor_test.cpp
index b53d0891f8c..ad6e27dd0c5 100644
--- a/be/test/load/memtable/memtable_flush_executor_test.cpp
+++ b/be/test/load/memtable/memtable_flush_executor_test.cpp
@@ -55,13 +55,16 @@ namespace {
class MockRowsetWriter final : public RowsetWriter {
public:
+ // `flush_enter_cnt`, when set, is incremented at the top of
flush_memtable() BEFORE the
+ // optional delay, so tests can deterministically wait for "task is inside
flush".
explicit MockRowsetWriter(std::atomic<int>* flush_cnt, bool fail_on_flush
= false,
const std::string& flush_error_msg = "mock flush
failed",
- int flush_delay_ms = 0)
+ int flush_delay_ms = 0, std::atomic<int>*
flush_enter_cnt = nullptr)
: _flush_cnt(flush_cnt),
_fail_on_flush(fail_on_flush),
_flush_error_msg(flush_error_msg),
- _flush_delay_ms(flush_delay_ms) {}
+ _flush_delay_ms(flush_delay_ms),
+ _flush_enter_cnt(flush_enter_cnt) {}
Status init(const RowsetWriterContext& ctx) override {
_context = ctx;
@@ -79,6 +82,9 @@ public:
Status flush_memtable(Block* block, int32_t segment_id, int64_t*
flush_size) override {
EXPECT_GT(block->rows(), 0);
+ if (_flush_enter_cnt != nullptr) {
+ ++(*_flush_enter_cnt);
+ }
if (_flush_delay_ms > 0) {
std::this_thread::sleep_for(std::chrono::milliseconds(_flush_delay_ms));
}
@@ -129,6 +135,7 @@ private:
bool _fail_on_flush;
std::string _flush_error_msg;
int _flush_delay_ms;
+ std::atomic<int>* _flush_enter_cnt;
int32_t _next_segment_id = 0;
int32_t _last_segment_id = -1;
ConstAllocatedLsnVectorSharedPtr _last_seg_lsn = nullptr;
@@ -553,6 +560,166 @@ TEST_F(MemTableFlushExecutorGroupFlushTest,
TestGroupFlushTokenPartialSuccess) {
EXPECT_EQ(1, flush_token->get_stats().flush_finish_count.load());
EXPECT_EQ(0, flush_token->get_stats().flush_submit_count.load());
+ // Flush-error path: the LSN entry inserted by the data subtask (whose
_memtable2block ran
+ // call_once before the binlog flush failed) must still be cleaned up when
the shared
+ // memtable is destroyed. wait() only observes the in-run() deferred
counters, so give the
+ // pool a moment to destroy the finished task objects first.
+ auto lsn_map = group_writer->context().allocated_lsn_map;
+ ASSERT_NE(lsn_map, nullptr);
+ int32_t seg_id = data_writer->last_segment_id();
+ std::this_thread::sleep_for(std::chrono::milliseconds(200));
+ EXPECT_FALSE(lsn_map->contains_segment(seg_id));
+
+ drop_tablet(ctx.request);
+}
+
+// Regression test for the use-after-free reported in issue #67428 (scenario
a): the last
+// external FlushToken/RowsetWriter owner is released while a group flush task
is still inside
+// flush_memtable(). When that task's run() returns, its local
shared_ptr<FlushToken> drops the
+// last reference, destroying the FlushToken -> GroupRowsetWriter ->
RowsetWriterContext, and
+// the subsequent task-object destruction runs ~SharedMemtable, which used to
dereference the
+// dangling raw RowsetWriterContext*. Now SharedMemtable owns the LSN map, so
this must be safe.
+TEST_F(MemTableFlushExecutorGroupFlushTest, TestGroupFlushTaskOutlivesToken) {
+ SCOPED_INIT_THREAD_CONTEXT();
+
+ GroupFlushTestContext ctx;
+ prepare_group_flush_test_context(10004, 270068376, {4000, 4001}, &ctx);
+
+ std::atomic<int> data_flush_cnt = 0;
+ std::atomic<int> binlog_flush_cnt = 0;
+ std::atomic<int> binlog_flush_enter_cnt = 0;
+ auto data_writer = std::make_shared<MockRowsetWriter>(&data_flush_cnt);
+ auto binlog_writer = std::make_shared<MockRowsetWriter>(&binlog_flush_cnt,
false, "", 500,
+
&binlog_flush_enter_cnt);
+ std::shared_ptr<GroupRowsetWriter> group_writer;
+ ASSERT_TRUE(create_group_rowset_writer(ctx, 4, data_writer, binlog_writer,
&group_writer).ok());
+
+ std::unique_ptr<ThreadPool> pool;
+ ASSERT_TRUE(ThreadPoolBuilder("MemTableGroupFlushTestPool")
+ .set_min_threads(2)
+ .set_max_threads(2)
+ .build(&pool)
+ .ok());
+
+ std::shared_ptr<FlushToken> flush_token;
+ ASSERT_TRUE(create_group_flush_token(ctx, group_writer, &flush_token,
pool.get()).ok());
+ ASSERT_TRUE(flush_token->submit(ctx.memtable).ok());
+
+ // Capture the LSN map and segment id while the writer is still alive,
then drop every
+ // external reference to the token and the group writer while the binlog
subtask is parked
+ // inside its delayed flush_memtable() and the data subtask has fully
finished.
+ auto lsn_map = group_writer->context().allocated_lsn_map;
+ ASSERT_NE(lsn_map, nullptr);
+ while (binlog_flush_enter_cnt.load() == 0 || data_flush_cnt.load() == 0) {
+ std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ }
+ flush_token.reset();
+ group_writer.reset();
+
+ // The binlog subtask finishes run(): its local token shared_ptr is now
the LAST reference,
+ // so the FlushToken and the GroupRowsetWriter (with RowsetWriterContext)
die at the end of
+ // run(), before the thread pool destroys the task and ~SharedMemtable()
runs. Before the
+ // fix this read freed memory (ASAN heap-use-after-free); after the fix
SharedMemtable owns
+ // the LSN map and the cleanup below is safe.
+ std::this_thread::sleep_for(std::chrono::milliseconds(1000));
+ EXPECT_EQ(1, binlog_flush_cnt.load());
+ EXPECT_FALSE(lsn_map->contains_segment(data_writer->last_segment_id()));
+
+ drop_tablet(ctx.request);
+}
+
+// Regression test for issue #67428 (scenario b): a queued group subtask runs
AFTER its weak
+// FlushToken reference has already expired. The token's last strong reference
dies at the end
+// of the first subtask's run(); the sibling subtask then observes the expired
token, skips its
+// flush, and its task-object destruction still runs ~SharedMemtable with LSNs
already inserted
+// by the first subtask — which must not touch freed state.
+TEST_F(MemTableFlushExecutorGroupFlushTest,
TestGroupFlushQueuedSubtaskAfterTokenExpired) {
+ SCOPED_INIT_THREAD_CONTEXT();
+
+ GroupFlushTestContext ctx;
+ prepare_group_flush_test_context(10005, 270068377, {5000, 5001}, &ctx);
+
+ std::atomic<int> data_flush_cnt = 0;
+ std::atomic<int> binlog_flush_cnt = 0;
+ std::atomic<int> data_flush_enter_cnt = 0;
+ // Single-threaded pool: the data subtask runs first (parked in its
delayed flush), the
+ // binlog subtask stays queued until the data subtask's run() has returned.
+ auto data_writer = std::make_shared<MockRowsetWriter>(&data_flush_cnt,
false, "", 500,
+
&data_flush_enter_cnt);
+ auto binlog_writer = std::make_shared<MockRowsetWriter>(&binlog_flush_cnt);
+ std::shared_ptr<GroupRowsetWriter> group_writer;
+ ASSERT_TRUE(create_group_rowset_writer(ctx, 5, data_writer, binlog_writer,
&group_writer).ok());
+
+ std::unique_ptr<ThreadPool> pool;
+ ASSERT_TRUE(ThreadPoolBuilder("MemTableGroupFlushTestPool")
+ .set_min_threads(1)
+ .set_max_threads(1)
+ .build(&pool)
+ .ok());
+
+ std::shared_ptr<FlushToken> flush_token;
+ ASSERT_TRUE(create_group_flush_token(ctx, group_writer, &flush_token,
pool.get()).ok());
+ ASSERT_TRUE(flush_token->submit(ctx.memtable).ok());
+
+ auto lsn_map = group_writer->context().allocated_lsn_map;
+ ASSERT_NE(lsn_map, nullptr);
+ // Wait until the data subtask is inside its delayed flush (the LSN entry
is already
+ // inserted by _memtable2block at this point), then drop the external
references while the
+ // data subtask's run() still holds the token alive. When its run()
returns, the local
+ // token shared_ptr is the last reference and the
FlushToken/GroupRowsetWriter die; the
+ // queued binlog subtask then runs with an expired weak token.
+ while (data_flush_enter_cnt.load() == 0) {
+ std::this_thread::sleep_for(std::chrono::milliseconds(10));
+ }
+ flush_token.reset();
+ group_writer.reset();
+
+ std::this_thread::sleep_for(std::chrono::milliseconds(1000));
+ EXPECT_EQ(1, data_flush_cnt.load());
+ // The binlog subtask observed the expired token and skipped its flush.
+ EXPECT_EQ(0, binlog_flush_cnt.load());
+ // The shared memtable was destroyed by the binlog task's destruction and
removed the
+ // segment's LSN entry through its OWNED map (no dangling
RowsetWriterContext access).
+ EXPECT_FALSE(lsn_map->contains_segment(data_writer->last_segment_id()));
+
+ drop_tablet(ctx.request);
+}
+
+// Cancelled path: after cancel(), submitted subtasks skip their flush
entirely, no LSN entry is
+// ever inserted, and the task teardown must stay clean.
+TEST_F(MemTableFlushExecutorGroupFlushTest,
TestGroupFlushTokenCancelledCleanup) {
+ SCOPED_INIT_THREAD_CONTEXT();
+
+ GroupFlushTestContext ctx;
+ prepare_group_flush_test_context(10006, 270068378, {6000, 6001}, &ctx);
+
+ std::atomic<int> data_flush_cnt = 0;
+ std::atomic<int> binlog_flush_cnt = 0;
+ auto data_writer = std::make_shared<MockRowsetWriter>(&data_flush_cnt);
+ auto binlog_writer = std::make_shared<MockRowsetWriter>(&binlog_flush_cnt);
+ std::shared_ptr<GroupRowsetWriter> group_writer;
+ ASSERT_TRUE(create_group_rowset_writer(ctx, 6, data_writer, binlog_writer,
&group_writer).ok());
+
+ std::unique_ptr<ThreadPool> pool;
+ ASSERT_TRUE(ThreadPoolBuilder("MemTableGroupFlushTestPool")
+ .set_min_threads(2)
+ .set_max_threads(2)
+ .build(&pool)
+ .ok());
+
+ std::shared_ptr<FlushToken> flush_token;
+ ASSERT_TRUE(create_group_flush_token(ctx, group_writer, &flush_token,
pool.get()).ok());
+ flush_token->cancel();
+ ASSERT_TRUE(flush_token->submit(ctx.memtable).ok());
+ ASSERT_TRUE(flush_token->wait().ok());
+
+ auto lsn_map = group_writer->context().allocated_lsn_map;
+ ASSERT_NE(lsn_map, nullptr);
+ // No flush ran, so no LSN entry was inserted and there is nothing stale
to leak.
+ std::this_thread::sleep_for(std::chrono::milliseconds(200));
+ EXPECT_EQ(0, data_flush_cnt.load());
+ EXPECT_EQ(0, binlog_flush_cnt.load());
+
drop_tablet(ctx.request);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
