[SYCL] Make initialization of host task thread pool thread-safe (#4985)

Author: sergei

sycl/source/detail/config.def

@@ -35,3 +35,4 @@ CONFIG(SYCL_CACHE_MIN_DEVICE_IMAGE_SIZE, 16, __SYCL_CACHE_MIN_DEVICE_IMAGE_SIZE)
 CONFIG(SYCL_CACHE_MAX_DEVICE_IMAGE_SIZE, 16, __SYCL_CACHE_MAX_DEVICE_IMAGE_SIZE)
 CONFIG(INTEL_ENABLE_OFFLOAD_ANNOTATIONS, 1, __SYCL_INTEL_ENABLE_OFFLOAD_ANNOTATIONS)
 CONFIG(SYCL_ENABLE_DEFAULT_CONTEXTS, 1, __SYCL_ENABLE_DEFAULT_CONTEXTS)
+CONFIG(SYCL_QUEUE_THREAD_POOL_SIZE, 4, __SYCL_QUEUE_THREAD_POOL_SIZE)

sycl/source/detail/config.hpp

@@ -18,6 +18,7 @@
 #include <algorithm>
 #include <array>
 #include <cstdlib>
+#include <mutex>
 #include <string>
 #include <utility>
 
@@ -320,6 +321,39 @@ template <> class SYCLConfig<SYCL_ENABLE_DEFAULT_CONTEXTS> {
   }
 };
 
+template <> class SYCLConfig<SYCL_QUEUE_THREAD_POOL_SIZE> {
+  using BaseT = SYCLConfigBase<SYCL_QUEUE_THREAD_POOL_SIZE>;
+
+public:
+  static int get() {
+    static int Value = [] {
+      const char *ValueStr = BaseT::getRawValue();
+
+      int Result = 1;
+
+      if (ValueStr)
+        try {
+          Result = std::stoi(ValueStr);
+        } catch (...) {
+          throw invalid_parameter_error(
+              "Invalid value for SYCL_QUEUE_THREAD_POOL_SIZE environment "
+              "variable: value should be a number",
+              PI_INVALID_VALUE);
+        }
+
+      if (Result < 1)
+        throw invalid_parameter_error(
+            "Invalid value for SYCL_QUEUE_THREAD_POOL_SIZE environment "
+            "variable: value should be larger than zero",
+            PI_INVALID_VALUE);
+
+      return Result;
+    }();
+
+    return Value;
+  }
+};
+
 } // namespace detail
 } // namespace sycl
 } // __SYCL_INLINE_NAMESPACE(cl)

sycl/source/detail/global_handler.cpp

@@ -9,11 +9,13 @@
 #include <CL/sycl/detail/device_filter.hpp>
 #include <CL/sycl/detail/pi.hpp>
 #include <CL/sycl/detail/spinlock.hpp>
+#include <detail/config.hpp>
 #include <detail/global_handler.hpp>
 #include <detail/platform_impl.hpp>
 #include <detail/plugin.hpp>
 #include <detail/program_manager/program_manager.hpp>
 #include <detail/scheduler/scheduler.hpp>
+#include <detail/thread_pool.hpp>
 #include <detail/xpti_registry.hpp>
 
 #ifdef _WIN32
@@ -89,6 +91,13 @@ std::mutex &GlobalHandler::getHandlerExtendedMembersMutex() {
   return getOrCreate(MHandlerExtendedMembersMutex);
 }
 
