Akka.NET: Creating a File Download Actor with Progress Reporting

In this post, I walk through the creation of a single Actor for Akka.NET. It will download files and report its progress to other Actors.

I'll go through all details but might skim over some of the basics of Akka.NET.

Actor Model

Here's a short introduction.

Each "actor" in the system has a clear responsibility. Actors communicate by sending and receiving immutable messages.

There can be many instances of a certain Actor. An Actor is cheap. They have an internal state, but do not share state with other Actors, or anything else.

This allows Akka.NET to run your code in parallell. At the same time the code you write in Actors runs sequentially, so you never have to worry about locks of some sort. Since the only state you change is internal and the code inside a single actor is synchronous, you will be fine.

The Download Actor

To put it in practice, we'll try to make an Actor that has one simple task: to download something from a URL.

It will:

Accept multiple download requests (executed sequentially)
Run a single download at a time
Report start, end and progress in the form of a Message to the requestor

The basics

First, we create an Actor called FileDownloadActor

public class FileDownloadActor : ReceiveActor
{
}

Then we can define the messages. We'll define one incoming message RequestDownload and three outgoing messages DownloadStarted, DownloadCompleted and DownloadProgressed:

public class FileDownloadActor : ReceiveActor
{
    #region Messages

    public class RequestDownload
    {
        public Uri Uri { get; private set; }
        public RequestDownload(Uri uri) { Uri = uri; }
    }

    public class DownloadStarted
    {
        public Uri Uri { get; private set; }
        public DownloadStarted(Uri uri) { Uri = uri; }
    }

    public class DownloadCompleted
    {
        public Uri Uri { get; private set; }
        public string Path { get; private set; }
        public DownloadCompleted(Uri uri, string path)
        {
            Uri = uri;
            Path = path;
        }
    }

    public class DownloadProgressed
    {
        public Uri Uri { get; private set; }
        public double Progress { get; private set; }
        public DownloadProgressed(Uri uri, double progress)
        {
            Uri = uri;
            Progress = progress;
        }
    }

    #endregion
}

Note that we put the message classes inside the Actor class. This is not a requierment but I find it adds clarity.

So everything starts with a RequestDownload, providing only a Uri.

When the downloading starts the requestor will be informed of this. The DownloadProgressed messages also includes the progress.

When the download completes, the path of the downloaded file is also provided. This is actually shared state between two Actors (the DownloadActor and the Sender). It's assumed the two actors can both access that path. But sending a large byte array in a message is bad for performance.

Keeping State

The DownloadActor will have to keep some state:

The Uri that is currently being downloaded
The path of the file that is being written
A reference to the requester of the current download, which is another Actor

There's some other stuff we'll need, but we can introduce that later. So we'll need a class member field for each of these items.

public class FileDownloadActor : ReceiveActor
{
    #region State

    private Uri _currentUri;
    private string _currentTargetPath;
    private IActorRef _currentDownloadRequestor;

    #endregion

    #region Messages
    ...
    #endregion
}

Initialization

Actors are always in a certain state. The state defines . In our case the DownloadActor can be ready to accept new download requests, or can be downloading.

Here's the states defined for the DownloadActor:

public class FileDownloadActor : ReceiveActor
{
    #region State

    ...

    #endregion

    #region Initialization

    public FileDownloadActor()
    {
        Become(Ready);
    }

    #endregion

    #region Messages

    ...

    #endregion

    #region States

    public void Ready()
    {
        Receive<RequestDownload>(message =>
        {
            HandleDownloadRequest(message);
            Become(Downloading);
        });
    }

    public void Downloading()
    {
        // TODO: Handle incoming Download Requests while downloading
        // TODO: Go back to Ready state when a download completes
    }

    #endregion
}

I will come back to the TODO's shortly, but first let's implement the HandleDownloadRequest method.

public class FileDownloadActor : ReceiveActor
{
    #region State

    ...

    #endregion

    #region Messages

    ...

    #endregion

    #region States

    ...

    #endregion

    #region Handlers

    private void HandleDownloadRequest(RequestDownload message)
    {
        _currentUri = message.Uri;
        _currentTargetPath = Path.GetTempFileName();
        _currentDownloadRequestor = Sender;

        _currentDownloadRequestor.Tell(new DownloadStarted(_currentUri));
    }

    #endregion
}

Note, in the message handler, we have access to the "Sender" property.

Doing the Download

A naive implementation could be the following. Careful though, in the next steps we'll be improving a lot of things.

public class FileDownloadActor : ReceiveActor
{
    #region State

    ...

    #endregion

    #region Messages

    ...

    #endregion

    #region States

    ...

    #endregion

    #region Handlers

    private void HandleDownloadRequest(RequestDownload message)
    {
        ...

        StartDownload();
    }

    #endregion

    #region Helper Functions

    private void StartDownload()
    {
        var client = new WebClient();

        // Careful, this is not the best way to do this!
        client.DownloadFile(_currentUri, _currentTargetPath);
        _currentDownloadRequestor.Tell(new DownloadCompleted(_currentUri, _currentTargetPath));
        Become(Ready);
    }

    #endregion
}

Testing

Currently, we have a basic Download Actor, with a lot of problems still, but we have the Actor, so we could start testing it.

Due to the parallell nature of Akka.NET, testing in Unit Test frameworks can be a bit tricky. It's not hard, but it's a little different and I won't go into it here. I'll give you a hint: use Akka.Test.

For now, we'll just test it in a simple Console application.

class Program
{
    static void Main(string[] args)
    {
        var system = ActorSystem.Create("downloader");

        var props = Props.Create(() => new FileDownloadActor());
        var actor = system.ActorOf(props);

        var requestDownload = new FileDownloadActor.RequestDownload(new Uri("http://getakka.net/images/akkalogo.png"));
        actor.Tell(requestDownload);

        Console.ReadLine();
    }
}

Issue with the current implementation

As said, there are quite some issues here:

How do we know when the download is completed and where the file is. Who handles the started, progress and completed messages?
The download action is a long-running synchronous operation. That's not a good thing in Akka. The action should be asynchronous.
While the actor is downloading, it ignores other incoming requests

Creating a TestActor

To solve the first issue, we can create a simple test actor that writes something on the console for each received message. The test actor will have a reference to a DownloadActor:

public class TestActor : ReceiveActor
{
    private readonly IActorRef _downloadActor;
    private readonly Uri _testUri;

    public TestActor(IActorRef downloadActor, Uri testUri)
    {
        _downloadActor = downloadActor;
        _testUri = testUri;

        Become(Ready);
    }

    public class StartTest { };

    public void Ready()
    {
        Receive<StartTest>(message => HandleStartTest(message));
        Receive<FileDownloadActor.DownloadStarted>(message => HandleDownloadStarted(message));
        Receive<FileDownloadActor.DownloadProgressed>(message => HandleDownloadProgressed(message));
        Receive<FileDownloadActor.DownloadCompleted>(message => HandleDownloadCompleted(message));
    }

    private void HandleStartTest(StartTest message)
    {
        Console.WriteLine($"Starting test for Uri '{_testUri}'");
        _downloadActor.Tell(new FileDownloadActor.RequestDownload(_testUri));
    }

    private void HandleDownloadStarted(FileDownloadActor.DownloadStarted message)
    {
        Console.WriteLine($"Download started Uri '{_testUri}'");
    }

    private void HandleDownloadProgressed(FileDownloadActor.DownloadProgressed message)
    {
        Console.WriteLine($"Download progressed for Uri '{_testUri}': {message.Progress}");
    }

    private void HandleDownloadCompleted(FileDownloadActor.DownloadCompleted message)
    {
        Console.WriteLine($"Download completed for Uri '{_testUri}'.");
    }
}

Usage is like so:

class Program
{
    static void Main(string[] args)
    {
        var system = ActorSystem.Create("downloader");

        var props = Props.Create(() => new FileDownloadActor());
        var actor = system.ActorOf(props);
        var testProps = Props.Create(() => new TestActor(actor, new Uri("http://getakka.net/images/akkalogo.png")));
        var testActor = system.ActorOf(testProps);
        testActor.Tell(new TestActor.StartTest());

        Console.ReadLine();
    }
}

The output of this application should now be:

Starting test for Uri 'http://getakka.net/images/akkalogo.png'.
Download started Uri 'http://getakka.net/images/akkalogo.png'.
Download completed for Uri 'http://getakka.net/images/akkalogo.png'.

Handle Asynchronous Downloads

There are several ways to make the WebClient do asynchornous downloads:

WebClient.DownloadFileAsync: uses callbacks
WebClient.DownloadFileTaskAsync: use the Task Parallell Library

Here, we'll use the DownloadFileAsync. For an example of using Task inside Actors, check out this example: Akka.NET PipeTo Sample.

The problem with the PipeTo approach is that our async process gives feedback when it's not done yet too. Getting progress reporting to work then seemed quite hard.

With callbacks, it's easier. We just handle the two callbacks (progress and success).

