diff --git a/src/paimon/common/executor/default_executor_test.cpp b/src/paimon/common/executor/default_executor_test.cpp index 07e78681b..80775f2f6 100644 --- a/src/paimon/common/executor/default_executor_test.cpp +++ b/src/paimon/common/executor/default_executor_test.cpp @@ -26,6 +26,10 @@ #include #include +#ifdef __linux__ +#include +#endif + #include "gtest/gtest.h" #include "paimon/common/executor/future.h" #include "paimon/executor.h" @@ -35,6 +39,89 @@ namespace paimon::test { +#ifdef __linux__ +// Number of threads of the current process according to /proc. +int32_t CountProcessThreads() { + DIR* dir = opendir("/proc/self/task"); + if (dir == nullptr) { + return -1; + } + int32_t count = 0; + while (struct dirent* entry = readdir(dir)) { + if (entry->d_name[0] != '.') { + ++count; + } + } + closedir(dir); + return count; +} + +// A worker joined by an earlier test can trail in /proc for a moment, so take +// the count only once two consecutive reads agree. +int32_t StableProcessThreadCount() { + int32_t last = CountProcessThreads(); + for (int32_t i = 0; i < 100; ++i) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + int32_t current = CountProcessThreads(); + if (current == last) { + return current; + } + last = current; + } + return last; +} +#endif + +TEST(DefaultExecutorTest, TestWorkersStartOnFirstTask) { +#ifdef __linux__ + const int32_t threads_before = StableProcessThreadCount(); + ASSERT_GT(threads_before, 0); +#endif + ASSERT_OK_AND_ASSIGN(std::unique_ptr executor, CreateDefaultExecutor(4)); + ASSERT_EQ(4u, executor->GetThreadNum()); +#ifdef __linux__ + // Constructing the executor does not spawn any worker thread. + ASSERT_LE(CountProcessThreads(), threads_before); +#endif + + std::atomic sum = {0}; + std::vector> futures; + for (int32_t index = 0; index < 8; ++index) { + futures.push_back(Via(executor.get(), [&sum]() { sum++; })); + } + Wait(futures); + ASSERT_EQ(8, sum.load()); +#ifdef __linux__ + // The first task started all four workers. + ASSERT_GE(CountProcessThreads(), threads_before + 4); +#endif + executor.reset(); +#ifdef __linux__ + // Destroying the executor joined them; the joined threads may trail in + // /proc for a moment, so poll briefly. + int32_t threads_after = CountProcessThreads(); + for (int32_t i = 0; i < 100 && threads_after > threads_before; ++i) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + threads_after = CountProcessThreads(); + } + ASSERT_LE(threads_after, threads_before); +#endif +} + +TEST(DefaultExecutorTest, TestShutdownWithoutTasks) { + // Shutting down or destroying an executor that never ran a task must not + // block or touch workers that were never started. + ASSERT_OK_AND_ASSIGN(std::unique_ptr executor, CreateDefaultExecutor(2)); + executor->ShutdownNow(); + std::atomic ran = {false}; + executor->Add([&ran]() { ran = true; }); + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + ASSERT_FALSE(ran.load()); + executor.reset(); + std::unique_ptr idle_executor = CreateDefaultExecutor(); + idle_executor.reset(); +} + TEST(DefaultExecutorTest, TestViaVoidFunc) { auto executor = GetGlobalDefaultExecutor(); std::atomic sum = {0}; diff --git a/src/paimon/common/executor/executor.cpp b/src/paimon/common/executor/executor.cpp index 45253fe1f..92aef4a09 100644 --- a/src/paimon/common/executor/executor.cpp +++ b/src/paimon/common/executor/executor.cpp @@ -52,15 +52,15 @@ class DefaultExecutor : public Executor { private: uint32_t thread_count_; + // Guarded by state_->mutex; populated on the first Add(). std::vector workers_; std::shared_ptr state_ = std::make_shared(); }; DefaultExecutor::DefaultExecutor(uint32_t thread_count) : thread_count_(thread_count) { assert(thread_count > 0); - for (uint32_t i = 0; i < thread_count_; ++i) { - workers_.emplace_back(&DefaultExecutor::WorkerThread, state_); - } + // Worker threads are started lazily by the first Add(): an executor that + // never receives a task never spawns a thread. } uint32_t DefaultExecutor::GetThreadNum() const { @@ -68,6 +68,7 @@ uint32_t DefaultExecutor::GetThreadNum() const { } void DefaultExecutor::ShutdownInternal(bool wait_for_pending_tasks) { + std::vector workers; { std::unique_lock lock(state_->mutex); if (state_->stop) { @@ -80,8 +81,9 @@ void DefaultExecutor::ShutdownInternal(bool wait_for_pending_tasks) { state_->tasks.swap(empty); } state_->condition.notify_all(); + workers.swap(workers_); } - for (std::thread& worker : workers_) { + for (std::thread& worker : workers) { if (worker.joinable()) { if (worker.get_id() == std::this_thread::get_id()) { worker.detach(); @@ -112,6 +114,12 @@ void DefaultExecutor::Add(std::function func) { 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_); + } + } } state_->condition.notify_one(); } diff --git a/src/paimon/core/operation/scan_context.cpp b/src/paimon/core/operation/scan_context.cpp index 16a80731d..0b2ab8558 100644 --- a/src/paimon/core/operation/scan_context.cpp +++ b/src/paimon/core/operation/scan_context.cpp @@ -67,7 +67,7 @@ class ScanContextBuilder::Impl { global_index_result_.reset(); realtime_context_.reset(); memory_pool_ = GetDefaultPool(); - executor_ = CreateDefaultExecutor(); + executor_.reset(); specific_file_system_.reset(); table_schema_ = std::nullopt; options_.clear(); @@ -84,7 +84,8 @@ class ScanContextBuilder::Impl { std::shared_ptr global_index_result_; std::shared_ptr realtime_context_; std::shared_ptr memory_pool_ = GetDefaultPool(); - std::shared_ptr executor_ = CreateDefaultExecutor(); + // Resolved in Finish(); a builder never owns an executor of its own. + std::shared_ptr executor_; std::shared_ptr specific_file_system_; std::optional table_schema_; std::map options_; @@ -178,13 +179,14 @@ Result> ScanContextBuilder::Finish() { if (impl_->path_.empty()) { return Status::Invalid("cannot scan with empty table path"); } + std::shared_ptr executor = + impl_->executor_ ? impl_->executor_ : CreateDefaultExecutor(); auto ctx = std::make_unique( impl_->path_, impl_->is_streaming_mode_, impl_->limit_, std::make_shared(impl_->predicates_, impl_->partition_filters_, impl_->bucket_filter_), - impl_->global_index_result_, impl_->realtime_context_, impl_->memory_pool_, - impl_->executor_, impl_->specific_file_system_, impl_->table_schema_, impl_->options_, - impl_->cache_); + impl_->global_index_result_, impl_->realtime_context_, impl_->memory_pool_, executor, + impl_->specific_file_system_, impl_->table_schema_, impl_->options_, impl_->cache_); impl_->Reset(); return ctx; } diff --git a/src/paimon/core/operation/scan_context_test.cpp b/src/paimon/core/operation/scan_context_test.cpp index 6c86c0498..3b9658892 100644 --- a/src/paimon/core/operation/scan_context_test.cpp +++ b/src/paimon/core/operation/scan_context_test.cpp @@ -101,4 +101,29 @@ TEST(ScanContextTest, TestSetOptionsOverridesAddedOptions) { ASSERT_EQ(expected_options, ctx->GetOptions()); } +TEST(ScanContextTest, TestDefaultExecutorIsCreatedPerContext) { + // A builder without WithExecutor() gives every context a default executor + // of its own; nothing is shared across contexts. + ScanContextBuilder first_builder("table_root_path"); + ASSERT_OK_AND_ASSIGN(auto first_ctx, first_builder.Finish()); + ScanContextBuilder second_builder("table_root_path"); + ASSERT_OK_AND_ASSIGN(auto second_ctx, second_builder.Finish()); + ASSERT_TRUE(first_ctx->GetExecutor()); + ASSERT_TRUE(second_ctx->GetExecutor()); + ASSERT_NE(first_ctx->GetExecutor(), second_ctx->GetExecutor()); + // Neither falls back to the process wide singleton. + ASSERT_NE(GetGlobalDefaultExecutor(), first_ctx->GetExecutor()); + ASSERT_NE(GetGlobalDefaultExecutor(), second_ctx->GetExecutor()); + + // Finish() resets the builder; an explicit executor set before does not + // leak into the next context built from the same builder. + std::shared_ptr executor = CreateDefaultExecutor(); + first_builder.WithExecutor(executor); + ASSERT_OK_AND_ASSIGN(auto explicit_ctx, first_builder.Finish()); + ASSERT_EQ(executor, explicit_ctx->GetExecutor()); + ASSERT_OK_AND_ASSIGN(auto reset_ctx, first_builder.Finish()); + ASSERT_TRUE(reset_ctx->GetExecutor()); + ASSERT_NE(executor, reset_ctx->GetExecutor()); +} + } // namespace paimon::test