Thread State Analysis (9:17:23 AM - 9:17:28 AM):
  BLOCKED:     47.3%  (synchronized wait on MetricsBuffer.lock)
  RUNNABLE:    31.2%  (actual work)
  WAITING:     18.4%  (condition.await in telemetry)
  TIMED_WAIT:   3.1%  (other)

public class BlockingMetricsBuffer {
    private final Object[] buffer;
    private int head = 0;
    private int tail = 0;
    private final Object lock = new Object();
    private final int capacity;
 
    public void record(MetricEvent event) {
        synchronized (lock) {
            while (isFull()) {
                try {
                    lock.wait();  // BLOCKING!
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    return;
                }
            }
            buffer[head] = event;
            head = (head + 1) % capacity;
            lock.notifyAll();
        }
    }
}

// Still problematic for high-frequency telemetry
ConcurrentLinkedQueue<MetricEvent> queue = new ConcurrentLinkedQueue<>();
 
public void record(MetricEvent event) {
    queue.offer(event);  // Lock-free, but...
}

ArrayBlockingQueue<MetricEvent> queue = new ArrayBlockingQueue<>(10000);
 
public void record(MetricEvent event) {
    try {
        queue.put(event);  // BLOCKS if full!
    } catch (InterruptedException e) {
        // ...
    }
}

public void record(MetricEvent event) {
    if (!queue.offer(event)) {
        // Queue full - what now?
        droppedMetrics.increment();
    }
}

Normal operation:
  - 4 telemetry producer threads
  - 1 consumer thread
  - Low contention
  - All algorithms perform similarly

Crisis scenario:
  - 4 telemetry producers + 20 business threads all recording metrics
  - Consumer overwhelmed, buffer filling up
  - High contention on shared state
  - OS scheduler under pressure

With blocking telemetry:
  - Producers block waiting for buffer space
  - Business threads (sharing thread pool) get delayed
  - System performance degrades
  - Telemetry shows nothing (threads are blocked!)

With lock-free telemetry:
  - Most producers make progress
  - Some producers may spin-wait
  - Under extreme contention, individual threads may starve
  - System performance partially preserved

With wait-free telemetry:
  - EVERY producer completes in bounded time
  - No thread ever waits or spins indefinitely
  - System performance guaranteed
  - Even under maximum load, telemetry keeps flowing

while (true) {
    int current = head.get();
    if (head.compareAndSet(current, current + 1)) {
        return current;  // Got a slot!
    }
    // CAS failed, retry
}

int mySlot = head.getAndIncrement();  // Always succeeds in one step!
// But now we need to handle overflow...

public class BlockingCircularBuffer<T> {
 
    private final Object[] buffer;
    private final int capacity;
 
    // Head: next position to write
    // Tail: next position to read
    private int head = 0;
    private int tail = 0;
    private int count = 0;
 
    private final Object lock = new Object();
 
    public BlockingCircularBuffer(int capacity) {
        this.capacity = capacity;
        this.buffer = new Object[capacity];
    }
 
    public void put(T element) throws InterruptedException {
        synchronized (lock) {
            // Wait while buffer is full
            while (count == capacity) {
                lock.wait();  // BLOCKING POINT
            }
 
            buffer[head] = element;
            head = (head + 1) % capacity;
            count++;
 
            lock.notifyAll();
        }
    }
 
    @SuppressWarnings("unchecked")
    public T take() throws InterruptedException {
        synchronized (lock) {
            // Wait while buffer is empty
            while (count == 0) {
                lock.wait();  // BLOCKING POINT
            }
 
            T element = (T) buffer[tail];
            buffer[tail] = null;  // Help GC
            tail = (tail + 1) % capacity;
            count--;
 
            lock.notifyAll();
            return element;
        }
    }
 
    public int size() {
        synchronized (lock) {
            return count;
        }
    }
}

Time    Thread-1        Thread-2        Thread-3        Thread-4
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
0ns     acquire lock    blocked         blocked         blocked
50ns    check full?     WAITING         WAITING         WAITING
55ns    write buffer    WAITING         WAITING         WAITING
60ns    update head     WAITING         WAITING         WAITING
65ns    notifyAll       WAITING         WAITING         WAITING
70ns    release lock    waking up...    waking up...    waking up...

// Context switch storm begins
3070ns  ---             acquire lock    blocked         blocked
3120ns  ---             check full?     WAITING         WAITING
3175ns  ---             write buffer    WAITING         WAITING
3230ns  ---             release lock    waking up...    waking up...

// And so on...

150,000 metric records/second
4 producer threads + 1 consumer thread
Buffer capacity: 10,000 entries

Observed behavior:
  Lock acquisitions/second:     150,000+
  Context switches/second:      50,000+
  Average time in synchronized: 200-300ns (uncontended)
  Average time in synchronized: 2,000-8,000ns (contended!)

  Total synchronization overhead: 300ms-1,200ms per second

  This leaves only 700ms-(-200ms) for actual work!

// Each wait() call may allocate:
// - AbstractQueuedSynchronizer$Node for wait queue
// - Condition queue nodes
// - Thread state objects
 
