1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

/**
 * Thread-safe Circular Buffer (Ring Buffer)
 * 
 * Uses modulo arithmetic for wrap-around indexing.
 * Fixed memory footprint, cache-friendly access pattern.
 */
class CircularBuffer<T> {
    private readonly buffer: (T | undefined)[];
    private readonly capacity: number;
    
    // Head points to next slot to read
    // Tail points to next slot to write
    private head = 0;
    private tail = 0;
    private count = 0;
    
    // Synchronization
    private readonly lock = new Mutex();
    private readonly notFull: ConditionVariable;
    private readonly notEmpty: ConditionVariable;
    
    constructor(capacity: number) {
        // Power of 2 capacity allows bitwise AND instead of modulo
        // (optimization: capacity & (size-1) instead of % size)
        this.capacity = capacity;
        this.buffer = new Array(capacity);
        this.notFull = new ConditionVariable(this.lock);
        this.notEmpty = new ConditionVariable(this.lock);
    }
    
    async produce(item: T): Promise<void> {
        await this.lock.acquire();
        try {
            // Wait while buffer is full
            while (this.count === this.capacity) {
                await this.notFull.wait();
            }
            
            // Add item at tail position
            this.buffer[this.tail] = item;
            
            // Advance tail with wrap-around
            this.tail = (this.tail + 1) % this.capacity;
            this.count++;
            
            // Signal consumers
            this.notEmpty.signal();
        } finally {
            await this.lock.release();
        }
    }
    
    async consume(): Promise<T> {
        await this.lock.acquire();
        try {
            // Wait while buffer is empty
            while (this.count === 0) {
                await this.notEmpty.wait();
            }
            
            // Remove item from head position
            const item = this.buffer[this.head]!;
            this.buffer[this.head] = undefined;  // Help GC
            
            // Advance head with wrap-around
            this.head = (this.head + 1) % this.capacity;
            this.count--;
            
            // Signal producers
            this.notFull.signal();
            
            return item;
        } finally {
            await this.lock.release();
        }
    }
    
    // Visualization of buffer state:
    // capacity=8, head=2, tail=6, count=4
    //
    // Index:  0   1   2   3   4   5   6   7
    // Buffer: [ ] [ ] [A] [B] [C] [D] [ ] [ ]
    //              ^head          ^tail
    //
    // Items A,B,C,D are in the buffer (count=4)
    // Consumer will read A next (at head)
    // Producer will write next item at position 6 (at tail)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

/**
 * Single-Producer, Single-Consumer (SPSC) Lock-Free Queue
 * 
 * When exactly one producer and one consumer are guaranteed,
 * we can eliminate ALL locking. The indices provide natural
 * synchronization via visibility ordering.
 * 
 * This is the fastest possible bounded queue implementation.
 */
class SPSCQueue<T> {
    private readonly buffer: (T | undefined)[];
    private readonly mask: number;  // capacity - 1 for power of 2
    
    // Volatile indices - changes visible across threads
    // Each is ONLY written by one thread:
    // - head written by consumer only
    // - tail written by producer only
    private head = 0;  // Consumer reads from here
    private tail = 0;  // Producer writes to here
    
    constructor(capacity: number) {
        // Capacity MUST be power of 2 for this implementation
        if ((capacity & (capacity - 1)) !== 0) {
            throw new Error("Capacity must be power of 2");
        }
        this.buffer = new Array(capacity);
        this.mask = capacity - 1;
    }
    
    /**
     * Producer: Try to add item
     * Returns false if queue is full (non-blocking)
     */
    tryProduce(item: T): boolean {
        const currentTail = this.tail;
        const nextTail = (currentTail + 1) & this.mask;
        
        // Check if queue is full
        // Full when: nextTail == head (would overwrite unread item)
        if (nextTail === this.head) {
            return false;  // Full, don't block
        }
        
        // Write item before updating tail
        // Memory barrier ensures item is visible before tail update
        this.buffer[currentTail] = item;
        
        // Update tail (signals consumer that item is ready)
        this.tail = nextTail;
        
        return true;
    }
    
    /**
     * Consumer: Try to remove item
     * Returns undefined if queue is empty (non-blocking)
     */
    tryConsume(): T | undefined {
        const currentHead = this.head;
        
        // Check if queue is empty
        // Empty when: head == tail (nothing to read)
        if (currentHead === this.tail) {
            return undefined;  // Empty, don't block
        }
        
        // Read item before updating head
        const item = this.buffer[currentHead];
        this.buffer[currentHead] = undefined;  // Help GC
        
        // Update head (signals producer that slot is free)
        this.head = (currentHead + 1) & this.mask;
        
        return item;
    }
    
    // Why this works without locks:
    // 1. Only producer writes tail, only consumer writes head
    // 2. Producer reads head (to check full), consumer reads tail (to check empty)
    // 3. No simultaneous writes to any variable
    // 4. Memory ordering ensures writes are visible before index updates
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86

/**
 * Partitioned Buffer for High-Throughput Processing
 * 
 * Distributes load across multiple independent queues.
 * Each partition can be consumed in parallel.
 */
class PartitionedBuffer<K, V> {
    private readonly partitions: CircularBuffer<V>[];
    private readonly numPartitions: number;
    
    constructor(numPartitions: number, partitionCapacity: number) {
        this.numPartitions = numPartitions;
        this.partitions = Array.from(
            { length: numPartitions },
            () => new CircularBuffer<V>(partitionCapacity)
        );
    }
    
    /**
     * Produce to specific partition based on key
     * Messages with same key always go to same partition
     */
    async produce(key: K, value: V): Promise<void> {
        const partitionId = this.getPartition(key);
        await this.partitions[partitionId].produce(value);
    }
    
    /**
     * Consume from specific partition
     * Each consumer is assigned one or more partitions
     */
    async consumeFromPartition(partitionId: number): Promise<V> {
        if (partitionId < 0 || partitionId >= this.numPartitions) {
            throw new Error(`Invalid partition: ${partitionId}`);
        }
        return await this.partitions[partitionId].consume();
    }
    
    /**
     * Hash key to partition ID
     * Same key always maps to same partition
     */
    private getPartition(key: K): number {
        // Simple hash function (in production, use murmur3 or similar)
        const hash = this.hashCode(key);
        // Ensure non-negative with bitwise AND
        return (hash & 0x7FFFFFFF) % this.numPartitions;
    }
    
    private hashCode(key: K): number {
        const str = String(key);
        let hash = 0;
        for (let i = 0; i < str.length; i++) {
            hash = ((hash << 5) - hash) + str.charCodeAt(i);
            hash = hash & hash;  // Convert to 32-bit integer
        }
        return hash;
    }
    
    getNumPartitions(): number {
        return this.numPartitions;
    }
}
 
// Usage: Order processing with customer ID as partition key
const orderQueue = new PartitionedBuffer<string, Order>(8, 1000);
 
// Producer: Orders for same customer go to same partition
async function processNewOrder(order: Order) {
    await orderQueue.produce(order.customerId, order);
}
 
// Consumers: Each consumer handles specific partitions
async function consumerForPartition(partitionId: number) {
    while (true) {
        const order = await orderQueue.consumeFromPartition(partitionId);
        await processOrder(order);
        // Orders for same customer processed by same consumer
        // Maintains order per-customer
    }
}
 
// Start 8 consumers, one per partition
for (let i = 0; i < orderQueue.getNumPartitions(); i++) {
    consumerForPartition(i);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

// Kafka Producer-Consumer Example
 
// Producer: Send order events to Kafka
import { Kafka } from 'kafkajs';
 
const kafka = new Kafka({
    clientId: 'order-service',
    brokers: ['kafka1:9092', 'kafka2:9092', 'kafka3:9092']
});
 
const producer = kafka.producer();
 
async function sendOrder(order: Order) {
    await producer.connect();
    
    await producer.send({
        topic: 'orders',
        messages: [{
            // Key determines partition (orders for same customer → same partition)
            key: order.customerId,
            value: JSON.stringify(order),
            // Optional headers for metadata
            headers: {
                'source': 'checkout-service',
                'timestamp': Date.now().toString()
            }
        }]
    });
}
 
// Consumer: Process orders from Kafka
const consumer = kafka.consumer({ 
    groupId: 'order-processing-group'
});
 
async function startConsumer() {
    await consumer.connect();
    
    // Subscribe to topic
    await consumer.subscribe({ 
        topic: 'orders',
        fromBeginning: false  // Start from current offset
    });
    
    // Process messages
    await consumer.run({
        eachMessage: async ({ topic, partition, message }) => {
            const order = JSON.parse(message.value!.toString());
            
            console.log(`Processing order from partition ${partition}`);
            
            await processOrder(order);
            
            // Kafka auto-commits offset after processing
            // (configurable for at-least-once vs exactly-once)
        }
    });
}
 
// Start multiple consumers in same group for parallel processing
// Kafka automatically assigns partitions to consumers

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

/**
 * Backpressure Policies for Bounded Buffers
 */
 
type BackpressurePolicy = 'BLOCK' | 'DROP_NEWEST' | 'DROP_OLDEST' | 'ERROR';
 
class PolicyBuffer<T> {
    private buffer: T[] = [];
    private readonly capacity: number;
    private readonly policy: BackpressurePolicy;
    
    constructor(capacity: number, policy: BackpressurePolicy) {
        this.capacity = capacity;
        this.policy = policy;
    }
    
    produce(item: T): ProduceResult {
        if (this.buffer.length < this.capacity) {
            this.buffer.push(item);
            return { success: true };
        }
        
        // Buffer is full - apply backpressure policy
        switch (this.policy) {
            case 'BLOCK':
                // In async version, would await until space available
                throw new Error('Blocking not shown in sync example');
                
            case 'DROP_NEWEST':
                // Reject the new item, keep buffer unchanged
                return { 
                    success: false, 
                    dropped: item,
                    reason: 'Buffer full, dropped newest'
                };
                
            case 'DROP_OLDEST':
                // Remove oldest item, add new item
                const dropped = this.buffer.shift();
                this.buffer.push(item);
                return { 
                    success: true, 
                    dropped,
                    reason: 'Buffer full, dropped oldest'
                };
                
            case 'ERROR':
                // Throw error - caller must handle
                throw new BufferFullError('Buffer capacity exceeded');
                
            default:
                throw new Error(`Unknown policy: ${this.policy}`);
        }
    }
}
 
// Usage Examples:
 
// Logging: Drop oldest, keep most recent logs
const logBuffer = new PolicyBuffer<LogEntry>(1000, 'DROP_OLDEST');
 
// Metrics: Drop newest, existing samples are valuable
const metricsBuffer = new PolicyBuffer<Metric>(5000, 'DROP_NEWEST');
 
// Critical transactions: Error, never lose data silently
const transactionBuffer = new PolicyBuffer<Transaction>(100, 'ERROR');