Many producer, many consumer, thread-safe, runtime configurable buffer size. When buffer is empty, consumers suspend and are resumed by producers. When buffer is full, producers suspend and are resumed by consumers.
pub fn Queue(Elem: type) typepub fn Queue(Elem: type) type {
return struct {
type_erased: TypeErasedQueue,
pub fn init(buffer: []Elem) @This() {
return .{ .type_erased = .init(@ptrCast(buffer)) };
}
/// After this is called, the queue enters a "closed" state. A closed
/// queue always returns `error.Closed` for put attempts even when
/// there is space in the buffer. However, existing elements of the
/// queue are retrieved before `error.Closed` is returned.
///
/// Threadsafe.
pub fn close(q: *@This(), io: Io) void {
q.type_erased.close(io);
}
/// Appends elements to the end of the queue, potentially blocking if
/// there is insufficient capacity. Returns when any one of the
/// following conditions is satisfied:
///
/// * At least `min` elements have been added to the queue
/// * The queue is closed
/// * The current task is canceled
///
/// Returns how many of `elements` have been added to the queue, if any.
/// If an error is returned, no elements have been added.
///
/// If the queue is closed or the task is canceled, but some items were
/// already added before the closure or cancelation, then `put` may
/// return a number lower than `min`, in which case future calls are
/// guaranteed to return `error.Canceled` or `error.Closed`.
///
/// A return value of 0 is only possible if `min` is 0, in which case
/// the call is guaranteed to queue as many of `elements` as is possible
/// *without* blocking.
///
/// Asserts that `elements.len >= min`.
pub fn put(q: *@This(), io: Io, elements: []const Elem, min: usize) (QueueClosedError || Cancelable)!usize {
return @divExact(try q.type_erased.put(io, @ptrCast(elements), min * @sizeOf(Elem)), @sizeOf(Elem));
}
/// Same as `put` but blocks until all elements have been added to the queue.
///
/// If the queue is closed or canceled, `error.Closed` or `error.Canceled`
/// is returned, and it is unspecified how many, if any, of `elements` were
/// added to the queue prior to cancelation or closure.
pub fn putAll(q: *@This(), io: Io, elements: []const Elem) (QueueClosedError || Cancelable)!void {
const n = try q.put(io, elements, elements.len);
if (n != elements.len) {
_ = try q.put(io, elements[n..], elements.len - n);
unreachable; // partial `put` implies queue was closed or we were canceled
}
}
/// Same as `put`, except does not introduce a cancelation point.
///
/// For a description of cancelation and cancelation points, see `Future.cancel`.
pub fn putUncancelable(q: *@This(), io: Io, elements: []const Elem, min: usize) QueueClosedError!usize {
return @divExact(try q.type_erased.putUncancelable(io, @ptrCast(elements), min * @sizeOf(Elem)), @sizeOf(Elem));
}
/// Appends `item` to the end of the queue, blocking if the queue is full.
pub fn putOne(q: *@This(), io: Io, item: Elem) (QueueClosedError || Cancelable)!void {
assert(try q.put(io, &.{item}, 1) == 1);
}
/// Same as `putOne`, except does not introduce a cancelation point.
///
/// For a description of cancelation and cancelation points, see `Future.cancel`.
pub fn putOneUncancelable(q: *@This(), io: Io, item: Elem) QueueClosedError!void {
assert(try q.putUncancelable(io, &.{item}, 1) == 1);
}
/// Receives elements from the beginning of the queue, potentially blocking
/// if there are insufficient elements currently in the queue. Returns when
/// any one of the following conditions is satisfied:
///
/// * At least `min` elements have been received from the queue
/// * The queue is closed and contains no buffered elements
/// * The current task is canceled
///
/// Returns how many elements of `buffer` have been populated, if any.
/// If an error is returned, no elements have been populated.
///
/// If the queue is closed or the task is canceled, but some items were
/// already received before the closure or cancelation, then `get` may
/// return a number lower than `min`, in which case future calls are
/// guaranteed to return `error.Canceled` or `error.Closed`.
///
/// A return value of 0 is only possible if `min` is 0, in which case
/// the call is guaranteed to fill as much of `buffer` as is possible
/// *without* blocking.
///
/// Asserts that `buffer.len >= min`.
pub fn get(q: *@This(), io: Io, buffer: []Elem, min: usize) (QueueClosedError || Cancelable)!usize {
return @divExact(try q.type_erased.get(io, @ptrCast(buffer), min * @sizeOf(Elem)), @sizeOf(Elem));
}
/// Same as `get`, except does not introduce a cancelation point.
///
/// For a description of cancelation and cancelation points, see `Future.cancel`.
pub fn getUncancelable(q: *@This(), io: Io, buffer: []Elem, min: usize) QueueClosedError!usize {
return @divExact(try q.type_erased.getUncancelable(io, @ptrCast(buffer), min * @sizeOf(Elem)), @sizeOf(Elem));
}
/// Receives one element from the beginning of the queue, blocking if the queue is empty.
pub fn getOne(q: *@This(), io: Io) (QueueClosedError || Cancelable)!Elem {
var buf: [1]Elem = undefined;
assert(try q.get(io, &buf, 1) == 1);
return buf[0];
}
/// Same as `getOne`, except does not introduce a cancelation point.
///
/// For a description of cancelation and cancelation points, see `Future.cancel`.
pub fn getOneUncancelable(q: *@This(), io: Io) QueueClosedError!Elem {
var buf: [1]Elem = undefined;
assert(try q.getUncancelable(io, &buf, 1) == 1);
return buf[0];
}
/// Returns buffer length in `Elem` units.
pub fn capacity(q: *const @This()) usize {
return @divExact(q.type_erased.buffer.len, @sizeOf(Elem));
}
};
}