[vm/concurrency] Make an isolate group use a custom ThreadPool with maximum size

Our TLAB sizes and maximum new space size constrain the number of
parallel mutator threads we can have. Having too many mutator threads
would cause constant races between threads to acquire TLABs.

In reality we should constrain the number of threads to be at most the
number of cores, since at most that many threads can run in parallel
(i.e. at the same time).

This CL extends the TreadPool implementation to be constrained by a
maximum size. Furthermore it makes each isolate group's have it's own
pool with constrained size and schedule all group member
mutator / message handler tasks on that pool.

Issue https://github.com/dart-lang/sdk/issues/36097

Change-Id: I095c749adad827ab892f33713a32be594d7606d1
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/145382
Commit-Queue: Martin Kustermann <kustermann@google.com>
Reviewed-by: Alexander Aprelev <aam@google.com>
Reviewed-by: Ryan Macnak <rmacnak@google.com>

runtime/include/dart_api.h

@@ -844,6 +844,9 @@ DART_EXPORT DART_WARN_UNUSED_RESULT char* Dart_Initialize(
  *
  * \return NULL if cleanup is successful. Returns an error message otherwise.
  *   The caller is responsible for freeing the error message.
+ *
+ * NOTE: This function must not be called on a thread that was created by the VM
+ * itself.
  */
 DART_EXPORT DART_WARN_UNUSED_RESULT char* Dart_Cleanup();
 

runtime/lib/isolate.cc

@@ -440,7 +440,8 @@ DEFINE_NATIVE_ENTRY(Isolate_spawnFunction, 0, 10) {
       // Since this is a call to Isolate.spawn, copy the parent isolate's code.
       state->isolate_flags()->copy_parent_code = true;
 
-      Dart::thread_pool()->Run<SpawnIsolateTask>(isolate, std::move(state));
+      isolate->group()->thread_pool()->Run<SpawnIsolateTask>(isolate,
+                                                             std::move(state));
       return Object::null();
     }
   }
@@ -554,7 +555,8 @@ DEFINE_NATIVE_ENTRY(Isolate_spawnUri, 0, 12) {
   // Since this is a call to Isolate.spawnUri, don't copy the parent's code.
   state->isolate_flags()->copy_parent_code = false;
 
-  Dart::thread_pool()->Run<SpawnIsolateTask>(isolate, std::move(state));
+  isolate->group()->thread_pool()->Run<SpawnIsolateTask>(isolate,
+                                                         std::move(state));
   return Object::null();
 }
 

runtime/vm/compiler/jit/compiler.cc

