Matthew Podwysocki's Blog : F#, Concurrency

Pondering Axum + F#

podwysocki — Fri, 25 Sep 2009 20:39:00 GMT

It’s been a while since I’ve posted about Axum as I’ve been posting about other asynchronous and parallel programming models. After two releases of the Axum standalone language and lots of good user feedback, it’s time to ponder what could be with Axum. Overall, the goals of Axum to provide an agent-based concurrency oriented system is important as we consider the emerging hardware trends. Many of these ideas would in fact benefit most languages, whether mainstream or not. With all the success that it has had, there have also been some issues as well, which leads me to wonder, what else could we do here? Let step through some of those issues today and see where we could go.

Should It Be Its Own Language?

One of the highlighted concerns was the need to have yet another language. Creating and maintaining languages is a large effort, not to be undertaken lightly at any organization. To base your language on C# while adding additional constructs was a great way to get started and create some buzz and buy-in, but it also brought its own set of problems.

One problem is that it is an enormous effort to keep on par with the C# compiler and add these constructs on a continuing basis. The Spec# project, also based upon C# realized that it wasn’t feasible to keep up with the rapid pace of the language and instead went as a language neutral approach. Not only keeping up with the language, but to enforce many functional programming constructs that Axum embraces such as immutability and purity are harder to enforce given a language which is built upon object-oriented and imperative approaches, an approach which encourage the encapsulation and manipulation of state. So, instead of creating a separate language, why not fold these constructs into an existing language instead?

The Case For F#

Given the impedance mismatch between the C# language and the overall goals of Axum for a safe, isolated, agent-based system that is great for concurrency, why not consider another language such as F#? Now that F# is a first-class citizen within Visual Studio going forward, there can be a strong case made. Given some of the features of F#, which we will discuss, could potentially be a happy marriage with Axum. The F# language and its associated libraries have many of the constructs that are essential to an agent-based messaging system, including immutability (albeit shallow), encouraging side effect free programming, and library functions such as the mailbox processor messaging classes.

In Praise Of Immutability

In the world of concurrency we find that shared mutable state is evil. Often we have to protect certain parts by using low level constructs such as locks, semaphores, mutexes. Because of this, we find that most people tend to avoid concurrency programming as it tends to be too hard to get just right. Instead, communication through message passing in languages such as Axum and Erlang, immutability is the standard. This way, we can freely reference the values without any worry of whether they will change underneath us. Providing first-class support for immutability goes a long way towards making safe concurrency programming easier for the masses. With the F# language, we get that approach by default and with rich types such as records, discriminated unions, lists, tuples and more.

Of particular note, Discriminated Unions are quite useful in messaging systems to define the types of messages your system accepts. For example, we could model our auction messages quite easily using discriminated unions to define which interactions are legal, such as the following code:

type AuctionMessage =
  | Offer of int * MailboxProcessor<AuctionReply>
  | Inquire of MailboxProcessor<AuctionReply>
and  AuctionReply =
  | Status of int * System.DateTime
  | BestOffer
  | BeatenOffer of int
  | AuctionConcluded of 
      MailboxProcessor<AuctionReply> * MailboxProcessor<AuctionReply>
  | AuctionFailed
  | AuctionOver

In this case, we have succinctly defined what messages can be processed. Adding to this, Axum, through the use of data-flow networks and channel protocols could be a nice compliment to determine how these messages are processed. Where else could they match up nicely?

Built For Concurrency

To talk about a potential marriage between F# and Axum, we need to talk about the language and its libraries and how they support concurrency. From first class events, to asynchronous workflows, to mailbox processors, F# has a wide array of asynchronous and parallel programming libraries available out of the box. In particular, the asynchronous workflows and the mailbox processor could match well with the Axum programming model. Interestingly, Luca Bolognese has built upon the mailbox processor for his LAgent framework to show some more interesting scenarios where agent based programming could work. This includes such things as advanced error handling, hot swapping, implementing MapReduce and so on.

With little effort, we can model such things as the standard ping-pong example using F# mailboxes:

let (<--) (m:MailboxProcessor<_>) msg = m.Post(msg)
type PingMessage = Ping of MailboxProcessor<PongMessage> | Stop
and  PongMessage = Pong

