Yesterday, I wrote a long article about asynchronous processing in Akka .NET actors, the dangers of async void methods, and PipeTo().

The article was written as a result of outdated Akka .NET documentation which claimed that async/await was evil in Akka .NET actors and recommended using PipeTo() instead (the documentation has since been updated). This was further exacerbated by the confusion between asynchronous and concurrent processing in the documentation.

I would like to thank Aaron Stannard and Roger Alsing from the Akka .NET team who, via separate channels, clarified a lot of the confusion. This new article covers asynchronous and concurrent processing in Akka .NET actors as a result.

Asynchronous vs Concurrent Processing

A lot of people confuse asynchronous and concurrent processing, more so given that you can do both with .NET Tasks.

This is asynchronous:

        static async Task RunAsync()
        {
            await Task.Delay(1000);
        }

RunAsync begins executing, but is suspended waiting for some external operation to occur (typically I/O such as reading from a database or a REST service). It can resume execution when the operation completes, but can’t continue doing other things in the meantime.

This, on the other hand, is concurrent:

        static async Task RunAsync()
        {
            Task.Delay(1000);
        }

Because the task is not awaited, the method proceeds with its execution while the task is running. In fact, we get a nice big warning when we do this in a method marked as async:

If we extend our example just a little bit, we can understand the behaviour better:

        static Task RunAsync()
        {
            Task.Delay(1000)
                .ContinueWith(x => Console.WriteLine("After Delay"));
            Console.WriteLine("End of RunAsync()");
        }

The output is as follows:

End of RunAsync()
After Delay

Now, I could go on about definitions of synchrony and asynchrony, concurrency and parallelism, interleaving, and task-switching. But that would bore you to tears, and it’s really not the point here.

The important thing is to realise that despite using very similar C# syntax, we’re doing two very different things here. And I need to make this clear because PipeTo() is really targeted at concurrent processing, although it is described within the context of asynchrony in Akka .NET documentation.

async/await in actors

You can do async/await in an actor by using the ReceiveActor‘s ReceiveAsync() method:

    public class MyActor : ReceiveActor
    {
        public MyActor()
        {
            this.ReceiveAsync<string>(HandleAsync);
        }

        public async Task HandleAsync(string str)
        {
            Console.WriteLine($"Begin {str}");

            await Task.Delay(2000);

            Console.WriteLine($"End {str}");
        }
    }

This is perfectly valid, and you will most likely resort to this when you need a guarantee on message order. In such situations, you don’t want to start processing the next message while waiting for an I/O operation to complete, as it is possible for the results of an older message to overwrite those of a newer message.

It is also very useful for sequential steps in I/O operations that depend directly on each other, such as when talking to a database:

        public async Task HandleAsync(string str)
        {
            const string connStr = @"Data Source=.\SQLEXPRESS;Database=test;Integrated Security=true";
            using (var conn = new SqlConnection(connStr))
            {
                await conn.OpenAsync();

                const string sql = "select * from person;";

                using (var command = new SqlCommand(sql, conn))
                using (var reader = await command.ExecuteReaderAsync())
                {
                    while (await reader.ReadAsync())
                    {
                        string id = reader["id"].ToString();
                        string name = reader["name"].ToString();

                        Console.WriteLine($"{id} {name}");
                    }
                }
            }
        }

However, this comes at a cost. First, there is a performance impact when compared to synchronous execution, because the Akka .NET pipeline needs to carry the current context across asynchronous steps.

Secondly, the actor will not be able to process the next message until the current one is finished. Sometimes, this is exactly what you want, such as when you need a guarantee on message order. In such situations, you don’t want to start processing the next message while waiting for an I/O operation to complete, as it is possible for the results of an older message to overwrite those of a newer message. But if your message processing does not depend on prior state, you will get much more throughput if you run tasks concurrently and use PipeTo() to collect the results (more on this later).

Using Ask()

