create threads out of the lock

src/Common/ThreadPool.cpp

@@ -7,6 +7,7 @@
 #include <Common/noexcept_scope.h>
 
 #include <cassert>
+#include <future>
 #include <type_traits>
 
 #include <Poco/Util/Application.h>
@@ -225,27 +226,37 @@ ReturnType ThreadPoolImpl<Thread>::scheduleImpl(Job job, Priority priority, std:
         {
             try
             {
-                threads.emplace_front();
-            }
-            catch (...)
-            {
-                /// Most likely this is a std::bad_alloc exception
-                return on_error("cannot allocate thread slot");
-            }
+                std::promise<typename std::list<Thread>::iterator> promise_thread_it;
+                std::shared_future<typename std::list<Thread>::iterator> future_thread_id = promise_thread_it.get_future().share();
+                lock.unlock();
 
-            try
-            {
                 Stopwatch watch2;
-                threads.front() = Thread([this, it = threads.begin()] { worker(it); });
+                ///  we use future here for 2 reasons:
+                ///  1) just passing a ref to list iterator after adding the thread to the list
+                ///  2) hold the thread work until the thread is added to the list, otherwise
+                ///     is can get the lock faster and then can wait for a cond_variable forever
+                auto thread = Thread([this, ft = std::move(future_thread_id)] mutable
+                {
+                    auto thread_it = ft.get();
+                    worker(thread_it);
+                });
                 ProfileEvents::increment(
                     std::is_same_v<Thread, std::thread> ? ProfileEvents::GlobalThreadPoolThreadCreationMicroseconds : ProfileEvents::LocalThreadPoolThreadCreationMicroseconds,
                     watch2.elapsedMicroseconds());
+                lock.lock();
                 ProfileEvents::increment(
                     std::is_same_v<Thread, std::thread> ? ProfileEvents::GlobalThreadPoolExpansions : ProfileEvents::LocalThreadPoolExpansions);
+                threads.push_front(std::move(thread));
+                promise_thread_it.set_value(threads.begin());
+            }
+            catch (const std::exception & e)
+            {
+                lock.lock();
+                return on_error(fmt::format("cannot start thread: {}", e.what()));
             }
             catch (...)
             {
-                threads.pop_front();
+                lock.lock();
                 return on_error("cannot allocate thread");
             }
         }