Luca Bolognese's WebLog

Luca Bolognese leaves Microsoft

lucabol — Fri, 19 Feb 2010 16:58:02 GMT

This is my last post on this blog. My new blog is here: http://lucabolognese.wordpress.com/

I accepted a role as Director for Credit Suisse in London. I’m excited by the opportunity to work in the financial industry, a long-standing desire of mine. I’m also excited to write more F# code and to be closer to Italy, where my extended family is.

The past ten years in Microsoft have been a wild ride. I’m proud to have been part of ObjectSpaces, Generics, LINQ and F# (and much more …). I’ve been lucky to be able to post on this blog and present at conferences about such innovative technologies. It has all been a lot of fun. I’m sure the next ten years will be as good.

I’ll see you guys on my new blog.

LChart: displaying charts in F# – Part I

lucabol — Wed, 17 Feb 2010 19:20:00 GMT

I want to use F# as a exploratory data analysis language (like R). But I don’t know how to get the same nice graphic capabilities. So I decided to create them. Here is a library to draw charts in F#. It steals ideas from this book and this R package. It is nothing more than a wrapper on top of the Microsoft Chart Controls to give it a more ‘exploratory’ one line calling syntax. It is also rough work in progress: I don’t wrap all the chart types and there are bugs in the ones I wrap. Also the architecture is all wrong (more on this in another post). But it’s a start and it kind of works. Attached the full code.

I will continue this series in my new blog at wordpress: http://lucabolognese.wordpress.com/. The reason I need a new blog will be explained in an upcoming post.

Part II is now here.

Ok, let’s start. How do I draw a chart?

let x = [1.;2.5;3.1;4.;4.8;6.0;7.5;8.;9.1;15.]
let y = [1.6;2.1;1.4;4.;2.3;1.9;2.4;1.4;5.;2.9]

lc.scatter(x, y) |> display

X and Y are just some make up data. lc is the name of a class (????) and scatter is a static method on it. scatter doesn’t display the chart, it just produces a an object that represents the chart. Display displays the chart. The reason for using the bizarre lc static class is that I want it to be short so that it is easy to type in the fsi.exe. At the same time it needs to support optional parameters (which are not supported on top level functions in F#).

You get a window with this chart on it. You can press CTRL+C to copy it (as I did to post it here).

You might want to customize the chart a bit by passing some of these famous optional parameters:

lc.scatter(x = x, y = y, markerSize = 10, markerStyle = MarkerStyle.Diamond,
    xname = "Players", yname = "Ratings", title = "Players' Ratings")  |> display

Or you might want to print different types of charts:

lc.line(y = y, markerSize = 10, markerStyle = MarkerStyle.Diamond, xname = "Players", yname = "Ratings", title = "Players' Ratings", isValueShownAsLabel = true,
    color = Color.Red) |> display

lc.spline(x = x, y = y, markerSize = 10, markerStyle = MarkerStyle.Diamond, xname = "Players", yname = "Ratings",
    title = "Players' Ratings", isValueShownAsLabel = true, color = Color.Red) |> display

lc.stepline(x = x, y = y, markerSize = 10, markerStyle = MarkerStyle.Diamond, xname = "Players", yname = "Ratings",
    title = "Players' Ratings", isValueShownAsLabel = true, color = Color.Red) |> display

lc.bar(y = y, xname = "Players", yname = "Ratings", title = "Players' Ratings", isValueShownAsLabel = true,
    drawingStyle = "Emboss") |> display

lc.column(y = y, xname = "Players", yname = "Ratings", title = "Players' Ratings",
    isValueShownAsLabel = true, drawingStyle = "Cylinder") |> display

lc.boxplot(y = y, xname = "Players", yname = "Ratings", title = "Players' Ratings", color = Color.Blue, whiskerPercentile = 5, percentile = 30,
    showAverage = false, showMedian = false, showUnusualValues = true) |> display

Ok, the last one is weird. You probably want more than one boxplot in a chart. I'll show you how to do that in the next post.

The next post will be on how to have more than one series on the same chart and more than one chart in the same windows. Something like the below:

Excel Financial Functions upgraded to Visual Studio 2010 RC

lucabol — Tue, 16 Feb 2010 22:55:00 GMT

Here is a link to the home page: http://code.msdn.microsoft.com/FinancialFunctions/Wiki/View.aspx?title=Home

New version of LAgent compatible with VS 2010 RC online

lucabol — Mon, 15 Feb 2010 17:36:54 GMT

My crazy agent framework now works with RC: http://code.msdn.microsoft.com/LAgent

A simpler F# MailboxProcessor

lucabol — Fri, 12 Feb 2010 23:29:46 GMT

I always forget the pattern to use to create a functioning MailboxProcessor in F#. I mean, which piece has to be async and how to structure the recursive loop. When I find myself in that kind of a situation situation, my instincts scream at me: “Wrap it and make it work how your mind expects it to work”. So here is a simplification of the paradigm.

Let’s see how some standard MailboxProcessor code looks like:

let counter0 =
    MailboxProcessor.Start(fun inbox ->
        let rec loop n =
            async { 
                    let! msg = inbox.Receive()
                    return! loop(n+msg) }
        loop 0)

This keeps a running sum of the messages it receives. The only part that is really unique to this guy is “n + msg”. All the rest is infrastructure.

You’d probably prefer to write code like the following:

let counter1 = MailboxProcessor.SpawnAgent( (fun msg n -> msg + n), 0)

Yep, just one line of code. But, is it possible? Let’s look at one way of doing it:

type AfterError<'state> =
| ContinueProcessing of 'state
| StopProcessing
| RestartProcessing
    
type MailboxProcessor<'a> with

    static member public SpawnAgent<'b>(messageHandler :'a->'b->'b,
                                        initialState : 'b, ?timeout:'b -> int,
                                        ?timeoutHandler:'b -> AfterError<'b>,
                                        ?errorHandler:
                                            Exception -> 'a option -> 'b -> AfterError<'b>)
                                        : MailboxProcessor<'a> =
        let timeout = defaultArg timeout (fun _ -> -1)
        let timeoutHandler = defaultArg timeoutHandler (fun state –>
                                                                  ContinueProcessing(state))
        let errorHandler = defaultArg errorHandler (fun _ _ state –>
                                                                  ContinueProcessing(state))
        MailboxProcessor.Start(fun inbox ->
            let rec loop(state) = async {
                let! msg = inbox.TryReceive(timeout(state))
                try
                    match msg with
                    | None      -> match timeoutHandler state with
                                    | ContinueProcessing(newState)    ->
                                                                   return! loop(newState)
                                    | StopProcessing        -> return ()
                                    | RestartProcessing     -> return! loop(initialState)
                    | Some(m)   -> return! loop(messageHandler m state)
                with
                | ex -> match errorHandler ex msg state with
                        | ContinueProcessing(newState)    -> return! loop(newState)
                        | StopProcessing        -> return ()
                        | RestartProcessing     -> return! loop(initialState)
                }
            loop(initialState))

The funny formatting is because I have to fit it in the small horizontal space of this blog. In any case, this is just a simple (?) wrapper for the MailboxProcessor pattern. The function takes two necessary parameters and two optional ones:

messageHandler: a function to execute when a message comes in, it takes the message and the current state as parameters and returns the new state.
initialState: the initial state for the MailboxProcessor
timeoutHandler: a function that is executed whenever a timeout occurs. It takes as a parameter the current state and returns one of ContinueProcessing(newState), StopProcessing or RestartProcessing
errorHandler: a function that gets call if an exception is generated inside the messageHandler function. It takes the exception, the message, the current state and returns ContinueProcessing(newState), StopProcessing or RestartProcessing

An example of how to use errorHandler to implement the CountingAgent in the Expert F# book follows:

type msg = Increment of int | Fetch of AsyncReplyChannel<int> | Stop

exception StopException

type CountingAgent() =
    let counter = MailboxProcessor.SpawnAgent((fun msg n ->
                    match msg with
                    | Increment m ->  n + m
                    | Stop -> raise(StopException)
                    | Fetch replyChannel ->
                        do replyChannel.Reply(n)
                        n
                  ), 0, errorHandler = (fun _ _ _ -> StopProcessing))
    member a.Increment(n) = counter.Post(Increment(n))
    member a.Stop() = counter.Post(Stop)
    member a.Fetch() = counter.PostAndReply(fun replyChannel -> Fetch(replyChannel))    
        
let counter2 = CountingAgent()
counter2.Increment(1)
counter2.Fetch()
counter2.Increment(2)
counter2.Fetch()
counter2.Stop()

Sometimes your agent doesn’t need a state, it is purely stateless. Something as simple as the following:

let echo = MailboxProcessor<_>.SpawnWorker(fun msg -> printfn "%s" msg)

You can easily make that happen by using this toned down version of an agent, called worker:

static member public SpawnWorker(messageHandler,  ?timeout, ?timeoutHandler,?errorHandler) =
    let timeout = defaultArg timeout (fun () -> -1)
    let timeoutHandler = defaultArg timeoutHandler (fun _ -> ContinueProcessing(()))
    let errorHandler = defaultArg errorHandler (fun _ _ -> ContinueProcessing(()))
    MailboxProcessor.SpawnAgent((fun msg _ -> messageHandler msg; ()),
                                 (), timeout, timeoutHandler,
                                 (fun ex msg _ -> errorHandler ex msg))

Given that they are parallel, you might want to run a whole bunch of them at the same time. You might want something that looks like a worker, but that, under the cover, execute each messageHandler in parallel. Something like:

type msg1 = Message1 | Message2 of int | Message3 of string
            
let a = MailboxProcessor.SpawnParallelWorker(function
                | Message1 -> printfn "Message1";
                | Message2 n -> printfn "Message2 %i" n;
                | Message3 _ -> failwith "I failed"
                , 10
                , errorHandler = (fun ex _ -> printfn "%A" ex; ContinueProcessing()))


a.Post(Message1)
a.Post(Message2(100))
a.Post(Message3("abc"))
a.Post(Message2(100))

In this example, the different messages, are likely to cause things to print out of order. Notice the number 10 above which is how many workers you want to process your messages. This is implemented by round-robin messages to the various workers:

static member public SpawnParallelWorker(messageHandler, howMany, ?timeout,
                                                        ?timeoutHandler,?errorHandler) =
    let timeout = defaultArg timeout (fun () -> -1)
    let timeoutHandler = defaultArg timeoutHandler (fun _ -> ContinueProcessing(()))
    let errorHandler = defaultArg errorHandler (fun _ _ -> ContinueProcessing(()))
    MailboxProcessor<'a>.SpawnAgent((fun msg (workers:MailboxProcessor<'a> array, index) ->
                                        workers.[index].Post msg
                                        (workers, (index + 1) % howMany))  
                                    , (Array.init howMany
                                      (fun _ -> MailboxProcessor<'a>.SpawnWorker(
                                                 messageHandler, timeout, timeoutHandler,
                                                 errorHandler)), 0))

One drawback with the current code is that it doesn’t supports cancellations. It should be possible to wrap that too, but I haven’t done it (yet). If you don’t want to cut and paste the code, it is inside the AgentSystem.fs file here.

The F#unctional Londoners Meetup Group

lucabol — Thu, 11 Feb 2010 17:24:05 GMT

If you live in London and work with F#, you should join http://www.meetup.com/FSharpLondon/

Also, if you don’t work with F#, but you’d like too, and you live in London you should join it. Also if you don’t live in London, but you’d like to and you’d like not to work on F# … Ok, you get the point.

LAgent: an agent framework in F# – Part X – ActiveObject

lucabol — Thu, 10 Dec 2009 18:09:14 GMT

Download framework here.

All posts are here:

If you stare long enough at agents, you start to realize that they are just ‘glorified locks’. They are a convenient programming model to protect a resource from concurrent access. The programming model is convenient because both the client and the server can write their code without worrying about concurrency problems, and yet the program runs in parallel. Protecting a resource sounds a lot like state encapsulation and the concept of state encapsulation is what object orientation is all about.

So you start thinking if there is a way to enhance vanilla objects to make them agents. You want to reuse all the concepts that you are familiar with (i.e. inheritance, visibility rules, etc…) and you want your clients to call agents as if they were calling normal objects. Obviously, under the cover, the method calls won’t execute immediately, but they would be queued. Let’s look at an example.

This is our simple counter agent:

type CounterMessage =
| Add of int
| Print

let counterF = fun msg count ->
    match msg with
    | Add(i)    -> count + i
    | Print     -> printfn "The value is %i" count; count
    
let c1 = spawnAgent counterF 0
c1 <-- Add(3)
c1 <—Print

As nice as this looks, there are unfamiliar things in this model:

The communication is through messages. This requires packing and unpacking which, albeit easy in F#, is unfamiliar and feels like machinery that we’d like to get rid off.
The management of state is bizarre, it gets passed into the lambda and returned from it instead of being represented as fields and properties on the agent

My best attempt at creating an object-like syntax follows:

type Counter() =
    let w = new WorkQueue()
    let mutable count = 0
    member c.Add x = w.Queue (fun () -> 
        count <- count + x
        )
    member c.Print () = w.Queue (fun () -> 
        printfn "The value is %i" count
        )

let c = new Counter()
c.Add 3
c.Print

With this syntax, you write your agents like you write your vanilla classes except:

You need a private field of type WorkQueue
You need to write your methods as lambdas passed to the WorkQueue.Queue function
Your methods cannot return values

The most worrisome of these constraints is 2. because you can easily forget about it. If you do forget, then everything compiles just fine, but it doesn’t do what you expect. That’s pure badness. I haven’t found a way to enforce it. This is a place where the language could help me. Other than that, the whole model works rather nicely.

Regarding the third point, you can concoct a programming model that allows you to return values from your methods. Here it is:

member c.CountTask = w.QueueWithTask(fun () ->
    count
    )
member c.CountAsync = w.QueueWithAsync(fun () ->
    count
    )

printfn "The count using Task is %i" (c.CountTask.Result)

The first method returns a Task; the second method returns an AsyncResultCell. Both are ways to represent a promise. The latter allows a natural integration with the async block in F# as in the following code:

Async.RunSynchronously (
            async {
                let! count = c.CountAsync
                printfn "The countusing Async is %i" count
            })

As for myself, I don’t like methods returning values. Every time I use them, I end up going back and thinking about my problem in a traditional way, aka as method calls that return results, instead of thinking about it in a more actor oriented fashion. I end up waiting for these promises to be materialized and, by doing so, I limit the amount of parallelism that I unleash. As a matter of fact, the whole business of hiding the message passing nature of the programming model is dubious. It makes for a nicer syntax, but you need to make an extra effort in your mind to translate it to what it really is: just message passing with a nice syntactical veneer. I haven’t decided yet which model I like the most.

You should have a sense of what WorkQueue is. In essence, it is a Mailbox of lambdas (look at the red bold code below).

type WorkQueue() =
    let workQueue = spawnWorker (fun f -> f())
    member w.Queue (f) = workQueue <-- f
    member w.QueueWithTask f : Task<'T> =
        let source = new TaskCompletionSource<_>()
        workQueue <-- (fun () -> f() |> source.SetResult)
        source.Task
    member w.QueueWithAsync (f:unit -> 'T) : Async<'T> =
        let result = new AsyncResultCell<'T>()
        workQueue <-- (fun () -> f() |> result.RegisterResult )
        result.AsyncWaitResult
    member w.Restart () = workQueue <-! Restart
    member w.Stop () = workQueue <-! Stop
    member w.SetErrorHandler(h) =
        let managerF = fun (_, name:string, ex:Exception, _, _, _) -> h name ex                             
        let manager = spawnWorker managerF
        workQueue <-! SetManager manager
    member w.SetName(name) = workQueue <-! SetName(name)
    member w.SetQueueHandler(g) = workQueue <-! SetWorkerHandler(g)
    member w.SetTimeoutHandler(timeout, f) = workQueue <-! SetTimeoutHandler(timeout, f)

I implemented all the services that are in the message passing model. The two are equivalent as expressing power goes. In case you wonder how a real piece of code looks like using this model, here is an ActiveObject version of the map reduce algorithm. One of these days, I will gather the strength to go trough this code and explain what it does, but not today :-)

#load "AgentSystem.fs"
open AgentSystem.LAgent
open System
open System.Collections
open System.Collections.Generic
open System.Threading

type IOutput<'out_key, 'out_value> =
    abstract Reduced: 'out_key -> seq<'out_value> -> unit
    abstract MapReduceDone: unit -> unit
    
type Mapper<'in_key, 'in_value, 'my_out_key, 'out_value when 'my_out_key : comparison>
                                                      (map:'in_key -> 'in_value -> seq<'my_out_key * 'out_value>, i, partitionF) =
    let w = new WorkQueue()
    let mutable reducerTracker: BitArray = null
    let mutable controller = Unchecked.defaultof<Controller<'in_key, 'in_value, 'my_out_key, 'out_value>>
    let mutable reducers = Unchecked.defaultof<Reducer<'in_key, 'in_value, 'my_out_key, 'out_value> array>
    member m.Init c reds =
        w.Queue (fun () ->
            controller <- c
            reducers <- reds
            reducerTracker <- new BitArray(reducers.Length, false))
    member m.Process inKey inValue =
        w.Queue (fun () ->
            let outKeyValues = map inKey inValue
            outKeyValues |> Seq.iter (fun (outKey, outValue) ->
                                        let reducerUsed = partitionF outKey (reducers.Length)
                                        reducerTracker.Set(reducerUsed, true)
                                        reducers.[reducerUsed].Add(outKey, outValue)))
    member m.Done () =
        w.Queue (fun () ->
            controller.MapDone i reducerTracker)
    member m.Stop () = w.Stop ()            
    
