Add thread pool as function parameters to C++ API #876
Conversation
kingcrimsontianyu
added
breaking
kingcrimsontianyu
changed the title
|
/ok to test afb670d |
|
/ok to test 19f8af9 |
madsbk
left a comment
madsbk
left a comment
There was a problem hiding this comment.
Looks good, but I think it would be useful to add some tests that mix different thread pools.
| std::size_t offset = 0, | ||
| std::size_t task_size = defaults::task_size()); | ||
| std::size_t task_size = defaults::task_size(), | ||
| ThreadPool* thread_pool = &defaults::thread_pool()); |
There was a problem hiding this comment.
Consider using std::shared_ptr<ThreadPool> throughout to encourage easier and safer lifetime management for users.
There was a problem hiding this comment.
Thanks. Thinking over passing thread pool as a shared pointer, when the asynchronous function pread/pwrite returns, the shared pointer is destroyed. So in order to properly extend the pool's lifetime for the async operation and prevent use-after-free, we need to further share its ownership with the I/O task, either each task or the last aggregate task. The pro is no concern over thread pool lifetime at the point the std::future 's result is being waited for. The con is the slight increase in implementation complexity and runtime overhead.
If we pass a raw pointer instead, we claim no ownership responsibility and require users to maintain the pool's lifetime throughout the I/O operations. The pro is simplicity, and the con is loss of bonus of smart pointers.
There was a problem hiding this comment.
There appears to be a very tricky problem.
To extend the lifetime of the thread pool properly during the asynchronous operations, we need to capture std::shared_ptr<ThreadPool> in the last task:
https://github.com/rapidsai/kvikio/blob/main/cpp/include/kvikio/detail/parallel_operation.hpp#L166
auto last_task = [=, thread_pool = thread_pool, tasks = std::move(tasks)]() mutable -> std::size_t {Suppose reference count is exactly 1 when the last task is being executed. When it is done, the task goes out of scope precisely at https://github.com/bshoshany/thread-pool/blob/v4.1.0/include/BS_thread_pool.hpp#L938, and the reference count will reach 0 and the pool start being destroyed. In the destructor, we wait (sleep) (https://github.com/bshoshany/thread-pool/blob/v4.1.0/include/BS_thread_pool.hpp#L336) for the condition that tasks_running == 0, which will not happen because --tasks_running takes place at the beginning of the worker's loop (https://github.com/bshoshany/thread-pool/blob/v4.1.0/include/BS_thread_pool.hpp#L915). So tasks_running will always be 1 and we are waiting forever in the destructor. Strangely, I haven't seen this in my unit test, but I fear that the deadlock may appear in production.
There was a problem hiding this comment.
So I think if we do want to extend the lifetime of the thread pool, we need to add it directly to the returned future's results ("shared state" in C++ terminology), i.e. instead of std::future<std::size_t> we probably need std::future<std::pair<std::size_t, std::shared_ptr<ThreadPool>>>.
There was a problem hiding this comment.
At this point, I'm inclined to go back to the raw pointer approach, and ask users to shoulder the responsibility of lifetime management for the thread pool. 🤔
|
/ok to test bbf240b |
|
/merge |
2 participants
This PR introduces two changes in KvikIO C++ API:
pread(andpwriteif applicable) method inFileHandle,RemoteHandleandMmapHandle's , this PR adds the thread pool as a function parameter. By default, the global thread pool is used.thread_pool_wrapperclass merely forwards calls without adding useful functionality. This PR removes it and adds a simple type aliasThreadPoolfor the underlyingBS:thread_pool.This PR is a dependency of #874 which facilitates investigation into a multi-drive scaling problem reported by #850.