Module Async_kernel.Mvar
An Mvar
is a mutable location that is either empty or contains a value. One can put
or set
the value, and wait on value_available
for the location to be filled in either way.
Having an Mvar.Read_write.t
gives the capability to mutate the mvar.
The key difference between an Mvar
and an Ivar
is that an Mvar
may be filled multiple times.
This implementation of Mvar
also allows one to replace the value without any guarantee that the reading side has seen it. This is useful in situations where last-value semantics are desired (e.g., you want to signal whenever a config file is updated, but only care about the most recent contents).
An Mvar
can also be used as a baton-passing mechanism between a producer and consumer. For instance, a producer reading from a socket and producing a set of deserialized messages can put
the batch from a single read into an Mvar
and can wait for taken
to return as a pushback mechanism. The consumer meanwhile waits on value_available
. This way the natural batch size is passed between the two sub-systems with minimal overhead.
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 untilis_empty t
, and then doesset t a
. If there are multiple concurrentput
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 int
toa
, even ifnot (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
appliesf
to the value int
andset
st
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 likeupdate
, except it raises ifis_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 deferredd
that becomes determined when a value is int
.d
does not include the value int
because that value may change afterd
becomes determined and before a deferred bind ond
gets to run.Repeated calls to
value_available t
will always return the same deferred until thet
is filled.
val take : ('a, [> Core_kernel.read ]) t -> 'a Deferred.t
take t
returns a deferred that, whent
is filled, becomes determined with the value oft
and and clearst
. If there are multiple concurrent calls totake
then only one of them will be fulfilled and the others will continue waiting on future values. There is no ordering guarantee for whichtake
call will be filled first.
val take_now : ('a, [> Core_kernel.read ]) t -> 'a option
take_now
is an immediate form oftake
.
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 timetake
clearst
.
val peek : ('a, [> Core_kernel.read ]) t -> 'a option
peek t
returns the value int
without clearingt
, or returnsNone
isis_empty t
.
val peek_exn : ('a, [> Core_kernel.read ]) t -> 'a
peek_exn t
is likepeek
, except it raises ifis_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 fromt
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 fromt
.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.