Institutional Low-Latency Telemetry
& Shared-Memory IPC.

// animus_core.hpp — zero-copy, lock-free ring buffer core, cache-line aligned end to end
#pragma once
#include <cstdint>
#include <atomic>
#include <memory>

extern "C" {

struct alignas(64) RingSlot {
    std::atomic<uint64_t> sequence;
    uint8_t payload[256];   // fixed-size, no allocation
};

// Vyukov MPMC ring buffer — head/tail counters alignas(64) to avoid false sharing
class alignas(64) LockFreeRingBuffer {
public:
    explicit LockFreeRingBuffer(uint32_t capacity)
        : slots_(std::make_unique<RingSlot[]>(capacity)), capacity_(capacity) {}

    bool push(const void* data, size_t len) noexcept;
    bool pop(void* out, size_t max_len) noexcept;

private:
    alignas(64) std::atomic<uint64_t> head_{0};
    alignas(64) std::atomic<uint64_t> tail_{0};
    std::unique_ptr<RingSlot[]> slots_;
    uint32_t capacity_;
};

// Direct C-ABI export — no RPC framework, no schema compiler
__declspec(dllexport) void* animus_create(uint32_t ring_capacity);
__declspec(dllexport) bool animus_record_batch(void* engine, const uint8_t* buf, size_t len);
__declspec(dllexport) size_t animus_poll_signals(void* engine, uint8_t* out, size_t max_len);

} // extern "C"

# animus_wrapper.py — zero-dependency, ctypes + memoryview, microsecond zero-copy interop
import ctypes
import time
from ctypes import c_void_p, c_uint32, c_size_t, c_uint8, c_bool, POINTER

class AnimusEngine:
    def __init__(self, lib_path="animus_core.dll"):
        self.lib = ctypes.CDLL(lib_path)
        self.lib.animus_poll_signals.restype = c_size_t
        # Buffer allocated once; memoryview gives a zero-copy window onto it
        self._out = (c_uint8 * 65536)()
        self._view = memoryview(self._out).cast('B')
        self.handle = None

    def init(self, ring_capacity=4096):
        self.handle = self.lib.animus_create(c_uint32(ring_capacity))

    # One raw byte buffer per batch, not one marshalled object per event
    def record_events_batch(self, buf: bytes) -> bool:
        arr = (c_uint8 * len(buf)).from_buffer_copy(buf)
        return self.lib.animus_record_batch(self.handle, arr, c_size_t(len(buf)))

    # memoryview over the same ctypes buffer — zero-copy read, no (de)serialization
    def poll_signals(self) -> memoryview:
        t0 = time.perf_counter()
        n = self.lib.animus_poll_signals(self.handle, self._out, c_size_t(len(self._out)))
        elapsed_us = (time.perf_counter() - t0) * 1e6  # microsecond payload consumption, no IPC hop
        return self._view[:n]  # consumer reads directly off native memory