val sexp_of_t : ('a -> Ppx_sexp_conv_lib.Sexp.t) -> ('phantom -> Ppx_sexp_conv_lib.Sexp.t) -> ('a, 'phantom) t -> Ppx_sexp_conv_lib.Sexp.t
module Read_write : sig ... end
module Read_only : sig ... end
val create : unit -> 'a Read_write.t
val is_empty : (_, _) t -> bool
val put : ('a, [> Core_kernel.write ]) t -> 'a -> unit Deferred.t
put t a
waits until is_empty t
, and then does set t a
. If there are multiple concurrent put
s, there is no fairness guarantee (i.e., put
s may happen out of order or may be starved).
val set : ('a, [> Core_kernel.write ]) t -> 'a -> unit
set t a
sets the value in t
to a
, even if not (is_empty t)
. This is useful if you want takers to have last-value semantics.
val update : ('a, Core_kernel.read_write) t -> f:('a option -> 'a) -> unit
update t ~f
applies f
to the value in t
and set
s t
to the result. This is useful if you want takers to have accumulated-value semantics.
val update_exn : ('a, Core_kernel.read_write) t -> f:('a -> 'a) -> unit
update_exn
is like update
, except it raises if is_empty t
.
val read_only : ('a, [> Core_kernel.read ]) t -> ('a, Core_kernel.read) t
val write_only : ('a, [> Core_kernel.write ]) t -> ('a, Core_kernel.write) t
val value_available : (_, [> Core_kernel.read ]) t -> unit Deferred.t
value_available t
returns a deferred d
that becomes determined when a value is in t
. d
does not include the value in t
because that value may change after d
becomes determined and before a deferred bind on d
gets to run.
Repeated calls to value_available t
will always return the same deferred until the t
is filled.
val take : ('a, [> Core_kernel.read ]) t -> 'a Deferred.t
take t
returns a deferred that, when t
is filled, becomes determined with the value of t
and and clears t
. If there are multiple concurrent calls to take
then only one of them will be fulfilled and the others will continue waiting on future values. There is no ordering guarantee for which take
call will be filled first.
val take_now : ('a, [> Core_kernel.read ]) t -> 'a option
take_now
is an immediate form of take
.
val take_now_exn : ('a, [> Core_kernel.read ]) t -> 'a
val taken : (_, [> Core_kernel.write ]) t -> unit Deferred.t
taken t
returns a deferred that is filled the next time take
clears t
.
val peek : ('a, [> Core_kernel.read ]) t -> 'a option
peek t
returns the value in t
without clearing t
, or returns None
is is_empty
t
.
val peek_exn : ('a, [> Core_kernel.read ]) t -> 'a
peek_exn t
is like peek
, except it raises if is_empty t
.
val pipe_when_ready : ('a, [> Core_kernel.read ]) t -> 'a Pipe.Reader.t
pipe_when_ready t
returns a pipe, then repeatedly takes a value from t
and writes it to the pipe. After each write, pipe_when_ready
waits for the pipe to be ready to accept another value before taking the next value. Once the pipe is closed, pipe_when_ready
will no longer take values from t
.
Notice that this implementation effectively creates an extra buffer of size 1, so when you read from the pipe you can read a stale value (even though a fresh one should come immediately afterwards), and a value will be taken from the MVar even if it's never read from the pipe.
There is no protection against creating multiple pipes or otherwise multiple things trying to take
concurrently. If that happens, it's not specified which of the pipes will get the value.