let ping iters (pong : MailboxProcessor<PingMessage>) =
  new MailboxProcessor<PongMessage>(fun inbox -> 
    pong <-- Ping inbox
    let rec loop pingsLeft = async { 
      let! msg = inbox.Receive()
      match msg with
      | Pong -> 
          if pingsLeft % 1000 = 0 then
            printfn "Ping: pong"
          if pingsLeft > 0 then
            pong <-- Ping inbox
            return! loop(pingsLeft - 1)
          else
            printfn "Ping: stop"
            pong <-- Stop
            return () }
    loop (iters - 1))
            
let pong () =
  new MailboxProcessor<PingMessage>(fun inbox -> 
    let rec loop pongCount = async { 
      let! msg = inbox.Receive()
      match msg with
      | Ping sender -> 
          if pongCount % 1000 = 0 then
            printfn "Pong: ping %d" pongCount
          sender <-- Pong
          return! loop (pongCount + 1) 
      | Stop -> printfn "Pong: stop"; return () }
    loop 0)

let ponger = pong()
let pinger = ping 100000 ponger
pinger.Start()
ponger.Start()

And you’ll notice from all of this, there are no mutable values anywhere nor shared global state and all communication happens through message passing. We have certain limitations here such as we cannot remote these calls, so all interaction happens inside the single application. Not only that, but how do we enforce no shared state?

What Would It Look Like?

So, now that we looked at some of the reasons why F# and Axum might be a good marriage, what might it actually look like? That’s a great question! Some of the concepts from Axum could map perfectly to F# whereas some might need a little more coaxing. Let’s go back to the basic building blocks of Axum. They consist of the following:

Agents and domains
Channels
Schema
Networks

The first three items are concerned mostly with state isolation and the exchange of information between our isolated regions of our application, while the last item deals more with the how messages are passed.

The domain is the most basic isolation concept where all data is encapsulated and only constructors are exposed outside of the domain definition. Agents are defined to run within the isolated space of a domain with each agent on a separate thread of control. These agents are exposed to one another through passing messages back and forth via channels which define ports through which our data flows. The data that passes between our domains can be defined in a schema, similar in nature of XML Schemas, to define structure and rules of our data. Now to think about message passing, F# already has asynchronous behavior already built in to the mailbox processor, so we could take advantage here and build on top of it and our networks could guide us as to what order messages are passed.

Let’s look at some possible code examples of what things might look like. I’ll choose just a couple as they might work nicely out of the box. Let’s first look at how we might define a port which accepts a given message.

type ProductInput = Multiply of int * int | Done
type ProductOutput = Product of int | AckDone

This way, we could define what our payload type for both incoming and outgoing. Another area would be to look how schemas might be done. In our earlier example, we used simple data types such as integers, but what about more complex types? Using F#, we could think of using records here to simulate the not only the data, but what is and is not required:

type PersonSchema = 
  { Name     : string
    Employer : string
    Age      : int option }

let matt =
  { Name = "Matthew Podwysocki"
    Employer = "Microsoft Corporation"
    Age = None }

By marking the fields we require as standard fields and those which are optional as option types works out nicely. Using F#, we’re able to have concise syntax with a language that is already defined and quite flexible.

Conclusion

As you can see from this post, a potential marriage between Axum and F# would certainly be a great fit. Instead of focusing on Axum as a separate language, focusing the effort on top of an existing language which matches many of the design goals can bear a lot of fruit. I believe Axum is important for the .NET programming stack, giving us more options on how to deal with concurrency and changing hardware architectures. What do you think?

F# – Async Running with Continuation Scissors

podwysocki — Sat, 20 Jun 2009 06:09:00 GMT

As you may have noticed, I’ve been covering a bit about concurrency on this blog lately, and for good reason. Between Axum, Erlang, Scala and F#, there is a lot to explore with actor model concurrency, task based concurrency, data parallel applications and so on. In the next couple of months, I have some talks coming up on concurrency and in particular with F#. I’ve been covering a bit of that with F# and mailbox processing, but I wanted to step back a bit into the asynchronous workflows and hitting against a well known API.

In this case, let’s walk through a simple example of using the Twitter HTTP API to get a user timeline using F# asynchronous workflows. Best of all, I’ve done all of the above code without ever once opening up Visual Studio and instead using the power of a simple text editor and F# interactive, I’m able to be quite productive. It certainly makes me rethink my utter dependence on certain tools…

Give me some tweets… eventually, when you get around to it

In order to get started, we need some helpers when dealing with LINQ to XML. In a previous post, I talked about the need for such a thing due to the lack of implicit operator support in F#. We can use the same operator that we had before such as the following:

let inline implicit arg =
    (^a : (static member op_Implicit : ^b -> ^a) arg)

