spsc_ring.c File Reference

#include "serene/rt/impl/spsc_ring.h"
#include "serene/rt/context.h"
#include "serene/utils.h"

Include dependency graph for spsc_ring.c:

Go to the source code of this file.

Macros
#define	MIN_MAG   2
#define	MAX_MAG   MAX_POW2_EXP_FOR_UNSIGNED(SRN_SPSC_RING_CAP_TYPE)

Functions
srn_spsc_ring_t *	srn_spsc_ring_make (srn_context_t *ctx, size_t magnitude, size_t elem_size)
	Set up a ring by allocating the buf to be the size of cap * elem_size.
bool	srn_spsc_ring_push (srn_spsc_ring_t *r, const void *elem)
	Producer side.
bool	srn_spsc_ring_pop (srn_spsc_ring_t *r, void *out)
	Consumer side.
bool	srn_spsc_ring_empty (const srn_spsc_ring_t *r)
	Whether the ring currently holds no elements.

Macro Definition Documentation

MAX_MAG

#define MAX_MAG   MAX_POW2_EXP_FOR_UNSIGNED(SRN_SPSC_RING_CAP_TYPE)

Definition at line 25 of file spsc_ring.c.

MIN_MAG

#define MIN_MAG   2

Definition at line 24 of file spsc_ring.c.

Function Documentation

srn_spsc_ring_empty()

bool srn_spsc_ring_empty

(

const srn_spsc_ring_t *

r

)

Whether the ring currently holds no elements.

For the consumer this is exact; for any other observer it is a hint.

Definition at line 103 of file spsc_ring.c.

                                                   {
  // Exact for the consumer, the only party that moves `head`. For any other
  // observer it is a hint, since the producer may publish right after the load.
  return atomic_load_explicit(&r->head, memory_order_acquire) ==
         atomic_load_explicit(&r->tail, memory_order_acquire);
}

Here is the caller graph for this function:

srn_spsc_ring_make()

srn_spsc_ring_t * srn_spsc_ring_make	(	srn_context_t *	ctx,
		size_t	magnitude,
		size_t	elem_size )

Set up a ring by allocating the buf to be the size of cap * elem_size.

And it will choose the cap to be 2 ^ magnitude.

Definition at line 27 of file spsc_ring.c.

                                                                                            {
  PANIC_IF_NULL(ctx);
 
  // Capacity is `2 ^ magnitude`, so it is always a power of two and a slot
  // index can be masked rather than divided. The magnitude must stay in range:
  // below MIN_MAG the ring is too small to be useful, and above MAX_MAG the
  // shift no longer fits `SRN_SPSC_RING_CAP_TYPE`. An out-of-range request
  // fails as a value carrying an error rather than aborting.
  if ((magnitude < MIN_MAG) || (magnitude > MAX_MAG)) {
    return FAIL(ctx, srn_spsc_ring_t, WRONG_RANGE, "Magnitude is not in the supported range.");
  }
 
  srn_spsc_ring_t *r = ALLOC(ctx, srn_spsc_ring_t);
 
  // A fresh ring is empty -- head and tail both start at zero -- and carries no
  // error. `mask` is `cap - 1`, the low bits that select a slot, and `buf`
  // holds `cap` slots of `elem_size` bytes each.
  atomic_init(&r->head, 0);
  atomic_init(&r->tail, 0);
 
  r->maybe_error = nullptr;
  r->cap         = ((SRN_SPSC_RING_CAP_TYPE)1U << magnitude);
  r->mask        = r->cap - 1;
  r->elem_size   = elem_size;
  // The buffer multiply must not wrap, a wrapped size would allocate a tiny
  // buffer while push/pop still index up to `cap` slots.
  PANIC_IF(elem_size == 0, "'elem_size' must be nonzero");
  PANIC_IF(r->cap > SIZE_MAX / elem_size, "ring buffer size overflows");
  r->buf = ALLOCN(ctx, unsigned char, (r->cap * elem_size));
  PANIC_IF_NULL(r->buf);
 
  return r;
}

Here is the caller graph for this function:

srn_spsc_ring_pop()

bool srn_spsc_ring_pop	(	srn_spsc_ring_t *	r,
		void *	out )

Consumer side.

Copy one element into *out. Returns false when the ring is empty, leaving *out untouched.

Definition at line 83 of file spsc_ring.c.

                                                      {
  // Mirror of push. The consumer alone advances `head` (relaxed), and reads the
  // producer's `tail` with acquire so it sees every element published so far.
  const size_t h = atomic_load_explicit(&r->head, memory_order_relaxed);
  const size_t t = atomic_load_explicit(&r->tail, memory_order_acquire);
 
  // Equal indices mean nothing new has been published since the last pop:
  // empty.
  if (h == t) {
    return false;
  }
 
  // Copy the element out, then free the slot by advancing `head` with release.
  // The release orders this read before a producer's acquire of `head`, so the
  // producer never reuses the slot while it is still being read here.
  memcpy(out, r->buf + ((h & r->mask) * r->elem_size), r->elem_size);
  atomic_store_explicit(&r->head, h + 1, memory_order_release);
  return true;
}

Here is the caller graph for this function:

srn_spsc_ring_push()

bool srn_spsc_ring_push	(	srn_spsc_ring_t *	r,
		const void *	elem )

Producer side.

Copy *elem into the ring. Returns false when the ring is full, leaving it unchanged.

Definition at line 61 of file spsc_ring.c.

                                                              {
  // The producer alone advances `tail`, so reading it needs no synchronisation.
  // `head` belongs to the consumer; the acquire load makes the slots it has
  // freed visible here.
  const size_t t = atomic_load_explicit(&r->tail, memory_order_relaxed);
  const size_t h = atomic_load_explicit(&r->head, memory_order_acquire);
 
  // `t - h` is the live count, correct across the unsigned wrap of the
  // counters. At `cap` the ring is full; refuse rather than overwrite a slot
  // the consumer has not read yet.
  if (t - h >= r->cap) {
    return false;
  }
 
  // Write the element, then publish it by advancing `tail` with release. A
  // consumer that later acquires this `tail` value is guaranteed to see the
  // bytes written just above, never a half-filled slot.
  memcpy(r->buf + ((t & r->mask) * r->elem_size), elem, r->elem_size);
  atomic_store_explicit(&r->tail, t + 1, memory_order_release);
  return true;
}

Here is the caller graph for this function: