Lock-Free Ring Buffer Implementation for Multi-Threaded Ingestion

Overview

This implementation provides high-performance, lock-free ring buffers for multi-threaded data ingestion in the StellarFlow backend. The system eliminates mutex contention through atomic operations and circular buffer algorithms, enabling massive throughput improvements for high-volume data streams.

Architecture

Core Components

LockFreeRingBuffer<T> - Single Producer, Single Consumer (SPSC) ring buffer
MPSCRingBuffer<T> - Multi-Producer, Single Consumer (MPSC) ring buffer
LockFreeIngestionChannel - Channel wrapper for ingestion use cases
LockFreeChannelManager - Multi-channel management system
WorkerThreadChannel - Worker thread integration for true multi-threading
UnifiedIngestionSystem - Integration layer with existing backpressure system

Design Principles

Lock-Free: No mutexes or locks, using atomic operations instead
Wait-Free: Operations complete in bounded time regardless of contention
Cache-Friendly: Power-of-2 capacity for fast modulo via bitmask
Memory Efficient: Pre-allocated circular buffer, no dynamic allocation
FIFO Guaranteed: Strict ordering for all operations

Performance Characteristics

Benchmarks

Based on performance benchmarks (scripts/benchmark-lockfree-queue.ts):

Implementation	Ops/sec (10k)	Avg Latency (μs)	Memory (MB)
SPSC Lock-Free	~2,500,000	~0.4	~0.1
MPSC Lock-Free	~2,200,000	~0.5	~0.1
Traditional Queue	~800,000	~1.2	~0.3
Backpressure Manager	~600,000	~1.5	~0.4

Performance Improvement: 200-300% over traditional queues

Key Metrics

Throughput: 2M+ ops/sec for SPSC, 2.2M+ ops/sec for MPSC
Latency: Sub-microsecond average latency
Memory: Constant memory usage, no GC pressure
Scalability: Linear scaling with producer count (MPSC)

Usage

Basic SPSC Ring Buffer

import { LockFreeRingBuffer } from "./queue/lockFreeRingBuffer";

const buffer = new LockFreeRingBuffer<IngestionPacket>({
  capacity: 1024,
  enableMetrics: true,
  enableBatching: false,
  batchSize: 1,
});

// Enqueue
const packet: IngestionPacket = {
  priority: PacketPriority.STANDARD,
  data: { value: 42 },
  timestamp: Date.now(),
};

if (buffer.tryEnqueue(packet)) {
  console.log("Enqueued successfully");
}

// Dequeue
const item = buffer.tryDequeue();
if (item) {
  console.log("Dequeued:", item);
}

MPSC Ring Buffer (Multiple Producers)

import { MPSCRingBuffer } from "./queue/lockFreeRingBuffer";

const buffer = new MPSCRingBuffer<IngestionPacket>({
  capacity: 1024,
  enableMetrics: true,
  enableBatching: true,
  batchSize: 32,
});

// Multiple producers can enqueue concurrently
producer1.tryEnqueue(packet);
producer2.tryEnqueue(packet);
producer3.tryEnqueue(packet);

// Single consumer dequeues in batches
const batch = buffer.tryDequeueBatch(32);

Ingestion Channel

import { LockFreeIngestionChannel } from "./queue/lockFreeRingBuffer";

const channel = new LockFreeIngestionChannel("price-updates", {
  capacity: 1024,
  enableMetrics: true,
  enableBatching: true,
  batchSize: 32,
});

// Publish to channel
channel.publish(packet);

// Subscribe from channel
const packets = channel.subscribe(32);

Channel Manager

import { LockFreeChannelManager } from "./queue/lockFreeRingBuffer";

const manager = new LockFreeChannelManager({
  capacity: 1024,
  enableMetrics: true,
  enableBatching: true,
  batchSize: 32,
});

// Get or create channels
const priceChannel = manager.getChannel("price-updates");
const cacheChannel = manager.getChannel("cache-invalidation");

// Get aggregate metrics
const metrics = manager.getAggregateMetrics();
console.log(`Total utilization: ${metrics.averageUtilization * 100}%`);

Unified Ingestion System

import { createIngestionSystem } from "./queue/ingestionIntegration";

const system = createIngestionSystem({
  useLockFree: true,
  useWorkerThreads: false,
  backpressureConfig: {
    maxCapacity: 1000,
    dropThreshold: 0.9,
    slowDownThreshold: 0.7,
  },
  ringBufferConfig: {
    capacity: 1024,
    batchSize: 32,
  },
});

// Enqueue with automatic method selection
await system.enqueue(packet, "price-updates");

// Dequeue
const item = await system.dequeue("price-updates");

// Get metrics
const metrics = await system.getMetrics();

Integration with Existing System

Backpressure Integration

The lock-free system integrates seamlessly with the existing backpressure manager:

import { UnifiedIngestionSystem } from "./queue/ingestionIntegration";

// Use lock-free for high-throughput channels
const system = new UnifiedIngestionSystem({
  useLockFree: true,
  useWorkerThreads: false,
  backpressureConfig: existingBackpressureConfig,
});

// Fallback to traditional queue when needed
system.updateConfig({ useLockFree: false });

Channel Constants

Use existing channel constants:

import { CHANNELS } from "./modules/constants/channels";

const priceChannel = manager.getChannel(CHANNELS.PRICE_UPDATES);
const cacheChannel = manager.getChannel(CHANNELS.CACHE_INVALIDATION);

Configuration

Ring Buffer Configuration

interface RingBufferConfig {
  capacity: number;        // Buffer size (auto-rounded to power of 2)
  enableMetrics: boolean;  // Enable performance metrics
  enableBatching: boolean; // Enable batch operations
  batchSize: number;       // Default batch size for operations
}

Recommended Configurations

High-Throughput Price Updates:

{
  capacity: 4096,
  enableMetrics: true,
  enableBatching: true,
  batchSize: 64,
}

Low-Latency Cache Invalidation:

{
  capacity: 256,
  enableMetrics: true,
  enableBatching: false,
  batchSize: 1,
}

General Purpose:

{
  capacity: 1024,
  enableMetrics: true,
  enableBatching: true,
  batchSize: 32,
}

Worker Thread Integration

For CPU-intensive ingestion, use worker threads:

import { WorkerChannelPool } from "./queue/workerThreadChannel";

const pool = new WorkerChannelPool({
  bufferSize: 1024,
  batchSize: 32,
});

// Get worker-backed channel
const channel = await pool.getChannel("heavy-processing");

// Publish (non-blocking, offloaded to worker)
await channel.publish(packet);

// Subscribe
const packets = await channel.subscribe();

// Cleanup
await pool.shutdown();

Note: Worker threads add overhead (~100μs per operation). Use only for CPU-intensive operations.

Metrics and Monitoring

Ring Buffer Metrics

interface RingBufferMetrics {
  size: number;              // Current buffer size
  capacity: number;          // Total capacity
  utilization: number;       // 0-1 utilization ratio
  totalEnqueued: number;     // Total items enqueued
  totalDequeued: number;     // Total items dequeued
  enqueueFailures: number;   // Failed enqueue attempts
  dequeueFailures: number;   // Failed dequeue attempts
  averageLatency: number;    // Average operation latency (ms)
  peakLatency: number;       // Peak operation latency (ms)
  batchesProcessed: number;  // Number of batches processed
}

Monitoring Example

const channel = manager.getChannel("price-updates");
const metrics = channel.getMetrics();

console.log(`Utilization: ${(metrics.utilization * 100).toFixed(2)}%`);
console.log(`Throughput: ${metrics.totalEnqueued / (metrics.totalDequeued || 1)}x`);
console.log(`Avg Latency: ${metrics.averageLatency.toFixed(3)}ms`);
console.log(`Failures: ${metrics.enqueueFailures + metrics.dequeueFailures}`);

Alerting Thresholds

Recommended alerting thresholds:

Utilization > 80%: Scale up capacity
Failures > 1%: Investigate contention
Latency > 1ms: Check system load
Peak Latency > 10ms: Immediate investigation

Testing