let (!!) : string -> XName = implicit

Alternatively, we could look at utilizing the XNamespace as well due to it also having an implicit operator to convert from a string. This is helpful as it has a defined the op_Addition which allows us to add an XNamespace to a string to create an XName. To take advantage we could add an additional operator to take care of this. Below is a simple session of F# interactive to show how it might work:

> let (!?) : string -> XNamespace = implicit;;
val ( !? ) : (string -> XNamespace)

> (+) !?"";;
val it : (string -> XName) = <fun:it@205>

> !?"" + "foo";;
val it : XName = foo {LocalName = "foo";
                      Namespace = ;
                      NamespaceName = "";}

If you’ll notice above, I curried the XNamespace implicit operator function and sure enough it recognizes it needs a string in order to produce an XName. Now that we’ve defined our helpers, let’s start looking at how to get a user’s timeline in Twitter. We’ll need to define a holder for our status data as well as the URL we will be using to post data.

type UserStatus = { UserName : string; ProfileImage : string; Status : string; StatusDate : DateTime }

let timelineUrl = "http://twitter.com/statuses/friends_timeline.xml"

We’re interested at this point just of the user name, their image file location, their status and the date of the tweet. Now let’s move onto retrieving and parsing the data.

let parseDocument xml =
  let document = XDocument.Parse(xml)
  document.Root.Descendants()
    |> Seq.choose(fun node -> 
         if node.Element(!! "user") <> null then
           let status = HttpUtility.HtmlDecode(node.Element(!! "text").Value)
           let image = node.Element(!! "user").Element(!! "profile_image_url").Value
           let userName = node.Element(!! "user").Element(!! "screen_name").Value
           let statusDate = DateTime.ParseExact(node.Element(!! "created_at").Value, 
                                                "ddd MMM dd HH:mm:ss +0000 yyyy", 
                                                CultureInfo.CurrentCulture)
           Some { UserName = userName; ProfileImage = image; Status = status; StatusDate = statusDate}
         else None)  

let getUserTimeline userName password =
  async { let request = WebRequest.Create timelineUrl :?> HttpWebRequest
          do request.Credentials <- new NetworkCredential(userName, password)
          
          use! response = request.AsyncGetResponse()
          use reader = new StreamReader(response.GetResponseStream())
          let! xml = reader.AsyncReadToEnd()
          
          return parseDocument xml
        }

In the above code, we create a WebRequest, setting the credentials through Basic Authentication. We’ll cover this approach for now with some attention to OAuth at some point in the relative near future. After setting the identity, we get the response stream and read it to the end. This gives us XML to parse and by using LINQ to XML, we’re able to iterate through each node, and if the user node is not null then we parse out each element for which we’re concerned.

Now that we have the ability to do some asynchronous behavior, what do we do about it? In previous posts, I’ve taken the road of running these operations in parallel, which then runs all instances and returns an answer. But, if we’re only executing one instance, how do we enlist in asynchronous behavior overall? To illustrate, I’ll create a simple Windows Forms application to display the tweets in a textbox and have a refresh button to call the above function to repopulate our textbox. First, let’s create the basic skeleton:

let form = new Form(Visible=true, TopMost = true, Width=500, Height=300)

let status = new TextBox(Multiline=true, ScrollBars=ScrollBars.Vertical)
form.Controls.Add status
status.Width <- 400
status.Height <- 300

let refresh = new Button(Text="Refresh")
form.Controls.Add refresh
refresh.Left <- 410
refresh.Top <- 10

Now that we have a basic skeleton, let’s add the behavior to the button to refresh our textbox. To do this, we simply add a handler to the button click event. In there, we will call Async.RunWithContinuations which gives us the ability to handle not only the successful case, but also should an exception be thrown or some form of cancelation occurs. Before we look at the implementing code, let’s look at the signature of RunWithContinuations:

