Daily Archives: August 8, 2016

Software development

Consistent Hashing in Akka .NET

7 Comments

The source code for this article is available at the Gigi Labs BitBucket repository.

Routers

In Akka .NET, a router is like a load balancer. It is a special actor that does not handle messages itself, but passes them on to other actors who can handle them. For this reason routers are the only kind of actor that can deal with several messages concurrently (whereas normal actors process messages sequentially, one by one).

The way routers forward messages to handling actors depends on the type of router you use. Some common routing strategies include broadcast, round robin, and random. In this article, we will deal with the ConsistentHashing router. Consistent hashing means that messages with the same (arbitrarily defined) key are always handled by the same actor.

Another important distinction between routers is that they fall under two categories: Group routers and Pool routers. “Pool” means the same as in the context of “Thread Pool” or “Connection Pool”: it is a dynamic set of resources that can adaptively grow and shrink as needed. A Pool router creates the actors that it will forward messages to for handling, and as such, it also supervises them. Group routers, on the other hand, passed a set of actors that are created beforehand. As such the handling actors are fixed in number and detached from the router; the Group router does not supervise them and often does not know when they die. For this reason Pool routers are preferred for most use cases.

There is a lot more to be said about routers. However, this section is intended only as a brief background. For more comprehensive references, check the links in the Further Reading section.

Consistent Hashing Example with Currency Pairs

In the financial industry, currency exchange is defined in terms of a currency pair, such as EUR/USD. This currency pair has a price, such as 1.1087. This means that 1 Euro is worth 1.1087 US Dollars. The currency exchange market is very volatile, and these prices can change several times per second.

In our example, we will be generating fictitious currency prices. We would like to have a pool of actors to handle these price updates. We would also like each currency pair to be always be handled by the same actor.

As always, we first need to install the Akka NuGet package:

Install-Package Akka

Then, in our Main() method, we will first add some trivial setup code:

            Console.Title = "Akka .NET Consistent Hashing";

            var random = new Random();
            var currencyPairs = new string[]
            {
                "EUR/GBP",
                "USD/CAD",
                "NZD/JPY",
                "EUR/USD",
                "USD/JPY",
                "NZD/EUR"
            };

Our program logic goes like this:

            using (var actorSystem = ActorSystem.Create("MyActorSystem"))
            {
                var pool = new ConsistentHashingPool(3);
                var props = Props.Create<CurrencyPriceChangeHandlerActor>()
                    .WithRouter(pool);
                var router = actorSystem.ActorOf(props, "MyPool");

                for (int i = 0; i < 20; i++)
                    SendRandomMessage(router, random, currencyPairs);

                Console.ReadLine();
            }

Here we’re setting up a Pool router using the Consistent Hashing strategy. A pool of 3 actors will be created, supervised by the router. We can send a message to the router as we would with any other actor, but it will actually be handled by one of its child actors.

The child actors are of type CurrencyPriceChangeHandlerActor. This type of actor simply writes the received message to the console, along with its own path so that we can distinguish between the child actors. The path is dynamically generated by the Pool router and we have no control over it.

    public class CurrencyPriceChangeHandlerActor
        : TypedActor, IHandle<CurrencyPriceChangeMessage>
    {
        public CurrencyPriceChangeHandlerActor()
        {
            
        }

        public void Handle(CurrencyPriceChangeMessage message)
        {
            Console.WriteLine($"{Context.Self.Path} received: {message}");
        }
    }

The message handled by this type of actor is a simple combination of currency pair and price. In line with best practices, the message is immutable by design. More importantly, it implements IConsistentHashable. This allows us to provide a key that will be used for the consistent hashing algorithm. In our case, the key is the currency pair.

    public class CurrencyPriceChangeMessage : IConsistentHashable
    {
        public string CurrencyPair { get; }
        public decimal Price { get; }

        public object ConsistentHashKey
        {
            get
            {
                return this.CurrencyPair;
            }
        }

        public CurrencyPriceChangeMessage(string currencyPair, decimal price)
        {
            this.CurrencyPair = currencyPair;
            this.Price = price;
        }

        public override string ToString()
        {
            return $"{this.CurrencyPair}: {this.Price}";
        }
    }

Note: this is just one of three ways how we can specify the key to use with consistent hashing. Refer to the documentation for more information.

All we have left is the implementation of SendRandomMessage(). It picks a random currency pair and a random price, and sends a message. It also introduces a random delay between each message. Without this delay, you’ll see a lot of the same currency pairs in sequence.

        private static void SendRandomMessage(IActorRef router, Random random,
            string[] currencyPairs)
        {
            var randomDelay = random.Next(100, 1500);
            var randomCurrencyId = random.Next(0, currencyPairs.Length);
            var randomPrice = Convert.ToDecimal(random.NextDouble());

            var currencyPair = currencyPairs[randomCurrencyId];

            var message = new CurrencyPriceChangeMessage(
                currencyPair, randomPrice);
            router.Tell(message);

            Thread.Sleep(TimeSpan.FromMilliseconds(randomDelay));
        }

Here’s what we get when we run the program:

akkanet-consistenthashing-output

You can see how although all three handling actors are in use, there is a direct correspondence between the currency pair and the actor that handles it. For example, USD/CAD is always handled by actor $c, whereas NZD/EUR is always handled by actor $b. This is what is implied by consistent hashing.

As far as I can tell, control of which actors handle which keys is up to the router. I would have liked to, for instance, create an actor to specifically handle each currency pair. But I don’t think that is possible, even with Group routers (correct me if I’m wrong). Just let the router worry about how to allocate the keys to the actors.

Further Reading