From 674162a5f71c6f2db389ba6cf5fbbf622ca87624 Mon Sep 17 00:00:00 2001 From: Aaron Wu Date: Fri, 16 Oct 2026 06:37:48 +0800 Subject: [PATCH 1/4] fix(executor): start lazy workers before accepting the callback --- src/paimon/CMakeLists.txt | 10 + src/paimon/common/executor/executor.cpp | 3 +- .../executor/executor_submission_test.cpp | 174 ++++++++++++++++++ 3 files changed, 186 insertions(+), 1 deletion(-) create mode 100644 src/paimon/common/executor/executor_submission_test.cpp diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index ba89aa825..022d2bbbe 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -536,6 +536,16 @@ add_subdirectory(common/file_index) add_subdirectory(common/global_index) if(PAIMON_BUILD_TESTS) + # Wrap actual executor allocations, not calls inside an already linked DSO. + if(PAIMON_BUILD_STATIC + AND CMAKE_SYSTEM_NAME STREQUAL "Linux" + AND CMAKE_CXX_COMPILER_ID MATCHES "^(GNU|Clang)$") + add_paimon_test(executor_submission_test + SOURCES common/executor/executor_submission_test.cpp + STATIC_LINK_LIBS paimon_static ${GTEST_LINK_TOOLCHAIN}) + target_link_options(paimon-executor-submission-test PRIVATE "LINKER:--wrap=_Znwm") + endif() + # The shared HTTP client is compiled only for the components that need libcurl, # so its test follows the same gate. set(PAIMON_COMMON_HTTP_CLIENT_TEST_SRCS) diff --git a/src/paimon/common/executor/executor.cpp b/src/paimon/common/executor/executor.cpp index 92aef4a09..11c280391 100644 --- a/src/paimon/common/executor/executor.cpp +++ b/src/paimon/common/executor/executor.cpp @@ -113,13 +113,14 @@ void DefaultExecutor::Add(std::function func) { if (state_->stop) { return; } - state_->tasks.emplace(std::move(func)); if (workers_.empty()) { workers_.reserve(thread_count_); for (uint32_t i = 0; i < thread_count_; ++i) { workers_.emplace_back(&DefaultExecutor::WorkerThread, state_); } } + // Startup may throw; do not accept the callback until it succeeds. + state_->tasks.emplace(std::move(func)); } state_->condition.notify_one(); } diff --git a/src/paimon/common/executor/executor_submission_test.cpp b/src/paimon/common/executor/executor_submission_test.cpp new file mode 100644 index 000000000..75b8cbb0b --- /dev/null +++ b/src/paimon/common/executor/executor_submission_test.cpp @@ -0,0 +1,174 @@ +/* + * 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 +#include +#include +#include +#include +#include +#include +#include +#include + +#include "gtest/gtest.h" +#include "paimon/common/utils/scope_guard.h" +#include "paimon/executor.h" + +namespace { +thread_local bool armed = false; +thread_local bool fired = false; +thread_local int32_t allocations = 0; +thread_local int32_t fail_at = 0; + +class AllocationFault { + public: + explicit AllocationFault(int32_t ordinal) { + allocations = 0; + fail_at = ordinal; + fired = false; + armed = true; + } + ~AllocationFault() { + armed = false; + } +}; +} // namespace + +extern "C" void* __real__Znwm(std::size_t size); +extern "C" void* __wrap__Znwm(std::size_t size) { + if (armed && ++allocations == fail_at) { + armed = false; + fired = true; + throw std::bad_alloc(); + } + return __real__Znwm(size); +} + +namespace paimon::test { +namespace { +struct SubmissionResult { + int32_t allocations; + bool fired; + bool caught; + bool recovered; + int32_t runs; +}; + +SubmissionResult Submit(uint32_t workers, int32_t ordinal, bool recover, bool shutdown) { + std::atomic runs = 0; + std::promise marker; + auto ready = marker.get_future(); + auto executor = CreateDefaultExecutor(workers).value(); + std::function task = [&]() { ++runs; }; + std::function recovery = [&]() { marker.set_value(); }; + bool caught = false; + { + AllocationFault fault(ordinal); + try { + executor->Add(std::move(task)); + } catch (const std::bad_alloc&) { + caught = true; + } + } + SubmissionResult result = {allocations, fired, caught, false, 0}; + if (recover) { + executor->Add(std::move(recovery)); + result.recovered = ready.wait_for(std::chrono::seconds(5)) == std::future_status::ready; + } + if (shutdown) { + executor->ShutdownNow(); + } + // Graceful destruction also drains callbacks on the multi-worker oracle. + executor.reset(); + result.runs = runs.load(); + return result; +} +} // namespace + +TEST(ExecutorSubmissionTest, StartupAllocationFailuresDoNotAcceptCallback) { + for (uint32_t workers : {1u, 4u}) { + const auto observation = Submit(workers, 0, true, false); + ASSERT_GT(observation.allocations, 0); + ASSERT_FALSE(observation.fired); + ASSERT_FALSE(observation.caught); + ASSERT_TRUE(observation.recovered); + ASSERT_EQ(observation.runs, 1); + for (int32_t ordinal = 1; ordinal <= observation.allocations; ++ordinal) { + SCOPED_TRACE(::testing::Message() << "workers=" << workers << ", ordinal=" << ordinal); + const auto result = Submit(workers, ordinal, true, false); + EXPECT_TRUE(result.fired); + EXPECT_TRUE(result.caught); + EXPECT_EQ(result.allocations, ordinal); + EXPECT_TRUE(result.recovered); + EXPECT_EQ(result.runs, 0); + for (bool shutdown : {false, true}) { + const auto cleanup = Submit(workers, ordinal, false, shutdown); + EXPECT_TRUE(cleanup.fired); + EXPECT_TRUE(cleanup.caught); + EXPECT_EQ(cleanup.runs, 0); + } + } + } +} + +TEST(ExecutorSubmissionTest, QueueAllocationFailurePreservesAcceptedPrefix) { + std::promise release; + auto gate = release.get_future().share(); + std::promise marker; + auto ready = marker.get_future(); + std::atomic runs = 0; + std::array, 128> tasks; + for (auto& task : tasks) { + task = [&]() { ++runs; }; + } + std::function recovery = [&]() { marker.set_value(); }; + auto executor = CreateDefaultExecutor(1).value(); + ScopeGuard release_on_error([&]() { release.set_value(); }); + executor->Add([gate]() { gate.wait(); }); + int32_t accepted = 0; + bool caught = false; + bool hit = false; + for (auto& task : tasks) { + { + AllocationFault fault(1); + try { + executor->Add(std::move(task)); + } catch (const std::bad_alloc&) { + caught = true; + } + } + if (fired) { + hit = true; + break; + } + ++accepted; + } + release.set_value(); + release_on_error.Release(); + executor->Add(std::move(recovery)); + const bool recovered = ready.wait_for(std::chrono::seconds(5)) == std::future_status::ready; + executor.reset(); + EXPECT_TRUE(hit); + EXPECT_TRUE(caught); + EXPECT_TRUE(recovered); + EXPECT_GT(accepted, 0); + EXPECT_EQ(runs.load(), accepted); +} +} // namespace paimon::test From 155f85edc09b95b69e9829fc6e33da8dc9dbceea Mon Sep 17 00:00:00 2001 From: Aaron Wu Date: Fri, 16 Oct 2026 06:39:25 +0800 Subject: [PATCH 2/4] test(executor): bound fault regressions and assert rejected capture release --- src/paimon/CMakeLists.txt | 2 ++ .../common/executor/executor_submission_test.cpp | 11 +++++++++-- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index 022d2bbbe..8bf086e76 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -539,11 +539,13 @@ if(PAIMON_BUILD_TESTS) # Wrap actual executor allocations, not calls inside an already linked DSO. if(PAIMON_BUILD_STATIC AND CMAKE_SYSTEM_NAME STREQUAL "Linux" + AND CMAKE_SIZEOF_VOID_P EQUAL 8 AND CMAKE_CXX_COMPILER_ID MATCHES "^(GNU|Clang)$") add_paimon_test(executor_submission_test SOURCES common/executor/executor_submission_test.cpp STATIC_LINK_LIBS paimon_static ${GTEST_LINK_TOOLCHAIN}) target_link_options(paimon-executor-submission-test PRIVATE "LINKER:--wrap=_Znwm") + set_tests_properties(paimon-executor-submission-test PROPERTIES TIMEOUT 30) endif() # The shared HTTP client is compiled only for the components that need libcurl, diff --git a/src/paimon/common/executor/executor_submission_test.cpp b/src/paimon/common/executor/executor_submission_test.cpp index 75b8cbb0b..b2030e798 100644 --- a/src/paimon/common/executor/executor_submission_test.cpp +++ b/src/paimon/common/executor/executor_submission_test.cpp @@ -24,6 +24,7 @@ #include #include #include +#include #include #include @@ -68,6 +69,7 @@ struct SubmissionResult { bool fired; bool caught; bool recovered; + bool released; int32_t runs; }; @@ -76,7 +78,10 @@ SubmissionResult Submit(uint32_t workers, int32_t ordinal, bool recover, bool sh std::promise marker; auto ready = marker.get_future(); auto executor = CreateDefaultExecutor(workers).value(); - std::function task = [&]() { ++runs; }; + auto owner = std::make_shared(0); + std::weak_ptr retained = owner; + std::function task = [&, owner]() { ++runs; }; + owner.reset(); std::function recovery = [&]() { marker.set_value(); }; bool caught = false; { @@ -87,7 +92,7 @@ SubmissionResult Submit(uint32_t workers, int32_t ordinal, bool recover, bool sh caught = true; } } - SubmissionResult result = {allocations, fired, caught, false, 0}; + SubmissionResult result = {allocations, fired, caught, false, retained.expired(), 0}; if (recover) { executor->Add(std::move(recovery)); result.recovered = ready.wait_for(std::chrono::seconds(5)) == std::future_status::ready; @@ -117,11 +122,13 @@ TEST(ExecutorSubmissionTest, StartupAllocationFailuresDoNotAcceptCallback) { EXPECT_TRUE(result.caught); EXPECT_EQ(result.allocations, ordinal); EXPECT_TRUE(result.recovered); + EXPECT_TRUE(result.released); EXPECT_EQ(result.runs, 0); for (bool shutdown : {false, true}) { const auto cleanup = Submit(workers, ordinal, false, shutdown); EXPECT_TRUE(cleanup.fired); EXPECT_TRUE(cleanup.caught); + EXPECT_TRUE(cleanup.released); EXPECT_EQ(cleanup.runs, 0); } } From 5159503817431e4917db0d2e24646c1f27502ac2 Mon Sep 17 00:00:00 2001 From: Aaron Wu Date: Fri, 16 Oct 2026 06:42:58 +0800 Subject: [PATCH 3/4] style: apply pinned CMake formatter to executor fault target --- src/paimon/CMakeLists.txt | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index 8bf086e76..0cafc164c 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -542,8 +542,11 @@ if(PAIMON_BUILD_TESTS) AND CMAKE_SIZEOF_VOID_P EQUAL 8 AND CMAKE_CXX_COMPILER_ID MATCHES "^(GNU|Clang)$") add_paimon_test(executor_submission_test - SOURCES common/executor/executor_submission_test.cpp - STATIC_LINK_LIBS paimon_static ${GTEST_LINK_TOOLCHAIN}) + SOURCES + common/executor/executor_submission_test.cpp + STATIC_LINK_LIBS + paimon_static + ${GTEST_LINK_TOOLCHAIN}) target_link_options(paimon-executor-submission-test PRIVATE "LINKER:--wrap=_Znwm") set_tests_properties(paimon-executor-submission-test PROPERTIES TIMEOUT 30) endif() From d14ead9ef02b87cb33c7b7ca1ad3b0e2371a14be Mon Sep 17 00:00:00 2001 From: Aaron Wu Date: Fri, 16 Oct 2026 06:43:58 +0800 Subject: [PATCH 4/4] test(executor): assert immediate capture release on queue refusal --- .../common/executor/executor_submission_test.cpp | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/src/paimon/common/executor/executor_submission_test.cpp b/src/paimon/common/executor/executor_submission_test.cpp index b2030e798..a735bac34 100644 --- a/src/paimon/common/executor/executor_submission_test.cpp +++ b/src/paimon/common/executor/executor_submission_test.cpp @@ -142,8 +142,11 @@ TEST(ExecutorSubmissionTest, QueueAllocationFailurePreservesAcceptedPrefix) { auto ready = marker.get_future(); std::atomic runs = 0; std::array, 128> tasks; - for (auto& task : tasks) { - task = [&]() { ++runs; }; + std::array, 128> retained; + for (std::size_t index = 0; index < tasks.size(); ++index) { + auto owner = std::make_shared(0); + retained[index] = owner; + tasks[index] = [&, owner]() { ++runs; }; } std::function recovery = [&]() { marker.set_value(); }; auto executor = CreateDefaultExecutor(1).value(); @@ -152,6 +155,7 @@ TEST(ExecutorSubmissionTest, QueueAllocationFailurePreservesAcceptedPrefix) { int32_t accepted = 0; bool caught = false; bool hit = false; + bool released = false; for (auto& task : tasks) { { AllocationFault fault(1); @@ -163,6 +167,7 @@ TEST(ExecutorSubmissionTest, QueueAllocationFailurePreservesAcceptedPrefix) { } if (fired) { hit = true; + released = retained[accepted].expired(); break; } ++accepted; @@ -174,6 +179,7 @@ TEST(ExecutorSubmissionTest, QueueAllocationFailurePreservesAcceptedPrefix) { executor.reset(); EXPECT_TRUE(hit); EXPECT_TRUE(caught); + EXPECT_TRUE(released); EXPECT_TRUE(recovered); EXPECT_GT(accepted, 0); EXPECT_EQ(runs.load(), accepted);