Cluster startup hard block

src/core/Akka.Cluster.Tests/StartupWithChannelExecutorSpec.cs

@@ -0,0 +1,74 @@
+//-----------------------------------------------------------------------
+// <copyright file="StartupWithOneThreadSpec.cs" company="Akka.NET Project">
+//     Copyright (C) 2009-2021 Lightbend Inc. <http://www.lightbend.com>
+//     Copyright (C) 2013-2021 .NET Foundation <https://github.com/akkadotnet/akka.net>
+// </copyright>
+//-----------------------------------------------------------------------
+
+using System;
+using System.Threading;
+using Akka.Actor;
+using Akka.Actor.Dsl;
+using Akka.Configuration;
+using Akka.Event;
+using Akka.TestKit;
+using Akka.Util;
+using Xunit;
+
+namespace Akka.Cluster.Tests
+{
+    public sealed class StartupWithChannelExecutorSpec : AkkaSpec
+    {
+        private static readonly Config Configuration = ConfigurationFactory.ParseString(@"
+            akka.actor.creation-timeout = 10s
+            akka.actor.provider = cluster
+            akka.actor.default-dispatcher.executor = channel-executor
+            akka.actor.internal-dispatcher.executor = channel-executor
+            akka.remote.default-remote-dispatcher.executor = channel-executor
+            akka.remote.backoff-remote-dispatcher.executor = channel-executor            
+        ").WithFallback(ConfigurationFactory.Default());
+
+        private long _startTime;
+
+        public StartupWithChannelExecutorSpec() : base(Configuration)
+        {
+            _startTime = MonotonicClock.GetTicks();
+        }
+
+        private Props TestProps
+        {
+            get
+            {
+                Action<IActorDsl> actor = (c =>
+                {
+                    c.ReceiveAny((o, context) => context.Sender.Tell(o));
+                    c.OnPreStart = context =>
+                    {
+                        var log = context.GetLogger();
+                        var cluster = Cluster.Get(context.System);
+                        log.Debug("Started {0} {1}", cluster.SelfAddress, Thread.CurrentThread.Name);
+                    };
+                });
+                return Props.Create(() => new Act(actor));
+            }
+        }
+
+        [Fact]
+        public void A_cluster_must_startup_with_channel_executor_dispatcher()
+        {
+            var totalStartupTime = TimeSpan.FromTicks(MonotonicClock.GetTicks() - _startTime).TotalMilliseconds;
+            Assert.True(totalStartupTime < (Sys.Settings.CreationTimeout - TimeSpan.FromSeconds(2)).TotalMilliseconds);
+            Sys.ActorOf(TestProps).Tell("hello");
+            Sys.ActorOf(TestProps).Tell("hello");
+            Sys.ActorOf(TestProps).Tell("hello");
+
+            var cluster = Cluster.Get(Sys);
+            totalStartupTime = TimeSpan.FromTicks(MonotonicClock.GetTicks() - _startTime).TotalMilliseconds;
+            Assert.True(totalStartupTime < (Sys.Settings.CreationTimeout - TimeSpan.FromSeconds(2)).TotalMilliseconds);
+
+            ExpectMsg("hello");
+            ExpectMsg("hello");
+            ExpectMsg("hello");
+        }
+    }
+}

src/core/Akka.Cluster/Cluster.cs

@@ -140,7 +140,11 @@ public Cluster(ActorSystemImpl system)
             _readView = new ClusterReadView(this);
 
             // force the underlying system to start
-            _clusterCore = GetClusterCoreRef().Result;
+            // and hard block the current thread
+            var clusterCoreTask = Task.Run(GetClusterCoreRef);
+            if (!clusterCoreTask.Wait(System.Settings.CreationTimeout))
+                throw new TimeoutException("cluster startup");
+            _clusterCore = clusterCoreTask.Result;
 
             system.RegisterOnTermination(Shutdown);
 

src/core/Akka/Dispatch/AbstractDispatcher.cs

@@ -6,7 +6,6 @@
 //-----------------------------------------------------------------------
 
 using System;
-using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Tasks;
 using Akka.Actor;
@@ -59,9 +58,9 @@ public sealed class DefaultDispatcherPrerequisites : IDispatcherPrerequisites
         /// <param name="settings">TBD</param>
         /// <param name="mailboxes">TBD</param>
         public DefaultDispatcherPrerequisites(
-            EventStream eventStream, 
-            IScheduler scheduler, 
-            Settings settings, 
+            EventStream eventStream,
+            IScheduler scheduler,
+            Settings settings,
             Mailboxes mailboxes)
         {
             Mailboxes = mailboxes;
@@ -120,8 +119,26 @@ internal sealed class ChannelExecutorConfigurator : ExecutorServiceConfigurator
     {
         public ChannelExecutorConfigurator(Config config, IDispatcherPrerequisites prerequisites) : base(config, prerequisites)
         {
-            var cfg = config.GetConfig("channel-executor");
-            Priority = (TaskSchedulerPriority)Enum.Parse(typeof(TaskSchedulerPriority), cfg.GetString("priority", "normal"), true);
+            var priorityName = config.GetString("channel-executor.priority", "None") ?? "None";
+            Priority = (TaskSchedulerPriority)Enum.Parse(typeof(TaskSchedulerPriority), priorityName, true);
+            if (Priority == TaskSchedulerPriority.None)
+            {
+                var dispatcherName = config.Root.GetObject().GetKey("name").ToString();
+                switch (dispatcherName)
+                {
+                    case "internal-dispatcher":
+                    case "default-remote-dispatcher":
+                        Priority = TaskSchedulerPriority.High;
+                        break;
+                    case "backoff-remote-dispatcher":
+                        Priority = TaskSchedulerPriority.Low;
+                        break;
+                    default:
+                        Priority = TaskSchedulerPriority.Normal;
+                        break;
+                };
+            }
+
         }
 
         public TaskSchedulerPriority Priority { get; }
@@ -133,6 +150,8 @@ public override ExecutorService Produce(string id)
             var scheduler = ChannelTaskScheduler.Get(Prerequisites.Settings.System).GetScheduler(Priority);
             return new TaskSchedulerExecutor(id, scheduler);
         }
+
+
     }
 
     /// <summary>
@@ -157,7 +176,7 @@ public override ExecutorService Produce(string id)
         /// </summary>
         /// <param name="config">TBD</param>
         /// <param name="prerequisites">TBD</param>
-        public DefaultTaskSchedulerExecutorConfigurator(Config config, IDispatcherPrerequisites prerequisites) 
+        public DefaultTaskSchedulerExecutorConfigurator(Config config, IDispatcherPrerequisites prerequisites)
             : base(config, prerequisites)
         {
         }
@@ -225,13 +244,13 @@ private static DedicatedThreadPoolSettings ConfigureSettings(Config config)
             {
                 var settings = new DedicatedThreadPoolSettings(
                     ThreadPoolConfig.ScaledPoolSize(
-                        fje.GetInt("parallelism-min"), 
-                        1.0, 
+                        fje.GetInt("parallelism-min"),
+                        1.0,
                         fje.GetInt("parallelism-max")),
-                        name:config.GetString("id"));
+                        name: config.GetString("id"));
                 return settings;
             }
