Module Lwt_pipe.Bounded

Type of queues holding values of type 'a.

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.

• raises {!Closed}

  if q is closed. More specifically, the promise is rejected with Closed if q is closed.

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.

• raises {!Closed}

  if q is empty and closed. More specifically, the promise is rejected with Closed if q is empty and closed.

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.

• raises {!Closed}

  if q is empty and closed. More specifically, the promise is rejected with Closed if q is empty and closed.

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.

• raises {!Closed}

  if q is empty and closed. More specifically, the promise is rejected with Closed if q is empty and closed.

  In practice, this function returns a promise that either:

  • is pending and will resolve with a single-element list,
  • is already resolved with a list of at least one element,
  • or will be rejected with Closed.

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.

• raises {!Closed}

  if q is empty and closed.

peek q returns the same value as pop q but does not remove the returned element.

• raises {!Closed}

  if q is empty and closed. More specifically, the promise is rejected with Closed if q is empty and closed.

peek_all_now q returns the elements in the q (oldest first), or [] if empty. It does not remove elements from q.

• raises {!Closed}

  if q is empty and closed.

push_now q v either

• adds v at the ends of q immediately and returns true, or
• if q is full, returns false.

• raises Closed

  if q is closed.

pop_now q may remove and return the first element in q if q contains at least one element.

• raises Closed

  if q is closed.

length q is the number of elements in q.

is_empty q is true if q is empty, false otherwise.

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.

is_closed q is true if close q has been called. It is false otherwise.

The number of bytes used in the internal representation to hold an element in the queue.