// Under high contention:
// 4 threads * 150,000 ops/sec * potential allocations =
// Hundreds of thousands of small allocations per second

Hot allocation sites (1 second sample):
  2,847,234 bytes: j.u.c.locks.AbstractQueuedSynchronizer$Node
    892,456 bytes: MetricEvent objects
    234,567 bytes: Various internal structures

  Total: 3.97 MB/second from synchronization alone

@BenchmarkMode(Mode.SampleTime)
@OutputTimeUnit(TimeUnit.NANOSECONDS)
@State(Scope.Benchmark)
public class BlockingBufferBenchmark {
 
    private BlockingCircularBuffer<Long> buffer;
    private AtomicLong counter;
 
    @Setup
    public void setup() {
        buffer = new BlockingCircularBuffer<>(10000);
        counter = new AtomicLong();
 
        // Start consumer thread
        Thread consumer = new Thread(() -> {
            while (!Thread.interrupted()) {
                try {
                    buffer.take();
                } catch (InterruptedException e) {
                    break;
                }
            }
        });
        consumer.setDaemon(true);
        consumer.start();
    }
 
    @Benchmark
    @Group("producers")
    @GroupThreads(4)
    public void produce() throws InterruptedException {
        buffer.put(counter.incrementAndGet());
    }
}

Benchmark                              Mode   Cnt    Score    Error  Units
BlockingBufferBenchmark.produce       sample  1000   287.43 ± 23.12  ns/op
BlockingBufferBenchmark.produce:p50   sample         198.00          ns/op
BlockingBufferBenchmark.produce:p90   sample         412.00          ns/op
BlockingBufferBenchmark.produce:p99   sample        1847.00          ns/op
BlockingBufferBenchmark.produce:p99.9 sample       12456.00          ns/op
BlockingBufferBenchmark.produce:max   sample       89234.00          ns/op

Throughput: ~3.5 million operations/second

// Traditional (lock-free but not wait-free):
while (true) {
    int current = head.get();
    if (head.compareAndSet(current, current + 1)) {
        break;  // Might loop forever under contention
    }
}
 
// Wait-free approach:
int myPosition = head.getAndIncrement();  // Always succeeds in one step!

long writePosition = head.getAndIncrement();
int index = (int) (writePosition & mask);
 
// Check if we're about to overwrite unread data
long currentTail = tail.get();
if (writePosition - currentTail >= capacity) {
    // We've caught up to (or passed) the tail
    // Advance the tail to make room
    tail.compareAndSet(currentTail, writePosition - capacity + 1);
}
 
// Now write - this slot is ours
buffer[index] = element;

private static final VarHandle HEAD;
private static final VarHandle TAIL;
private static final VarHandle BUFFER;
 
static {
    try {
        MethodHandles.Lookup lookup = MethodHandles.lookup();
        HEAD = lookup.findVarHandle(WaitFreeBuffer.class, "head", long.class);
        TAIL = lookup.findVarHandle(WaitFreeBuffer.class, "tail", long.class);
        BUFFER = MethodHandles.arrayElementVarHandle(Object[].class);
    } catch (Exception e) {
        throw new ExceptionInInitializerError(e);
    }
}

// Padding before head
long p01, p02, p03, p04, p05, p06, p07;
 
private volatile long head;
 
// Padding between head and tail
long p11, p12, p13, p14, p15, p16, p17;
 
private volatile long tail;
 
// Padding after tail
long p21, p22, p23, p24, p25, p26, p27;

package com.trading.telemetry;
 
import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;
 
/**
 * Wait-Free Overwrite Buffer for High-Frequency Telemetry
 *
 * Design principles:
 * 1. Wait-free guarantee: Every operation completes in bounded steps
 * 2. Overwrite semantics: Old data is overwritten when buffer is full
 * 3. Zero allocation: No allocations after initialization
 * 4. Cache-line padding: Prevents false sharing between producers/consumer
 *
 * Performance characteristics:
 * - Producer: ~10-20ns per write (wait-free)
 * - Consumer: ~15-25ns per read
 * - Throughput: 50+ million ops/sec with 4 producers
 *
 * Trade-offs:
 * - Data loss under heavy load (by design)
 * - Consumer may see gaps in sequence
 * - Not suitable for ordered/reliable messaging
 *
 * @param <T> Element type stored in the buffer
 */
public class WaitFreeOverwriteBuffer<T> {
 
    // ========== VarHandle Setup ==========
 
    private static final VarHandle HEAD;
    private static final VarHandle TAIL;
    private static final VarHandle BUFFER;
 
    static {
        try {
            MethodHandles.Lookup lookup = MethodHandles.lookup();
            HEAD = lookup.findVarHandle(
                WaitFreeOverwriteBuffer.class, "head", long.class);
            TAIL = lookup.findVarHandle(
                WaitFreeOverwriteBuffer.class, "tail", long.class);
            BUFFER = MethodHandles.arrayElementVarHandle(Object[].class);
        } catch (ReflectiveOperationException e) {
            throw new ExceptionInInitializerError(e);
        }
    }
 
