ZeroMQ #7: A Simple Actor Model

Last time we looked at using ZeroMQ to use a “Divide And Conquer” pattern to distribute work to a number of workers and then combine the results again.

Since I wrote that last post I have had a bit of think about this series of posts, and realised that nothing I can say here would be as good or as thorough as the guide, so I have has to rethink my strategy a bit for the posts that I may write on ZeroMQ from here on in. So rather than me regurgitate what has already been said by Pieter on the guide web site, I will instead only be writing about stuff that I think is new, or worthy of a post. Now this could mean that the posts are less frequent, but I hope when there is one it will be of more interest, than me just saying here is a NetMQ version of the “Paranoid Pirate Pattern”, go check this link at the guide for more information.

So where does that leave this series of posts? Well to be honest slightly in limbo, but I have also been in contact with Pieter Hintjens, who was kind enough to give me a little push into something that may be of interest.

Pieter notified by of a Actor Model that was part of the high level C library for ZeroMQ called “czmq”, which is not contained in the NetMQ GitHub repository. So I had a call with Pieter, and looked into that.

This post will discuss a very simple actor model, that I have written to work with NetMQ, Pieter has given it the once over, and I have also talked it through with a regular ZeroMQ user at work, so I think it an ok version of the original C ZeroMQ “czmq” version.

Where Is The Code?

As always before we start, it’s only polite to tell you where the code is, and it is as before on GitHub:

https://github.com/sachabarber/ZeroMqDemos

What Is An Actor Model?

Here is what Wikipedia has to same in the introduction to what an Actor Model is.

The actor model in computer science is a mathematical model of concurrent computation that treats “actors” as the universal primitives of concurrent digital computation: in response to a message that it receives, an actor can make local decisions, create more actors, send more messages, and determine how to respond to the next message received.

….

….

The Actor model adopts the philosophy that everything is an actor. This is similar to the everything is an object philosophy used by some object-oriented programming languages, but differs in that object-oriented software is typically executed sequentially, while the Actor model is inherently concurrent.

An actor is a computational entity that, in response to a message it receives, can concurrently:

  • send a finite number of messages to other actors;
  • create a finite number of new actors;
  • designate the behavior to be used for the next message it receives.

There is no assumed sequence to the above actions and they could be carried out in parallel.

Decoupling the sender from communications sent was a fundamental advance of the Actor model enabling asynchronous communication and control structures as patterns of passing messages.[8]

Recipients of messages are identified by address, sometimes called “mailing address”. Thus an actor can only communicate with actors whose addresses it has. It can obtain those from a message it receives, or if the address is for an actor it has itself created.

The Actor model is characterized by inherent concurrency of computation within and among actors, dynamic creation of actors, inclusion of actor addresses in messages, and interaction only through direct asynchronous message passing with no restriction on message arrival order.

http://en.wikipedia.org/wiki/Actor_model

 

How I like to think of Actors is that they may be used to alleviate some of synchronization concerns of using shared data structures. This is achieved by your application code talking to actors via message passing/receiving. The actor itself may pass messages to other actors, or work on the passed message itself. By using message passing rather than using shared data structures, it may help to think of the actor (or any subsequent actors its send messages to) working on a copy of the data rather than working on the same shared structures. Which kind of gets rid of the need to worry about nasty things like lock(s) and any nasty timing issues that may arise from carrying out multi threaded code. If the actor is working with its own copy of the data then we should have no issues with other threads wanting to work with the data  the actor has, as the only place that data can be is within the actor itself, that is unless we pass another message to a different actor. If we were to do that though the new message to the other actor would also be a copy of the data, so would also be thread safe.

I hope you see what I am trying to explain there, may be a diagram may help.

Multi Threaded Application Using Shared Data Structure

A fairly common thing to do is have multiple threads running to speed things up, but then you realise that your threads need to mutate the state of some shared data structure, so then you have to involve threading synchronization primitives (most commonly lock(..) statements, to create your user defined critical sections). This will work, but now you are introducing artificial delays due to having to wait for the lock to be released so you can run Thread X’s code.

image 

To take this one step further, lets see some code that may illustrate this further, imagine we had this sort of data structure representing a very slim bank account

