Data queues similar to the
Pipe module in Jane Street's
Async library. They are implemented with
Queues, 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
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
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
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
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
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
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
val push_now : 'a t -> 'a -> bool
push_now q v either
vat the ends of
qimmediately and returns
qis full, returns
val pop_now : 'a t -> 'a option
pop_now q may remove and return the first element in
q contains at least one element.
val length : 'a t -> int
length q is the number of elements in
val is_empty : 'a t -> bool
is_empty q is
q is empty,
val close : 'a t -> unit
close q the write-end of
- Pending and future write attempts will fail with
- 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
- If there is no data left in the pipe, then pending and future reads will fail with
close function is idempotent.
val is_closed : 'a t -> bool
is_closed q is
close q has been called. It is