A naive approach - that actually works - is the following:

    private void StartDownload()
    {
        var client = new WebClient();

        client.DownloadProgressChanged += HandleWebClientDownloadProgressChanged;
        client.DownloadFileCompleted += HandleWebClientDownloadCompleted;
        client.DownloadFileAsync(_currentUri, _currentTargetPath);
        Become(Ready);
    }

    private void HandleWebClientDownloadProgressChanged(object sender, DownloadProgressChangedEventArgs e)
    {
        _currentDownloadRequestor.Tell(new DownloadProgressed(_currentUri, e.ProgressPercentage));
    }

    private void HandleWebClientDownloadCompleted(object sender, AsyncCompletedEventArgs e)
    {
        _currentDownloadRequestor.Tell(new DownloadCompleted(_currentUri, _currentTargetPath));
    }

The event handlers run outside the context of the actor. Using the state could be an issue. The best implementation would be to make the handlers send a simple message to the actor itself and access the state in those message handlers.

The output should now be something like this:

Starting test for Uri 'http://getakka.net/images/akkalogo.png'.
Download started Uri 'http://getakka.net/images/akkalogo.png'.
Download progressed for Uri 'http://getakka.net/images/akkalogo.png': 100.
Download progressed for Uri 'http://getakka.net/images/akkalogo.png': 100.
Download completed for Uri 'http://getakka.net/images/akkalogo.png'.

It seems like WebClient fires the progress event twice. But that's not an issue.

Queueing Requests

In the current implementation requests that come in when the Actor is already downloading are ignored.

Luckily, Akka.NET has a way to "stash" these messages for later use. We have to do a few steps to make this work:

Implement the IWithUnboundedStash in the Actor
While downloading, push incoming RequestDownload messages on the stash
When the download is completed make the stash send the queued messages

Implement the `IWithUnboundedStash` in the Actor

public class FileDownloadActor : ReceiveActor, IWithUnboundedStash
{
    ...
    public IStash Stash { get; set; }
    ...
}

While downloading, push incoming `RequestDownload` messages on the stash

public class FileDownloadActor : ReceiveActor, IWithUnboundedStash
{
    ...

    #region States

    ...

    public void Downloading()
    {
        Receive<RequestDownload>(message => Stash.Stash());
        // TODO: Go back to Ready state when a download completes
    }

    #endregion

    ...
}

When the download is completed make the stash send the queued messages

public class FileDownloadActor : ReceiveActor, IWithUnboundedStash
{
    ...

    #region States

    ...

    public void Downloading()
    {
        Receive<RequestDownload>(message => Stash.Stash());
        Receive<DownloadCompleted>(message => {
            Become(Ready);
            Stash.UnstashAll();
        });
    }

    #endregion

    ...
}

Now, the output is something like the following, which shows the final problem:

Starting test for Uri 'http://getakka.net/images/akkalogo.png'.
Download started Uri 'http://getakka.net/images/akkalogo.png'.
Download progressed for Uri 'http://getakka.net/images/akkalogo.png': 100.
Download progressed for Uri 'http://getakka.net/images/akkalogo.png': 100.
Download completed for Uri 'http://getakka.net/images/akkalogo.png'.

The Actor itself doesn't know yet when to come out of the "Downloading" state.

To do that, it has to send itself a DownloadCompleted message. A naive way - that doesn't work this time - could be the following:

public class FileDownloadActor : ReceiveActor, IWithUnboundedStash
{
    ...

    #region Helper Functions

    ...

    private void HandleWebClientDownloadCompleted(object sender, AsyncCompletedEventArgs e)
    {
        _currentDownloadRequestor.Tell(new DownloadCompleted(_currentUri, _currentTargetPath));
        Self.Tell(new DownloadCompleted(_currentUri, _currentTargetPath));
    }

    #endregion
}

Running this produces the following exception:

An unhandled exception of type 'System.NotSupportedException' occurred in Akka.dll

Additional information: There is no active ActorContext, this is most likely due to use of async operations from within this actor.

We cannot use the Self property, but we have to send ourselves a message somehow.

However we can create a field _self and assign it the Self property.

public class FileDownloadActor : ReceiveActor, IWithUnboundedStash
{
    #region State

    ...

    private IActorRef _self;

    ...

    #endregion

    #region Initialization

    public FileDownloadActor()
    {
        _self = Self;
        Become(Ready);
    }

    #endregion

    ...

    #region Helper Functions

    ...

    private void HandleWebClientDownloadCompleted(object sender, AsyncCompletedEventArgs e)
    {
        _currentDownloadRequestor.Tell(new DownloadCompleted(_currentUri, _currentTargetPath));
        _self.Tell(new DownloadCompleted(_currentUri, _currentTargetPath));
    }

