[C++][Acero] thread pool resources cannot be released when using DeclarationToStatusAsync
- Dominant language
- C++
- Stars
- 17.1k
- Forks
- 4.3k
- Avg merge
- 3d 13h
- Merged PRs (30d)
- 88
Description
### Describe the bug, including details regarding any error messages, version, and platform.
I noticed that when using the DeclarationToStatusAsync method, if the argument use_threads=false, the thread pool with a capacity of 1 created before the program exits cannot terminate properly.
```cpp
Future<> DeclarationToStatusAsync(Declaration declaration, bool use_threads,
MemoryPool* memory_pool,
FunctionRegistry* function_registry) {
QueryOptions query_options = QueryOptionsFromArgs(memory_pool, function_registry);
if (use_threads) {
return DeclarationToStatusImpl(std::move(declaration), query_options,
::arrow::internal::GetCpuThreadPool());
} else {
ARROW_ASSIGN_OR_RAISE(std::shared_ptr tpool, ThreadPool::Make(1));
return DeclarationToStatusImpl(std::move(declaration), query_options, tpool.get())
.Then([tpool]() {});
}
}
// arrow/util/thread_pool.cc
ThreadPool::~ThreadPool() {
if (shutdown_on_destroy_) {
ARROW_UNUSED(Shutdown(false /* wait */));
}
}
Status ThreadPool::Shutdown(bool wait) {
std::unique_lock lock(state_->mutex_);
if (state_->please_shutdown_) {
return Status::Invalid("Shutdown() already called");
}
state_->please_shutdown_ = true;
state_->quick_shutdown_ = !wait;
state_->cv_.notify_all();
state_->cv_shutdown_.wait(lock, [this] { // block here, task is not finished, so workers_ length is 1
return state_->workers_.empty(); });
if (!state_->quick_shutdown_) {
DCHECK_EQ(state_->pending_tasks_.size(), 0);
} else {
std::priority_queue empty;
std::swap(state_->pending_tasks_, empty);
}
CollectFinishedWorkersUnlocked();
return Status::OK();
}
```
I believe the original intention of placing tpool in the Then callback function was to keep it alive, but this also introduces another hidden issue: the destructor of the ThreadPool will also be executed within the callback function. Since this callback function itself is a task of the thread pool, it cannot complete properly (because when the thread pool calls Shutdown(), it needs to wait for state_->workers_ to be empty, which, in this case, is clearly not empty). As a result, the thread pool also fails to be destroyed after the task finishes running.
As a brief summary, I think the lifecycle of the current thread pool should not be terminated within a thread pool task.
### Component(s)
C++
Contributor guide
Assessment
This issue has not been assessed yet.