[PHI decoupling] move "thread pool" from fluid to phi (#48075)

* move threadpool

fix cmake

* fix make

paddle/fluid/framework/CMakeLists.txt

@@ -225,10 +225,6 @@ cc_test(
   SRCS reader_test.cc
   DEPS reader)
 
-cc_library(
-  threadpool
-  SRCS threadpool.cc
-  DEPS enforce)
 cc_test(
   threadpool_test
   SRCS threadpool_test.cc

paddle/fluid/framework/scope.cc

@@ -155,7 +155,7 @@ void Scope::DeleteScope(Scope* scope) const {
     if (FLAGS_benchmark || FLAGS_eager_delete_scope) {
       delete scope;
     } else {
-      Async([scope] { delete scope; });
+      phi::Async([scope] { delete scope; });
     }
   }
 }

paddle/fluid/framework/threadpool.h

@@ -14,138 +14,16 @@ limitations under the License. */
 
 #pragma once
 
-#include <condition_variable>  // NOLINT
-#include <functional>
-#include <future>  // NOLINT
-#include <memory>
-#include <mutex>  // NOLINT
-#include <queue>
-#include <thread>  // NOLINT
-#include <utility>
-#include <vector>
-
-#include "glog/logging.h"
-#include "paddle/fluid/platform/enforce.h"
-#include "paddle/fluid/platform/macros.h"  // for DISABLE_COPY_AND_ASSIGN
+#include "paddle/phi/core/threadpool.h"
 
 namespace paddle {
 namespace framework {
 
-struct ExceptionHandler {
-  mutable std::future<std::unique_ptr<platform::EnforceNotMet>> future_;
-  explicit ExceptionHandler(
-      std::future<std::unique_ptr<platform::EnforceNotMet>>&& f)
-      : future_(std::move(f)) {}
-  void operator()() const {
-    auto ex = this->future_.get();
-    if (ex != nullptr) {
-      PADDLE_THROW(platform::errors::Fatal(
-          "The exception is thrown inside the thread pool. You "
-          "should use RunAndGetException to handle the exception."
-          "The exception is:\n %s.",
-          ex->what()));
-    }
-  }
-};
-
-// ThreadPool maintains a queue of tasks, and runs them using a fixed
-// number of threads.
-class ThreadPool {
- public:
-  explicit ThreadPool(int num_threads);
-
-  using Task = std::packaged_task<std::unique_ptr<platform::EnforceNotMet>()>;
-
-  // Returns the singleton of ThreadPool.
-  static ThreadPool* GetInstance();
-
-  ~ThreadPool();
-
-  // Run pushes a function to the task queue and returns a std::future
-  // object. To wait for the completion of the task, call
-  // std::future::wait().
-  template <typename Callback>
-  std::future<void> Run(Callback fn) {
-    auto f = this->RunAndGetException(fn);
-    return std::async(std::launch::deferred, ExceptionHandler(std::move(f)));
-  }
-
-  template <typename Callback>
-  std::future<std::unique_ptr<platform::EnforceNotMet>> RunAndGetException(
-      Callback fn) {
-    Task task([fn]() -> std::unique_ptr<platform::EnforceNotMet> {
-      try {
-        fn();
-      } catch (platform::EnforceNotMet& ex) {
-        return std::unique_ptr<platform::EnforceNotMet>(
-            new platform::EnforceNotMet(ex));
-      } catch (const std::exception& e) {
-        PADDLE_THROW(platform::errors::Fatal(
-            "Unexpected exception is catched in thread pool. All "
-            "throwable exception in Paddle should be an EnforceNotMet."
-            "The exception is:\n %s.",
-            e.what()));
-      }
-      return nullptr;
-    });
-    std::future<std::unique_ptr<platform::EnforceNotMet>> f = task.get_future();
-    {
-      std::unique_lock<std::mutex> lock(mutex_);
-      if (!running_) {
-        PADDLE_THROW(platform::errors::Unavailable(
-            "Task is enqueued into stopped ThreadPool."));
-      }
-      tasks_.push(std::move(task));
-    }
-    scheduled_.notify_one();
-    return f;
-  }
-
- private:
-  DISABLE_COPY_AND_ASSIGN(ThreadPool);
-
-  // The constructor starts threads to run TaskLoop, which retrieves
-  // and runs tasks from the queue.
-  void TaskLoop();
-
-  // Init is called by GetInstance.
-  static void Init();
-
- private:
-  static std::unique_ptr<ThreadPool> threadpool_;
-  static std::once_flag init_flag_;
-
-  std::vector<std::unique_ptr<std::thread>> threads_;
-
-  std::queue<Task> tasks_;
-  std::mutex mutex_;
-  bool running_;
-  std::condition_variable scheduled_;
-};
-
-class ThreadPoolIO : ThreadPool {
- public:
-  static ThreadPool* GetInstanceIO();
-  static void InitIO();
-
- private:
-  // NOTE: threadpool in base will be inhereted here.
-  static std::unique_ptr<ThreadPool> io_threadpool_;
-  static std::once_flag io_init_flag_;
-};
+using ExceptionHandler = phi::ExceptionHandler;
 
-// Run a function asynchronously.
-// NOTE: The function must return void. If the function need to return a value,
-// you can use lambda to capture a value pointer.
-template <typename Callback>
-std::future<void> Async(Callback callback) {
-  return ThreadPool::GetInstance()->Run(callback);
-}
+using ThreadPool = phi::ThreadPool;
 
-template <typename Callback>
-std::future<void> AsyncIO(Callback callback) {
-  return ThreadPoolIO::GetInstanceIO()->Run(callback);
-}
+using ThreadPoolIO = phi::ThreadPoolIO;
 
 }  // namespace framework
 }  // namespace paddle

paddle/fluid/framework/threadpool_test.cc

@@ -26,7 +26,7 @@ void do_sum(std::vector<std::future<void>>* fs,
             int cnt) {
   for (int i = 0; i < cnt; ++i) {
     std::lock_guard<std::mutex> l(*mu);
-    fs->push_back(framework::Async([sum]() { sum->fetch_add(1); }));
+    fs->push_back(phi::Async([sum]() { sum->fetch_add(1); }));
   }
 }
 

paddle/phi/core/CMakeLists.txt

@@ -43,6 +43,10 @@ cc_library(
   lod_utils
   SRCS lod_utils.cc
   DEPS phi_enforce)
+cc_library(
+  threadpool
+  SRCS threadpool.cc
+  DEPS phi_enforce)
 
 cc_library(
   dense_tensor

paddle/fluid/framework/threadpool.cc → paddle/phi/core/threadpool.cc

-/* Copyright (c) 2016 PaddlePaddle Authors. All Rights Reserved.
-
-   Licensed 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 "paddle/fluid/framework/threadpool.h"
+// Copyright (c) 2022 PaddlePaddle Authors. All Rights Reserved.
+//
+// Licensed 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 "paddle/phi/core/threadpool.h"
 
 #include <thread>
 
 #include "gflags/gflags.h"
 #include "glog/logging.h"
-#include "paddle/fluid/platform/enforce.h"
+#include "paddle/phi/core/enforce.h"
 
+DECLARE_int32(dist_threadpool_size);
 DEFINE_int32(io_threadpool_size,
              100,
              "number of threads used for doing IO, default 100");
 
-DECLARE_int32(dist_threadpool_size);
+namespace phi {
 
-namespace paddle {
-namespace framework {
 std::unique_ptr<ThreadPool> ThreadPool::threadpool_(nullptr);
 std::once_flag ThreadPool::init_flag_;
 
@@ -47,7 +46,7 @@ void ThreadPool::Init() {
     PADDLE_ENFORCE_GT(
         num_threads,
         0,
-        platform::errors::InvalidArgument("The number of threads is 0."));
+        phi::errors::InvalidArgument("The number of threads is 0."));
     threadpool_.reset(new ThreadPool(num_threads));
   }
 }
@@ -88,8 +87,8 @@ void ThreadPool::TaskLoop() {
       }
 
       if (tasks_.empty()) {
-        PADDLE_THROW(platform::errors::Unavailable(
-            "Current thread has no task to Run."));
+        PADDLE_THROW(
+            phi::errors::Unavailable("Current thread has no task to Run."));
       }
 
       // pop a task from the task queue
@@ -115,6 +114,4 @@ void ThreadPoolIO::InitIO() {
     io_threadpool_.reset(new ThreadPool(FLAGS_io_threadpool_size));
   }
 }
-
-}  // namespace framework
-}  // namespace paddle
+}  // namespace phi

paddle/phi/core/threadpool.h

+// Copyright (c) 2022 PaddlePaddle Authors. All Rights Reserved.
+//
+// Licensed 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.
+
+#pragma once
+
+#include <condition_variable>  // NOLINT
+#include <functional>
+#include <future>  // NOLINT
+#include <memory>
+#include <mutex>  // NOLINT
+#include <queue>
+#include <thread>  // NOLINT
+#include <utility>
+#include <vector>
+
+#include "paddle/phi/core/enforce.h"
+#include "paddle/phi/core/macros.h"  // for DISABLE_COPY_AND_ASSIGN
+
+namespace phi {
+
+struct ExceptionHandler {
+  mutable std::future<std::unique_ptr<phi::enforce::EnforceNotMet>> future_;
+  explicit ExceptionHandler(
+      std::future<std::unique_ptr<phi::enforce::EnforceNotMet>>&& f)
+      : future_(std::move(f)) {}
+  void operator()() const {
+    auto ex = this->future_.get();
+    if (ex != nullptr) {
+      PADDLE_THROW(phi::errors::Fatal(
+          "The exception is thrown inside the thread pool. You "
+          "should use RunAndGetException to handle the exception."
+          "The exception is:\n %s.",
+          ex->what()));
+    }
+  }
+};
+
+// ThreadPool maintains a queue of tasks, and runs them using a fixed
+// number of threads.
+class ThreadPool {
+ public:
+  explicit ThreadPool(int num_threads);
+
+  using Task =
+      std::packaged_task<std::unique_ptr<phi::enforce::EnforceNotMet>()>;
+
+  // Returns the singleton of ThreadPool.
+  static ThreadPool* GetInstance();
+
+  ~ThreadPool();
+
+  // Run pushes a function to the task queue and returns a std::future
+  // object. To wait for the completion of the task, call
+  // std::future::wait().
+  template <typename Callback>
+  std::future<void> Run(Callback fn) {
+    auto f = this->RunAndGetException(fn);
+    return std::async(std::launch::deferred, ExceptionHandler(std::move(f)));
+  }
+
+  template <typename Callback>
+  std::future<std::unique_ptr<phi::enforce::EnforceNotMet>> RunAndGetException(
+      Callback fn) {
+    Task task([fn]() -> std::unique_ptr<phi::enforce::EnforceNotMet> {
+      try {
+        fn();
+      } catch (phi::enforce::EnforceNotMet& ex) {
+        return std::unique_ptr<phi::enforce::EnforceNotMet>(
+            new phi::enforce::EnforceNotMet(ex));
+      } catch (const std::exception& e) {
+        PADDLE_THROW(phi::errors::Fatal(
+            "Unexpected exception is catched in thread pool. All "
+            "throwable exception in Paddle should be an EnforceNotMet."
+            "The exception is:\n %s.",
+            e.what()));
+      }
+      return nullptr;
+    });
+    std::future<std::unique_ptr<phi::enforce::EnforceNotMet>> f =
+        task.get_future();
+    {
+      std::unique_lock<std::mutex> lock(mutex_);
+      if (!running_) {
+        PADDLE_THROW(phi::errors::Unavailable(
+            "Task is enqueued into stopped ThreadPool."));
+      }
+      tasks_.push(std::move(task));
+    }
+    scheduled_.notify_one();
+    return f;
+  }
+
+ private:
+  DISABLE_COPY_AND_ASSIGN(ThreadPool);
+
+  // The constructor starts threads to run TaskLoop, which retrieves
+  // and runs tasks from the queue.
+  void TaskLoop();
+
+  // Init is called by GetInstance.
+  static void Init();
+
+ private:
+  static std::unique_ptr<ThreadPool> threadpool_;
+  static std::once_flag init_flag_;
+
+  std::vector<std::unique_ptr<std::thread>> threads_;
+
+  std::queue<Task> tasks_;
+  std::mutex mutex_;
+  bool running_;
+  std::condition_variable scheduled_;
+};
+
+class ThreadPoolIO : ThreadPool {
+ public:
+  static ThreadPool* GetInstanceIO();
+  static void InitIO();
+
+ private:
+  // NOTE: threadpool in base will be inhereted here.
+  static std::unique_ptr<ThreadPool> io_threadpool_;
+  static std::once_flag io_init_flag_;
+};
+
+// Run a function asynchronously.
+// NOTE: The function must return void. If the function need to return a value,
+// you can use lambda to capture a value pointer.
+template <typename Callback>
+std::future<void> Async(Callback callback) {
+  return ThreadPool::GetInstance()->Run(callback);
+}
+
+template <typename Callback>
+std::future<void> AsyncIO(Callback callback) {
+  return ThreadPoolIO::GetInstanceIO()->Run(callback);
+}
+
+}  // namespace phi

paddle/phi/kernels/selected_rows/cpu/adam_kernel.cc

@@ -15,10 +15,10 @@
 #include "paddle/phi/kernels/selected_rows/adam_kernel.h"
 
 #include "paddle/fluid/framework/tensor_util.h"
-#include "paddle/fluid/framework/threadpool.h"
 #include "paddle/phi/backends/cpu/cpu_context.h"
 #include "paddle/phi/core/kernel_registry.h"
 #include "paddle/phi/core/tensor_utils.h"
+#include "paddle/phi/core/threadpool.h"
 #include "paddle/phi/kernels/funcs/adam_functors.h"
 #include "paddle/phi/kernels/funcs/selected_rows_functor.h"
 
@@ -201,7 +201,7 @@ void AdamDenseParamSparseGradKernel(
       if (end > static_cast<int64_t>(param_row_count)) {
         end = static_cast<int64_t>(param_row_count);
       }
-      fs.push_back(paddle::framework::Async([&functor,
+      fs.push_back(phi::Async([&functor,
                                &row_id_to_grad_row_offset,
                                &grad_data,
                                row_numel,