> Async.RunWithContinuations;;
val it :
  ('a -> unit) * (exn -> unit) * (OperationCanceledException -> unit) *
  Async<'a> -> unit = <fun:clo@0-42>

What this tells us is that it takes a success continuation as the first argument, the exception continuation as the second argument, the cancelation continuation as the third, and then finally the asynchronous operation. This allows us to uniquely handle each type of event, should they occur. In this case, our success continuation should populate the textbox, whereas both the cancelation and the exceptional case should show a message box containing the exception.

refresh.Click.Add
   (fun args ->
      status.Text <- ""
      Async.RunWithContinuations(
        (fun res -> 
           res |> Seq.iter(fun item -> status.Text <- sprintf "User: %s - %s\r\n\r\n%s" item.UserName item.Status status.Text )),
        (fun exn -> MessageBox.Show(sprintf "An error occurred: %A" exn) |> ignore),
        (fun cxn -> MessageBox.Show(sprintf "A cancellation error ocurred: %A" cxn) |> ignore),
        (getUserTimeline "username" "password")))

As you can see, the behavior is rather straight forward to define handlers for our three cases. Lastly, we provide our credentials to the getUserTimeline function and then when the button is clicked, we’ll get results much like the following screenshot.

Then we can keep pressing the refresh button and then asynchronously, it will handle the behavior appropriately, either in success or in failure.

Conclusion

This is a pretty simply and naive application, but it can show how you can take advantage of asynchronous behavior using the Async Workflows in F#. By being able to have the F# libraries handle the exceptions and cancelation, a major source of errors is reduced by quite a bit. Giving us the power then to handle each one of these scenarios, either in success, failure or cancelation is a great concept. There is still more yet to be covered including Futures and the inclusion of the Task Parallel Library as well as other adventures into Twitter and Bing APIs.

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.

When Side Effects and Laziness Collide

podwysocki — Tue, 02 Jun 2009 16:34:00 GMT

While working on a side project recently, I came to rediscover some of the consequences of one of my earlier posts on “Side Effects and Functional Programming”. It’s important that we realize that when we are creating our programs to beware of lazy evaluation and side effects.

Adding a Wrinkle

Consider the following example. We need to be able to get the stream data from a given web request for a given URL. From there, we will take the stream of data and read each line and return it as a collection. Typically if we’re going to do this through F#, we’d use a sequence expression to lazily read through each line and return it as an IEnumerable<string>. But, let’s add an extra wrinkle to this. What if we want to do this in a non-thread blocking asynchronous scenario? This adds an extra wrinkle to this in terms of managing the lifetime of our objects. We might write the code such as the following:

open System.IO
open System.Net

let getStreamData (uri:string) =
  async { let request = WebRequest.Create uri 
          let! response = request.AsyncGetResponse()
          
          return seq { use stream = response.GetResponseStream()
                       use reader = new StreamReader(stream)
                       while not reader.EndOfStream do yield reader.ReadLine() } 
        }

What we’re doing is putting a sequence expression inside of an asynchronous expression. We must bind the response outside of the sequence expression due to the fact that we are doing this in an asynchronous fashion. Once inside a sequence expression, a bind has a different meaning. But, there’s a problem with this code. Can you spot it?

The problem lies in the fact that we’re not explicitly disposing the response, which implements IDisposable. My rule of thumb when flushing out my design is that if a class implements IDisposable and I ultimately control its lifetime, then I must wrap it in a using block. If you read the documentation, you’ll realize that it’s not totally necessary when dealing with an HttpWebRequest to call Dispose, but it is when you’re dealing with an FtpWebRequest. Because I don’t make any distinction in this code, calling Dispose is a must. We might be tempted then to change the code to the following:

open System.IO
open System.Net

let getStreamData (uri:string) =
  async { let request = WebRequest.Create uri 
          use! response = request.AsyncGetResponse()
          
          return seq { use stream = response.GetResponseStream()
                       use reader = new StreamReader(stream)
                       while not reader.EndOfStream do yield reader.ReadLine() } 
        }

Running through a quick example using F# interactive, we’d get the following result:

> Async.RunSynchronously (getStreamData "http://www.bing.com");;
val it : seq<string> =
  Error: Cannot access a disposed object.
Object name: 'System.Net.HttpWebResponse'.

The problem now becomes that our response binding now will be disposed by the time that I iterate the results of the sequence expression. Remember that our evaluation of the sequence expression is lazy, so by the time I’m ready to iterate, the response has already had its Dispose called. So, where does that leave us?

Well, there are several things that we could do here. First, we could copy the response inside of our sequence expression and bind that to a using block. This would ensure that by the time we’re finished, we would properly dispose the response. That code might look like the following:

open System.IO
open System.Net

let getStreamData (uri:string) =
  async { let request = WebRequest.Create uri 
          let! response = request.AsyncGetResponse()
          
          return seq { use response' = response
                       use stream = response'.GetResponseStream()
                       use reader = new StreamReader(stream)
                       while not reader.EndOfStream do yield reader.ReadLine() } 
        }

This approach will work, unlike the previous example. By copying in this reference, we can ensure that the response will be disposed by the time we’re finished with our sequence expression. Although this works, it has a certain code smell to me. This code smell revolves around copying these references and there is a certain danger in doing so in terms of managing the lifetime. Instead, could we try another approach, perhaps more eager?

From Lazy to Eager

Another approach we could take with this is to do away with the idea of lazy I/O in this solution altogether. Instead, how about something as simple as a list comprehension instead of a sequence expression? The code might then look like the following:

open System.IO
open System.Net

let getStreamData (uri:string) =
  async { let request = WebRequest.Create uri 
          use! response = request.AsyncGetResponse()
          
          return [ use stream = response.GetResponseStream()
                   use reader = new StreamReader(stream)
                   while not reader.EndOfStream do yield reader.ReadLine() ] 
        }

Now what we have is that we have the use binding on our response and stream. At the end of our function, we simply use a list comprehension instead of the sequence comprehension which in this case gives us the benefit of eager evaluation.

Conclusion

When dealing with asynchronous expressions mixed with sequence expressions can cause problems due to object lifetimes. As I’ve had in my earlier post, it’s best to avoid such scenarios if possible and instead, focus on an eager approach. The savings you get here is you have to worry less about the resource lifetimes which may or may not be in scope by the time you’re ready to execute the sequence expression. The approach using the list comprehension should be the preferred solution.

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.

Talking Functional Programming with Erik Meijer

podwysocki — Wed, 25 Mar 2009 04:58:00 GMT

When I was last out in Redmond, I had the opportunity to sit down with Erik Meijer to speak about something for which I’m passionate, functional programming. After such recent appearances on Channel9 as Joe Duffy and Anders Hejlsberg, I was quite flattered to be asked to appear. I gave a short as possible introduction to start talking about any number of topics including my passion for functional programming, the Haskell and F# languages, and many other ideas.

One of the topics I talked about was monads. In some of my previous posts, I talked about them extensively in my Much Ado About Monads series. When I got the email that this video was posted, I was already writing my series installment on the asynchronous monad, so I hope this video will serve as a fine addition to that conversation when I get finished with it later this week.

So without further ado, here is the video.

Which Is More Beautiful Architecture - FP or OOP?

podwysocki — Mon, 09 Mar 2009 16:31:00 GMT

Recently, upon the recommendations of a few people, I picked up a copy of the book “Beautiful Architecture: Leading Thinkers Reveal the Hidden Beauty in Software”. This book is a great read and includes essays from some of the top minds in software today. Some of the topics covered are:

How Facebook's architecture is the basis for a data-centric application ecosystem
The effect of Xen's well-designed architecture on the way operating systems evolve
How community processes within the KDE project help software architectures evolve from rough sketches to beautiful systems
How feature creep has helped GNU Emacs gain unanticipated functionality
The magic behind the Jikes RVM self-optimizable, self-hosting runtime
The design choices and building blocks that made Tandem the choice platform in high-availability environments for over two decades
The differences and similarities between object-oriented and functional architectural views
How architectures can affect the software's evolution and the developers' engagement

But one essay in particular caught my attention was Chapter 13 Software Architecture: Object-Oriented Versus Functional by Bertrand Meyer. In this writing, he steps through an example application of financial contracts, as presented in both a presentation called “Composing contracts: An adventure in financial engineering” by Simon Peyton-Jones, Jean-Marc Eber and Julian Seward, along with an associated presentation by Jean-Marc Eber called “Compositional Description, Valuation, and Management of Financial Contracts: The MLFi Language”.

Bertrand Meyer’s overall conclusion is that given the above functional solution, an object-oriented design, especially supporting such features as closures is better than the functional approach as it can provide higher-level abstractions more supportive of extension and reuse.

Laying Out The Case

Dr. Meyer defines what architecture “beauty” is from his 1997 hallmark book, “Object-Oriented Software Construction, Second Edition” as three objective criteria:

Reliability
Does the architecture help establish the correctness and robustness of the software?
Extendibility
How easy is it to accommodate change?
Reusability
Is the solution general, or better yet, can we turn it into a component to be plugged in directly, off-the-shelf, into a new application?

Using the above criteria and the functional examples cited, there are a few issues as noted with his essay such as the lack of data points, lack of detail, and a specific focus on modularity, whereas the functional programming examples had a focus on other criteria such as elegance of a declarative approach. The biggest drawback to this essay is that most, if not all Object Oriented examples revolve around the use of Eiffel, which is a fringe language and not in the mainstream thought in the Object Oriented world. To focus on this versus a mainstream functional language such as Haskell has some serious drawbacks that I don’t think can be reconciled in this essay.

Now, I’m not going to go through the example as I linked them above for you to dissect at your leisure. Instead, we’re going to focus on the single issue of modularity between the two approaches.

Extendibility/Reusability

To speak to the reuse and extendibility issue, the functional example is heavy in the use of combinators, which are functions that take functions as arguments and return new functions. The idea is that the financial world has a lot of jargon and instead of specifying a large set of contracts ahead of time, we could compose them using a strict set of combinators. This way, we have the ability to define new unforeseen financial contracts with relative ease in a declarative manner. The paper is quite compelling for how to build composable Domain Specific Languages around financial modeling, for which functional programming is widely used.

To the author the idea of defining combinators does not scale to large applications and must be divided into modules. In turn, the extendibility problem arises on how you modularize said code while affecting the fewest modules possible. Another issue is that with a one module approach, this does not help reusability as you wish to have only a subset of those operations imported. Careful modularization can get around these issues and I do not necessarily see this as a stumbling block. Instead with a good strategy on modularization and qualifying imports can solve a lot of these issues. Also, the use of type classes can add extendibility to your architecture by defining a set of operations in which other types can

Managing State

Another issue raised by the author is state management. The notion of stateless programming is essential to the pure functional programmer. State and unmanaged side effects are indeed the bane of concurrency. Instead, Haskell has the notion of monads which allow for the modeling of not only state, but exception management, as well as input/output operations.

Dr. Meyer doesn’t have a concern about the whether these concepts are hard to learn, as they shouldn’t be, but instead, if they are worth the effort at all. Instead, through the use of Command Query Separation, where queries return results but do not alter the world, whereas commands alter the state of the world and do not return a value. I believe, to rely on this design pattern instead of enforcement through the type system is his undoing. Most imperative/object-oriented programmers do not follow this rule and thus should only confine his comments to Eiffel alone. I believe once again, his focus on Eiffel is undercutting his credibility that well known OOP principles can subsume a Functional Programming example.

This analysis could go on and on, but must this be an either/or battle?

The False Battle

With the exclusion of Java, most modern languages have been evolving to include such concepts as closures, which come from the functional world. Even the Eiffel language that has been written by the author, has incorporated “agents” which are his term for closures. And then we have multi-paradigm languages such as F# and Scala which take more of a functional programming approach with the full support of imperative and object oriented features.

Although at this point in history, functional programming languages are less popular in mainstream development, it’s hard to now avoid their influence. With the increased attention to concurrency and parallelism, the ideas from functional programming such as lack of state and immutability are gaining momentum as we realize the peril of shared state. Instead of this battle, where can object oriented technologies learn from functional programming and vice versa?

How Functional Programming Influences

Let’s try another tactic with this and talk about where functional programming techniques have their biggest influence. Brian McNamara, of the F# team, laid out the argument well in his post “How does functional programming affect the structure of your code” in which he describes its effects in three distinct areas:

In the Large
Covers overall architecture and high level design. This is where the modularization of our code has its biggest influences. To instill laziness as a default architecture, compose message passing systems and high performance computing concerns is where functional programming can have its greatest impact.
In the Medium
The medium covers more lower level API decisions that we make. When combined with a good modularization strategy, can yield great results. Many times as well the object oriented design patterns such as the command, builder, strategy and visitor patterns can better be expressed through functions.
Also, the use of discriminated unions, records and collection structures can create concise and rich domain models using pure functional structures.
In the Small
By programming in the small, we mean at the actual function level. With an immutable by default posture, we are better able to express our values without type annotations and the compiler is better able to make decisions based upon that. Functional composition is also very much in effect at this level as we build more powerful functions from the small pieces to create rich solutions.

By understanding these concepts help us better understand where functional programming has its greatest influence and where we can use the techniques even if not in a pure functional language.

Conclusion

Overall, the arguments presented by Bertrand Meyer were persuasive in some ways, yet lacking in others. Unable to cite more real world applications leaves me wanting. Without this, the modularization issue cannot be addressed in full. Other areas such as insisting on Command-Query-Separation as a default falls largely on deaf ears outside of the Eiffel community. I do realize the more progressive teachers are starting to focus on this, but for the most part, CQS is largely untaught. Instead having the type system express our side effects is a much more powerful option. Overall, I thought a lot of the arguments given his bias and the lack of better examples, fell flat. Using Eiffel, a non-mainstream OOP langauge doesn’t help his case and should have focused on more mainstream approaches as in C#, and Ruby among others.

So, who wins? That’s a good question. If you’re in a world resigned to side effects then OOP might just be for you. Else, Haskell and the lessons learned from the language can be powerful instruments in creating rich and beautiful architectures using pure functional techniques. I feel with time and maturation, the functional programming community can contribute more to the architecture picture. The battle continues?

Exploring MapReduce with F#

podwysocki — Tue, 03 Mar 2009 07:09:00 GMT

With my exploration into mass concurrency and big data problems, I’m always finding challenges to give myself on how I might solve a given issue. Such examples that have intrigued me along the way as PLINQ, MPI/.NET, Data Parallel Haskell, but one in particular has intrigued me more – MapReduce. A challenge I gave myself is to fully understand this paradigm and implement a version using F#.

What Is MapReduce?

MapReduce is a Google programming model and implementation for processing and generating large data sets. Programs written using this more functional style can be parallelized over a large cluster of machines without needing the knowledge of concurrent programming. The actual runtime can then partition the data, schedule and handle any potential failure. The Google implementation of it runs on a large cluster of machines and can process terabytes of data at a time.

The programming model is based upon five simple concepts:

Iteration over input
Computation of key/value pairs from each input
Grouping of all intermediate values by key
Iteration over the resulting groups
Reduction of each group

So, where does this Map/Reduce wording come in? Well, let’s explore each:

Map
Function written by the user to take an input pair and produce a result of intermediate key/value pairs. The intermediate values are then grouped with the same intermediate key and passed to the Reduce function.
Reduce
Function also written by the user to take the intermediate key and sequence of values for that key. The sequence of values are then merged to produce a possibly smaller set of values. Zero or one output value is produced per our Reduce function invocation. In order to manage memory, an iterator is used for potentially large data sets.

How useful is this programming model? Using this model, we’re able to express some interesting problems such as the following:

Word counts in large documents
Distributed Grep – Find patterns in a document
Count of URL Access Frequency
Reverse Web-Link Graph – Find all source URLs for a given destination
Term-Vector per Host - Summarize most important words in a document
Inverted Index
Distributed Sort

That’s only a start to what we can do. After the paper which explained the technology was published in 2004, there have been several implementations including the open-source Hadoop project and the Microsoft Research Project Dryad. Of course, it doesn’t come without some criticism about the resource usage such as this post called “eBay doesn’t love MapReduce”.

Dryad is more interesting in that its functionality subsumes MapReduce and can do in regards to job creation, resource management, job monitoring, visualization and more. What’s even more interesting is the associated project DryadLINQ, which allows us to compile LINQ expressions to have sent across the cluster so that I could perform intense data analysis using LINQ operators that we’re used for our other data tasks. What does the future hold for Dryad remains to be seen, but the possibilities seem plenty.

F# Implementation

The challenge I gave myself was to understand the basics of the programming model. Ralf Lämmel, formerly of the Microsoft Programmability Team, published a paper entitled Google’s MapReduce Programming Model – Revisited which I used as a source for how my F# implementation works. If you want to understand not only the deep internal details, but also how an implementation in Haskell would look, I highly recommend this paper.

In our example, we’re going to use a MapReduce implementation to look for the most frequent IP addresses from our IIS logs. Let’s first start by defining the outer skeleton needed. Let’s define what the map_reduce function and associated steps will look like:

#light  

namespace CodeBetter.MapReduceExample  

module MapReduceModule =  

  let map_reduce   
    // Map function take pair and create sequence of key/value pairs  

    (m:'k1 -> 'v1 -> seq<'k2 * 'v2>)   
    // Reduce function takes key and sequence to produce optional value  

    (r:'k2 -> seq<'v2> -> 'v3 option) 

    // Takes an input of key/value pairs to produce an output key/value pairs

    : Map<'k1, 'v1> -> Map<'k2, 'v3> =  

    map_per_key >>     // 1. Apply map function to each key/value pair  

      group_by_key >>  // 2. Group intermediate data per key  

      reduce_per_key   // 3. Apply reduce to each group

I’m using the standard sequences instead of lists for this because the data may be large and the List<T> in F# by default is eagerly evaluated. The case could be made to use the LazyList<T> to model the data, but as I might want to use the Parallel Extensions and the PSeq module I used before, I’ll stick with the IEnumerable<T>.

The comments here in the code say a lot about what’s going on. Our map_reduce function takes two functions, the map and the reduce which are explained in the code comments. It’s important to understand what happens last is our series of three functions, our map_per_key, group_by_key and reduce_per_key. Each of those steps fall into line with the explanation from above. At this point, we haven’t actually yet implemented the functions, but let’s go ahead and do that now to flush out our map_per_key, group_by_key and reduce_per_key. The following code comes before the last statement which ties our three functions together:

    let map_per_key : Map<'k1, 'v1> -> seq<('k2 * 'v2)> = 

      Map.to_seq >>                   // 1. Map into a sequence 

        Seq.map (Tuple.uncurry m) >>  // 2. Map m over a list of pairs 

        Seq.concat                    // 3. Concat per-key lists 

    let group_by_key (l:seq<('k2 * 'v2)>) : Map<'k2,seq<'v2>> = 

      let insert d (k2, v2) = Map.insert_with Seq.append k2 (seq [v2]) d 

      let func (f:Map<'a, seq<'b>> -> Map<'a, seq<'b>>) (c:'a * 'b)  
        : (Map<'a, seq<'b>> -> Map<'a, seq<'b>>) =  
        fun x -> f(insert x c) 

      (Seq.fold func (fun x -> x) l) Map.empty 

    let reduce_per_key : Map<'k2, seq<'v2>> -> Map<'k2,'v3> = 

      let un_some k (Some v) = v // Remove optional type 

      let is_some k = function 

        | Some _ -> true         // Keep entires 

        | None   -> false        // Remove entries 

      Map.mapi r >>           // 1. Apply reduce per key 

        Map.filter is_some >> // 2. Remove None entries 

        Map.mapi un_some      // 3. Transform to remove option

The first function is the map_per_key which first turns a map into a sequence. Then we map the m function over our list of pairs, and then finally concat the per-key lists. Next our group by key might look a little interesting, but it’s simply a right fold over our insert function and our sequence of key/value pairs. As I had in a previous post about how you can do right-folds in the terms of a left-fold. Finally, our reduce per key applies the reduce function, then filters the removable entries, and finally transforms to remove the optional type.

There you have it, our implementation of a MapReduce. Now let’s step ahead with our implementation to step through our IIS logs. Let’s first retrieve the data for our logs from a given directory. We can use the async workflows in F# to retrieve this data in parallel given a base directory. Next, we need a way to harvest the data from the given logs so that we can get the IP address from it.

module LogCount = 

  open System.IO 

  let processed_logs : (string * string) array =  
    Async.Run ( 

      Async.Parallel  
        [for file in Directory.GetFiles @"C:\Logs\" ->  
           async { return file, File.ReadAllText file }]) 

  let harvest_data (data:string) : seq<string> = 

    seq { for line in String.lines data do 

            if not (line.StartsWith "#") then 

              yield line.Split([|' '|]).[2] 

        }

Finally, we can count get the IP occurrence count by implementing our map and reduce functions and calling our map_reduce implementation.

  let ip_occurrence_count : Map<string, string> -> Map<string, int> = 

    let m = const' (harvest_data >> Seq.map(flip Tuple.curry 1)) 

    let r = const' (Seq.sum >> Some)  
    MapReduce.map_reduce m r

module MainModule = 

  [<EntryPoint>] 

  let main(args:string array) = 

    printfn "%A" 

      (LogCount.ip_occurrence_count 

        (Map.of_seq LogCount.processed_logs)) 

    0

The m function above harvests the data from the string value and then mark each IP address with an occurrence of 1. Then our reduce function will sum the sequence of 1s for each given key to produce our result. There is some boilerplate code that has been omitted due to space constraints but will be available at the end of this post. Once we run this against a known set of data, we might get the following results:

seq 
  
  [[192.168.1.2, 46]; [192.168.1.8, 339]; 

   [192.168.1.10, 11]; [192.168.1.15, 43]] 

Press any key to continue . . .

With this simple example, we can see some of the power and possibilities such a programming model can bring. This exercise is more of an idea of how one might be written using the programming model, as I didn’t write any system to handle concurrency, scheduling and so on.

Conclusion

In many of my adventures in big data analysis and data parallel concurrency, MapReduce is one of the more interesting solutions. By utilizing functional programming constructs, we are able to write rich code to analyze our data that may not as structured as we wish. But, does our adventure stop here with this implementation? Not necessarily, but either way it was a lot of fun.

You can find the source code for this example here.