    #endregion
}

This will make the Actor able to handle multiple requests, sequentially.

Taking Advantage of Actor's single threaded nature

Because we know the Actor will do one download at a time, we can re-use a single instance of the WebClient class.

public class FileDownloadActor : ReceiveActor, IWithUnboundedStash
{
    #region State

    ...

    private readonly WebClient _client = new WebClient();

    ...

    #endregion

    #region Initialization

    public FileDownloadActor()
    {
        _self = Self;
        _client.DownloadProgressChanged += HandleWebClientDownloadProgressChanged;
        _client.DownloadFileCompleted += HandleWebClientDownloadCompleted;
        Become(Ready);
    }

    #endregion

    ...

    #region Helper Functions

    private void StartDownload()
    {
        _client.DownloadFileAsync(_currentUri, _currentTargetPath);
        Become(Ready);
    }

    ...

    #endregion
}

Running in Parallell

You might be thinking that this is very much not parallell. After all, the Download Actor only downloads one URI at a time.

That is true, but we could create a "pool" of such download actors and use them simultaniously. In fact, Akka.NET makes that a one-liner, as shown here:

In our test application we have the following code to create our Download Actor:

    var props = Props.Create(() => new FileDownloadActor());
    var actor = system.ActorOf(props);

The actor is then passed to the TestActor who sends a number of download requests like so:

    private void HandleStartTest(StartTest message)
    {
        Console.WriteLine($"Starting test for Uri '{_testUri}'.");
        _downloadActor.Tell(new FileDownloadActor.RequestDownload(_testUri));
        _downloadActor.Tell(new FileDownloadActor.RequestDownload(_testUri));
        _downloadActor.Tell(new FileDownloadActor.RequestDownload(_testUri));
    }

Running this will show the three downloads happen in sequence.

To allow multiple download requests to be handled simutaniously, we can simply add a WithRouter statement to the Actor creation:

        var props = Props.Create(() => new FileDownloadActor()).WithRouter(new RoundRobinPool(10));
        var actor = system.ActorOf(props);

This is also why IActorRef is passed around and not just and instance of actor: IActorRef can refer to one instance or even a pool of instances of an Actor, optionally spread over multiple machines when that makes sense. This is where using Actors starts to pay off. It handles some of the complexity of running things in parallell.

Another approach could be to create a dispatching Actor which will create one Download Actor per hostname, for example.

Open Issues

There are some things that can improve this Actor:

If, for some reason, the same Uri is requested twice or more by the same Actor, it will be unclear to them which of the requests was completed. This is because they only get the Uri. Better would be if the requested would provide a unique ID to each request. For example a Guid.
When a download goes wrong, the Actor should inform the requester. Either by a message or by simply crashing and using Akka.NET's supervision capabilities.
I'm not a 100% sure of keeping a reference to Self. Since Self is an IActorRef, and IActorRef doesn't guarantee to point that instance of the Actor class I'm not sure it's fool proof.

Conclusion

With this post, I wanted to explain to you, and to myself, how to create a Download Actor that handles asynchronous calls inside itself. Also, I hope it introduces Akka.NET in a practical way.

I find it an enjoyable and different way to think about applications. And it takes away a lot of pain from parallell programming in .NET ("threading").

Even if you don't create massively distributed applications, Akka.NET could be used in small scale applications, just to abstract away the technical details of running things in parallell. Which is exactly what I'll be trying with our download actor.

Akka.NET: Creating a File Download Actor with Progress Reporting

Actor Model

The Download Actor

The basics

Keeping State

Initialization

Doing the Download

Testing

Issue with the current implementation

Creating a TestActor

Handle Asynchronous Downloads

Queueing Requests

Implement the `IWithUnboundedStash` in the Actor

While downloading, push incoming `RequestDownload` messages on the stash

When the download is completed make the stash send the queued messages

Taking Advantage of Actor's single threaded nature

Running in Parallell

Open Issues

Conclusion

FlacLibSharp Found a User

Integrating WPF and Akka.NET

Actor Model

The Download Actor

The basics

Keeping State

Initialization

Doing the Download

Testing

Issue with the current implementation

Creating a TestActor

Handle Asynchronous Downloads

Queueing Requests

Implement the IWithUnboundedStash in the Actor

While downloading, push incoming RequestDownload messages on the stash

When the download is completed make the stash send the queued messages

Taking Advantage of Actor's single threaded nature

Running in Parallell

Open Issues

Conclusion

Implement the `IWithUnboundedStash` in the Actor

While downloading, push incoming `RequestDownload` messages on the stash