Implement a simple threadpool #6684

Yancey1989 · 2017-12-18T01:54:48Z

Maybe the simple threadpool would be used in parallel_do Op, such as:

ParallelDo::Run(...) {
  BlockingCounter bc(4);  // means parallel execute the block with 4 threads
  tp = ThreadPoool::Instance();
  for (int i = 0; i < 4; ++i){
    tp.Run([](){
      executor::Run(block);
      bc.Desc();
    })
  }
  bc.Wait(); // wait for the blocking counter become 0.
}

reyoung

I think a simple thread pool is cool.
Maybe @tonyyang-svail need to pay attention to this PR and it may help to implement the Parallel.for operator

typhoonzero · 2017-12-18T03:11:57Z

paddle/framework/CMakeLists.txt

@@ -58,3 +58,5 @@ cc_test(var_type_inference_test SRCS var_type_inference_test.cc DEPS op_registry
        proto_desc)
 cc_library(selected_rows SRCS selected_rows.cc DEPS tensor)
 cc_test(selected_rows_test SRCS selected_rows_test.cc DEPS selected_rows)
+
+cc_test(threadpool_test SRCS threadpool_test.cc )


The space before trailing ) seems not needed.

typhoonzero · 2017-12-18T03:21:21Z

paddle/framework/threadpool.h

+    condition_.notify_one();
+  }
+
+  // wait unitle all the function are completed


Typo "unitle".

typhoonzero · 2017-12-18T03:22:18Z

paddle/framework/threadpool.h

+  std::queue<Func> tasks_;
+  std::vector<std::unique_ptr<std::thread>> threads_;
+  std::mutex mutex_;
+  std::condition_variable condition_;


condition_ doesn't mean anything, should have a proper name.

condition_variable must use with a mutex as a pair, mutex_ is used every where, and don't know which variable it's protecting.

condition_ doesn't mean anything, should have a proper name

condition_variable must use with a mutex as a pair, mutex_ is used every where, and don't know which variable it's protecting

the mutex just protects the task queue, maybe we only need one mutex for now?

typhoonzero · 2017-12-18T03:28:00Z

@reyoung @tonyyang-svail The ThreadPool is also needed for multi-GPU, each GPU thread needs a corresponding CPU thread to launch its CUDA kernel.

gongweibao · 2017-12-18T03:43:39Z

I think also:

I think a simple thread pool is cool

Do we need to investigate BRPC threading pool?It's a similar implement with GO scheduler:
https://github.com/brpc/brpc/blob/master/docs/cn/threading_overview.md#%E5%A4%9A%E7%BA%BF%E7%A8%8Breactor
https://github.com/brpc/brpc/blob/master/docs/cn/bthread.md
http://morsmachine.dk/go-scheduler

gongweibao · 2017-12-18T05:13:51Z

paddle/framework/threadpool.h

+
+  // push a function to the queue, and will be scheduled and
+  // executed if a thread is available
+  void Start(const Func& fn) {


Start=>Run?

…threadpool

helinwang · 2017-12-21T01:21:31Z

paddle/framework/threadpool.h

+   *            If num_threads <= 0, the thread pool wil be initilized
+   *            with the number of concurrent threads supported.
+   */
+  static ThreadPool* Instance(size_t num_threads) {


Maybe just bake in num_threads (not as argument, std::thread::hardware_concurrency is probably a good choice. But it could return 0, so need to check for 0), otherwise the first time Instance(10) is called, latter Instance(20) is called, but returns one threadpool with 10 threads rather than 20.

The user may specify the max CPU cores which will be used, such as GOMAXPROCS in Go, but maybe I can remove the argument num_threads and support MAXPROCS for future.

helinwang · 2017-12-21T01:26:19Z

paddle/framework/threadpool.h

+   */
+  static ThreadPool* Instance(size_t num_threads) {
+    static std::unique_ptr<ThreadPool> threadpool;
+    if (threadpool.get() == nullptr) {


I believe here is a race condition (as well as GetNumThreads and GetAvailable). We can either use a mutex to protect it, or do not do lazy initialization. If it's initialized when the program starts, it will happen before all concurrent access, thus thread safe.

The num_threads_ will be initialized for the first call GetInstance(), and never changed, so maybe it's not a race condition, and I add a lock in functionGetAvailabel(), thanks :)

I think if Instance is not called before, and called concurrently, it will be a race condition. Maybe try this: https://stackoverflow.com/a/19907903/852385

Thanks, I think call_once is a good choice, and use a enhancement version in https://github.com/PaddlePaddle/Paddle/blob/develop/paddle/platform/call_once.h

helinwang · 2017-12-21T01:43:27Z

paddle/framework/threadpool.h

+        std::unique_lock<std::mutex> lock(mutex_);
+        ++available_;
+        if (tasks_.empty() && available_ == num_threads_) {
+          done_ = true;


How do you think about removing done_ and create a function done() which returns tasks_.empty() && available_ == num_threads_?

typhoonzero · 2017-12-21T15:18:50Z