package octez-libs
Data queues similar to the Pipe
module in Jane Street's Async
library. They are implemented with Queue
s, limited in size, and use Lwt primitives for concurrent access.
val create : max_size:int -> compute_size:('a -> int) -> unit -> 'a t
create ~max_size ~compute_size ()
is an empty queue that can hold at most max_size
"bytes" of data, using compute_size
to compute the number of "bytes" in a datum.
Note that you can use max_size
/compute_size
to actually limit the size in byte (i.e., the memory footprint of the structure (in this case, consider using push_overhead
to account for the boilerplate memory), but you can also use max_size
/compute_size
to limit the footprint of the structure for some other resource. E.g., you can spin up tasks in separate processes and limit the number of concurrently running processes.
Also note that the size bound is not inclusive. So with max_size
set to 2
and compute_size
to fun _ -> 1
you can add one (1) element and then the pipe is full. (It is full because adding any other element would take the total size to 2
which is not strictly smaller than the max_size
bound.)
Finally, note that when the pipe is empty, inserting an element will always succeed immediately, even if its size exceed the size bound. For the same reason, inserting an element bigger than the size bound will eventually succeeds, but only once the pipe has been emptied. Once such an element has been inserted, no other element can be pushed until the bigger-than-bound element is popped. This behaviour breaks the invariant guaranteed by the module, but is kept for backwards compatibility. It may be changed in the future, but only after a careful review of the potential impact.
push q v
is a promise that is pending until there is enough space in q
to accommodate v
. When this happens v
is added to the end of q
and the promise resolves.
If there is enough space in q
to accommodate v
when the call is made, then the v
is added immediately and an already resolved promise is returned.
Note that if several writes are stuck because the pipe is full. These writes will succeed in an order that might be different from the order the write attempts were made. Specifically, when pushing elements of different computed sizes, smaller pushes may be resolved earlier if enough space is freed.
pop q
is a promise that is pending until there is an element in q
. When this happens an element is removed and the promise is fulfilled with it.
If there is already an element in q
when the call is made, the element is removed immediately and an already resolved promise is returned.
pop_with_timeout t q
is a promise that behaves similarly to pop q
except that it resolves with None
if t
resolves before there is an element in q
to pop.
Note that there can be multiple promises that are awaiting for an element to pop from the queue. As a result, it is possible that pop_with_timeout
is fulfilled with None
even though values have been pushed to the q
.
t
is canceled (i.e., it fails with Canceled
) if an element is returned.
pop_all q
is a promise that is pending until there is at least one element in q
. When this happens, all the elements of q
are removed and the promise is fulfilled with the list of elements (in the order in which they were inserted).
If there is already one or more elements in q
when the call is made, the elements are removed immediately and an already resolved promise is returned.
val pop_all_now : 'a t -> 'a list
pop_all_now q
removes and returns all the elements in q
(in the order in which they were inserted). If q
is empty, []
is returned.
peek q
returns the same value as pop q
but does not remove the returned element.
val peek_all_now : 'a t -> 'a list
peek_all_now q
returns the elements in the q
(oldest first), or []
if empty. It does not remove elements from q
.
val push_now : 'a t -> 'a -> bool
push_now q v
either
- adds
v
at the ends ofq
immediately and returnstrue
, or - if
q
is full, returnsfalse
.
val pop_now : 'a t -> 'a option
pop_now q
may remove and return the first element in q
if q
contains at least one element.
val length : 'a t -> int
length q
is the number of elements in q
.
val is_empty : 'a t -> bool
is_empty q
is true
if q
is empty, false
otherwise.
val close : 'a t -> unit
close q
the write-end of q
:
- Pending and future write attempts will fail with
Closed
. - If there is data left in the pipe, then future read attempts will be resolved until the remaining data is drained, after which further reads will fail with
Closed
. - If there is no data left in the pipe, then pending and future reads will fail with
Closed
.
The close
function is idempotent.
val is_closed : 'a t -> bool
is_closed q
is true
if close q
has been called. It is false
otherwise.