A multi-producer, single-consumer bounded queue implemented as a ring buffer.

Imports

ops, sipsic

Types

InvalidCallDefect = object of Defect: Raised by Mupsic.push(), Mupmuc.push(), and Mupmuc.pop() because those should happen via Producer.push() or Consumer.pop(). Source Edit
Mupsic[N; P; T] = object of Sipsic[N, T] prevProducerIdx*: Atomic[int] ## The ID (index) of the most recent producer producerTails*: array[P, Atomic[int]] ## Array of producer tails producerThreadIds*: array[P, Atomic[int]] ## \ ## Array of producer thread IDs by index: A multi-producer, single-consumer bounded queue implemented as a ring buffer. Popping is wait-free.
N is the capacity of the queue.

P is the number of producer threads.

T is the type of data the queue will hold.

Source Edit
NoProducersAvailableError = object of CatchableError: Raised by getProducer() if all producers have been assigned to other threads. Source Edit
Producer[N; P; T] = object idx*: int ## The producer's unique identifier. queue*: ptr Mupsic[N, P, T] ## A reference to the producer's queue.: A per-thread interface for pushing items to a queue. Retrieved via a call to Mupsic.getProducer() Source Edit

Consts

NoProducerIdx = -1: The initial value of Mupsic.prevProducerIdx. Source Edit

Procs

proc capacity[N, P: static int; T](self: var Mupsic[N, P, T]): int {.inline.}: Returns the queue's storage capacity (N). Source Edit
proc getProducer[N, P: static int; T](self: var Mupsic[N, P, T]; idx: int = NoProducerIdx): Producer[N, P, T] {....raises: [NoProducersAvailableError].}: Assigns and returns a Producer instance for the current thread. Source Edit
proc initMupsic[N, P: static int; T](): Mupsic[N, P, T]: Initialize a new Mupsic queue. Source Edit
proc producerCount[N, P: static int; T](self: var Mupsic[N, P, T]): int {.inline.}: Returns the queue's number of producers (P). Source Edit
proc push[N, P: static int; T](self: Producer[N, P, T]; item: T): bool: Append a single item to the queue. If the queue is full, false is returned. If item is appended, true is returned. Source Edit
proc push[N, P: static int; T](self: Producer[N, P, T]; items: openArray[T]): Option[ HSlice[int, int]]: Append multiple items to the queue. If the queue is already full or is filled by this call, some(unpushed) is returned, where unpushed is an HSlice corresponding to the chunk of items which could not be pushed. If all items are appended, none(HSlice[int, int]) is returned. Source Edit
proc push[N, P: static int; T](self: var Mupsic[N, P, T]; item: T): bool: Overload of Sipsic.push() that simply raises InvalidCallDefect. Pushes should happen via Producer.push(). Source Edit
proc push[N, P: static int; T](self: var Mupsic[N, P, T]; items: openArray[T]): Option[ HSlice[int, int]]: Overload of Sipsic.push() that simply raises InvalidCallDefect. Pushes should happen via Producer.push(). Source Edit

src/lockfreequeues/mupsic

Imports

Types

Consts

Procs