diff --git a/runtime/vm/isolate.cc b/runtime/vm/isolate.cc index 45dbeb6f35f..11cb004f448 100644 --- a/runtime/vm/isolate.cc +++ b/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 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); diff --git a/runtime/vm/isolate.h b/runtime/vm/isolate.h index ba7ac135b28..9f5d27fb404 100644 --- a/runtime/vm/isolate.h +++ b/runtime/vm/isolate.h @@ -540,9 +540,10 @@ class IsolateGroup : public IntrusiveDListEntry { 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_; } diff --git a/runtime/vm/thread.cc b/runtime/vm/thread.cc index 5b37119122e..a323c2378fb 100644 --- a/runtime/vm/thread.cc +++ b/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() { diff --git a/runtime/vm/thread_pool.cc b/runtime/vm/thread_pool.cc index d8536103919..48b75f8fe9e 100644 --- a/runtime/vm/thread_pool.cc +++ b/runtime/vm/thread_pool.cc @@ -139,8 +139,11 @@ bool ThreadPool::CurrentThreadIsWorker() { } void ThreadPool::MarkCurrentWorkerAsBlocked() { - auto worker = - static_cast(OSThread::Current()->owning_thread_pool_worker_); + MarkWorkerAsBlocked(OSThread::Current()); +} + +void ThreadPool::MarkWorkerAsBlocked(OSThread* thread) { + auto worker = static_cast(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(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) { diff --git a/runtime/vm/thread_pool.h b/runtime/vm/thread_pool.h index 99fee30c10e..db8d684e6f0 100644 --- a/runtime/vm/thread_pool.h +++ b/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 diff --git a/runtime/vm/thread_registry.cc b/runtime/vm/thread_registry.cc index 9b85b19484d..b4b09f79746 100644 --- a/runtime/vm/thread_registry.cc +++ b/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_; diff --git a/runtime/vm/thread_registry.h b/runtime/vm/thread_registry.h index 813afd30ce4..db54f944881 100644 --- a/runtime/vm/thread_registry.h +++ b/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(); }