[vm] Let active mutator stealing mark thread pool workers as blocked instead of setting the thread pool to unlimited workers.

TEST=ci (flaky resource exhaustion)
Bug: https://github.com/dart-lang/sdk/issues/54687
Change-Id: I4ba7b6ae4d5ceb460c7db883c0733434ae76de19
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/410641
Reviewed-by: Martin Kustermann <kustermann@google.com>
Commit-Queue: Ryan Macnak <rmacnak@google.com>
Reviewed-by: Alexander Aprelev <aam@google.com>

runtime/vm/isolate.cc

@@ -94,7 +94,7 @@ DEFINE_FLAG_HANDLER(DeterministicModeHandler,
 
 DEFINE_FLAG(bool,
             disable_thread_pool_limit,
-            true,
+            false,
             "Disables the limit of the thread pool (simulates custom embedder "
             "with custom message handler on unlimited number of threads).");
 
@@ -380,10 +380,17 @@ IsolateGroup::IsolateGroup(std::shared_ptr<IsolateGroupSource> source,
 {
   FlagsCopyFrom(api_flags);
   if (!is_vm_isolate) {
-    thread_pool_.reset(
-        new MutatorThreadPool(this, FLAG_disable_thread_pool_limit
-                                        ? 0
-                                        : Scavenger::MaxMutatorThreadCount()));
+    intptr_t max_worker_threads;
+    if (FLAG_disable_thread_pool_limit) {
+      max_worker_threads = 0;
+    } else {
+      // There needs to be at least one more thread than active mutators slots
+      // so that there is a thread waiting in IncreaseMutatorCount (instead of
+      // unscheduled task sitting in the thread pool's queue) to eventually
+      // timeout and trigger StealActiveMutators.
+      max_worker_threads = Scavenger::MaxMutatorThreadCount() + 2;
+    }
+    thread_pool_.reset(new MutatorThreadPool(this, max_worker_threads));
   }
   {
     WriteRwLocker wl(ThreadState::Current(), isolate_groups_rwlock_);
@@ -587,14 +594,12 @@ void IsolateGroup::set_saved_unlinked_calls(const Array& saved_unlinked_calls) {
 
 static constexpr intptr_t kActiveMutatorPreemptionTimeout = 120;
 
-void IsolateGroup::IncreaseMutatorCount(Isolate* mutator,
-                                        bool is_nested_reenter) {
-  ASSERT(mutator->group() == this);
-
+void IsolateGroup::IncreaseMutatorCount(Thread* thread,
+                                        bool is_nested_reenter,
+                                        bool was_stolen) {
   // If the mutator was temporarily blocked on a worker thread, we have to
   // unblock the worker thread again.
-  if (is_nested_reenter) {
-    ASSERT(mutator->mutator_thread() != nullptr);
+  if (is_nested_reenter || was_stolen) {
     thread_pool()->MarkCurrentWorkerAsUnBlocked();
   }
 
@@ -610,17 +615,28 @@ void IsolateGroup::IncreaseMutatorCount(Isolate* mutator,
       waiting_mutators_++;
       bool timed_out = false;
       if (has_timeout_waiter_) {
-        ml.Wait();
+        if (was_stolen) {
+          ml.WaitWithSafepointCheck(thread);
+        } else {
+          ml.Wait();
+        }
       } else {
         has_timeout_waiter_ = true;
-        timed_out =
-            ml.Wait(kActiveMutatorPreemptionTimeout) == Monitor::kTimedOut;
+        if (was_stolen) {
+          timed_out = ml.WaitWithSafepointCheck(
+                          thread, kActiveMutatorPreemptionTimeout) ==
+                      Monitor::kTimedOut;
+        } else {
+          timed_out =
+              ml.Wait(kActiveMutatorPreemptionTimeout) == Monitor::kTimedOut;
+        }
         has_timeout_waiter_ = false;
       }
       waiting_mutators_--;
 
       if (timed_out) {
-        active_mutators_ -= thread_registry()->StealActiveMutators();
+        active_mutators_ -=
+            thread_registry()->StealActiveMutators(thread_pool());
         ASSERT(active_mutators_ >= 0);
       }
     }
@@ -636,39 +652,6 @@ void IsolateGroup::IncreaseMutatorCount(Isolate* mutator,
   }
 }
 
-void IsolateGroup::ReincreaseMutatorCount(Thread* thread) {
-  MonitorLocker ml(active_mutators_monitor_.get());
-  ASSERT(active_mutators_ <= max_active_mutators_);
-  while (active_mutators_ == max_active_mutators_) {
-    waiting_mutators_++;
-    bool timed_out = false;
-    if (has_timeout_waiter_) {
-      ml.WaitWithSafepointCheck(thread);
-    } else {
-      has_timeout_waiter_ = true;
-      timed_out =
-          ml.WaitWithSafepointCheck(thread, kActiveMutatorPreemptionTimeout) ==
-          Monitor::kTimedOut;
-      has_timeout_waiter_ = false;
-    }
-    waiting_mutators_--;
-
-    if (timed_out) {
-      active_mutators_ -= thread_registry()->StealActiveMutators();
-      ASSERT(active_mutators_ >= 0);
-    }
-  }
-  active_mutators_++;
-
-  // StealActiveMutators may cause multiple slots to become available, but
-  // does not do a NotifyAll to prevent the case of thousands of threads
-  // waking up to claim a ~dozen slots, so we keep notifying while there are
-  // both available slots and waiters.
-  if ((active_mutators_ != max_active_mutators_) && (waiting_mutators_ > 0)) {
-    ml.Notify();
-  }
-}
-
 void IsolateGroup::DecreaseMutatorCount(Isolate* mutator, bool is_nested_exit) {
   ASSERT(mutator->group() == this);
 

runtime/vm/isolate.h

@@ -540,9 +540,10 @@ class IsolateGroup : public IntrusiveDListEntry<IsolateGroup> {
     return thread == nullptr ? nullptr : thread->isolate_group();
   }
 
-  void IncreaseMutatorCount(Isolate* mutator, bool is_nested_reenter);
+  void IncreaseMutatorCount(Thread* thread,
+                            bool is_nested_reenter,
+                            bool was_stolen);
   void DecreaseMutatorCount(Isolate* mutator, bool is_nested_exit);
-  void ReincreaseMutatorCount(Thread* thread);
   NO_SANITIZE_THREAD
   intptr_t MutatorCount() const { return active_mutators_; }
 

runtime/vm/thread.cc

@@ -369,7 +369,7 @@ void Thread::EnterIsolate(Isolate* isolate) {
 
   auto group = isolate->group();
   if (!(is_nested_reenter && isolate->mutator_thread()->OwnsSafepoint())) {
-    group->IncreaseMutatorCount(isolate, is_nested_reenter);
+    group->IncreaseMutatorCount(nullptr, is_nested_reenter, false);
   }
 
   // Two threads cannot enter isolate at same time.
@@ -1366,7 +1366,8 @@ void Thread::UnwindScopes(uword stack_marker) {
 }
 
 void Thread::HandleStolen() {
-  isolate_group()->ReincreaseMutatorCount(this);
+  isolate_group()->IncreaseMutatorCount(this, /*is_nested_reenter=*/false,
+                                        /*was_stolen=*/true);
 }
 
 void Thread::EnterSafepointUsingLock() {

runtime/vm/thread_pool.cc

@@ -139,8 +139,11 @@ bool ThreadPool::CurrentThreadIsWorker() {
 }
 
 void ThreadPool::MarkCurrentWorkerAsBlocked() {
-  auto worker =
-      static_cast<Worker*>(OSThread::Current()->owning_thread_pool_worker_);
+  MarkWorkerAsBlocked(OSThread::Current());
+}
+
+void ThreadPool::MarkWorkerAsBlocked(OSThread* thread) {
+  auto worker = static_cast<Worker*>(thread->owning_thread_pool_worker_);
   Worker* new_worker = nullptr;
   if (worker != nullptr) {
     MutexLocker ml(&pool_mutex_);
@@ -152,7 +155,7 @@ void ThreadPool::MarkCurrentWorkerAsBlocked() {
       // If we have pending tasks and there are no idle workers, we will spawn a
       // new thread (temporarily allow exceeding the maximum pool size) to
       // handle the pending tasks.
-      if (idle_workers_.IsEmpty() && pending_tasks_ > 0) {
+      if (pending_tasks_ > count_idle_) {
         new_worker = new Worker(this);
         idle_workers_.Append(new_worker);
         count_idle_++;
@@ -169,6 +172,7 @@ void ThreadPool::MarkCurrentWorkerAsUnBlocked() {
       static_cast<Worker*>(OSThread::Current()->owning_thread_pool_worker_);
   if (worker != nullptr) {
     MutexLocker ml(&pool_mutex_);
+    ASSERT(worker->is_blocked_);
     if (worker->is_blocked_) {
       worker->is_blocked_ = false;
       if (max_pool_size_ > 0) {

runtime/vm/thread_pool.h

@@ -54,6 +54,7 @@ class ThreadPool {
   // Mark the current thread as being blocked (e.g. in native code). This might
   // temporarily increase the max thread pool size.
   void MarkCurrentWorkerAsBlocked();
+  void MarkWorkerAsBlocked(OSThread* thread);
 
   // Mark the current thread as being unblocked. Must be called iff
   // [MarkCurrentWorkerAsBlocked] was called before and the thread is now ready

runtime/vm/thread_registry.cc

@@ -115,13 +115,14 @@ void ThreadRegistry::FlushMarkingStacks() {
   }
 }
 
-intptr_t ThreadRegistry::StealActiveMutators() {
+intptr_t ThreadRegistry::StealActiveMutators(ThreadPool* pool) {
   MonitorLocker ml(threads_lock());
   intptr_t count = 0;
   Thread* thread = active_list_;
   while (thread != nullptr) {
     if (thread->TryStealActiveMutator()) {
       ASSERT(thread->IsDartMutatorThread());
+      pool->MarkWorkerAsBlocked(thread->os_thread());
       count++;
     }
     thread = thread->next_;

runtime/vm/thread_registry.h

@@ -38,7 +38,7 @@ class ThreadRegistry {
   void AcquireMarkingStacks();
   void ReleaseMarkingStacks();
   void FlushMarkingStacks();
-  intptr_t StealActiveMutators();
+  intptr_t StealActiveMutators(ThreadPool* pool);
 
   // Concurrent-approximate number of active isolates in the active_list
   intptr_t active_isolates_count() { return active_isolates_count_.load(); }