diff --git a/runtime/include/dart_api.h b/runtime/include/dart_api.h index e5ad04f59e0..51bb6169bde 100644 --- a/runtime/include/dart_api.h +++ b/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(); diff --git a/runtime/lib/isolate.cc b/runtime/lib/isolate.cc index 7833f4a248e..a723fbc664c 100644 --- a/runtime/lib/isolate.cc +++ b/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(isolate, std::move(state)); + isolate->group()->thread_pool()->Run(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(isolate, std::move(state)); + isolate->group()->thread_pool()->Run(isolate, + std::move(state)); return Object::null(); } diff --git a/runtime/vm/compiler/jit/compiler.cc b/runtime/vm/compiler/jit/compiler.cc index 1364dd0579a..befaa2a14f2 100644 --- a/runtime/vm/compiler/jit/compiler.cc +++ b/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(this); if (!task_started) { running_ = false; diff --git a/runtime/vm/dart.cc b/runtime/vm/dart.cc index 08ae0763c82..7edb36e0c79 100644 --- a/runtime/vm/dart.cc +++ b/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; diff --git a/runtime/vm/dart_api_impl.cc b/runtime/vm/dart_api_impl.cc index 91d3d636a5e..d8e352e8d83 100644 --- a/runtime/vm/dart_api_impl.cc +++ b/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(&data)); while (!data.done) { ml.Wait(); diff --git a/runtime/vm/heap/scavenger.h b/runtime/vm/heap/scavenger.h index 56baac3ff20..29469776b3f 100644 --- a/runtime/vm/heap/scavenger.h +++ b/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 { diff --git a/runtime/vm/isolate.cc b/runtime/vm/isolate.cc index ccca676c6f0..c31b429a4e7 100644 --- a/runtime/vm/isolate.cc +++ b/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 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 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(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(isolate_group); + } } else { if (FLAG_enable_isolate_groups) { // TODO(dartbug.com/36097): An isolate just died. A significant amount of diff --git a/runtime/vm/isolate.h b/runtime/vm/isolate.h index 04870399e4b..73455aa7773 100644 --- a/runtime/vm/isolate.h +++ b/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 { public: @@ -515,6 +531,8 @@ class IsolateGroup : public IntrusiveDListEntry { 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 { bool is_vm_isolate_heap_ = false; void* embedder_data_ = nullptr; + IdleTimeHandler idle_time_handler_; + std::unique_ptr thread_pool_; std::unique_ptr isolates_lock_; IntrusiveDList isolates_; intptr_t isolate_count_ = 0; @@ -598,7 +618,6 @@ class IsolateGroup : public IntrusiveDListEntry { 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, diff --git a/runtime/vm/message_handler.cc b/runtime/vm/message_handler.cc index 10126303004..28a89a2e660 100644 --- a/runtime/vm/message_handler.cc +++ b/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) { diff --git a/runtime/vm/os_thread.h b/runtime/vm/os_thread.h index 849d47217e5..193a5f8fe9a 100644 --- a/runtime/vm/os_thread.h +++ b/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 diff --git a/runtime/vm/thread_pool.cc b/runtime/vm/thread_pool.cc index d307f5f5bef..0199a1b2bdb 100644 --- a/runtime/vm/thread_pool.cc +++ b/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) { @@ -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(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) { diff --git a/runtime/vm/thread_pool.h b/runtime/vm/thread_pool.h index 08fc19de042..c0c532cd2a4 100644 --- a/runtime/vm/thread_pool.h +++ b/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(new T(std::forward(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; using WorkerList = IntrusiveDList; @@ -103,6 +119,8 @@ class ThreadPool { Monitor exit_monitor_; std::atomic all_workers_dead_; + uintptr_t max_pool_size_ = 0; + DISALLOW_COPY_AND_ASSIGN(ThreadPool); };