Programming model

The MBrace programming model is a language-integrated cloud programming DSL for use from F#. It offers a concise and elegant programming model which extends F# asynchronous workflows to the domain of distributed cloud computation.

The following tutorials demonstrate different aspects of the core programming model. All the tutorials and samples can be found in the MBrace Starter Kit

• Hello World with MBrace
• Introduction to cloud combinators
• Introduction to CPU-intensive cloud parallelism
• Introduction to data parallel cloud flows
• Exceptions and Fault tolerance
• Local Cloud Workflows
• Using C# and native components
• Using cloud value storage
• Using cloud file storage
• Using cloud queue storage
• Using cloud key/value storage
• Provisioning MBrace clusters on Azure

Examples:

• Example: Using Data Parallel Cloud Flows and F# Type Providers for House Prices

• Example: Distributed Image Processing

• Example: KMeans Clustering with Incremental Notifications

• Example: Incremental Stock Trading Analysis

• Example: Extracting Statistics for a Spelling Corrector

• Example: Using the R type provider with MBrace

• Example: Parallel Web Download

• Example: Word Count

• Example: Monte Carlo Pi Approximation

• Example: Extracting Statistics for a Spelling Corrector

• Example: KNN Digit Recognizer

More Techniques:

• Launching Python and Other Executables
• Prototyping a WebServer to Control Your Cluster
• Running MBrace in the Local Machine Thread Pool
• Cloud Gotchas

What follows is a general overview.

Cloud Workflows

In MBrace, the unit of computation is a cloud workflow:

1: 

let myFirstCloudWorkflow = cloud { return 42 }

Cloud workflows generate objects of type Cloud<'T>, which denote a delayed computation that once executed will yield a result of type 'T. Cloud workflows are language-integrated and can freely intertwine with native code.

1: 
2: 
3: 
4: 

let getDataCenterTime() = cloud {
    let now = System.DateTime.Now
    return now
}

Simple cloud computations can be composed into larger ones using the let! keyword:

1: 
2: 
3: 
4: 
5: 
6: 
7: 
8: 

let first = cloud { return 15 }
let second = cloud { return 27 }

cloud {
    let! x = first
    let! y = second
    return x + y
}

This creates bindings to the first and second workflows respectively, to be consumed by the continuations of the main computation. Once executed, this will sequentually perform the computations in first and second, resuming once the latter has completed.

For loops and while loops are possible:

1: 
2: 
3: 
4: 
5: 
6: 
7: 
8: 
9: 

/// Sequentially iterate and loop 
cloud {
    for i in [ 1 .. 100 ] do
        do! Cloud.Logf "Logging entry %d of 100" i

    while true do
        do! Cloud.Sleep 200
        do! Cloud.Log "Logging forever..."
}

MBrace workflows also integrate with exception handling:

1: 
2: 
3: 
4: 
5: 
6: 
7: 

cloud {
    try
        let! x = cloud { return 1 / 0 }
        return true

    with :? System.DivideByZeroException -> return false
}

Asynchronous workflows can be embedded into cloud workflows:

 1: 
 2: 
 3: 
 4: 
 5: 
 6: 
 7: 
 8: 
 9: 
10: 
11: 
12: 

let downloadAsync (url : string) = async {
    use webClient = new System.Net.WebClient()
    let! html = webClient.AsyncDownloadString(System.Uri url) 
    return html.Split('\n')
}


cloud {
    let! t1 = downloadAsync "http://www.nessos.gr/" |> Cloud.OfAsync
    let! t2 = downloadAsync "http://www.mbrace.io/" |> Cloud.OfAsync
    return t1.Length + t2.Length
}

Parallelism Combinators

Cloud workflows as discussed so far enable asynchronous computation but do not suffice in describing parallelism and distribution. To control this, MBrace uses a collection of primitive combinators that act on the distribution/parallelism semantics of execution in cloud workflows.

The previous example could be altered so that downloading happens in parallel:

 1: 
 2: 
 3: 
 4: 
 5: 
 6: 
 7: 
 8: 
 9: 
10: 
11: 
12: 

let downloadCloud url = downloadAsync url |> Cloud.OfAsync

