diff --git a/runtime/vm/heap/compactor.cc b/runtime/vm/heap/compactor.cc index 2ea66661ad5..5775b4af132 100644 --- a/runtime/vm/heap/compactor.cc +++ b/runtime/vm/heap/compactor.cc @@ -122,7 +122,7 @@ struct Partition { Page* tail; }; -class CompactorTask : public ThreadPool::Task { +class CompactorTask : public SafepointTask { public: CompactorTask(IsolateGroup* isolate_group, GCCompactor* compactor, @@ -147,11 +147,14 @@ class CompactorTask : public ThreadPool::Task { free_page_(nullptr), free_current_(0), free_end_(0) {} + ~CompactorTask() { barrier_->Release(); } - void Run(); - void RunEnteredIsolateGroup(); + void Run() override; + void RunBlockedAtSafepoint() override; + void RunMain() override; private: + void RunEnteredIsolateGroup(); void PlanPage(Page* page); void SlidePage(Page* page); uword PlanBlock(uword first_object, ForwardingPage* forwarding_page); @@ -292,24 +295,14 @@ void GCCompactor::Compact(Page* pages, FreeList* freelist, Mutex* pages_lock) { RelaxedAtomic next_sliding_task = {0}; RelaxedAtomic next_forwarding_task = {0}; - for (intptr_t task_index = 0; task_index < num_tasks; task_index++) { - if (task_index < (num_tasks - 1)) { - // Begin compacting on a helper thread. - Dart::thread_pool()->Run( - thread()->isolate_group(), this, barrier, &next_planning_task, - &next_setup_task, &next_sliding_task, &next_forwarding_task, - num_tasks, partitions, freelist); - } else { - // Last worker is the main thread. - CompactorTask task(thread()->isolate_group(), this, barrier, - &next_planning_task, &next_setup_task, - &next_sliding_task, &next_forwarding_task, num_tasks, - partitions, freelist); - task.RunEnteredIsolateGroup(); - barrier->Sync(); - barrier->Release(); - } + IntrusiveDList tasks; + for (intptr_t i = 0; i < num_tasks; i++) { + tasks.Append(new CompactorTask(thread()->isolate_group(), this, barrier, + &next_planning_task, &next_setup_task, + &next_sliding_task, &next_forwarding_task, + num_tasks, partitions, freelist)); } + thread()->isolate_group()->safepoint_handler()->RunTasks(&tasks); } // Update inner pointers in typed data views (needs to be done after all @@ -399,7 +392,6 @@ void GCCompactor::Compact(Page* pages, FreeList* freelist, Mutex* pages_lock) { void CompactorTask::Run() { if (!barrier_->TryEnter()) { - barrier_->Release(); return; } @@ -414,7 +406,28 @@ void CompactorTask::Run() { // This task is done. Notify the original thread. barrier_->Sync(); - barrier_->Release(); +} + +void CompactorTask::RunBlockedAtSafepoint() { + if (!barrier_->TryEnter()) { + return; + } + + Thread* thread = Thread::Current(); + Thread::TaskKind saved_task_kind = thread->task_kind(); + thread->set_task_kind(Thread::kCompactorTask); + + RunEnteredIsolateGroup(); + + thread->set_task_kind(saved_task_kind); + + barrier_->Sync(); +} + +void CompactorTask::RunMain() { + RunEnteredIsolateGroup(); + + barrier_->Sync(); } void CompactorTask::RunEnteredIsolateGroup() { diff --git a/runtime/vm/heap/gc_shared.h b/runtime/vm/heap/gc_shared.h index 84964e8eeba..4ef57b28f2d 100644 --- a/runtime/vm/heap/gc_shared.h +++ b/runtime/vm/heap/gc_shared.h @@ -215,7 +215,7 @@ void MournFinalizerEntry(GCVisitorType* visitor, // workers are not, but they bypass safepoint because the main // worker is at a safepoint already. ASSERT(Thread::Current()->OwnsGCSafepoint() || - Thread::Current()->BypassSafepoints()); + (Thread::Current()->task_kind() == Thread::kScavengerTask)); if (finalizer.IsNativeFinalizer()) { NativeFinalizerPtr native_finalizer = diff --git a/runtime/vm/heap/incremental_compactor.cc b/runtime/vm/heap/incremental_compactor.cc index c93d656e9a3..f871844af04 100644 --- a/runtime/vm/heap/incremental_compactor.cc +++ b/runtime/vm/heap/incremental_compactor.cc @@ -99,7 +99,7 @@ struct PrologueState { intptr_t freelist_limit; }; -class PrologueTask : public ThreadPool::Task { +class PrologueTask : public SafepointTask { public: PrologueTask(ThreadBarrier* barrier, IsolateGroup* isolate_group, @@ -109,10 +109,10 @@ class PrologueTask : public ThreadPool::Task { isolate_group_(isolate_group), old_space_(old_space), state_(state) {} + ~PrologueTask() { barrier_->Release(); } - void Run() { + void Run() override { if (!barrier_->TryEnter()) { - barrier_->Release(); return; } @@ -126,7 +126,28 @@ class PrologueTask : public ThreadPool::Task { Thread::ExitIsolateGroupAsHelper(/*bypass_safepoint=*/true); barrier_->Sync(); - barrier_->Release(); + } + + void RunBlockedAtSafepoint() override { + if (!barrier_->TryEnter()) { + return; + } + + Thread* thread = Thread::Current(); + Thread::TaskKind saved_task_kind = thread->task_kind(); + thread->set_task_kind(Thread::kIncrementalCompactorTask); + + RunEnteredIsolateGroup(); + + thread->set_task_kind(saved_task_kind); + + barrier_->Sync(); + } + + void RunMain() override { + RunEnteredIsolateGroup(); + + barrier_->Sync(); } void RunEnteredIsolateGroup() { @@ -257,20 +278,11 @@ bool GCIncrementalCompactor::SelectEvacuationCandidates(PageSpace* old_space) { isolate_group->heap()->new_space()->NumScavengeWorkers(); RELEASE_ASSERT(num_tasks > 0); ThreadBarrier* barrier = new ThreadBarrier(num_tasks, 1); + IntrusiveDList tasks; for (intptr_t i = 0; i < num_tasks; i++) { - if (i < (num_tasks - 1)) { - // Begin compacting on a helper thread. - bool result = Dart::thread_pool()->Run( - barrier, isolate_group, old_space, &state); - ASSERT(result); - } else { - // Last worker is the main thread. - PrologueTask task(barrier, isolate_group, old_space, &state); - task.RunEnteredIsolateGroup(); - barrier->Sync(); - barrier->Release(); - } + tasks.Append(new PrologueTask(barrier, isolate_group, old_space, &state)); } + isolate_group->safepoint_handler()->RunTasks(&tasks); for (intptr_t i = PageSpace::kDataFreelist, n = old_space->num_freelists_; i < n; i++) { @@ -627,7 +639,7 @@ class EpilogueState { RelaxedAtomic new_free_size_ = {0}; }; -class EpilogueTask : public ThreadPool::Task { +class EpilogueTask : public SafepointTask { public: EpilogueTask(ThreadBarrier* barrier, IsolateGroup* isolate_group, @@ -639,8 +651,9 @@ class EpilogueTask : public ThreadPool::Task { old_space_(old_space), freelist_(freelist), state_(state) {} + ~EpilogueTask() { barrier_->Release(); } - void Run() { + void Run() override { bool result = Thread::EnterIsolateGroupAsHelper( isolate_group_, Thread::kIncrementalCompactorTask, /*bypass_safepoint=*/true); @@ -651,7 +664,24 @@ class EpilogueTask : public ThreadPool::Task { Thread::ExitIsolateGroupAsHelper(/*bypass_safepoint=*/true); barrier_->Sync(); - barrier_->Release(); + } + + void RunBlockedAtSafepoint() override { + Thread* thread = Thread::Current(); + Thread::TaskKind saved_task_kind = thread->task_kind(); + thread->set_task_kind(Thread::kIncrementalCompactorTask); + + RunEnteredIsolateGroup(); + + thread->set_task_kind(saved_task_kind); + + barrier_->Sync(); + } + + void RunMain() override { + RunEnteredIsolateGroup(); + + barrier_->Sync(); } void RunEnteredIsolateGroup() { @@ -882,21 +912,12 @@ void GCIncrementalCompactor::Evacuate(PageSpace* old_space) { isolate_group->heap()->new_space()->NumScavengeWorkers(); RELEASE_ASSERT(num_tasks > 0); ThreadBarrier* barrier = new ThreadBarrier(num_tasks, num_tasks); + IntrusiveDList tasks; for (intptr_t i = 0; i < num_tasks; i++) { - // Begin compacting on a helper thread. - FreeList* freelist = old_space->DataFreeList(i); - if (i < (num_tasks - 1)) { - bool result = Dart::thread_pool()->Run( - barrier, isolate_group, old_space, freelist, &state); - ASSERT(result); - } else { - // Last worker is the main thread. - EpilogueTask task(barrier, isolate_group, old_space, freelist, &state); - task.RunEnteredIsolateGroup(); - barrier->Sync(); - barrier->Release(); - } + tasks.Append(new EpilogueTask(barrier, isolate_group, old_space, + old_space->DataFreeList(i), &state)); } + isolate_group->safepoint_handler()->RunTasks(&tasks); old_space->heap_->new_space()->set_freed_in_words(state.NewFreeSize() >> kWordSizeLog2); diff --git a/runtime/vm/heap/marker.cc b/runtime/vm/heap/marker.cc index df733a2b6a1..38b83893664 100644 --- a/runtime/vm/heap/marker.cc +++ b/runtime/vm/heap/marker.cc @@ -901,7 +901,7 @@ void GCMarker::ProcessRememberedSet(Thread* thread) { store_buffer->PushBlock(writing, StoreBuffer::kIgnoreThreshold); } -class ParallelMarkTask : public ThreadPool::Task { +class ParallelMarkTask : public SafepointTask { public: ParallelMarkTask(GCMarker* marker, IsolateGroup* isolate_group, @@ -915,10 +915,10 @@ class ParallelMarkTask : public ThreadPool::Task { barrier_(barrier), visitor_(visitor), num_busy_(num_busy) {} + ~ParallelMarkTask() { barrier_->Release(); } - virtual void Run() { + void Run() override { if (!barrier_->TryEnter()) { - barrier_->Release(); return; } @@ -931,7 +931,28 @@ class ParallelMarkTask : public ThreadPool::Task { Thread::ExitIsolateGroupAsHelper(/*bypass_safepoint=*/true); barrier_->Sync(); - barrier_->Release(); + } + + void RunBlockedAtSafepoint() override { + if (!barrier_->TryEnter()) { + return; + } + + Thread* thread = Thread::Current(); + Thread::TaskKind saved_task_kind = thread->task_kind(); + thread->set_task_kind(Thread::kMarkerTask); + + RunEnteredIsolateGroup(); + + thread->set_task_kind(saved_task_kind); + + barrier_->Sync(); + } + + void RunMain() override { + RunEnteredIsolateGroup(); + + barrier_->Sync(); } void RunEnteredIsolateGroup() { @@ -1343,8 +1364,8 @@ void GCMarker::MarkObjects(PageSpace* page_space) { ResetSlices(); // Used to coordinate draining among tasks; all start out as 'busy'. RelaxedAtomic num_busy = 0; - // Phase 1: Iterate over roots and drain marking stack in tasks. + IntrusiveDList tasks; for (intptr_t i = 0; i < num_tasks; ++i) { SyncMarkingVisitor* visitor = visitors_[i]; // Visitors may or may not have already been created depending on @@ -1363,23 +1384,12 @@ void GCMarker::MarkObjects(PageSpace* page_space) { // such a visitor's local blocks. visitor->Flush(&global_list_); // Need to move weak property list too. - - if (i < (num_tasks - 1)) { - // Begin marking on a helper thread. - bool result = Dart::thread_pool()->Run( - this, isolate_group_, &old_marking_stack_, barrier, visitor, - &num_busy); - ASSERT(result); - } else { - // Last worker is the main thread. - visitor->Adopt(&global_list_); - ParallelMarkTask task(this, isolate_group_, &old_marking_stack_, - barrier, visitor, &num_busy); - task.RunEnteredIsolateGroup(); - barrier->Sync(); - barrier->Release(); - } + tasks.Append(new ParallelMarkTask(this, isolate_group_, + &old_marking_stack_, barrier, visitor, + &num_busy)); } + visitors_[0]->Adopt(&global_list_); + isolate_group_->safepoint_handler()->RunTasks(&tasks); for (intptr_t i = 0; i < num_tasks; i++) { SyncMarkingVisitor* visitor = visitors_[i]; diff --git a/runtime/vm/heap/safepoint.cc b/runtime/vm/heap/safepoint.cc index 2c3f501cefd..e018e25b70e 100644 --- a/runtime/vm/heap/safepoint.cc +++ b/runtime/vm/heap/safepoint.cc @@ -60,7 +60,7 @@ ForceGrowthSafepointOperationScope::~ForceGrowthSafepointOperationScope() { } SafepointHandler::SafepointHandler(IsolateGroup* isolate_group) - : isolate_group_(isolate_group) { + : isolate_group_(isolate_group), tasks_() { handlers_[SafepointLevel::kGC] = new LevelHandler(isolate_group, SafepointLevel::kGC); handlers_[SafepointLevel::kGCAndDeopt] = @@ -70,6 +70,7 @@ SafepointHandler::SafepointHandler(IsolateGroup* isolate_group) } SafepointHandler::~SafepointHandler() { + ASSERT(tasks_.IsEmpty()); for (intptr_t level = 0; level < SafepointLevel::kNumLevels; ++level) { ASSERT(handlers_[level]->owner_ == nullptr); delete handlers_[level]; @@ -359,12 +360,68 @@ void SafepointHandler::LevelHandler::NotifyWeAreParked(Thread* T) { void SafepointHandler::ExitSafepointLocked(Thread* T, MonitorLocker* tl, SafepointLevel level) { + ASSERT(T == Thread::Current()); while (T->IsSafepointRequestedLocked(level)) { T->SetBlockedForSafepoint(true); tl->Wait(); T->SetBlockedForSafepoint(false); + + MonitorLeaveScope mls(tl); + SafepointTask* task = nullptr; + { + MonitorLocker ml(threads_lock()); + if (!tasks_.IsEmpty()) { + task = tasks_.RemoveFirst(); + } + } + if (task != nullptr) { + task->RunBlockedAtSafepoint(); + delete task; + } } T->SetAtSafepoint(false, level); } +void SafepointHandler::RunTasks(IntrusiveDList* tasks) { + ASSERT(Thread::Current()->OwnsSafepoint()); + + // Withold one task for the main thread. + ASSERT(!tasks->IsEmpty()); + SafepointTask* main = tasks->RemoveFirst(); + + // First use threads blocked at this safepoint. + { + MonitorLocker tl(threads_lock()); + ASSERT(tasks_.IsEmpty()); + for (auto current = isolate_group()->thread_registry()->active_list(); + current != nullptr; current = current->next()) { + if (tasks->IsEmpty()) break; + + MonitorLocker tl(current->thread_lock()); + if (current->IsBlockedForSafepoint()) { + tasks_.Append(tasks->RemoveFirst()); + tl.Notify(); + } + } + } + + // Then use thread pool workers. + while (!tasks->IsEmpty()) { + bool result = Dart::thread_pool()->Run(tasks->RemoveFirst()); + ASSERT(result); + } + + // Run one task on the main thread. + main->RunMain(); + delete main; + + // Clean up any tasks that took too long to start. + { + MonitorLocker tl(threads_lock()); + while (!tasks_.IsEmpty()) { + delete tasks_.RemoveFirst(); + } + } +} + } // namespace dart diff --git a/runtime/vm/heap/safepoint.h b/runtime/vm/heap/safepoint.h index 3bb11184b71..2ce0e6f49e0 100644 --- a/runtime/vm/heap/safepoint.h +++ b/runtime/vm/heap/safepoint.h @@ -76,6 +76,22 @@ class ForceGrowthSafepointOperationScope : public ThreadStackResource { DISALLOW_COPY_AND_ASSIGN(ForceGrowthSafepointOperationScope); }; +// Subclasses of SafepointTask are able to run on thread blocked at a safepoint. +class SafepointTask : public ThreadPool::Task, + public IntrusiveDListEntry { + protected: + SafepointTask() {} + + public: + virtual ~SafepointTask() {} + + virtual void RunBlockedAtSafepoint() = 0; + virtual void RunMain() = 0; + + private: + DISALLOW_COPY_AND_ASSIGN(SafepointTask); +}; + // Implements handling of safepoint operations for all threads in an // IsolateGroup. class SafepointHandler { @@ -114,6 +130,8 @@ class SafepointHandler { return false; } + void RunTasks(IntrusiveDList* tasks); + private: class LevelHandler { public: @@ -201,6 +219,7 @@ class SafepointHandler { IsolateGroup* isolate_group_; LevelHandler* handlers_[SafepointLevel::kNumLevels]; + IntrusiveDList tasks_; friend class Isolate; friend class IsolateGroup; diff --git a/runtime/vm/heap/scavenger.cc b/runtime/vm/heap/scavenger.cc index 6f4eb07f4c0..7326edfb508 100644 --- a/runtime/vm/heap/scavenger.cc +++ b/runtime/vm/heap/scavenger.cc @@ -670,7 +670,7 @@ class ScavengerWeakVisitor : public HandleVisitor { DISALLOW_COPY_AND_ASSIGN(ScavengerWeakVisitor); }; -class ParallelScavengerTask : public ThreadPool::Task { +class ParallelScavengerTask : public SafepointTask { public: ParallelScavengerTask(IsolateGroup* isolate_group, ThreadBarrier* barrier, @@ -680,10 +680,10 @@ class ParallelScavengerTask : public ThreadPool::Task { barrier_(barrier), visitor_(visitor), num_busy_(num_busy) {} + ~ParallelScavengerTask() { barrier_->Release(); } - virtual void Run() { + void Run() override { if (!barrier_->TryEnter()) { - barrier_->Release(); return; } @@ -696,7 +696,28 @@ class ParallelScavengerTask : public ThreadPool::Task { Thread::ExitIsolateGroupAsHelper(/*bypass_safepoint=*/true); barrier_->Sync(); - barrier_->Release(); + } + + void RunBlockedAtSafepoint() override { + if (!barrier_->TryEnter()) { + return; + } + + Thread* thread = Thread::Current(); + Thread::TaskKind saved_task_kind = thread->task_kind(); + thread->set_task_kind(Thread::kScavengerTask); + + RunEnteredIsolateGroup(); + + thread->set_task_kind(saved_task_kind); + + barrier_->Sync(); + } + + void RunMain() override { + RunEnteredIsolateGroup(); + + barrier_->Sync(); } void RunEnteredIsolateGroup() { @@ -2039,28 +2060,21 @@ intptr_t Scavenger::ParallelScavenge(SemiSpace* from) { ThreadBarrier* barrier = new ThreadBarrier(num_tasks, 1); RelaxedAtomic num_busy = 0; + IsolateGroup* isolate_group = heap_->isolate_group(); + ParallelScavengerVisitor** visitors = new ParallelScavengerVisitor*[num_tasks]; + IntrusiveDList tasks; for (intptr_t i = 0; i < num_tasks; i++) { FreeList* freelist = heap_->old_space()->DataFreeList(i); - visitors[i] = new ParallelScavengerVisitor( - heap_->isolate_group(), this, from, freelist, &promotion_stack_); - if (i < (num_tasks - 1)) { - // Begin scavenging on a helper thread. - bool result = Dart::thread_pool()->Run( - heap_->isolate_group(), barrier, visitors[i], &num_busy); - ASSERT(result); - } else { - // Last worker is the main thread. - ParallelScavengerTask task(heap_->isolate_group(), barrier, visitors[i], - &num_busy); - task.RunEnteredIsolateGroup(); - barrier->Sync(); - barrier->Release(); - } + visitors[i] = new ParallelScavengerVisitor(isolate_group, this, from, + freelist, &promotion_stack_); + tasks.Append(new ParallelScavengerTask(isolate_group, barrier, visitors[i], + &num_busy)); } + isolate_group->safepoint_handler()->RunTasks(&tasks); - StoreBuffer* store_buffer = heap_->isolate_group()->store_buffer(); + StoreBuffer* store_buffer = isolate_group->store_buffer(); for (intptr_t i = 0; i < num_tasks; i++) { ParallelScavengerVisitor* visitor = visitors[i]; visitor->Finalize(store_buffer); diff --git a/runtime/vm/thread.cc b/runtime/vm/thread.cc index 462cd423fcd..80cdd2db63d 100644 --- a/runtime/vm/thread.cc +++ b/runtime/vm/thread.cc @@ -396,11 +396,6 @@ void Thread::EnterIsolate(Isolate* isolate) { ASSERT(thread->scheduled_dart_mutator_isolate_ == isolate); ASSERT(thread->isolate() == isolate); ASSERT(thread->isolate_group() == isolate->group()); - { - // Descheduled isolates are reloadable (if nothing else prevents it). - RawReloadParticipationScope enable_reload(thread); - thread->ExitSafepoint(); - } } else { thread = AddActiveThread(group, isolate, /*is_dart_mutator*/ true, /*bypass_safepoint=*/false); @@ -411,6 +406,14 @@ void Thread::EnterIsolate(Isolate* isolate) { isolate->scheduled_mutator_thread_ = thread; ResumeDartMutatorThreadInternal(thread); + + if (is_resumable) { + // Descheduled isolates are reloadable (if nothing else prevents it). + RawReloadParticipationScope enable_reload(thread); + thread->ExitSafepoint(); + } + + ASSERT(!thread->IsAtSafepoint()); } static bool ShouldSuspend(bool isolate_shutdown, Thread* thread) { @@ -556,7 +559,6 @@ void Thread::SuspendDartMutatorThreadInternal(Thread* thread, } void Thread::ResumeThreadInternal(Thread* thread) { - ASSERT(!thread->IsAtSafepoint()); ASSERT(thread->isolate_group() != nullptr); ASSERT(thread->execution_state() == Thread::kThreadInNative); ASSERT(thread->vm_tag() == VMTag::kInvalidTagId || diff --git a/runtime/vm/thread.h b/runtime/vm/thread.h index 37c5928c0e8..260505730ff 100644 --- a/runtime/vm/thread.h +++ b/runtime/vm/thread.h @@ -487,6 +487,7 @@ class Thread : public ThreadState { } TaskKind task_kind() const { return task_kind_; } + void set_task_kind(TaskKind kind) { task_kind_ = kind; } // Retrieves and clears the stack overflow flags. These are set by // the generated code before the slow path runtime routine for a diff --git a/runtime/vm/thread_pool.h b/runtime/vm/thread_pool.h index f67cc4dea22..74479acc682 100644 --- a/runtime/vm/thread_pool.h +++ b/runtime/vm/thread_pool.h @@ -45,6 +45,7 @@ class ThreadPool { bool Run(Args&&... args) { return RunImpl(std::unique_ptr(new T(std::forward(args)...))); } + bool Run(Task* task) { return RunImpl(std::unique_ptr(task)); } // Returns `true` if the current thread is running on the [this] thread pool. bool CurrentThreadIsWorker();