diff --git a/runtime/bin/dartutils.h b/runtime/bin/dartutils.h index b414d731ca9..228779000fc 100644 --- a/runtime/bin/dartutils.h +++ b/runtime/bin/dartutils.h @@ -651,20 +651,6 @@ class ScopedBlockingCall { DISALLOW_COPY_AND_ASSIGN(ScopedBlockingCall); }; -// Remove once we remove the limitation on the number of running mutators. -// https://github.com/dart-lang/sdk/issues/54687 -class LeaveIsolateScope { - public: - LeaveIsolateScope() : isolate_(Dart_CurrentIsolate()) { Dart_ExitIsolate(); } - ~LeaveIsolateScope() { Dart_EnterIsolate(isolate_); } - - private: - Dart_Isolate isolate_; - - DISALLOW_ALLOCATION(); - DISALLOW_COPY_AND_ASSIGN(LeaveIsolateScope); -}; - struct MagicNumberData { static constexpr intptr_t kMaxLength = 8; diff --git a/runtime/bin/process.cc b/runtime/bin/process.cc index dfb0b77e403..027734d06fa 100644 --- a/runtime/bin/process.cc +++ b/runtime/bin/process.cc @@ -285,8 +285,6 @@ void FUNCTION_NAME(Process_Sleep)(Dart_NativeArguments args) { int64_t milliseconds = 0; // Ignore result if passing invalid argument and just set exit code to 0. DartUtils::GetInt64Value(Dart_GetNativeArgument(args, 0), &milliseconds); - - LeaveIsolateScope leave_isolate; TimerUtils::Sleep(milliseconds); } diff --git a/runtime/bin/process_linux.cc b/runtime/bin/process_linux.cc index da95e5a1ecd..14f7e86d311 100644 --- a/runtime/bin/process_linux.cc +++ b/runtime/bin/process_linux.cc @@ -861,8 +861,6 @@ bool Process::Wait(intptr_t pid, int alive = 3; while (alive > 0) { - LeaveIsolateScope leave_isolate; - // Blocking call waiting for events from the child process. if (TEMP_FAILURE_RETRY(poll(fds, alive, -1)) <= 0) { return CloseProcessBuffers(fds, alive); diff --git a/runtime/bin/process_macos.cc b/runtime/bin/process_macos.cc index 94568e3e176..5f3c8a5f316 100644 --- a/runtime/bin/process_macos.cc +++ b/runtime/bin/process_macos.cc @@ -826,8 +826,6 @@ bool Process::Wait(intptr_t pid, int alive = 3; while (alive > 0) { - LeaveIsolateScope leave_isolate; - // Blocking call waiting for events from the child process. if (TEMP_FAILURE_RETRY(poll(fds, alive, -1)) <= 0) { return CloseProcessBuffers(fds, alive); diff --git a/runtime/bin/process_win.cc b/runtime/bin/process_win.cc index b9949c9c52a..df4061ebdbf 100644 --- a/runtime/bin/process_win.cc +++ b/runtime/bin/process_win.cc @@ -866,8 +866,6 @@ bool Process::Wait(intptr_t pid, // Continue until all handles are closed. int alive = kHandles; while (alive > 0) { - LeaveIsolateScope leave_isolate; - // Blocking call waiting for events from the child process. DWORD wait_result = WaitForMultipleObjects(alive, events, FALSE, INFINITE); diff --git a/runtime/tests/vm/dart/isolates/many_isolates_blocked_at_ffi_test.dart b/runtime/tests/vm/dart/isolates/many_isolates_blocked_at_ffi_test.dart new file mode 100644 index 00000000000..1c8e5b24778 --- /dev/null +++ b/runtime/tests/vm/dart/isolates/many_isolates_blocked_at_ffi_test.dart @@ -0,0 +1,33 @@ +// Copyright (c) 2025, the Dart project authors. Please see the AUTHORS file +// for details. All rights reserved. Use of this source code is governed by a +// BSD-style license that can be found in the LICENSE file. + +import "dart:isolate"; +import "dart:io"; +import "dart:ffi"; + +typedef CSleep = Void Function(Int32); +typedef DartSleep = void Function(int); + +child(replyPort) { + replyPort.send(null); + + var sleep = DynamicLibrary.process().lookupFunction( + "sleep", + ); + sleep(60 * 60); +} + +main() async { + var pending = 0; + var port; + port = new RawReceivePort((msg) { + pending--; + if (pending == 0) exit(0); + }); + + for (var i = 0; i < 20; i++) { + Isolate.spawn(child, port.sendPort); + pending++; + } +} diff --git a/runtime/tests/vm/vm.status b/runtime/tests/vm/vm.status index 7f3bf68a67e..ef354ca8935 100644 --- a/runtime/tests/vm/vm.status +++ b/runtime/tests/vm/vm.status @@ -56,6 +56,9 @@ dart/kernel_determinism_test: SkipSlow dart/minimal_kernel_test: SkipSlow # gen_kernel is too slow with optimization_counter_threshold [ $compiler == app_jitk ] +dart/isolates/many_isolates_blocked_at_ffi_test: Skip # App snapshot creation needs regular isolate shutdown +dart/isolates/many_isolates_blocked_at_process_run_sync_test: Skip # App snapshot creation needs regular isolate shutdown +dart/isolates/many_isolates_blocked_at_sleep_test: Skip # App snapshot creation needs regular isolate shutdown dart/snapshot_version_test: RuntimeError [ $compiler == dartk ] diff --git a/runtime/vm/compiler/assembler/assembler_arm.cc b/runtime/vm/compiler/assembler/assembler_arm.cc index ffae85ede15..31869f0cf61 100644 --- a/runtime/vm/compiler/assembler/assembler_arm.cc +++ b/runtime/vm/compiler/assembler/assembler_arm.cc @@ -590,10 +590,10 @@ void Assembler::EnterFullSafepoint(Register addr, Register state) { add(addr, THR, Operand(addr)); Bind(&retry); ldrex(state, addr); - cmp(state, Operand(target::Thread::full_safepoint_state_unacquired())); + cmp(state, Operand(target::Thread::native_safepoint_state_unacquired())); b(&slow_path, NE); - mov(state, Operand(target::Thread::full_safepoint_state_acquired())); + mov(state, Operand(target::Thread::native_safepoint_state_acquired())); strex(TMP, state, addr); cmp(TMP, Operand(0)); // 0 means strex was successful. b(&done, EQ); @@ -651,10 +651,10 @@ void Assembler::ExitFullSafepoint(Register tmp1, add(addr, THR, Operand(addr)); Bind(&retry); ldrex(state, addr); - cmp(state, Operand(target::Thread::full_safepoint_state_acquired())); + cmp(state, Operand(target::Thread::native_safepoint_state_acquired())); b(&slow_path, NE); - mov(state, Operand(target::Thread::full_safepoint_state_unacquired())); + mov(state, Operand(target::Thread::native_safepoint_state_unacquired())); strex(TMP, state, addr); cmp(TMP, Operand(0)); // 0 means strex was successful. b(&done, EQ); @@ -690,8 +690,8 @@ void Assembler::TransitionNativeToGenerated(Register addr, ASSERT(!ignore_unwind_in_progress); #if defined(DEBUG) // Ensure we've already left the safepoint. - ASSERT(target::Thread::full_safepoint_state_acquired() != 0); - LoadImmediate(state, target::Thread::full_safepoint_state_acquired()); + ASSERT(target::Thread::native_safepoint_state_acquired() != 0); + LoadImmediate(state, target::Thread::native_safepoint_state_acquired()); ldr(TMP, Address(THR, target::Thread::safepoint_state_offset())); ands(TMP, TMP, Operand(state)); Label ok; diff --git a/runtime/vm/compiler/assembler/assembler_arm64.cc b/runtime/vm/compiler/assembler/assembler_arm64.cc index a335f456df8..253a2b88adc 100644 --- a/runtime/vm/compiler/assembler/assembler_arm64.cc +++ b/runtime/vm/compiler/assembler/assembler_arm64.cc @@ -1564,10 +1564,10 @@ void Assembler::EnterFullSafepoint(Register state) { add(addr, THR, Operand(addr)); Bind(&retry); ldxr(state, addr); - cmp(state, Operand(target::Thread::full_safepoint_state_unacquired())); + cmp(state, Operand(target::Thread::native_safepoint_state_unacquired())); b(&slow_path, NE); - movz(state, Immediate(target::Thread::full_safepoint_state_acquired()), 0); + movz(state, Immediate(target::Thread::native_safepoint_state_acquired()), 0); stxr(TMP, state, addr); cbz(&done, TMP); // 0 means stxr was successful. @@ -1637,10 +1637,11 @@ void Assembler::ExitFullSafepoint(Register state, add(addr, THR, Operand(addr)); Bind(&retry); ldxr(state, addr); - cmp(state, Operand(target::Thread::full_safepoint_state_acquired())); + cmp(state, Operand(target::Thread::native_safepoint_state_acquired())); b(&slow_path, NE); - movz(state, Immediate(target::Thread::full_safepoint_state_unacquired()), 0); + movz(state, Immediate(target::Thread::native_safepoint_state_unacquired()), + 0); stxr(TMP, state, addr); cbz(&done, TMP); // 0 means stxr was successful. @@ -1674,8 +1675,8 @@ void Assembler::TransitionNativeToGenerated(Register state, ASSERT(!ignore_unwind_in_progress); #if defined(DEBUG) // Ensure we've already left the safepoint. - ASSERT(target::Thread::full_safepoint_state_acquired() != 0); - LoadImmediate(state, target::Thread::full_safepoint_state_acquired()); + ASSERT(target::Thread::native_safepoint_state_acquired() != 0); + LoadImmediate(state, target::Thread::native_safepoint_state_acquired()); ldr(TMP, Address(THR, target::Thread::safepoint_state_offset())); and_(TMP, TMP, Operand(state)); Label ok; diff --git a/runtime/vm/compiler/assembler/assembler_ia32.cc b/runtime/vm/compiler/assembler/assembler_ia32.cc index c80db898018..e58d76da871 100644 --- a/runtime/vm/compiler/assembler/assembler_ia32.cc +++ b/runtime/vm/compiler/assembler/assembler_ia32.cc @@ -2495,12 +2495,12 @@ void Assembler::EnterFullSafepoint(Register scratch) { } pushl(EAX); - movl(EAX, Immediate(target::Thread::full_safepoint_state_unacquired())); - movl(scratch, Immediate(target::Thread::full_safepoint_state_acquired())); + movl(EAX, Immediate(target::Thread::native_safepoint_state_unacquired())); + movl(scratch, Immediate(target::Thread::native_safepoint_state_acquired())); LockCmpxchgl(Address(THR, target::Thread::safepoint_state_offset()), scratch); movl(scratch, EAX); popl(EAX); - cmpl(scratch, Immediate(target::Thread::full_safepoint_state_unacquired())); + cmpl(scratch, Immediate(target::Thread::native_safepoint_state_unacquired())); if (!FLAG_use_slow_path) { j(EQUAL, &done); @@ -2551,12 +2551,12 @@ void Assembler::ExitFullSafepoint(Register scratch, } pushl(EAX); - movl(EAX, Immediate(target::Thread::full_safepoint_state_acquired())); - movl(scratch, Immediate(target::Thread::full_safepoint_state_unacquired())); + movl(EAX, Immediate(target::Thread::native_safepoint_state_acquired())); + movl(scratch, Immediate(target::Thread::native_safepoint_state_unacquired())); LockCmpxchgl(Address(THR, target::Thread::safepoint_state_offset()), scratch); movl(scratch, EAX); popl(EAX); - cmpl(scratch, Immediate(target::Thread::full_safepoint_state_acquired())); + cmpl(scratch, Immediate(target::Thread::native_safepoint_state_acquired())); if (!FLAG_use_slow_path) { j(EQUAL, &done); @@ -2589,7 +2589,7 @@ void Assembler::TransitionNativeToGenerated(Register scratch, #if defined(DEBUG) // Ensure we've already left the safepoint. movl(scratch, Address(THR, target::Thread::safepoint_state_offset())); - andl(scratch, Immediate(target::Thread::full_safepoint_state_acquired())); + andl(scratch, Immediate(target::Thread::native_safepoint_state_acquired())); Label ok; j(ZERO, &ok); Breakpoint(); diff --git a/runtime/vm/compiler/assembler/assembler_riscv.cc b/runtime/vm/compiler/assembler/assembler_riscv.cc index a89553c9d4a..5b893574db2 100644 --- a/runtime/vm/compiler/assembler/assembler_riscv.cc +++ b/runtime/vm/compiler/assembler/assembler_riscv.cc @@ -4408,8 +4408,8 @@ void Assembler::TransitionNativeToGenerated(Register state, ASSERT(!ignore_unwind_in_progress); #if defined(DEBUG) // Ensure we've already left the safepoint. - ASSERT(target::Thread::full_safepoint_state_acquired() != 0); - li(state, target::Thread::full_safepoint_state_acquired()); + ASSERT(target::Thread::native_safepoint_state_acquired() != 0); + li(state, target::Thread::native_safepoint_state_acquired()); lx(RA, Address(THR, target::Thread::safepoint_state_offset())); and_(RA, RA, state); Label ok; @@ -4476,10 +4476,10 @@ void Assembler::EnterFullSafepoint(Register state) { addi(addr, THR, target::Thread::safepoint_state_offset()); Bind(&retry); lr(state, Address(addr, 0)); - subi(state, state, target::Thread::full_safepoint_state_unacquired()); + subi(state, state, target::Thread::native_safepoint_state_unacquired()); bnez(state, &slow_path, Assembler::kNearJump); - li(state, target::Thread::full_safepoint_state_acquired()); + li(state, target::Thread::native_safepoint_state_acquired()); sc(state, state, Address(addr, 0)); beqz(state, &done, Assembler::kNearJump); // 0 means sc was successful. @@ -4512,10 +4512,10 @@ void Assembler::ExitFullSafepoint(Register state, addi(addr, THR, target::Thread::safepoint_state_offset()); Bind(&retry); lr(state, Address(addr, 0)); - subi(state, state, target::Thread::full_safepoint_state_acquired()); + subi(state, state, target::Thread::native_safepoint_state_acquired()); bnez(state, &slow_path, Assembler::kNearJump); - li(state, target::Thread::full_safepoint_state_unacquired()); + li(state, target::Thread::native_safepoint_state_unacquired()); sc(state, state, Address(addr, 0)); beqz(state, &done, Assembler::kNearJump); // 0 means sc was successful. diff --git a/runtime/vm/compiler/assembler/assembler_x64.cc b/runtime/vm/compiler/assembler/assembler_x64.cc index 8009bbbc5be..1cf323049b9 100644 --- a/runtime/vm/compiler/assembler/assembler_x64.cc +++ b/runtime/vm/compiler/assembler/assembler_x64.cc @@ -155,12 +155,12 @@ void Assembler::EnterFullSafepoint() { // Compare and swap the value at Thread::safepoint_state from // unacquired to acquired. If the CAS fails, go to a slow-path stub. pushq(RAX); - movq(RAX, Immediate(target::Thread::full_safepoint_state_unacquired())); - movq(TMP, Immediate(target::Thread::full_safepoint_state_acquired())); + movq(RAX, Immediate(target::Thread::native_safepoint_state_unacquired())); + movq(TMP, Immediate(target::Thread::native_safepoint_state_acquired())); LockCmpxchgq(Address(THR, target::Thread::safepoint_state_offset()), TMP); movq(TMP, RAX); popq(RAX); - cmpq(TMP, Immediate(target::Thread::full_safepoint_state_unacquired())); + cmpq(TMP, Immediate(target::Thread::native_safepoint_state_unacquired())); if (!FLAG_use_slow_path && !FLAG_target_thread_sanitizer) { j(EQUAL, &done); @@ -231,12 +231,12 @@ void Assembler::ExitFullSafepoint(bool ignore_unwind_in_progress) { // fallthrough. pushq(RAX); - movq(RAX, Immediate(target::Thread::full_safepoint_state_acquired())); - movq(TMP, Immediate(target::Thread::full_safepoint_state_unacquired())); + movq(RAX, Immediate(target::Thread::native_safepoint_state_acquired())); + movq(TMP, Immediate(target::Thread::native_safepoint_state_unacquired())); LockCmpxchgq(Address(THR, target::Thread::safepoint_state_offset()), TMP); movq(TMP, RAX); popq(RAX); - cmpq(TMP, Immediate(target::Thread::full_safepoint_state_acquired())); + cmpq(TMP, Immediate(target::Thread::native_safepoint_state_acquired())); if (!FLAG_use_slow_path && !FLAG_target_thread_sanitizer) { j(EQUAL, &done); @@ -272,7 +272,7 @@ void Assembler::TransitionNativeToGenerated(bool exit_safepoint, #if defined(DEBUG) // Ensure we've already left the safepoint. movq(TMP, Address(THR, target::Thread::safepoint_state_offset())); - andq(TMP, Immediate(target::Thread::full_safepoint_state_acquired())); + andq(TMP, Immediate(target::Thread::native_safepoint_state_acquired())); Label ok; j(ZERO, &ok); Breakpoint(); diff --git a/runtime/vm/compiler/runtime_api.cc b/runtime/vm/compiler/runtime_api.cc index 902833a7c4b..d60213f4006 100644 --- a/runtime/vm/compiler/runtime_api.cc +++ b/runtime/vm/compiler/runtime_api.cc @@ -900,12 +900,12 @@ word Thread::stack_overflow_shared_stub_entry_point_offset(bool fpu_regs) { : stack_overflow_shared_without_fpu_regs_entry_point_offset(); } -uword Thread::full_safepoint_state_unacquired() { - return dart::Thread::full_safepoint_state_unacquired(); +uword Thread::native_safepoint_state_unacquired() { + return dart::Thread::native_safepoint_state_unacquired(); } -uword Thread::full_safepoint_state_acquired() { - return dart::Thread::full_safepoint_state_acquired(); +uword Thread::native_safepoint_state_acquired() { + return dart::Thread::native_safepoint_state_acquired(); } uword Thread::generated_execution_state() { diff --git a/runtime/vm/compiler/runtime_api.h b/runtime/vm/compiler/runtime_api.h index 1ffe184f676..56a438275a9 100644 --- a/runtime/vm/compiler/runtime_api.h +++ b/runtime/vm/compiler/runtime_api.h @@ -1197,8 +1197,8 @@ class Thread : public AllStatic { static uword vm_tag_dart_id(); static word safepoint_state_offset(); - static uword full_safepoint_state_unacquired(); - static uword full_safepoint_state_acquired(); + static uword native_safepoint_state_unacquired(); + static uword native_safepoint_state_acquired(); static word execution_state_offset(); static uword vm_execution_state(); diff --git a/runtime/vm/dart_api_impl.cc b/runtime/vm/dart_api_impl.cc index bd104c7f4f2..1e00e011764 100644 --- a/runtime/vm/dart_api_impl.cc +++ b/runtime/vm/dart_api_impl.cc @@ -1422,7 +1422,7 @@ DART_EXPORT void Dart_ShutdownIsolate() { // scope objects as the original transition happened outside this scope in // Dart_EnterIsolate/Dart_CreateIsolateGroup. ASSERT(T->execution_state() == Thread::kThreadInNative); - T->ExitSafepoint(); + T->ExitSafepointFromNative(); T->set_execution_state(Thread::kThreadInVM); I->WaitForOutstandingSpawns(); @@ -1861,7 +1861,7 @@ DART_EXPORT void Dart_ExitIsolate() { // scope objects as the original transition happened outside this scope in // Dart_EnterIsolate/Dart_CreateIsolateGroup. ASSERT(T->execution_state() == Thread::kThreadInNative); - T->ExitSafepoint(); + T->ExitSafepointFromNative(); T->set_execution_state(Thread::kThreadInVM); Thread::ExitIsolate(); } diff --git a/runtime/vm/heap/safepoint.h b/runtime/vm/heap/safepoint.h index f121ddc1ae8..91095b3579b 100644 --- a/runtime/vm/heap/safepoint.h +++ b/runtime/vm/heap/safepoint.h @@ -336,14 +336,14 @@ class TransitionGeneratedToNative : public TransitionSafepointState { // accordingly. ASSERT(T->execution_state() == Thread::kThreadInGenerated); T->set_execution_state(Thread::kThreadInNative); - T->EnterSafepoint(); + T->EnterSafepointToNative(); } ~TransitionGeneratedToNative() { // We are returning to generated code and so we are not at a safepoint // anymore. ASSERT(thread()->execution_state() == Thread::kThreadInNative); - thread()->ExitSafepoint(); + thread()->ExitSafepointFromNative(); thread()->set_execution_state(Thread::kThreadInGenerated); } @@ -392,7 +392,7 @@ class TransitionVMToNative : public TransitionSafepointState { ~TransitionVMToNative() { // We are returning to vm code and so we are not at a safepoint anymore. ASSERT(thread()->execution_state() == Thread::kThreadInNative); - thread()->ExitSafepoint(); + thread()->ExitSafepointFromNative(); thread()->set_execution_state(Thread::kThreadInVM); } @@ -435,7 +435,7 @@ class TransitionNativeToVM : public TransitionSafepointState { // We are about to execute vm code and so we are not at a safepoint anymore. ASSERT(T->execution_state() == Thread::kThreadInNative); if (T->no_callback_scope_depth() == 0) { - T->ExitSafepoint(); + T->ExitSafepointFromNative(); } T->set_execution_state(Thread::kThreadInVM); } @@ -445,7 +445,7 @@ class TransitionNativeToVM : public TransitionSafepointState { ASSERT(thread()->execution_state() == Thread::kThreadInVM); thread()->set_execution_state(Thread::kThreadInNative); if (thread()->no_callback_scope_depth() == 0) { - thread()->EnterSafepoint(); + thread()->EnterSafepointToNative(); } } @@ -502,7 +502,7 @@ class TransitionToVM : public TransitionSafepointState { ASSERT((execution_state_ == Thread::kThreadInVM) || (execution_state_ == Thread::kThreadInNative)); if (execution_state_ == Thread::kThreadInNative) { - T->ExitSafepoint(); + T->ExitSafepointFromNative(); T->set_execution_state(Thread::kThreadInVM); } ASSERT(T->execution_state() == Thread::kThreadInVM); @@ -534,7 +534,7 @@ class TransitionToNative : public TransitionSafepointState { (execution_state_ == Thread::kThreadInNative)); if (execution_state_ == Thread::kThreadInVM) { T->set_execution_state(Thread::kThreadInNative); - T->EnterSafepoint(); + T->EnterSafepointToNative(); } ASSERT(T->execution_state() == Thread::kThreadInNative); } @@ -542,7 +542,7 @@ class TransitionToNative : public TransitionSafepointState { ~TransitionToNative() { ASSERT(thread()->execution_state() == Thread::kThreadInNative); if (execution_state_ == Thread::kThreadInVM) { - thread()->ExitSafepoint(); + thread()->ExitSafepointFromNative(); thread()->set_execution_state(Thread::kThreadInVM); } } diff --git a/runtime/vm/isolate.cc b/runtime/vm/isolate.cc index 79308e28545..b9e6c292646 100644 --- a/runtime/vm/isolate.cc +++ b/runtime/vm/isolate.cc @@ -95,7 +95,7 @@ DEFINE_FLAG_HANDLER(DeterministicModeHandler, DEFINE_FLAG(bool, disable_thread_pool_limit, - false, + true, "Disables the limit of the thread pool (simulates custom embedder " "with custom message handler on unlimited number of threads)."); @@ -581,6 +581,8 @@ void IsolateGroup::set_saved_unlinked_calls(const Array& saved_unlinked_calls) { saved_unlinked_calls_ = saved_unlinked_calls.ptr(); } +static constexpr intptr_t kActiveMutatorPreemptionTimeout = 120; + void IsolateGroup::IncreaseMutatorCount(Isolate* mutator, bool is_nested_reenter) { ASSERT(mutator->group() == this); @@ -602,10 +604,64 @@ void IsolateGroup::IncreaseMutatorCount(Isolate* mutator, ASSERT(active_mutators_ <= max_active_mutators_); while (active_mutators_ == max_active_mutators_) { waiting_mutators_++; - ml.Wait(); + bool timed_out = false; + if (has_timeout_waiter_) { + ml.Wait(); + } else { + has_timeout_waiter_ = true; + timed_out = + ml.Wait(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::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(); } } @@ -626,6 +682,7 @@ void IsolateGroup::DecreaseMutatorCount(Isolate* mutator, bool is_nested_exit) { // max_active_mutators) and only use monitors in the uncommon case. MonitorLocker ml(active_mutators_monitor_.get()); ASSERT(active_mutators_ <= max_active_mutators_); + ASSERT(active_mutators_ > 0); active_mutators_--; if (waiting_mutators_ > 0) { ml.Notify(); diff --git a/runtime/vm/isolate.h b/runtime/vm/isolate.h index d3736802d97..209bd0e605f 100644 --- a/runtime/vm/isolate.h +++ b/runtime/vm/isolate.h @@ -541,10 +541,9 @@ class IsolateGroup : public IntrusiveDListEntry { void IncreaseMutatorCount(Isolate* mutator, bool is_nested_reenter); void DecreaseMutatorCount(Isolate* mutator, bool is_nested_exit); - intptr_t MutatorCount() const { - MonitorLocker ml(active_mutators_monitor_.get()); - return active_mutators_; - } + void ReincreaseMutatorCount(Thread* thread); + NO_SANITIZE_THREAD + intptr_t MutatorCount() const { return active_mutators_; } bool HasTagHandler() const { return library_tag_handler() != nullptr; } ObjectPtr CallTagHandler(Dart_LibraryTag tag, @@ -925,6 +924,7 @@ class IsolateGroup : public IntrusiveDListEntry { intptr_t active_mutators_ = 0; intptr_t waiting_mutators_ = 0; intptr_t max_active_mutators_ = 0; + bool has_timeout_waiter_ = false; NOT_IN_PRODUCT(GroupDebugger* debugger_ = nullptr); diff --git a/runtime/vm/runtime_entry.cc b/runtime/vm/runtime_entry.cc index b5df184e5a7..941a44f81c3 100644 --- a/runtime/vm/runtime_entry.cc +++ b/runtime/vm/runtime_entry.cc @@ -4329,7 +4329,7 @@ extern "C" void DFLRT_EnterSafepoint(NativeArguments __unusable_) { Thread* thread = Thread::Current(); ASSERT(thread->top_exit_frame_info() != 0); ASSERT(thread->execution_state() == Thread::kThreadInNative); - thread->EnterSafepoint(); + thread->EnterSafepointToNative(); TRACE_RUNTIME_CALL("%s", "EnterSafepoint done"); } DEFINE_RAW_LEAF_RUNTIME_ENTRY(EnterSafepoint, @@ -4355,7 +4355,7 @@ extern "C" void DFLRT_ExitSafepoint(NativeArguments __unusable_) { thread->isolate()->isolate_object_store()->preallocated_unwind_error(); Exceptions::PropagateError(unwind_error); } - thread->ExitSafepoint(); + thread->ExitSafepointFromNative(); TRACE_RUNTIME_CALL("%s", "ExitSafepoint done"); } @@ -4379,7 +4379,7 @@ extern "C" void DFLRT_ExitSafepointIgnoreUnwindInProgress( // is_unwind_in_progress flag because this is called as part of JumpToFrame // exception handler - we want this transition to complete so that the next // safepoint check does error propagation. - thread->ExitSafepoint(); + thread->ExitSafepointFromNative(); TRACE_RUNTIME_CALL("%s", "ExitSafepointIgnoreUnwindInProgress done"); } @@ -4439,7 +4439,7 @@ extern "C" Thread* DLRT_GetFfiCallbackMetadata( if (current_thread->execution_state() != Thread::kThreadInNative) { FATAL("Cannot invoke native callback from a leaf call."); } - current_thread->ExitSafepoint(); + current_thread->ExitSafepointFromNative(); current_thread->set_execution_state(Thread::kThreadInVM); } @@ -4494,7 +4494,7 @@ extern "C" Thread* DLRT_GetFfiCallbackMetadata( // in native code anymore. See tests/ffi/function_gc_test.dart for example. current_thread->set_execution_state(Thread::kThreadInVM); - current_thread->ExitSafepoint(); + current_thread->ExitSafepointFromNative(); current_thread->set_unboxed_int64_runtime_arg(metadata.context()); diff --git a/runtime/vm/thread.cc b/runtime/vm/thread.cc index ad38d537181..a0d88117ddb 100644 --- a/runtime/vm/thread.cc +++ b/runtime/vm/thread.cc @@ -436,6 +436,8 @@ void Thread::ExitIsolate(bool isolate_shutdown) { isolate->scheduled_mutator_thread_ = nullptr; + ASSERT(!ActiveMutatorStolenField::decode(thread->safepoint_state_.load())); + // Right now we keep the [Thread] object across the isolate's lifetime. This // makes entering/exiting quite fast as it mainly boils down to safepoint // transitions. Though any operation that walks over all active threads will @@ -1358,6 +1360,10 @@ void Thread::UnwindScopes(uword stack_marker) { } } +void Thread::HandleStolen() { + isolate_group()->ReincreaseMutatorCount(this); +} + void Thread::EnterSafepointUsingLock() { isolate_group()->safepoint_handler()->EnterSafepointUsingLock(this); } diff --git a/runtime/vm/thread.h b/runtime/vm/thread.h index 42a00336eaf..86c0cc48c4a 100644 --- a/runtime/vm/thread.h +++ b/runtime/vm/thread.h @@ -969,11 +969,11 @@ class Thread : public ThreadState { static bool IsSafepointLevelRequested(uword state, SafepointLevel level) { switch (level) { case SafepointLevel::kGC: - return (state & SafepointRequestedField::mask_in_place()) != 0; + return SafepointRequestedField::decode(state); case SafepointLevel::kGCAndDeopt: - return (state & DeoptSafepointRequestedField::mask_in_place()) != 0; + return DeoptSafepointRequestedField::decode(state); case SafepointLevel::kGCAndDeoptAndReload: - return (state & ReloadSafepointRequestedField::mask_in_place()) != 0; + return ReloadSafepointRequestedField::decode(state); default: UNREACHABLE(); } @@ -1071,13 +1071,48 @@ class Thread : public ThreadState { (execution_state() == kThreadInGenerated); } - static uword full_safepoint_state_unacquired() { - return (0 << AtSafepointField::shift()) | - (0 << AtDeoptSafepointField::shift()); + static uword native_safepoint_state_unacquired() { + return AtSafepointField::encode(false) | + AtDeoptSafepointField::encode(false) | + ActiveMutatorStealableField::encode(false); } - static uword full_safepoint_state_acquired() { - return (1 << AtSafepointField::shift()) | - (1 << AtDeoptSafepointField::shift()); + static uword native_safepoint_state_acquired() { + return AtSafepointField::encode(true) | + AtDeoptSafepointField::encode(true) | + ActiveMutatorStealableField::encode(true); + } + + bool TryStealActiveMutator() { + uword old_state = safepoint_state_.load(); + if (!ActiveMutatorStealableField::decode(old_state)) return false; + ASSERT(!ActiveMutatorStolenField::decode(old_state)); + ASSERT(AtSafepointField::decode(old_state)); + ASSERT(AtDeoptSafepointField::decode(old_state)); + uword new_state = old_state; + new_state = ActiveMutatorStealableField::update(false, new_state); + new_state = ActiveMutatorStolenField::update(true, new_state); + return safepoint_state_.compare_exchange_strong(old_state, new_state, + std::memory_order_relaxed); + } + + bool TryEnterSafepointToNative() { + uword old_state = 0; + uword new_state = AtSafepointBits(current_safepoint_level()) | + ActiveMutatorStealableField::encode(true); + return safepoint_state_.compare_exchange_strong(old_state, new_state, + std::memory_order_release); + } + + void EnterSafepointToNative() { + ASSERT(no_safepoint_scope_depth() == 0); + // First try a fast update of the thread state to indicate it is at a + // safepoint. + if (!TryEnterSafepointToNative()) { + // Fast update failed which means we could potentially be in the middle + // of a safepoint operation. + EnterSafepointUsingLock(); + safepoint_state_.fetch_or(ActiveMutatorStealableField::encode(true)); + } } bool TryEnterSafepoint() { @@ -1115,11 +1150,36 @@ class Thread : public ThreadState { } } + bool TryExitSafepointFromNative() { + uword old_state = AtSafepointBits(current_safepoint_level()) | + ActiveMutatorStealableField::encode(true); + uword new_state = 0; + return safepoint_state_.compare_exchange_strong(old_state, new_state, + std::memory_order_acquire); + } + + void ExitSafepointFromNative() { + if (!TryExitSafepointFromNative()) { + ExitSafepointUsingLock(); + uword old = safepoint_state_.fetch_and( + ~(ActiveMutatorStealableField::encode(true) | + ActiveMutatorStolenField::encode(true))); + if (ActiveMutatorStolenField::decode(old)) { + set_execution_state(Thread::kThreadInVM); + HandleStolen(); + } + } + } + void HandleStolen(); + void CheckForSafepoint() { // If we are in a runtime call that doesn't support lazy deopt, we will only // respond to gc safepointing requests. ASSERT(no_safepoint_scope_depth() == 0); if (IsSafepointRequested()) { + bool stolen = ActiveMutatorStolenField::decode(safepoint_state_.load()); + ASSERT(!stolen); + BlockForSafepoint(); } } @@ -1382,8 +1442,12 @@ class Thread : public ThreadState { BitField; using ReloadSafepointRequestedField = BitField; - using BlockedForSafepointField = + using ActiveMutatorStealableField = BitField; + using ActiveMutatorStolenField = + BitField; + using BlockedForSafepointField = + BitField; using BypassSafepointsField = BitField; using UnwindErrorInProgressField = @@ -1392,14 +1456,14 @@ class Thread : public ThreadState { static uword AtSafepointBits(SafepointLevel level) { switch (level) { case SafepointLevel::kGC: - return AtSafepointField::mask_in_place(); + return AtSafepointField::encode(true); case SafepointLevel::kGCAndDeopt: - return AtSafepointField::mask_in_place() | - AtDeoptSafepointField::mask_in_place(); + return AtSafepointField::encode(true) | + AtDeoptSafepointField::encode(true); case SafepointLevel::kGCAndDeoptAndReload: - return AtSafepointField::mask_in_place() | - AtDeoptSafepointField::mask_in_place() | - AtReloadSafepointField::mask_in_place(); + return AtSafepointField::encode(true) | + AtDeoptSafepointField::encode(true) | + AtReloadSafepointField::encode(true); default: UNREACHABLE(); } diff --git a/runtime/vm/thread_registry.cc b/runtime/vm/thread_registry.cc index 6fe6746a8d1..9b85b19484d 100644 --- a/runtime/vm/thread_registry.cc +++ b/runtime/vm/thread_registry.cc @@ -115,6 +115,20 @@ void ThreadRegistry::FlushMarkingStacks() { } } +intptr_t ThreadRegistry::StealActiveMutators() { + MonitorLocker ml(threads_lock()); + intptr_t count = 0; + Thread* thread = active_list_; + while (thread != nullptr) { + if (thread->TryStealActiveMutator()) { + ASSERT(thread->IsDartMutatorThread()); + count++; + } + thread = thread->next_; + } + return count; +} + void ThreadRegistry::AddToActiveListLocked(Thread* thread) { ASSERT(thread != nullptr); ASSERT(threads_lock()->IsOwnedByCurrentThread()); diff --git a/runtime/vm/thread_registry.h b/runtime/vm/thread_registry.h index 24b9468d9c5..813afd30ce4 100644 --- a/runtime/vm/thread_registry.h +++ b/runtime/vm/thread_registry.h @@ -38,6 +38,7 @@ class ThreadRegistry { void AcquireMarkingStacks(); void ReleaseMarkingStacks(); void FlushMarkingStacks(); + intptr_t StealActiveMutators(); // Concurrent-approximate number of active isolates in the active_list intptr_t active_isolates_count() { return active_isolates_count_.load(); }