chore: revert "feat: expire idle task pool threads after 5 seconds" (#14043)

Reverts #13123

Author: Kha

src/runtime/object.cpp

@@ -726,17 +726,16 @@ struct scoped_current_task_object : flet<lean_task_object *> {
 
 class task_manager {
     mutex                                         m_mutex;
-    unsigned                                      m_num_std_workers{0};
+    std::vector<std::unique_ptr<lthread>>         m_std_workers;
     unsigned                                      m_idle_std_workers{0};
     unsigned                                      m_max_std_workers{0};
     unsigned                                      m_num_dedicated_workers{0};
     std::deque<lean_task_object *>                m_queues[LEAN_MAX_PRIO+1];
     unsigned                                      m_queues_size{0};
     unsigned                                      m_max_prio{0};
-    condition_variable                            m_queue_cv;            // notified on work arrival or shutdown
+    condition_variable                            m_queue_cv;
     condition_variable                            m_task_finished_cv;
-    condition_variable                            m_capacity_cv;         // notified when std-worker capacity may have opened up, or shutdown
-    condition_variable                            m_worker_finished_cv;  // notified on std/dedicated worker exit or shutdown
+    condition_variable                            m_dedicated_finished_cv;
     bool                                          m_shutting_down{false};
 
     lean_task_object * dequeue() {
@@ -771,7 +770,7 @@ class task_manager {
             m_max_prio = prio;
         m_queues[prio].push_back(t);
         m_queues_size++;
-        if (!m_idle_std_workers && m_num_std_workers < m_max_std_workers)
+        if (!m_idle_std_workers && m_std_workers.size() < m_max_std_workers)
             spawn_worker();
         else
             m_queue_cv.notify_one();
@@ -795,58 +794,46 @@ class task_manager {
         lock.lock();
     }
 
-    static constexpr unsigned WORKER_IDLE_TIMEOUT_MS = 5000;
-
     void spawn_worker() {
         if (m_shutting_down)
             return;
 
-        // NOTE: always called inside lock
-        m_num_std_workers++;
-        // The `lthread` object is immediately destroyed, which detaches the thread.
-        lthread([this]() {
+        m_std_workers.emplace_back(new lthread([this]() {
             save_stack_info(false);
             unique_lock<mutex> lock(m_mutex);
             m_idle_std_workers++;
             while (true) {
                 if (m_queues_size == 0) {
                     if (m_shutting_down) {
+                        // We're done
                         break;
                     }
-                    // Wait for new tasks, with a timeout so idle threads can exit
-                    if (!m_queue_cv.wait_for(lock, chrono::milliseconds(WORKER_IDLE_TIMEOUT_MS),
-                            [&]() { return m_queues_size > 0 || m_shutting_down; })) {
-                        break;  // Exit due to timeout
-                    }
+                    // Wait for new tasks
+                    m_queue_cv.wait(lock);
                     continue;
                 }
 
                 // There's work to be done.
                 // If we have reached the maximum number of standard workers (because the
                 // maximum was decreased by `task_get`), wait for someone else to become
                 // idle before picking up new work.
-                // During shutdown we skip this throttling so remaining queued work drains
-                // promptly rather than blocking on capacity that will not free up.
+                // But during shutdown, we skip this throttling:
+                // because the finalizer might have called m_queue_cv.notify_all() for the last
+                // time, we don't want to get stuck behind the wait().
                 if (!m_shutting_down &&
-                    m_num_std_workers - m_idle_std_workers >= m_max_std_workers) {
-                    m_capacity_cv.wait(lock, [&]() {
-                        return m_shutting_down ||
-                               m_num_std_workers - m_idle_std_workers < m_max_std_workers;
-                    });
+                    m_std_workers.size() - m_idle_std_workers >= m_max_std_workers) {
+                    m_queue_cv.wait(lock);
                     continue;
                 }
 
                 lean_task_object * t = dequeue();
                 m_idle_std_workers--;
                 run_task(lock, t);
                 m_idle_std_workers++;
-                m_capacity_cv.notify_one();
                 reset_heartbeat();
             }
             m_idle_std_workers--;
-            m_num_std_workers--;
-            m_worker_finished_cv.notify_all();
-        });
+        }));
     }
 
     void spawn_dedicated_worker(lean_task_object * t) {
@@ -856,7 +843,7 @@ class task_manager {
             unique_lock<mutex> lock(m_mutex);
             run_task(lock, t);
             m_num_dedicated_workers--;
-            m_worker_finished_cv.notify_all();
+            m_dedicated_finished_cv.notify_all();
         });
         // `lthread` will be implicitly freed, which frees up its control resources but does not terminate the thread
     }
@@ -951,17 +938,16 @@ class task_manager {
         {
             unique_lock<mutex> lock(m_mutex);
             m_shutting_down = true;
+            // we can assume that `m_std_workers` will not be changed after this line
         }
         m_queue_cv.notify_all();
-        m_capacity_cv.notify_all();
 #ifndef LEAN_EMSCRIPTEN
         // wait for all workers to finish
-        {
-            unique_lock<mutex> lock(m_mutex);
-            m_worker_finished_cv.wait(lock, [&]() {
-                return m_num_std_workers == 0 && m_num_dedicated_workers == 0;
-            });
-        }
+        for (auto & t : m_std_workers)
+            t->join();
+
+        unique_lock<mutex> lock(m_mutex);
+        m_dedicated_finished_cv.wait(lock, [&]() { return m_num_dedicated_workers == 0; });
         // never seems to terminate under Emscripten
 #endif
     }
@@ -1018,7 +1004,6 @@ class task_manager {
                 spawn_worker();
             else
                 m_queue_cv.notify_one();
-            m_capacity_cv.notify_one();
         }
         m_task_finished_cv.wait(lock, [&]() { return t->m_value != nullptr; });
         if (in_pool) {