cloud {

    let! results =
        [ "http://www.mbrace.io/"
          "http://www.nessos.gr/" ]
        |> List.map downloadCloud
        |> Cloud.Parallel

    return results |> Array.sumBy(fun r -> r.Length)
}

Here is a second example of Cloud.Parallel:

1: 
2: 
3: 
4: 
5: 
6: 

cloud {

    let n = System.Random().Next(50,100) // number of parallel jobs determined at runtime
    let! results = Cloud.Parallel [ for x in 1..n -> cloud { return x * x } ]
    return Array.sum results
}

Exception handling in workflows

For exception handling, consider the workflow:

 1: 
 2: 
 3: 
 4: 
 5: 
 6: 
 7: 
 8: 
 9: 
10: 
11: 
12: 
13: 
14: 
15: 
16: 
17: 

cloud {

    try
        let! results =
            [ "http://www.mbrace.io/"
              "http://www.nessos.gr/"
              "http://non.existent.domain/" ]
            |> List.map downloadCloud
            |> Cloud.Parallel

        return results |> Array.sumBy(fun r -> r.Length)

    with :? System.Net.WebException as e ->
        // log and reraise
        do! Cloud.Logf "Encountered error %O" e
        return raise e
}

In this case, one of the child computations will fail on account of an invalid url, creating an exception. In general, uncaught exceptions bubble up through Cloud.Parallel triggering cancellation of all outstanding child computations (just like Async.Parallel).

The exception handling clause will almost certainly be executed in a different machine than the one in which it was originally thrown. This is due to the cloud workflow, which allows exceptions, environments, closures to be passed around worker machines in a largely transparent manner.

Non-deterministic parallelism

MBrace provides the Cloud.Choice combinator that utilizes parallelism for non-deterministic algorithms. Cloud workflows of type Cloud<'T option> are said to be non-deterministic in the sense that their return type indicates either success with Some result or a negative answer with None.

Cloud.Choice combines a collection of arbitrary nondeterministic computations into one in which everything executes in parallel: it either returns Some result whenever a child happens to complete with Some result (cancelling all pending jobs) or None when all children have completed with None. It can be thought of as a distributed equivalent to the Seq.tryPick function found in the F# core library.

The following example defines a distributed search function based on Cloud.Choice:

1: 
2: 
3: 
4: 
5: 
6: 
7: 
8: 
9: 

let tryFind (f : 'T -> bool) (ts : 'T list) = cloud {
    let select t = cloud {
        return
            if f t then Some t
            else None
    }

    return! ts |> List.map select |> Cloud.Choice
}

Composing distributed workflows

Recursive and higher-order composition of cloud workflows is possible:

1: 
2: 
3: 
4: 
5: 
6: 
7: 
8: 

/// Sequentially fold along a set of jobs
let rec foldLeftCloud (f : 'State -> 'T -> Cloud<'State>) state ts = cloud {
    match ts with
    | [] -> return state
    | t :: ts' ->
        let! s' = f state t
        return! foldLeftCloud f s' ts'
}

Distributed Data

Cloud workflows offer a programming model for distributed computation. But what happens when it comes to data?

Small-to-medium-scale data can be transported implicitly as part of a cloud computation. This offers a limited (though extremely convenient) form of data distribution. However, it will not scale to all needs, particularly computations involving gigabytes of data.

MBrace offers a range of mechanisms for managing large-scale data in a more global and massive scale. These provide an essential decoupling between distributed computation and distributed data.

See also this summary of the MBrace abstractions for cloud data.

Cloud Values

The mbrace programming model offers access to persistable and cacheable distributed data entities known as cloud values. Cloud values very much resemble immutable data values found in F# but are stored in persisted cloud storage. The following workflow stores the downloaded content of a web page and returns a cloud value to it:

1: 
2: 
3: 
4: 
5: 
6: 
7: 
8: 

let textCell = cloud {
    // download a piece of data
    let! text = downloadCloud "http://www.mbrace.io/"
    // store data to a new CloudValue
    let! cref = CloudValue.New text
    // return the ref
    return cref
}

Dereferencing a cloud value can be done by getting its .Value property:

1: 
2: 
3: 
4: 

let dereference (data : CloudValue<byte []>) = cloud {
    let v = data.Value
    return v.Length
}

