The ring buffer — one contiguous view of a stream¶

/**
 * ring_write_policy_demo.c — two ways to write, and what each promises.
 *
 * dp_f32_write() is ALL-OR-NOTHING: a block that does not fit is refused
 * whole. Nothing is copied, the caller still holds every sample, and the
 * refusal is counted in ->dropped -- which is therefore a count of refused
 * samples, not of lost ones. Samples are lost only if the caller then
 * throws them away.
 *
 * dp_f32_write_some() takes what fits and says how much. It never refuses,
 * so it never counts a drop, and it is the only way to feed a block larger
 * than the ring.
 *
 * dp_f32_space() is the room a write is guaranteed to find, so a producer
 * that sizes its block from it is never refused at all. dp_f32_consume()
 * is bounded the same way from the other side: it will not release more
 * than is readable. And
 * dp_f32_wait_status() answers "why can I not have n samples?" without
 * blocking, in one call, for a caller that wants to say so.
 *
 * Build:
 *   make build
 *   ./build/native/examples/ring_write_policy_demo
 */
#include "doppler/buffer/buffer.h"
#include <stdio.h>
#include <stdlib.h>

#define CHECK(cond)                                                           \
  do                                                                          \
    {                                                                         \
      if (!(cond))                                                            \
        {                                                                     \
          fprintf (stderr, "FAIL %s:%d  %s\\n", __FILE__, __LINE__, #cond);   \
          return 1;                                                           \
        }                                                                     \
    }                                                                         \
  while (0)

int
main (void)
{
  dp_f32_t *ring = dp_f32_create (1024);
  CHECK (ring != NULL);
  size_t cap = ring->capacity;
  CHECK (cap == 1024); /* exactly what was asked, on every machine */

  float *block = calloc (2 * (cap + 1), sizeof *block); /* I/Q interleaved */
  CHECK (block != NULL);

  /* An empty ring has room for exactly its capacity. */
  CHECK (dp_f32_space (ring) == cap);
  CHECK (dp_f32_available (ring) == 0);

  /* One sample too many: refused WHOLE, and counted. */
  CHECK (!dp_f32_write (ring, block, cap + 1));
  CHECK (dp_f32_available (ring) == 0); /* nothing was copied       */
  CHECK (ring->dropped == cap + 1);     /* refused, not lost        */

  /* The same block through write_some: the ring takes what fits. */
  CHECK (dp_f32_write_some (ring, block, cap + 1) == cap);
  CHECK (dp_f32_space (ring) == 0);
  CHECK (dp_f32_write_some (ring, block, 8) == 0); /* full: 0, not an error */
  CHECK (ring->dropped == cap + 1);                /* write_some never drops */

  /* Why can I not have n samples? Asked without blocking. */
  CHECK (dp_f32_wait_status (ring, cap) == DP_WAIT_OK);
  CHECK (dp_f32_wait_status (ring, cap + 1) == DP_WAIT_TOO_LARGE);

  /* The release is bounded the same way: more than is there is refused,
     and releases nothing. A caller that gives back only what wait() or
     peek() lent never meets this, and may ignore the return. */
  CHECK (dp_f32_consume (ring, cap + 1) == DP_ERR_INVALID);
  CHECK (dp_f32_available (ring) == cap);
  CHECK (dp_f32_consume (ring, cap) == DP_OK);
  CHECK (dp_f32_wait_status (ring, 1) == DP_WAIT_PENDING); /* just not yet */

  /* A producer that sizes from space() is never refused. */
  size_t before = ring->dropped;
  for (int k = 0; k < 100; k++)
    {
      size_t n = dp_f32_space (ring) < 300 ? dp_f32_space (ring) : 300;
      CHECK (dp_f32_write (ring, block, n));
      if (dp_f32_space (ring) == 0)
        dp_f32_consume (ring, dp_f32_available (ring));
    }
  CHECK (ring->dropped == before);

  /* close() says no more is coming; what is there can still be read. */
  dp_f32_close (ring);
  CHECK (dp_f32_closed (ring));
  size_t left = dp_f32_available (ring);
  CHECK (dp_f32_wait_status (ring, left + 1) == DP_WAIT_CLOSED);
  CHECK (dp_f32_peek (ring, left) != NULL || left == 0);

  /* reset() empties AND reopens, so the mapping carries a second stream.
     ->dropped is a lifetime count and is kept. */
  dp_f32_reset (ring);
  CHECK (!dp_f32_closed (ring));
  CHECK (dp_f32_available (ring) == 0 && dp_f32_space (ring) == cap);
  CHECK (ring->dropped == before);

  printf ("write policy: capacity %zu, %zu samples refused by write(), "
          "0 by write_some(), 0 when sized from space()\n",
          cap, (size_t)ring->dropped);
  free (block);
  dp_f32_destroy (ring);
  return 0;
}

/**
 * ring_drip_feed_demo.c — accumulate small arrivals until a frame is there.
 *
 * Samples arrive in pieces smaller than a frame, on ONE thread. The blocking
 * dp_f32_wait() would deadlock here -- the thread that would produce the
 * samples is the one waiting for them -- so the question "is there a frame
 * yet?" is asked with dp_f32_peek(), which never blocks: it returns the
 * frame, contiguous and zero-copy, or NULL.
 *
 * Self-validating: exits non-zero if a frame is missing, late, or starts at
 * the wrong sample. (Checks are explicit rather than assert(): a Release
 * build defines NDEBUG, and an example that validates nothing is a listing.)
 *
 * Build:
 *   make build
 *   ./build/native/examples/ring_drip_feed_demo
 */
#include "doppler/buffer/buffer.h"
#include <stdio.h>

#define CHECK(cond)                                                           \
  do                                                                          \
    {                                                                         \
      if (!(cond))                                                            \
        {                                                                     \
          fprintf (stderr, "FAIL %s:%d  %s\n", __FILE__, __LINE__, #cond);    \
          return 1;                                                           \
        }                                                                     \
    }                                                                         \
  while (0)

enum
{
  PIECE    = 100,  /* complex samples per arrival          */
  FRAME    = 1024, /* complex samples the consumer wants   */
  ARRIVALS = 25    /* 2500 samples in all: two full frames */
};

int
main (void)
{
  dp_f32_t *ring = dp_f32_create (4096);
  CHECK (ring != NULL);

  /* A ramp, so a frame's first sample says where in the stream it starts. */
  float  piece[2 * PIECE];
  size_t sent = 0, frames = 0;

  for (int arrival = 0; arrival < ARRIVALS; arrival++)
    {
      for (size_t k = 0; k < PIECE; k++)
        {
          piece[2 * k]     = (float)(sent + k); /* I = stream position */
          piece[2 * k + 1] = 0.0f;
        }
      CHECK (dp_f32_write_some (ring, piece, PIECE) == PIECE);
      sent += PIECE;

      /* NULL until FRAME samples have accumulated; then the frame itself. */
      float *frame = dp_f32_peek (ring, FRAME);
      if (frame)
        {
          CHECK (frame[0] == (float)(frames * FRAME)); /* starts on time  */
          CHECK (frame[2 * (FRAME - 1)]
                 == (float)(frames * FRAME + FRAME - 1));
          dp_f32_consume (ring, FRAME);
          frames++;
        }
    }

  CHECK (frames == (size_t)(ARRIVALS * PIECE) / FRAME);
  CHECK (dp_f32_available (ring) == (size_t)(ARRIVALS * PIECE) % FRAME);
  printf ("drip-feed: %d arrivals of %d -> %zu frames of %d, %zu left over\n",
          ARRIVALS, PIECE, frames, FRAME, dp_f32_available (ring));
  dp_f32_destroy (ring);
  return 0;
}

/**
 * ring_chunking_demo.c — any chunk in, fixed frames out.
 *
 * The shape every FFT front end needs: input arrives in large or irregular
 * chunks, the transform wants exactly NFFT samples, possibly overlapped.
 * Feed with dp_f32_write_some(), drain with dp_f32_peek(), and alternate the
 * two -- which is also how a chunk LARGER THAN THE RING goes through it, and
 * why all-or-nothing dp_f32_write() cannot do this job at all.
 *
 *   - NFFT does not divide the ring's capacity, so frames straddle the end
 *     of the ring; the double mapping hands each one back contiguous anyway.
 *   - HOP < NFFT reads overlapped frames: consume() releases HOP, not NFFT.
 *
 * Self-validating: every sample of every frame is checked against the input
 * stream, so a dropped, repeated or torn sample anywhere exits non-zero.
 *
 * Build:
 *   make build
 *   ./build/native/examples/ring_chunking_demo
 */
#include "doppler/buffer/buffer.h"
#include <stdio.h>
#include <stdlib.h>

#define CHECK(cond)                                                           \
  do                                                                          \
    {                                                                         \
      if (!(cond))                                                            \
        {                                                                     \
          fprintf (stderr, "FAIL %s:%d  %s\n", __FILE__, __LINE__, #cond);    \
          return 1;                                                           \
        }                                                                     \
    }                                                                         \
  while (0)

enum
{
  NFFT = 1000, /* frame length; does not divide the capacity */
  HOP  = 250   /* 75% overlap                                 */
};

int
main (void)
{
  dp_f32_t *ring = dp_f32_create (1024);
  CHECK (ring != NULL);

  /* One input chunk, more than three times what the ring can hold. */
  const size_t total = 3 * ring->capacity + 777;
  float       *in    = (float *)malloc (2 * total * sizeof *in);
  CHECK (in != NULL);
  for (size_t i = 0; i < total; i++)
    {
      in[2 * i]     = (float)i; /* I = stream position */
      in[2 * i + 1] = -(float)i;
    }

  size_t off = 0, frames = 0, torn = 0, across_the_end = 0;
  while (off < total)
    {
      /* Takes what fits and says how much; 0 only when the ring is full. */
      off += dp_f32_write_some (ring, in + 2 * off, total - off);

      float *frame;
      while ((frame = dp_f32_peek (ring, NFFT)) != NULL)
        {
          /* NFFT contiguous samples -- even when they wrap the ring. */
          if ((ring->tail & ring->mask) + NFFT > ring->capacity)
            across_the_end++;
          for (size_t k = 0; k < NFFT; k++)
            if (frame[2 * k] != (float)(frames * HOP + k)
                || frame[2 * k + 1] != -(float)(frames * HOP + k))
              torn++;
          dp_f32_consume (ring, HOP); /* overlap: release HOP, keep the rest */
          frames++;
        }
    }

  CHECK (torn == 0);
  CHECK (frames == (total - NFFT) / HOP + 1);
  CHECK (across_the_end > 0); /* the wrap really was exercised */
  printf ("chunking: one %zu-sample chunk through a %zu-sample ring -> %zu "
          "frames of %d (hop %d), %zu read across the end of the ring\n",
          total, ring->capacity, frames, NFFT, HOP, across_the_end);
  free (in);
  dp_f32_destroy (ring);
  return 0;
}

/**
 * ring_threaded_demo.c — one producer thread, one consumer thread, an end.
 *
 * The shape the ring was built for. The producer hands over whatever block
 * sizes it has; the consumer asks for exactly FRAME samples with the
 * blocking dp_f32_wait(), which spins until they are there and returns them
 * contiguous even when they straddle the end of the ring.
 *
 * Two things make it a complete program rather than a loop:
 *
 *   - BACKPRESSURE is the producer's. dp_f32_write() never blocks; it
 *     refuses. So the producer waits for dp_f32_space() before it writes,
 *     and nothing is ever refused.
 *   - THE END is said out loud. An empty ring cannot tell a slow producer
 *     from a finished one, so the producer calls dp_f32_close(). After that
 *     dp_f32_wait() returns NULL instead of spinning forever, and
 *     dp_f32_wait_status() says the NULL means "closed", not "interrupted".
 *     What was written before the close is still delivered first.
 *
 * Self-validating: the stream is a ramp, and every sample of every frame is
 * checked against its position.
 *
 * Build:
 *   make build
 *   ./build/native/examples/ring_threaded_demo
 */
#include "doppler/buffer/buffer.h"
#include <pthread.h>
#include <stdio.h>

#define CHECK(cond)                                                           \
  do                                                                          \
    {                                                                         \
      if (!(cond))                                                            \
        {                                                                     \
          fprintf (stderr, "FAIL %s:%d  %s\\n", __FILE__, __LINE__, #cond);   \
          return 1;                                                           \
        }                                                                     \
    }                                                                         \
  while (0)

enum
{
  FRAME = 1024,
  TOTAL = 50 * FRAME + 300 /* not a whole number of frames, on purpose */
};

static void *
producer (void *arg)
{
  dp_f32_t           *ring    = arg;
  static const size_t sizes[] = { 3000, 700, 4096, 129, 2048 };
  float               block[2 * 4096];
  size_t              sent = 0, turn = 0;

  while (sent < TOTAL)
    {
      size_t n = sizes[turn++ % 5];
      if (n > TOTAL - sent)
        n = TOTAL - sent;
      for (size_t k = 0; k < n; k++)
        {
          block[2 * k]     = (float)(sent + k); /* I = stream position */
          block[2 * k + 1] = 0.0f;
        }
      /* Wait for ROOM rather than retrying a refused write: the consumer
         can only free space, so once it is there it stays there. */
      while (dp_f32_space (ring) < n)
        ;
      dp_f32_write (ring, block, n);
      sent += n;
    }
  dp_f32_close (ring); /* the consumer's only way to tell slow from done */
  return NULL;
}

int
main (void)
{
  dp_f32_t *ring = dp_f32_create (8192);
  CHECK (ring != NULL);

  pthread_t tid;
  CHECK (pthread_create (&tid, NULL, producer, ring) == 0);

  size_t got = 0;
  float *frame;
  while ((frame = dp_f32_wait (ring, FRAME)) != NULL)
    {
      for (size_t k = 0; k < FRAME; k++)
        CHECK (frame[2 * k] == (float)(got + k));
      dp_f32_consume (ring, FRAME);
      got += FRAME;
    }

  /* NULL has more than one meaning; ask which. */
  CHECK (dp_f32_wait_status (ring, FRAME) == DP_WAIT_CLOSED);

  /* The tail is under a frame but it is still there, and still in order. */
  size_t tail = dp_f32_available (ring);
  CHECK (tail == TOTAL % FRAME);
  float *rest = dp_f32_peek (ring, tail);
  CHECK (rest != NULL);
  for (size_t k = 0; k < tail; k++)
    CHECK (rest[2 * k] == (float)(got + k));
  dp_f32_consume (ring, tail);

  pthread_join (tid, NULL);
  CHECK (ring->dropped == 0);
  printf ("threaded: %d samples -> %zu frames of %d + a tail of %zu, "
          "0 refused, ended by close()\n",
          TOTAL, got / FRAME, FRAME, tail);
  dp_f32_destroy (ring);
  return 0;
}

/**
 * ring_element_view_demo.c — one element per sample, on every width.
 *
 * The ring stores SCALARS, two per complex sample, and dp_f32_wait()
 * returns a `float *`. A caller that thinks in samples would rather have
 * one ELEMENT per sample, and the `_view` functions are that: the same
 * four calls, the same memory, the same counts -- typed as the element.
 *
 *     f32   float _Complex
 *     f64   double _Complex
 *     i16   dp_iq16_t  { int16_t i, q; }
 *
 * C has no complex integer, so the 16-bit element is a two-field record.
 * It is the same 4 bytes an ADC delivers (I, Q, I, Q, ...), so a capture
 * buffer can be handed over with a cast and no copy.
 *
 * Each is a cast and nothing more: every count in buffer.h is already in
 * samples, so the two faces cannot disagree about a length.
 *
 * Build:
 *   make build
 *   ./build/native/examples/ring_element_view_demo
 */
#include "doppler/f32_buffer/f32_buffer_core.h"
#include "doppler/i16_buffer/i16_buffer_core.h"
#include <stdio.h>

#define CHECK(cond)                                                           \
  do                                                                          \
    {                                                                         \
      if (!(cond))                                                            \
        {                                                                     \
          fprintf (stderr, "FAIL %s:%d  %s\\n", __FILE__, __LINE__, #cond);   \
          return 1;                                                           \
        }                                                                     \
    }                                                                         \
  while (0)

int
main (void)
{
  /* ── complex float: write elements, read elements ─────────────────── */
  dp_f32_t *f = dp_f32_create (1024);
  CHECK (f != NULL);

  float _Complex tone[4] = { 1.0f, 2.0f, 3.0f, 4.0f };
  CHECK (dp_f32_write_view (f, tone, 4)); /* 4 SAMPLES, not 8 floats */
  CHECK (dp_f32_available (f) == 4);

  float _Complex *v   = dp_f32_peek_view (f, 4);
  float          *raw = dp_f32_peek (f, 4);
  CHECK (v != NULL && (void *)v == (void *)raw); /* one memory, two types */
  CHECK (v[2] == 3.0f && raw[2 * 2] == 3.0f && raw[2 * 2 + 1] == 0.0f);
  dp_f32_consume (f, 4);
  dp_f32_destroy (f);

  /* ── 16-bit I/Q: an ADC buffer goes in with a cast ────────────────── */
  dp_i16_t *q = dp_i16_create (1024);
  CHECK (q != NULL);

  int16_t adc[6] = { 10, 11, 20, 21, 30, 31 }; /* I, Q, I, Q, I, Q */
  CHECK (dp_i16_write_some_view (q, (const dp_iq16_t *)adc, 3) == 3);

  dp_iq16_t *s = dp_i16_wait_view (q, 3); /* already there: returns at once */
  CHECK (s != NULL);
  CHECK (s[0].i == 10 && s[0].q == 11);
  CHECK (s[2].i == 30 && s[2].q == 31);
  dp_i16_consume (q, 3);
  dp_i16_destroy (q);

  printf ("element view: f32 as float _Complex, i16 as {i, q} -- "
          "same memory as the scalar face, counts in samples\n");
  return 0;
}

/**
 * ring_backed_demo.c — a ring whose samples are a file.
 *
 * dp_f32_create_backed() is dp_f32_create() over a path instead of
 * anonymous memory. The mapping is shared, so the ring's samples ARE the
 * file's contents: there is no separate write-to-disk step and no copy,
 * and the two cannot disagree. That is what lets a history outlive the
 * process that recorded it.
 *
 * Two calls matter beyond the ordinary ones:
 *
 *   - `existed` says whether the file already held a ring of this size.
 *     1 means its samples are now this ring's; 0 means it was created (or
 *     resized) and is zeroed.
 *   - dp_f32_sync() flushes to disk. Until then the samples are in the
 *     page cache, where a crash can still lose them -- so call it where a
 *     checkpoint is taken, not after every write.
 *
 * What is persisted is the SAMPLES. The read and write positions live in
 * the ring's struct, not in the file, so a re-attached ring starts empty
 * over its old contents: a caller that wants to resume records where it
 * was, beside the file, and that bookkeeping is its own.
 *
 * Build:
 *   make build
 *   ./build/native/examples/ring_backed_demo
 */
#include "doppler/buffer/buffer.h"
#include <stdio.h>
#include <stdlib.h>

#define CHECK(cond)                                                           \
  do                                                                          \
    {                                                                         \
      if (!(cond))                                                            \
        {                                                                     \
          fprintf (stderr, "FAIL %s:%d  %s\\n", __FILE__, __LINE__, #cond);   \
          return 1;                                                           \
        }                                                                     \
    }                                                                         \
  while (0)

enum
{
  N = 4096
};

int
main (void)
{
  const char *dir = getenv ("TMPDIR");
  char        path[512];
  snprintf (path, sizeof path, "%s/dp_ring_backed_demo.bin",
            dir && *dir ? dir : "/tmp");
  remove (path); /* a clean start, so `existed` below means something */

  /* ── first process: record, checkpoint, go away ───────────────────── */
  int       existed = -1;
  dp_f32_t *ring    = dp_f32_create_backed (N, path, &existed);
  CHECK (ring != NULL);
  CHECK (existed == 0); /* created, and zeroed */
  size_t cap = ring->capacity;

  static float block[2 * N];
  for (size_t k = 0; k < N; k++)
    {
      block[2 * k]     = (float)k;
      block[2 * k + 1] = -(float)k;
    }
  CHECK (dp_f32_write (ring, block, N));
  dp_f32_sync (ring);    /* the checkpoint: now it is on disk */
  dp_f32_destroy (ring); /* unmaps; the FILE stays             */

  /* ── second process: the history is simply there ──────────────────── */
  ring = dp_f32_create_backed (N, path, &existed);
  CHECK (ring != NULL);
  CHECK (existed == 1); /* same size: mapped as it stands */
  CHECK (ring->capacity == cap);
  CHECK (dp_f32_available (ring) == 0); /* positions are NOT in the file */

  for (size_t k = 0; k < N; k++)
    {
      CHECK (ring->data[2 * k] == (float)k);
      CHECK (ring->data[2 * k + 1] == -(float)k);
    }

  printf ("backed: %d samples written, synced, unmapped -- and read back "
          "from %s by a fresh mapping\n",
          N, path);
  dp_f32_destroy (ring);
  remove (path);
  return 0;
}