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 realized that nothing I can say here would be as good or as thorough as the guide, so I have had 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:
What Is An Actor Model?
Here is what Wikipedia has to say 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 the 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 it's 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.
To take this one step further, let's 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 let's 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 possibly 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:
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 passed 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.
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 therefore 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 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
{
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;
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);
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)
{
cts.Cancel();
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
{
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)
{
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
{
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 and 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
{
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)
{
NetMQMessage msg = null;
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.
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();
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");
}
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:
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) and 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 deserializes 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
{
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)
{
NetMQMessage msg = null;
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:
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 receive are NOT the same object. This is due to the fact they have been sent down the wire using NetMQ sockets.
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.
Submit an article or tip