namespace ConsoleApplication1
{
    public class Account
    {

        public Account()
        {

        }

        public Account(int id, string name,
            string sortCode, decimal balance)
        {
            Id = id;
            Name = name;
            SortCode = sortCode;
            Balance = balance;
        }

        public int Id { get; set; }
        public string Name { get; set; }
        public string SortCode { get; set; }
        public decimal Balance { get; set; }
    }
}

 

Nothing fancy there, just some fields. So lets now move onto looking at some threading code, I have chosen to just show two threads acting on a shared Account instance.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Management.Instrumentation;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ConsoleApplication1
{
    class Program
    {
        private object syncLock = new object();
        private Account clientBankAccount;
        public Program()
        {
            clientBankAccount = new Account(1,"sacha barber","112233",0);
        }

        public async Task Run()
        {
            try
            {
                await Task.Run(() =>
                {
                    Console.WriteLine("Tread Id {0}, Account balance before: {1}",
                        Thread.CurrentThread.ManagedThreadId, clientBankAccount.Balance);
                        
                    lock (syncLock)
                    {
                        Console.WriteLine("Tread Id {0}, Adding 10 to balance",
                           Thread.CurrentThread.ManagedThreadId);
                        clientBankAccount.Balance += 10;
                        Console.WriteLine("Tread Id {0}, Account balance before: {1}",
                            Thread.CurrentThread.ManagedThreadId, clientBankAccount.Balance);
                    }
                });

                await Task.Run(() =>
                {
                    Console.WriteLine("Tread Id {0}, Account balance before: {1}",
                        Thread.CurrentThread.ManagedThreadId, clientBankAccount.Balance);
                    lock (syncLock)
                    {
                        Console.WriteLine("Tread Id {0}, Subtracting 4 to balance",
                           Thread.CurrentThread.ManagedThreadId);
                        clientBankAccount.Balance -= 4;
                        Console.WriteLine("Tread Id {0}, Account balance before: {1}",
                            Thread.CurrentThread.ManagedThreadId, clientBankAccount.Balance);
                    }
                });
            }
            catch (Exception e)
            {
                Console.WriteLine(e);
            }

        }

        static void Main(string[] args)
        {
           Program p = new Program();
           p.Run().Wait();
           Console.ReadLine();
        }
    }
}

 

I have possible picked an example that you think may not actually happen in real life, and to be honest this scenario may not popup in real life, as who would do something as silly as crediting an account in one thread, and debiting it in another…we are all diligent developers, we would not let this one into the code would we?

To be honest whether the actual example has real world merit or not, the point remains the same, since we have more than one thread accessing a shared data structure, access to it must be synchronized, which is typically done using a lock(..) statement, as can be seen in the code.

Now don’t get me wrong the above code does work, as shown in the output below:

image

Perhaps there might be a more interesting way though!

Actor Model

The actor model, takes a different approach, where by message passing is used, which may involve some form of serialization as the messages are pass down the wire, which kind of guarantees no shared structures to contend with. Now I am not saying all Actor frameworks use message passing down the wire (serialization) but the code presented in this article does.

The basic idea is that each thread would talk to an Actor, and send/receive message with the actor.

If you wanted to get even more isolation, you could use thread local storage where each thread could have its own copy of the actor which it, and it alone talks to.

image

Anyway enough talking, I am sure some of you want to see the code right?

The Implementation

The idea is that the Actor itself may be treated like a ZeroMQ (NetMQ in my case) sockets, and may therefor be used to Send/Receive messages. Now you may be wondering if I send the Actor a message, who is listening to that message, and how is it that I am able to receive a message from the Actor?

The answer to that lies inside the implementation of the simple Actor framework in this post. Internally the Actor spins up another thread. Within that new thread is one end of a PairSocket where the Actor itself is the other end of the pipe which is also a PairSocket (recall I said the Actor is able to act as a socket). The Actor and the other end of the pipe communicate via message passing, and they use an in process (inproc) protocol to do so.

