sweek's implementation of an mvar in async

mvar.ml

open Core.Std
open Async.Std

module Mvar : sig
  type 'a t with sexp_of
  include Invariant.S1 with type 'a t := 'a t
  val create : unit -> _ t
  val put : 'a t -> 'a -> unit Deferred.t
  val take : 'a t -> 'a Deferred.t
end = struct
  type 'a t =
    { putters : ('a * unit Ivar.t) Queue.t;
      takers : 'a Ivar.t Queue.t;
    }
  with fields, sexp_of

  let invariant invariant_a t : unit =
    try
      let check f = fun field -> f (Field.get field t) in
      Fields.iter
        ~putters:(check (fun putters ->
            if not (Queue.is_empty putters) then assert
              (Queue.is_empty t.takers);
            Queue.iter putters ~f:(fun (a, ivar) ->
                assert (Ivar.is_empty ivar);
                invariant_a a)))
        ~takers:(check (fun takers ->
            if not (Queue.is_empty takers) then assert
              (Queue.is_empty t.putters);
            Queue.iter takers ~f:(fun ivar ->
                assert (Ivar.is_empty ivar))))
    with exn ->
      failwiths "Mvar.invariant failed" (exn, t) <:sexp_of< exn * _ t >>

  let create () =
    { putters = Queue.create ();
      takers  = Queue.create ();
    }

  let take t =
    match Queue.dequeue t.putters with
    | Some (a, ivar) -> Ivar.fill ivar (); return a
    | None -> Deferred.create (fun ivar -> Queue.enqueue t.takers ivar)

  let put t a =
    match Queue.dequeue t.takers with
    | Some ivar -> Ivar.fill ivar a; return ()
    | None -> Deferred.create (fun ivar -> Queue.enqueue t.putters (a, ivar))
end