Packet Queue

📬 Thread-safe per-client packet queues More...

Data Structures
struct	queued_packet_t
	Single packet ready to send (header already in network byte order) More...
struct	packet_node
	Node in the packet queue linked list. More...
struct	node_pool
	Memory pool for packet nodes to reduce malloc/free overhead. More...
struct	packet_queue_t
	Thread-safe packet queue for producer-consumer communication. More...

Typedefs
typedef struct packet_node	packet_node_t
	Forward declaration for packet queue node.
typedef struct node_pool	node_pool_t
	Memory pool for packet nodes to reduce malloc/free overhead.

Variables
packet_header_t	queued_packet_t::header
	Complete packet header (already in network byte order)
void *	queued_packet_t::data
	Packet payload data (can be NULL for header-only packets)
size_t	queued_packet_t::data_len
	Length of payload data in bytes.
bool	queued_packet_t::owns_data
	If true, free data when packet is freed.
buffer_pool_t *	queued_packet_t::buffer_pool
	Pool that allocated the data (NULL if malloc'd)
queued_packet_t	packet_node::packet
	The queued packet data.
atomic_ptr_t	packet_node::next
	Pointer to next node in linked list (NULL for tail) - atomic for lock-free operations.
atomic_ptr_t	node_pool::free_list
	Stack of free nodes (LIFO for cache locality) - atomic for lock-free access.
packet_node_t *	node_pool::nodes
	Pre-allocated array of all nodes.
size_t	node_pool::pool_size
	Total number of nodes in pool.
atomic_t	node_pool::used_count
	Number of nodes currently in use - atomic for lock-free tracking.
atomic_ptr_t	packet_queue_t::head
	Front of queue (dequeue from here) - atomic for lock-free access.
atomic_ptr_t	packet_queue_t::tail
	Back of queue (enqueue here) - atomic for lock-free access.
atomic_t	packet_queue_t::count
	Number of packets currently in queue - atomic for lock-free access.
size_t	packet_queue_t::max_size
	Maximum queue size (0 = unlimited)
atomic_t	packet_queue_t::bytes_queued
	Total bytes of data queued (for monitoring) - atomic for lock-free access.
node_pool_t *	packet_queue_t::node_pool
	Optional memory pool for nodes (NULL = use malloc/free)
buffer_pool_t *	packet_queue_t::buffer_pool
	Optional memory pool for data buffers (NULL = use malloc/free)
atomic_t	packet_queue_t::packets_enqueued
	Total packets enqueued (statistics) - atomic for lock-free access.
atomic_t	packet_queue_t::packets_dequeued
	Total packets dequeued (statistics) - atomic for lock-free access.
atomic_t	packet_queue_t::packets_dropped
	Total packets dropped due to queue full (statistics) - atomic for lock-free access.
atomic_t	packet_queue_t::shutdown
	Shutdown flag (true = dequeue returns NULL) - atomic for lock-free access.

Node Pool Functions
node_pool_t *	node_pool_create (size_t pool_size)
	Create a memory pool for packet queue nodes.
void	node_pool_destroy (node_pool_t *pool)
	Destroy a node pool and free all memory.
packet_node_t *	node_pool_get (node_pool_t *pool)
	Get a free node from the pool.
void	node_pool_put (node_pool_t *pool, packet_node_t *node)
	Return a node to the pool.

Queue Management Functions
packet_queue_t *	packet_queue_create (size_t max_size)
	Create a new packet queue.
packet_queue_t *	packet_queue_create_with_pool (size_t max_size, size_t pool_size)
	Create a packet queue with node pool.
packet_queue_t *	packet_queue_create_with_pools (size_t max_size, size_t node_pool_size, bool use_buffer_pool)
	Create a packet queue with both node and buffer pools.
void	packet_queue_stop (packet_queue_t *queue)
	Destroy a packet queue and free all resources.

Queue Operations
int	packet_queue_enqueue (packet_queue_t *queue, packet_type_t type, const void *data, size_t data_len, uint32_t client_id, bool copy_data)
	Enqueue a packet into the queue.
int	packet_queue_enqueue_packet (packet_queue_t *queue, const queued_packet_t *packet)
	Enqueue a pre-built packet (for special cases like compressed frames)
queued_packet_t *	packet_queue_dequeue (packet_queue_t *queue)
	Dequeue a packet from the queue (non-blocking)
queued_packet_t *	packet_queue_try_dequeue (packet_queue_t *queue)
	Try to dequeue a packet without blocking.
void	packet_queue_free_packet (queued_packet_t *packet)
	Free a dequeued packet.

Queue Status Functions
size_t	packet_queue_size (packet_queue_t *queue)
	Get current number of packets in queue.
bool	packet_queue_is_empty (packet_queue_t *queue)
	Check if queue is empty.
bool	packet_queue_is_full (packet_queue_t *queue)
	Check if queue is full.

Queue Control Functions
void	packet_queue_destroy (packet_queue_t *queue)
	Signal queue shutdown (causes dequeue to return NULL)
void	packet_queue_clear (packet_queue_t *queue)
	Clear all packets from queue.

Statistics Functions
void	packet_queue_get_stats (packet_queue_t *queue, uint64_t *enqueued, uint64_t *dequeued, uint64_t *dropped)
	Get queue statistics.
bool	packet_queue_validate_packet (const queued_packet_t *packet)
	Validate packet integrity.

Detailed Description

📬 Thread-safe per-client packet queues

This module implements a high-performance thread-safe queue for network packets, enabling producer threads (audio mixer, video broadcast) to enqueue packets while consumer threads (per-client send threads) dequeue and transmit them. This design eliminates shared bottlenecks and enables linear scaling across multiple clients.

CORE RESPONSIBILITIES:

1. Thread-safe packet queuing for producer-consumer architecture
2. Per-client queue isolation (each client has separate audio/video queues)
3. Memory-efficient packet node pooling to reduce malloc/free overhead
4. Queue size limits with automatic packet dropping under load
5. Graceful shutdown handling for clean thread termination
6. Comprehensive statistics collection for performance monitoring

ARCHITECTURAL OVERVIEW:

PRODUCER-CONSUMER MODEL:

• Producer threads (audio mixer, video render): Enqueue packets asynchronously
• Consumer threads (per-client send threads): Dequeue packets and transmit
• Lock-free operations with atomic retry patterns (no busy-waiting with backoff)

QUEUE DESIGN:

• Linked-list implementation with head/tail pointers for O(1) enqueue/dequeue
• Optional node pool reduces malloc/free overhead for high-frequency operations
• Optional buffer pool integration for zero-allocation packet data management
• Size limits prevent unbounded memory growth under load

MEMORY MANAGEMENT:

• Node pool pre-allocates queue nodes to eliminate per-packet allocations
• Buffer pool integration allows zero-allocation packet data (when available)
• Automatic cleanup when queue is destroyed
• Statistics track memory efficiency

THREAD SAFETY:

SYNCHRONIZATION PRIMITIVES:

• Atomic operations for lock-free queue state (head, tail, count, etc.)
• Atomic statistics fields for performance monitoring
• Atomic shutdown flag for graceful termination
• No mutexes or condition variables (lock-free design)

LOCK-FREE PATTERN:

• All queue operations use atomic operations for thread-safe access
• Memory ordering (acquire/release) ensures proper synchronization
• Non-blocking dequeue operations (caller should retry if needed)
• Lock-free operations eliminate lock contention bottlenecks

INTEGRATION WITH OTHER MODULES:

• buffer_pool.h: Optional integration for zero-allocation packet data
• network/packet.h: Uses packet_header_t and packet_type_t
• Per-client send threads: Consume packets from queues
• Audio mixer / Video render: Produce packets into queues

PERFORMANCE CHARACTERISTICS:

• O(1) enqueue/dequeue operations (amortized)
• Zero-allocation operation when pools are configured
• Minimal lock contention due to per-client queue isolation
• Automatic backpressure via queue size limits

Note

Each client should have separate queues for audio and video to enable prioritization and prevent one stream from blocking the other.

When max_size > 0, full queues drop oldest packets automatically. This prevents memory exhaustion but may cause frame drops under load.

Queue shutdown causes dequeue operations to return NULL, allowing consumer threads to exit gracefully without blocking forever.

Warning

Always free dequeued packets with packet_queue_free_packet() to properly release memory (either to pool or free()).

Queue destruction does NOT free packets still in the queue. Clear the queue first or ensure all packets are dequeued.

Author

Zachary Fogg me@zf.nosp@m.o.gg

Date

September 2025

Version

2.0 (Post-Modularization)

Packet Queue

Overview

Welcome! Let's talk about packet queues—the unsung heroes that keep ascii-chat's network communication smooth and organized.

Imagine you're at a busy restaurant. Orders are coming in from multiple tables, and the kitchen needs to process them one at a time. You don't want orders getting mixed up or lost, right? That's exactly what packet queues do for network data—they line up incoming packets so each client's data gets processed in order, without chaos.

Each client connected to the server gets their own dedicated packet queue. When packets arrive over the network, they get placed in the appropriate client's queue. Then, a separate thread can process packets at its own pace without blocking the network thread. It's like having a separate order ticket for each table—clean, organized, and efficient.

Implementation: lib/packet_queue.c/h

What makes packet queues useful?

• Per-client queues: Each client has their own queue (no cross-contamination!)
• Thread-safe: Multiple threads can safely interact with the queue
• Flexible ownership: Choose whether the queue copies data or just references it
• Buffer pool integration: Works with our buffer pool for efficient memory usage
• Statistics tracking: See how well your queues are performing

Architecture

Queue Structure

// Single packet node in the queue
typedef struct packet_node {
    packet_header_t header;   // Type, length, CRC, client_id
    void *data;               // Payload
    struct packet_node *next; // Next packet in queue
} packet_node_t;
 
// Per-client packet queue
typedef struct packet_queue {
    packet_node_t *head;   // Front of queue (dequeue here)
    packet_node_t *tail;   // Back of queue (enqueue here)
    size_t size;           // Current packet count
    size_t max_size;       // Maximum allowed packets (0 = unlimited)
    bool owns_data;        // If true, queue frees packet data
    mutex_t mutex;         // Thread safety
 
    // Statistics
    uint64_t packets_enqueued;
    uint64_t packets_dequeued;
    uint64_t packets_dropped;  // Dropped due to max_size limit
    size_t peak_size;          // Peak queue depth
} packet_queue_t;

Data Ownership

Queues can either copy packet data or reference it:

Deep copy mode (owns_data = true):

• Queue allocates its own copy of packet data
• Safe for producer to free original data immediately
• Queue frees data when packet dequeued
• Higher memory usage, but safer

Reference mode (owns_data = false):

• Queue stores pointer to original data
• Producer must keep data alive until dequeued
• Consumer responsible for freeing data
• Lower memory usage, but requires careful lifetime management

// Deep copy mode (safer) - copy_data=true
packet_queue_t *q = packet_queue_create(100);
 
char *data = SAFE_MALLOC(1024, char*);
strcpy(data, "Hello");
packet_queue_enqueue(q, PACKET_TYPE_ASCII_FRAME, data, strlen(data), client_id, true);
SAFE_FREE(data);  // Can free immediately since copy_data=true!
 
// Reference mode (lower overhead) - copy_data=false
packet_queue_t *q2 = packet_queue_create(100);
 
char *data2 = SAFE_MALLOC(1024, char*);
strcpy(data2, "Hello");
packet_queue_enqueue(q2, PACKET_TYPE_ASCII_FRAME, data2, strlen(data2), client_id, false);
// data2 must stay alive until dequeued since copy_data=false!
 
// Dequeue returns a queued_packet_t pointer
queued_packet_t *packet = packet_queue_dequeue(q2);
if (packet) {
    process_packet(packet->data);
    packet_queue_free_packet(q2, packet);  // Free packet properly
}

API Reference

Creation/Destruction

// Create packet queue
packet_queue_t *queue = packet_queue_create(100, true);
if (!queue) {
    log_error("Failed to create packet queue");
    return NULL;
}
 
// Use queue...
 
// Destroy queue
packet_queue_destroy(queue);

Integration

Buffer Pool Integration:

• Packet queues use buffer pool for efficient memory allocation
• Node pool pre-allocates packet nodes
• Data pool pre-allocates packet data buffers
• Reduces malloc/free overhead in high-throughput scenarios

Network Integration:

• Each client has a dedicated packet queue
• Network receive thread enqueues packets
• Client processing thread dequeues packets
• Decouples network I/O from packet processing

Performance

Throughput:

• Enqueue: ~1M packets/second (mutex-protected)
• Dequeue: ~1M packets/second (mutex-protected)
• Memory overhead: ~40 bytes per queued packet

Scalability:

• Per-client queues eliminate contention between clients
• Statistics tracking has minimal overhead
• Buffer pool reduces allocation overhead

Thread Safety

Mutex Protection:

• All queue operations are mutex-protected
• Thread-safe for concurrent enqueue/dequeue
• No lock-free optimizations (simplicity over speed)

Concurrent Access:

• Multiple threads can enqueue concurrently
• Only one thread should dequeue (per queue)
• Statistics are updated atomically

See also

packet_queue.h

buffer_pool.h

Typedef Documentation

node_pool_t

typedef struct node_pool node_pool_t

#include <queue.h>

Memory pool for packet nodes to reduce malloc/free overhead.

Pre-allocates a fixed-size array of packet nodes and maintains a lock-free free list using atomic compare-and-swap operations. This eliminates per-packet malloc/free calls and avoids lock contention for high-frequency packet queue operations.

Note

Pool operations are thread-safe via lock-free atomics (no mutexes).

If pool is exhausted, operations fall back to malloc/free.

packet_node_t

typedef struct packet_node packet_node_t

#include <queue.h>

Forward declaration for packet queue node.

Definition at line 137 of file queue.h.

Function Documentation

node_pool_create()

node_pool_t * node_pool_create

(

size_t

pool_size

)

#include <queue.h>

Create a memory pool for packet queue nodes.

Parameters

pool_size

Number of nodes to pre-allocate

Returns

Pointer to node pool, or NULL on failure

Pre-allocates a fixed-size array of packet nodes to eliminate per-packet malloc/free overhead. All nodes are initialized and added to the free list.

Note

Pool size should be chosen based on expected queue depth. Too small causes fallback to malloc, too large wastes memory.

Definition at line 25 of file queue.c.

                                                {
  if (pool_size == 0) {
    return NULL;
  }
 
  node_pool_t *pool;
  pool = SAFE_MALLOC(sizeof(node_pool_t), node_pool_t *);
 
  // Allocate all nodes at once
  pool->nodes = SAFE_MALLOC(sizeof(packet_node_t) * pool_size, packet_node_t *);
 
  // Link all nodes into free list (using atomic stores for consistency with atomic type)
  for (size_t i = 0; i < pool_size - 1; i++) {
    atomic_ptr_store(&pool->nodes[i].next, (void *)&pool->nodes[i + 1]);
  }
  atomic_ptr_store(&pool->nodes[pool_size - 1].next, NULL);
 
  // Initialize atomics for lock-free operations
  atomic_ptr_store(&pool->free_list, (void *)&pool->nodes[0]);
  pool->pool_size = pool_size;
  atomic_store_u64(&pool->used_count, 0);
 
  NAMED_REGISTER_NODE_POOL(pool, "node_pool", NULL);
 
  return pool;
}

References atomic_ptr_store(), atomic_store_u64(), buffer_pool::free_list, NAMED_REGISTER_NODE_POOL, pool, and SAFE_MALLOC.

Referenced by packet_queue_create_with_pools().

node_pool_destroy()

void node_pool_destroy

(

node_pool_t *

pool

)

#include <queue.h>

Destroy a node pool and free all memory.

Parameters

pool	Node pool to destroy (can be NULL)

Frees all pre-allocated nodes and the pool structure itself.

Warning

Pool must not be in use by any queues when destroyed.

Definition at line 52 of file queue.c.

                                          {
  if (!pool) {
    return;
  }
 
  NAMED_UNREGISTER(pool);
 
  SAFE_FREE(pool->nodes);
  SAFE_FREE(pool);
}

References NAMED_UNREGISTER, pool, and SAFE_FREE.

Referenced by packet_queue_destroy().

node_pool_get()

packet_node_t * node_pool_get

(

node_pool_t *

pool

)

#include <queue.h>

Get a free node from the pool.

Parameters

pool

Node pool

Returns

Pointer to free node, or NULL if pool is exhausted

Removes a node from the free list. If pool is exhausted, returns NULL (caller should fall back to malloc).

Note

Thread-safe (protected by pool_mutex).

Definition at line 63 of file queue.c.

                                                {
  if (!pool) {
    // No pool, fallback to malloc
    packet_node_t *node;
    node = SAFE_MALLOC(sizeof(packet_node_t), packet_node_t *);
    return node;
  }
 
  // Lock-free pop from free_list stack using CAS (same pattern as buffer_pool)
  packet_node_t *node = (packet_node_t *)atomic_ptr_load(&pool->free_list);
  while (node) {
    packet_node_t *next = (packet_node_t *)atomic_ptr_load(&node->next);
    if (atomic_ptr_cas(&pool->free_list, (void **)&node, next)) {
      // Successfully popped - clear next pointer and increment used_count
      atomic_ptr_store(&node->next, (void *)(packet_node_t *)NULL);
      atomic_fetch_add_u64(&pool->used_count, 1);
      return node;
    }
    // CAS failed - reload and retry (another thread grabbed the node)
    node = (packet_node_t *)atomic_ptr_load(&pool->free_list);
  }
 
  // Pool exhausted, fallback to malloc
  node = SAFE_MALLOC(sizeof(packet_node_t), packet_node_t *);
  size_t used = atomic_load_u64(&pool->used_count);
  log_debug("Memory pool exhausted, falling back to SAFE_MALLOC (used: %zu/%zu)", used, pool->pool_size);
 
  return node;
}

References atomic_fetch_add_u64(), atomic_load_u64(), atomic_ptr_cas(), atomic_ptr_load(), atomic_ptr_store(), buffer_pool::free_list, log_debug, packet_node::next, pool, and SAFE_MALLOC.

Referenced by packet_queue_enqueue(), and packet_queue_enqueue_packet().

node_pool_put()

void node_pool_put	(	node_pool_t *	pool,
		packet_node_t *	node
	)

#include <queue.h>

Return a node to the pool.

Parameters

pool	Node pool
node	Node to return (must have been allocated from this pool)

Returns a node to the free list for reuse. Node should not be in use by any queue when returned.

Note

Thread-safe (protected by pool_mutex).

Warning

Do not return nodes that are still in use by queues.

Definition at line 93 of file queue.c.

                                                           {
  if (!node) {
    return;
  }
 
  if (!pool) {
    // No pool, just free
    SAFE_FREE(node);
    return;
  }
 
  // Check if this node is from our pool
  bool is_pool_node = (node >= pool->nodes && node < pool->nodes + pool->pool_size);
 
  if (is_pool_node) {
    // Lock-free push to free_list stack using CAS
    packet_node_t *head = (packet_node_t *)atomic_ptr_load(&pool->free_list);
    do {
      atomic_ptr_store(&node->next, (void *)head);
    } while (!atomic_ptr_cas(&pool->free_list, (void **)&head, node));
    atomic_fetch_sub_u64(&pool->used_count, 1);
  } else {
    // This was malloc'd, so free it
    SAFE_FREE(node);
  }
}

References atomic_fetch_sub_u64(), atomic_ptr_cas(), atomic_ptr_load(), atomic_ptr_store(), buffer_pool::free_list, packet_node::next, pool, and SAFE_FREE.

Referenced by packet_queue_enqueue(), packet_queue_enqueue_packet(), and packet_queue_try_dequeue().

packet_queue_clear()

void packet_queue_clear

(

packet_queue_t *

queue

)

#include <queue.h>

Clear all packets from queue.

Parameters

queue

Packet queue

Removes and frees all packets currently in the queue. Useful for cleanup before queue destruction or when resetting queue state.

Note

Thread-safe (lock-free using atomic operations).

This function frees all queued packets, so ensure nothing is still referencing them.

Definition at line 626 of file queue.c.

                                               {
  if (!queue)
    return;
 
  // Lock-free clear: drain queue by repeatedly dequeuing until empty
  queued_packet_t *packet;
  while ((packet = packet_queue_try_dequeue(queue)) != NULL) {
    packet_queue_free_packet(packet);
  }
}

References packet_queue_free_packet(), and packet_queue_try_dequeue().

Referenced by packet_queue_destroy().

packet_queue_create()

packet_queue_t * packet_queue_create

(

size_t

max_size

)

#include <queue.h>

Create a new packet queue.

Parameters

max_size

Maximum queue size (0 = unlimited)

Returns

Pointer to new queue, or NULL on failure

Creates a packet queue without memory pools. Queue will use malloc/free for all node and data allocations.

Note

For high-performance scenarios, use packet_queue_create_with_pools() instead to enable zero-allocation operation.

Definition at line 125 of file queue.c.

                                                     {
  return packet_queue_create_with_pool(max_size, 0); // No pool by default
}

References packet_queue_create_with_pool().

Referenced by add_client().

packet_queue_create_with_pool()

packet_queue_t * packet_queue_create_with_pool	(	size_t	max_size,
		size_t	pool_size
	)

#include <queue.h>

Create a packet queue with node pool.

Parameters

max_size	Maximum queue size (0 = unlimited)
pool_size	Number of nodes to pre-allocate in pool

Returns

Pointer to new queue, or NULL on failure

Creates a packet queue with optional node pool to reduce malloc overhead. Queue will still use malloc/free for packet data allocations.

Note

For zero-allocation operation, use packet_queue_create_with_pools() with use_buffer_pool = true.

Definition at line 129 of file queue.c.

                                                                                 {
  return packet_queue_create_with_pools(max_size, pool_size, false);
}

References packet_queue_create_with_pools().

Referenced by packet_queue_create().

packet_queue_create_with_pools()

packet_queue_t * packet_queue_create_with_pools	(	size_t	max_size,
		size_t	node_pool_size,
		bool	use_buffer_pool
	)

#include <queue.h>

Create a packet queue with both node and buffer pools.

Parameters

max_size	Maximum queue size (0 = unlimited)
node_pool_size	Number of nodes to pre-allocate
use_buffer_pool	If true, use global buffer pool for packet data

Returns

Pointer to new queue, or NULL on failure

Creates a high-performance packet queue with both node and buffer pools enabled. This enables zero-allocation operation when pools are configured.

Note

When use_buffer_pool is true, queue uses global buffer pool singleton. Buffer pool must be initialized before queue creation.

Definition at line 133 of file queue.c.

                                                                                                             {
  packet_queue_t *queue;
  queue = SAFE_MALLOC(sizeof(packet_queue_t), packet_queue_t *);
 
  // Initialize atomic fields
  // For atomic pointer types, use atomic_store with relaxed ordering for initialization
  atomic_ptr_store(&queue->head, (void *)(packet_node_t *)NULL);
  atomic_ptr_store(&queue->tail, (void *)(packet_node_t *)NULL);
  atomic_store_u64(&queue->count, 0);
  queue->max_size = max_size;
  atomic_store_u64(&queue->bytes_queued, 0);
 
  // Create memory pools if requested
  queue->node_pool = node_pool_size > 0 ? node_pool_create(node_pool_size) : NULL;
  queue->buffer_pool = use_buffer_pool ? buffer_pool_create(0, 0) : NULL;
 
  // Initialize atomic statistics
  atomic_store_u64(&queue->packets_enqueued, 0);
  atomic_store_u64(&queue->packets_dequeued, 0);
  atomic_store_u64(&queue->packets_dropped, 0);
  atomic_store_bool(&queue->shutdown, false);
 
  NAMED_REGISTER_PACKET_QUEUE(queue, "packet_queue", NULL);
  if (queue->node_pool) {
    NAMED_REGISTER_NODE_POOL(queue->node_pool, "node_pool", (uintptr_t)(const void *)(queue));
  }
 
  // Register atomic fields for sync state monitoring - descriptive names for lock-free queue tracking
  NAMED_REGISTER_ATOMIC(&queue->count, "current_packet_count", (uintptr_t)(const void *)(queue));
  NAMED_REGISTER_ATOMIC(&queue->bytes_queued, "total_bytes_currently_queued", (uintptr_t)(const void *)(queue));
  NAMED_REGISTER_ATOMIC(&queue->packets_enqueued, "lifetime_packets_enqueued_total", (uintptr_t)(const void *)(queue));
  NAMED_REGISTER_ATOMIC(&queue->packets_dequeued, "lifetime_packets_dequeued_total", (uintptr_t)(const void *)(queue));
  NAMED_REGISTER_ATOMIC(&queue->packets_dropped, "lifetime_packets_dropped_overflow", (uintptr_t)(const void *)(queue));
  NAMED_REGISTER_ATOMIC(&queue->shutdown, "is_shutdown_requested", (uintptr_t)(const void *)(queue));
  NAMED_REGISTER_ATOMIC_PTR(&queue->head, "head_node_pointer", (uintptr_t)(const void *)(queue));
  NAMED_REGISTER_ATOMIC_PTR(&queue->tail, "tail_node_pointer", (uintptr_t)(const void *)(queue));
 
  return queue;
}

References atomic_ptr_store(), atomic_store_bool(), atomic_store_u64(), packet_queue_t::buffer_pool, buffer_pool_create(), packet_queue_t::bytes_queued, packet_queue_t::count, packet_queue_t::head, packet_queue_t::max_size, NAMED_REGISTER_ATOMIC, NAMED_REGISTER_ATOMIC_PTR, NAMED_REGISTER_NODE_POOL, NAMED_REGISTER_PACKET_QUEUE, packet_queue_t::node_pool, node_pool_create(), packet_queue_t::packets_dequeued, packet_queue_t::packets_dropped, packet_queue_t::packets_enqueued, SAFE_MALLOC, packet_queue_t::shutdown, and packet_queue_t::tail.

Referenced by add_webrtc_client(), and packet_queue_create_with_pool().

packet_queue_dequeue()

queued_packet_t * packet_queue_dequeue

(

packet_queue_t *

queue

)

#include <queue.h>

Dequeue a packet from the queue (non-blocking)

Parameters

queue

Packet queue

Returns

Pointer to dequeued packet, or NULL if queue is empty or shutdown

Removes and returns the packet at the head of the queue. Returns NULL immediately if queue is empty or shutdown (non-blocking operation).

Note

Caller must free the returned packet with packet_queue_free_packet() to properly release memory (either to pool or free()).

Returns NULL when queue is empty or shutdown (allows consumer threads to exit). Caller should retry periodically if queue is expected to have data.

Thread-safe (lock-free using atomic operations).

Warning

Always free dequeued packets with packet_queue_free_packet().

Definition at line 479 of file queue.c.

                                                             {
  // Non-blocking dequeue (same as try_dequeue for lock-free design)
  return packet_queue_try_dequeue(queue);
}

References packet_queue_try_dequeue().

packet_queue_destroy()

void packet_queue_destroy

(

packet_queue_t *

queue

)

#include <queue.h>

Signal queue shutdown (causes dequeue to return NULL)

Parameters

queue

Packet queue

Sets the shutdown flag, causing all dequeue operations to return NULL. This allows consumer threads to exit gracefully without blocking forever.

Note

After shutdown, enqueue operations still succeed, but dequeue returns NULL.

Thread-safe (wakes all waiting dequeue threads).

Definition at line 173 of file queue.c.

                                                 {
  if (!queue)
    return;
 
  // Unregister atomic fields
  NAMED_UNREGISTER(&queue->count);
  NAMED_UNREGISTER(&queue->bytes_queued);
  NAMED_UNREGISTER(&queue->packets_enqueued);
  NAMED_UNREGISTER(&queue->packets_dequeued);
  NAMED_UNREGISTER(&queue->packets_dropped);
  NAMED_UNREGISTER(&queue->shutdown);
  NAMED_UNREGISTER(&queue->head);
  NAMED_UNREGISTER(&queue->tail);
 
  NAMED_UNREGISTER(queue);
 
  // Signal shutdown first
  packet_queue_stop(queue);
 
  // Clear any remaining packets
  packet_queue_clear(queue);
 
  // Destroy memory pools if present
  if (queue->node_pool) {
    node_pool_destroy(queue->node_pool);
  }
  if (queue->buffer_pool) {
    buffer_pool_log_stats(queue->buffer_pool, "packet_queue");
    buffer_pool_destroy(queue->buffer_pool);
  }
 
  // No mutex/cond to destroy (lock-free design)
 
  SAFE_FREE(queue);
}

References packet_queue_t::buffer_pool, buffer_pool_destroy(), buffer_pool_log_stats(), packet_queue_t::bytes_queued, packet_queue_t::count, packet_queue_t::head, NAMED_UNREGISTER, packet_queue_t::node_pool, node_pool_destroy(), packet_queue_clear(), packet_queue_stop(), packet_queue_t::packets_dequeued, packet_queue_t::packets_dropped, packet_queue_t::packets_enqueued, SAFE_FREE, packet_queue_t::shutdown, and packet_queue_t::tail.

Referenced by add_client(), and cleanup_client_packet_queues().

packet_queue_enqueue()

int packet_queue_enqueue	(	packet_queue_t *	queue,
		packet_type_t	type,
		const void *	data,
		size_t	data_len,
		uint32_t	client_id,
		bool	copy_data
	)

#include <queue.h>

Enqueue a packet into the queue.

Parameters

queue	Packet queue
type	Packet type (from packet_types.h)
data	Packet payload data (can be NULL)
data_len	Length of payload data in bytes
client_id	Client ID for packet header
copy_data	If true, copy data into pool/malloc buffer; if false, use data pointer directly

Returns

0 on success, -1 on error

Enqueues a new packet at the tail of the queue. If copy_data is true, packet data is copied into a buffer (from pool if available, otherwise malloc). If copy_data is false, the data pointer is used directly (caller must ensure data remains valid until packet is dequeued and freed).

Note

If queue is full (max_size > 0 and count >= max_size), oldest packet is automatically dropped to make room (packet count does not exceed max_size).

Thread-safe (lock-free using atomic operations).

Warning

When copy_data is false, caller must ensure data pointer remains valid until packet is freed.

Definition at line 209 of file queue.c.

                                                             {
  if (!queue)
    return -1;
 
  // Check if shutdown (atomic read with acquire semantics)
  if (atomic_load_bool(&queue->shutdown)) {
    return -1;
  }
 
  // Check if queue is full and drop oldest packet if needed (lock-free)
  size_t current_count = atomic_load_u64(&queue->count);
  if (queue->max_size > 0 && current_count >= queue->max_size) {
    // Drop oldest packet (head) using atomic compare-and-swap
    packet_node_t *head = (packet_node_t *)atomic_ptr_load(&queue->head);
    if (head) {
      packet_node_t *next = (packet_node_t *)atomic_ptr_load(&head->next);
      // Atomically update head pointer
      if (atomic_ptr_cas(&queue->head, (void **)&head, next)) {
        // Successfully claimed head node
        if (next == NULL) {
          // Queue became empty, also update tail
          atomic_ptr_store(&queue->tail, (void *)(packet_node_t *)NULL);
        }
 
        // Update counters atomically
        size_t bytes = head->packet.data_len;
        atomic_fetch_sub_u64(&queue->bytes_queued, bytes);
        atomic_fetch_sub_u64(&queue->count, 1);
        atomic_fetch_add_u64(&queue->packets_dropped, 1);
 
        // Free dropped packet data
        if (head->packet.owns_data && head->packet.data) {
          buffer_pool_free(head->packet.buffer_pool, head->packet.data, head->packet.data_len);
        }
        node_pool_put(queue->node_pool, head);
 
        log_dev_every(4500 * US_PER_MS_INT, "Dropped packet from queue (full): type=%d, client=%u", type, client_id);
      }
      // If CAS failed, another thread already dequeued - continue to enqueue
    }
  }
 
  // Create new node (use pool if available)
  packet_node_t *node = node_pool_get(queue->node_pool);
  if (!node) {
    SET_ERRNO(ERROR_MEMORY, "Failed to allocate packet node");
    return -1;
  }
 
  // Build packet header
  node->packet.header.magic = HOST_TO_NET_U64(PACKET_MAGIC);
  node->packet.header.type = HOST_TO_NET_U16((uint16_t)type);
  node->packet.header.length = HOST_TO_NET_U32((uint32_t)data_len);
  node->packet.header.client_id = HOST_TO_NET_U32(client_id);
  // Calculate CRC32 for the data (0 for empty packets)
  node->packet.header.crc32 = HOST_TO_NET_U32(data_len > 0 ? asciichat_crc32(data, data_len) : 0);
 
  // Handle data
  if (data_len > 0 && data) {
    if (copy_data) {
      // Try to allocate from buffer pool (local or global)
      if (queue->buffer_pool) {
        node->packet.data = buffer_pool_alloc(queue->buffer_pool, data_len);
        node->packet.buffer_pool = queue->buffer_pool;
      } else {
        // Use global pool if no local pool
        node->packet.data = buffer_pool_alloc(NULL, data_len);
        node->packet.buffer_pool = buffer_pool_get_global();
      }
      SAFE_MEMCPY(node->packet.data, data_len, data, data_len);
      node->packet.owns_data = true;
    } else {
      // Use the data pointer directly (caller must ensure it stays valid)
      node->packet.data = (void *)data;
      node->packet.owns_data = false;
      node->packet.buffer_pool = NULL;
    }
  } else {
    node->packet.data = NULL;
    node->packet.owns_data = false;
    node->packet.buffer_pool = NULL;
  }
 
  node->packet.data_len = data_len;
  atomic_ptr_store(&node->next, (void *)(packet_node_t *)NULL);
 
  // Add to queue using lock-free CAS-based enqueue (Michael-Scott algorithm)
  while (true) {
    packet_node_t *tail = (packet_node_t *)atomic_ptr_load(&queue->tail);
 
    if (tail == NULL) {
      // Empty queue - atomically set both head and tail
      packet_node_t *expected = NULL;
      if (atomic_ptr_cas(&queue->head, (void **)&expected, node)) {
        // Successfully set head (queue was empty)
        atomic_ptr_store(&queue->tail, (void *)node);
        break; // Enqueue successful
      }
      // CAS failed - another thread initialized queue, retry
      continue;
    }
 
    // Queue is non-empty - try to append to tail
    packet_node_t *next = (packet_node_t *)atomic_ptr_load(&tail->next);
    packet_node_t *current_tail = (packet_node_t *)atomic_ptr_load(&queue->tail);
 
    // Verify tail hasn't changed (ABA problem mitigation)
    if (tail != current_tail) {
      // Tail was updated by another thread, retry
      continue;
    }
 
    if (next == NULL) {
      // Tail is actually the last node - try to link new node
      packet_node_t *expected_null = NULL;
      if (atomic_ptr_cas(&tail->next, (void **)&expected_null, node)) {
        // Successfully linked node - try to swing tail forward (best-effort, ignore failure)
        atomic_ptr_cas(&queue->tail, (void **)&tail, node);
        break; // Enqueue successful
      }
      // CAS failed - another thread appended to tail, retry
    } else {
      // Tail is lagging behind - help advance it
      atomic_ptr_cas(&queue->tail, (void **)&tail, next);
      // Retry with new tail
    }
  }
 
  // Update counters atomically
  atomic_fetch_add_u64(&queue->count, (size_t)1);
  atomic_fetch_add_u64(&queue->bytes_queued, data_len);
  atomic_fetch_add_u64(&queue->packets_enqueued, (uint64_t)1);
 
  return 0;
}

References asciichat_crc32, atomic_fetch_add_u64(), atomic_fetch_sub_u64(), atomic_load_bool(), atomic_load_u64(), atomic_ptr_cas(), atomic_ptr_load(), atomic_ptr_store(), queued_packet_t::buffer_pool, packet_queue_t::buffer_pool, buffer_pool_alloc(), buffer_pool_free(), buffer_pool_get_global(), packet_queue_t::bytes_queued, packet_header_t::client_id, packet_queue_t::count, packet_header_t::crc32, queued_packet_t::data, queued_packet_t::data_len, ERROR_MEMORY, packet_queue_t::head, queued_packet_t::header, HOST_TO_NET_U16, HOST_TO_NET_U32, HOST_TO_NET_U64, packet_header_t::length, log_dev_every, packet_header_t::magic, packet_queue_t::max_size, packet_node::next, packet_queue_t::node_pool, node_pool_get(), node_pool_put(), queued_packet_t::owns_data, packet_node::packet, PACKET_MAGIC, packet_queue_t::packets_dropped, packet_queue_t::packets_enqueued, SAFE_MEMCPY, SET_ERRNO, packet_queue_t::shutdown, packet_queue_t::tail, packet_header_t::type, and US_PER_MS_INT.

Referenced by client_audio_render_thread(), client_receive_thread(), and queue_audio_for_client().

packet_queue_enqueue_packet()

int packet_queue_enqueue_packet	(	packet_queue_t *	queue,
		const queued_packet_t *	packet
	)

#include <queue.h>

Enqueue a pre-built packet (for special cases like compressed frames)

Parameters

queue	Packet queue
packet	Pre-built packet structure

Returns

0 on success, -1 on error

Enqueues a packet that has already been built (header, data, etc.). Useful for special cases like compressed frames where packet construction is done outside the normal enqueue path.

Note

Packet is copied into the queue, but data ownership semantics follow the packet's owns_data flag.

Thread-safe (lock-free using atomic operations).

Definition at line 346 of file queue.c.

                                                                                      {
  if (!queue || !packet) {
    SET_ERRNO(ERROR_INVALID_PARAM, "Invalid parameters: queue=%p, packet=%p", queue, packet);
    return -1;
  }
 
  // Validate packet before enqueueing
  if (!packet_queue_validate_packet(packet)) {
    SET_ERRNO(ERROR_INVALID_PARAM, "Refusing to enqueue invalid packet");
    return -1;
  }
 
  // Check if shutdown (atomic read with acquire semantics)
  if (atomic_load_bool(&queue->shutdown)) {
    return -1;
  }
 
  // Check if queue is full and drop oldest packet if needed (lock-free)
  size_t current_count = atomic_load_u64(&queue->count);
  if (queue->max_size > 0 && current_count >= queue->max_size) {
    // Drop oldest packet (head) using atomic compare-and-swap
    packet_node_t *head = (packet_node_t *)atomic_ptr_load(&queue->head);
    if (head) {
      packet_node_t *next = (packet_node_t *)atomic_ptr_load(&head->next);
      // Atomically update head pointer
      if (atomic_ptr_cas(&queue->head, (void **)&head, next)) {
        // Successfully claimed head node
        if (next == NULL) {
          // Queue became empty, also update tail
          atomic_ptr_store(&queue->tail, (void *)(packet_node_t *)NULL);
        }
 
        // Update counters atomically
        size_t bytes = head->packet.data_len;
        atomic_fetch_sub_u64(&queue->bytes_queued, bytes);
        atomic_fetch_sub_u64(&queue->count, 1);
        atomic_fetch_add_u64(&queue->packets_dropped, 1);
 
        // Free dropped packet data
        if (head->packet.owns_data && head->packet.data) {
          buffer_pool_free(head->packet.buffer_pool, head->packet.data, head->packet.data_len);
        }
        node_pool_put(queue->node_pool, head);
      }
      // If CAS failed, another thread already dequeued - continue to enqueue
    }
  }
 
  // Create new node (use pool if available)
  packet_node_t *node = node_pool_get(queue->node_pool);
  if (!node) {
    SET_ERRNO(ERROR_MEMORY, "Failed to allocate packet node");
    return -1;
  }
 
  // Copy the packet header
  SAFE_MEMCPY(&node->packet, sizeof(queued_packet_t), packet, sizeof(queued_packet_t));
 
  // Deep copy the data if needed
  if (packet->data && packet->data_len > 0 && packet->owns_data) {
    // If the packet owns its data, we need to make a copy
    // Try to allocate from buffer pool (local or global)
    void *data_copy;
    if (queue->buffer_pool) {
      data_copy = buffer_pool_alloc(queue->buffer_pool, packet->data_len);
      node->packet.buffer_pool = queue->buffer_pool;
    } else {
      // Use global pool if no local pool
      data_copy = buffer_pool_alloc(NULL, packet->data_len);
      node->packet.buffer_pool = buffer_pool_get_global();
    }
    SAFE_MEMCPY(data_copy, packet->data_len, packet->data, packet->data_len);
    node->packet.data = data_copy;
    node->packet.owns_data = true;
  } else {
    // Either no data or packet doesn't own it (shared reference is OK)
    node->packet.data = packet->data;
    node->packet.owns_data = packet->owns_data;
    node->packet.buffer_pool = packet->buffer_pool; // Preserve original pool reference
  }
 
  atomic_ptr_store(&node->next, (void *)(packet_node_t *)NULL);
 
  // Add to queue using lock-free CAS-based enqueue (Michael-Scott algorithm)
  while (true) {
    packet_node_t *tail = (packet_node_t *)atomic_ptr_load(&queue->tail);
 
    if (tail == NULL) {
      // Empty queue - atomically set both head and tail
      packet_node_t *expected = NULL;
      if (atomic_ptr_cas(&queue->head, (void **)&expected, node)) {
        // Successfully set head (queue was empty)
        atomic_ptr_store(&queue->tail, (void *)node);
        break; // Enqueue successful
      }
      // CAS failed - another thread initialized queue, retry
      continue;
    }
 
    // Queue is non-empty - try to append to tail
    packet_node_t *next = (packet_node_t *)atomic_ptr_load(&tail->next);
    packet_node_t *current_tail = (packet_node_t *)atomic_ptr_load(&queue->tail);
 
    // Verify tail hasn't changed (ABA problem mitigation)
    if (tail != current_tail) {
      // Tail was updated by another thread, retry
      continue;
    }
 
    if (next == NULL) {
      // Tail is actually the last node - try to link new node
      packet_node_t *expected_null = NULL;
      if (atomic_ptr_cas(&tail->next, (void **)&expected_null, node)) {
        // Successfully linked node - try to swing tail forward (best-effort, ignore failure)
        atomic_ptr_cas(&queue->tail, (void **)&tail, node);
        break; // Enqueue successful
      }
      // CAS failed - another thread appended to tail, retry
    } else {
      // Tail is lagging behind - help advance it
      atomic_ptr_cas(&queue->tail, (void **)&tail, next);
      // Retry with new tail
    }
  }
 
  // Update counters atomically
  atomic_fetch_add_u64(&queue->count, (size_t)1);
  atomic_fetch_add_u64(&queue->bytes_queued, packet->data_len);
  atomic_fetch_add_u64(&queue->packets_enqueued, (uint64_t)1);
 
  return 0;
}

References atomic_fetch_add_u64(), atomic_fetch_sub_u64(), atomic_load_bool(), atomic_load_u64(), atomic_ptr_cas(), atomic_ptr_load(), atomic_ptr_store(), queued_packet_t::buffer_pool, packet_queue_t::buffer_pool, buffer_pool_alloc(), buffer_pool_free(), buffer_pool_get_global(), packet_queue_t::bytes_queued, packet_queue_t::count, queued_packet_t::data, queued_packet_t::data_len, ERROR_INVALID_PARAM, ERROR_MEMORY, packet_queue_t::head, packet_queue_t::max_size, packet_node::next, packet_queue_t::node_pool, node_pool_get(), node_pool_put(), queued_packet_t::owns_data, packet_node::packet, packet_queue_validate_packet(), packet_queue_t::packets_dropped, packet_queue_t::packets_enqueued, SAFE_MEMCPY, SET_ERRNO, packet_queue_t::shutdown, and packet_queue_t::tail.

packet_queue_free_packet()

void packet_queue_free_packet

(

queued_packet_t *

packet

)

#include <queue.h>

Free a dequeued packet.

Parameters

packet

Packet to free (can be NULL)

Properly frees packet memory, returning data to buffer pool if applicable, or calling free() if allocated with malloc. Also frees packet structure itself.

Note

Safe to call with NULL (no-op).

Always call this for packets returned by dequeue functions.

Definition at line 568 of file queue.c.

                                                       {
  if (!packet)
    return;
 
  // Check if packet was already freed (detect double-free)
  if (packet->header.magic != HOST_TO_NET_U64(PACKET_MAGIC)) {
    log_warn("Attempted double-free of packet (magic=0x%llx, expected=0x%llx)", NET_TO_HOST_U64(packet->header.magic),
             PACKET_MAGIC);
    return;
  }
 
  if (packet->owns_data && packet->data) {
    // Return to appropriate pool or free
    if (packet->buffer_pool) {
      buffer_pool_free(packet->buffer_pool, packet->data, packet->data_len);
    } else {
      // This was allocated from global pool or malloc, use buffer_pool_free which handles both
      buffer_pool_free(NULL, packet->data, packet->data_len);
    }
  }
 
  // Mark as freed to detect future double-free attempts
  // Use network byte order for consistency on big-endian systems
  packet->header.magic = HOST_TO_NET_U64(0xBEEFDEADULL); // Different magic in network byte order
  SAFE_FREE(packet);
}

References queued_packet_t::buffer_pool, buffer_pool_free(), queued_packet_t::data, queued_packet_t::data_len, queued_packet_t::header, HOST_TO_NET_U64, log_warn, packet_header_t::magic, NET_TO_HOST_U64, queued_packet_t::owns_data, PACKET_MAGIC, and SAFE_FREE.

Referenced by client_dispatch_thread(), client_send_thread_func(), and packet_queue_clear().

packet_queue_get_stats()

void packet_queue_get_stats	(	packet_queue_t *	queue,
		uint64_t *	enqueued,
		uint64_t *	dequeued,
		uint64_t *	dropped
	)

#include <queue.h>

Get queue statistics.

Parameters

queue	Packet queue
enqueued	Output: Total packets enqueued
dequeued	Output: Total packets dequeued
dropped	Output: Total packets dropped (due to queue full)

Retrieves cumulative statistics for the queue. Useful for performance monitoring and debugging.

Note

Thread-safe (lock-free using atomic operations).

All output parameters must be non-NULL.

Definition at line 637 of file queue.c.

                                                                                                              {
  if (!queue)
    return;
 
  // Lock-free atomic reads (acquire semantics for consistency)
  if (enqueued)
    *enqueued = atomic_load_u64(&queue->packets_enqueued);
  if (dequeued)
    *dequeued = atomic_load_u64(&queue->packets_dequeued);
  if (dropped)
    *dropped = atomic_load_u64(&queue->packets_dropped);
}

References atomic_load_u64(), packet_queue_t::packets_dequeued, packet_queue_t::packets_dropped, and packet_queue_t::packets_enqueued.

Referenced by stats_logger_thread().

packet_queue_is_empty()

bool packet_queue_is_empty

(

packet_queue_t *

queue

)

#include <queue.h>

Check if queue is empty.

Parameters

queue

Packet queue

Returns

true if queue is empty, false otherwise

Returns true if queue contains no packets. Snapshot value may change immediately after return due to concurrent operations.

Note

Thread-safe (lock-free using atomic operations).

Definition at line 603 of file queue.c.

                                                  {
  return packet_queue_size(queue) == 0;
}

References packet_queue_size().

packet_queue_is_full()

bool packet_queue_is_full

(

packet_queue_t *

queue

)

#include <queue.h>

Check if queue is full.

Parameters

queue

Packet queue

Returns

true if queue is full (max_size > 0 and count >= max_size), false otherwise

Returns true if queue has reached its maximum size. Unlimited queues (max_size == 0) never return true from this function.

Note

Thread-safe (lock-free using atomic operations).

Definition at line 607 of file queue.c.

                                                 {
  if (!queue || queue->max_size == 0)
    return false;
 
  // Lock-free atomic read
  size_t count = atomic_load_u64(&queue->count);
  return (count >= queue->max_size);
}

References atomic_load_u64(), packet_queue_t::count, and packet_queue_t::max_size.

packet_queue_size()

size_t packet_queue_size

(

packet_queue_t *

queue

)

#include <queue.h>

Get current number of packets in queue.

Parameters

queue

Packet queue

Returns

Number of packets currently queued

Returns the current queue size (count field). This is a snapshot and may change immediately after return due to concurrent enqueue/dequeue operations.

Note

Thread-safe (lock-free using atomic operations).

Definition at line 595 of file queue.c.

                                                {
  if (!queue)
    return 0;
 
  // Lock-free atomic read
  return atomic_load_u64(&queue->count);
}

References atomic_load_u64(), and packet_queue_t::count.

Referenced by client_audio_render_thread(), and packet_queue_is_empty().

packet_queue_stop()

void packet_queue_stop

(

packet_queue_t *

queue

)

#include <queue.h>

Destroy a packet queue and free all resources.

Parameters

queue

Queue to destroy (can be NULL)

Destroys the queue and all associated resources (node pool, etc.).

Warning

Packets still in the queue are NOT freed. Clear the queue first or ensure all packets are dequeued before destruction.

Definition at line 616 of file queue.c.

                                              {
  if (!queue) {
    SET_ERRNO(ERROR_INVALID_PARAM, "Invalid parameters: queue=%p", queue);
    return;
  }
 
  // Lock-free atomic store (release semantics ensures visibility to other threads)
  atomic_store_u64(&queue->shutdown, true);
}

References atomic_store_u64(), ERROR_INVALID_PARAM, SET_ERRNO, and packet_queue_t::shutdown.

Referenced by disconnect_client_for_bad_data(), packet_queue_destroy(), and remove_client().

packet_queue_try_dequeue()

queued_packet_t * packet_queue_try_dequeue

(

packet_queue_t *

queue

)

#include <queue.h>

Try to dequeue a packet without blocking.

Parameters

queue

Packet queue

Returns

Pointer to dequeued packet, or NULL if queue is empty or shutdown

Non-blocking version of packet_queue_dequeue(). Returns immediately with NULL if queue is empty or shutdown, otherwise returns next packet.

Note

Caller must free the returned packet with packet_queue_free_packet().

Thread-safe (lock-free using atomic operations).

Definition at line 484 of file queue.c.

                                                                 {
  if (!queue)
    return NULL;
 
  // Check if shutdown (atomic read with acquire semantics)
  if (atomic_load_bool(&queue->shutdown)) {
    return NULL;
  }
 
  // Check if queue is empty (atomic read with acquire semantics)
  size_t current_count = atomic_load_u64(&queue->count);
  if (current_count == 0) {
    return NULL;
  }
 
  // Remove from head atomically (lock-free dequeue)
  packet_node_t *head = (packet_node_t *)atomic_ptr_load(&queue->head);
  if (!head) {
    return NULL;
  }
 
  // Atomically update head pointer
  packet_node_t *next = (packet_node_t *)atomic_ptr_load(&head->next);
  if (atomic_ptr_cas(&queue->head, (void **)&head, next)) {
    // Successfully claimed head node
    if (next == NULL) {
      // Queue became empty, also update tail atomically
      atomic_ptr_store(&queue->tail, (void *)(packet_node_t *)NULL);
    }
 
    // Update counters atomically
    size_t bytes = head->packet.data_len;
    atomic_fetch_sub_u64(&queue->bytes_queued, bytes);
    atomic_fetch_sub_u64(&queue->count, 1);
    atomic_fetch_add_u64(&queue->packets_dequeued, (uint64_t)1);
 
    // Verify packet magic number for corruption detection
    uint64_t magic = NET_TO_HOST_U64(head->packet.header.magic);
    if (magic != PACKET_MAGIC) {
      SET_ERRNO(ERROR_BUFFER, "CORRUPTION: Invalid magic in try_dequeued packet: 0x%llx (expected 0x%llx), type=%u",
                magic, PACKET_MAGIC, NET_TO_HOST_U16(head->packet.header.type));
      // Still return node to pool but don't return corrupted packet
      node_pool_put(queue->node_pool, head);
      return NULL;
    }
 
    // Validate CRC if there's data
    if (head->packet.data_len > 0 && head->packet.data) {
      uint32_t expected_crc = NET_TO_HOST_U32(head->packet.header.crc32);
      uint32_t actual_crc = asciichat_crc32(head->packet.data, head->packet.data_len);
      if (actual_crc != expected_crc) {
        SET_ERRNO(ERROR_BUFFER,
                  "CORRUPTION: CRC mismatch in try_dequeued packet: got 0x%x, expected 0x%x, type=%u, len=%zu",
                  actual_crc, expected_crc, NET_TO_HOST_U16(head->packet.header.type), head->packet.data_len);
        // Free data if packet owns it
        if (head->packet.owns_data && head->packet.data) {
          // Use buffer_pool_free for global pool allocations, buffer_pool_free for local pools
          if (head->packet.buffer_pool) {
            buffer_pool_free(head->packet.buffer_pool, head->packet.data, head->packet.data_len);
          } else {
            // This was allocated from global pool or malloc, use buffer_pool_free which handles both
            buffer_pool_free(NULL, head->packet.data, head->packet.data_len);
          }
          // Clear pointer to prevent double-free when packet is copied later.
          head->packet.data = NULL;
          head->packet.owns_data = false;
        }
        node_pool_put(queue->node_pool, head);
        return NULL;
      }
    }
 
    // Extract packet and return node to pool
    queued_packet_t *packet;
    packet = SAFE_MALLOC(sizeof(queued_packet_t), queued_packet_t *);
    SAFE_MEMCPY(packet, sizeof(queued_packet_t), &head->packet, sizeof(queued_packet_t));
    node_pool_put(queue->node_pool, head);
    return packet;
  }
 
  // CAS failed - another thread dequeued, retry if needed (or return NULL for non-blocking)
  return NULL;
}

References asciichat_crc32, atomic_fetch_add_u64(), atomic_fetch_sub_u64(), atomic_load_bool(), atomic_load_u64(), atomic_ptr_cas(), atomic_ptr_load(), atomic_ptr_store(), queued_packet_t::buffer_pool, buffer_pool_free(), packet_queue_t::bytes_queued, packet_queue_t::count, packet_header_t::crc32, queued_packet_t::data, queued_packet_t::data_len, ERROR_BUFFER, packet_queue_t::head, queued_packet_t::header, packet_header_t::magic, NET_TO_HOST_U16, NET_TO_HOST_U32, NET_TO_HOST_U64, packet_node::next, packet_queue_t::node_pool, node_pool_put(), queued_packet_t::owns_data, packet_node::packet, PACKET_MAGIC, packet_queue_t::packets_dequeued, SAFE_MALLOC, SAFE_MEMCPY, SET_ERRNO, packet_queue_t::shutdown, packet_queue_t::tail, and packet_header_t::type.

Referenced by client_dispatch_thread(), client_send_thread_func(), packet_queue_clear(), and packet_queue_dequeue().

packet_queue_validate_packet()

bool packet_queue_validate_packet

(

const queued_packet_t *

packet

)

#include <queue.h>

Validate packet integrity.

Parameters

packet

Packet to validate

Returns

true if packet is valid, false otherwise

Validates packet structure and header integrity. Useful for debugging and detecting memory corruption.

Note

This is a basic validation - does not verify payload contents.

Definition at line 650 of file queue.c.

                                                                 {
  if (!packet) {
    return false;
  }
 
  // Check magic number
  uint64_t magic = NET_TO_HOST_U64(packet->header.magic);
  if (magic != PACKET_MAGIC) {
    SET_ERRNO(ERROR_BUFFER, "Invalid packet magic: 0x%llx (expected 0x%llx)", magic, PACKET_MAGIC);
    return false;
  }
 
  // Check packet type is valid (must be non-zero and within reasonable range)
  uint16_t type = NET_TO_HOST_U16(packet->header.type);
  if (type == 0 || type > 10000) {
    SET_ERRNO(ERROR_BUFFER, "Invalid packet type: %u", type);
    return false;
  }
 
  // Check length matches data_len
  uint32_t length = NET_TO_HOST_U32(packet->header.length);
  if (length != packet->data_len) {
    SET_ERRNO(ERROR_BUFFER, "Packet length mismatch: header says %u, data_len is %zu", length, packet->data_len);
    return false;
  }
 
  // Check CRC if there's data
  if (packet->data_len > 0 && packet->data) {
    uint32_t expected_crc = NET_TO_HOST_U32(packet->header.crc32);
    uint32_t actual_crc = asciichat_crc32(packet->data, packet->data_len);
    if (actual_crc != expected_crc) {
      SET_ERRNO(ERROR_BUFFER, "Packet CRC mismatch: got 0x%x, expected 0x%x", actual_crc, expected_crc);
      return false;
    }
  }
 
  return true;
}

References asciichat_crc32, packet_header_t::crc32, queued_packet_t::data, queued_packet_t::data_len, ERROR_BUFFER, queued_packet_t::header, packet_header_t::length, packet_header_t::magic, NET_TO_HOST_U16, NET_TO_HOST_U32, NET_TO_HOST_U64, PACKET_MAGIC, SET_ERRNO, and packet_header_t::type.

Referenced by packet_queue_enqueue_packet().

Variable Documentation

buffer_pool [1/2]

buffer_pool_t* queued_packet_t::buffer_pool

Pool that allocated the data (NULL if malloc'd)

Definition at line 130 of file queue.h.

Referenced by packet_queue_enqueue(), packet_queue_enqueue_packet(), packet_queue_free_packet(), and packet_queue_try_dequeue().

buffer_pool [2/2]

buffer_pool_t* packet_queue_t::buffer_pool

Optional memory pool for data buffers (NULL = use malloc/free)

Definition at line 226 of file queue.h.

Referenced by packet_queue_create_with_pools(), packet_queue_destroy(), packet_queue_enqueue(), and packet_queue_enqueue_packet().

bytes_queued

atomic_t packet_queue_t::bytes_queued

Total bytes of data queued (for monitoring) - atomic for lock-free access.

Definition at line 221 of file queue.h.

Referenced by packet_queue_create_with_pools(), packet_queue_destroy(), packet_queue_enqueue(), packet_queue_enqueue_packet(), and packet_queue_try_dequeue().

count

atomic_t packet_queue_t::count

Number of packets currently in queue - atomic for lock-free access.

Definition at line 217 of file queue.h.

Referenced by packet_queue_create_with_pools(), packet_queue_destroy(), packet_queue_enqueue(), packet_queue_enqueue_packet(), packet_queue_is_full(), packet_queue_size(), and packet_queue_try_dequeue().

data

void* queued_packet_t::data

Packet payload data (can be NULL for header-only packets)

Definition at line 124 of file queue.h.

Referenced by client_dispatch_thread(), client_send_thread_func(), packet_queue_enqueue(), packet_queue_enqueue_packet(), packet_queue_free_packet(), packet_queue_try_dequeue(), and packet_queue_validate_packet().

data_len

size_t queued_packet_t::data_len

Length of payload data in bytes.

Definition at line 126 of file queue.h.

Referenced by client_dispatch_thread(), client_send_thread_func(), packet_queue_enqueue(), packet_queue_enqueue_packet(), packet_queue_free_packet(), packet_queue_try_dequeue(), and packet_queue_validate_packet().

free_list

atomic_ptr_t node_pool::free_list

Stack of free nodes (LIFO for cache locality) - atomic for lock-free access.

Definition at line 169 of file queue.h.

head

atomic_ptr_t packet_queue_t::head

Front of queue (dequeue from here) - atomic for lock-free access.

Definition at line 213 of file queue.h.

Referenced by packet_queue_create_with_pools(), packet_queue_destroy(), packet_queue_enqueue(), packet_queue_enqueue_packet(), and packet_queue_try_dequeue().

header

packet_header_t queued_packet_t::header

Complete packet header (already in network byte order)

Definition at line 122 of file queue.h.

Referenced by packet_queue_enqueue(), packet_queue_free_packet(), packet_queue_try_dequeue(), and packet_queue_validate_packet().

max_size

size_t packet_queue_t::max_size

Maximum queue size (0 = unlimited)

Definition at line 219 of file queue.h.

Referenced by packet_queue_create_with_pools(), packet_queue_enqueue(), packet_queue_enqueue_packet(), and packet_queue_is_full().

next

atomic_ptr_t packet_node::next

Pointer to next node in linked list (NULL for tail) - atomic for lock-free operations.

Definition at line 151 of file queue.h.

Referenced by node_pool_get(), node_pool_put(), packet_queue_enqueue(), packet_queue_enqueue_packet(), and packet_queue_try_dequeue().

node_pool

node_pool_t* packet_queue_t::node_pool

Optional memory pool for nodes (NULL = use malloc/free)

Definition at line 224 of file queue.h.

Referenced by packet_queue_create_with_pools(), packet_queue_destroy(), packet_queue_enqueue(), packet_queue_enqueue_packet(), and packet_queue_try_dequeue().

nodes

packet_node_t* node_pool::nodes

Pre-allocated array of all nodes.

Definition at line 171 of file queue.h.

owns_data

bool queued_packet_t::owns_data

If true, free data when packet is freed.

Definition at line 128 of file queue.h.

Referenced by packet_queue_enqueue(), packet_queue_enqueue_packet(), packet_queue_free_packet(), and packet_queue_try_dequeue().

packet

queued_packet_t packet_node::packet

The queued packet data.

Definition at line 149 of file queue.h.

Referenced by packet_queue_enqueue(), packet_queue_enqueue_packet(), and packet_queue_try_dequeue().

packets_dequeued

atomic_t packet_queue_t::packets_dequeued

Total packets dequeued (statistics) - atomic for lock-free access.

Definition at line 231 of file queue.h.

Referenced by packet_queue_create_with_pools(), packet_queue_destroy(), packet_queue_get_stats(), and packet_queue_try_dequeue().

packets_dropped

atomic_t packet_queue_t::packets_dropped

Total packets dropped due to queue full (statistics) - atomic for lock-free access.

Definition at line 233 of file queue.h.

Referenced by packet_queue_create_with_pools(), packet_queue_destroy(), packet_queue_enqueue(), packet_queue_enqueue_packet(), and packet_queue_get_stats().

packets_enqueued

atomic_t packet_queue_t::packets_enqueued

Total packets enqueued (statistics) - atomic for lock-free access.

Definition at line 229 of file queue.h.

Referenced by packet_queue_create_with_pools(), packet_queue_destroy(), packet_queue_enqueue(), packet_queue_enqueue_packet(), and packet_queue_get_stats().

pool_size

size_t node_pool::pool_size

Total number of nodes in pool.

Definition at line 173 of file queue.h.

shutdown

atomic_t packet_queue_t::shutdown

Shutdown flag (true = dequeue returns NULL) - atomic for lock-free access.

Definition at line 236 of file queue.h.

Referenced by packet_queue_create_with_pools(), packet_queue_destroy(), packet_queue_enqueue(), packet_queue_enqueue_packet(), packet_queue_stop(), and packet_queue_try_dequeue().

tail

atomic_ptr_t packet_queue_t::tail

Back of queue (enqueue here) - atomic for lock-free access.

Definition at line 215 of file queue.h.

Referenced by packet_queue_create_with_pools(), packet_queue_destroy(), packet_queue_enqueue(), packet_queue_enqueue_packet(), and packet_queue_try_dequeue().

used_count

atomic_t node_pool::used_count

Number of nodes currently in use - atomic for lock-free tracking.

Definition at line 175 of file queue.h.