The initial message passed to the Actor forms some some of protocol  that both the other end of the Actor pipe (i.e. the thread the Actor created ZeroMQ czmq implementation), which I am calling the “shim” (borrowed from the) MUST know how to deal with the protocol that the user code sends via the Actor.

The way I have chosen to do this, is you (i.e. the user of the simple Actor library will need to create a “shim” handler class. This “shim” handler class may be a very simple protocol or a very complicated one (for the demo I have stuck to very simple ones), as long as it understands the message/command being sent from the Actor, and knows what to do with it. That is up to you to come up with, I have no silver bullet for that

One final thing to explain is that the “shim” handler may be passed some initial arguments (outside of the message passing) should you want to make use of this feature. It is something you may not want/need to use, but it is there should you want to use it.

Actor Code

Here is the code for the Actor itself, where it can be seen that it may be treated as a socket using the Send/Receive methods. The Actor also creates a new thread which is used to run the code in the shim handler. The Actor also creates the shim and the pair of PairSocket(s) for message passing.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using NetMQ;
using NetMQ.Sockets;
using NetMQ.zmq;

namespace NetMQActors
{
    /// <summary>
    /// The Actor represents one end of a two way pipe between 2 PairSocket(s). Where
    /// the actor may be passed messages, that are sent to the other end of the pipe
    /// which I am calling the "shim"
    /// </summary>
    public class Actor : IOutgoingSocket, IReceivingSocket, IDisposable
    {
        private readonly PairSocket self;
        private readonly Shim shim;
        private Random rand = new Random();
        private CancellationTokenSource cts = new CancellationTokenSource();

        private string GetEndPointName()
        {
            return string.Format("inproc://zactor-{0}-{1}",
                rand.Next(0, 10000), rand.Next(0, 10000));
        }

        public Actor(NetMQContext context, IShimHandler shimHandler, object[] args)
        {
            this.self = context.CreatePairSocket();
            this.shim = new Shim(shimHandler, context.CreatePairSocket());
            this.self.Options.SendHighWatermark = 1000;
            this.self.Options.SendHighWatermark = 1000;

            //now binding and connect pipe ends
            string endPoint = string.Empty;
            while (true)
            {
                Action bindAction = () =>
                {
                    endPoint = GetEndPointName();
                    self.Bind(endPoint);
                };

                try
                {
                    bindAction();
                    break;
                }
                catch (NetMQException nex)
                {
                    if (nex.ErrorCode == ErrorCode.EFAULT)
                    {
                        bindAction();
                    }
                }

            }

            shim.Pipe.Connect(endPoint);

            //Create Shim thread handler
            CreateShimThread(args);
        }

        private void CreateShimThread(object[] args)
        {
            Task shimTask = Task.Factory.StartNew(
                (state) => this.shim.Handler.Run(this.shim.Pipe, (object[])state, cts.Token),
                args,
                cts.Token,
                TaskCreationOptions.LongRunning,
                TaskScheduler.Default);

            shimTask.ContinueWith(ant =>
            {
                if (ant.Exception == null) return;

                Exception baseException = ant.Exception.Flatten().GetBaseException();
                if (baseException.GetType() == typeof (NetMQException))
                {
                    Console.WriteLine(string.Format("NetMQException caught : {0}",
                        baseException.Message));
                }
                else if (baseException.GetType() == typeof (ObjectDisposedException))
                {
                    Console.WriteLine(string.Format("ObjectDisposedException caught : {0}",
                        baseException.Message));
                }
                else
                {
                    Console.WriteLine(string.Format("Exception caught : {0}",
                        baseException.Message));
                }
            }, TaskContinuationOptions.OnlyOnFaulted);
        }

        ~Actor()
        {
            Dispose(false);
        }

        public void Dispose(){
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        protected virtual void Dispose(bool disposing)
        {
            //cancel shim thread
            cts.Cancel();

            // release other disposable objects
            if (disposing)
            {
                if (self != null) self.Dispose();
                if (shim != null) shim.Dispose();
            }
        }

        public void Send(byte[] data, int length, bool dontWait = false, bool sendMore = false)
        {
            self.Send(data, length, dontWait, sendMore);
        }

        public byte[] Receive(bool dontWait, out bool hasMore)
        {
            return self.Receive(dontWait, out hasMore);
        }
    }
}

 

Shim Code

The shim represents the other end of the pipe, the shim essentially is a property bag, but it does hold a reference to the IShimHandler that the thread in the actual Actor will run. The IShimHandler is the one that MUST understand the protocol, and carry out any work.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

using NetMQ.Sockets;

namespace NetMQActors
{
    /// <summary>
    /// Shim represents one end of the in process pipe, where the Shim expects
    /// to be supplied with a <c>IShimHandler</c> that it would use for running the pipe
    /// protocol with the original Actor PairSocket the other end of the pipe
    /// </summary>
    public class Shim : IDisposable
    {
        public Shim(IShimHandler shimHandler, PairSocket pipe)
        {
            this.Handler = shimHandler;
            this.Pipe = pipe;
        }

        public IShimHandler Handler { get; private set; }
        public PairSocket Pipe { get; private set; }

        ~Shim()
        {
            Dispose(false);
        }

        public void Dispose(){
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        protected virtual void Dispose(bool disposing)
        {
            if (disposing)
            {
                // release disposable objects
                if (Pipe != null) Pipe.Dispose();
            }
        }
    }
}

 

Handler Interface

You shim handler code MUST implement this interface

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using NetMQ.Sockets;

namespace NetMQActors
{
    /// <summary>
    /// Simple interface that all shims should implement
    /// </summary>
    public interface IShimHandler
    {
        void Run(PairSocket shim, object[] args, CancellationToken token);
    }
}

An Example : Simple EchoShim Handler

This example shows how to create a simple echo shim handler that can be used with standard actor code above. The EchoShimHandler presented here, uses an EXTREMELY simple protocol is does the following:

  • It expects the initial arguments to be 1 in length
  • It expects the initial argument element 0 to be “Hello World”
  • It expects a multi part message, where the 1st message frame is the command string “ECHO”

If all of those criteria are satisfied, then the EchoShimHandler will write the its end of the PairSocket pipe. The user of the Actor at the other end of the pipe (i.e. the other PairSocket), can then receive the value from the EchoShimHandler. Remember in this mini library the Actor may act as a regular NetMQ socket.

Here is the code for the EchoShimHandler

using System;
using System.Linq;
using System.Threading;
using NetMQ;
using NetMQ.Sockets;
using NetMQ.zmq;

namespace NetMQActors
{
    /// <summary>
    /// This hander class is specific implementation that you would need
    /// to implement per actor. This essentially contains your commands/protocol
    /// and should deal with any command workload, as well as sending back to the
    /// other end of the PairSocket which calling code would receive by using the
    /// Actor classes various RecieveXXX() methods
    ///
    /// This is a VERY simple protocol but it just demonstrates what you would need
    /// to do to implement your own Shim handler
    /// </summary>
    public class EchoShimHandler : IShimHandler
    {
        public void Run(PairSocket shim, object[] args, CancellationToken token)
        {
            if (args == null || args.Count() != 1 || (string)args[0] != "Hello World")
                throw new InvalidOperationException(
                    "Args were not correct, expected 'Hello World'");

            while (!token.IsCancellationRequested)
            {
                //Message for this actor/shim handler is expected to be
                //Frame[0] : Command
                //Frame[1] : Payload
                //
                //Result back to actor is a simple echoing of the Payload, where
                //the payload is prefixed with "ECHO BACK "
                NetMQMessage msg = null;

                //this may throw NetMQException if we have disposed of the actor
                //end of the pipe, and the CancellationToken.IsCancellationRequested
                //did not get picked up this loop cycle
                msg = shim.ReceiveMessage();

                if (msg == null)
                    break;

                if (msg[0].ConvertToString() == "ECHO")
                {
                    shim.Send(string.Format("ECHO BACK : {0}",
                        msg[1].ConvertToString()));
                }
                else
                {
                    throw NetMQException.Create("Unexpected command",
                        ErrorCode.EFAULT);
                }
            }
        }
    }
}

 

And here is the Actor test code that goes with this, where it can be seen that we are able send/receive using the Actor. There is also an example here that shows us trying to use a previously disposed Actor, which we expect to fail, and it does

//Round 1 : Should work fine
EchoShimHandler echoShimHandler = new EchoShimHandler();

Actor actor = new Actor(NetMQContext.Create(), echoShimHandler, new object[] { "Hello World" });
actor.SendMore("ECHO");
string actorMessage = "This is a string";
actor.Send(actorMessage);
var result = actor.ReceiveString();
Console.WriteLine("ROUND1");
Console.WriteLine("========================");
string expectedEchoHandlerResult = string.Format("ECHO BACK : {0}", actorMessage);
Console.WriteLine("ExpectedEchoHandlerResult: '{0}'\r\nGot : '{1}'\r\n",
    expectedEchoHandlerResult, result);
actor.Dispose();

//Round 2 : Should NOT work, as we are now using Disposed actor
try
{
    Console.WriteLine("ROUND2");
    Console.WriteLine("========================");
    actor.SendMore("ECHO");
    actor.Send("This is a string");
    result = actor.ReceiveString();
}
catch (NetMQException nex)
{
    Console.WriteLine("NetMQException : Actor has been disposed so this is expected\r\n");
}

//Round 3 : Should work fine
echoShimHandler = new EchoShimHandler();

actor = new Actor(NetMQContext.Create(), echoShimHandler, new object[] { "Hello World" });
actor.SendMore("ECHO");
actorMessage = "Another Go";
actor.Send(actorMessage);
result = actor.ReceiveString();
Console.WriteLine("ROUND3");
Console.WriteLine("========================");
expectedEchoHandlerResult = string.Format("ECHO BACK : {0}", actorMessage);
Console.WriteLine("ExpectedEchoHandlerResult: '{0}'\r\nGot : '{1}'\r\n",
    expectedEchoHandlerResult, result);
actor.Dispose();

 

Which would give output something like this

image

 

Another Example : Sending JSON Objects

This example shows how to create a simple account shim handler that can be used with standard actor code above. The AccountShimHandler presented here, uses another simple protocol (on purpose, you may choose to make this as simple or as complex as you wish) is does the following:

  • It expects the initial arguments to be 1 in length
  • It expects the initial argument element 0 to be a JSON serialized string of an AccountAction
  • It expects a multi part message, where the 1st message frame is the command string “AMEND ACCOUNT”
  • It expects the 2nd message frame to be a JSON serialized string of an Account

If all these criteria are met, then the AccountShimHandler deserialize the JSON Account object into an actual Account object, and will either debit/credit the Account that was passed into the AccountShimHandler, and then serialize the modified Account object back into JSON and send it back to the Actor via the PairSocket in the AccountShimHandler

The AccountAction class looks like this

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace NetMQActors.Models
{
    public enum TransactionType {  Debit=1, Credit=2}
    public class AccountAction
    {
        public AccountAction()
        {
            
        }

        public AccountAction(TransactionType transactionType, decimal amount)
        {
            TransactionType = transactionType;
            Amount = amount;
        }

        public TransactionType TransactionType { get; set; }
        public decimal Amount { get; set; }
    }
}

 

The Account class looks like this

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace NetMQActors.Models
{
    public class Account
    {

        public Account()
        {
            
        }

        public Account(int id, string name, string sortCode, decimal balance)
        {
            Id = id;
            Name = name;
            SortCode = sortCode;
            Balance = balance;
        }

        public int Id { get; set; }
        public string Name { get; set; }
        public string SortCode { get; set; }
        public decimal Balance { get; set; }
    }
}

 

Here is the code for the AccountShimHandler

using System;
using System.Linq;
using System.Threading;
using NetMQ;
using NetMQ.Sockets;
using NetMQ.zmq;
using NetMQActors.Models;
using Newtonsoft.Json;

namespace NetMQActors
{
    /// <summary>
    /// This hander class is specific implementation that you would need
    /// to implement per actor. This essentially contains your commands/protocol
    /// and should deal with any command workload, as well as sending back to the
    /// other end of the PairSocket which calling code would receive by using the
    /// Actor classes various RecieveXXX() methods
    ///
    /// This is a VERY simple protocol but it just demonstrates what you would need
    /// to do to implement your own Shim handler
    /// </summary>
    public class AccountShimHandler : IShimHandler
    {

        private void AmmendAccount(AccountAction action, Account account)
        {
            decimal currentAmount = account.Balance;
            account.Balance = action.TransactionType == TransactionType.Debit
                ? currentAmount – action.Amount
                : currentAmount + action.Amount;
        }

        public void Run(PairSocket shim, object[] args, CancellationToken token)
        {
            if (args == null || args.Count() != 1)
                throw new InvalidOperationException(
                    "Args were not correct, expected one argument");

            AccountAction accountAction = JsonConvert.DeserializeObject<AccountAction>(args[0].ToString());

            while (!token.IsCancellationRequested)
            {
                //Message for this actor/shim handler is expected to be
                //Frame[0] : Command
                //Frame[1] : Payload
                //
                //Result back to actor is a simple echoing of the Payload, where
                //the payload is prefixed with "AMEND ACCOUNT"
                NetMQMessage msg = null;

                //this may throw NetMQException if we have disposed of the actor
                //end of the pipe, and the CancellationToken.IsCancellationRequested
                //did not get picked up this loop cycle
                msg = shim.ReceiveMessage();

                if (msg == null)
                    break;

                if (msg[0].ConvertToString() == "AMEND ACCOUNT")
                {
                    string json = msg[1].ConvertToString();
                    Account account = JsonConvert.DeserializeObject<Account>(json);
                    AmmendAccount(accountAction, account);
                    shim.Send(JsonConvert.SerializeObject(account));
                }
                else
                {
                    throw NetMQException.Create("Unexpected command",
                        ErrorCode.EFAULT);
                }
            }
        }
    }
}

 

And here is the relevant Actor code

//Round 4 : Should work fine
AccountShimHandler accountShimHandler = new AccountShimHandler();

AccountAction accountAction = new AccountAction(TransactionType.Credit, 10);
Account account = new Account(1, "Test Account", "11223", 0);

Actor accountActor = new Actor(NetMQContext.Create(), accountShimHandler,
    new object[] { JsonConvert.SerializeObject(accountAction) });
accountActor.SendMore("AMEND ACCOUNT");
accountActor.Send(JsonConvert.SerializeObject(account));
Account updatedAccount =
    JsonConvert.DeserializeObject<Account>(accountActor.ReceiveString());
Console.WriteLine("ROUND4");
Console.WriteLine("========================");
decimal expectedAccountBalance = 10.0m;
Console.WriteLine(
    "Exected Account Balance: '{0}'\r\nGot : '{1}'\r\n" +
    "Are Same Account Object : '{2}'\r\n",
    expectedAccountBalance, updatedAccount.Balance,
    ReferenceEquals(accountActor, updatedAccount));
accountActor.Dispose();

 

Which gives a result something like this, if you read the code above you will see the Account object we send, and the receive are NOT the same object. This is due to the fact they have been sent down the wire using NetMQ sockets.

image

Some Actor Frameworks To Look At

There are a couple of Actor frameworks out there that I am aware of. Namely the following ones, there will be more, but these are the main ones I am aware of.

I have only really given Akka a cursory look, but I remember when Axum first came out and gave it a good try, and thought this is neat, no concurrency hell to worry about here. Cool.

For me at least, I wish there was an Actor model in .NET

Advertisements

3 thoughts on “ZeroMQ #7: A Simple Actor Model

  1. somdoron says:

    Why not send creating NetMQActor and send a pull request to NetMQ?

    • sachabarber says:

      Hey the man himself. Cool

      I’ll try and do that over next few days, though am expecting new baby any day now, so it will dependent on that baby arriving or not

  2. somdoron says:

    Good luck!

Leave a Reply

Fill in your details below or click an icon to log in:

WordPress.com Logo

You are commenting using your WordPress.com account. Log Out / Change )

Twitter picture

You are commenting using your Twitter account. Log Out / Change )

Facebook photo

You are commenting using your Facebook account. Log Out / Change )

Google+ photo

You are commenting using your Google+ account. Log Out / Change )

Connecting to %s

%d bloggers like this: