spsc_ring.c

Go to the documentation of this file.

/* -*- C -*-
 * Serene programming language
 * Copyright (C) 2019-2026 Sameer Rahmani <[email protected]>
 *
 * This library is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this library.  If not, see <https://www.gnu.org/licenses/>.
 */
 
#include "serene/rt/impl/spsc_ring.h"
 
#include "serene/rt/context.h"
#include "serene/utils.h"
 
#define MIN_MAG 2
#define MAX_MAG MAX_POW2_EXP_FOR_UNSIGNED(SRN_SPSC_RING_CAP_TYPE)
 
srn_spsc_ring_t *srn_spsc_ring_make(srn_context_t *ctx, size_t magnitude, size_t elem_size) {
  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;
}
 
bool srn_spsc_ring_push(srn_spsc_ring_t *r, const void *elem) {
  // 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;
}
 
bool srn_spsc_ring_pop(srn_spsc_ring_t *r, void *out) {
  // 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;
}
 
bool srn_spsc_ring_empty(const srn_spsc_ring_t *r) {
  // 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);
}