RTQueue - Lock-Free SPMC Queue

A high-performance, lock-free Single Producer Multiple Consumers (SPMC) queue implementation for real-time applications. The queue is "lossy" by design - when full, the producer automatically drops the oldest items to make room for new ones, ensuring non-blocking operation.

Features

• Lock-free: Uses atomic operations for thread-safe access without locks
• SPMC: Single producer, multiple consumers architecture
• Lossy by design: Automatically drops old items when full to maintain real-time performance
• Cache-aligned: Internal structure optimized to prevent false sharing
• Cross-platform: Supports Linux, macOS, Windows, and FreeBSD
• Dual API: Available as both C library and Python module

Requirements

Python

• Python 3.6 or higher
• C11-compatible compiler

C

• C11-compatible compiler with atomics support
• CMake 3.10 or higher (for building tests and benchmarks)

Installation

Python Module

Install directly from source:

pip install .

Or for development:

pip install -e .

C Library

Include the source files directly in your project:

#include"src/SPMCQueue.h"

Link with src/SPMCQueue.c during compilation.

For building tests and benchmarks:

mkdir build
cd build
cmake ..
make

Python API

Basic Usage

fromLossyQueueimportLossyQueue# Create a queue (size must be a power of 2)queue=LossyQueue(64)
# Producer: Put items into the queuequeue.put("message 1")
queue.put("message 2")
queue.put({"key": "value"})
# Consumer: Get items from the queueitem=queue.get() # Returns the item or None if queue is emptyifitemisnotNone:
print(f"Received: {item}")

Multi-threaded Example

fromLossyQueueimportLossyQueueimportthreadingimporttimequeue=LossyQueue(256)
defproducer():
"""Single producer thread"""foriinrange(1000):
queue.put(f"message-{i}")
time.sleep(0.001)
defconsumer(name):
"""Multiple consumer threads"""whileTrue:
item=queue.get()
ifitemisnotNone:
print(f"{name} received: {item}")
else:
time.sleep(0.001)
# Start producerproducer_thread=threading.Thread(target=producer)
producer_thread.start()
# Start multiple consumersconsumers= []
foriinrange(3):
t=threading.Thread(target=consumer, args=(f"Consumer-{i}",), daemon=True)
t.start()
consumers.append(t)
producer_thread.join()

API Reference

LossyQueue(size)

Constructor to create a new queue.

• Parameters:
  • size (int): Queue capacity. Must be a power of 2 (e.g., 64, 128, 256, 1024).
• Raises:
  • ValueError: If size is not a power of 2.
  • RuntimeError: If queue initialization fails.

put(item)

Add an item to the queue. If the queue is full, automatically removes the oldest item.

• Parameters:
  • item: Any Python object to store in the queue.
• Returns: None

put_many(items)

Add multiple items to the queue. If the queue is full, automatically removes the oldest items so every supplied item is enqueued.

• Parameters:
  • items: Any iterable of Python objects to store in the queue.
• Raises:
  • ValueError: If the iterable contains more items than the queue capacity.
• Returns: None

get()

Retrieve and remove an item from the queue.

• Returns: The next item from the queue, or None if the queue is empty.

C API

Basic Usage

#include<stdio.h>#include"SPMCQueue.h"intmain() {
// Create a queue (size must be a power of 2)SPMCQueue*queue=create_queue(64);
if (queue==NULL) {
fprintf(stderr, "Failed to create queue\n");
return1;
}
// Push itemsintvalue1=42;
intvalue2=100;
if (try_push(queue, &value1)) {
printf("Pushed value1\n");
}
if (try_push(queue, &value2)) {
printf("Pushed value2\n");
}
// Pop itemsvoid*item;
if (try_pop(queue, &item)) {
printf("Popped: %d\n", *(int*)item);
}
// Clean updestroy_queue(queue);
return0;
}

Multi-threaded Example

