Introduce ChannelExecutor

[Aaronontheweb]

Working on deprecating the ole' ForkJoinDispatcher and thought I'd try using System.Threading.Channels - results are pretty disappointing so far. Current built-in implementations of the ThreadPool dispatchers absolutely stomp these ones in terms of performance.

[Aaronontheweb]

This number looked suspiciously good, so I decided to check it against several other benchmarks - for whatever reason Channels did well here but got absolutely destroyed in the original PingPong benchmark as well as in the NBench suite.

[Aaronontheweb]

RemotePingPong numbers running two different Channel<T> dispatchers were about half of what they are in dev

[Aaronontheweb]

I suspect this is the biggest performance problem, how the ChannelReader<IRunnable> actually performs consumption via the ThreadPool (or in this case, the default TaskScheduler) but this is how a lot of other projects and implementations do it. I suspect their performance requirements aren't nearly as high as ours.

[Horusiath]

Take a look at how UnboundedChannel is implemented. The basic approach is that it's a ConcurrentQueue<> + a deque with lock for awaiting readers subscription.

If I understand correctly major incentive over just dispatching tasks to threadpool is control over degreeOfParallelism.

You could probably just replace channel with concurrent queue with asynchronous version of AutoResetEvent (it can have very simple construction as it's never awaited more than once).

[Aaronontheweb]

Yeah, the basic goal is to bound the degree of parallelism for multiple workloads in order to prevent starvation of /system and /remote actors - either the queue has to exist at the level of the OS' thread scheduler (which is what happens today) or it has to exist above it in the form of a virtual work queue. Our working theory is that we're probably better off letting the .NET ThreadPool manage the former and for us to apply an abstraction above the latter.

[Aaronontheweb]

You could probably just replace channel with concurrent queue with asynchronous version of AutoResetEvent (it can have very simple construction as it's never awaited more than once).

I'll give that a try!

[Aaronontheweb]

Made some changes to how we execute the read-side of the ChannelDispatcher and it is now significantly faster. Have some more tweaks I want to make, but this is progress.

[Aaronontheweb]

So latest results from the benchmarks:

1. PingPong benchmark - need to re-run these, but replacing the default dispatcher and the ForkJoinDispatcher with a ChannelExecutor resulted in a performance increase. Looks promising.
2. RemotePingPong benchmark - a clean dev branch ran showed about 80-81 active threads running on my machine (16 cores) and an upper limit of 192,000 msg/s. Running with the ChannelExecutor showed 50 active threads (bear in mind this is the sum of two ActorSystems running inside the same process for both cases) but peak performance dropped to 164,000 msg/s.

This is a big performance increase from my first attempt at using System.Threading.Channels and my goal was to cut down on the number of threads waiting for work in Akka.NET - that appears to be a success so far but it's come at the cost of 20% throughput. Not sure that's worth it. Going to keep tweaking this to see if I can improve performance on it.

[Aaronontheweb]

Related: #4537

[Aaronontheweb]

Using https://github.com/akkadotnet/akka.net-integration-tests/tree/master/src/ClusterPingPong to measure the efficacy of these changes, running as separate .NET 5 processes on the same host machine. 2 node cluster.

I tried running these benchmarks inside docker-compose initially, but the overhead of WSL / docker Vnet / docker itself made it impossible to get a clear signal on the throughput (numbers were pretty similar for both.)

Running on bare metal itself tells a different story though.

With TaskScheduler(MaxConcurrency = VirtualCpuCount)

OSVersion:                         Microsoft Windows NT 10.0.19041.0
ProcessorCount:                    16
Actor Count:                       32
Node Count:                        2
Active Connections:                1
Msgs sent/received per connection: 200000  (2e5)
Is Server GC:                      True
Thread count per node:             45

Connections, Actors/node, Total [actor], Total [msg], Msgs/sec, Total [ms]
       1,        5,       10,  1,200,000,    179,543,    6683.62
       1,       10,       20,  2,200,000,    170,565,   12898.27
       1,       15,       30,  3,200,000,    158,513,   20187.56
       1,       20,       40,  4,200,000,    156,176,   26892.67
       1,       25,       50,  5,200,000,    156,961,   33129.11
       1,       30,       60,  6,200,000,    153,478,   40396.54

With v1.4.18 (ForkJoinExecutor and ThreadPoolExecutor)

OSVersion:                         Microsoft Windows NT 10.0.19041.0
ProcessorCount:                    16
Actor Count:                       32
Node Count:                        2
Active Connections:                1
Msgs sent/received per connection: 200000  (2e5)
Is Server GC:                      True
Thread count per node:             62

Connections, Actors/node, Total [actor], Total [msg], Msgs/sec, Total [ms]
       1,        5,       10,  1,200,000,    214,524,    5593.78
       1,       10,       20,  2,200,000,    198,502,   11082.99
       1,       15,       30,  3,200,000,    193,198,   16563.26
       1,       20,       40,  4,200,000,    188,137,   22324.13
       1,       25,       50,  5,200,000,    186,125,   27938.12
       1,       30,       60,  6,200,000,    182,807,   33915.48

Looks like the TaskSchedulerExecutor got dwarfed by our control - need to try using the ChannelExecutor next.

[Aaronontheweb]

Made one small change:

ThreadPool.SetMinThreads(0,0) - this tells .NET to not set a minimum number of I/O and worker threads and instead allows the pool to scale up and down dynamically.

OSVersion:                         Microsoft Windows NT 10.0.19041.0
ProcessorCount:                    16
Actor Count:                       32
Node Count:                        2
Active Connections:                1
Msgs sent/received per connection: 200000  (2e5)
Is Server GC:                      True
Thread count per node:             45

Connections, Actors/node, Total [actor], Total [msg], Msgs/sec, Total [ms]
       1,        5,       10,  1,200,000,    190,670,    6293.58
       1,       10,       20,  2,200,000,    176,469,   12466.75
       1,       15,       30,  3,200,000,    185,703,   17231.78
       1,       20,       40,  4,200,000,    199,370,   21066.26
       1,       25,       50,  5,200,000,    203,247,   25584.53
       1,       30,       60,  6,200,000,    200,973,   30849.79

What a difference that makes! And these numbers were all generated at well below 100% CPU, which means that the bottleneck is now somewhere closer to the I/O pipeline:

[Aaronontheweb]

Performance numbers using our control for ThreadPool.SetMinThreads(0,0) were pretty much the same - including the gradual performance degradation as load increased.

[Aaronontheweb]

Trying this out using the System.Threading.Channels implementation instead:

OSVersion:                         Microsoft Windows NT 10.0.19041.0
ProcessorCount:                    16
Actor Count:                       32
Node Count:                        2
Active Connections:                1
Msgs sent/received per connection: 200000  (2e5)
Is Server GC:                      True
Thread count per node:             45

Connections, Actors/node, Total [actor], Total [msg], Msgs/sec, Total [ms]
       1,        5,       10,  1,200,000,    178,031,    6740.37
       1,       10,       20,  2,200,000,    170,268,   12920.74
       1,       15,       30,  3,200,000,    167,911,   19057.66
       1,       20,       40,  4,200,000,    177,717,   23632.95
       1,       25,       50,  5,200,000,    175,869,   29567.32
       1,       30,       60,  6,200,000,    193,511,   32039.38

Going to try re-running that one more time since the last figure has me scratching my head...

[Aaronontheweb]

Looked better the second time around:

OSVersion:                         Microsoft Windows NT 10.0.19041.0
ProcessorCount:                    16
Actor Count:                       32
Node Count:                        2
Active Connections:                1
Msgs sent/received per connection: 200000  (2e5)
Is Server GC:                      True
Thread count per node:             45

Connections, Actors/node, Total [actor], Total [msg], Msgs/sec, Total [ms]
       1,        5,       10,  1,200,000,    194,142,    6181.02
       1,       10,       20,  2,200,000,    176,987,   12430.24
       1,       15,       30,  3,200,000,    179,937,   17783.94
       1,       20,       40,  4,200,000,    192,664,   21799.51
       1,       25,       50,  5,200,000,    189,140,   27492.84
       1,       30,       60,  6,200,000,    189,949,   32640.24

[Aaronontheweb]

Numbers using the latest https://github.com/akkadotnet/akka.net-integration-tests/tree/master/src/ClusterPingPong

Docker

Using v1.4.18

seed_1          | OSVersion:                         Unix 5.4.72.2
seed_1          | ProcessorCount:                    16
seed_1          | Node Count:                        2
seed_1          | Active Connections:                1
seed_1          | Msgs sent/received per actor:      200000  (2e5)
seed_1          | Is Server GC:                      True
seed_1          | Thread count per node:             41
seed_1          |
seed_1          | Connections, Actors/node, Total [actor], Total [msg], Msgs/sec, Total [ms], Threads
seed_1          |        1,        5,             10,    1,000,000,    118,759,    8420.36,      3 -> 57
seed_1          |        1,       10,             20,    2,000,000,    133,184,   15016.80,      1 -> 51
seed_1          |        1,       15,             30,    3,000,000,    135,418,   22153.53,      1 -> 50
seed_1          |        1,       20,             40,    4,000,000,    135,450,   29530.99,      1 -> 50
seed_1          |        1,       25,             50,    5,000,000,    140,668,   35544.44,      2 -> 50
seed_1          |        1,       30,             60,    6,000,000,    133,377,   44985.23,      2 -> 50

Using TaskSchedulerExecutor w/ Bounded Concurrency in v1.4.19

seed_1          | OSVersion:                         Unix 5.4.72.2
seed_1          | ProcessorCount:                    16
seed_1          | Node Count:                        2
seed_1          | Active Connections:                1
seed_1          | Msgs sent/received per actor:      200000  (2e5)
seed_1          | Is Server GC:                      True
seed_1          | Thread count per node:             41
seed_1          |
seed_1          | Connections, Actors/node, Total [actor], Total [msg], Msgs/sec, Total [ms], Threads
seed_1          |        1,        5,             10,    1,000,000,    111,815,    8943.31,    3 -> 57
seed_1          |        1,       10,             20,    2,000,000,    136,361,   14666.87,    2 -> 51
seed_1          |        1,       15,             30,    3,000,000,    142,152,   21104.12,    2 -> 50
seed_1          |        1,       20,             40,    4,000,000,    141,230,   28322.51,    1 -> 50
seed_1          |        1,       25,             50,    5,000,000,    138,405,   36125.68,    2 -> 50
seed_1          |        1,       30,             60,    6,000,000,    135,683,   44220.70,    2 -> 50

Numbers inside Docker look very similar, including the thread counts. But I was able to verify that the ChannelExecutor is indeed running.

Bare Metal

Using v1.4.18

OSVersion:                         Microsoft Windows NT 10.0.19041.0                  
ProcessorCount:                    16                                                 
Node Count:                        2                                                  
Active Connections:                1                                                  
Msgs sent/received per actor:      200000  (2e5)                                      
Is Server GC:                      True                                               
Thread count per node:             61                                                 
                                                                                      
Connections, Actors/node, Total [actor], Total [msg], Msgs/sec, Total [ms], Threads   
       1,        5,             10,    1,000,000,    174,766,    5721.91,      5 -> 84
       1,       10,             20,    2,000,000,    179,417,   11147.18,      7 -> 83
       1,       15,             30,    3,000,000,    177,977,   16856.02,      3 -> 79
       1,       20,             40,    4,000,000,    181,384,   22052.61,      3 -> 71
       1,       25,             50,    5,000,000,    184,619,   27082.80,      2 -> 70
       1,       30,             60,    6,000,000,    176,577,   33979.39,      2 -> 68

Using TaskSchedulerExecutor w/ Bounded Concurrency in v1.4.19

OSVersion:                         Microsoft Windows NT 10.0.19041.0
ProcessorCount:                    16
Node Count:                        2
Active Connections:                1
Msgs sent/received per actor:      200000  (2e5)
Is Server GC:                      True
Thread count per node:             44

Connections, Actors/node, Total [actor], Total [msg], Msgs/sec, Total [ms], Threads
       1,        5,             10,    1,000,000,    141,820,    7051.18,    9 -> 78
       1,       10,             20,    2,000,000,    154,477,   12946.90,    9 -> 78
       1,       15,             30,    3,000,000,    177,678,   16884.40,    6 -> 65
       1,       20,             40,    4,000,000,    193,058,   20719.06,    8 -> 57
       1,       25,             50,    5,000,000,    193,859,   25791.82,    5 -> 54
       1,       30,             60,    6,000,000,    195,050,   30761.33,    5 -> 51

Much more visible results here - the TaskSchedulerExecutor wins in terms of overall performance once it has a chance to tune its ThreadPool by round 4 - but it also is producing these numbers using ~40% CPU.

[Aaronontheweb]

Going to try this one more time using the TaskSchedulerExecutor instead, since that will save us the trouble of having to install another dependency and is largely backwards-compatible with what we have today.

edit: Whoops, all of that data was with the TaskSchedulerExecutor - need to try ChannelExecutor instead.

[Aaronontheweb]

ChannelExecutor Performance

Docker

seed_1          | OSVersion:                         Unix 5.4.72.2
seed_1          | ProcessorCount:                    16
seed_1          | Node Count:                        2
seed_1          | Active Connections:                1
seed_1          | Msgs sent/received per actor:      200000  (2e5)
seed_1          | Is Server GC:                      True
seed_1          | Thread count per node:             41
seed_1          |
seed_1          | Connections, Actors/node, Total [actor], Total [msg], Msgs/sec, Total [ms], Threads
seed_1          |        1,        5,             10,    1,000,000,    114,908,    8702.59,    1 -> 57
seed_1          |        1,       10,             20,    2,000,000,    134,613,   14857.36,    1 -> 50
seed_1          |        1,       15,             30,    3,000,000,    140,762,   21312.48,    2 -> 50
seed_1          |        1,       20,             40,    4,000,000,    138,127,   28958.82,    2 -> 50
seed_1          |        1,       25,             50,    5,000,000,    138,427,   36119.92,    2 -> 50
seed_1          |        1,       30,             60,    6,000,000,    138,373,   43360.90,    1 -> 50

Bare Metal

OSVersion:                         Microsoft Windows NT 10.0.19041.0
ProcessorCount:                    16
Node Count:                        2
Active Connections:                1
Msgs sent/received per actor:      200000  (2e5)
Is Server GC:                      True
Thread count per node:             45

Connections, Actors/node, Total [actor], Total [msg], Msgs/sec, Total [ms], Threads
       1,        5,             10,    1,000,000,    154,427,    6475.54,    9 -> 79
       1,       10,             20,    2,000,000,    157,232,   12719.98,   13 -> 79
       1,       15,             30,    3,000,000,    165,067,   18174.39,    6 -> 61
       1,       20,             40,    4,000,000,    188,741,   21192.97,    6 -> 54
       1,       25,             50,    5,000,000,    194,141,   25754.36,    6 -> 54
       1,       30,             60,    6,000,000,    187,075,   32072.54,    8 -> 54

Looks about even with the TaskScheduler but I think the former's numbers are slightly better...

[Aaronontheweb]

Full ChannelExecutor source code, even though I'm removing it from this PR in favor of the FixedConcurrencyTaskScheduler implementation:

internal sealed class ChannelExecutor : ExecutorService
    {
        private Channel<IRunnable> _channel;
        private CancellationTokenSource _cts = new CancellationTokenSource();

        private int _readers = 0;

        public int DegreeOfParallelism { get; }

        public ChannelExecutor(string id, int degreeOfParallelism) : base(id)
        {
            _channel = Channel.CreateUnbounded<IRunnable>(new UnboundedChannelOptions()
            {
                AllowSynchronousContinuations = true,
                SingleReader = false,
                SingleWriter = true
            });

            DegreeOfParallelism = degreeOfParallelism;
        }

        private static WaitCallback Executor = o => ReadChannel((ChannelExecutor)o);

        private static void ReadChannel(ChannelExecutor executor)
        {
            try
            {
                while ( executor._channel.Reader.TryRead(out var runnable))
                {
                    runnable.Run();
                }
            }
            catch
            {
                // suppress exceptions
            }
            finally
            {
                Interlocked.Decrement(ref  executor._readers);
            }
        }

        public override void Execute(IRunnable run)
        {
            if (_channel.Writer.TryWrite(run))
            {                
                if (_readers < DegreeOfParallelism)
                {
                    var initial = _readers;
                    var newVale = _readers + 1;
                    if (initial == Interlocked.CompareExchange(ref _readers, newVale, initial))
                    {
                        // try to start a new worker
                        ThreadPool.UnsafeQueueUserWorkItem(Executor, this);
                    }
                }
            }
        }

        public override void Shutdown()
        {
            _channel.Writer.Complete();
            _cts.Cancel();
        }
    }

[Aaronontheweb]

Writing up some documentation for this now, but once that's done this should be good to go. Going to make this an opt-in feature as part of Akka.NET v1.4.19.

[Aaronontheweb]

Described my changes.

[Aaronontheweb]

This is really the right configuration, rather than a hard thread count of 4.

[Aaronontheweb]

The differences in the benchmarks with this and without it are massive. However, if this causes problems for users they can easily reset it by calling ThreadPool.SetMinThreads(yourValue, yourValue) after the ActorSystem is created. I doubt many users will need to do that and changing this default likely works to the benefit of most.

[Aaronontheweb]

Used to configure the TaskSchedulerExecutor(id, new FixedConcurrencyTaskScheduler(MaxParallelism));

[Aaronontheweb]

This is how we actually pass the new executor into dispatcher configurations.

[Aaronontheweb]

Debug event you can listen for if you want to be certain that the ChannelExecutor is loaded.

[Aaronontheweb]

Re-use ForkJoinDispatcher configuration block, since it expresses what we need.

[Arkatufus]

We might actually need to copy HOCON lines over and make them part of the executor config in the future, because having one part of the system depending on a config of another part can be a point of confusion for the end user

[Aaronontheweb]

Whoops - this was actually from a previous PR.

[Arkatufus]

Code looks good, just a few changes needed in the documentation

[Arkatufus]

We might actually need to copy HOCON lines over and make them part of the executor config in the future, because having one part of the system depending on a config of another part can be a point of confusion for the end user

[Aaronontheweb]

We might actually need to copy HOCON lines over and make them part of the executor config in the future, because having one part of the system depending on a config of another part can be a point of confusion for the end user

Agree, but not going to do that today since this is meant to be a drop-in replacement for ForkJoinDispatcher. Goal is to eventually purge DedicatedThreadPool out of the code base for good.

[Aaronontheweb]

We might actually need to copy HOCON lines over and make them part of the executor config in the future, because having one part of the system depending on a config of another part can be a point of confusion for the end user

Agree, but not going to do that today since this is meant to be a drop-in replacement for ForkJoinDispatcher. Goal is to eventually purge DedicatedThreadPool out of the code base for good.

Should also point out, this is a sub-config of the current HOCON node anyway

[Aaronontheweb]

ThreadPool.SetMinThreads(0,0) definitely skewed some of the tests inside the TestKit - going to need to take a look at that.

[Zetanova]

I would not recommend to set ThreadPool.SetMinThreads(0,0) or do any global ThreadPool settings.
Why set both limits work and IO threads?

I don't realy understand the effect that would have other then the ThreadPool schedule algo would be used instantly
and the first messages/IRunnable would be a bit delayed.

An idea would be first to add a totalReaders counter to count how may readers got called over the test-run.
Less would be better. I believe only to set ThreadPool.SetMinThreads(0,0) would have the effect to reduce the totalReaders count.
And other counters to count worker executions that got 0, 1 or many work-items.
Counters: totalReaders, zeroReader, singleReaders

To have the same effect as ThreadPool.SetMinThreads(0,0) a wrapped IRunnable could be used to start the next worker from one of the currently running wrokers.

[Aaronontheweb]

I would not recommend to set ThreadPool.SetMinThreads(0,0) or do any global ThreadPool settings.
Why set both limits work and IO threads?

This setting removes the lower bound on the number of minimum threads - now there are no limits, as the lower bound is zero. The previous value was Environment.ProcessorCount, which as our benchmarks showed was probably more than even most busy systems need.

In the event of an extremely busy system, this change probably won't have much impact as those threads will continue to be used. The one downside might be that a bunch of threads might be killed and recreated right away at startup, and Ben Adams pointed out some potential memory overhead from work-stealing that might be incurred by this change too: https://twitter.com/ben_a_adams/status/1387236911018819596

If the setting has adverse affects for a lot of our users, we'll roll it back and make it toggleable - but changing that setting improved all of the Akka.NET v1.4.18 defaults too.

An idea would be first to add a totalReaders counter to count how may readers got called over the test-run.
Less would be better. I believe only to set ThreadPool.SetMinThreads(0,0) would have the effect to reduce the totalReaders count.
And other counters to count worker executions that got 0, 1 or many work-items.
Counters: totalReaders, zeroReader, singleReaders

We'd need to keep track of how busy all of the current readers are and factor that in too, the time component. But I'm open to auto-tuning our use of workers on the thread pool as well. Definitely interested in doing more work in that area.

[Zetanova]

My proposed coutners would only for perf messurement.
I believe that the new work item will get often "stolen" by an other worker
and the queued up worker would run empty. zeroReaders++

Most likely the effect of ThreadPool.SetMinThreads(0,0) would be that
worker-1, worker-2, worker-3 would be queued in the ThreadPool queue
and worker-0 would process all work-items more or less by himself.

[Aaronontheweb]

@Zetanova I'd be very interested in seeing some experiments with this construct from contributors like you - your original Channel<T> implementation was one of the inspirations for this!

[Zetanova]

@Aaronontheweb
I will do one and test the current one at the weekend.

The refactor of the HashWheelTaskScheduler is on my todo list too.

[Aaronontheweb]

The refactor of the HashWheelTaskScheduler is on my todo list too.

That's another good one - as @to11mtm often says, adding support for "Schedule with fixed delay" is something we should aim for by the time Akka.NET v1.5 is released.

[Zetanova]

@Aaronontheweb I didn't used the new dotnet counters for now, but would it be not an idea to use implement them here
https://docs.microsoft.com/en-gb/dotnet/core/diagnostics/dotnet-counters
Of course about the overhead we would need to be very careful.

The benefit would be that somebody could exec the dotnet counter output on production / kubernetes cluster
without a spezial build