2.1-lock-and-acquire.fsx

#!/usr/bin/env fsharpi

#I "../../../.nuget/packages/hopac/0.3.21/lib/net45/"
#r "Hopac.Platform"
#r "Hopac.Core"
#r "Hopac"

open System.Collections.Generic

open Hopac
open Hopac.Infixes

let ps o = printfn "%A" o

type Sem = MVar<unit>
let P sem = MVar.take sem
let V sem = MVar.fill sem ()

type Buffer<'a> =
    { mutable data: 'a option
      mutable waiting: int
      lockSem: MVar<unit>
      waitSem: MVar<unit> }


[<AutoOpen>]
module Buffer =
    let write (buf:Buffer<'a>) (v:'a) = job { buf.data <- Some v }
    let clear (buf:Buffer<'a>) = job { buf.data <- None }

    let create () =
        { data = Option<'a>.None
          waiting = 0
          lockSem = Sem (())
          waitSem = Sem  ()}

    let wait (buf:Buffer<'a>) = job {
        do buf.waiting <- buf.waiting + 1
        return! V buf.lockSem >>=. P buf.waitSem }

    let rec signal (buf:Buffer<'a>) =
        Job.whenDo (buf.waiting > 0) ((job { buf.waiting <- buf.waiting - 1 })
        >>=. V buf.waitSem
        >>= (fun x -> signal buf))

    let rec insert (buf:Buffer<'a>) (v:'a) =
        P buf.lockSem
        >>=. job {
            do! match buf.data with
                | None ->
                          write buf v
                          >>=. signal buf
                          >>=. V buf.lockSem
                | Some _ ->
                            wait buf
                       >>=. insert buf v }

    let rec remove (buf:Buffer<'a>) =
        P buf.lockSem >>=. job {
            return! match (buf.data) with
                    | Some v -> clear buf >>=. signal buf >>=. V buf.lockSem >>=. Job.result v
                    | None -> wait buf >>=. remove buf }

let b = create ()

start <| Extensions.Seq.iterJobIgnore (fun x -> insert b x)  [0..99]

run   <| Job.forever (remove b >>= (fun x -> job { printfn "%i" x }))
0