-            
+
         }
     }
 

src/core/Akka/Dispatch/ChannelSchedulerExtension.cs

@@ -62,14 +62,17 @@ public ChannelTaskScheduler(ExtendedActorSystem system)
             //config channel-scheduler
             var config = system.Settings.Config.GetConfig("akka.channel-scheduler");
             _maximumConcurrencyLevel = ThreadPoolConfig.ScaledPoolSize(
-                        config.GetInt("parallelism-min"),
-                        config.GetDouble("parallelism-factor", 1.0D), // the scalar-based factor to scale the threadpool size to 
-                        config.GetInt("parallelism-max"));
+                        config?.GetInt("parallelism-min", 4) ?? 4,
+                        config?.GetDouble("parallelism-factor", 1.0D) ?? 1.0D, // the scalar-based factor to scale the threadpool size to 
+                        config?.GetInt("parallelism-max", 64) ?? 64);
             _maximumConcurrencyLevel = Math.Max(_maximumConcurrencyLevel, 1);
-            _maxWork = Math.Max(config.GetInt("work-max", _maxWork), 3); //min 3 normal work in work-loop
-
-            _workInterval = config.GetInt("work-interval", _workInterval);
-            _workStep = config.GetInt("work-step", _workStep);
+
+            if (config != null)
+            {
+                _maxWork = Math.Max(config.GetInt("work-max", _maxWork), 3); //min 3 normal work in work-loop
+                _workInterval = config.GetInt("work-interval", _workInterval);
+                _workStep = config.GetInt("work-step", _workStep);
+            }                
 
             //create task schedulers
             var channelOptions = new UnboundedChannelOptions()
@@ -276,7 +279,7 @@ private int DoWork(int workerId)
             //the work loop
             _threadPriority = TaskSchedulerPriority.Idle;
             try
-            {   
+            {
                 do
                 {
                     rounds++;