@@ -1296,6 +1296,10 @@ void BackgroundCompiler::Start() {
   if (running_ || !done_) return;
   running_ = true;
   done_ = false;
+  // If we ever wanted to run the BG compiler on the
+  // `IsolateGroup::mutator_pool()` we would need to ensure the BG compiler
+  // stops when it's idle - otherwise the [MutatorThreadPool]-based idle
+  // notification would not work anymore.
   bool task_started = Dart::thread_pool()->Run<BackgroundCompilerTask>(this);
   if (!task_started) {
     running_ = false;

runtime/vm/dart.cc

@@ -559,6 +559,7 @@ char* Dart::Cleanup() {
     OS::PrintErr("[+%" Pd64 "ms] SHUTDOWN: Deleting thread pool\n",
                  UptimeMillis());
   }
+  thread_pool_->Shutdown();
   delete thread_pool_;
   thread_pool_ = NULL;
 

runtime/vm/dart_api_impl.cc

@@ -1956,7 +1956,7 @@ DART_EXPORT Dart_Handle Dart_RunLoop() {
     RunLoopData data;
     data.monitor = &monitor;
     data.done = false;
-    I->message_handler()->Run(Dart::thread_pool(), NULL, RunLoopDone,
+    I->message_handler()->Run(I->group()->thread_pool(), NULL, RunLoopDone,
                               reinterpret_cast<uword>(&data));
     while (!data.done) {
       ml.Wait();

runtime/vm/heap/scavenger.h

@@ -112,6 +112,9 @@ class ScavengeStats {
 };
 
 class Scavenger {
+ private:
+  static const intptr_t kTLABSize = 512 * KB;
+
  public:
   Scavenger(Heap* heap, intptr_t max_semi_capacity_in_words);
   ~Scavenger();
@@ -194,6 +197,17 @@ class Scavenger {
 
   bool scavenging() const { return scavenging_; }
 
+  // The maximum number of Dart mutator threads we allow to execute at the same
+  // time.
+  static intptr_t MaxMutatorThreadCount() {
+    // With a max new-space of 16 MB and 512kb TLABs we would allow up to 8
+    // mutator threads to run at the same time.
+    const intptr_t max_parallel_tlab_usage =
+        (FLAG_new_gen_semi_max_size * MB) / Scavenger::kTLABSize;
+    const intptr_t max_pool_size = max_parallel_tlab_usage / 4;
+    return max_pool_size;
+  }
+
  private:
   // Ids for time and data records in Heap::GCStats.
   enum {

runtime/vm/isolate.cc

@@ -230,10 +230,55 @@ class FinalizeWeakPersistentHandlesVisitor : public HandleVisitor {
   DISALLOW_COPY_AND_ASSIGN(FinalizeWeakPersistentHandlesVisitor);
 };
 
+void MutatorThreadPool::OnEnterIdleLocked(MonitorLocker* ml) {
+  if (FLAG_idle_timeout_micros == 0) return;
+
+  // If the isolate has not started running application code yet, we ignore the
+  // idle time.
+  if (!isolate_group_->initial_spawn_successful()) return;
+
+  int64_t idle_expiry = 0;
+  // Obtain the idle time we should wait.
+  if (isolate_group_->idle_time_handler()->ShouldNotifyIdle(&idle_expiry)) {
+    MonitorLeaveScope mls(ml);
+    NotifyIdle();
+    return;
+  }
+
+  // Wait for the recommended idle timeout.
+  // We can be woken up because of a), b) or c)
+  const auto result =
+      ml->WaitMicros(idle_expiry - OS::GetCurrentMonotonicMicros());
+
+  // a) If there are new tasks we have to run them.
+  if (TasksWaitingToRunLocked()) return;
+
+  // b) If the thread pool is shutting down we're done.
+  if (ShuttingDownLocked()) return;
+
+  // c) We timed out and should run the idle notifier.
+  if (result == Monitor::kTimedOut &&
+      isolate_group_->idle_time_handler()->ShouldNotifyIdle(&idle_expiry)) {
+    MonitorLeaveScope mls(ml);
+    NotifyIdle();
+    return;
+  }
+
+  // There must've been another thread doing active work in the meantime.
+  // If that thread becomes idle and is the last idle thread it will run this
+  // code again.
+}
+
+void MutatorThreadPool::NotifyIdle() {
+  EnterIsolateGroupScope isolate_group_scope(isolate_group_);
+  isolate_group_->idle_time_handler()->NotifyIdleUsingDefaultDeadline();
+}
+
 IsolateGroup::IsolateGroup(std::shared_ptr<IsolateGroupSource> source,
                            void* embedder_data,
                            ObjectStore* object_store)
     : embedder_data_(embedder_data),
+      thread_pool_(),
       isolates_lock_(new SafepointRwLock()),
       isolates_(),
       start_time_micros_(OS::GetCurrentMonotonicMicros()),
@@ -255,6 +300,11 @@ IsolateGroup::IsolateGroup(std::shared_ptr<IsolateGroupSource> source,
       store_buffer_(new StoreBuffer()),
       heap_(nullptr),
       saved_unlinked_calls_(Array::null()) {
+  const bool is_vm_isolate = Dart::VmIsolateNameEquals(source_->name);
+  if (!is_vm_isolate) {
+    thread_pool_.reset(
+        new MutatorThreadPool(this, Scavenger::MaxMutatorThreadCount()));
+  }
   {
     WriteRwLocker wl(ThreadState::Current(), isolate_groups_rwlock_);
     id_ = isolate_group_random_->NextUInt64();
@@ -366,6 +416,14 @@ void IsolateGroup::Shutdown() {
     }
   }
 
+  // Ensure to join all threads before starting to delete the members.
+  // (for vm-isolate we don't have a thread pool)
+  if (!Dart::VmIsolateNameEquals(source()->name)) {
+    ASSERT(thread_pool_ != nullptr);
+    thread_pool_->Shutdown();
+    thread_pool_.reset();
+  }
+
   delete this;
 }
 
@@ -2216,7 +2274,7 @@ void Isolate::SetStickyError(ErrorPtr sticky_error) {
 }
 
 void Isolate::Run() {
-  message_handler()->Run(Dart::thread_pool(), RunIsolate, ShutdownIsolate,
+  message_handler()->Run(group()->thread_pool(), RunIsolate, ShutdownIsolate,
                          reinterpret_cast<uword>(this));
 }
 
@@ -2462,7 +2520,24 @@ void Isolate::LowLevelCleanup(Isolate* isolate) {
   const bool shutdown_group =
       isolate_group->UnregisterIsolateDecrementCount(isolate);
   if (shutdown_group) {
-    isolate_group->Shutdown();
+    if (!OSThread::CurrentThreadRunsOn(isolate_group->thread_pool())) {
+      isolate_group->Shutdown();
+    } else {
+      class ShutdownGroupTask : public ThreadPool::Task {
+       public:
+        explicit ShutdownGroupTask(IsolateGroup* isolate_group)
+            : isolate_group_(isolate_group) {}
+
+        virtual void Run() { isolate_group_->Shutdown(); }
+
+       private:
+        IsolateGroup* isolate_group_;
+      };
+      // The current thread is running on the isolate group's thread pool.
+      // So we cannot safely delete the isolate group (and it's pool).
+      // Instead we will destroy the isolate group on the VM-global pool.
+      Dart::thread_pool()->Run<ShutdownGroupTask>(isolate_group);
+    }
   } else {
     if (FLAG_enable_isolate_groups) {
       // TODO(dartbug.com/36097): An isolate just died. A significant amount of

runtime/vm/isolate.h

@@ -33,6 +33,7 @@
 #include "vm/random.h"
 #include "vm/tags.h"
 #include "vm/thread.h"
+#include "vm/thread_pool.h"
 #include "vm/thread_stack_resource.h"
 #include "vm/token_position.h"
 #include "vm/virtual_memory.h"
@@ -274,6 +275,21 @@ class DisableIdleTimerScope : public ValueObject {
   IdleTimeHandler* handler_;
 };
 
+class MutatorThreadPool : public ThreadPool {
+ public:
+  MutatorThreadPool(IsolateGroup* isolate_group, intptr_t max_pool_size)
+      : ThreadPool(max_pool_size), isolate_group_(isolate_group) {}
+  virtual ~MutatorThreadPool() {}
+
+ protected:
+  virtual void OnEnterIdleLocked(MonitorLocker* ml);
+
+ private:
+  void NotifyIdle();
+
+  IsolateGroup* isolate_group_ = nullptr;
+};
+
 // Represents an isolate group and is shared among all isolates within a group.
 class IsolateGroup : public IntrusiveDListEntry<IsolateGroup> {
  public:
@@ -515,6 +531,8 @@ class IsolateGroup : public IntrusiveDListEntry<IsolateGroup> {
     reverse_pc_lookup_cache_ = table;
   }
 
+  MutatorThreadPool* thread_pool() { return thread_pool_.get(); }
+
  private:
   friend class Dart;  // For `object_store_ = ` in Dart::Init
   friend class Heap;
@@ -548,6 +566,8 @@ class IsolateGroup : public IntrusiveDListEntry<IsolateGroup> {
   bool is_vm_isolate_heap_ = false;
   void* embedder_data_ = nullptr;
 
+  IdleTimeHandler idle_time_handler_;
+  std::unique_ptr<MutatorThreadPool> thread_pool_;
   std::unique_ptr<SafepointRwLock> isolates_lock_;
   IntrusiveDList<Isolate> isolates_;
   intptr_t isolate_count_ = 0;
@@ -598,7 +618,6 @@ class IsolateGroup : public IntrusiveDListEntry<IsolateGroup> {
   ReversePcLookupCache* reverse_pc_lookup_cache_ = nullptr;
   ArrayPtr saved_unlinked_calls_;
 
-  IdleTimeHandler idle_time_handler_;
   uint32_t isolate_group_flags_ = 0;
 };
 
@@ -1592,6 +1611,26 @@ class StartIsolateScope {
   DISALLOW_COPY_AND_ASSIGN(StartIsolateScope);
 };
 
+class EnterIsolateGroupScope {
+ public:
+  explicit EnterIsolateGroupScope(IsolateGroup* isolate_group)
+      : isolate_group_(isolate_group) {
+    ASSERT(IsolateGroup::Current() == nullptr);
+    const bool result = Thread::EnterIsolateGroupAsHelper(
+        isolate_group_, Thread::kUnknownTask, /*bypass_safepoint=*/false);
+    ASSERT(result);
+  }
+
+  ~EnterIsolateGroupScope() {
+    Thread::ExitIsolateGroupAsHelper(/*bypass_safepoint=*/false);
+  }
+
+ private:
+  IsolateGroup* isolate_group_;
+
+  DISALLOW_COPY_AND_ASSIGN(EnterIsolateGroupScope);
+};
+
 class IsolateSpawnState {
  public:
   IsolateSpawnState(Dart_Port parent_port,

runtime/vm/message_handler.cc

@@ -438,18 +438,9 @@ void MessageHandler::TaskCallback() {
         ml.Enter();
       }
 
-      bool handle_messages = true;
-      while (handle_messages) {
-        handle_messages = false;
-
-        // Handle any pending messages for this message handler.
-        if (status != kShutdown) {
-          status = HandleMessages(&ml, (status == kOK), true);
-        }
-
-        if (status == kOK && HasLivePorts()) {
-          handle_messages = CheckIfIdleLocked(&ml);
-        }
+      // Handle any pending messages for this message handler.
+      if (status != kShutdown) {
+        status = HandleMessages(&ml, (status == kOK), true);
       }
     }
 
@@ -522,48 +513,6 @@ void MessageHandler::TaskCallback() {
   }
 }
 
-bool MessageHandler::CheckIfIdleLocked(MonitorLocker* ml) {
-  if (isolate() == nullptr ||
-      !isolate()->group()->idle_time_handler()->ShouldCheckForIdle()) {
-    // No idle task to schedule.
-    return false;
-  }
-  if (!isolate()->group()->initial_spawn_successful()) {
-    // The isolate has not started running application code yet.
-    return false;
-  }
-  int64_t idle_expirary = 0;
-  if (isolate()->group()->idle_time_handler()->ShouldNotifyIdle(
-          &idle_expirary)) {
-    // We've been without a message long enough to hope we can do some
-    // cleanup before the next message arrives.
-    RunIdleTaskLocked(ml);
-    // We may have received new messages while running idle task, so return
-    // true so that the handle messages loop is run again.
-    return true;
-  }
-
-  // We wait here for the scheduled idle time to expire or
-  // new messages or OOB messages to arrive.
-  paused_for_messages_ = true;
-  ml->WaitMicros(idle_expirary - OS::GetCurrentMonotonicMicros());
-  paused_for_messages_ = false;
-  // We want to loop back in order to handle the new messages
-  // or run the idle task.
-  return true;
-}
-
-void MessageHandler::RunIdleTaskLocked(MonitorLocker* ml) {
-  // Idle tasks may take a while: don't block other isolates sending
-  // us messages.
-  ml->Exit();
-  {
-    StartIsolateScope start_isolate(isolate());
-    isolate()->group()->idle_time_handler()->NotifyIdleUsingDefaultDeadline();
-  }
-  ml->Enter();
-}
-
 void MessageHandler::ClosePort(Dart_Port port) {
   MonitorLocker ml(&monitor_);
   if (FLAG_trace_isolates) {

runtime/vm/os_thread.h

@@ -39,6 +39,7 @@ class Log;
 class Mutex;
 class ThreadState;
 class TimelineEventBlock;
+class ThreadPool;
 
 class Mutex {
  public:
@@ -236,6 +237,11 @@ class OSThread : public BaseThread {
   static void DisableOSThreadCreation();
   static void EnableOSThreadCreation();
 
+  static bool CurrentThreadRunsOn(ThreadPool* pool) {
+    auto owning_pool = OSThread::Current()->owning_thread_pool_;
+    return owning_pool != nullptr && owning_pool == pool;
+  }
+
   static const intptr_t kStackSizeBufferMax = (16 * KB * kWordSize);
   static constexpr float kStackSizeBufferFraction = 0.5;
 
@@ -295,6 +301,10 @@ class OSThread : public BaseThread {
   uword stack_limit_;
   uword stack_headroom_;
   ThreadState* thread_;
+  // The ThreadPool which owns this OSThread. If this OSThread was not started
+  // by a ThreadPool it will be nullptr. This TLS value is not protected and
+  // should only be read/written by the OSThread itself.
+  ThreadPool* owning_thread_pool_ = nullptr;
 
   // thread_list_lock_ cannot have a static lifetime because the order in which
   // destructors run is undefined. At the moment this lock cannot be deleted
@@ -313,6 +323,7 @@ class OSThread : public BaseThread {
   friend class OSThreadIterator;
   friend class ThreadInterrupterWin;
   friend class ThreadInterrupterFuchsia;
+  friend class ThreadPool;  // to access owning_thread_pool_
 };
 
 // Note that this takes the thread list lock, prohibiting threads from coming

runtime/vm/thread_pool.cc

@@ -34,10 +34,28 @@ static int64_t ComputeTimeout(int64_t idle_start) {
   }
 }
 
-ThreadPool::ThreadPool() : all_workers_dead_(false) {}
+ThreadPool::ThreadPool(uintptr_t max_pool_size)
+    : all_workers_dead_(false), max_pool_size_(max_pool_size) {}
 
 ThreadPool::~ThreadPool() {
-  TriggerShutdown();
+  Shutdown();
+}
+
+void ThreadPool::Shutdown() {
+  {
+    MonitorLocker ml(&pool_monitor_);
+
+    // Prevent scheduling of new tasks.
+    shutting_down_ = true;
+
+    if (running_workers_.IsEmpty() && idle_workers_.IsEmpty()) {
+      // All workers have already died.
+      all_workers_dead_ = true;
+    } else {
+      // Tell workers to drain remaining work and then shut down.
+      ml.NotifyAll();
+    }
+  }
 
   // Wait until all workers are dead. Any new death will notify the exit
   // monitor.
@@ -47,14 +65,20 @@ ThreadPool::~ThreadPool() {
       eml.Wait();
     }
   }
+  ASSERT(count_idle_ == 0);
+  ASSERT(count_running_ == 0);
+  ASSERT(idle_workers_.IsEmpty());
+  ASSERT(running_workers_.IsEmpty());
 
-  // Join all dead workers.
   WorkerList dead_workers_to_join;
   {
     MonitorLocker ml(&pool_monitor_);
     ObtainDeadWorkersLocked(&dead_workers_to_join);
   }
   JoinDeadWorkersLocked(&dead_workers_to_join);
+
+  ASSERT(count_dead_ == 0);
+  ASSERT(dead_workers_.IsEmpty());
 }
 
 bool ThreadPool::RunImpl(std::unique_ptr<Task> task) {
@@ -91,6 +115,14 @@ void ThreadPool::WorkerLoop(Worker* worker) {
       RunningToIdleLocked(worker);
     }
 
+    if (running_workers_.IsEmpty()) {
+      ASSERT(tasks_.IsEmpty());
+      OnEnterIdleLocked(&ml);
+      if (!tasks_.IsEmpty()) {
+        continue;
+      }
+    }
+
     if (shutting_down_) {
       ObtainDeadWorkersLocked(&dead_workers_to_join);
       IdleToDeadLocked(worker);
@@ -125,21 +157,6 @@ void ThreadPool::WorkerLoop(Worker* worker) {
   JoinDeadWorkersLocked(&dead_workers_to_join);
 }
 
-void ThreadPool::TriggerShutdown() {
-  MonitorLocker ml(&pool_monitor_);
-
-  // Prevent scheduling of new tasks.
-  shutting_down_ = true;
-
-  if (running_workers_.IsEmpty() && idle_workers_.IsEmpty()) {
-    // All workers have already died.
-    all_workers_dead_ = true;
-  } else {
-    // Tell workers to drain remaining work and then shut down.
-    ml.NotifyAll();
-  }
-}
-
 void ThreadPool::IdleToRunningLocked(Worker* worker) {
   ASSERT(idle_workers_.ContainsForDebugging(worker));
   idle_workers_.Remove(worker);
@@ -209,6 +226,15 @@ ThreadPool::Worker* ThreadPool::ScheduleTaskLocked(MonitorLocker* ml,
     return nullptr;
   }
 
+  // If we have maxed out the number of threads running, we will not start a
+  // new one.
+  if (max_pool_size_ > 0 && (count_idle_ + count_running_) >= max_pool_size_) {
+    if (!idle_workers_.IsEmpty()) {
+      ml->Notify();
+    }
+    return nullptr;
+  }
+
   // Otherwise start a new worker.
   auto new_worker = new Worker(this);
   idle_workers_.Append(new_worker);
@@ -234,6 +260,8 @@ void ThreadPool::Worker::Main(uword args) {
   Worker* worker = reinterpret_cast<Worker*>(args);
   ThreadPool* pool = worker->pool_;
 
+  os_thread->owning_thread_pool_ = pool;
+
   // Once the worker quits it needs to be joined.
   worker->join_id_ = OSThread::GetCurrentThreadJoinId(os_thread);
 
@@ -246,6 +274,8 @@ void ThreadPool::Worker::Main(uword args) {
 
   pool->WorkerLoop(worker);
 
+  os_thread->owning_thread_pool_ = nullptr;
+
   // Call the thread exit hook here to notify the embedder that the
   // thread pool thread is exiting.
   if (Dart::thread_exit_callback() != NULL) {

runtime/vm/thread_pool.h

@@ -34,7 +34,10 @@ class ThreadPool {
     DISALLOW_COPY_AND_ASSIGN(Task);
   };
 
-  ThreadPool();
+  explicit ThreadPool(uintptr_t max_pool_size = 0);
+
+  // Prevent scheduling of new tasks, wait until all pending tasks are done
+  // and join worker threads.
   virtual ~ThreadPool();
 
   // Runs a task on the thread pool.
@@ -43,8 +46,8 @@ class ThreadPool {
     return RunImpl(std::unique_ptr<Task>(new T(std::forward<Args>(args)...)));
   }
 
-  // Trigger shutdown, prevents scheduling of new tasks.
-  void TriggerShutdown();
+  // Triggers shutdown, prevents scheduling of new tasks.
+  void Shutdown();
 
   // Exposed for unit test in thread_pool_test.cc
   uint64_t workers_started() const { return count_idle_ + count_running_; }
@@ -74,6 +77,19 @@ class ThreadPool {
     DISALLOW_COPY_AND_ASSIGN(Worker);
   };
 
+ protected:
+  // Called when the thread pool turns idle.
+  //
+  // Subclasses can override this to perform some action.
+  // NOTE: While this function is running the thread pool will be locked.
+  virtual void OnEnterIdleLocked(MonitorLocker* ml) {}
+
+  // Whether a shutdown was requested.
+  bool ShuttingDownLocked() { return shutting_down_; }
+
+  // Whether new tasks are ready to be run.
+  bool TasksWaitingToRunLocked() { return !tasks_.IsEmpty(); }
+
  private:
   using TaskList = IntrusiveDList<Task>;
   using WorkerList = IntrusiveDList<Worker>;
@@ -103,6 +119,8 @@ class ThreadPool {
   Monitor exit_monitor_;
   std::atomic<bool> all_workers_dead_;
 
+  uintptr_t max_pool_size_ = 0;
+
   DISALLOW_COPY_AND_ASSIGN(ThreadPool);
 };