diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index ba89aa825..0cafc164c 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -536,6 +536,21 @@ 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_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, # 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..a735bac34 --- /dev/null +++ b/src/paimon/common/executor/executor_submission_test.cpp @@ -0,0 +1,187 @@ +/* + * 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 + +#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; + bool released; + 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(); + 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; + { + AllocationFault fault(ordinal); + try { + executor->Add(std::move(task)); + } catch (const std::bad_alloc&) { + caught = true; + } + } + 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; + } + 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_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); + } + } + } +} + +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; + 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(); + ScopeGuard release_on_error([&]() { release.set_value(); }); + executor->Add([gate]() { gate.wait(); }); + int32_t accepted = 0; + bool caught = false; + bool hit = false; + bool released = 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; + released = retained[accepted].expired(); + 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(released); + EXPECT_TRUE(recovered); + EXPECT_GT(accepted, 0); + EXPECT_EQ(runs.load(), accepted); +} +} // namespace paimon::test