threading cleanup

Author: pavelsavara

src/libraries/System.Runtime.InteropServices.JavaScript/src/System/Runtime/InteropServices/JavaScript/JSSynchronizationContext.cs

@@ -11,22 +11,26 @@
 using System.Runtime.CompilerServices;
 using QueueType = System.Threading.Channels.Channel<System.Runtime.InteropServices.JavaScript.JSSynchronizationContext.WorkItem>;
 
-namespace System.Runtime.InteropServices.JavaScript {
+namespace System.Runtime.InteropServices.JavaScript
+{
     /// <summary>
     /// Provides a thread-safe default SynchronizationContext for the browser that will automatically
     ///  route callbacks to the main browser thread where they can interact with the DOM and other
     ///  thread-affinity-having APIs like WebSockets, fetch, WebGL, etc.
     /// Callbacks are processed during event loop turns via the runtime's background job system.
     /// </summary>
-    internal sealed unsafe class JSSynchronizationContext : SynchronizationContext {
+    internal sealed unsafe class JSSynchronizationContext : SynchronizationContext
+    {
         public readonly Thread MainThread;
 
-        internal readonly struct WorkItem {
+        internal readonly struct WorkItem
+        {
             public readonly SendOrPostCallback Callback;
             public readonly object? Data;
             public readonly ManualResetEventSlim? Signal;
 
-            public WorkItem (SendOrPostCallback callback, object? data, ManualResetEventSlim? signal) {
+            public WorkItem(SendOrPostCallback callback, object? data, ManualResetEventSlim? signal)
+            {
                 Callback = callback;
                 Data = data;
                 Signal = signal;
@@ -35,31 +39,33 @@ public WorkItem (SendOrPostCallback callback, object? data, ManualResetEventSlim
 
         private static JSSynchronizationContext? MainThreadSynchronizationContext;
         private readonly QueueType Queue;
-        private readonly Action _DataIsAvailable;
 
-        private JSSynchronizationContext (Thread mainThread)
-            : this (
-                mainThread,
+        private JSSynchronizationContext()
+            : this(
+                Thread.CurrentThread,
                 Channel.CreateUnbounded<WorkItem>(
                     new UnboundedChannelOptions { SingleWriter = false, SingleReader = true, AllowSynchronousContinuations = true }
                 )
             )
         {
         }
 
-        private JSSynchronizationContext (Thread mainThread, QueueType queue) {
+        private JSSynchronizationContext(Thread mainThread, QueueType queue)
+        {
             MainThread = mainThread;
             Queue = queue;
-            _DataIsAvailable = DataIsAvailable;
         }
 
-        public override SynchronizationContext CreateCopy () {
+        public override SynchronizationContext CreateCopy()
+        {
             return new JSSynchronizationContext(MainThread, Queue);
         }
 
-        private void AwaitNewData () {
+        private void AwaitNewData()
+        {
             var vt = Queue.Reader.WaitToReadAsync();
-            if (vt.IsCompleted) {
+            if (vt.IsCompleted)
+            {
                 DataIsAvailable();
                 return;
             }
@@ -69,31 +75,34 @@ private void AwaitNewData () {
             //  fire a callback that will schedule a background job to pump the queue on the main thread.
             var awaiter = vt.AsTask().ConfigureAwait(false).GetAwaiter();
             // UnsafeOnCompleted avoids spending time flowing the execution context (we don't need it.)
-            awaiter.UnsafeOnCompleted(_DataIsAvailable);
+            awaiter.UnsafeOnCompleted(DataIsAvailable);
         }
 
-        private void DataIsAvailable () {
-            // While we COULD pump here, we don't want to. We want the pump to happen on the next event loop turn.
-            // Otherwise we could get a chain where a pump generates a new work item and that makes us pump again, forever.
-            ScheduleBackgroundJob((void*)(delegate* unmanaged[Cdecl]<void>)&BackgroundJobHandler);
+        private void DataIsAvailable()
+        {
+            MainThreadScheduleBackgroundJob((void*)(delegate* unmanaged[Cdecl]<void>)&BackgroundJobHandler);
         }
 
-        public override void Post (SendOrPostCallback d, object? state) {
+        public override void Post(SendOrPostCallback d, object? state)
+        {
             var workItem = new WorkItem(d, state, null);
             if (!Queue.Writer.TryWrite(workItem))
                 throw new Exception("Internal error");
         }
 
         // This path can only run when threading is enabled
-        #pragma warning disable CA1416
+#pragma warning disable CA1416
 
-        public override void Send (SendOrPostCallback d, object? state) {
-            if (Thread.CurrentThread == MainThread) {
+        public override void Send(SendOrPostCallback d, object? state)
+        {
+            if (Thread.CurrentThread == MainThread)
+            {
                 d(state);
                 return;
             }
 
-            using (var signal = new ManualResetEventSlim(false)) {
+            using (var signal = new ManualResetEventSlim(false))
+            {
                 var workItem = new WorkItem(d, state, signal);
                 if (!Queue.Writer.TryWrite(workItem))
                     throw new Exception("Internal error");
@@ -102,40 +111,51 @@ public override void Send (SendOrPostCallback d, object? state) {
             }
         }
 
-        internal static void Install () {
-            MainThreadSynchronizationContext ??= new JSSynchronizationContext(Thread.CurrentThread);
-
-            SynchronizationContext.SetSynchronizationContext(MainThreadSynchronizationContext);
-            MainThreadSynchronizationContext.AwaitNewData();
+        internal static void Install()
+        {
+            var ctx = Current as JSSynchronizationContext;
+            if (ctx == null)
+            {
+                ctx = new JSSynchronizationContext();
+                MainThreadSynchronizationContext = ctx;
+                SetSynchronizationContext(ctx);
+            }
+            ctx.AwaitNewData();
         }
 
         [MethodImplAttribute(MethodImplOptions.InternalCall)]
-        internal static extern unsafe void ScheduleBackgroundJob(void* callback);
+        internal static extern unsafe void MainThreadScheduleBackgroundJob(void* callback);
 
 #pragma warning disable CS3016 // Arrays as attribute arguments is not CLS-compliant
         [UnmanagedCallersOnly(CallConvs = new[] { typeof(CallConvCdecl) })]
-        private static unsafe void BackgroundJobHandler () {
+#pragma warning restore CS3016
+        // this callback will arrive on the bound thread
+        private static unsafe void BackgroundJobHandler()
+        {
             MainThreadSynchronizationContext!.Pump();
         }
 
-        [UnmanagedCallersOnly(CallConvs = new[] { typeof(CallConvCdecl) })]
-        private static unsafe void RequestPumpCallback () {
-            ScheduleBackgroundJob((void*)(delegate* unmanaged[Cdecl]<void>)&BackgroundJobHandler);
-        }
-
-        private void Pump () {
-            try {
-                while (Queue.Reader.TryRead(out var item)) {
-                    try {
+        private void Pump()
+        {
+            try
+            {
+                while (Queue.Reader.TryRead(out var item))
+                {
+                    try
+                    {
                         item.Callback(item.Data);
                         // While we would ideally have a catch block here and do something to dispatch/forward unhandled
                         //  exceptions, the standard threadpool (and thus standard synchronizationcontext) have zero
                         //  error handling, so for consistency with them we do nothing. Don't throw in SyncContext callbacks.
-                    } finally {
+                    }
+                    finally
+                    {
                         item.Signal?.Set();
                     }
                 }
-            } finally {
+            }
+            finally
+            {
                 // If an item throws, we want to ensure that the next pump gets scheduled appropriately regardless.
                 AwaitNewData();
             }

src/mono/System.Private.CoreLib/src/System/Threading/ThreadPool.Browser.Mono.cs

@@ -28,7 +28,7 @@ public bool Unregister(WaitHandle? waitObject)
         }
     }
 
-    public static partial class ThreadPool
+    public static unsafe partial class ThreadPool
     {
         // Indicates whether the thread pool should yield the thread from the dispatch loop to the runtime periodically so that
         // the runtime may use the thread for processing other work
@@ -79,7 +79,7 @@ internal static void RequestWorkerThread()
             if (_callbackQueued)
                 return;
             _callbackQueued = true;
-            QueueCallback();
+            MainThreadScheduleBackgroundJob((void*)(delegate* unmanaged[Cdecl]<void>)&BackgroundJobHandler);
         }
 
         internal static void NotifyWorkItemProgress()
@@ -110,12 +110,13 @@ private static RegisteredWaitHandle RegisterWaitForSingleObject(
             throw new PlatformNotSupportedException();
         }
 
-        [DynamicDependency("Callback")]
         [MethodImplAttribute(MethodImplOptions.InternalCall)]
-        private static extern void QueueCallback();
+        internal static extern unsafe void MainThreadScheduleBackgroundJob(void* callback);
 
-        private static void Callback()
-        {
+#pragma warning disable CS3016 // Arrays as attribute arguments is not CLS-compliant
+        [UnmanagedCallersOnly(CallConvs = new[] { typeof(CallConvCdecl) })]
+#pragma warning restore CS3016
+        private static unsafe void BackgroundJobHandler () {
             _callbackQueued = false;
             ThreadPoolWorkQueue.Dispatch();
         }

src/mono/System.Private.CoreLib/src/System/Threading/TimerQueue.Browser.Mono.cs

@@ -5,6 +5,7 @@
 using System.Diagnostics;
 using System.Runtime.CompilerServices;
 using System.Diagnostics.CodeAnalysis;
+using System.Runtime.InteropServices;
 
 namespace System.Threading
 {
@@ -13,7 +14,7 @@ namespace System.Threading
     // Based on TimerQueue.Portable.cs
     // Not thread safe
     //
-    internal partial class TimerQueue
+    internal unsafe partial class TimerQueue
     {
         private static List<TimerQueue>? s_scheduledTimers;
         private static List<TimerQueue>? s_scheduledTimersToFire;
@@ -27,19 +28,20 @@ private TimerQueue(int _)
         {
         }
 
-        [DynamicDependency("TimeoutCallback")]
         // This replaces the current pending setTimeout with shorter one
         [MethodImplAttribute(MethodImplOptions.InternalCall)]
-        private static extern void SetTimeout(int timeout);
+        private static extern unsafe void MainThreadScheduleTimer(void* callback, int shortestDueTimeMs);
 
-        // Called by mini-wasm.c:mono_set_timeout_exec
-        private static void TimeoutCallback()
-        {
+#pragma warning disable CS3016 // Arrays as attribute arguments is not CLS-compliant
+        [UnmanagedCallersOnly(CallConvs = new[] { typeof(CallConvCdecl) })]
+#pragma warning restore CS3016
+        // Called by mini-wasm.c:mono_wasm_execute_timer
+        private static unsafe void TimerHandler () {
             // always only have one scheduled at a time
             s_shortestDueTimeMs = long.MaxValue;
 
             long currentTimeMs = TickCount64;
-            ReplaceNextSetTimeout(PumpTimerQueue(currentTimeMs), currentTimeMs);
+            ReplaceNextTimer(PumpTimerQueue(currentTimeMs), currentTimeMs);
         }
 
         // this is called with shortest of timers scheduled on the particular TimerQueue
@@ -57,13 +59,13 @@ private bool SetTimer(uint actualDuration)
 
             _scheduledDueTimeMs = currentTimeMs + (int)actualDuration;
 
-            ReplaceNextSetTimeout(ShortestDueTime(), currentTimeMs);
+            ReplaceNextTimer(ShortestDueTime(), currentTimeMs);
 
             return true;
         }
 
         // shortest time of all TimerQueues
-        private static void ReplaceNextSetTimeout(long shortestDueTimeMs, long currentTimeMs)
+        private static void ReplaceNextTimer(long shortestDueTimeMs, long currentTimeMs)
         {
             if (shortestDueTimeMs == long.MaxValue)
             {
@@ -77,7 +79,7 @@ private static void ReplaceNextSetTimeout(long shortestDueTimeMs, long currentTi
                 int shortestWait = Math.Max((int)(shortestDueTimeMs - currentTimeMs), 0);
                 // this would cancel the previous schedule and create shorter one
                 // it is expensive call
-                SetTimeout(shortestWait);
+                MainThreadScheduleTimer((void*)(delegate* unmanaged[Cdecl]<void>)&TimerHandler, shortestWait);
             }
         }
 

src/mono/mono/metadata/gc.c

@@ -695,7 +695,7 @@ mono_wasm_gc_finalize_notify (void)
 	/* use this if we are going to start the finalizer thread on wasm. */
 	mono_coop_sem_post (&finalizer_sem);
 #else
-	mono_threads_schedule_background_job (mono_runtime_do_background_work);
+	mono_main_thread_schedule_background_job (mono_runtime_do_background_work);
 #endif
 }
 

src/mono/mono/metadata/sgen-mono.c

@@ -2897,7 +2897,7 @@ sgen_client_binary_protocol_collection_end (int minor_gc_count, int generation,
 void
 sgen_client_schedule_background_job (void (*cb)(void))
 {
-	mono_threads_schedule_background_job (cb);
+	mono_main_thread_schedule_background_job (cb);
 }
 
 #endif