Running Tests

# Run lock-free queue tests
npm run test:jest lockFreeRingBuffer.test.ts

# Run performance benchmarks
tsx scripts/benchmark-lockfree-queue.ts

Test Coverage

The test suite covers:

Basic enqueue/dequeue operations
FIFO order guarantees
Buffer full/empty conditions
Metrics tracking
Batch operations
Channel management
Aggregate metrics
Concurrency scenarios
Buffer wraparound

Performance Optimization Tips

1. Choose Right Capacity

Too small: Frequent failures, backpressure
Too large: Memory waste, cache misses
Optimal: 70-80% average utilization

2. Enable Batching

Batching reduces operation overhead:

{
  enableBatching: true,
  batchSize: 32,  // Tune based on workload
}

3. Use Appropriate Buffer Type

SPSC: Single producer, single consumer (fastest)
MPSC: Multiple producers, single consumer (most flexible)
Worker Threads: CPU-intensive processing (highest overhead)

4. Monitor Metrics

Regularly review metrics to identify bottlenecks:

setInterval(() => {
  const metrics = channel.getMetrics();
  if (metrics.utilization > 0.9) {
    console.warn("High utilization detected");
  }
}, 5000);

Troubleshooting

High Failure Rate

Symptom: Many enqueue/dequeue failures

Causes:

Buffer too small for workload
Consumer too slow
Burst traffic patterns

Solutions:

Increase buffer capacity
Add more consumers
Enable backpressure

High Latency

Symptom: Operations taking >1ms

Causes:

System under heavy load
Memory pressure
GC pauses

Solutions:

Check system resources
Reduce batch size
Disable metrics temporarily

Memory Growth

Symptom: Memory usage increasing over time

Causes:

Packets not consumed
Metrics not reset
Channel leaks

Solutions:

Ensure consumer is running
Reset metrics periodically
Remove unused channels

Migration Guide

From Traditional Queue

Before:

import { AsyncBoundedQueue } from "./queue/backpressure";

const queue = new AsyncBoundedQueue<IngestionPacket>(1000);
await queue.put(packet);
const item = await queue.get();

After:

import { LockFreeIngestionChannel } from "./queue/lockFreeRingBuffer";

const channel = new LockFreeIngestionChannel("channel", {
  capacity: 1024,
  enableMetrics: true,
});
channel.publish(packet);
const items = channel.subscribe(1);

From Backpressure Manager

Before:

import { BackpressureManager } from "./queue/backpressure";

const manager = new BackpressureManager(config);
await manager.enqueue(packet);
const item = await manager.dequeue();

After:

import { UnifiedIngestionSystem } from "./queue/ingestionIntegration";

const system = createIngestionSystem({
  useLockFree: true,
  backpressureConfig: config,
});
await system.enqueue(packet, "channel");
const item = await system.dequeue("channel");

Security Considerations

Memory Safety

Pre-allocated buffer prevents heap exhaustion
Power-of-2 capacity prevents overflow
Type-safe generics prevent data corruption

DoS Protection

Buffer limits prevent memory exhaustion
Failure tracking detects abuse patterns
Backpressure integration prevents overload

Data Integrity

FIFO ordering guarantees message sequence
Atomic operations prevent data races
Metrics track data loss

Best Practices

1. Start with SPSC

Use SPSC when possible for maximum performance:

const buffer = new LockFreeRingBuffer<T>(config);

Upgrade to MPSC only when multiple producers are needed.

2. Monitor Continuously

Set up monitoring for all channels:

const metrics = manager.getAllMetrics();
for (const [name, m] of metrics) {
  console.log(`${name}: ${m.utilization * 100}%`);
}

3. Handle Failures Gracefully

Always check return values:

if (!channel.publish(packet)) {
  // Handle backpressure
  await backoff();
}

4. Use Appropriate Batch Sizes

Tune batch size based on workload:

Small items: Batch size 32-64
Large items: Batch size 8-16
Mixed: Batch size 16-32

5. Clean Up Resources

Always shutdown channels when done:

await pool.shutdown();
manager.shutdown();

API Reference