Universal Construction

universal_construction.fs

open Marvel
open System.Collections.Concurrent
open System.Collections.Generic

type Pid = int

type Op = {
  op : string
  sn : SN
  }

and SN = int

and Exec = {
  op : string
  pid : Pid
  }

let start
  (pid:Pid) (N:int)
  (s:'s) (d:'s -> string -> 's * 'o)
  (LAST_OP:IDictionary<Pid, Op>) /// a MRSW shared variable containg the last operation invoked by each process
  (CONS:IDictionary<int, _>)     /// a sequence of consensus objects indexed by round
    = async {

  /// A list of all processes
  let procs : Pid list = undef

  /// The last sequence number
  let lastSN : IDictionary<Pid, SN> = undef

  /// A local variable containing the result of the operation
  let res = MVar.create ()

  let helper = async {
    let k = ref 0 // the round number
    let s = ref s // the current object state
    while true do
      // the proposal is the list of operations that
      // invoked by processes who's sequence number is greater
      // than the last sequence number known by the calling process
      let prop =
        procs
        |> List.choose (fun p ->
          if LAST_OP.[p].sn > lastSN.[p] then
            Some LAST_OP.[p].op
          else
            None)
      match prop with
      | [] -> ()
      | prop ->
        incr k
        let execs : Exec list = CONS.[!k] prop
        for exec in execs do
          let (s',r) = d !s exec.op
          s := s'
          lastSN.[exec.pid] <- lastSN.[exec.pid] + 1
          if (exec.pid = pid) then
            do! MVar.put r res |> Async.Ignore
    }

  Async.Start helper

  let perform args = async {
    LAST_OP.[pid] <- { Op.op = "" ; sn = lastSN.[pid] + 1 }
    return! res |> MVar.take }

  return perform

}