Matthew Podwysocki's Blog : Erlang

[ANN] DC ALT.NET 9/10/2009 – Webmachine with Kevin Smith

podwysocki — Fri, 04 Sep 2009 19:27:00 GMT

This month, DC ALT.NET is moving the meeting a little forward this month and teaming with NOVALANG and the Erlang Users of Arlington/DC to talk about the Webmachine REST Toolkit. We at DC ALT.NET continue to strive to push developers to look outside their normal surroundings and comfort zones and this month is no different. Details are below:

Getting Some REST with Erlang and Webmachine

Kevin Smith of Hypothetical Labs will be speaking about WebMachine: an erlang REST toolkit which makes it easy to develop HTTP interfaces using Erlang. Providing RESTful interfaces is quickly becoming the norm for modern websites. There's more to REST than handling GETs and POSTs, though. This talk will illustrate how the combination of Erlang and webmachine simplifies and speeds the development of REST resources.

About Kevin Smith

Kevin Smith has, at various times, been a network administrator, DBA, developer, team lead and trainer over his 14 year career. He ﬁrst learned about Erlang in 2006 via Joe Armstrong's excellent “Programming Erlang” and has never looked back. He attended the first (and only?) Erlang Studio where he was asked to produce what became the popular PragProg screencast series "Erlang In Practice". Until recently, Kevin was a developer at Engine Yard where he worked on a distributed systems management application using Erlang. Kevin founded Hypothetical Labs in April of this year to focus on Erlang full time. He is also the co-author of the upcoming book "Erlang Web Application Development".

Date/Time: 9/10/2009 – 7-9PM

Location:
The Motley Fool
2000 Duke Street
Alexandria, VA 22314

The “Anti-For” Campaign

podwysocki — Fri, 26 Jun 2009 04:38:00 GMT

Recently, there has been an effort launched called the “Anti-If Campaign” in which they deride the use of if statements and instead, focus on Object Oriented Principles in order to create more flexible designs. Now certainly, I have a sympathetic ear to this cause as I’ve seen code that literally walks off the side of the screen due to nesting of if statements. Pattern matching to me, especially at the top level of the function is actually quite beautiful in a way, such as the implementations in Haskell:

-- Haskell
lucas :: Int -> Integer
lucas 0 = 2
lucas 1 = 1
lucas n = lucas (n - 2) + lucas (n - 1)

And in Erlang, this also holds true:

% Erlang
-module(lucascalc).
-export([lucas/1]).

lucas(0) -> 2;
lucas(1) -> 1;
lucas(N) -> lucas (N - 2) + lucas (N - 1).

Simple, easy to understand and best of all, no if statements. But, instead of focus on this debate, I’d like to propose another which strikes closer to this functional programmers heart, the “Anti-For Campaign”. This is simply to say that we should create and use composable functions instead of explicit for loops. This is actually an old post I had written months ago and until now had been unfinished, but now with some inspiration, it’ll finally be done.

What and Why?

Before you throw all sorts of questions asking what and why, let me instead ask a question. When you’re writing a loop, ask yourself the question, “What am I accomplishing in this loop?” Chances are, it might be one of the following:

Query (Map, Filter, etc)
Aggregation (Sum, Count, etc)
Perform some side effect

If you’re doing more than one of those in a single loop, then well, you’re probably doing too much. In fact, Martin Fowler’s Refactoring site has a refactoring called “Split Loop” which would solve that issue. It is better for future refactorings and readability if we keep those loops pointed to do one thing, and one thing only. Better yet, we could rid ourselves of that loop altogether, and that’s what we’ll focus on here.

Looking at first two bullet points, you’ll notice most of LINQ is indeed built around those two to be able to query data as well as aggregate. The final bullet point, we perform some sort of side effect, perhaps writing to a file, printing to the console, or even perhaps sending messages.

So, what’s my problem with them? My problem is that it focuses more on the How instead of the What. Let’s look at a quick example down below of what I mean. First, we’ll attempt to find all prime numbers under 100 using C# as an example language. First in the How:

static bool IsPrime(int i) {
    var lim = Math.Sqrt(i);
    
    Func<int, bool> check = null;
    check = j =>
        j > lim || (i % j != 0 && check(j + 1);
        
    return check(2);
}

static void Main(string[] args) {
    var numbers = Enumerable.Range(1, 100);
    var output = new List<int>();

    foreach(var number in numbers)
        if(IsPrime(number)) output.Add(number);
}

The problem is that if we then want to compose another operation, well, it’s really hard to do inside of these for loops. We’re too focused on the how at this point. Instead, getting to know generics and lazy evaluation, in .NET 2.0 and beyond, we were able to write generic functions to take advantage of some functional constructs. This will give us a more declarative style that we can now focus on the what.

static IEnumerable<T> Filter<T>(
    this IEnumerable<T> items,
    Func<T, bool> predicate) {
    
    foreach(var item in items)
        if(predicate(item))
            yield return item;
}

static void Main(string[] args) {
    var numbers = Enumerable.Range(1, 100);
    
    var primes = items.Filter(x => IsPrime(x));
}

Of course we realize that LINQ already has such constructs built in, so we could rewrite the entire code above in just one statement.

var primes = Enumerable.Range(1, 100)
    .Where(x => IsPrime(x));

And what’s better is that it is composable that we could do other operations such as aggregations (sum, count, etc) without much additional code:

var primesCount = Enumerable.Range(1, 100)
    .Where(x => IsPrime(x))
    .Count();

And there you have it. We not only have a query, but also an aggregation. Try doing that with those non-composable loops! Justin Etheredge has a nice writeup as well recently on the subject.

Coping Strategies

Functional languages tend to deemphasize the use of such constructs. In fact, Haskell has neither a for loop nor a while loop, and languages such as F# and OCaml have limited support for such constructs in terms of no break and continue. We tend to look at those two in particular with suspicion due to the fact that it cannot return a value and instead it mutates state in some fashion. With that in mind, how do we cope with the fact that those aren’t available to us? Above I showed a basic concept of a filter instead of an explicit loop, but what about some other considerations?

Some things we might want to consider with some links to some of my previous posts on the subject:

Explicit Recursion
Transforming each item (Map/Select & SelectMany in LINQ)
Selecting items (Filter/Where in LINQ)
Aggregating Data (Folds/Aggregate in LINQ)

In fact, many times that people could consider using explicit recursion could instead use a fold to aggregate the data which then cuts out the issue of tail call optimization. By truly understanding the goals of LINQ as well as the concepts of functional programming, we can realize that most of the looping that we do can indeed be replaced by the above, outside of side effects of course.

Now What About Those Side Effects?

Is there a place where we draw the line and say that explicit loops are ok? Eric Lippert was recently asked about the reasoning of the lack of the ForEach extension method on IEnumerable<T>. His response was that he was philosophically opposed to such an endeavor as the whole approach is to cause a side effect. As IEnumerable<T> collections are immutable, he doesn’t believe it makes as much sense because you wouldn’t be side effecting the collection itself. Not only that, but introducing closures can complicate object lifetimes and all sorts of potential reference issues.

What about me? I understand his concerns, and in C#, I can certainly see where he is coming from. However, in F# we have such constructs readily available to us in the iter and iteri functions as shown below in F# Interactive:

> let flip f y x = f x y
- [1..10]
-   |> List.map((*) 2)
-   |> List.filter(flip (%) 3 >> (=) 0)
-   |> List.iter(printfn "%d");;
6
12
18
> [1..10]
-   |> List.map((*) 2)
-   |> List.filter(flip (%) 3 >> (=) 0)
-   |> List.iteri(printfn "%d\t%d");;
0       6
1       12
2       18

Outside of logging, writing to the console and such are rare in functional programming, so once again, I can certainly understand the concern.

Conclusion

I hope by going through some basic scenarios that we will indeed question our code the next time we see that we are writing that explicit loop with a for and a while. With a tool chest filled with such functions as transforming every item, to filtering content, to aggregating data and so on, we can realize that we can create composable solutions instead of creating mutable collections or mutable variables and aggregating to them which are not as much. So, come and join me in the “Anti-For Campaign”.

Axum – Ping Pong with Ordered Interaction Points

podwysocki — Thu, 04 Jun 2009 06:22:00 GMT

UPDATE: Removed code and explained that what I had was not intended behavior

After a slight diversion into F# mailbox processing, it’s time to come back to talk a little bit more about Axum. In our last Axum post, we discussed using dataflow networks to clean up our canonical Ping-Pong example. This time, we’ll try another approach using ordered interaction points.

A Little Housekeeping

Before we get started with our full demonstration, let’s take a brief look at some concepts that will come into play for this solution. In our previous example using data flow networks, we a shared-nothing approach between these two agents. This time, let’s introduce the concept of domains. A domain in Axum allows for a group of agents to safely share data while shielding this state from the outside world. Like normal classes, domains can have such things as fields and methods, but also, we can declare agents. We’ll discuss how this comes into play later on in the post.

domain A {
    private string fieldName;
    ...
}

Another key concept to understand is reader and writer agents. Typically, when dealing with shared state of some sort, we have a notion of a reader-writer lock. This synchronization technique allows access to the shared state as one to many readers or a single writer to ensure consistency of that shared resource. Agents in Axum also follow this notion where a reader agent is only allowed to read shared state and a writer agent can both read and write to domain state.

domain A {
    private string fieldName;
    
    public reader agent A1 { ... } // Can read fieldName
    
    public writer agent A2 { ... } // Can read/write fieldName
}

One more concept that needs to be covered before our solution is the idea of hosts. By using a host, we are able to associate an agent’s type with a particular instance in the domain. For example when we look at the following code, what we’re doing is when someone asks for our A1 domain with an address of myA1, we attach it to the current domain and then return the associated C channel endpoint.

Host<A1>("myA1");

We can connect to this host then through a mechanism such as the following:

var chan = new C("myA1");

Bringing this all together, let’s look at the Ping-Pong example through these eyes.

Getting to the Solution

In this solution, we’ll introduce the aspect of ordered interaction points. This is an interaction point that acts as both a source and a target and the keyword being ordered which means that the order of messages is preserved. To declare one, use the OrderedInteractionPoint<T> class such as the following:

var ip = new OrderedInteractionPoint<int>();

Now, let’s get to our solution. The main application is no different, but where we will find differences is the interaction between the Ping and Pong agents. As the code for that part hasn’t changed, let’s focus on the interaction between the ping and pong inside our domain. First, let’s define the domain and we’ll call it the PingPongDomain (imaginative, I know). In this domain, we need to set up the hosting for the Pong agent at the “Pong” address as well as the ordered interaction points for the ping and the pong.

public domain PingPongDomain
{
    public PingPongDomain()
    {
        Host<Pong>("Pong");
    }
    
    OrderedInteractionPoint<Signal> ping = new OrderedInteractionPoint<Signal>();
    OrderedInteractionPoint<Signal> pong = new OrderedInteractionPoint<Signal>();

Once we have this defined for our domain, our Ping and Pong agents inside this domain will have access to these points. Since the Ping agent will ultimately both read and write messages, but using the immutable data type as the message type, we need not define it as a writer, but instead as a reader. Let’s define that below.

    public reader agent Ping : channel PingPongStatus
    {
        public Ping()
        {
            PrimaryChannel::HowMany ==> Process;
        }
        
        private Signal Process(int iters)
        {
            // Create pong
            var chan = new PingPong("Pong");

           // Send pings and receive pongs
           for (int i = 0; i < iters; i++)
           {
               ping <-- Signal.Value;
               receive(pong);
               Console.WriteLine("ping received pong");
           }

            chan::Done <-- Signal.Value;
            return Signal.Value;
        }
    }

You’ll notice that our code doesn’t look too much different than our previous example, except for connecting to the PingPongChannel with the “Pong” address that we defined in our domain above. The other change is that we are sending a message on the ping ordered interaction point and receiving a message on the pong instance.

Moving along to our Pong agent, once again, the code does not deviate much. But, as we are reading and writing our domain state by sending messages that are immutable, we could declare ourselves as a reader agent. We will continuously receive messages either on the ping ordered interaction point or we will receive a message on the Done port on the primary channel. The code will look something like the following.

    public reader agent Pong : channel PingPong
    {
        public Pong()
        {
            while (true) receive
            {
                from PrimaryChannel::Done:
                    return;
                from ping:
                    Console.WriteLine("pong received ping");
                    pong <-- Signal.Value;
                    break;
            }
        }
    }

Now that we have a solution, but, unfortunately due to a bug, this doesn’t work because the compiler cannot distinguish whether the state is immutable or not. But, we can get it to run with an escape hatch.

Being Unsafe in a Safe Way

As I stated before, Axum does have a notion of an escape hatch when it comes to code that allows you to execute code that might modify shared state through the use of the unsafe keyword. Most times it’s better to be safe than sorry in this regard, but in this example, it fits quite well. You can think of this unsafe escape hatch much like the Haskell unsafePerformIO function. This function in Haskell is the back door into the IO monad which allows an IO computation to be performed at any time. And like our unsafe keyword, in order for this to be safe, we should be free of side effects and not dependent upon the environment.

With this in mind, let’s rewrite the example using the unsafe construct. First, let’s look at the Ping agent and how we might make that a little unsafe, but yet fully operational:

public agent Ping : channel PingPongStatus
{
    public Ping()
    {
        PrimaryChannel::HowMany ==> Process;
    }
    
    private Signal Process(int iters)
    {
        // Create pong
        var chan = DefaultCommunicationProvider.Connect<PingPong>("Pong");
        unsafe
        {
            // Send pings and receive pongs
            for (int i = 0; i < iters; i++)
            {
                ping <-- Signal.Value;
                receive(pong);
                Console.WriteLine("ping received pong");
            }
        }
        chan::Done <-- Signal.Value;
        return Signal.Value;
    }
}

By simply wrapping our for loop in the unsafe block, we are able to send messages to our ping ordered interaction point and receive from our pong ordered interaction point where we weren’t allowed to in the above example. The same applies for our Pong agent as well. We can wrap the while loop receiving messages in an unsafe block so that we can receive from the ping ordered interaction point and then send a signal to the pong ordered interaction point.

public agent Pong : channel PingPong
{
    public Pong()
    {
        unsafe 
        {
            while (true) receive
            {
                from PrimaryChannel::Done:
                    return;
                from ping:
                    Console.WriteLine("pong received ping");
                    pong <-- Signal.Value;
                    break;
            }
        }
    }
}

Running our sample, we get our expected results. You can find the complete source code to this example here.

Conclusion

Unlike Erlang and other shared-nothing messaging approaches, Axum allows for shared domain state through tightly controlled access. By doing so, we keep the safety aspect of messaging through such notions of a reader-writer lock. Of course, there is an escape hatch for those who truly understand their application and the ramifications of what they are doing. As you can see, Axum has a wide array of solving even the most simple and canonical examples of actor model concurrency. There is still more yet to cover including asynchronous behavior and an exploration of the Auction example that I gave in F#.

Download it today, use it in anger and give the team feedback as to help shape what actor model concurrency might look like on the .NET platform.

Actors in F# – The Bounded Buffer Problem

podwysocki — Thu, 28 May 2009 06:58:00 GMT

In the previous post, I covered an example of an auction simulation using asynchronous message passing and a shared nothing approach using the MailboxProcessor class in F#. The auction example was a great piece to demonstrate scalability by adding additional clients to create a sort of bidding war between them. Once again, with this approach, we’ve eliminated the need for locks and other concurrency primitives.

This time, let’s take another canonical example of a Bounded Buffer and look at some of the design patterns around this.

The Bounded Buffer

The goal of this post is to walk through an example of actor model concurrency of the canonical Bounded Buffer which is another example given in Scala. The intent of this demo is to store and retrieve items in a buffer (rather simple actually). Given this example, we’ll walk through how we might implement this using the constructs in F#. One important aspect of this solution is to not post messages asynchronously as before, but instead, to post a message and await the reply.

Without further ado, let’s get into the code. As before, we have a few utility functions that will be quite handy for this journey. Much like we defined the (<—) operator last time for posting messages, I’d like one for posting and waiting for a reply. In addition, I need a way to accomplish currying for reasons you will see later. One thing that has irked me on occasion is the confusion between partial application and currying, which I’ve covered earlier. Getting back to the issue at hand, let’s look at the code for that:

// Curry the arguments
let curry f x y = f (x, y)

// Asynchronous post
let (<--) (m:_ MailboxProcessor) msg = m.Post msg

// Post and reply operator
let (<->) (m:_ MailboxProcessor) msg = m.PostAndReply(fun replyChannel -> msg replyChannel)

As you’ll notice, I put the two operators, the asynchronous post and the post and reply operators, the former not being needed for this post. The PostAndReply method gives a way to post a message and wait for the reply. A temporary reply channel is created and that forms part of our message. This reply channel is an AsyncReplyChannel<T> which supports one function of Reply which we will use later. This message is then sent back to the caller as the result.

Next, we need to define the messages we will be processing as part of this bounded buffer. Each of these messages define operations that our buffer supports, namely put, get and stop. Let’s take a look at these in detail:

type 'a BufferMessage = Put of 'a * unit AsyncReplyChannel 
                      | Get of 'a AsyncReplyChannel 
                      | Stop of unit AsyncReplyChannel

As you will notice, each of these has an associated AsyncReplyChannel part to the defined message. This is to allow me to reply to each of the callers in turn. The Put and Stop both have reply channels that take no associated data, so we can create them as an AsyncReplyChannel<unit>. The Put message allows us to put a value into the buffer, the Get allows us to retrieve those values in turn, and the Stop allows us to stop the mailbox.

Let’s move on to the heart of the matter, the actual bounded buffer. This class takes in a buffer size and then we expose methods that allow us to put values in the buffer, get values from the buffer and stop the mailbox. Below is how the code might look:

type 'a BoundedBuffer(N:int) =
  
  let buffer =     
    MailboxProcessor.Start(fun inbox ->
      let buf:'a array = Array.zeroCreate N
      let rec loop in' out n =
        async { let! msg = inbox.Receive()
                match msg with
                | Put (x, replyChannel) when n < N ->
                    Array.set buf in' x
                    replyChannel.Reply ()
                    return! loop ((in' + 1) % N) out (n + 1)
                    
                | Get replyChannel when n > 0 ->
                    let r = Array.get buf out
                    replyChannel.Reply r
                    return! loop in' ((out + 1) % N) (n - 1)
                    
                | Stop replyChannel -> replyChannel.Reply(); return () }
      loop 0 0 0)
      
  member this.Put(x:'a) = buffer <-> curry Put x
  member this.Get() = buffer <-> Get
  member this.Stop() = buffer <-> Stop

Inside our BoundedBuffer class, we create the buffer which then creates an initialized array. Because array contents are mutable, there is no sense in putting this as part of our processing loop. Instead, we’ll focus on the input index, the output index and the number of items in the buffer as part of our processing loop. When we receive the Put message when the number of items in the buffer is less than the buffer size, we set the value at the specified input index, return a reply back to the caller, and then loop with an increment to our index with a modulo of the buffer size as well as the number of items in the buffer. In receiving a Get message when the number of items in the buffer is greater than zero, we get the item at the output index, send the reply back to the caller with the value, and then loop with an increment to the output index with a modulo as well as decrementing the number of items in our buffer. Finally, should we receive a Stop, we simply reply back to the caller and return.

We created three methods to wrap this functionality for outside consumption. The Put method takes in the item to post to the buffer, and then we simply do a PostAndReply with our Put message and our item to post. I used currying here because the Put message requires two parameters, the item to put as well as the reply channel. In this case, my operator already provides that reply channel, so I only need to supply the item to put. Both the Get and the Stop methods are fairly straight forward as they post their respective messages with their private reply channels.

How does this work? Let’s fire up F# interactive and take a look with an example of posting a few items to our buffer and then retrieving them.

> let buffer = new int BoundedBuffer 42;;

val buffer : int BoundedBuffer

> buffer.Put 12;;
val it : unit = ()
> buffer.Put 34;;
val it : unit = ()
> buffer.Put 56;;
val it : unit = ()
> buffer.Get();;
val it : int = 12
> buffer.Get();;
val it : int = 34
> buffer.Get();;
val it : int = 56
> buffer.Stop();;
val it : unit = ()

What I did was create a BoundedBuffer that handled integers with a buffer size of 42. Then I posted three values, 12, 34 and 56. After putting these values into our buffer, I then retrieved each in the order in which it was placed into our buffer. Finally, I stopped the buffer. The complete source code to this example can be found here.

Conclusion

Once again, we can create rather interesting solutions using this shared nothing asynchronous message passing approach in F#. This solution involving the bounded buffer is no exception. How might this solution look in Axum? In due time, we will approach this as well as our Auction example from the previous post. There are a lot of Axum items to cover especially in regards to asynchronous methods and ordered interaction points, so stay tuned.

F# Actors Revisited

podwysocki — Wed, 20 May 2009 05:25:00 GMT

UPDATE: Removed ref cells to use two recursive loops

In the previous post, I covered briefly about the actor model in F#. This style of concurrency, using asynchronous message passing and a shared-nothing approach through the use of mailboxes is a pretty powerful mechanism for achieving scalability. With a shared-nothing approach, we can remove the need for such concurrency primitives such as locks.

Taking cues from Erlang, F# has the capabilities as well through the use of the MailboxProcessor class. In this post, let’s walk through a canonical example and explain some of the design patterns around this.

Mailboxes, Etc

Before we begin, let’s make sure we have some of the basics down. As I stated above, the approach here is that inter-agent communication is accomplished through a shared-nothing asynchronous message passing system. Each of these agents have a mailbox, which is nothing more than a queue of messages sent by other processes. These messages are then retrieved via either a scan or receive approach, depending if you wish to continue receiving messages. To determine which message we received, we do basic pattern matching against our given message types and decide how we want to handle it. To continue processing, we then recurse the function again to keep sending and receiving. We can send any number of pieces of data as part of these messages. For our example, we’ll implement the Auction example that comes from Scala.

Going Once… Going Twice… Sold!

Our goal in this post is to walk through implementing the Auction example which is an example given in Scala. The intent of this demo is to show an online auction service using the shared-nothing asynchronous message passing. We’ll have two clients bidding against each other to see who wins the particular item. Given that Scala has library support for actors, this is an attempt to see how well F# can handle this as well. The first impressions of the Scala model is that it’s much more focused on classes and objects than the F# version will be.

Without further ado, let’s start digging through the code. First up, let’s get some utility functions out of the way to help us on our journey. Instead of using the dot operator on some objects to send messages, I prefer the <—operator, as shown also in the Expert F# book. Also, I need a way to deconstruct an option type so that I can deal with a bidder that may not exist yet.

let (<--) (m:_ MailboxProcessor) x = m.Post x
let unSome (Some x) = x

Both of these functions should be rather self explanatory. The unSome takes in an option type with Some value and returns the value. Let’s move on to the messages that we need to define. I have comments beside each as to define what each message signifies.

type AuctionMessage =
  | Offer of int * AuctionReply MailboxProcessor // Make a bid
  | Inquire of AuctionReply MailboxProcessor     // Check the status
and AuctionReply =
  | Status of int * DateTime // Asked sum and expiration
  | BestOffer                // Ours is the best offer
  | BeatenOffer of int       // Yours is beaten by another offer
  | AuctionConcluded of      // Auction concluded
      AuctionReply MailboxProcessor * AuctionReply MailboxProcessor
  | AuctionFailed            // Failed without any bids
  | AuctionOver              // Bidding is closed

I decided to use a recursive type definition as to allow my AuctionMessage discriminated union to appear before my AuctionReply discriminated union, because both messages in AuctionMessage references a MailboxProcessor that handles the AuctionReply messages. In F#, order does matter in which values are declared, so this is a quick way to get a readable output. Next, we need to define some constants such as time to shutdown and our allowed bid increments:

let timeToShutdown = 3000
let bidIncrement = 10

It’s now time to define our auction agent. This will handle the overall orchestration of how the auction is handled. This includes handling both the Offer and the Inquire messages as well as what happens when we fail to get a message within a certain time period. Let’s step through the code now and then explain down below:

let auctionAgent seller minBid closing =
  new AuctionMessage MailboxProcessor(fun inbox ->
    let rec loop maxBid maxBidder =
      async { let! msg = inbox.TryReceive((closing - DateTime.Now).Milliseconds)
              match msg with
                | Some ( Offer(bid, client) ) ->
                    if bid >= maxBid + bidIncrement then
                      if maxBid >= minBid then unSome maxBidder <-- BeatenOffer bid                  
                      client <-- BestOffer
                      return! loop bid (Some client)
                    else
                      client <-- BeatenOffer maxBid
                      return! loop maxBid maxBidder
                
                | Some ( Inquire client ) ->
                    client <-- Status(maxBid, closing)
                    return! loop maxBid maxBidder
                
                | None ->
                    if maxBid >= minBid then
                      let reply = AuctionConcluded(seller, unSome maxBidder)
                      unSome maxBidder <-- reply
                      seller <-- reply
                    else seller <-- AuctionFailed
                    
                    let! msg' = inbox.TryReceive timeToShutdown
                    match msg' with
                    | Some ( Offer (_, client) ) -> 
                        client <-- AuctionOver
                        return! loop maxBid maxBidder
                    | None -> return ()         
            }
    loop (minBid - bidIncrement) None)

It’s a bit of code to digest, but the gist of this is rather simple. When creating an instance of this auction agent, we need a seller, a minimum bid and a closing time for the auction. From this, we create our MailboxProcessor to handle AuctionMessage types. Since we are continuously in a cycle of receiving messages and sending results, we have a recursive loop which carries our state for us. This state, which we carry from recursive call to the next is our maximum bid and our maximum bidder at the time. When we begin our call to this loop function, we start with the max bid of the minimum bid minus the bid increment and no maximum bidder.

When we are receiving messages, it’s important to handle any potential timeouts. In this case, we use this timeout as a mechanism for determining when our auction is over. In order to do this, we must use the TryReceive function instead of the Receive. Let’s look at the differences below:

// Throws an exception if timeout is exceeded
member MailboxProcessor.Receive : ?timeout : int -> Async<'msg>

// Returns None if timeout is exceeded
member MailboxProcessor.TryReceive : ?timeout : int -> Async<'msg option>

Since we need to know how to handle timeouts, it’s best to use the TryReceive approach. If we receive an offer, we determine whether the bid is acceptable, meaning greater than our current maximum plus increment. If it is, then we check whether the maximum bid is greater than the minimum, and if so, then send a message to the current maximum bidder with the BeatenOffer message. If we receive an inquiry, we simply return a Status message back to the client with our current bid and how much time they have left. Finally, if we don’t receive a message in the given timeframe, it’s time to wind down our auction. If the maximum bid is greater than our minimum, our auction is a success and we let our maximum bidder and our seller know. If, however, we don’t receive any bids, then our auction is considered a failure. Finally, and additional offers received will be responded to with an AuctionOver message to indicate to the client that they are not the winner. If this operation times out, we simply exit. A bit of a mouthful, but fairly straight forward.

Let’s move on to the client aspect. First, let’s set up a few things necessary for our client, encapsulated in the Auction module. We’ll set such parameters as our minimum bid, our closing date, our seller and our auction itself.

module Auction =
  let random = new Random()
  
  let minBid = 100
  let closing = DateTime.Now.AddMilliseconds 10000.
  
  let seller = new AuctionReply MailboxProcessor(fun inbox ->
    let rec loop() =
      async { let! _ = inbox.Receive() 
              return! loop()}
    loop())
  let auction = auctionAgent seller minBid closing

In order to create a seller for our auction, we need to stub out basic features of this agent. In this case, we simply have a processing loop which receives a message, does nothing with it, and continues with the loop again. As I have stated above, we need both our seller agent, and our auction agent, but what about our clients? In our same module, we define them as the following:

  let client i increment top = 
    let name = sprintf "Client %i" i
    let log msg = Console.WriteLine("{0}: {1}", name, msg)
    
    new AuctionReply MailboxProcessor(fun inbox ->
    
      let rec startAuction() =
        async { log "started"
                auction <-- Inquire inbox
                let! curMsg = inbox.Receive()
                match curMsg with
                | Status(maxBid,_) ->
                    log <| sprintf "status(%d)" maxBid
                    return! loop 0 maxBid }
      and loop current max =
        async { if max >= top then log "too high for me"
                    
                let current' =
                  if current < max then
                    let current' = max + increment
                    Thread.Sleep (1 + random.Next 1000)
                    auction <-- Offer(current', inbox)
                    current'
                  else current
                
                let! msg = inbox.TryReceive timeToShutdown
                match msg with
                | Some BestOffer -> 
                    log <| sprintf "bestOffer(%d)" current'
                    return! loop current' max
                | Some (BeatenOffer maxBid) ->
                    log <| sprintf "beatenOffer(%d)" maxBid
                    return! loop current' maxBid
                | Some ( AuctionConcluded(seller, maxBidder) ) ->
                    log "auctionConcluded"; return ()
                | Some AuctionOver ->
                    log "auctionOver"; return ()
                | None -> return () }
      startAuction())

What we’re doing here is having two mutually recursive loops. The first loop, the startAuction loop will get the max bid for us and then go to the main loop where we do our main work. From there, we’ll receive a message sent from the auction to the inbox and determine whether we’re the highest bidder, we’re the maximum or it’s time to call it a day. Once again, fairly straight forward code.

Lastly, it’s time to bring it all together. Let’s set up the calling code and see the results:

open Auction

seller.Start()
auction.Start()
(client 1 20 200).Start()
(client 2 10 300).Start()
Console.ReadLine() |> ignore

Executing this code gives us the following for our results:

Client 2: started
Client 1: started
Client 2: status(90)
Client 1: status(90)
Client 2: bestOffer(100)
Client 1: bestOffer(110)
Client 2: beatenOffer(110)
Client 1: beatenOffer(120)
Client 2: bestOffer(120)
Client 2: auctionConcluded
Client 1: auctionOver

What this tells us is that client 2 starts first and bids first. Client 2 then bids 100, only to be outdone by client 1 and goes back and forth until client 2 comes in with the best and final offer. The auction is then concluded and client 1 is notified that the auction is over. We can add additional client agents to the list and have just as much fun as before. To give you an idea, if we add another agent with an increment of 30 and a max bid of 150, we get the following results:

Client 1: started
Client 2: started
Client 3: started
Client 1: status(90)
Client 2: status(90)
Client 3: status(90)
Client 1: bestOffer(110)
Client 1: beatenOffer(120)
Client 3: bestOffer(120)
Client 2: beatenOffer(120)
Client 3: auctionConcluded
Client 1: auctionOver
Client 2: bestOffer(130)
Client 2: auctionConcluded

Now what we have here is that client 2 wins with an offer of 130 and the rest are notified of their loss. You can find the complete code for this example here.

Conclusion

Solutions such as these using shared-nothing asynchronous message passing can create rather scalable architectures. Utilizing the MailboxProcessor class, we’re able to implement actor model concurrency to tackle some of these problems with coordination of concurrent processes. How might Axum handle this scenario? Good question and I hope to have that answered shortly.

Axum – Ping Pong with Dataflow Networks

podwysocki — Fri, 15 May 2009 12:48:00 GMT

In the previous post, I gave the canonical Ping-Pong example in Axum and how it compared with Axum. I want to revisit this post because there are some areas in which we can rework it in addition to the other solutions we’ll visit. Some parts were needlessly chatty and instead we’ll work in some other language features to help clean up our solution.

Actors in F#

But, before we begin, Nicklas Gustafsson, the architect behind Axum, was inspired by my post and made an F# version of the canonical Ping-Pong example. F# supports a concept of mailbox processing as a first class citizen, much like other languages such as Scala and Erlang, although not as central to the core as it is in Erlang.

To use these, simply use the MailboxProcessor class. Below is an implementation as given by Nicklas and we can compare with my previous post with the Erlang implementation. The Start function creates a new MailboxProcessor instance and starts it with the given inbox. Much like the Erlang solution, this relies upon recursion as a technique for keeping the mailboxes alive. In order for us to communicate from one mailbox to the other, we must as part of our message, include a reference to our mailbox of origin in order to communicate.

#light

type Message = Finished | Msg of int * Message MailboxProcessor

let ping iters (outbox : Message MailboxProcessor) =
    MailboxProcessor.Start(fun inbox -> 
        let rec loop n = async { 
            match n with
            | 0 -> outbox.Post Finished
                   printfn "ping finished"
                   return ()
            | _ -> outbox.Post <| Msg(n, inbox)
                   let! msg = inbox.Receive()
                   printfn "ping received pong"
                   return! loop(n-1)}
        loop iters)
            
let pong() =
    MailboxProcessor.Start(fun inbox -> 
        let rec loop () = async { 
            let! msg = inbox.Receive()
            match msg with
            | Finished -> 
                printfn "pong finished"
                return ()
            | Msg(n, outbox) -> 
                printfn "pong received ping"
                outbox.Post <| Msg(n, inbox)
                return! loop() }
                    
        loop())

ping 100 <| pong() |> ignore
System.Console.ReadLine() |> ignore

It’s a fairly straight forward and follows the mold of Erlang quite nicely. Scala as well also follows this mold in their Ping-Pong example. Using the actor package in Haskell, we can also simulate the Ping-Pong example.

Of the criticisms mentioned in the post, I have no real issues with these such as the issues with no clear endpoints and how to distinguish the actors from one another. The type safety of Axum does give us some advantages as to define an explicit contract between actors.

Getting back to the issue at hand, what could we do to clean up our example in Axum?

Cleaning Up with Dataflow Networks

In our previous example, where we had the two agents communicating with each other, the Ping and the Pong, we pretty much driven by the data received from other agents. Instead, we can turn our attention to using dataflow networks, which is to be driven only by the availability of data entering our network and the computations that are performed as we move through this network.

Using Axum, we have several operators that can help us. The Forward Operator ( ==> ) sends each message produced by the source to the target. We also have the Forward Once ( –> ) operator which forwards a message from the source to the target, then disconnects after the first message. Using these operators, we can build pretty rich pipelines for our data to flow, but just as well, we could use it to build an event-driven system where we respond to events.

Let’s take a look at the rewrite of the main agent in which we create the endpoints for Ping and then send the message on the HowMany port. After that, we then forward the received value of Done to the Done state. Much cleaner than before.

using System;
using System.Concurrency;
using Microsoft.Axum;
using System.Concurrency.Messaging;

public agent Program : channel Application
{
    public Program()
    {
        // Create instance of ping and send msg
        var chan = Ping.CreateInNewDomain();
        chan::HowMany <-- 10;
        chan::Done ==> Done;
    }
}

Next up, let’s take a look at the Ping agent. You’ll note no change in our PingPongStatus channel and the associated ports. The interesting part is in the Ping constructor in which we forward the message from the HowMany port to the Process method which takes the number of iterations to perform. In Process method, we create the Pong instance and communication channels, then we loop through our number of iterations to send the Signal to the Ping port, and receive the return Signal on the Pong port. After the loop, we send a message to the Done to close the interaction and then return a Signal to end our Ping. Below is the implementation of the code.

public channel PingPongStatus
{
    input int HowMany : Signal;
    output Signal Done;
    Start: { HowMany -> End; }
}

public agent Ping : channel PingPongStatus
{
    public Ping()
    {
        PrimaryChannel::HowMany ==> Process;
    }
   
    private Signal Process(int iters)
    {
        // Create pong
        var chan = Pong.CreateInNewDomain();      
        
        // Send pings and receive pongs
        for (int i = 0; i < iters; i++)
        {
            chan::Ping <-- Signal.Value;
            receive(chan::Pong);
            Console.WriteLine("ping received pong");
        }
        chan::Done <-- Signal.Value;
        return Signal.Value;
    }  
}

You’ll notice that we’re no longer interested in sending the number of iterations to the Pong. It shouldn’t be needed and instead, we can send the Signal message on the Done port when we’re complete. The Pong can listen in an infinite loop until it receives the message on the Done port.

Lastly, let’s take a look at the Pong agent. We’ll clean up the channel to rid ourselves of the HowMany port. Instead, all we need are Signals to communicate at this point inside the infinite loop that I mentioned above. Let’s take a look at the code required.

public channel PingPong
{
    input Signal Done;
    
    input Signal Ping;
    output Signal Pong;
}

public agent Pong : channel PingPong
{
    public Pong()
    {
        while (true) receive
        {
          from PrimaryChannel::Done:
            return;
          from PrimaryChannel::Ping: 
            Console.WriteLine("pong received ping");
            PrimaryChannel::Pong <-- Signal.Value;
            break;    
        }
    }
}

As you can see, inside the Pong constructor, we’re performing an infinite loop in which we call receive each time, then we can react to the messages coming on the specified ports, in this case, the Done and the Ping ports. If we receive a message on the Ping port, we then send a Signal back on the Pong port and then break. Else, if we receive a message on the Done port, then we exit the agent. This approach is a bit more cleaner than our previous attempt at the Ping-Pong example.

Conclusion

As I indicated before, are other angles I wish to tackle with this example, including using asynchronous methods, ordered interaction points and so on, which I will get into in subsequent posts. With this little cleanup, I hope this shows some of the power of the language that we can indeed create concise, yet safe concurrent apps using message passing. Download it, use it in anger, and give the team feedback on the MSDN forum. Once again, with your help we can help shape the future of an actor based concurrency model on .NET.

Axum – Introduction and Ping Pong Example

podwysocki — Tue, 12 May 2009 05:37:00 GMT

As it was announced last week, Axum, a .NET Domain Specific Language around safe, scalable parallel programming through the actor model and message passing was released to the world as a CTP. It was noted, that although this is an initial release, that this is still an incubation project which may or may not lead to a final product. This post will serve as a kind of introduction to the language and some basic concepts.

Just A Brief Introduction

Before we get started, I wanted to make sure that we had a basic introduction to the language. The question often comes up, why another .NET language? Many people feel that with such others as C#, VB.NET, now F#, IronPython and IronRuby, that maybe we have enough languages already to solve most of our programming needs. But there remains one important, yet untackled problem around making coordinated parallelization easy. Sure, we have such things as the Parallel Extensions for .NET being baked into .NET 4.0, which gives us the Task Parallel Library, Parallel LINQ and so on, but none of these really address the intricate nature of coordinated parallelization.

Typically, when we’re developing these parallel applications, we could partition the data as usual and then scatter them out as parallel tasks. Some jobs were meant for solutions such as this, but others might require a bit more coordination. The Axum language, built upon the Microsoft Concurrency and Coordination Runtime from the Microsoft Robotics Studio, allows for a pretty natural language around such concepts as messaging, side effect free programming, but just as well, free yourself of many concurrency related bugs. Because isolation is built-in as a default, one of the major sources of concurrency related bugs is taken away. Just as well, when we design our applications with messaging in mind, we are building a scalable solution where concurrency isn’t an afterthought.

So, the final word is, why a language? Built-in isolation, easy to understand syntax around messaging, easy asynchronous programming model and many other benefits we get by stepping outside of the C# bounds. The real question then becomes, so what? What can we do with it?

The Erlang Way

When learning language concepts, I find it easier to spend some time in the language which most strongly implements that concept. This way, I learn the concepts best when they are first-class citizens in the space. For learning functional programming, I don’t think it’s as wise to spend your time hacking on C# as to instead learn on a more functional language such as Haskell. Often when people ask how best to learn FP concepts, often Haskell comes into the equation and has obviously improved my F# code as well as some of my C# as well. The same applies to a language with actor model concurrency as a built-in concept, where I can learn from my time spent in Erlang as a good precursor to what I can do in Axum.

One of the really elementary things to do in Erlang is a simple ping-pong example. In this simple example, we create two processes, a pinger and a ponger, and send messages between them a number of times. Such code might look like the following:

-module(tut15).

-export([start/0, ping/2, pong/0]).

ping(0, Pong_PID) ->
    Pong_PID ! finished,
    io:format("ping finished~n", []);

ping(N, Pong_PID) ->
    Pong_PID ! {ping, self()},
    receive
        pong ->
            io:format("Ping received pong~n", [])
    end,
    ping(N - 1, Pong_PID).

pong() ->
    receive
        finished ->
            io:format("Pong finished~n", []);
        {ping, Ping_PID} ->
            io:format("Pong received ping~n", []),
            Ping_PID ! pong,
            pong()
    end.

start() ->
    Pong_PID = spawn(tut15, pong, []),
    spawn(tut15, ping, [3, Pong_PID]).

From this example, we have two functions, a ping and a pong. Our ping has an arity with two arguments, the iteration number and the Process ID (PID) of the pong. If we are at zero iterations left, then we sent the finished message back to the pong’s PID, else we send a ping to the pong, then receive our pong message and then recurse to decrement the iteration number. Pretty simple, elegant and straight forward example. Running this example, we might get the following:

Erlang R13B (erts-5.7.1) [smp:2:2] [rq:2] [async-threads:0]

Eshell V5.7.1  (abort with ^G)
1> c(tut15).
{ok,tut15}
2> tut15:start().
Pong received ping
<0.40.0>
Ping received pong
Pong received ping
Ping received pong
Pong received ping
Ping received pong
ping finished
Pong finished

The <0.40.0> value you see above is the return value from the start() function which is nothing more than the PID of the ping process. Now that we’ve seen this in action in Erlang, how could we do the same in Axum?

The Axum Solution

We have a couple of options when implementing this ping-pong example in Axum. First, we have approach using channels, and another using an ordered interaction point. Let’s walk through each of these solutions as a possible implementation. The Axum Programming Guide is a great reference just in case you want to dive deeper into some of these topics.

Before we get started here, let’s first take a look at a simple hello world example to get you familiar with the syntax. Below is a quick example of getting the command line arguments, and then setting the channel to say we are finished. Alternatively, we could set the error code

using Microsoft.Axum;

public agent Program : channel Application
{
    public Program()
    {
        // Wait for the command line arguments
        var cmdLine = PrimaryChannel::CommandLine;
        var cmdArgs = receive(cmdLine);  
        
        // Tell the application we're done
        PrimaryChannel::Done <-- Signal.Value;
    }
}

The first piece you will notice is the lack of the word class, and instead the keyword agent. The agent concept comes from the actor model of concurrency in which it is an autonomous entity that communicates with others via messages they receives from these other “actors” and can spawn off other “actors” as well. These agents aren’t like normal classes in which you have fields that other classes can modify, nor have any direct method calls you can make.

The second piece you’ll notice after the agent declaration is that our Program implements the Microsoft.Axum.Application Channel. When we implement a channel, we are attached at the implementing end of that channel and we become the server of messages on that channel. This channel has a public property called PrimaryChannel which gives access to the channel being implemented. The double-colon operator gives us access to the channel’s port.

Now to actually dive through the agent constructor, we wait for the command line arguments to be passed. This receive is a blocking call, so we will wait until we receive the data from the CommandLine port, which in this case will be a string array. We then tell the application we are done by sending a message through the <—operator to the Done port, which is done asynchronously, as opposed to the receive. Now that we looked at a basic example, let’s move on to a channel based approach.

The Channels Example

The first example we’re going to look at is using channels and protocols. In this example, we will use some of the above constructs with the addition of channel protocols. First, let’s look at the main code which starts the program:

using System;
using System.Concurrency;
using System.Collections.Generic;
using Microsoft.Axum;
using System.Concurrency.Messaging;

public agent Program : channel Application
{
    public Program()
    {
        // Wait for our command args
        var cmdArgs = receive(PrimaryChannel::CommandLine);
        
        // Set the iterations to be some number
        var iters = 3;
        
        // Create instance of ping and send msg
        var chan = Ping.CreateInNewDomain();
        chan::HowMany <-- iters;
        
        // Receive how many done
        receive(chan::Done);
        
        PrimaryChannel::Done <-- Signal.Value;
    }
}

This code first waits for our command arguments, then we set the iterations to 3 for now, but we could set it as high as we wish, maybe higher than 5000 if we care. Sure, that’s a high number, but our system should be good for it. We then call Ping.CreateInNewDomain() which instantiates two ends of the PingPongStatus channel, creates an instance of Ping that implements the channel, then returns the using end of the channel. After this, we send a message on the HowMany port with our number of iterations desired. Then, we’ll block until we’ve received our message on the Done port to indicate that we are finished. Finally, we wrap up with setting the overall Done on the PrimaryChannel to indicate our program completion. Let’s move onto the Ping implementation next.

public channel PingPongStatus
{
    input int HowMany;
    output int Done;
    
    Start: { HowMany, Done -> End; }
}

public agent Ping : channel PingPongStatus
{
    public Ping()
    {
        // How many are we supposed to do?
        var iters = receive(PrimaryChannel::HowMany);
        
        // Create pong and send how many to do
        var chan = Pong.CreateInNewDomain();      
        chan::HowMany <-- iters;
        
        // Send pings and receive pongs
        for (int i = 0; i < iters; i++)
        {
            chan::Ping <-- Signal.Value;
            receive(chan::Pong);
            Console.WriteLine("Ping received Pong");
        }
        
        // How many did Pong process?
        int pongIters = receive(chan::Done);

        PrimaryChannel::Done <-- 0;
    }  
}

First, we created a PingPongStatus channel to get an input of how many to process, and then we have an output message which says we’re complete. You’ll notice there’s something more to our channel which begins with Start:. This is an implementation of a protocol, which allows us to specify the order in which messages are sent to which ports. All cases start with the Start as we have above, and then we transition to the HowMany port, then from the Done, it signifies the end of our interaction on this channel.

Now to analyze the Ping agent. Quite simply, we receive the number of iterations message from our Program agent from above. From there, we call Pong.CreateInNewDomain() which instantiates two ends of the PingPong channel, creates an instance of Pong that implements the channel, then returns the using end of the channel. We send a message to the HowMany port to specify the number of iterations our ponger should do. Going through each iteration, we send a message on the Ping port of a Signal, receive the message back on the Pong port. Finally, we receive the number of iterations that our pong did and then set our Done flag to 0. Now, let’s move onto our Pong implementation.

public channel PingPong
{
    input int HowMany;
    output int Done;
    
    input Signal Ping;
    output Signal Pong;
}

public agent Pong : channel PingPong
{
    public Pong()
    {
        // Get how many we're supposed to do
        var iters = receive(PrimaryChannel::HowMany);

        int i = 0;

        // Receive our ping and send back our pong
        for (; i < iters; i++)
        {
            receive(PrimaryChannel::Ping);
            Console.WriteLine("Pong received Ping");
            PrimaryChannel::Pong <-- Signal.Value;    
        }
    
        PrimaryChannel::Done <-- i;
    }
}

Notice that we have defined our PingPong channel to have ports to indicate how many to do, an output of how many we did, plus an input of a ping and an output of a pong. Pretty straight forward and self explanatory. Our Pong agent receives the number of iterations we’re supposed to do. Then we loop through our iterations, receiving a ping and sending a pong. After we finish the loop, we send a message on the Done port to indicate the number of iterations done.

If we stick with our three iterations, the output from our application should look like the following:

Pong received Ping
Ping Received Pong
Pong received Ping
Ping Received Pong
Pong received Ping
Ping Received Pong

Pretty straight forward and easy example as we dip our toe into what we can do with Axum. You can find the complete source code here.

Conclusion

There are other angles I wish to tackle with this example, including using asynchronous methods, ordered interaction points and so on, which I will get into in subsequent posts. Until then, I hope this brief foray into some of the capabilities of Axum will inspire you to give Axum a look. Download it, use it in anger, and give the team feedback on the MSDN forum. With your help we can help shape the future of an actor based concurrency model on .NET.