#include<pthread.h>#include<stdio.h>#include<stdint.h>#include"SPMCQueue.h"#defineQUEUE_SIZE 1024
#defineNUM_CONSUMERS 4
SPMCQueue*queue;
void*producer_thread(void*arg) {
for (uintptr_ti=0; i<10000; i++) {
while (!try_push(queue, (void*)i)) {
// Queue full, try again (or handle overflow)
}
}
returnNULL;
}
void*consumer_thread(void*arg) {
intid=*(int*)arg;
size_tcount=0;
while (1) {
void*value;
if (try_pop(queue, &value)) {
printf("Consumer %d got: %lu\n", id, (uintptr_t)value);
count++;
}
}
returnNULL;
}
intmain() {
queue=create_queue(QUEUE_SIZE);
pthread_tprod, cons[NUM_CONSUMERS];
intids[NUM_CONSUMERS];
// Start consumersfor (inti=0; i<NUM_CONSUMERS; i++) {
ids[i] =i;
pthread_create(&cons[i], NULL, consumer_thread, &ids[i]);
}
// Start producerpthread_create(&prod, NULL, producer_thread, NULL);
pthread_join(prod, NULL);
destroy_queue(queue);
return0;
}

Batch Operations

#include<stdio.h>#include"SPMCQueue.h"#defineBATCH_SIZE 16
voidbatch_consumer_example(SPMCQueue*queue) {
void*items[BATCH_SIZE];
// Pop multiple items at once for better performancesize_tcount=try_pop_many(queue, items, BATCH_SIZE);
printf("Popped %zu items in one batch\n", count);
for (size_ti=0; i<count; i++) {
// Process items[i]printf("Item %zu: %lu\n", i, (uintptr_t)items[i]);
}
}
voidbatch_producer_example(SPMCQueue*queue) {
void*items[BATCH_SIZE] = {
(void*)1, (void*)2, (void*)3, (void*)4,
(void*)5, (void*)6, (void*)7, (void*)8,
(void*)9, (void*)10, (void*)11, (void*)12,
(void*)13, (void*)14, (void*)15, (void*)16,
};
// Push as many items as will fit in one batchsize_tcount=try_push_many(queue, items, BATCH_SIZE);
printf("Pushed %zu items in one batch\n", count);
}

API Reference

SPMCQueue* create_queue(size_t capacity)

Create a new SPMC queue.

• Parameters:
  • capacity: Queue capacity. Must be a power of 2.
• Returns: Pointer to the queue, or NULL on failure.

void destroy_queue(SPMCQueue* queue)

Destroy a queue and free its memory.

• Parameters:
  • queue: Queue to destroy.

bool try_push(SPMCQueue* queue, void* value)

Attempt to push a value onto the queue.

• Parameters:
  • queue: The queue.
  • value: Pointer to store in the queue.
• Returns:true if successful, false if queue is full.

size_t try_push_many(SPMCQueue* queue, void** values, size_t howmany)

Attempt to push multiple values at once (batch operation).

• Parameters:
  • queue: The queue.
  • values: Array of pointers to store in the queue.
  • howmany: Maximum number of items to push.
• Returns: Number of items actually pushed (0 to howmany).

size_t try_push_many_pre(SPMCQueue* queue, void** values, size_t howmany, SPMCPrePushFunc pre_queue, void *cb_arg)

Attempt to push multiple values at once while running a callback for each item that is actually accepted.

• Parameters:
  • queue: The queue.
  • values: Array of pointers to store in the queue.
  • howmany: Maximum number of items to push.
  • pre_queue: Optional callback invoked after capacity has been confirmed for an item and before that item is made visible to consumers.
  • cb_arg: Opaque callback context pointer passed to pre_queue.
• Returns: Number of items actually pushed (0 to howmany).

bool try_pop(SPMCQueue* queue, void** value)

Attempt to pop a value from the queue.

• Parameters:
  • queue: The queue.
  • value: Pointer to store the retrieved value.
• Returns:true if successful, false if queue is empty.

size_t try_pop_many(SPMCQueue* queue, void** values, size_t howmany)

Attempt to pop multiple values at once (batch operation).

• Parameters:
  • queue: The queue.
  • values: Array to store retrieved values.
  • howmany: Maximum number of items to pop.
• Returns: Number of items actually popped (0 to howmany).

Performance Considerations

• Queue size should be a power of 2 for optimal performance
• Use larger queue sizes to reduce the chance of dropped items
• The try_push_many() and try_pop_many() functions are more efficient for high-throughput scenarios
• Internal structures are cache-line aligned to prevent false sharing
• No dynamic memory allocation during operation (all allocations happen at queue creation)

License

BSD 2-Clause License

Copyright (c) 2023, Maksym Sobolyev

See LICENSE file for details.

Contributing

Contributions are welcome! Please feel free to submit pull requests or open issues.

Testing

Run Python tests:

python -m pytest python/test_lossyqueue.py

Run C benchmarks:

cd build
make
./spmc_bench_test