    // ========== Cache Line Padding for Head ==========
 
    // 7 longs = 56 bytes padding before head
    @SuppressWarnings("unused")
    private long p01, p02, p03, p04, p05, p06, p07;
 
    /**
     * Next position for producers to claim.
     * Monotonically increasing; wraps via modulo for array access.
     * Uses getAndIncrement for wait-free slot claiming.
     */
    private volatile long head = 0;
 
    // 7 longs = 56 bytes padding between head and tail
    @SuppressWarnings("unused")
    private long p11, p12, p13, p14, p15, p16, p17;
 
    // ========== Cache Line Padding for Tail ==========
 
    /**
     * Next position for consumer to read.
     * May be advanced by producers during overflow.
     * Consumer is sole reader; producers may advance via CAS.
     */
    private volatile long tail = 0;
 
    // 7 longs = 56 bytes padding after tail
    @SuppressWarnings("unused")
    private long p21, p22, p23, p24, p25, p26, p27;
 
    // ========== Buffer Storage ==========
 
    /** Pre-allocated buffer array. Size is always power of 2. */
    private final Object[] buffer;
 
    /** Buffer capacity (power of 2 for fast modulo). */
    private final int capacity;
 
    /** Bit mask for index calculation: capacity - 1. */
    private final int mask;
 
    // ========== Metrics (optional, for monitoring) ==========
 
    /** Count of items written. */
    private volatile long writtenCount = 0;
 
    /** Count of items overwritten before being read. */
    private volatile long overwrittenCount = 0;
 
    // ========== Constructor ==========
 
    /**
     * Creates a new wait-free overwrite buffer.
     *
     * @param requestedCapacity Minimum capacity (will be rounded up to power of 2)
     * @throws IllegalArgumentException if capacity less than 2
     */
    public WaitFreeOverwriteBuffer(int requestedCapacity) {
        if (requestedCapacity < 2) {
            throw new IllegalArgumentException(
                "Capacity must be at least 2, got: " + requestedCapacity);
        }
 
        this.capacity = roundUpToPowerOf2(requestedCapacity);
        this.mask = this.capacity - 1;
        this.buffer = new Object[this.capacity];
    }
 
    private static int roundUpToPowerOf2(int value) {
        int highBit = Integer.highestOneBit(value);
        return (highBit == value) ? value : highBit << 1;
    }
 
    // ========== Producer Operations ==========
 
    /**
     * Records a telemetry event to the buffer.
     *
     * WAIT-FREE GUARANTEE: This method always completes in bounded time,
     * regardless of other threads or buffer state.
     *
     * If the buffer is full, the oldest unread data will be overwritten.
     * This is by design - for telemetry, recent data is more valuable.
     *
     * @param element The element to record (should not be null)
     */
    public void record(T element) {
        // Step 1: Claim a position (WAIT-FREE - always succeeds immediately)
        // getAndIncrement is a single atomic operation that cannot fail
        long writePosition = (long) HEAD.getAndAdd(this, 1L);
 
        // Step 2: Calculate array index using bitwise AND (faster than modulo)
        int index = (int) (writePosition & mask);
 
        // Step 3: Check for overflow condition
        // If writePosition has caught up to tail, we're overwriting unread data
        long currentTail = (long) TAIL.getOpaque(this);
 
        if (writePosition - currentTail >= capacity) {
            // Overflow detected! We're about to overwrite unread data.
            // Advance the tail to "forget" the oldest entries.
 
            // Calculate where tail should be to make room for our write
            long newTail = writePosition - capacity + 1;
 
            // Try to advance tail. Use CAS because:
            // - Multiple overflow producers might race
            // - Consumer might have already advanced tail
            // - We only advance if tail hasn't moved past our target
 
            long observedTail = currentTail;
            while (observedTail < newTail) {
                // Try to advance tail
                if (TAIL.compareAndSet(this, observedTail, newTail)) {
                    // Successfully advanced tail
                    // Count how many entries we skipped
                    overwrittenCount += (newTail - observedTail);
                    break;
                }
                // CAS failed - someone else advanced tail
                // Re-read and check if we still need to advance
                observedTail = (long) TAIL.getOpaque(this);
            }
        }
 
        // Step 4: Write the element
        // Use setRelease to ensure the write is visible to the consumer
        // after it sees the tail advance past this position
        BUFFER.setRelease(buffer, index, element);
 
        // Step 5: Update metrics
        writtenCount++;
    }
 
    /**
     * Batch record multiple elements.
     * More efficient than individual record() calls due to reduced overhead.
     *
     * @param elements Array of elements to record
     * @param offset Starting offset in the array
     * @param length Number of elements to record
     */
    public void recordBatch(T[] elements, int offset, int length) {
        for (int i = 0; i < length; i++) {
            record(elements[offset + i]);
        }
    }
 
    // ========== Consumer Operations ==========
 
    /**
     * Retrieves and removes the next element from the buffer.
     *
     * This method is designed for SINGLE CONSUMER use only.
     * Using multiple consumers will result in data corruption.
     *
     * @return The next element, or null if buffer is empty
     */
    @SuppressWarnings("unchecked")
    public T poll() {
        // Read current positions
        long currentTail = tail;
        long currentHead = (long) HEAD.getOpaque(this);
 
        // Check if buffer is empty
        if (currentTail >= currentHead) {
            return null;
        }
 
        // Calculate array index
        int index = (int) (currentTail & mask);
 
        // Read the element with acquire semantics
        // This ensures we see all writes that happened before the producer's release
        T element = (T) BUFFER.getAcquire(buffer, index);
 
        // Clear the slot to help GC (optional, but recommended)
        BUFFER.setRelease(buffer, index, null);
 
        // Advance tail (plain write is safe - single consumer)
        tail = currentTail + 1;
 
        return element;
    }
 
    /**
     * Drains all available elements to the provided consumer function.
     * More efficient than repeated poll() calls.
     *
     * @param consumer Function to process each element
     * @return Number of elements drained
     */
    @SuppressWarnings("unchecked")
    public int drain(java.util.function.Consumer<T> consumer) {
        long currentTail = tail;
        long currentHead = (long) HEAD.getOpaque(this);
 
        int count = 0;
        while (currentTail < currentHead) {
            int index = (int) (currentTail & mask);
            T element = (T) BUFFER.getAcquire(buffer, index);
 
            if (element != null) {
                consumer.accept(element);
                BUFFER.setRelease(buffer, index, null);
                count++;
            }
 
            currentTail++;
        }
 
        // Batch update tail
        tail = currentTail;
        return count;
    }
 
    /**
     * Drains up to maxElements to the provided consumer.
     * Useful for rate-limited processing.
     *
     * @param consumer Function to process each element
     * @param maxElements Maximum number of elements to drain
     * @return Number of elements actually drained
     */
    @SuppressWarnings("unchecked")
    public int drainTo(java.util.function.Consumer<T> consumer, int maxElements) {
        long currentTail = tail;
        long currentHead = (long) HEAD.getOpaque(this);
        long available = currentHead - currentTail;
 
        int toDrain = (int) Math.min(available, maxElements);
 
        for (int i = 0; i < toDrain; i++) {
            int index = (int) (currentTail & mask);
            T element = (T) BUFFER.getAcquire(buffer, index);
 
            if (element != null) {
                consumer.accept(element);
                BUFFER.setRelease(buffer, index, null);
            }
 
            currentTail++;
        }
 
        tail = currentTail;
        return toDrain;
    }
 
    // ========== Query Operations ==========
 
    /**
     * Returns approximate size of unread elements.
     * May be stale due to concurrent modifications.
     */
    public int size() {
        long currentHead = (long) HEAD.getOpaque(this);
        long currentTail = tail;
        long size = currentHead - currentTail;
 
        if (size < 0) return 0;
        if (size > capacity) return capacity;
        return (int) size;
    }
 
    /**
     * Returns true if buffer appears empty.
     */
    public boolean isEmpty() {
        return (long) HEAD.getOpaque(this) <= tail;
    }
 
    /**
     * Returns the buffer's capacity.
     */
    public int capacity() {
        return capacity;
    }
 
    /**
     * Returns total number of items written since creation.
     */
    public long getWrittenCount() {
        return writtenCount;
    }
 
    /**
     * Returns number of items overwritten before being read.
     */
    public long getOverwrittenCount() {
        return overwrittenCount;
    }
 
    /**
     * Returns the data loss ratio (overwritten / written).
     * A high ratio indicates the consumer can't keep up.
     */
    public double getDataLossRatio() {
        long written = writtenCount;
        if (written == 0) return 0.0;
        return (double) overwrittenCount / written;
    }
}

// Wait-free (always succeeds in one step)
long writePosition = (long) HEAD.getAndAdd(this, 1L);
 
// vs. Lock-free (might retry indefinitely)
while (true) {
    long current = head;
    if (HEAD.compareAndSet(this, current, current + 1)) {
        break;
    }
}

long currentTail = (long) TAIL.getOpaque(this);

while (observedTail < newTail) {
    if (TAIL.compareAndSet(this, observedTail, newTail)) {
        break;
    }
    observedTail = (long) TAIL.getOpaque(this);
}

BUFFER.setRelease(buffer, index, element);

// 1. Plain access - no barriers
long localCopy = someField;
someField = newValue;
 
// 2. Opaque access - prevents compiler caching
long fresh = (long) VARHANDLE.getOpaque(this);
 
// 3. Release/Acquire - establishes happens-before
VARHANDLE.setRelease(this, value);  // Writer
value = (long) VARHANDLE.getAcquire(this);  // Reader

private volatile Object[] buffer;  // Doesn't help - array reference is volatile
                                   // but elements are not!

@BenchmarkMode(Mode.SampleTime)
@OutputTimeUnit(TimeUnit.NANOSECONDS)
@Warmup(iterations = 5, time = 2)
@Measurement(iterations = 10, time = 5)
@Fork(value = 3, jvmArgs = {
    "-Xms4g", "-Xmx4g",
    "-XX:+UseG1GC",
    "-XX:+AlwaysPreTouch"
})
@State(Scope.Benchmark)
public class TelemetryBufferBenchmark {
 
    @Param({"1", "4", "8", "16"})
    private int producerCount;
 
    private BlockingCircularBuffer<Long> blockingBuffer;
    private WaitFreeOverwriteBuffer<Long> waitFreeBuffer;
    private AtomicLong counter;
    private volatile boolean running;
    private Thread consumerThread;
 
    @Setup(Level.Trial)
    public void setup() {
        blockingBuffer = new BlockingCircularBuffer<>(65536);
        waitFreeBuffer = new WaitFreeOverwriteBuffer<>(65536);
        counter = new AtomicLong();
        running = true;
 
        // Background consumer that drains both buffers
        consumerThread = new Thread(() -> {
            while (running) {
                try {
                    blockingBuffer.take();
                } catch (InterruptedException e) {
                    break;
                }
                waitFreeBuffer.poll();
            }
        });
        consumerThread.start();
    }
 
    @TearDown(Level.Trial)
    public void teardown() throws InterruptedException {
        running = false;
        consumerThread.interrupt();
        consumerThread.join(1000);
    }
 
    @Benchmark
    @Group("blocking")
    @GroupThreads(4)
    public void blockingRecord() throws InterruptedException {
        blockingBuffer.put(counter.incrementAndGet());
    }
 
    @Benchmark
    @Group("waitfree")
    @GroupThreads(4)
    public void waitFreeRecord() {
        waitFreeBuffer.record(counter.incrementAndGet());
    }
}

Latency Distribution (4 producers, log scale)

Blocking Buffer:
10ns   |
20ns   |
50ns   |█
100ns  |████████████████
200ns  |██████████████████████████████████████████████ (peak)
500ns  |████████████████████
1μs    |████████
2μs    |███
5μs    |██
10μs   |█
50μs+  |█ (tail)

Wait-Free Buffer:
10ns   |████████████████████████ (peak)
20ns   |██████████████████████████████████████████████████████
50ns   |████
100ns  |█
200ns+ | (rare outliers)

L1-dcache-load-misses:    1,247,891,234
LLC-load-misses:              23,456,789
cycles per operation:               ~690

L1-dcache-load-misses:      187,234,567
LLC-load-misses:               3,123,456
cycles per operation:                ~45

public class TelemetryMonitor {
 
    private final WaitFreeOverwriteBuffer<?> buffer;
    private final ScheduledExecutorService scheduler;
 
    private long lastWrittenCount = 0;
    private long lastOverwrittenCount = 0;
 
    public TelemetryMonitor(WaitFreeOverwriteBuffer<?> buffer) {
        this.buffer = buffer;
        this.scheduler = Executors.newSingleThreadScheduledExecutor();
    }
 
    public void start() {
        scheduler.scheduleAtFixedRate(this::checkHealth,
            1, 1, TimeUnit.SECONDS);
    }
 
    private void checkHealth() {
        long currentWritten = buffer.getWrittenCount();
        long currentOverwritten = buffer.getOverwrittenCount();
 
        long writtenDelta = currentWritten - lastWrittenCount;
        long overwrittenDelta = currentOverwritten - lastOverwrittenCount;
 
        double lossRate = writtenDelta > 0
            ? (double) overwrittenDelta / writtenDelta
            : 0.0;
 
        if (lossRate > 0.01) {  // > 1% loss
            logger.warn("Telemetry data loss: {:.2f}% ({} of {} events)",
                lossRate * 100, overwrittenDelta, writtenDelta);
        }
 
        if (lossRate > 0.10) {  // > 10% loss
            alerting.fire("TELEMETRY_HIGH_DATA_LOSS",
                "Loss rate: " + lossRate * 100 + "%");
        }
 
        // Export metrics for dashboards
        metrics.gauge("telemetry.write_rate", writtenDelta);
        metrics.gauge("telemetry.loss_rate", lossRate);
        metrics.gauge("telemetry.buffer_size", buffer.size());
 
        lastWrittenCount = currentWritten;
        lastOverwrittenCount = currentOverwritten;
    }
}

Too small:
  - High data loss under load
  - Consumer can't keep up
  - Metrics gaps during spikes

Too large:
  - Higher memory usage
  - Longer drain times
  - Older data when backlogged

Sizing formula:
  buffer_size = peak_write_rate * acceptable_latency * safety_factor

Example:
  peak_write_rate = 200,000 events/sec
  acceptable_latency = 100ms (time for consumer to catch up)
  safety_factor = 2x

  buffer_size = 200,000 * 0.1 * 2 = 40,000 entries

  Round up to power of 2: 65,536 (64K entries)

public class TelemetryConsumer implements Runnable {
 
    private final WaitFreeOverwriteBuffer<MetricEvent> buffer;
    private final MetricsSink sink;
    private volatile boolean running = true;
 
    @Override
    public void run() {
        while (running) {
            int drained = buffer.drain(event -> {
                try {
                    sink.record(event);
                } catch (Exception e) {
                    // Log but don't crash - telemetry shouldn't kill the system
                    logger.warn("Failed to record metric", e);
                }
            });
 
            if (drained == 0) {
                // Buffer empty - back off to reduce CPU
                LockSupport.parkNanos(100_000);  // 100μs
            }
        }
    }
 
    public void stop() {
        running = false;
    }
}

public class BatchingConsumer {
 
    private static final int BATCH_SIZE = 1000;
    private final List<MetricEvent> batch = new ArrayList<>(BATCH_SIZE);
 
    public void processBatch(WaitFreeOverwriteBuffer<MetricEvent> buffer) {
        batch.clear();
 
        buffer.drainTo(batch::add, BATCH_SIZE);
 
        if (!batch.isEmpty()) {
            // Process entire batch together - more efficient
            metricsSink.recordBatch(batch);
        }
    }
}

public class SamplingConsumer {
 
    private final WaitFreeOverwriteBuffer<MetricEvent> buffer;
    private final Random random = ThreadLocalRandom.current();
 
    public void process() {
        int size = buffer.size();
 
        // If buffer is getting full, sample instead of draining everything
        double sampleRate = size > buffer.capacity() * 0.8
            ? 0.1  // Sample 10% when buffer is 80%+ full
            : 1.0; // Process everything otherwise
 
        buffer.drain(event -> {
            if (random.nextDouble() < sampleRate) {
                sink.record(event);
            }
        });
    }
}

// Using Java Thread Affinity library
AffinityLock producerLock = AffinityLock.acquireCore();
try {
    // Producer runs on dedicated core
    while (running) {
        buffer.record(generateEvent());
    }
} finally {
    producerLock.release();
}

# Pin producer JVM to cores 0-3
taskset -c 0-3 java -jar producer.jar
 
# Pin consumer JVM to core 4
taskset -c 4 java -jar consumer.jar

java \
  -Xms4g -Xmx4g \                    # Fixed heap size
  -XX:+UseZGC \                       # Low-latency GC
  -XX:+AlwaysPreTouch \               # Pre-fault heap pages
  -XX:-UseBiasedLocking \             # Disable biased locking
  -XX:+UseNUMA \                      # NUMA awareness
  -XX:+PerfDisableSharedMem \         # Disable perf shared memory
  -Djava.lang.Integer.IntegerCache.high=10000 \  # Larger integer cache
  -jar telemetry.jar

public class TypedMetricEvent {
    public enum Type { COUNTER, GAUGE, HISTOGRAM, TIMER }
 
    private final Type type;
    private final String name;
    private final long value;
    private final long timestamp;
    private final String[] tags;
 
    // Use object pooling to avoid allocation
    private static final ThreadLocal<TypedMetricEvent> POOL =
        ThreadLocal.withInitial(TypedMetricEvent::new);
 
    public static TypedMetricEvent acquire() {
        return POOL.get();
    }
 
    public TypedMetricEvent set(Type type, String name, long value, String... tags) {
        this.type = type;
        this.name = name;
        this.value = value;
        this.timestamp = System.nanoTime();
        this.tags = tags;
        return this;
    }
}

public class ShardedTelemetryBuffer<T> {
 
    private final WaitFreeOverwriteBuffer<T>[] shards;
    private final ThreadLocal<Integer> shardIndex;
 
    @SuppressWarnings("unchecked")
    public ShardedTelemetryBuffer(int shardCount, int shardCapacity) {
        shards = new WaitFreeOverwriteBuffer[shardCount];
        for (int i = 0; i < shardCount; i++) {
            shards[i] = new WaitFreeOverwriteBuffer<>(shardCapacity);
        }
 
        AtomicInteger counter = new AtomicInteger();
        shardIndex = ThreadLocal.withInitial(() ->
            counter.getAndIncrement() % shardCount);
    }
 
    public void record(T element) {
        // Each thread writes to its own shard - zero contention!
        shards[shardIndex.get()].record(element);
    }
 
    public void drainAll(java.util.function.Consumer<T> consumer) {
        for (WaitFreeOverwriteBuffer<T> shard : shards) {
            shard.drain(consumer);
        }
    }
}

public class WindowedAggregator {
 
    private final WaitFreeOverwriteBuffer<MetricEvent> buffer;
    private final Duration windowSize;
    private final Map<String, LongAdder> currentWindow = new ConcurrentHashMap<>();
    private volatile long windowStart = System.currentTimeMillis();
 
    public void recordCounter(String name, long delta) {
        buffer.record(new MetricEvent(name, delta, System.nanoTime()));
    }
 
    public void aggregate() {
        long now = System.currentTimeMillis();
 
        if (now - windowStart >= windowSize.toMillis()) {
            // Window complete - emit aggregates
            Map<String, Long> snapshot = new HashMap<>();
            currentWindow.forEach((k, v) -> snapshot.put(k, v.sumThenReset()));
 
            emitAggregates(snapshot);
            windowStart = now;
        }
 
        // Drain events into current window
        buffer.drain(event -> {
            currentWindow
                .computeIfAbsent(event.getName(), k -> new LongAdder())
                .add(event.getValue());
        });
    }
}

public class BackpressureAwareTelemetry {
 
    private final WaitFreeOverwriteBuffer<MetricEvent> buffer;
    private final AtomicBoolean backpressureSignal = new AtomicBoolean(false);
 
    public void record(MetricEvent event) {
        buffer.record(event);
 
        // Signal backpressure when buffer is getting full
        double utilization = (double) buffer.size() / buffer.capacity();
        backpressureSignal.set(utilization > 0.8);
    }
 
    public boolean isBackpressured() {
        return backpressureSignal.get();
    }
 
    // Upstream can use this to shed load
    public void recordIfNotBackpressured(MetricEvent event) {
        if (!isBackpressured()) {
            record(event);
        }
        // Silently drop if backpressured
    }
}

// Creating the buffer
WaitFreeOverwriteBuffer<MetricEvent> buffer =
    new WaitFreeOverwriteBuffer<>(65536);
 
// Recording metrics (from any thread)
buffer.record(new MetricEvent("request.latency", latencyNs));
 
// Consuming metrics (single consumer thread)
buffer.drain(event -> sink.record(event));
 
// Monitoring health
double lossRate = buffer.getDataLossRatio();
if (lossRate > 0.01) {
    alert("Telemetry data loss: " + (lossRate * 100) + "%");
}

package com.trading.benchmark;
 
import org.openjdk.jmh.annotations.*;
import org.openjdk.jmh.infra.Blackhole;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
 
@BenchmarkMode(Mode.SampleTime)
@OutputTimeUnit(TimeUnit.NANOSECONDS)
@Warmup(iterations = 5, time = 2)
@Measurement(iterations = 10, time = 5)
@Fork(value = 3, jvmArgs = {"-Xms4g", "-Xmx4g", "-XX:+UseG1GC"})
@State(Scope.Benchmark)
public class FullTelemetryBenchmark {
 
    @Param({"4", "8", "16"})
    private int producers;
 
    @Param({"16384", "65536", "262144"})
    private int bufferSize;
 
    private BlockingCircularBuffer<Long> blockingBuffer;
    private WaitFreeOverwriteBuffer<Long> waitFreeBuffer;
    private AtomicLong counter;
    private volatile boolean running;
    private Thread consumer;
 
    @Setup(Level.Trial)
    public void setup() {
        blockingBuffer = new BlockingCircularBuffer<>(bufferSize);
        waitFreeBuffer = new WaitFreeOverwriteBuffer<>(bufferSize);
        counter = new AtomicLong();
        running = true;
 
        consumer = new Thread(() -> {
            while (running) {
                try {
                    blockingBuffer.take();
                } catch (InterruptedException e) {
                    break;
                }
                waitFreeBuffer.poll();
            }
        });
        consumer.setDaemon(true);
        consumer.start();
    }
 
    @TearDown(Level.Trial)
    public void teardown() throws InterruptedException {
        running = false;
        consumer.interrupt();
        consumer.join(1000);
    }
 
    @Benchmark
    @Group("blocking")
    @GroupThreads(4)
    public void blockingWrite(Blackhole bh) throws InterruptedException {
        bh.consume(blockingBuffer.put(counter.incrementAndGet()));
    }
 
    @Benchmark
    @Group("waitfree")
    @GroupThreads(4)
    public void waitFreeWrite(Blackhole bh) {
        waitFreeBuffer.record(counter.incrementAndGet());
        bh.consume(true);
    }
 
    @Benchmark
    @Group("blocking_throughput")
    @GroupThreads(4)
    @BenchmarkMode(Mode.Throughput)
    @OutputTimeUnit(TimeUnit.SECONDS)
    public void blockingThroughput() throws InterruptedException {
        blockingBuffer.put(counter.incrementAndGet());
    }
 
    @Benchmark
    @Group("waitfree_throughput")
    @GroupThreads(4)
    @BenchmarkMode(Mode.Throughput)
    @OutputTimeUnit(TimeUnit.SECONDS)
    public void waitFreeThroughput() {
        waitFreeBuffer.record(counter.incrementAndGet());
    }
}

// WRONG: No memory ordering guarantee
buffer[index] = element;
head = newHead;  // Consumer might see new head before element is written!
 
// CORRECT: Release semantics ensure element is visible before head update
BUFFER.setRelease(buffer, index, element);
// Or if using volatile:
buffer[index] = element;
VarHandle.releaseFence();
head = newHead;  // Now correctly ordered

// Dangerous if positions can wrap
if (writePosition - currentTail >= capacity) {
    // Overflow detected
}

private volatile long head = 0;  // 64-bit, practically never wraps
private volatile long tail = 0;

// WRONG: Counters share cache line with hot fields
private volatile long head;
private volatile long writtenCount;  // False sharing!
private volatile long overwrittenCount;  // False sharing!
private volatile long tail;
 
// CORRECT: Pad counters onto separate cache lines
private volatile long head;
long p1, p2, p3, p4, p5, p6, p7;  // Padding
 
private volatile long writtenCount;
long p8, p9, p10, p11, p12, p13, p14;
 
private volatile long overwrittenCount;
long p15, p16, p17, p18, p19, p20, p21;
 
private volatile long tail;

// DANGEROUS: Two places that might call poll()
scheduler.scheduleAtFixedRate(() -> {
    while (buffer.poll() != null) { /* process */ }
}, 0, 100, TimeUnit.MILLISECONDS);
 
// Somewhere else, on shutdown:
void shutdown() {
    while (buffer.poll() != null) { /* drain remaining */ }
}

public class TelemetryConsumer {
    private final AtomicBoolean running = new AtomicBoolean(true);
    private final Thread consumerThread;
 
    public TelemetryConsumer(WaitFreeOverwriteBuffer<?> buffer) {
        consumerThread = new Thread(() -> {
            while (running.get()) {
                buffer.drain(this::process);
                LockSupport.parkNanos(100_000);
            }
            // Final drain on shutdown
            buffer.drain(this::process);
        });
        consumerThread.start();
    }
 
    public void shutdown() {
        running.set(false);
        consumerThread.interrupt();
        try {
            consumerThread.join(5000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

// Problem: null could mean empty OR actual null value
T element = (T) BUFFER.getAcquire(buffer, index);
if (element == null) {
    return null;  // Is buffer empty, or was null stored?
}

public void record(T element) {
    if (element == null) {
        throw new NullPointerException("Null elements not permitted");
    }
    // ...
}

private static final Object EMPTY = new Object();
 
// Store wrapper instead of raw value
buffer[index] = (element == null) ? NULL_WRAPPER : element;

private final long[] slotState;  // Parallel array tracking filled/empty

@Test
void verifyWaitFreeProperty() {
    WaitFreeOverwriteBuffer<Long> buffer = new WaitFreeOverwriteBuffer<>(1024);
    AtomicLong maxOperationTime = new AtomicLong(0);
 
    // Run many operations and track worst-case time
    for (int i = 0; i < 10_000_000; i++) {
        long start = System.nanoTime();
        buffer.record((long) i);
        long elapsed = System.nanoTime() - start;
 
        maxOperationTime.updateAndGet(max -> Math.max(max, elapsed));
    }
 
    // Wait-free means bounded worst case
    // Anything over 1us suggests blocking or unbounded spinning
    assertTrue(maxOperationTime.get() < 1000,
        "Worst case time " + maxOperationTime.get() + "ns exceeds wait-free bound");
}

public class DualTelemetryRecorder {
    private final BlockingCircularBuffer<MetricEvent> oldBuffer;
    private final WaitFreeOverwriteBuffer<MetricEvent> newBuffer;
    private final AtomicLong oldCount = new AtomicLong();
    private final AtomicLong newCount = new AtomicLong();
 
    public void record(MetricEvent event) {
        // Always record to new buffer (never blocks)
        newBuffer.record(event);
        newCount.incrementAndGet();
 
        // Try to record to old buffer (might block)
        try {
            if (oldBuffer.offer(event, 1, TimeUnit.MILLISECONDS)) {
                oldCount.incrementAndGet();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

public class TelemetryRouter {
    private final WaitFreeOverwriteBuffer<MetricEvent> newBuffer;
    private final BlockingCircularBuffer<MetricEvent> oldBuffer;
    private volatile int newBufferPercentage = 0;  // 0-100
    private final Random random = ThreadLocalRandom.current();
 
    public void record(MetricEvent event) {
        if (random.nextInt(100) < newBufferPercentage) {
            newBuffer.record(event);
        } else {
            try {
                oldBuffer.put(event);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
 
    // Called by feature flag system
    public void setNewBufferPercentage(int percentage) {
        this.newBufferPercentage = Math.max(0, Math.min(100, percentage));
    }
}

taskset -c 7 java -jar telemetry-consumer.jar

// SLOW: Aggregation inline with consumption
buffer.drain(event -> {
    String key = event.getName() + ":" + Arrays.toString(event.getTags());
    aggregates.computeIfAbsent(key, k -> new LongAdder()).add(event.getValue());
});

// FAST: Just collect, aggregate later
List<MetricEvent> batch = new ArrayList<>(10000);
buffer.drainTo(batch::add, 10000);
 
// Aggregate in bulk (more cache-friendly)
for (MetricEvent event : batch) {
    // ... aggregation logic
}

telemetry:
  buffer:
    capacity: 262144  # 256K entries
    type: wait-free-overwrite
 
  consumer:
    thread-affinity: core-7
    drain-batch-size: 10000
    idle-park-nanos: 100000  # 100us
 
  monitoring:
    loss-alert-threshold: 0.01  # 1%
    loss-critical-threshold: 0.05  # 5%
 
  jvm:
    gc: ZGC
    heap: 8g
    flags:
      - -XX:+AlwaysPreTouch
      - -XX:-UseBiasedLocking