File buffer.h¶
Go to the documentation of this file
#ifndef DP_BUFFER_H
#define DP_BUFFER_H
/* -------------------------------------------------------------------------
* Platform includes
* ---------------------------------------------------------------------- */
#ifdef _WIN32
#ifndef _WIN32_WINNT
#define _WIN32_WINNT 0x0A00 /* Windows 10 – required for VirtualAlloc2 */
#elif _WIN32_WINNT < 0x0A00
#undef _WIN32_WINNT
#define _WIN32_WINNT 0x0A00
#endif
#ifndef WIN32_LEAN_AND_MEAN
#define WIN32_LEAN_AND_MEAN
#endif
#include <windows.h>
#else /* POSIX ------------------------------------------------------------ \
*/
/* NO feature-test macro here. _GNU_SOURCE / _DARWIN_C_SOURCE are defined on
the compile line (see the top-level CMakeLists.txt), because glibc's
features.h latches on its first inclusion: a #define in this header is a
no-op for every translation unit that reached libc before including it,
which is most of them. This file carried one and it was inert exactly that
way -- doppler#986. A downstream compiling the installed headers needs the
same definition on ITS compile line. */
#include <fcntl.h> /* shm_open on macOS; open() for a file-backed ring */
#include <stdio.h> /* snprintf */
#include <sys/mman.h> /* mmap, msync */
#include <sys/stat.h> /* fstat -- is this file already a ring? */
#include <unistd.h>
#endif /* _WIN32 */
/* The ring's wait consults the same interrupt flag as every other
blocking wait in doppler; see docs/design/io-termination.md. */
#include "doppler/dp_interrupt.h"
#include <stdbool.h>
#include <stddef.h>
#include <stdint.h>
#include "doppler/clib_common.h" /* DP_NODISCARD */
#include "doppler/dp_state.h" /* the framer's snapshot envelope */
#include "doppler/jm_perf.h" /* JM_FORCEINLINE */
#include "doppler/util/util_core.h" /* next_pow_two */
/* macOS uses MAP_ANON, Linux uses MAP_ANONYMOUS. Normalize to MAP_ANONYMOUS.
*/
#if !defined(_WIN32) && !defined(MAP_ANONYMOUS) && defined(MAP_ANON)
#define MAP_ANONYMOUS MAP_ANON
#endif
#include <stdlib.h>
#include <string.h>
/* -------------------------------------------------------------------------
* Architecture: spin-hint
* ---------------------------------------------------------------------- */
#if defined(__x86_64__) || defined(_M_X64) || defined(__i386__)
#include <immintrin.h>
#define DP_SPIN_HINT() _mm_pause ()
#else
#define DP_SPIN_HINT() ((void)0)
#endif
/* -------------------------------------------------------------------------
* Compat: alignment
* ---------------------------------------------------------------------- */
#if defined(__STDC_VERSION__) && __STDC_VERSION__ >= 201112L
#include <stdalign.h>
#define DP_ALIGN(n) alignas (n)
#else
#define DP_ALIGN(n) __attribute__ ((aligned (n)))
#endif
/* -------------------------------------------------------------------------
* Compat: compile-time power-of-two assert
* ---------------------------------------------------------------------- */
#if defined(__STDC_VERSION__) && __STDC_VERSION__ >= 201112L
#define DP_ASSERT_PWR2(n) \
_Static_assert (((n) & ((n) - 1)) == 0, "Size must be power of 2")
#else
#define DP_ASSERT_PWR2(n) \
typedef char dp_assert_pwr2_##n[((n) & ((n) - 1)) == 0 ? 1 : -1]
#endif
/* -------------------------------------------------------------------------
* Atomics (GCC / Clang __atomic built-ins)
* ---------------------------------------------------------------------- */
#if defined(__GNUC__) || defined(__clang__)
#define DP_LOAD_ACQ(ptr) \
__extension__ ({ \
size_t _v; \
__atomic_load ((const size_t *)(ptr), &_v, __ATOMIC_ACQUIRE); \
_v; \
})
#define DP_LOAD_RLX(ptr) \
__extension__ ({ \
size_t _v; \
__atomic_load ((const size_t *)(ptr), &_v, __ATOMIC_RELAXED); \
_v; \
})
#define DP_STORE_REL(ptr, val) \
__extension__ ({ \
size_t _v = (val); \
__atomic_store ((size_t *)(ptr), &_v, __ATOMIC_RELEASE); \
})
#else
#error "doppler requires GCC or Clang for atomic built-ins."
#endif
#define DP_CACHELINE 64
/* =========================================================================
* Platform-agnostic memory helpers
* All platform-specific logic lives HERE, outside any macro, so that
* #ifdef / #else can be used freely.
* ====================================================================== */
static inline size_t
dp__page_size (void)
{
#ifdef _WIN32
SYSTEM_INFO si;
GetSystemInfo (&si);
return (size_t)si.dwAllocationGranularity;
#else
return (size_t)sysconf (_SC_PAGESIZE);
#endif
}
#ifdef _WIN32
static inline void *
dp__win_map_twice (HANDLE mapping, size_t bytes)
{
/*
* Windows ring-buffer using only kernel32.dll functions (Win7+).
*
* Strategy: CreateFileMapping + MapViewOfFileEx at adjacent addresses.
* Reserve 2*bytes of address space to find a hole BOTH views fit in,
* release it, then map the two views into it. A retry loop handles the
* one way that can still fail: another thread taking the address between
* the release and the maps.
*
* The hole has to be probed at 2*bytes. This used to probe with a
* `bytes`-long MapViewOfFile, which proves only that the FIRST view fits;
* whether the mirror did depended on whatever sat after it, which ASLR
* moves on every run. When it did not, every retry found the same hint,
* all 1024 failed, and create() returned NULL -- seen in CI as acq and
* dp_tlm constructors returning NULL, a dp_xnn abort (0xc0000409), heap
* corruption and segfaults, on commits that passed the run before.
*/
void *addr = NULL;
for (int attempt = 0; attempt < 1024; ++attempt)
{
/* 1. Find a free region big enough for BOTH views. VirtualAlloc
returns it aligned to the allocation granularity, which is what
MapViewOfFileEx needs of each view's address. */
void *hint = VirtualAlloc (NULL, 2 * bytes, MEM_RESERVE, PAGE_NOACCESS);
if (!hint)
break;
VirtualFree (hint, 0, MEM_RELEASE);
/* 2. Remap first view at the hint address. */
void *v1
= MapViewOfFileEx (mapping, FILE_MAP_ALL_ACCESS, 0, 0, bytes, hint);
if (!v1)
continue; /* someone stole the address – retry */
/* 3. Remap second view immediately after the first. */
void *v2 = MapViewOfFileEx (mapping, FILE_MAP_ALL_ACCESS, 0, 0, bytes,
(char *)v1 + bytes);
if (v2)
{
addr = v1;
break;
} /* success */
UnmapViewOfFile (v1); /* second map failed – retry with new hint */
}
return addr;
}
#endif /* _WIN32 */
static inline void *
dp__buf_alloc (size_t bytes, void **handle_out)
{
*handle_out = NULL;
#ifdef _WIN32
DWORD size_hi = (DWORD)((DWORD64)bytes >> 32);
DWORD size_lo = (DWORD)(bytes & 0xFFFFFFFFULL);
HANDLE h = CreateFileMappingA (INVALID_HANDLE_VALUE, NULL, PAGE_READWRITE,
size_hi, size_lo, NULL);
if (!h)
return NULL;
void *addr = dp__win_map_twice (h, bytes);
if (!addr)
{
CloseHandle (h);
return NULL;
}
*handle_out = (void *)h;
return addr;
#else /* POSIX ----------------------------------------------------------- */
/* Obtain an anonymous fd for `bytes` of shared memory. */
int fd;
#ifdef __linux__
/* Use memfd_create when available; otherwise fall back to the raw
* syscall to avoid implicit-declaration warnings on older libc. */
#if defined(__GLIBC__)
#include <features.h>
#endif
#include <sys/syscall.h>
#ifndef SYS_memfd_create
#if defined(__x86_64__)
#define SYS_memfd_create 319
#elif defined(__aarch64__)
#define SYS_memfd_create 279
#elif defined(__i386__)
#define SYS_memfd_create 356
#endif
#endif
fd = (int)syscall (SYS_memfd_create, "dp_shm", 0);
#else
char name[64];
/* PID + pointer makes name unique even across concurrent tests. */
(void)snprintf (name, sizeof (name), "/dp_%d_%p", (int)getpid (),
(void *)&bytes);
fd = shm_open (name, O_RDWR | O_CREAT | O_EXCL, 0600);
if (fd != -1)
shm_unlink (name); /* unlink immediately; fd keeps it alive */
#endif
if (fd == -1)
return NULL;
if (ftruncate (fd, (off_t)bytes) == -1)
{
close (fd);
return NULL;
}
/* Reserve 2*bytes of virtual address space. */
void *addr
= mmap (NULL, 2 * bytes, PROT_NONE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
if (addr == MAP_FAILED)
{
close (fd);
return NULL;
}
/* Map lower half. */
if (mmap (addr, bytes, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_FIXED, fd, 0)
== MAP_FAILED)
{
munmap (addr, 2 * bytes);
close (fd);
return NULL;
}
/* Map upper half (same physical pages). */
if (mmap ((char *)addr + bytes, bytes, PROT_READ | PROT_WRITE,
MAP_SHARED | MAP_FIXED, fd, 0)
== MAP_FAILED)
{
munmap (addr, 2 * bytes);
close (fd);
return NULL;
}
close (fd); /* fd no longer needed once both views are established */
return addr;
#endif /* _WIN32 */
}
static inline void *
dp__buf_alloc_file (size_t bytes, void **handle_out, const char *path,
int *existed)
{
*handle_out = NULL;
if (existed)
*existed = 0;
#ifdef _WIN32
if (!path || !*path)
return NULL;
HANDLE f = CreateFileA (path, GENERIC_READ | GENERIC_WRITE,
FILE_SHARE_READ | FILE_SHARE_WRITE, NULL,
OPEN_ALWAYS, FILE_ATTRIBUTE_NORMAL, NULL);
if (f == INVALID_HANDLE_VALUE)
return NULL;
/* Same rule as POSIX: exactly this long is a ring somebody filled, and is
mapped as it stands; anything else is cut to zero and regrown, which
sizes it AND zeroes it -- NTFS reads an extension back as zeros. */
LARGE_INTEGER have, zero, want;
zero.QuadPart = 0;
want.QuadPart = (LONGLONG)bytes;
if (GetFileSizeEx (f, &have) && (size_t)have.QuadPart == bytes)
{
if (existed)
*existed = 1;
}
else if (!SetFilePointerEx (f, zero, NULL, FILE_BEGIN) || !SetEndOfFile (f)
|| !SetFilePointerEx (f, want, NULL, FILE_BEGIN)
|| !SetEndOfFile (f))
{
CloseHandle (f);
return NULL;
}
HANDLE h = CreateFileMappingA (f, NULL, PAGE_READWRITE,
(DWORD)((DWORD64)bytes >> 32),
(DWORD)(bytes & 0xFFFFFFFFULL), NULL);
CloseHandle (f); /* the mapping holds the file open from here */
if (!h)
return NULL;
void *addr = dp__win_map_twice (h, bytes);
if (!addr)
{
CloseHandle (h);
return NULL;
}
*handle_out = (void *)h;
return addr;
#else
if (!path || !*path)
return NULL;
int fd = open (path, O_RDWR | O_CREAT, 0600);
if (fd == -1)
return NULL;
/* A file already exactly this long is a ring somebody filled: map it as it
stands. Anything else -- absent, short, or a different geometry -- is
cut to ZERO and regrown, which sizes it AND zeroes it.
Two truncates, not one. ftruncate() to a LARGER size zeroes only the
extension and keeps what was there; to a smaller one it keeps the
leading bytes. Either way a ring of a new geometry came back holding
the old one's samples while `existed` said 0 -- "created, and zeroed".
The Windows branch above has always done it this way. */
struct stat st;
if (fstat (fd, &st) == 0 && (size_t)st.st_size == bytes)
{
if (existed)
*existed = 1;
}
else if (ftruncate (fd, 0) == -1 || ftruncate (fd, (off_t)bytes) == -1)
{
close (fd);
return NULL;
}
/* Reserve 2*bytes of address space, then place both views in it -- the
same sequence dp__buf_alloc() uses, over a different fd. */
void *addr
= mmap (NULL, 2 * bytes, PROT_NONE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
if (addr == MAP_FAILED)
{
close (fd);
return NULL;
}
if (mmap (addr, bytes, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_FIXED, fd, 0)
== MAP_FAILED
|| mmap ((char *)addr + bytes, bytes, PROT_READ | PROT_WRITE,
MAP_SHARED | MAP_FIXED, fd, 0)
== MAP_FAILED)
{
munmap (addr, 2 * bytes);
close (fd);
return NULL;
}
close (fd); /* both views hold the file open */
return addr;
#endif
}
static inline void
dp__buf_sync (void *addr, size_t bytes)
{
#ifdef _WIN32
/* Writes the dirty pages to the file, which is what survives the PROCESS.
msync(MS_SYNC) also waits for the disk; the Windows equivalent needs
FlushFileBuffers on a file handle this mapping no longer keeps. */
(void)FlushViewOfFile (addr, bytes);
#else
(void)msync (addr, bytes, MS_SYNC);
#endif
}
static inline void
dp__buf_free (void *addr, size_t bytes, void *handle)
{
#ifdef _WIN32
UnmapViewOfFile (addr);
UnmapViewOfFile ((char *)addr + bytes);
if (handle)
CloseHandle ((HANDLE)handle);
#else
munmap (addr, 2 * bytes);
(void)handle;
#endif
}
typedef enum
{
DP_WAIT_OK = 0,
DP_WAIT_PENDING = 1,
DP_WAIT_TOO_LARGE = 2,
DP_WAIT_CLOSED = 3,
DP_WAIT_INTERRUPTED = 4
} dp_wait_status_t;
/* =========================================================================
* DECLARE_DP_BUFFER(name, type)
*
* Generates a complete type-specific circular-buffer implementation.
* The macro itself contains NO #ifdef, keeping it portable by delegating
* all platform work to dp__buf_alloc / dp__buf_free above.
* ====================================================================== */
#define DECLARE_DP_BUFFER(name, type) \
\
\
typedef struct \
{ \
type *data; \
size_t mask; \
size_t capacity; \
void *_handle; \
DP_ALIGN (DP_CACHELINE) volatile size_t head; \
/* Shares the producer's line on purpose: the producer writes both, \
and a consumer reading them together reads one line. */ \
volatile int closed; \
DP_ALIGN (DP_CACHELINE) volatile size_t tail; \
\
DP_ALIGN (DP_CACHELINE) volatile size_t dropped; \
} dp_##name##_t; \
\
\
static inline dp_##name##_t *dp_##name##_wrap_ ( \
void *addr, size_t capacity, size_t mapped, void *handle) \
{ \
dp_##name##_t *ab = (dp_##name##_t *)calloc (1, sizeof (dp_##name##_t)); \
if (!ab) \
{ \
dp__buf_free (addr, mapped * sizeof (type) * 2, handle); \
return NULL; \
} \
ab->data = (type *)addr; \
ab->capacity = capacity; \
ab->mask = mapped - 1; \
ab->_handle = handle; \
return ab; \
} \
\
\
static inline size_t dp_##name##_mapped_for_ (size_t n_samples) \
{ \
size_t mapped = n_samples ? dp_next_pow_two (n_samples) : 0; \
size_t page = dp__page_size (); \
size_t elem = sizeof (type) * 2; \
/* elem and page are powers of two and so is `mapped`, so doubling until \
the unit reaches a page lands on an exact page multiple. */ \
while (mapped && mapped * elem < page) \
mapped <<= 1; \
return mapped; \
} \
\
\
static inline dp_##name##_t *dp_##name##_create (size_t n_samples) \
{ \
size_t mapped = dp_##name##_mapped_for_ (n_samples); \
if (!mapped) \
return NULL; \
void *handle = NULL; \
void *addr = dp__buf_alloc (mapped * sizeof (type) * 2, &handle); \
if (!addr) \
return NULL; \
return dp_##name##_wrap_ (addr, n_samples, mapped, handle); \
} \
\
\
static inline dp_##name##_t *dp_##name##_create_backed ( \
size_t n_samples, const char *path, int *existed) \
{ \
size_t mapped = dp_##name##_mapped_for_ (n_samples); \
if (!mapped) \
return NULL; \
void *handle = NULL; \
void *addr = dp__buf_alloc_file (mapped * sizeof (type) * 2, &handle, \
path, existed); \
if (!addr) \
return NULL; \
return dp_##name##_wrap_ (addr, n_samples, mapped, handle); \
} \
\
\
static inline void dp_##name##_sync (dp_##name##_t *ab) \
{ \
dp__buf_sync (ab->data, (ab->mask + 1) * sizeof (type) * 2); \
} \
\
\
static inline void dp_##name##_destroy (dp_##name##_t *ab) \
{ \
dp__buf_free (ab->data, (ab->mask + 1) * sizeof (type) * 2, \
ab->_handle); \
free (ab); \
} \
\
\
JM_FORCEINLINE bool dp_##name##_write (dp_##name##_t *ab, \
const type *src, size_t n) \
{ \
size_t h = DP_LOAD_RLX (&ab->head); \
size_t t = DP_LOAD_ACQ (&ab->tail); \
/* n > capacity FIRST, and not only because it can never fit: the room \
below is computed from two indices, and if they are ever wrong -- \
tail past head -- it comes out LARGER than the ring. Believing it \
copied past the mapping. consume() now refuses the one way in, but \
a memcpy must not depend on that: capacity is already loaded, so \
this costs the hot path one compare and no atomic. */ \
if (n > ab->capacity || (ab->capacity - (h - t)) < n) \
{ \
__atomic_fetch_add (&ab->dropped, n, __ATOMIC_RELAXED); \
return false; \
} \
memcpy (&ab->data[(h & ab->mask) * 2], src, n * sizeof (type) * 2); \
DP_STORE_REL (&ab->head, h + n); \
return true; \
} \
\
\
static inline void dp_##name##_close (dp_##name##_t *ab) \
{ \
__atomic_store_n (&ab->closed, 1, __ATOMIC_RELEASE); \
} \
\
\
static inline int dp_##name##_closed (const dp_##name##_t *ab) \
{ \
return __atomic_load_n (&ab->closed, __ATOMIC_ACQUIRE); \
} \
\
\
static inline type *dp_##name##_wait (dp_##name##_t *ab, size_t n) \
{ \
size_t h, t; \
/* Unsatisfiable by construction -- the ring holds at most `capacity`, \
so head - tail can never reach n. Checked before the loop because \
the loop has no exit for it. */ \
if (n > ab->capacity) \
return NULL; \
while (((h = DP_LOAD_ACQ (&ab->head)) - (t = DP_LOAD_RLX (&ab->tail))) \
< n) \
{ \
/* The re-load after observing `closed` is required, not tidy. \
close() is a RELEASE store and closed() an ACQUIRE load, so a \
consumer that observes the flag is guaranteed to see every write \
that preceded it -- but only if it looks again. Using the `h` \
read before the acquire would discard a final batch the producer \
had already published. \
\
Narrow enough that no deterministic test reaches it: the branch \
needs the consumer to see the flag and NOT yet the data, and if \
the write is already visible the loop exits above without coming \
here at all. Sabotaging it leaves the suite green, which is why \
the reasoning is written down rather than left to a test. */ \
if (dp_##name##_closed (ab)) \
{ \
h = DP_LOAD_ACQ (&ab->head); \
t = DP_LOAD_RLX (&ab->tail); \
if (h - t < n) \
return NULL; \
break; \
} \
if (dp_interrupted ()) \
return NULL; \
DP_SPIN_HINT (); \
} \
(void)h; \
return &ab->data[(t & ab->mask) * 2]; \
} \
\
\
static inline size_t dp_##name##_available (const dp_##name##_t *ab) \
{ \
size_t h = DP_LOAD_ACQ (&ab->head); \
size_t t = DP_LOAD_RLX (&ab->tail); \
return h - t; \
} \
\
\
static inline size_t dp_##name##_space (const dp_##name##_t *ab) \
{ \
size_t h = DP_LOAD_RLX (&ab->head); \
size_t t = DP_LOAD_ACQ (&ab->tail); \
return ab->capacity - (h - t); \
} \
\
\
JM_FORCEINLINE size_t dp_##name##_write_some (dp_##name##_t *ab, \
const type *src, size_t n) \
{ \
size_t h = DP_LOAD_RLX (&ab->head); \
size_t t = DP_LOAD_ACQ (&ab->tail); \
size_t space = ab->capacity - (h - t); \
if (space > ab->capacity) /* indices corrupt (see write()): stay in */ \
space = ab->capacity; /* the mapping whatever they claim */ \
if (n > space) \
n = space; \
if (n == 0) \
return 0; \
memcpy (&ab->data[(h & ab->mask) * 2], src, n * sizeof (type) * 2); \
DP_STORE_REL (&ab->head, h + n); \
return n; \
} \
\
\
JM_FORCEINLINE type *dp_##name##_peek (dp_##name##_t *ab, size_t n) \
{ \
size_t h = DP_LOAD_ACQ (&ab->head); \
size_t t = DP_LOAD_RLX (&ab->tail); \
if (h - t < n) \
return NULL; \
return &ab->data[(t & ab->mask) * 2]; \
} \
\
\
static inline dp_wait_status_t dp_##name##_wait_status ( \
const dp_##name##_t *ab, size_t n) \
{ \
if (n > ab->capacity) \
return DP_WAIT_TOO_LARGE; \
if (dp_##name##_available (ab) >= n) \
return DP_WAIT_OK; \
if (dp_##name##_closed (ab)) \
return dp_##name##_available (ab) >= n ? DP_WAIT_OK : DP_WAIT_CLOSED; \
if (dp_interrupted ()) \
return DP_WAIT_INTERRUPTED; \
return DP_WAIT_PENDING; \
} \
\
\
static inline void dp_##name##_reset (dp_##name##_t *ab) \
{ \
DP_STORE_REL (&ab->head, 0); \
DP_STORE_REL (&ab->tail, 0); \
__atomic_store_n (&ab->closed, 0, __ATOMIC_RELEASE); \
} \
\
\
static inline int dp_##name##_consume (dp_##name##_t *ab, size_t n) \
{ \
size_t h = DP_LOAD_ACQ (&ab->head); \
size_t t = DP_LOAD_RLX (&ab->tail); \
if (n > h - t) \
return DP_ERR_INVALID; \
DP_STORE_REL (&ab->tail, t + n); \
return DP_OK; \
}
/* -------------------------------------------------------------------------
* The element-typed face
* ---------------------------------------------------------------------- */
#if defined(__STDC_VERSION__) && __STDC_VERSION__ >= 201112L
#define DP_ASSERT_2X(tag, elem, type) \
_Static_assert (sizeof (elem) == 2 * sizeof (type), \
"view element must span two stored scalars")
#else
#define DP_ASSERT_2X(tag, elem, type) \
typedef char dp_assert_2x_##tag[sizeof (elem) == 2 * sizeof (type) ? 1 : -1]
#endif
#define DECLARE_DP_BUFFER_VIEW(name, type, elem) \
\
DP_ASSERT_2X (name, elem, type); \
\
static inline elem *dp_##name##_wait_view (dp_##name##_t *ab, size_t n) \
{ \
return (elem *)dp_##name##_wait (ab, n); \
} \
\
static inline elem *dp_##name##_peek_view (dp_##name##_t *ab, size_t n) \
{ \
return (elem *)dp_##name##_peek (ab, n); \
} \
\
static inline bool dp_##name##_write_view (dp_##name##_t *ab, \
const elem *src, size_t n) \
{ \
return dp_##name##_write (ab, (const type *)src, n); \
} \
\
static inline size_t dp_##name##_write_some_view ( \
dp_##name##_t *ab, const elem *src, size_t n) \
{ \
return dp_##name##_write_some (ab, (const type *)src, n); \
}
/* -------------------------------------------------------------------------
* The framed face
* ---------------------------------------------------------------------- */
#define DP_FRAMER_STATE_MAGIC DP_FOURCC ('F', 'R', 'M', 'R')
#define DP_FRAMER_STATE_WRITTEN_OFFSET(sample_t, frame_n) \
(sizeof (dp_state_hdr_t) + 2 * sizeof (uint64_t) \
+ ((frame_n) - 1) * 2 * sizeof (sample_t))
/* A stamped face is DEFINED in the includer's translation unit. clang warns
* about any unused static function defined in the MAIN file -- inline or not
* -- so a program that stamps the framed face in a .c file and calls some of
* it would be warned about the rest. The typed headers stamp it in a header,
* where no compiler warns; this keeps a .c that stamps it quiet too. */
#if defined(__GNUC__) || defined(__clang__)
#define DP_BUFFER_UNUSED __attribute__ ((unused))
#else
#define DP_BUFFER_UNUSED
#endif
#define DECLARE_DP_BUFFER_FRAMES(name, type, elem) \
DP_ASSERT_2X (name##_frames, elem, type); \
\
\
typedef struct \
{ \
dp_##name##_t *ring; \
size_t frame_n; \
size_t hop; \
size_t owed; \
uint64_t written; \
uint64_t frames; \
} dp_##name##_framer_t; \
\
\
static inline DP_BUFFER_UNUSED DP_NODISCARD int \
dp_##name##_framer_init (dp_##name##_framer_t *fr, dp_##name##_t *ring, \
size_t frame_n, size_t hop) \
{ \
if (!fr || !ring || hop == 0 || hop > frame_n || frame_n > ring->capacity \
|| dp_##name##_available (ring) != 0) \
return DP_ERR_INVALID; \
fr->ring = ring; \
fr->frame_n = frame_n; \
fr->hop = hop; \
fr->owed = 0; \
fr->written = 0; \
fr->frames = 0; \
return DP_OK; \
} \
\
\
static inline DP_BUFFER_UNUSED size_t \
dp_##name##_framer_frames_in (const dp_##name##_framer_t *fr, size_t n) \
{ \
size_t held = dp_##name##_available (fr->ring) - fr->owed; \
size_t avail = n > SIZE_MAX - held ? SIZE_MAX : held + n; \
return avail >= fr->frame_n ? (avail - fr->frame_n) / fr->hop + 1 : 0; \
} \
\
\
static inline DP_BUFFER_UNUSED void \
dp_##name##_framer_settle (dp_##name##_framer_t *fr) \
{ \
if (fr->owed) \
{ \
dp_##name##_consume (fr->ring, fr->owed); \
fr->owed = 0; \
} \
} \
\
\
static inline DP_BUFFER_UNUSED type * \
dp_##name##_framer_next (dp_##name##_framer_t *fr) \
{ \
dp_##name##_framer_settle (fr); \
type *p = dp_##name##_peek (fr->ring, fr->frame_n); \
if (p) \
{ \
fr->owed = fr->hop; \
fr->frames++; \
} \
return p; \
} \
\
\
static inline DP_BUFFER_UNUSED size_t \
dp_##name##_framer_feed (dp_##name##_framer_t *fr, const type *in, size_t n, \
size_t max_frames) \
{ \
dp_##name##_framer_settle (fr); \
size_t avail = dp_##name##_available (fr->ring); \
size_t room = n; \
if (max_frames <= (SIZE_MAX - fr->frame_n) / fr->hop) \
{ \
/* Largest buffered total that still yields <= max_frames frames. */ \
size_t a_max = fr->frame_n - 1 + max_frames * fr->hop; \
room = a_max > avail ? a_max - avail : 0; \
} \
if (n > room) \
n = room; \
n = dp_##name##_write_some (fr->ring, in, n); \
fr->written += n; \
return n; \
} \
\
\
static inline DP_BUFFER_UNUSED size_t \
dp_##name##_framer_frames_for (const dp_##name##_framer_t *fr, size_t n) \
{ \
size_t held = dp_##name##_available (fr->ring) - fr->owed; \
size_t room = fr->ring->capacity - held; \
return dp_##name##_framer_frames_in (fr, n < room ? n : room); \
} \
\
\
static inline DP_BUFFER_UNUSED size_t \
dp_##name##_framer_pending (const dp_##name##_framer_t *fr) \
{ \
uint64_t covered \
= fr->frames ? (fr->frames - 1) * fr->hop + fr->frame_n : 0; \
return (size_t)(fr->written - covered); \
} \
\
\
static inline DP_BUFFER_UNUSED size_t \
dp_##name##_framer_feed_frames (dp_##name##_framer_t *fr, const type *in, \
size_t n, size_t max_frames) \
{ \
if (max_frames == 0) \
return 0; \
dp_##name##_framer_settle (fr); \
if (max_frames - 1 <= (SIZE_MAX - fr->frame_n) / fr->hop) \
{ \
size_t end = fr->frame_n + (max_frames - 1) * fr->hop; \
size_t held = dp_##name##_available (fr->ring); \
size_t room = end > held ? end - held : 0; \
if (n > room) \
n = room; \
} \
return dp_##name##_framer_feed (fr, in, n, max_frames); \
} \
\
\
static inline DP_BUFFER_UNUSED size_t \
dp_##name##_framer_feed_carry (dp_##name##_framer_t *fr, const type *in, \
size_t n) \
{ \
return dp_##name##_framer_frames_for (fr, n) == 0 \
? dp_##name##_framer_feed (fr, in, n, 0) \
: 0; \
} \
\
\
static inline DP_BUFFER_UNUSED int \
dp_##name##_framer_drained (const dp_##name##_framer_t *fr) \
{ \
return dp_##name##_available (fr->ring) - fr->owed < fr->frame_n; \
} \
\
\
static inline DP_BUFFER_UNUSED void \
dp_##name##_framer_reset (dp_##name##_framer_t *fr) \
{ \
dp_##name##_reset (fr->ring); \
fr->owed = 0; \
fr->written = 0; \
fr->frames = 0; \
} \
\
\
static inline DP_BUFFER_UNUSED int \
dp_##name##_framer_flush (dp_##name##_framer_t *fr, type *row) \
{ \
dp_##name##_framer_settle (fr); \
if (!dp_##name##_framer_drained (fr)) \
return DP_ERR_INVALID; \
int emit = dp_##name##_framer_pending (fr) != 0; \
if (emit) \
{ \
size_t live = dp_##name##_available (fr->ring); \
memcpy (row, dp_##name##_peek (fr->ring, live), \
live * 2 * sizeof (type)); \
memset (row + live * 2, 0, (fr->frame_n - live) * 2 * sizeof (type)); \
} \
dp_##name##_framer_reset (fr); \
return emit; \
} \
\
\
\
static inline DP_BUFFER_UNUSED size_t \
dp_##name##_framer_state_bytes (const dp_##name##_framer_t *fr) \
{ \
return sizeof (dp_state_hdr_t) + 2 * sizeof (uint64_t) \
+ (fr->frame_n - 1) * 2 * sizeof (type) + 3 * sizeof (uint64_t); \
} \
\
\
static inline DP_BUFFER_UNUSED void \
dp_##name##_framer_get_state (const dp_##name##_framer_t *fr, void *blob) \
{ \
size_t total = dp_##name##_framer_state_bytes (fr); \
if (!dp_##name##_framer_drained (fr)) \
{ \
memset (blob, 0, total); \
return; \
} \
DP_GET_OPEN (DP_FRAMER_STATE_MAGIC, 1u, total); \
size_t t = DP_LOAD_RLX (&fr->ring->tail) + fr->owed; \
size_t live = dp_##name##_available (fr->ring) - fr->owed; \
dp_w_u64 (&_w, live); \
dp_w_u64 (&_w, sizeof (type)); \
dp_w_bytes (&_w, &fr->ring->data[(t & fr->ring->mask) * 2], \
live * 2 * sizeof (type)); \
/* The unused slots are written, as zeros: a byte get_state leaves alone is \
a byte of the caller's heap shipped in the blob (doppler#1471). */ \
dp_w_zeros (&_w, (fr->frame_n - 1 - live) * 2 * sizeof (type)); \
dp_w_u64 (&_w, fr->written); \
dp_w_u64 (&_w, fr->frames); \
dp_w_u64 (&_w, fr->hop); \
} \
\
/* Every refusal a snapshot can earn, in ONE place: set_state applies what \
this accepts and state_frames reports it, so the two cannot disagree on \
which blobs are snapshots. Reads, never writes. */ \
static inline DP_BUFFER_UNUSED int \
dp_##name##_framer_check_state_ (const dp_##name##_framer_t *fr, \
const void *blob, uint64_t *live, \
const void **src, uint64_t *written, \
uint64_t *frames) \
{ \
size_t want = dp_##name##_framer_state_bytes (fr); \
DP_SET_OPEN (DP_FRAMER_STATE_MAGIC, 1u, want); \
*live = dp_r_u64 (&_r); \
uint64_t elem_bytes = dp_r_u64 (&_r); \
*src = dp_r_reserve (&_r, (fr->frame_n - 1) * 2 * sizeof (type)); \
*written = dp_r_u64 (&_r); \
*frames = dp_r_u64 (&_r); \
uint64_t hop = dp_r_u64 (&_r); \
/* Frames retired exactly frames*hop samples, so what remains is written \
minus that, and a drained framer holds fewer than a frame. Once a \
frame has been handed out it also holds at least frame_n - hop: the \
last frame covered frame_n samples and only its hop was retired. \
Anything else is a corrupt blob, not a snapshot. */ \
if (_r.err || elem_bytes != sizeof (type) || hop != fr->hop \
|| *live > fr->frame_n - 1 \
|| (*frames && *live < fr->frame_n - fr->hop) \
|| *frames > UINT64_MAX / fr->hop \
|| *written < *frames * fr->hop \
|| *written - *frames * fr->hop != *live) \
return DP_ERR_INVALID; \
return DP_OK; \
} \
\
\
static inline DP_BUFFER_UNUSED int \
dp_##name##_framer_state_frames (const dp_##name##_framer_t *fr, \
const void *blob, uint64_t *frames) \
{ \
uint64_t live, written, f; \
const void *src; \
int rc = dp_##name##_framer_check_state_ (fr, blob, &live, &src, \
&written, &f); \
if (rc == DP_OK) \
*frames = f; \
return rc; \
} \
\
\
static inline DP_BUFFER_UNUSED int \
dp_##name##_framer_set_state (dp_##name##_framer_t *fr, const void *blob) \
{ \
uint64_t live, written, frames; \
const void *src; \
int rc = dp_##name##_framer_check_state_ (fr, blob, &live, &src, \
&written, &frames); \
if (rc != DP_OK) \
return rc; \
dp_##name##_framer_reset (fr); \
dp_##name##_write_some (fr->ring, (const type *)src, (size_t)live); \
fr->written = written; \
fr->frames = frames; \
return DP_OK; \
} \
\
/* The element face: one ELEMENT per sample, as the VIEW macro does. */ \
static inline DP_BUFFER_UNUSED elem * \
dp_##name##_framer_next_view (dp_##name##_framer_t *fr) \
{ \
return (elem *)dp_##name##_framer_next (fr); \
} \
\
static inline DP_BUFFER_UNUSED size_t \
dp_##name##_framer_feed_view (dp_##name##_framer_t *fr, const elem *in, \
size_t n, size_t max_frames) \
{ \
return dp_##name##_framer_feed (fr, (const type *)in, n, max_frames); \
} \
\
static inline DP_BUFFER_UNUSED size_t \
dp_##name##_framer_feed_frames_view (dp_##name##_framer_t *fr, \
const elem *in, size_t n, \
size_t max_frames) \
{ \
return dp_##name##_framer_feed_frames (fr, (const type *)in, n, \
max_frames); \
} \
\
static inline DP_BUFFER_UNUSED size_t \
dp_##name##_framer_feed_carry_view (dp_##name##_framer_t *fr, \
const elem *in, size_t n) \
{ \
return dp_##name##_framer_feed_carry (fr, (const type *)in, n); \
} \
\
static inline DP_BUFFER_UNUSED int \
dp_##name##_framer_flush_view (dp_##name##_framer_t *fr, elem *row) \
{ \
return dp_##name##_framer_flush (fr, (type *)row); \
}
/* --- Type instantiations --- */
DECLARE_DP_BUFFER (f32, float)
DECLARE_DP_BUFFER (f64, double)
DECLARE_DP_BUFFER (i16, int16_t)
#endif /* DP_BUFFER_H */