src/bounded_queue.zig

Página generada desde zkit@zkit-0.0.0-_fk2iAa4BQCqGFKJeeY9WvMQClX5kiZQ2iOZrVsxVyyN (src/root.zig). No se edita a mano: bun run docs:gen la regenera y bun run docs:check falla si difiere.

BoundedQueue(T) — cola FIFO acotada y thread-safe, con la política de desborde como PARÁMETRO.

Es la versión concurrente de SubscriberQueue (nodo zkit/handle-concurrent, r44 #9 de styx) y la generalización de PendingDeliveryQueue de styx (moqt_pending_delivery.zig), que llevaba mezclados la cola, la política live/VOD, la espera por backpressure (con nanosleep de 0.5 ms en bucle) y los contadores de dominio. Aquí queda la parte genérica:

• Anillo de capacidad fija reservado en init: push NUNCA reserva memoria, así que el hot path no tiene un error.OutOfMemory que convertir en pérdida silenciosa.
• Overflow.reject: lleno ⇒ .full (o pushWait espera con un Condition, sin sondeo, hasta un plazo).
• Overflow.drop_oldest: lleno ⇒ se expulsa el más viejo y se DEVUELVE al llamador (.evicted) para que libere lo que posea — la versión single-thread lo tiraba sin devolverlo, un leak para T con recursos. La siguiente pop devuelve .discontinuity (contrato de SubscriberQueue). El productor nunca espera ni es rechazado.
• Drenado con RESERVA (beginDrain/settleDrain): lo que el consumidor no pudo entregar vuelve al FRENTE en orden. · .reject: lo sacado SIGUE contando contra la capacidad hasta que se liquida. Invariante en todo instante: len + reserved <= capacity, así que settleDrain nunca desborda ni pierde nada. · .drop_oldest: lo sacado NO bloquea al productor (lo fresco manda). Invariante: len <= capacity. Si al liquidar no cabe todo, se descartan los más viejos de lo devuelto (que son más viejos que todo lo encolado) y se entregan al llamador.
• close() despierta a todos los que esperan; wakeAll() también, para que un productor que espera re-mire su flag de cancelación.

Los contadores de dominio (por pista, por sesión) se quedan en el consumidor; aquí sólo hay agregados baratos en stats().

bounded_queue.OverflowCopiar enlace

type · línea 39

pub const Overflow = enum

Sin ///.

bounded_queue.OptionsCopiar enlace

type · línea 47

pub const Options = struct

Sin ///.

bounded_queue.BoundedQueueCopiar enlace

fn · línea 52

pub fn BoundedQueue(comptime T: type) type

Sin ///.

bounded_queue.BoundedQueue.InitErrorCopiar enlace

const · línea 56

pub const InitError = error{ InvalidParameter, OutOfMemory }

Sin ///.

bounded_queue.BoundedQueue.ErrorCopiar enlace

const · línea 57

pub const Error = error{Closed}

Sin ///.

bounded_queue.BoundedQueue.WaitErrorCopiar enlace

const · línea 58

pub const WaitError = error{ Closed, Timeout, Cancelled }

Sin ///.

bounded_queue.BoundedQueue.PushResultCopiar enlace

type · línea 60

pub const PushResult = union

Sin ///.

bounded_queue.BoundedQueue.PopResultCopiar enlace

type · línea 70

pub const PopResult = union

Sin ///.

bounded_queue.BoundedQueue.StatsCopiar enlace

type · línea 77

pub const Stats = struct

Sin ///.

bounded_queue.BoundedQueue.initCopiar enlace

fn · línea 97

pub fn init(allocator: std.mem.Allocator, opts: Options) InitError!Self

Sin ///.

bounded_queue.BoundedQueue.deinitCopiar enlace

fn · línea 108

pub fn deinit(self: *Self) void

Libera el anillo. Los items que queden NO se destruyen: si T posee recursos, vacíala antes con drainInto.

bounded_queue.BoundedQueue.capacityCopiar enlace

fn · línea 117

pub fn capacity(self: *const Self) usize

Sin ///.

bounded_queue.BoundedQueue.lenCopiar enlace

fn · línea 121

pub fn len(self: *Self) usize

Sin ///.

bounded_queue.BoundedQueue.reservedCountCopiar enlace

fn · línea 127

pub fn reservedCount(self: *Self) usize

Sin ///.

bounded_queue.BoundedQueue.statsCopiar enlace

fn · línea 133

pub fn stats(self: *Self) Stats

Sin ///.

bounded_queue.BoundedQueue.closeCopiar enlace

fn · línea 142

pub fn close(self: *Self) void

Cierra la cola: los push posteriores fallan con error.Closed, y los que esperan se despiertan. Lo ya encolado se puede seguir sacando.

bounded_queue.BoundedQueue.wakeAllCopiar enlace

fn · línea 154

pub fn wakeAll(self: *Self) void

Despierta a todos los que esperan para que re-evalúen su plazo o su flag de cancelación.

bounded_queue.BoundedQueue.pushCopiar enlace

fn · línea 165

pub fn push(self: *Self, item: T) Error!PushResult

Encola sin bloquear.

bounded_queue.BoundedQueue.pushWaitCopiar enlace

fn · línea 180

pub fn pushWait( self: *Self, item: T, deadline: time.Deadline, cancel: ?*const std.atomic.Value(bool), ) WaitError!PushResult

Encola; con .reject y la cola llena espera (sin sondeo) a que haya sitio hasta deadline. cancel se re-evalúa en cada despertar: quien lo pone a true debe llamar a wakeAll(). Con .drop_oldest no espera nunca.

bounded_queue.BoundedQueue.popCopiar enlace

fn · línea 242

pub fn pop(self: *Self) PopResult

Saca el más viejo sin bloquear.

bounded_queue.BoundedQueue.popWaitCopiar enlace

fn · línea 254

pub fn popWait(self: *Self, deadline: time.Deadline) error{ Closed, Timeout }!PopResult

Como pop, esperando hasta deadline si está vacía. error.Closed sólo cuando está cerrada Y vacía.

bounded_queue.BoundedQueue.takeDiscontinuityCopiar enlace

fn · línea 283

pub fn takeDiscontinuity(self: *Self) bool

true (una vez) si hubo expulsiones desde la última consulta. Para consumidores por lotes que no usan pop.

bounded_queue.BoundedQueue.drainIntoCopiar enlace

fn · línea 291

pub fn drainInto(self: *Self, out: []T) usize

Mueve hasta out.len items a out (propiedad plena, sin reserva).

bounded_queue.BoundedQueue.beginDrainCopiar enlace

fn · línea 300

pub fn beginDrain(self: *Self, out: []T) usize

Como drainInto, pero lo sacado SIGUE contando contra la capacidad hasta settleDrain. Un único consumidor: begin/settle estrictamente emparejados.

bounded_queue.BoundedQueue.settleDrainCopiar enlace

fn · línea 311

pub fn settleDrain(self: *Self, held: []const T, evicted: []T) usize

Liquida el drenado en curso: held (lo que NO se pudo entregar, en su orden original, prefijo o subconjunto ordenado de lo sacado) vuelve al FRENTE. En .reject siempre cabe (invariante len + reserved <= capacity) y devuelve 0. En .drop_oldest lo que no quepa se descarta empezando por el más viejo de held: esos items se copian a evicted (el llamador es su dueño; evicted.len debe alcanzar held.len) y se devuelve cuántos.

BoundedQueueCopiar enlace

fn · línea 52

pub fn BoundedQueue(comptime T: type) type

Sin ///.

BoundedQueue.InitErrorCopiar enlace

const · línea 56

pub const InitError = error{ InvalidParameter, OutOfMemory }

Sin ///.

BoundedQueue.ErrorCopiar enlace

const · línea 57

pub const Error = error{Closed}

Sin ///.

BoundedQueue.WaitErrorCopiar enlace

const · línea 58

pub const WaitError = error{ Closed, Timeout, Cancelled }

Sin ///.

BoundedQueue.PushResultCopiar enlace

type · línea 60

pub const PushResult = union

Sin ///.

BoundedQueue.PopResultCopiar enlace

type · línea 70

pub const PopResult = union

Sin ///.

BoundedQueue.StatsCopiar enlace

type · línea 77

pub const Stats = struct

Sin ///.

BoundedQueue.initCopiar enlace

fn · línea 97

pub fn init(allocator: std.mem.Allocator, opts: Options) InitError!Self

Sin ///.

BoundedQueue.deinitCopiar enlace

fn · línea 108

pub fn deinit(self: *Self) void

Libera el anillo. Los items que queden NO se destruyen: si T posee recursos, vacíala antes con drainInto.

BoundedQueue.capacityCopiar enlace

fn · línea 117

pub fn capacity(self: *const Self) usize

Sin ///.

BoundedQueue.lenCopiar enlace

fn · línea 121

pub fn len(self: *Self) usize

Sin ///.

BoundedQueue.reservedCountCopiar enlace

fn · línea 127

pub fn reservedCount(self: *Self) usize

Sin ///.

BoundedQueue.statsCopiar enlace

fn · línea 133

pub fn stats(self: *Self) Stats

Sin ///.

BoundedQueue.closeCopiar enlace

fn · línea 142

pub fn close(self: *Self) void

Cierra la cola: los push posteriores fallan con error.Closed, y los que esperan se despiertan. Lo ya encolado se puede seguir sacando.

BoundedQueue.wakeAllCopiar enlace

fn · línea 154

pub fn wakeAll(self: *Self) void

Despierta a todos los que esperan para que re-evalúen su plazo o su flag de cancelación.

BoundedQueue.pushCopiar enlace

fn · línea 165

pub fn push(self: *Self, item: T) Error!PushResult

Encola sin bloquear.

BoundedQueue.pushWaitCopiar enlace

fn · línea 180

pub fn pushWait( self: *Self, item: T, deadline: time.Deadline, cancel: ?*const std.atomic.Value(bool), ) WaitError!PushResult

Encola; con .reject y la cola llena espera (sin sondeo) a que haya sitio hasta deadline. cancel se re-evalúa en cada despertar: quien lo pone a true debe llamar a wakeAll(). Con .drop_oldest no espera nunca.

BoundedQueue.popCopiar enlace

fn · línea 242

pub fn pop(self: *Self) PopResult

Saca el más viejo sin bloquear.

BoundedQueue.popWaitCopiar enlace

fn · línea 254

pub fn popWait(self: *Self, deadline: time.Deadline) error{ Closed, Timeout }!PopResult

Como pop, esperando hasta deadline si está vacía. error.Closed sólo cuando está cerrada Y vacía.

BoundedQueue.takeDiscontinuityCopiar enlace

fn · línea 283

pub fn takeDiscontinuity(self: *Self) bool

true (una vez) si hubo expulsiones desde la última consulta. Para consumidores por lotes que no usan pop.

BoundedQueue.drainIntoCopiar enlace

fn · línea 291

pub fn drainInto(self: *Self, out: []T) usize

Mueve hasta out.len items a out (propiedad plena, sin reserva).

BoundedQueue.beginDrainCopiar enlace

fn · línea 300

pub fn beginDrain(self: *Self, out: []T) usize

Como drainInto, pero lo sacado SIGUE contando contra la capacidad hasta settleDrain. Un único consumidor: begin/settle estrictamente emparejados.

BoundedQueue.settleDrainCopiar enlace

fn · línea 311

pub fn settleDrain(self: *Self, held: []const T, evicted: []T) usize

Liquida el drenado en curso: held (lo que NO se pudo entregar, en su orden original, prefijo o subconjunto ordenado de lo sacado) vuelve al FRENTE. En .reject siempre cabe (invariante len + reserved <= capacity) y devuelve 0. En .drop_oldest lo que no quepa se descarta empezando por el más viejo de held: esos items se copian a evicted (el llamador es su dueño; evicted.len debe alcanzar held.len) y se devuelve cuántos.