Ask() lets you do request/response between actors:

    public class ServiceActor : ReceiveActor
    {
        public ServiceActor()
        {
            this.ReceiveAsync<string>(HandleAsync);
        }

        public async Task HandleAsync(string str)
        {
            await Task.Delay(2000);
            Sender.Tell(str + " done");
        }
    }

    public class MyActor : ReceiveActor
    {
        private IActorRef serviceActor;

        public MyActor(IActorRef serviceActor)
        {
            this.serviceActor = serviceActor;

            this.ReceiveAsync<string>(HandleAsync);
        }

        public async Task HandleAsync(string str)
        {
            Console.WriteLine($"Begin {str}");

            var result = await this.serviceActor.Ask(str);

            Console.WriteLine($"End {result}");
        }
    }

Here is the output for this:

The approach above is something you typically want to avoid, for the same reason outlined in the previous section. If your actor is waiting for a response, it can’t process other messages in the meantime. Most of the time you should spawn a concurrent task as shown in the documentation, unless you have a good reason for not wanting to process the next message before the current one has finished. Try to design your system in a push fashion, rather than request/response.

Concurrent Execution and PipeTo()

If you have no reason to process messages in a strictly sequential manner, then you can do long-running tasks and I/O operations in a spawned task.

    public class MyActor : ReceiveActor
    {
        public MyActor()
        {
            this.Receive<string>(x => Handle(x));
        }

        public void Handle(string str)
        {
            Task.Run(async () =>
            {
                Console.WriteLine($"Begin {str}");

                await Task.Delay(2000);

                Console.WriteLine($"End {str}");
            });
        }
    }

Because of this, the actual processing you do within the task will be interleaved, as in yesterday’s article:

But this is okay, because we’re working on the assumption that the messages don’t need to be processed strictly in sequence.

Now if you want to send the result of your concurrent tasks somewhere, you can do that with PipeTo():

    public class MyActor : ReceiveActor
    {
        public MyActor()
        {
            this.Receive<string>(x => Handle(x));
            this.Receive<int>(x => Handle(x));
        }

        public void Handle(string str)
        {
            Task.Run(async () =>
            {
                Console.WriteLine($"Begin {str}");

                await Task.Delay(2000);

                Console.WriteLine($"End {str}");

                return 42;
            }).PipeTo(Self);
        }

        public void Handle(int result)
        {
            Console.WriteLine($"Got result: {result}");
        }
    }

The result of the concurrent operation is sent to the actor you specify (in this case to itself) and processed as any other message in its mailbox. You can also do post-processing (e.g. check HTTP status code after an HTTP GET operation) by adding a ContinueWith(); see the PipeTo() article on the Petabridge blog for an example.

More Advanced Concurrent Operations

Given that you can use both tasks and async/await in your actors, you can use all of the typical patterns you would normally use with the Task Parallel Library (TPL).

Here’s an example representing when you’d aggregate data from multiple external resources:

    public class MyActor : ReceiveActor
    {
        public MyActor()
        {
            this.ReceiveAsync<string>(x => HandleAsync(x));
        }

        public async Task HandleAsync(string str)
        {
            var task1 = Task.Delay(1000).ContinueWith(x => { return 1; });
            var task2 = Task.Delay(2000).ContinueWith(x => { return 2; });
            var task3 = Task.Delay(3000).ContinueWith(x => { return 3; });

            var results = await Task.WhenAll<int>(task1, task2, task3);
            var sum = results.Sum();
            Console.WriteLine(sum);
        }
    }

WhenAll() will wait for all the tasks to complete before the method can proceed with its execution. Here’s the output:

Here’s another example which takes the result of whichever task completes first:

        public async Task HandleAsync(string str)
        {
            var task1 = Task.Delay(1000).ContinueWith(x => { return 1; });
            var task2 = Task.Delay(2000).ContinueWith(x => { return 2; });
            var task3 = Task.Delay(3000).ContinueWith(x => { return 3; });

            var result = await await Task.WhenAny<int>(task1, task2, task3);
            Console.WriteLine(result);
        }

In this example, WhenAny() suspends execution of the method until any of the tasks completes. The result from the fastest task is taken.