and Reducer<'in_key, 'in_value, 'out_key, 'out_value when 'out_key :
                     comparison>(reduce:'out_key -> seq<'out_value> -> seq<'out_value>, i, output:IOutput<'out_key, 'out_value>) =
    let w = new WorkQueue()
    let mutable workItems = new List<'out_key * 'out_value>()
    let mutable controller = Unchecked.defaultof<Controller<'in_key, 'in_value, 'out_key, 'out_value>>
    member r.Init c =
        w.Queue (fun () ->
            controller <- c)
    member r.StartReduction () =
        w.Queue (fun () ->
            workItems
            |> Seq.groupBy fst
            |> Seq.sortBy fst
            |> Seq.map (fun (key, values) -> (key, reduce key (values |> Seq.map snd)))
            |> Seq.iter (fun (key, value) -> output.Reduced key value)
            controller.ReductionDone i) 
    member r.Add (outKey:'out_key, outValue:'out_value) : unit =
        w.Queue (fun () ->
            workItems.Add((outKey, outValue)))
    member m.Stop () = w.Stop ()            
                
and Controller<'in_key, 'in_value, 'out_key, 'out_value when 'out_key : comparison>(output:IOutput<'out_key, 'out_value>) =
    let w = new WorkQueue()
    let mutable mapperTracker: BitArray = null
    let mutable reducerUsedByMappers: BitArray = null
    let mutable reducerDone: BitArray = null
    let mutable mappers = Unchecked.defaultof<Mapper<'in_key, 'in_value, 'out_key, 'out_value> array>
    let mutable reducers = Unchecked.defaultof<Reducer<'in_key, 'in_value, 'out_key, 'out_value> array>
    let BAtoSeq (b:BitArray) = [for x in b do yield x]
    member c.Init maps reds =
        w.Queue (fun () ->
            mappers <- maps
            reducers <- reds
            mapperTracker <- new BitArray(mappers.Length, false)
            reducerUsedByMappers <- new BitArray(reducers.Length, false)
            reducerDone <- new BitArray(reducers.Length, false))
    member c.MapDone (i : int) (reducerTracker : BitArray) : unit =
        w.Queue (fun () ->
            mapperTracker.Set(i, true)
            let reducerUsedByMappers = reducerUsedByMappers.Or(reducerTracker)
            if not( BAtoSeq mapperTracker |> Seq.exists(fun bit -> bit = false)) then
                BAtoSeq reducerUsedByMappers |> Seq.iteri (fun i r -> if r = true then reducers.[i].StartReduction ())
                mappers |> Seq.iter (fun m -> m.Stop ())
              )
    member c.ReductionDone (i: int) : unit =
        w.Queue (fun () ->
            reducerDone.Set(i, true)
            if BAtoSeq reducerDone |> Seq.forall2 (fun x y -> x = y) (BAtoSeq reducerUsedByMappers) then
                output.MapReduceDone ()
                reducers |> Seq.iter (fun r -> r.Stop ())
                c.Stop()
             )         
    member m.Stop () = w.Stop ()
                
let mapReduce   (inputs:seq<'in_key * 'in_value>)
                (map:'in_key -> 'in_value -> seq<'out_key * 'out_value>)
                (reduce:'out_key -> seq<'out_value> -> seq<'out_value>)
                (output:IOutput<'out_key, 'out_value>)
                M R partitionF =
                    
    let len = inputs |> Seq.length
    let M = if len < M then len else M
    
    let mappers = Array.init M (fun i -> new Mapper<'in_key, 'in_value, 'out_key, 'out_value>(map, i, partitionF))
    let reducers = Array.init R (fun i -> new Reducer<'in_key, 'in_value, 'out_key, 'out_value>(reduce, i, output))
    let controller = new Controller<'in_key, 'in_value, 'out_key, 'out_value>(output)
    
    mappers |> Array.iter (fun m -> m.Init controller reducers)
    reducers |> Array.iter (fun r -> r. Init controller )
    controller.Init mappers reducers
    
    inputs |> Seq.iteri (fun i (inKey, inValue) -> mappers.[i % M].Process inKey inValue)
    mappers |> Seq.iter (fun m -> m.Done ())    

let partitionF = fun key M -> abs(key.GetHashCode()) % M 

let map = fun (fileName:string) (fileContent:string) ->
            let l = new List<string * int>()
            let wordDelims = [|' ';',';';';'.';':';'?';'!';'(';')';'\n';'\t';'\f';'\r';'\b'|]
            fileContent.Split(wordDelims) |> Seq.iter (fun word -> l.Add((word, 1)))
            l :> seq<string * int>
                                  
let reduce = fun key (values:seq<int>) -> [values |> Seq.sum] |> seq<int>

let printer () =
  { new IOutput<string, int> with
        member o.Reduced key values = printfn "%A %A" key values
        member o.MapReduceDone () = printfn "All done!!"}
    
let testInput =
     ["File1", "I was going to the airport when I saw someone crossing"; "File2", "I was going home when I saw you coming toward me"]   
mapReduce testInput map reduce (printer ()) 2 2 partitionF

open System.IO
open System.Text

let gatherer(step) =
  let w = new WorkQueue()
  let data = new List<string * int>()
  let counter = ref 0 
  { new IOutput<string, int> with
        member o.Reduced key values =
            w.Queue (fun () ->
                if !counter % step = 0 then
                    printfn "Processed %i words. Now processing %s" !counter key 
                data.Add((key, values |> Seq.hd))
                counter := !counter + 1)
        member o.MapReduceDone () =
            w.Queue (fun () ->
                data
                |> Seq.distinctBy (fun (key, _) -> key.ToLower())
                |> Seq.filter (fun (key, _) -> not(key = "" || key = "\"" || (fst (Double.TryParse(key)))))
                |> Seq.to_array
                |> Array.sortBy snd
                |> Array.rev
                |> Seq.take 20
                |> Seq.iter (fun (key, value) -> printfn "%A\t\t%A" key value)
                printfn "All done!!")           
        }
                        
let splitBook howManyBlocks fileName =
    let buffers = Array.init howManyBlocks (fun _ -> new StringBuilder())
    fileName
    |> File.ReadAllLines
    |> Array.iteri (fun i line -> buffers.[i % (howManyBlocks)].Append(line) |> ignore)
    buffers

let blocks1 = __SOURCE_DIRECTORY__ + "\kjv10.txt" |> splitBook 100
let blocks2 = __SOURCE_DIRECTORY__ + "\warandpeace.txt" |> splitBook 100
let input =
    blocks1
    |> Array.append blocks2
    |> Array.mapi (fun i b -> i.ToString(), b.ToString())
        
//mapReduce input map reduce (gatherer(1000)) 20 20 partitionF

type BookSplitter () =
    let blocks = new List<string * string>()
    member b.Split howManyBlocks fileName =
            let b =
                fileName
                |> splitBook howManyBlocks
                |> Array.mapi (fun i b -> i.ToString(), b.ToString())
            blocks.AddRange(b)
    member b.Blocks () =
            blocks.ToArray() :> seq<string * string>

type WordCounter () =
    let w = new WorkQueue()
    let words = new Dictionary<string,int>()
    let worker(wordCounter:WordCounter, ev:EventWaitHandle) =
          let w1 = new WorkQueue()
          { new IOutput<string, int> with
                member o.Reduced key values =
                    w1.Queue (fun() ->
                        wordCounter.AddWord key (values |> Seq.hd))
                member o.MapReduceDone () =
                    w1.Queue(fun () ->
                        ev.Set() |> ignore)
           }
    member c.AddWord word count =
            let exist, value = words.TryGetValue(word)
            if exist then
                words.[word] <- value + count
            else
                words.Add(word, count)
    member c.Add fileName =
        w.Queue (fun () ->
            let s = new BookSplitter()
            fileName |> s.Split 100
            let ev = new EventWaitHandle(false, EventResetMode.AutoReset)
            let blocks = s.Blocks ()
            mapReduce blocks map reduce (worker(c, ev)) 20 20 partitionF
            ev.WaitOne() |> ignore
            )  
    member c.Words =
        w.QueueWithAsync (fun () ->
            words |> Seq.to_array |> Array.map (fun kv -> kv.Key, kv.Value)
        )    

let wc = new WordCounter()
wc.Add (__SOURCE_DIRECTORY__ + "\kjv10.txt")
wc.Add (__SOURCE_DIRECTORY__ + "\warandpeace.txt")

let wordsToPrint = async {
                    let! words = wc.Words
                    return words
                        |> Seq.distinctBy (fun (key, _) -> key.ToLower())
                        |> Seq.filter (fun (key, _) -> not(key = "" || key = "\"" || (fst (Double.TryParse(key)))))
                        |> Seq.to_array
                        |> Array.sortBy snd
                        |> Array.rev
                        |> Seq.take 20
                        |> Seq.iter (fun (key, value) -> printfn "%A\t\t%A" key value)}


Async.RunSynchronously wordsToPrint
                        
Thread.Sleep(15000)
printfn "Closed session"

My PDC session is online - “Future directions for C# and Visual Basic”

lucabol — Mon, 23 Nov 2009 21:17:44 GMT

In case you are training to play the part of a first generation Italian immigrant in a Broadway show or if you want to understand why I never short msft stock: http://microsoftpdc.com/Sessions/FT11

Luca at PDC 2009 next week

lucabol — Fri, 13 Nov 2009 18:49:18 GMT

I’ll be in Los Angeles next week for PDC 2009. My session is called “Future Directions for C# and Visual Basic” and it comes on Tuesday immediately after the first keynote.

I’m planning on spending the first half of the session talking about the biggest trends influencing C# and VB. The second part will be about more future looking features we are playing with these days.

My deck has just two slides, as always. It’s all about watching me typing code on stage, hoping that I make crazy errors …

There are several other interesting language related sessions at PDC. Here is a list of them (please notice that the times might change). Also feel free to drop by the languages booth and chat. My team and I will be there most of the time.

Day	Time	Title	Presenter	Room
Tues 11/17	11:00 - 12:00	Future Directions for C# and Visual Basic	Luca Bolognese	Hall F
Tues 11/17	11:00 - 12:00	Accelerated Windows Application Development with Microsoft Visual C++ 2010	Boris Jabes	408B
Tues 11/17	12:30 - 13:15	Dynamic Binding in C# 4	Mads Torgersen	Hall F
Tues 11/17	12:30 - 13:15	Using Dynamic Languages to Build Scriptable Applications	Dino Viehland	403AB
Tues 11/17	13:30 - 14:30	C++ Forever: Interactive Applications in the Age of Manycore	Rick Molloy	515B
Tues 11/17	16:30 - 17:30	Manycore and the Microsoft .NET Framework 4: A Match Made in Microsoft Visual Studio 2010	Stephen Toub	502A
Tues 11/17	16:30 - 17:30	Code Contracts and Pex: Power Charge Your Assertions and Unit Tests	Mike Barnett, Nikolai Tillmann	408A
Wed 11/18	11:30 - 12:30	Microsoft Perspectives on the Future of Programming	Panel	Petree Hall C
Wed 11/18	12:00 - 13:00	F# for Architects – Hitting the Sweet Spot	Chris Smith	VS and .NET Pavillion
Wed 11/18	13:00 - 13:45	Code Like the Wind with Microsoft Visual Basic 2010	Lucian Wischik	Petree Hall D
Wed 11/18	13:00 - 13:45	Future of Garbage Collection	Patrick Dussud	Petree Hall C
Wed 11/18	13:00 - 13:45	Microsoft Project Code Name “M”: The Data and Modeling Language	Don Box, Jeff Pinkston	408A
Thu 11/19	08:30 - 09:30	PLINQ: LINQ, but Faster!	Ed Essey, Igor Ostrovsky	515A
Thu 11/19	10:00 - 11:00	A Lap around Microsoft Visual Studio 2010 for the Visual Basic Developer	Lisa Feigenbaum	VS and .NET Pavillion
Thu 11/19	10:00 - 11:00	Axum: A .NET Language for Safe and Scalable Concurrency	Niklas Gustafsson	515A
Thu 11/19	11:30 - 12:30	F# for Parallel and Asynchronous Programming	Luke Hoban	515A
Thu 11/19	12:45 - 13:30	Microsoft Visual C# IDE Tips and Tricks	DJ Park	Petree Hall D
Thu 11/19	12:45 - 13:30	Microsoft Visual Basic IDE Tips and Tricks	Dustin Campbell	Petree Hall C

Becoming really rich with C#

lucabol — Wed, 23 Sep 2009 02:40:00 GMT

Or maybe not, please do not hold me responsible if you lose money following this system. Having said that, it is my opinion that there are very few concepts that are important in investing. Three big ones are value, diversification and momentum. This post is about the latter two and how to use C# to create a simple trading system that uses both.

Diversification is ‘not put all your eggs in one basket’ (contrary to ‘put all of them in one basket and watch that basket’). I don’t believe you can ‘watch’ very much in financial markets, so I tend to prefer diversification.

Momentum is a mysterious tendency of financial prices that have risen the most in the recent past, to continue outperforming in the close future. In essence, buying the top stocks/sectors/asset classes tends to outperform buying the bottom ones over horizons from three months to one year.

The idea then is to rank some assets (i.e. ETFs) by how fast they have risen in the past, go long the top ones and short the bottom ones. There are hundreds of variations of this basic strategy, we’ll add the rule that we won’t buy assets that are below their 200 days moving average or sell short assets that are above it.

I’m writing this code with VS 2010 Beta 2 (which hasn’t shipped yet). It should be trivial to modify it to run on B1 (or maybe it does run on it already). I attach the code and data files to this post.

struct Event {
    internal Event(DateTime date, double price) { Date = date; Price = price; }
    internal readonly DateTime Date;
    internal readonly double Price;
}

We’ll use this simple structure to load the closing price for a particular date. My use of internal is kind of bizarre. Actually the whole code might look strange. It is an interesting (maybe un-elegant) mix of object orientation and functional programming.

class Summary {
    internal Summary(string ticker, string name, string assetClass,
                    string assetSubClass, double? weekly, double? fourWeeks,
                    double? threeMonths, double? sixMonths, double? oneYear,
                    double? stdDev, double price, double? mav200) {
        Ticker = ticker;
        Name = name;
        AssetClass = assetClass;
        AssetSubClass = assetSubClass;
        // Abracadabra ...
        LRS = (fourWeeks + threeMonths + sixMonths + oneYear) / 4;
        Weekly = weekly;
        FourWeeks = fourWeeks;
        ThreeMonths = threeMonths;
        SixMonths = sixMonths;
        OneYear = oneYear;
        StdDev = stdDev;
        Mav200 = mav200;
        Price = price;
    }
    internal readonly string Ticker;
    internal readonly string Name;
    internal readonly string AssetClass;
    internal readonly string AssetSubClass;
    internal readonly double? LRS;
    internal readonly double? Weekly;
    internal readonly double? FourWeeks;
    internal readonly double? ThreeMonths;
    internal readonly double? SixMonths;
    internal readonly double? OneYear;
    internal readonly double? StdDev;
    internal readonly double? Mav200;
    internal double Price;

    internal static void Banner() {
        Console.Write("{0,-6}", "Ticker");
        Console.Write("{0,-50}", "Name");
        Console.Write("{0,-12}", "Asset Class");
        //Console.Write("{0,-30}\t", "Asset SubClass";
        Console.Write("{0,4}", "RS");
        Console.Write("{0,4}", "1Wk");
        Console.Write("{0,4}", "4Wk");
        Console.Write("{0,4}", "3Ms");
        Console.Write("{0,4}", "6Ms");
        Console.Write("{0,4}", "1Yr");
        Console.Write("{0,6}", "Vol");
        Console.WriteLine("{0,2}", "Mv");
        //Console.Write("{0,6}", "Pr");
        //Console.WriteLine("{0,6}", "M200");
    }

    internal void Print() {

        Console.Write("{0,-6}", Ticker);
        Console.Write("{0,-50}", new String(Name.Take(48).ToArray()));
        Console.Write("{0,-12}", new String(AssetClass.Take(10).ToArray()));
        //Console.Write("{0,-30}\t", new String(AssetSubClass.Take(28).ToArray()));
        Console.Write("{0,4:N0}", LRS * 100);
        Console.Write("{0,4:N0}", Weekly * 100);
        Console.Write("{0,4:N0}", FourWeeks * 100);
        Console.Write("{0,4:N0}", ThreeMonths * 100);
        Console.Write("{0,4:N0}", SixMonths * 100);
        Console.Write("{0,4:N0}", OneYear * 100);
        Console.Write("{0,6:N0}", StdDev * 100);
        if (Price <= Mav200)
            Console.WriteLine("{0,2}", "X");
        else
            Console.WriteLine();
        //Console.Write("{0,6:N2}", Price);
        //Console.WriteLine("{0,6:N2}", Mav200);
    }
}

The class Summary above is how I want to present my results. A few comments on the code. I use Nullable<T> because some of this values can be null (i.e. not enough history), but I still don’t want to worry about it. It ends up working rather neatly.

I also print the results out to Console, which is crazy. I really should be using WPF/Silverlight as the presentation layer. Also the {0,4:N0} notation might be unfamiliar to some of you, but this is how mad Console guys like myself avoid using real UI frameworks. Sometimes we print things in color too.

The real meat is in the following line:

LRS = (fourWeeks + threeMonths + sixMonths + oneYear) / 4;

That is our highway to richness. It’s a very elaborated quant formula, never before shown, that calculate a magick relative strength (aka momentum) factor as the average of the performance of four weeks, three months, six months and one year.

class TimeSeries {
    internal readonly string Ticker;
    readonly DateTime _start;
    readonly Dictionary<DateTime, double> _adjDictionary;
    readonly string _name;
    readonly string _assetClass;
    readonly string _assetSubClass;

    internal TimeSeries(string ticker, string name, string assetClass, string assetSubClass, 
                                                                IEnumerable<Event> events) {
        Ticker = ticker;
        _name = name;
        _assetClass = assetClass;
        _assetSubClass = assetSubClass;
        _start = events.Last().Date;
        _adjDictionary = events.ToDictionary(e => e.Date, e => e.Price);
    }

I then built myself a little TimeSeries class that represents a series of (date, price). I choose a dictionary to store it because of my assumption that I will be accessing it by date a lot. In retrospect, I was kind of right and kind of wrong. It doesn’t really matter much.

bool GetPrice(DateTime when, out double price, out double shift) {
    // To nullify the effect of hours/min/sec/millisec being different from 0
    when = new DateTime(when.Year, when.Month, when.Day);
    var found = false;
    shift = 1;
    double aPrice = 0;
    while (when >= _start && !found) {
        if (_adjDictionary.TryGetValue(when, out aPrice)) {
            found = true;
        }
        when = when.AddDays(-1);
        shift -= 1;
    }
    price = aPrice;
    return found;
}

A TimeSeries can give you back the price at a particular date. This looks bizarre and complex, but there is a reason for it. I might ask for a date that doesn’t have a price associated with it (i.e. holidays, week-ends). In such cases I want to return the previous price which could be N days in the past.

I also want to return how many days in the past I had to go, so that other calculations (i.e. Return) can modify their end date by the same amount. Also I might not find such a price at all, in which case I don’t want to throw an exception, but instead notify the caller. In retrospect, I should have used double? to signify ‘price not found’.

double? GetReturn(DateTime start, DateTime end) {
    var startPrice = 0.0;
    var endPrice = 0.0;
    var shift = 0.0;
    var foundEnd = GetPrice(end, out endPrice, out shift);
    var foundStart = GetPrice(start.AddDays(shift), out startPrice, out shift);
    if (!foundStart || !foundEnd)
        return null;
    else
        return endPrice / startPrice - 1;
}

We can now go and calculate the return between two dates. Also the TimeSeries object needs to perform a little more calculations.

    internal double? LastWeekReturn() {
        return GetReturn(DateTime.Now.AddDays(-7), DateTime.Now);
    }
    internal double? Last4WeeksReturn() {
        return GetReturn(DateTime.Now.AddDays(-28), DateTime.Now);
    }
    internal double? Last3MonthsReturn() {
        return GetReturn(DateTime.Now.AddMonths(-3), DateTime.Now);
    }
    internal double? Last6MonthsReturn() {
        return GetReturn(DateTime.Now.AddMonths(-6), DateTime.Now);
    }
    internal double? LastYearReturn() {
        return GetReturn(DateTime.Now.AddYears(-1), DateTime.Now);
    }
    internal double? StdDev() {
        var now = DateTime.Now;
        now = new DateTime(now.Year, now.Month, now.Day);
        var limit = now.AddYears(-3);
        var rets = new List<double>();
        while (now >= _start.AddDays(12) && now >= limit) {
            var ret = GetReturn(now.AddDays(-7), now);
            rets.Add(ret.Value);
            now = now.AddDays(-7);
        }
        var mean = rets.Average();
        var variance = rets.Select(r => Math.Pow(r - mean, 2)).Sum();
        var weeklyStdDev = Math.Sqrt(variance / rets.Count);
        return weeklyStdDev * Math.Sqrt(40);
    }
    internal double? MAV200() {
        return _adjDictionary
               .ToList()
               .OrderByDescending(k => k.Key)
               .Take(200)
               .Average(k => k.Value);
    }
    internal double TodayPrice() {
        var price = 0.0;
        var shift = 0.0;
        GetPrice(DateTime.Now, out price, out shift);
        return price;
    }
    internal Summary GetSummary() {
        return new Summary(Ticker, _name, _assetClass, _assetSubClass, 
                           LastWeekReturn(), Last4WeeksReturn(), Last3MonthsReturn(),
                           Last6MonthsReturn(), LastYearReturn(), StdDev(), TodayPrice(),
                           MAV200());
    }
}

Nothing particularly interesting in this code. Just a bunch of calculations. The MAV200 is the 200 days moving average of closing prices. It shows a more functional way of doing things. The StdDev function is instead very imperative.

We now can work on downloading the prices. This is how you construct the right URL:

static string CreateUrl(string ticker, DateTime start, DateTime end) {
    return @"http://ichart.finance.yahoo.com/table.csv?s=" + ticker + "&a="
            + (start.Month - 1).ToString() + "&b=" + start.Day.ToString() + "&c="
            + start.Year.ToString() + "&d=" + (end.Month - 1).ToString() + "&e="
            + end.Day.ToString() + "&f=" + end.Year.ToString() + "&g=d&ignore=.csv";
}

And let’s set how many concurrent connections we are going to use …

ServicePointManager.DefaultConnectionLimit = 10;

On my machine, setting this number too high causes errors to be returned. I’m not sure on which side of the connection the problem lies.

We can then load all the tickers we want to load from a file. One of the files has Leveraged ETFs, which I want to filter out because they tend to pop up always at the top.

var tickers =
    //File.ReadAllLines("ETFs.csv")
    //File.ReadAllLines("ETFTest.csv")
    File.ReadAllLines("AssetClasses.csv")
    .Skip(1)
    .Select(l => l.Split(new[] { ',' }))
    .Where(v => v[2] != "Leveraged")
    .Select(values => Tuple.Create(values[0], values[1], values[2], values[3]))
    .ToArray();
var len = tickers.Length;

var start = DateTime.Now.AddYears(-2);
var end = DateTime.Now;
var cevent = new CountdownEvent(len);
var summaries = new Summary[len];

And then load all of them, making sure to make an asynchronous call so not to keep the thread busy.

for(var i = 0; i < len; i++)  {
    var t = tickers[i];
    var url = CreateUrl(t.Item1, start, end);
    using (var webClient = new WebClient()) {
        webClient.DownloadStringCompleted +=
                          new DownloadStringCompletedEventHandler(downloadStringCompleted);
        webClient.DownloadStringAsync(new Uri(url), Tuple.Create(t, cevent, summaries, i));
    }
}

cevent.Wait();

Notice the use of a Countdown event to wait for all the thread to complete before printing out the results. Also notice the new Tuple<T> class used to package things to send around.

We can then print out the top and bottom 15%:

var top15perc =
        summaries
        .Where(s => s.LRS.HasValue)
        .OrderByDescending(s => s.LRS)
        .Take((int)(len * 0.15));
var bottom15perc =
        summaries
        .Where(s => s.LRS.HasValue)
        .OrderBy(s => s.LRS)
        .Take((int)(len * 0.15));

Console.WriteLine();
Summary.Banner();
Console.WriteLine("TOP 15%");
foreach(var s in top15perc)
    s.Print();

Console.WriteLine();
Console.WriteLine("Bottom 15%");
foreach (var s in bottom15perc)
    s.Print();

Here is what we do when a request comes back with data:

static void downloadStringCompleted(object sender, DownloadStringCompletedEventArgs e) {
    var bigTuple =
             (Tuple<Tuple<string, string, string, string>, CountdownEvent, Summary[], int>)
              e.UserState;
    var tuple = bigTuple.Item1;
    var cevent = bigTuple.Item2;
    var summaries = bigTuple.Item3;
    var i = bigTuple.Item4;
    var ticker = tuple.Item1;
    var name = tuple.Item2;
    var asset = tuple.Item3;
    var subAsset = tuple.Item4;

    if (e.Error == null) {
        var adjustedPrices =
                e.Result
                .Split(new[] { '\n' })
                .Skip(1)
                .Select(l => l.Split(new[] { ',' }))
                .Where(l => l.Length == 7)
                .Select(v => new Event(DateTime.Parse(v[0]), Double.Parse(v[6])));

        var timeSeries = new TimeSeries(ticker, name, asset, subAsset, adjustedPrices);
        summaries[i] = timeSeries.GetSummary();
        cevent.Signal();
        Console.Write("{0} ", ticker);
    }
    else {
        Console.WriteLine("[{0} ERROR] ", ticker);
        //Console.WriteLine(e.Error);
        summaries[i] = new Summary(ticker, name, "ERROR", "ERROR", 0, 0, 0, 0, 0, 0,0,0); 
        cevent.Signal();
    }
}

We first unpack the Tuple we sent out originally, we then extract the Date and Price, create a Summary object and store it in the summaries array. It’s important to remember to Signal to the cevent in the error case as well because we want to print out the results even if some downloading failed.

And here is what you get for your effort:

LAgent: an agent framework in F# – Part IX – Counting words …

lucabol — Fri, 18 Sep 2009 21:43:00 GMT

Download framework here.

All posts are here:

Let’s now use our mapReduce to do something more interesting, for example finding the frequency of words in several books. Now the agent that processes the output needs to be a bit more complex.

let gathererF = fun msg (data:List<string * int>, counter, step) ->
                    match msg with
                    | Reduced(key, value)   ->
                        if counter % step = 0 then
                            printfn "Processed %i words. Now processing %s" counter key 
                        data.Add((key, value |> Seq.hd))
                        data, counter + 1, step
                    | MapReduceDone         ->
                        data
                        |> Seq.distinctBy (fun (key, _) -> key.ToLower())
                        |> Seq.filter (fun (key, _) -> not(key = "" || key = "\"" ||
                                                             (fst (Double.TryParse(key)))))
                        |> Seq.to_array
                        |> Array.sortBy snd
                        |> Array.rev
                        |> Seq.take 20
                        |> Seq.iter (fun (key, value) -> printfn "%A\t\t%A" key value)
                        printfn "All done!!"
                        data, counter, step                           
                    
let gatherer = spawnAgent gathererF (new List<string * int>(), 0, 1000)

Every time a new word is reduced, a message is printed out and the result is added to a running list. When everything is done such a list is printed out by first manipulating it to reduce weirdness and limit the number of items. BTW: there are at least two bugs in this code, maybe more (late night quick-and-dirty-see-if-the-algo-works kind of coding).

We want to maximize the number of processors to use, so let’s split the books in chunks so that they can be operated in parallel. The code below roughly does it (I say roughly because it doesn’t chunk the lines in the right order, but for this particular case it doesn’t matter).

let gatherer = spawnAgent gathererF (new List<string * int>(), 0, 1000)

let splitBook howManyBlocks fileName =
    let buffers = Array.init howManyBlocks (fun _ -> new StringBuilder())
    fileName
    |> File.ReadAllLines
    |> Array.iteri (fun i line -> buffers.[i % (howManyBlocks)].Append(line) |> ignore)
    buffers

let blocks1 = "C:\Users\lucabol\Desktop\Agents\Agents\kjv10.txt" |> splitBook 100
let blocks2 = "C:\Users\lucabol\Desktop\Agents\Agents\warandpeace.txt" |> splitBook 100
let input =
    blocks1
    |> Array.append blocks2
    |> Array.mapi (fun i b -> i.ToString(), b.ToString())

And let’s execute!!

mapReduce input map reduce gatherer 20 20 partitionF

On my machine I get the following, which could be the right result.

"a"        16147
"And"        13071
"I"        11349
"unto"        8125
"as"        6400
"her"        5865
"which"        5544
"from"        5378
"at"        5175
"on"        5155
"have"        5135
"me"        5068
"my"        4629
"this"        3782
"out"        3653
"ye"        3399
"when"        3312
"an"        2841
"upon"        2558
"so"        2489
All done!!

LAgent: an agent framework in F# – Part VIII - Implementing MapReduce (user model)

lucabol — Fri, 04 Sep 2009 20:57:48 GMT

Download framework here.

All posts are here:

For this post I use a newer version of the framework that I just uploaded on CodeGallery. In the process of using LAgent I grew more and more unhappy with the weakly typed way of sending messages. The code that implements that feature is nasty: full of upcasts and downcasts. I was losing faith in it. Bugs were cropping up in all sorts of scenarios (i.e. using generic union types as messages).

In the end I decided to re-architecture the framework so to make it strongly typed. In essence now each agent can just receive messages of a single type. The limitations that this design choice introduces (i.e. more limited hot swapping) are compensated by the catching of errors at compile time and the streamlining of the code. I left the old framework on the site in case you disagree with me.

In any case, today’s post is about MapReduce. It assumes that you know what it is (link to the original Google paper that served as inspiration is here: Google Research Publication- MapReduce). What would it take to implement an in-memory MapReduce using my agent framework?

Let’s start with the user model.

let mapReduce   (inputs:seq<'in_key * 'in_value>)
                (map:'in_key -> 'in_value -> seq<'out_key * 'out_value>)
                (reduce:'out_key -> seq<'out_value> -> seq<'reducedValues>)
                outputAgent
                M R partitionF =

mapReduce takes seven parameters:

inputs: a sequence of input key/value pairs.
map: this function operates on each input key/value pair. It returns a sequence of output key/value pairs. The type of the output sequence can be different from the type of the inputs.
reduce: this function operates on an output key and all the values associated with it. It returns a sequence of reduced values (i.e. the average of all the values for this key)
ouputAgent: this is the agent that gets notified every time a new output key has been reduced and at the end when all the operation ends.
M: how many mapper agents to instantiate
R: how many reducer agents to instantiate
partitionF: the partition function used to choose which of the reducers is associated with a key

Let’s look at how to use this function to find how often each word is used in a set of files. First a simple partition function can be defined as:

let partitionF = fun key M -> abs(key.GetHashCode()) % M

Given a key and some buckets, it picks one of the buckets. Its type is: ‘a –> int –> int, so it’s fairly reusable.

Let’s also create a basic agent that just prints out the reduced values:

let printer = spawnWorker (fun msg ->
                            match msg with
                            | Reduced(key, value)   -> printfn "%A %A" key value
                            | MapReduceDone         -> printfn "All done!!")

The agent gets notified whenever a new key is reduced or the algorithm ends. It is useful to be notified immediately instead of waiting for everything to be done. If I hadn’t written this code using agents I would have not realized that possibility. I would simply have framed the problem as a function that takes an input and returns an output. Agents force you to think explicitly about the parallelism in your app. That’s a good thing.

The mapping function simply split the content of a file into words and adds a word/1 pair to the list. I know that there are much better ways to do this (i.e. regular expressions for the parsing and summing words counts inside the function), but I wanted to test the basic framework capabilities and doing it this way does it better.

let map = fun (fileName:string) (fileContent:string) ->
            let l = new List<string * int>()
            let wordDelims = [|' ';',';';';'.';':';'?';'!';'(';')';'\n';'\t';'\f';'\r';'\b'|]
            fileContent.Split(wordDelims) |> Seq.iter (fun word -> l.Add((word, 1)))
            l :> seq<string * int>

The reducer function simply sums the various word statistics sent by the mappers:

let reduce = fun key (values:seq<int>) -> [values |> Seq.sum] |> seq<int>

Now we can create some fake input to check that it works:

let testInput = ["File1", "I was going to the airport when I saw someone crossing";
                               "File2", "I was going home when I saw you coming toward me"]

And execute the mapReduce:

mapReduce testInput map reduce printer 2 2 partitionF

On my machine I get the following. You might get a different order because of the async/parallel processing involved. If I wanted a stable order I would need to change the printer agent to cache results on Reduced and process them on MapReduceDone (see next post).

"I" [4]
"crossing" [1]
"going" [2]
"home" [1]
"me" [1]
"the" [1]
"toward" [1]
"airport" [1]
"coming" [1]
"saw" [2]
"someone" [1]
"to" [1]
"was" [2]
"when" [2]
"you" [1]

In the next post we’ll process some real books …

LAgent: an agent framework in F# – part VII – An auction application

lucabol — Fri, 10 Jul 2009 23:14:00 GMT

Download framework here.

All posts are here:

Here is an application that uses the framework we have been creating. It is an auction application and it is described in more detail here.

Let’s go through it.

type AuctionMessage =
  | Offer of int * AsyncAgent // Make a bid
  | Inquire of AsyncAgent     // Check the status
and AuctionReply =
  | StartBidding
  | 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
      AsyncAgent * AsyncAgent
  | AuctionFailed            // Failed without any bids
  | AuctionOver              // Bidding is closed
  
let timeToShutdown = 3000
let bidIncrement = 10

This is the format of the messages that the clients can send and the action agent can reply to. F# is really good at this sort of thing. First, we need an auction agent:

let auctionAgent seller minBid closing =
    let agent = spawnAgent (fun msg (isConcluded, maxBid, maxBidder) ->
                            match msg with
                            | Offer (_, client) when isConcluded ->
                                client <-- AuctionOver
                                (isConcluded, maxBid, maxBidder)
                            | Offer(bid, client) when not(isConcluded) ->
                                if bid >= maxBid + bidIncrement then
                                    if maxBid >= minBid then maxBidder <-- BeatenOffer bid                  
                                    client <-- BestOffer
                                    (isConcluded, bid, client)
                                else
                                    client <-- BeatenOffer maxBid
                                    (isConcluded, maxBid, maxBidder)
                            | Inquire client    ->
                                client <-- Status(maxBid, closing)
                                (isConcluded, maxBid, maxBidder))
                            (false, (minBid - bidIncrement), spawnWorker (fun _ -> ()))

Notice that, if the action is concluded, the agent replies to offers by sending an AuctionOver message. If the auction is still open, then, in case the bid is higher than the max, it sets a new max and notify the two parties involved; otherwise it notifies the bidder that the offer wasn’t successful. Also you can ask for the status of the auction.

This is what the code above says. Maybe the code is simpler than words. Anyhow, we need to treat the case where no message is received for some amount of time.

agent <-- SetTimeoutHandler
            (closing - DateTime.Now).Milliseconds
            (fun (isConcluded: bool, maxBid, maxBidder) ->
                if maxBid >= minBid then
                  let reply = AuctionConcluded(seller, maxBidder)
                  maxBidder <-- reply
                  seller <-- reply
                else seller <-- AuctionFailed
                agent <-- SetTimeoutHandler
                    timeToShutdown
                    (fun (_:bool, _:int,_:AsyncAgent) -> StopProcessing)
                ContinueProcessing (true, maxBid, maxBidder))   
agent

We start by waiting for the amount of time to the closing of the auction. If we get no messages, then two things might happen: we have an offer that is more than the minimum or we don’t. If we do, we tell everyone that it’s finished. Otherwise, we tell the seller that its item wasn’t successful. In any case, we prepare the agent to shutdown by setting its next timeout to be timeoutToShutdown.

It is interesting that we set the timeout handler inside the timeout handler. This is not a problem because of the nature of message processing (aka it processes one message at the time).

We then need a bunch of of symbols …

module Auction =
  let random = new Random()
  
  let minBid = 100
  let closing = DateTime.Now.AddMilliseconds 10000.
  
  let seller = spawnWorker (fun (msg:AuctionReply) -> ())
  let auction = auctionAgent seller minBid closing

Not a very smart seller we have here … Next up is our definition of a client.

let rec c = spawnAgent (
                fun msg (max, current) ->
                    let processBid (aMax, aCurrent) =
                        if aMax >= top then
                            log "too high for me"
                            (aMax, aCurrent)
                        elif aCurrent < aMax then
                              let aCurrent = aMax + increment
                              Thread.Sleep (1 + random.Next 1000)
                              auction <-- Offer(aCurrent, c)
                              (aMax, aCurrent)
                        else (aMax, aCurrent)                       
                    match msg with
                    | StartBidding      ->
                        auction <-- Inquire c
                        (max, current)
                    | Status(maxBid,_)  ->
                        log <| sprintf "status(%d)" maxBid
                        let s = processBid (maxBid, current)
                        c <-- SetTimeoutHandler timeToShutdown (fun _ -> StopProcessing) 
                        s
                    | BestOffer ->
                        log <| sprintf "bestOffer(%d)" current
                        processBid(max, current)
                    | BeatenOffer maxBid ->
                        log <| sprintf "beatenOffer(%d)" maxBid
                        processBid(maxBid, current)
                    | AuctionConcluded(seller, maxBidder) ->
                        log "auctionConcluded"
                        c <-- Stop
                        (max, current)
                    | AuctionOver ->
                        log "auctionOver"
                        c <-- Stop
                        (max, current))
                 (0,0)
c

Something that I like about agents is the fact that you need to understand just small snippets of code at the time. For example, you can read the processing for BestOffer and figure out if it makes sense. I have an easy time personalizing them as in : “Ok, the guy just got a notification that there has been a better offer, what is he going to do next?”.

The code should be self explanatory for the most part. In essence, if you can offer more, do it otherwise wait for the auction to end. I’m not even sure the processing is completely right. I confess I’m just trying to do the same as Matthews code from the link above.

We can then start up the whole thing and enjoy the cool output.

open Auction

(client 1 20 200) <-- StartBidding
(client 2 10 300) <-- StartBidding
(client 3 30 150) <-- StartBidding
Console.ReadLine() |> ignore

Now for the nasty part. Implementing the framework.

LAgent: an agent framework in F# – Part VI – Hot swapping of code (and something silly)

lucabol — Fri, 03 Jul 2009 22:23:00 GMT

Download framework here.

All posts are here:

Hot swapping of code

Let’s get back a couple of steps and consider what happens when you get an error. Sure, your agent will continue processing messages, but it might be doing the wrong thing. Your message handling code might be buggy.

Ideally you’d want to patch things on the fly. You’d want to replace the message processing code for an agent without stopping it.

Here is how you do it:

let counter2 = spawnAgent (fun msg state -> printfn "From %i to %i" state (state + msg);
                                                                              state + msg) 0
counter2 <-- 2
                                         
counter2 <-- SetAgentHandler(fun msg state –>
                printfn "From %i to %i via multiplication" state (state * msg); msg * state)
counter2 <-- 3

Which generates:

From 0 to 2
From 2 to 6 via multiplication

After the agent receives a SetAgentHandler message, it switch from a ‘+’ agent to a ‘*’ agent on the fly!! All the messages that come after that one gets multiplied to the state. Also, the state is preserved between changes in behavior.

It might not be immediately apparent how to load a function at runtime, but it is really simple. Imagine that I get the data on the function to load from somewhere (i.e. a management console UI).

let assemblyNameFromSomewhere, typeNameFromSomewhere, methodNameFromSomewhere = 
                                                "mscorlib.dll", "System.Console", "WriteLine"

I can then use it to dynamically load a message handler (in this case Console.Writeline).

let a = Assembly.Load(assemblyNameFromSomewhere)
let c = a.GetType(typeNameFromSomewhere)
let m = c.GetMethod(methodNameFromSomewhere, [|"".GetType()|])
let newF = fun (msg:string) (state:obj) -> m.Invoke(null, [| (msg:>obj) |])

And then it is as simple as posting a SetAgentHandler.

counter2 <-- SetAgentHandler(newF)
counter2 <-- "blah"

Now our counter2 agent has become an echo agent on the fly, having loaded Console.WriteLine dynamically. Note how the agent moved from being a ‘+’ agent taking integers to being a ‘*’ agent taking integers to being an ‘echo’ agent taking strings. And it didn’t stop processing messages for the whole time.

Obviously, you can do the same thing with workers:

echo <-- SetWorkerHandler(fun msg -> printfn "I'm an echo and I say: %s" msg)
echo <-- "Hello"

And parallelWorkers:

parallelEcho <-- SetWorkerHandler(fun msg -> tprint ("I'm new and " + msg))
messages |> Seq.iter (fun msg -> parallelEcho <-- msg)

A silly interlude

As a way to show some agents talking to each other, here is a simple program that simulates marital interactions (of the worst kind):

let rec husband = spawnWorker (fun (To, msg) -> printfn "Husband says: %s" msg; To <-- msg)
let rec wife = spawnWorker (fun msg -> printfn "Wife says: screw you and your '%s'" msg)
husband <-- (wife, "Hello")
husband <-- (wife, "But darling ...")
husband <-- (wife, "ok")

Which produces:

Husband says: Hello
Husband says: But darling ...
Wife says: screw you and your 'Hello'
Wife says: screw you and your 'But darling ...'
Husband says: ok
Wife says: screw you and your 'ok'

And yes, you cannot expect messages to be in the right sequence … Next up is an auction application.

LAgent: an agent framework in F# – Part V – Timeout management

lucabol — Fri, 26 Jun 2009 21:56:00 GMT

Download framework here.

All posts are here:

Timeout management

Timeouts are very important in message based systems. In essence, if you are not getting messages for a certain period of time, that usually means something. It might be that something crashed, that other agents think that you are not online, or any other number of things. Hence the need to set timeouts and react when they are triggered.

You do that by writing the following:

counter1 <--SetTimeoutHandler 1000 
            (fun state -> printfn "I'm still waiting for a message in state %A, come on ..." 
                                                            state; ContinueProcessing(state))

Which generates the following message every second:

I'm still waiting for a message in state 2, come on ...
I'm still waiting for a message in state 2, come on .…

The first parameter to SetTimeoutHandler is how long to wait before triggering the handler. The second parameter is the handler that gets called whenever no message is received for that amount of time. Notice that the handler takes the current state of the agent and returns ContinueProcessing(state). This tells the agent to continue processing messages and sets the current state to state.

The following code:

counter1 <-- 2

Then generates:

I'm still waiting for a message in state 4, come on ...
I'm still waiting for a message in state 4, come on ...

ContinueProcessing is just one of the three possible values of the (terribly named) AfterError:

type AfterError =
| ContinueProcessing of obj
| StopProcessing
| RestartProcessing

Let’s see what RestartProcessing does.

counter1 <-- SetTimeoutHandler 1000  (fun state -> printfn "Restart from state %A" state
                                                                        ; RestartProcessing)

Which, as expected, generates a nice stream of:

Restart from state 0
Restart from state 0

To bring things back to normal (aka no timeout) you can just pass –1 as in:

counter1 <-- SetTimeoutHandler -1  (fun state -> ContinueProcessing(state))

Also, you can stop the agent when a timeout occurs by returning the aptly named StopProcessing:

counter1 <-- SetTimeoutHandler 1000  (fun state -> printfn "Restart from state %A" state; 
                                                                             StopProcessing)

Another interesting thing you might want to do is hot swapping of code. More on that in the next part …

LAgent: an agent framework in F# – Part IV – Custom error management

lucabol — Fri, 19 Jun 2009 21:36:00 GMT

Download framework here.

All posts are here:

Custom error management

In the last part we saw what happens by default in the framework when an error occurs. But that might not be what you want. You might want to have your sophisticated error detection and recovery distributed algorithm.

To make such a thing possible each agent has a manager. The manager is an agent that gets called whenever an error occurs in the agent it is monitoring.

In code:

let manager = spawnWorker (fun (agent, name:string, ex:Exception, msg:obj,
                 state, initialState) -> printfn "%s restarting ..." name; agent <-- Restart)
counter1 <-- SetManager(manager)

Whenever an error is generated the manager receives a tuple of:

(agent, name, exception, message, currentState, inititialState)

This manager prints out something and then restarts the agent. Let’s trigger an error by posting the wrong message:

counter1 <-- "afdaf"
counter1 <-- 2

The expectation is that the counter will restart from 0 whenever an error is triggered. This is what happens:

Bob restarting ...
From 0 to 2

Which is what we expected. Obviously this is not a very sophisticated error recovery algorithm. You might want to do something more meaningful. Hopefully you have enough information to build whatever you need.

A particularly important class of unexpected event is timeouts. We’ll talk about them next.

LAgent: an agent framework in F# – Part III – Default error management

lucabol — Fri, 12 Jun 2009 21:07:00 GMT

Download framework here.

All posts are here:

Default error management

What happens when an error occurs? Well, ideally you want to notify someone and continue processing messages. By default you want to print the error and as much information as you can about it.

Let’s first see what happens if you pass the wrong message type:

counter1 <-- "fst"

Generates:

> The exception below occurred on agent Undefined at state 3 with message "fst". The agent was started with state 0.
System.InvalidCastException: Specified cast is not valid.
   at Microsoft.FSharp.Core.LanguagePrimitives.IntrinsicFunctions.UnboxGeneric[T](Object source)
   at FSI_0003.AgentSystem.f@158.Invoke(Object a, Object b)
   at Microsoft.FSharp.Core.FastFunc`2.InvokeFast[V](FastFunc`2 func, T arg1, TResult arg2)
   at Microsoft.FSharp.Core.FastFunc`2.InvokeFast[V](FastFunc`2 func, T arg1, TResult arg2)
   at FSI_0003.AgentSystem.loop@20-3.Invoke(Unit unitVar)
   at Microsoft.FSharp.Control.AsyncBuilderImpl.callA@245.Invoke(AsyncParams`1 args)

You get information about the current state of the agent, the message that generated the error, the initial state of the agent and the exception that was generated. But, in a system with several agents, you’d like to know which one agent failed. Then you need to name your agent:

counter1 <-- SetName("Bob")
counter1 <-- "fadfad"

Now you get (important part in blue):

> The exception below occurred on agent Bob at state 3 with message "fadfad". The agent was started with state 0.
System.InvalidCastException: Specified cast is not valid.
   at Microsoft.FSharp.Core.LanguagePrimitives.IntrinsicFunctions.UnboxGeneric[T](Object source)
   at FSI_0003.AgentSystem.f@158.Invoke(Object a, Object b)
   at Microsoft.FSharp.Core.FastFunc`2.InvokeFast[V](FastFunc`2 func, T arg1, TResult arg2)
   at Microsoft.FSharp.Core.FastFunc`2.InvokeFast[V](FastFunc`2 func, T arg1, TResult arg2)
   at FSI_0003.AgentSystem.loop@20-3.Invoke(Unit unitVar)
   at Microsoft.FSharp.Control.AsyncBuilderImpl.callA@245.Invoke(AsyncParams`1 args)

The important thing is that the agent continues running. It lives to fight another day. Hence:

counter1 <-- 3

Produces:

From 3 to 6

Which shows that the agent is running and that it has kept its current state. Also errors can occur inside the message handler with a similar result:

(spawnAgent (fun msg state -> state / msg) 100) <-- 0

Produces:

> The exception below occurred on agent Undefined at state 100 with message 0. The agent was started with state 100.
System.DivideByZeroException: Attempted to divide by zero.
   at FSI_0013.it@48-3.Invoke(Int32 msg, Int32 state)
   at Microsoft.FSharp.Core.FastFunc`2.InvokeFast[V](FastFunc`2 func, T arg1, TResult arg2)
   at FSI_0003.AgentSystem.f@158.Invoke(Object a, Object b)
   at Microsoft.FSharp.Core.FastFunc`2.InvokeFast[V](FastFunc`2 func, T arg1, TResult arg2)
   at Microsoft.FSharp.Core.FastFunc`2.InvokeFast[V](FastFunc`2 func, T arg1, TResult arg2)
   at FSI_0003.AgentSystem.loop@20-3.Invoke(Unit unitVar)
   at Microsoft.FSharp.Control.AsyncBuilderImpl.callA@245.Invoke(AsyncParams`1 args)

But this might not be what you want. You might want to customize what happens when an error occurs. We’ll talk about that next.

LAgent : an agent framework in F# – Part II – Agents and control messages

lucabol — Fri, 05 Jun 2009 21:03:00 GMT

Download framework here.

All posts are here:

Agents

Agents are entities that process messages and keep state between one message and the next. As such they need to be initialized with a lambda that takes a message and a state and returns a new state. In F# pseudo code: msg –> state –> newState. For example the following:

let counter = spawnAgent (fun msg state -> state + msg) 0

This is a counter that starts from 0 and gets incremented by the value of the received message. Let’s make it print something when it receives a message:

let counter1 = spawnAgent
                (fun msg state -> printfn "From %i to %i" state (state + msg); state + msg) 0
counter1 <-- 3 
counter1 <-- 4

Which produces:

From 0 to 3
From 3 to 7

There is no spawnParallelAgent, because I couldn’t figure out its usage patterns. Maybe I don’t have enough creativity. Obviously msg and state could be of whatever type (in real application they end up being tuples more often than not).

Control messages

You can do things to agents. I’m always adding to them but at this stage they are:

type Command =
| Restart
| Stop
| SetManager of AsyncAgent
| SetName of string

Plus some others. I’ll describe most of them later on, right now I want to talk about Restart and Stop. You use the former like this:

counter1 <-- Restart
counter1 <-- 3

Which produces:

From 0 to 3

This should be somehow surprising to you. You would have thought that you could just post integers to a counter. This is not the case. You can post whatever object. This is useful because it allows to have a common model for passing all sort of messages, it allows for the agent not to be parameterized by the type of the message (and of state) so that you can store them in data structures and allows advanced scenarios (i.e. hot swapping of code).

This is a debatable decision. I tried to get the best of strongly typing and dynamic typing, while keeping simplicity of usage. The implementation of this is kind of a mess though. We’ll get there.

BTW: you use Stop just by posting Stop, which stops the agent (forever).

LAgent : an agent framework in F# – Part I – Workers and ParallelWorkers

lucabol — Fri, 29 May 2009 20:26:00 GMT

Download framework here.

All posts are here:

Introduction

I like to try out different programming paradigms. I started out as an object oriented programmer. In university, I used Prolog. I then learned functional programming. I also experimented with various shared memory parallel paradigms (i.e. async, tasks and such). I now want to learn more about message based parallel programming (Erlang style). I’m convinced that doing so makes me a better programmer. Plus, I enjoy it …

My usual learning style is to build a framework that replicates a particular programming model and then write code using it. In essence, I build a very constrained environment. For example, when learning functional programming, I didn’t use any OO construct for a while even if my programming language supports them.

In this case, I built myself a little agent framework based on F# MailboxProcessors. I could have used MailboxProcessors directly, but they are too flexible for my goal. Even to write a simple one of these guys, you need to use async and recursion in a specific pattern, which I always forget. Also, there are multiple ways to to do Post. I wanted things to be as simple as possible. I was willing to sacrifice flexibility for that.

Notice that there are serious efforts in this space (as Axum). This is not one of them. It’s just a simple thing I enjoy working on between one meeting and the next.

Workers and ParallelWorkers

The two major primitives are spawning an agent and posting a message.

let echo = spawnWorker (fun msg -> printfn "%s" msg)
echo <-- "Hello guys!"

There are two kinds of agents in my system. A worker is an agent that doesn’t keep any state between consecutive messages. It is a stateless guy. Notice that the lambda that you pass to create the agent is strongly typed (aka msg is of type string). Also notice that I overloaded the <— operator to mean Post.

Given that a worker is stateless, you can create a whole bunch of them and, when a message is posted, route it to one of them transparently.

let parallelEcho = spawnParallelWorker(fun s -> printfn "%s" s) 10
parallelEcho <-- "Hello guys!”

For example, in the above code, 10 workers are created and, when a message is posted, it gets routed to one of them (using a super duper innovative dispatching algorithm I’ll describe in the implementation part). This parallelWorker guy is not really needed, you could easily built it out of the other primitives, but it is kind of cute.

To show the difference between a worker and a parallelWorker, consider this:

let tprint s = printfn "%s running on thread %i" s Thread.CurrentThread.ManagedThreadId
let echo1 = spawnWorker (fun s -> tprint s)
let parallelEcho1 = spawnParallelWorker(fun s -> tprint s) 10

let messages = ["a";"b";"c";"d";"e";"f";"g";"h";"i";"l";"m";"n";"o";"p";"q";"r";"s";"t"]
messages |> Seq.iter (fun msg -> echo1 <-- msg)
messages |> Seq.iter (fun msg -> parallelEcho1 <-- msg)

The result of the echo1 iteration is:

a running on thread 11
b running on thread 11
c running on thread 11
d running on thread 11
…

While the result of the parallelEcho1 iteration is:

a running on thread 13
c running on thread 14
b running on thread 12
o running on thread 14
m running on thread 13
…

Notice how the latter executes on multiple threads (but not in order). Next time I’ll talk about agents, control messages and error management.

Luke talks about F# on Channel9

lucabol — Wed, 27 May 2009 19:47:00 GMT

Here you go: http://channel9.msdn.com/posts/Charles/Luke-Hoban-Latest-version-of-F-Released-Whats-the-story-Whats-next/

A version of the AsyncCache found its way into the Parallel Programming samples …

lucabol — Thu, 21 May 2009 20:28:13 GMT

Go here to download them. It is in \ParallelExtensionsExtras\CoordinationDataStructures. It has a slightly different design in that it returns Tasks. I’m trying to get Stephen to blog about it so that you can compare them.

I talk about C# and VB Co-Evolution on Channel9 (and some F# …)

lucabol — Thu, 14 May 2009 20:26:18 GMT

The title says it all. If you are interested, go here.

An Async Html cache – part II – Testing the cache

lucabol — Fri, 08 May 2009 18:34:00 GMT

An Async Html cache – Part I - Writing the cache

lucabol — Tue, 28 Apr 2009 02:57:00 GMT

Luca at NDC in Oslo 17 – 19 June 2009

lucabol — Tue, 24 Mar 2009 19:46:27 GMT

I’ll be speaking about the future of C# and F#. Oslo brings back so many memories …

“one of the world’s most important conferences for IT developers and leaders”