+ThreadPool &GlobalHandler::getHostTaskThreadPool() {
+  int Size = SYCLConfig<SYCL_QUEUE_THREAD_POOL_SIZE>::get();
+  ThreadPool &TP = getOrCreate(MHostTaskThreadPool, Size);
+
+  return TP;
+}
+
 void releaseDefaultContexts() {
   // Release shared-pointers to SYCL objects.
 #ifndef _WIN32
@@ -112,6 +121,11 @@ void GlobalHandler::registerDefaultContextReleaseHandler() {
 }
 
 void shutdown() {
+  // Ensure neither host task is working so that no default context is accessed
+  // upon its release
+  if (GlobalHandler::instance().MHostTaskThreadPool.Inst)
+    GlobalHandler::instance().MHostTaskThreadPool.Inst->finishAndWait();
+
   // If default contexts are requested after the first default contexts have
   // been released there may be a new default context. These must be released
   // prior to closing the plugins.

sycl/source/detail/global_handler.hpp

@@ -25,6 +25,7 @@ class Sync;
 class plugin;
 class device_filter_list;
 class XPTIRegistry;
+class ThreadPool;
 
 using PlatformImplPtr = std::shared_ptr<platform_impl>;
 using ContextImplPtr = std::shared_ptr<context_impl>;
@@ -67,6 +68,7 @@ class GlobalHandler {
   device_filter_list &getDeviceFilterList(const std::string &InitValue);
   XPTIRegistry &getXPTIRegistry();
   std::mutex &getHandlerExtendedMembersMutex();
+  ThreadPool &getHostTaskThreadPool();
 
   static void registerDefaultContextReleaseHandler();
 
@@ -101,6 +103,8 @@ class GlobalHandler {
   InstWithLock<XPTIRegistry> MXPTIRegistry;
   // The mutex for synchronizing accesses to handlers extended members
   InstWithLock<std::mutex> MHandlerExtendedMembersMutex;
+  // Thread pool for host task and event callbacks execution
+  InstWithLock<ThreadPool> MHostTaskThreadPool;
 };
 } // namespace detail
 } // namespace sycl

sycl/source/detail/queue_impl.cpp

@@ -322,30 +322,6 @@ void queue_impl::wait(const detail::code_location &CodeLoc) {
 #endif
 }
 
-void queue_impl::initHostTaskAndEventCallbackThreadPool() {
-  if (MHostTaskThreadPool)
-    return;
-
-  int Size = 1;
-
-  if (const char *Val = std::getenv("SYCL_QUEUE_THREAD_POOL_SIZE"))
-    try {
-      Size = std::stoi(Val);
-    } catch (...) {
-      throw invalid_parameter_error(
-          "Invalid value for SYCL_QUEUE_THREAD_POOL_SIZE environment variable",
-          PI_INVALID_VALUE);
-    }
-
-  if (Size < 1)
-    throw invalid_parameter_error(
-        "Invalid value for SYCL_QUEUE_THREAD_POOL_SIZE environment variable",
-        PI_INVALID_VALUE);
-
-  MHostTaskThreadPool.reset(new ThreadPool(Size));
-  MHostTaskThreadPool->start();
-}
-
 pi_native_handle queue_impl::getNative() const {
   const detail::plugin &Plugin = getPlugin();
   if (Plugin.getBackend() == backend::opencl)

sycl/source/detail/queue_impl.hpp

@@ -24,6 +24,7 @@
 #include <detail/context_impl.hpp>
 #include <detail/device_impl.hpp>
 #include <detail/event_impl.hpp>
+#include <detail/global_handler.hpp>
 #include <detail/kernel_impl.hpp>
 #include <detail/plugin.hpp>
 #include <detail/scheduler/scheduler.hpp>
@@ -397,16 +398,7 @@ class queue_impl {
   }
 
   ThreadPool &getThreadPool() {
-    if (!MHostTaskThreadPool)
-      initHostTaskAndEventCallbackThreadPool();
-
-    return *MHostTaskThreadPool;
-  }
-
-  void stopThreadPool() {
-    if (MHostTaskThreadPool) {
-      MHostTaskThreadPool->finishAndWait();
-    }
+    return GlobalHandler::instance().getHostTaskThreadPool();
   }
 
   /// Gets the native handle of the SYCL queue.
@@ -495,8 +487,6 @@ class queue_impl {
   void instrumentationEpilog(void *TelementryEvent, std::string &Name,
                              int32_t StreamID, uint64_t IId);
 
-  void initHostTaskAndEventCallbackThreadPool();
-
   /// queue_impl.addEvent tracks events with weak pointers
   /// but some events have no other owners. addSharedEvent()
   /// follows events with a shared pointer.
@@ -535,10 +525,6 @@ class queue_impl {
   // Assume OOO support by default.
   bool MSupportOOO = true;
 
-  // Thread pool for host task and event callbacks execution.
-  // The thread pool is instantiated upon the very first call to getThreadPool()
-  std::unique_ptr<ThreadPool> MHostTaskThreadPool;
-
   // Buffer to store assert failure descriptor
   buffer<AssertHappened, 1> MAssertHappenedBuffer;
 

sycl/source/detail/scheduler/scheduler.cpp

@@ -364,7 +364,6 @@ Scheduler::Scheduler() {
 }
 
 Scheduler::~Scheduler() {
-  DefaultHostQueue->stopThreadPool();
   // By specification there are several possible sync points: buffer
   // destruction, wait() method of a queue or event. Stream doesn't introduce
   // any synchronization point. It is guaranteed that stream is flushed and

sycl/source/detail/thread_pool.hpp

@@ -51,11 +51,6 @@ class ThreadPool {
     }
   }
 
-public:
-  ThreadPool(unsigned int ThreadCount = 1) : MThreadCount(ThreadCount) {}
-
-  ~ThreadPool() { finishAndWait(); }
-
   void start() {
     MLaunchedThreads.reserve(MThreadCount);
 
@@ -65,6 +60,13 @@ class ThreadPool {
       MLaunchedThreads.emplace_back([this] { worker(); });
   }
 
+public:
+  ThreadPool(unsigned int ThreadCount = 1) : MThreadCount(ThreadCount) {
+    start();
+  }
+
+  ~ThreadPool() { finishAndWait(); }
+
   void finishAndWait() {
     MStop.store(true);