Note: if you’re looking to do this kind of concurrent fastest-query operation, you might want to look at Akka .NET Routers with Routing Strategies such as ScatterGatherFirstCompleted.

Update 21st August 2016: I wrote this article based on outdated Akka .NET documentation that discouraged async/await within actors and suggested using PipeTo() instead. Akka .NET now does support async/await (thanks to the ReceiveAsync() method), and PipeTo() is not a replacement for it. Aaron Stannard (in a comment on this post) and Roger Alsing (on Reddit) from the Akka .NET team were very prompt in correcting various misconceptions, and Aaron Stannard has since updated the Petabridge blog post about PipeTo(). See my followup post for the latest best practices.

Tasks and the more recent async/await syntactic sugar have been a blessing for .NET developers aiming to keep their applications responsive despite increasing requirements for I/O and CPU-intensive requests.

Thus it was really odd for me to learn that Akka .NET, an emergent framework for distributed computing, not only does not support async/await directly within actors, but actually discourages its use (going as far as calling them “code smell”).

In fact, they implemented this PipeTo() workaround that you need to use, sending the result of a task to an actor for processing. You can’t use async/await; you have to resort to the old ContinueWith() way of chaining tasks if you want to do any post-execution logic. If you’ve worked with ContinueWith() in the past, you’ll know it can get ugly really fast.

Why is it such a problem to have elegant asynchrony in our actors, seeing how competitor Microsoft Orleans has no problem with it? As Natan Vivo said in the comments of The Top 7 Mistakes Newbies Make with Akka.NET:

“The fact I decided to use DbCommand.ExecuteNonQueryAsync() instead of DbCommand.ExecuteNonQuery() shouldn’t force me to break a single message into multiple messages with PipeTo.”

Update 20th August 2016: Thanks to the Reddit user who brought to my attention that there actually is proper async support (though apparently not yet documented anywhere). Use the ReceiveActor’s ReceiveAsync() method.

Why Akka .NET Discourages async/await

To learn why awaiting in an actor is bad, let’s break the rules and do it.

    public class BusyActor : UntypedActor
    {
        protected override void OnReceive(object message)
        {
            Console.WriteLine($"Begin processing {message.ToString()}");

            Thread.Sleep(2000);

            Console.WriteLine($"End processing {message.ToString()}");
        }
    }

I have this example actor. For now it’s doing synchronous stuff, sleeping for a couple of seconds and writing something before and after so we can see the behaviour.

        static void Main(string[] args)
        {
            using (var actorSystem = ActorSystem.Create("MyActorSystem"))
            {
                var actor = actorSystem.ActorOf(Props.Create<BusyActor>(), "BusyActor");

                actor.Tell("Task 1");
                actor.Tell("Task 2");
                actor.Tell("Task 3");

                Console.ReadLine();
            }
        }

The main program simply creates the actor system and the actor, and then sends it three messages in succession.

As you can see, the messages are handled sequentially and there is no overlap.

Now let’s change the actor to work asynchronously instead:

    public class BusyActor : UntypedActor
    {
        protected override async void OnReceive(object message)
        {
            Console.WriteLine($"Begin processing {message.ToString()}");

            await Task.Delay(2000);

            Console.WriteLine($"End processing {message.ToString()}");
        }
    }

Run it again…

What happened here? All three messages were processed in quick succession, and they have been interleaved. This is very bad, and in fact we were warned about it. Quoting the questions on the official PipeTo() sample:

“Await breaks the “actors process one message at a time” guarantee, and suddenly your actor’s context might be different. Variables such as the Sender of the previous message may be different, or the actor might even be shutting down when the await call returns to the previous context.”

Why Processing Messages Asynchronously Causes Interleaving

