Fix #7130: Contention scheduling actions in HashedWheelTimerScheduler

src/core/Akka.Tests/Actor/Scheduler/HashedWheelTimerSchedulerContentionSpec.cs

@@ -0,0 +1,136 @@
+// -----------------------------------------------------------------------
+//  <copyright file="SchedulerHeavyUse.cs" company="Akka.NET Project">
+//      Copyright (C) 2009-2024 Lightbend Inc. <http://www.lightbend.com>
+//      Copyright (C) 2013-2024 .NET Foundation <https://github.com/akkadotnet/akka.net>
+//  </copyright>
+// -----------------------------------------------------------------------
+
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
+using System.Threading;
+using Akka.Actor;
+using Akka.Event;
+using FluentAssertions;
+using FluentAssertions.Extensions;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace Akka.Tests.Actor.Scheduler;
+
+public class HashedWheelTimerSchedulerContentionSpec: TestKit.Xunit2.TestKit
+{
+    private const int TotalActor = 5000;
+    private const int TotalThreads = 10;
+    private const int ActorsPerThread = TotalActor / TotalThreads;
+
+    public HashedWheelTimerSchedulerContentionSpec(ITestOutputHelper output) : base("{}", output)
+    {
+    }
+
+    [Fact]
+    public void SchedulerContentionTest()
+    {
+        var collector = CreateTestProbe();
+
+        foreach (var i in Enumerable.Range(0, TotalActor))
+        {
+            Sys.ActorOf(Props.Create(() => new DoStuffActor(TestActor, collector)), i.ToString());
+        }
+
+        Within(10.Seconds(), () =>
+        {
+            for (var i = 0; i < TotalActor; i++)
+            {
+                ExpectMsg<Done>();
+            }
+        });
+
+        object? received = null;
+        do
+        {
+            received = collector.ReceiveOne(TimeSpan.Zero);
+            if (received is long value)
+            {
+                value.Should().BeLessThan(200, "Scheduler should not experience resource contention");
+            }
+        } while (received is not null);
+
+    }
+
+    [Fact]
+    public void SchedulerContentionThreadedTest()
+    {
+        var collector = CreateTestProbe();
+        var threads = new List<Thread>();
+
+        foreach (var j in Enumerable.Range(0, TotalThreads))
+        {
+            threads.Add(new Thread(() => RunThread(j)));
+        }
+
+        foreach (var thread in threads)
+        {
+            thread.Start();
+        }
+
+        foreach (var thread in threads)
+        {
+            thread.Join();
+        }
+
+        Within(10.Seconds(), () =>
+        {
+            for (var i = 0; i < TotalActor; i++)
+            {
+                ExpectMsg<Done>();
+            }
+        });
+
+        object? received = null;
+        do
+        {
+            received = collector.ReceiveOne(TimeSpan.Zero);
+            if (received is long value)
+            {
+                value.Should().BeLessThan(200, "Scheduler should not experience resource contention");
+            }
+        } while (received is not null);
+
+        return;
+
+        void RunThread(int n)
+        {
+            n *= ActorsPerThread;
+            for (var i = 0; i < ActorsPerThread; i++)
+            {
+                Sys.ActorOf(Props.Create(() => new DoStuffActor(TestActor, collector)), (n + i).ToString());
+            }
+        }
+    }
+
+    public class DoStuffActor : ReceiveActor, IWithTimers
+    {
+        public ITimerScheduler Timers { get; set; }
+
+        public DoStuffActor(IActorRef probe, IActorRef collector)
+        {
+            Receive<Done>(d =>
+            {
+                Context.Stop(Self);
+                probe.Tell(d);
+            });
+
+            var sw = Stopwatch.StartNew();
+            Timers.StartSingleTimer("Test", Done.Instance, TimeSpan.FromSeconds(3));
+            sw.Stop();
+
+            if (sw.ElapsedMilliseconds > 0)
+            {
+                Context.GetLogger().Info($"{sw.ElapsedMilliseconds}");
+                collector.Tell(sw.ElapsedMilliseconds);
+            }
+        }
+    }
+}

src/core/Akka/Actor/Scheduler/HashedWheelTimerScheduler.cs

@@ -137,35 +137,41 @@ private static int NormalizeTicksPerWheel(int ticksPerWheel)
         private readonly HashSet<SchedulerRegistration> _rescheduleRegistrations = new();
 
 #if NET6_0_OR_GREATER
+        private readonly object _lock = new ();
         private PeriodicTimer? _timer;
         private readonly CancellationTokenSource _cts = new();
 
         private void Start()
         {
-            if (_workerState == WORKER_STATE_STARTED)
+            var start = false;
+            lock (_lock)
             {
-            } // do nothing
-            else if (_workerState == WORKER_STATE_INIT)
-            {
-                if (Interlocked.CompareExchange(ref _workerState, WORKER_STATE_STARTED, WORKER_STATE_INIT) ==
-                    WORKER_STATE_INIT)
+                switch (_workerState)
                 {
-                    var t = TimeSpan.FromTicks(_tickDuration);
-                    _timer = new PeriodicTimer(t);
-
-                    Task.Run(() => RunAsync(_cts.Token)); // start the clock
+                    case WORKER_STATE_STARTED:
+                        // do nothing
+                        break;
+
+                    case WORKER_STATE_INIT:
+                        _workerState = WORKER_STATE_STARTED;
+                        start = true;
+                        break;
+
+                    case WORKER_STATE_SHUTDOWN:
+                        throw new SchedulerException("cannot enqueue after timer shutdown");
+
+                    default:
+                        throw new InvalidOperationException($"Worker in invalid state: {_workerState}");
                 }
             }
-            else if (_workerState == WORKER_STATE_SHUTDOWN)
+
+            if (start)
             {
-                throw new SchedulerException("cannot enqueue after timer shutdown");
+                var timerDuration = TimeSpan.FromTicks(_tickDuration);
+                _timer ??= new PeriodicTimer(timerDuration);
+                Task.Run(() => RunAsync(_cts.Token)); // start the clock
             }
             else
-            {
-                throw new InvalidOperationException($"Worker in invalid state: {_workerState}");
-            }
-
-            while (_startTime == 0)
             {
                 _workerInitialized.Wait();
             }
@@ -248,7 +254,7 @@ private void Start()
             if (_workerState == WORKER_STATE_STARTED) { } // do nothing
             else if (_workerState == WORKER_STATE_INIT)
             {
-                _worker = new Thread(Run) { IsBackground = true };
+                _worker ??= new Thread(Run) { IsBackground = true };
                 if (Interlocked.CompareExchange(ref _workerState, WORKER_STATE_STARTED, WORKER_STATE_INIT) ==
                     WORKER_STATE_INIT)
                 {