It is possible to define explicitly specify the StorageLevel used for the specific CloudValue instance:

1: 
2: 
3: 
4: 
5: 

cloud {
    let! text = downloadCloud "http://www.m-mbrace.net/largeFile.txt"
    let! cref = CloudValue.New(text, storageLevel = StorageLevel.MemoryAndDisk)
    return cref
}

This indicates the cloud value should be persisted using both disk storage and in-memory for dereferencing worker instances.

Example: Defining a MapReduce workflow

Cloud worklows in conjunctions with parallel combinators can be used to articulate MapReduce-like workflows. A simplistic version follows:

 1: 
 2: 
 3: 
 4: 
 5: 
 6: 
 7: 
 8: 
 9: 
10: 
11: 
12: 
13: 
14: 

let mapReduce (map : 'T -> 'R) (reduce : 'R -> 'R -> 'R)
              (identity : 'R) (inputs : 'T list) =

    let rec aux inputs = cloud {
        match inputs with
        | [] -> return identity
        | [t] -> return map t
        | _ ->
            let left,right = List.split inputs
            let! results = Cloud.Parallel [ aux left;  aux right ]
            return reduce results.[0] results.[1]
    }

    aux inputs

The workflow follows a divide-and-conquer approach, recursively partitioning input data until trivial cases are met. Recursive calls are passed through Cloud.Parallel, thus achieving the effect of distributed parallelism.

This is a naive conception of mapReduce, as it does not enable data parallelism nor does it take into account cluster granularity.

Cloud Files

Like cloud values, sequences and vectors, CloudFile is an immutable storage primitive that a references a file saved in the global store. In other words, it is an interface for storing or accessing binary blobs in the runtime.

 1: 
 2: 
 3: 
 4: 
 5: 
 6: 
 7: 
 8: 
 9: 
10: 
11: 
12: 
13: 
14: 
15: 
16: 
17: 
18: 
19: 
20: 
21: 

cloud {
    // enumerate all files from underlying storage container
    let! files = CloudFile.Enumerate "path/to/container"

    // read a cloud file and return its word count
    let wordCount (f : CloudFileInfo) = cloud {
        let! text = CloudFile.ReadAllText f.Path
        let count =
            text.Split(' ')
            |> Seq.groupBy id
            |> Seq.map (fun (token,instances) -> token, Seq.length instances)
            |> Seq.toArray

        return f.Path, count
    }

    // perform computation in parallel
    let! results = files |> Array.map wordCount |> Cloud.Parallel

    return results
}

Mutable cloud values (CloudAtom)

The CloudAtom primitive is, like CloudValue, a reference to data stored in the underlying cloud service. However, CloudAtoms are mutable. The value of a cloud atom can be updated and, as a result, its values are never cached. Mutable cloud values can be updated transactionally using the CloudAtom.Transact methods or forcibly using the CloudAtom.Force method.

The CloudAtom is a powerful primitive that can be used to create runtime-wide synchronization mechanisms like locks, semaphores, etc.

The following demonstrates simple use of the cloud atom:

1: 
2: 
3: 
4: 
5: 
6: 
7: 
8: 
9: 

let race () = cloud {
    let! ca = CloudAtom.New(0)
    let! _ =
        cloud { ca.Force 2 }
            <||>
        cloud { ca.Force 1 }

    return ca.Value
}

The snippet will return a result of either 1 or 2, depending on which update operation was run last.

The following snippet implements an transactionally incrementing function acting on a cloud atom:

1: 
2: 
3: 

let increment (counter : CloudAtom<int>) = cloud {
    do! Cloud.OfAsync <| counter.UpdateAsync(fun c -> c + 1)
}

Other Primitives

For more information and examples on the programming model, please refer to the API Reference.

• MBrace.Core and MBrace.Azure
• Home page
• Get Library via NuGet
• Source Code on GitHub
• License
• Release Notes
• Getting Started
• Local cluster
• Azure cluster
• PLOS'13 Publication
• Documentation
• Programming model
• API Reference (MBrace.Core)
• API Reference (MBrace.Runtime)
• API Reference (MBrace.Flow)
• Reference (Local)
• Reference (Azure)
• Reference (AWS)