We can learn a lot about how actors process messages by investigating the Akka .NET source code. This method in Mailbox.cs seems to be more or less where actors begin to process their messages:

        private void ProcessMailbox(int left, long deadlineTicks)
        {
            while (ShouldProcessMessage())
            {
                Envelope next;
                if (!TryDequeue(out next)) return;

                DebugPrint("{0} processing message {1}", Actor.Self, next);

                // not going to bother catching ThreadAbortExceptions here, since they'll get rethrown anyway
                Actor.Invoke(next);
                ProcessAllSystemMessages();
                if (left > 1 && (Dispatcher.ThroughputDeadlineTime.HasValue == false || (MonotonicClock.GetTicks() - deadlineTicks) < 0))
                {
                    left = left - 1;
                    continue;
                }
                break;
            }
        }

From Actor.Invoke(), there is a succession of method calls that ends in a method called Receive() in UntypedActor.cs:

        protected sealed override bool Receive(object message)
        {
            OnReceive(message);
            return true;
        }

Our OnReceive() method, where implement our message-handling logic for our actors, is subsequently called.

Now, the code above may look confusing, but the point here is not to understand what it’s doing exactly. Take a closer look. The methods in the call stack are mostly void (or otherwise returning simple types). There are no Tasks to be seen anywhere.

What does this mean for us? It means that we’re doing something very bad when we declare our message handler as async void.

Understanding async void

In order to better understand why the approach we took earlier will never work, it’s best to look at a much simpler example:

    class Program
    {
        static void Main(string[] args)
        {
            RunAll();
            Console.ReadLine();
        }

        static void RunAll()
        {
            RunJob("Job 1");
            RunJob("Job 2");
            RunJob("Job 3");
        }

        static void RunJob(string str)
        {
            Console.WriteLine("Start " + str);

            Thread.Sleep(2000);

            Console.WriteLine("End " + str);
        }
    }

Here we’ve reproduced the earlier scenario, but with no Akka .NET. And with the synchronous implementation, it works just fine:

Let’s change RunJob() to run asynchronously:

        static async void RunJob(string str)
        {
            Console.WriteLine("Start " + str);

            await Task.Delay(2000);

            Console.WriteLine("End " + str);
        }

When we run it, the following happens:

This is exactly the same interleaving problem we had with Akka .NET, except that this time we have no Akka .NET.

The real reason why we have this problem is due to an incorrect use of asynchrony. As you can read in Stephen Cleary’s MSDN Magazine article, “Async/Await – Best Practices in Asynchronous Programming” (March 2013), async void methods can be pretty dangerous to work with. When you call an async void method, you have two main problems: you have no way of awaiting completion of the method, and exceptions can bring the whole application down.

But here, we have also seen a third problem: that the method effectively exits when you await, returning execution control to the caller. In Akka .NET, this means that the next message will begin processing while the current one hasn’t finished yet.

async void methods should be restricted to methods at the beginning of the call chain (such as event handlers and WPF command handlers). You can’t sneak asynchrony into an otherwise synchronous call stack by introducing an async void. If you do async, it has to be all the way.

So it really seems that the problem with having asynchronous actor logic is simply that Akka .NET was never really designed to work with asynchronous methods.

Asynchrony in Akka .NET with PipeTo()

It should be clear by now that doing async/await in actors is not an option. So how do we go about doing our asynchronous work? We do that by using the PipeTo() pattern (because in Akka .NET, everything is called a pattern).

Let’s go back to our original example with the BusyActor. We left off with this code:

    public class BusyActor : UntypedActor
    {
        protected override async void OnReceive(object message)
        {
            Console.WriteLine($"Begin processing {message.ToString()}");

            await Task.Delay(2000);

            Console.WriteLine($"End processing {message.ToString()}");
        }
    }

Now, we need to refactor this to do the asynchronous operation (in this case Task.Delay()) in a fire-and-forget manner, and send the result as a separate message to an actor. We’re going to need separate messages for this:

    public class TaskMessage
    {
        public string Message { get; }

        public TaskMessage(string message)
        {
            this.Message = message;
        }

        public override string ToString()
        {
            return this.Message;
        }
    }

    public class ResultMessage
    {
        public string Message { get; }

        public ResultMessage(string message)
        {
            this.Message = message;
        }

        public override string ToString()
        {
            return this.Message;
        }
    }

Since our message handling is going to grow a little, UntypedActor is no longer suitable for what we need. Instead, we’ll refactor BusyActor as follows:

    public class BusyActor : ReceiveActor
    {
        public BusyActor()
        {
            this.Receive<TaskMessage>(m => Handle(m));
            this.Receive<ResultMessage>(m => Handle(m));
        }

        public void Handle(TaskMessage message)
        {
            Console.WriteLine($"Begin processing {message.ToString()}");

            Task.Delay(2000)
                .ContinueWith(x => new ResultMessage(message.Message),
                    TaskContinuationOptions.AttachedToParent
                    & TaskContinuationOptions.ExecuteSynchronously)
                .PipeTo(Self);
        }

        public void Handle(ResultMessage message)
        {
            Console.WriteLine($"End processing {message.ToString()}");
        }
    }

Similarly to the official example (which shows how to do an HTTP GET request within an actor), we are firing off an asynchronous request but not awaiting it. This happens in fire-and-forget manner as far as the actor is concerned. When the asynchronous operation is done, we create a new message and send it to the same actor so that he can log the end of the task.

Note that we have those two TaskContinuationOptions settings. You can read more about them in the official PipeTo() blog post, but the point I want to make here is that you need to remember to include them, and this makes this approach pretty error-prone.

Back in our main program, we need to send a TaskMessage instead of a simple string now:

        static void Main(string[] args)
        {
            using (var actorSystem = ActorSystem.Create("MyActorSystem"))
            {
                var actor = actorSystem.ActorOf(Props.Create<BusyActor>(), "BusyActor");

                actor.Tell(new TaskMessage("Task 1"));
                actor.Tell(new TaskMessage("Task 2"));
                actor.Tell(new TaskMessage("Task 3"));

                Console.ReadLine();
            }
        }

Let us now run this code:

This is bad. Even with PipeTo(), we still have the same interleaving problem as before. If you think about it, it makes sense.

What we are doing is firing off a fire-and-forget task, and the method can return immediately, thus allowing the next message to be processed before the asynchronous operation has completed. This is exactly the same problem we had when using async void.

If you’re firing off an asynchronous operation that doesn’t touch anything else and you just want to take its result, then the suggested PipeTo() approach will work. But if you need a guarantee on message order because your message processing is touching some state (thus an older message might overwrite the results of a newer message), then this is going to be a problem.

Coupling and Cohesion

Another problem with using PipeTo() is that it… complicates things. You can already see how our original example has been bloated into something a lot less easy to work with, just for the sake of not using async/await. But there’s more.

One common pitfall I see when developers begin to understand the importance of decoupled software is that they go to the other extreme: they split up components into extremely granular classes. In doing so, they breaking the companion principle of coupling: cohesion. While coupling dictates that software components should have minimal dependencies between themselves, cohesion suggests that components with strong direct interrelations should work closely together. Making classes too granular, for instance, is another way to end up with messy software.

At the beginning of this article, I quoted Natan Vivo’s comment about having to break a database operation into multiple operations. Typically, in ADO .NET, a database operation would look something like this:

1. Open a connection to the database.
2. Execute a command (query, nonquery, etc) against the database.
3. In case of a query, iterate over the rows and do something with them.

Each of the three operations above can be done asynchronously in sequence. They are meant to be together because they are part of the same cohesive operation. But if you break each of these operations into different messages and different message handlers, you’re going to scatter this otherwise contiguous operation all over the place. And that makes software a lot harder to maintain.

So when I see something like (to again quote the questions from the official Akka .NET PipeTo() sample) this:

“So just don’t do it. Await is evil inside an actor. Await is just syntactic sugar anyway. Use ContinueWith and PipeTo instead.”

…I feel the need to remind people that syntactic sugar is really important to make our software easier to write, but more importantly, easier to maintain.

For the reasons outlined above, I believe that the PipeTo() ‘pattern’ is really an anti-pattern, and I appeal for native asynchronous support in Akka .NET rather than quirky workarounds.