packet_queue.c File Reference

📬 Lock-free packet queue with per-client isolation and memory pooling More...

Go to the source code of this file.

Functions
node_pool_t *	node_pool_create (size_t pool_size)
void	node_pool_destroy (node_pool_t *pool)
packet_node_t *	node_pool_get (node_pool_t *pool)
void	node_pool_put (node_pool_t *pool, packet_node_t *node)
packet_queue_t *	packet_queue_create (size_t max_size)
packet_queue_t *	packet_queue_create_with_pool (size_t max_size, size_t pool_size)
packet_queue_t *	packet_queue_create_with_pools (size_t max_size, size_t node_pool_size, bool use_buffer_pool)
void	packet_queue_destroy (packet_queue_t *queue)
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)
int	packet_queue_enqueue_packet (packet_queue_t *queue, const queued_packet_t *packet)
queued_packet_t *	packet_queue_dequeue (packet_queue_t *queue)
queued_packet_t *	packet_queue_try_dequeue (packet_queue_t *queue)
void	packet_queue_free_packet (queued_packet_t *packet)
size_t	packet_queue_size (packet_queue_t *queue)
bool	packet_queue_is_empty (packet_queue_t *queue)
bool	packet_queue_is_full (packet_queue_t *queue)
void	packet_queue_stop (packet_queue_t *queue)
void	packet_queue_clear (packet_queue_t *queue)
void	packet_queue_get_stats (packet_queue_t *queue, uint64_t *enqueued, uint64_t *dequeued, uint64_t *dropped)
bool	packet_queue_validate_packet (const queued_packet_t *packet)

Detailed Description

📬 Lock-free packet queue with per-client isolation and memory pooling

Definition in file packet_queue.c.

Function Documentation

node_pool_create()

node_pool_t * node_pool_create

(

size_t

pool_size

)

Definition at line 25 of file packet_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_store_explicit(&pool->nodes[i].next, &pool->nodes[i + 1], memory_order_relaxed);
  }
  atomic_store_explicit(&pool->nodes[pool_size - 1].next, (packet_node_t *)NULL, memory_order_relaxed);
 
  pool->free_list = &pool->nodes[0];
  pool->pool_size = pool_size;
  pool->used_count = 0;
 
  if (mutex_init(&pool->pool_mutex) != 0) {
    SET_ERRNO(ERROR_PLATFORM_INIT, "Failed to initialize mutex for node pool");
    node_pool_destroy(pool);
    return NULL;
  }
 
  return pool;
}

References mutex_init(), and node_pool_destroy().

Referenced by packet_queue_create_with_pools().

node_pool_destroy()

void node_pool_destroy

(

node_pool_t *

pool

)

Definition at line 55 of file packet_queue.c.

                                          {
  if (!pool) {
    return;
  }
 
  mutex_destroy(&pool->pool_mutex);
  SAFE_FREE(pool->nodes);
  SAFE_FREE(pool);
}

References mutex_destroy().

Referenced by node_pool_create(), and packet_queue_destroy().

node_pool_get()

packet_node_t * node_pool_get

(

node_pool_t *

pool

)

Definition at line 65 of file packet_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;
  }
 
  mutex_lock(&pool->pool_mutex);
 
  packet_node_t *node = pool->free_list;
  if (node) {
    pool->free_list = atomic_load_explicit(&node->next, memory_order_relaxed);
    pool->used_count++;
    atomic_store_explicit(&node->next, (packet_node_t *)NULL, memory_order_relaxed); // Clear next pointer
  }
 
  mutex_unlock(&pool->pool_mutex);
 
  if (!node) {
    // Pool exhausted, fallback to malloc
    node = SAFE_MALLOC(sizeof(packet_node_t), packet_node_t *);
    // Remove extra text from format string that had no matching argument
    log_debug("Memory pool exhausted, falling back to SAFE_MALLOC (used: %zu/%zu)", pool->used_count, pool->pool_size);
  }
 
  return node;
}

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
	)

Definition at line 94 of file packet_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) {
    mutex_lock(&pool->pool_mutex);
 
    // Return to free list
    atomic_store_explicit(&node->next, pool->free_list, memory_order_relaxed);
    pool->free_list = node;
    pool->used_count--;
 
    mutex_unlock(&pool->pool_mutex);
  } else {
    // This was malloc'd, so free it
    SAFE_FREE(node);
  }
}

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

)

Definition at line 606 of file packet_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

)

Definition at line 128 of file packet_queue.c.

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

References packet_queue_create_with_pool().

packet_queue_create_with_pool()

packet_queue_t * packet_queue_create_with_pool	(	size_t	max_size,
		size_t	pool_size
	)

Definition at line 132 of file packet_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
	)

Definition at line 136 of file packet_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_store_explicit(&queue->head, (packet_node_t *)NULL, memory_order_relaxed);
  atomic_store_explicit(&queue->tail, (packet_node_t *)NULL, memory_order_relaxed);
  atomic_init(&queue->count, (size_t)0);
  queue->max_size = max_size;
  atomic_init(&queue->bytes_queued, (size_t)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_init(&queue->packets_enqueued, (uint64_t)0);
  atomic_init(&queue->packets_dequeued, (uint64_t)0);
  atomic_init(&queue->packets_dropped, (uint64_t)0);
  atomic_init(&queue->shutdown, false);
 
  return queue;
}

References buffer_pool_create(), and node_pool_create().

Referenced by add_client(), add_webrtc_client(), and packet_queue_create_with_pool().

packet_queue_dequeue()

queued_packet_t * packet_queue_dequeue

(

packet_queue_t *

queue

)

Definition at line 459 of file packet_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

)

Definition at line 161 of file packet_queue.c.

                                                 {
  if (!queue)
    return;
 
  // 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 buffer_pool_destroy(), buffer_pool_log_stats(), node_pool_destroy(), packet_queue_clear(), and packet_queue_stop().

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
	)

Definition at line 185 of file packet_queue.c.

                                                             {
  if (!queue)
    return -1;
 
  // Check if shutdown (atomic read with acquire semantics)
  if (atomic_load_explicit(&queue->shutdown, memory_order_acquire)) {
    return -1;
  }
 
  // Check if queue is full and drop oldest packet if needed (lock-free)
  size_t current_count = atomic_load_explicit(&queue->count, memory_order_acquire);
  if (queue->max_size > 0 && current_count >= queue->max_size) {
    // Drop oldest packet (head) using atomic compare-and-swap
    packet_node_t *head = atomic_load_explicit(&queue->head, memory_order_acquire);
    if (head) {
      packet_node_t *next = atomic_load_explicit(&head->next, memory_order_acquire);
      // Atomically update head pointer
      if (atomic_compare_exchange_weak(&queue->head, &head, next)) {
        // Successfully claimed head node
        if (next == NULL) {
          // Queue became empty, also update tail
          atomic_store_explicit(&queue->tail, (packet_node_t *)NULL, memory_order_release);
        }
 
        // Update counters atomically
        size_t bytes = head->packet.data_len;
        atomic_fetch_sub(&queue->bytes_queued, bytes);
        atomic_fetch_sub(&queue->count, (size_t)1);
        atomic_fetch_add(&queue->packets_dropped, (uint64_t)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_store_explicit(&node->next, (packet_node_t *)NULL, memory_order_relaxed);
 
  // Add to queue using lock-free CAS-based enqueue (Michael-Scott algorithm)
  while (true) {
    packet_node_t *tail = atomic_load_explicit(&queue->tail, memory_order_acquire);
 
    if (tail == NULL) {
      // Empty queue - atomically set both head and tail
      packet_node_t *expected = NULL;
      if (atomic_compare_exchange_weak_explicit(&queue->head, &expected, node, memory_order_release,
                                                memory_order_acquire)) {
        // Successfully set head (queue was empty)
        atomic_store_explicit(&queue->tail, node, memory_order_release);
        break; // Enqueue successful
      }
      // CAS failed - another thread initialized queue, retry
      continue;
    }
 
    // Queue is non-empty - try to append to tail
    packet_node_t *next = atomic_load_explicit(&tail->next, memory_order_acquire);
    packet_node_t *current_tail = atomic_load_explicit(&queue->tail, memory_order_acquire);
 
    // 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_compare_exchange_weak_explicit(&tail->next, &expected_null, node, memory_order_release,
                                                memory_order_acquire)) {
        // Successfully linked node - try to swing tail forward (best-effort, ignore failure)
        atomic_compare_exchange_weak_explicit(&queue->tail, &tail, node, memory_order_release, memory_order_relaxed);
        break; // Enqueue successful
      }
      // CAS failed - another thread appended to tail, retry
    } else {
      // Tail is lagging behind - help advance it
      atomic_compare_exchange_weak_explicit(&queue->tail, &tail, next, memory_order_release, memory_order_relaxed);
      // Retry with new tail
    }
  }
 
  // Update counters atomically
  atomic_fetch_add(&queue->count, (size_t)1);
  atomic_fetch_add(&queue->bytes_queued, data_len);
  atomic_fetch_add(&queue->packets_enqueued, (uint64_t)1);
 
  return 0;
}

References buffer_pool_alloc(), buffer_pool_free(), buffer_pool_get_global(), node_pool_get(), and node_pool_put().

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
	)

Definition at line 324 of file packet_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_explicit(&queue->shutdown, memory_order_acquire)) {
    return -1;
  }
 
  // Check if queue is full and drop oldest packet if needed (lock-free)
  size_t current_count = atomic_load_explicit(&queue->count, memory_order_acquire);
  if (queue->max_size > 0 && current_count >= queue->max_size) {
    // Drop oldest packet (head) using atomic compare-and-swap
    packet_node_t *head = atomic_load_explicit(&queue->head, memory_order_acquire);
    if (head) {
      packet_node_t *next = atomic_load_explicit(&head->next, memory_order_acquire);
      // Atomically update head pointer
      if (atomic_compare_exchange_weak(&queue->head, &head, next)) {
        // Successfully claimed head node
        if (next == NULL) {
          // Queue became empty, also update tail
          atomic_store_explicit(&queue->tail, (packet_node_t *)NULL, memory_order_release);
        }
 
        // Update counters atomically
        size_t bytes = head->packet.data_len;
        atomic_fetch_sub(&queue->bytes_queued, bytes);
        atomic_fetch_sub(&queue->count, (size_t)1);
        atomic_fetch_add(&queue->packets_dropped, (uint64_t)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_store_explicit(&node->next, (packet_node_t *)NULL, memory_order_relaxed);
 
  // Add to queue using lock-free CAS-based enqueue (Michael-Scott algorithm)
  while (true) {
    packet_node_t *tail = atomic_load_explicit(&queue->tail, memory_order_acquire);
 
    if (tail == NULL) {
      // Empty queue - atomically set both head and tail
      packet_node_t *expected = NULL;
      if (atomic_compare_exchange_weak_explicit(&queue->head, &expected, node, memory_order_release,
                                                memory_order_acquire)) {
        // Successfully set head (queue was empty)
        atomic_store_explicit(&queue->tail, node, memory_order_release);
        break; // Enqueue successful
      }
      // CAS failed - another thread initialized queue, retry
      continue;
    }
 
    // Queue is non-empty - try to append to tail
    packet_node_t *next = atomic_load_explicit(&tail->next, memory_order_acquire);
    packet_node_t *current_tail = atomic_load_explicit(&queue->tail, memory_order_acquire);
 
    // 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_compare_exchange_weak_explicit(&tail->next, &expected_null, node, memory_order_release,
                                                memory_order_acquire)) {
        // Successfully linked node - try to swing tail forward (best-effort, ignore failure)
        atomic_compare_exchange_weak_explicit(&queue->tail, &tail, node, memory_order_release, memory_order_relaxed);
        break; // Enqueue successful
      }
      // CAS failed - another thread appended to tail, retry
    } else {
      // Tail is lagging behind - help advance it
      atomic_compare_exchange_weak_explicit(&queue->tail, &tail, next, memory_order_release, memory_order_relaxed);
      // Retry with new tail
    }
  }
 
  // Update counters atomically
  atomic_fetch_add(&queue->count, (size_t)1);
  atomic_fetch_add(&queue->bytes_queued, packet->data_len);
  atomic_fetch_add(&queue->packets_enqueued, (uint64_t)1);
 
  return 0;
}

References buffer_pool_alloc(), buffer_pool_free(), buffer_pool_get_global(), node_pool_get(), node_pool_put(), and packet_queue_validate_packet().

packet_queue_free_packet()

void packet_queue_free_packet

(

queued_packet_t *

packet

)

Definition at line 548 of file packet_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 buffer_pool_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
	)

Definition at line 617 of file packet_queue.c.

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

Referenced by stats_logger_thread().

packet_queue_is_empty()

bool packet_queue_is_empty

(

packet_queue_t *

queue

)

Definition at line 583 of file packet_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

)

Definition at line 587 of file packet_queue.c.

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

packet_queue_size()

size_t packet_queue_size

(

packet_queue_t *

queue

)

Definition at line 575 of file packet_queue.c.

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

Referenced by client_audio_render_thread(), and packet_queue_is_empty().

packet_queue_stop()

void packet_queue_stop

(

packet_queue_t *

queue

)

Definition at line 596 of file packet_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_explicit(&queue->shutdown, true, memory_order_release);
}

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

)

Definition at line 464 of file packet_queue.c.

                                                                 {
  if (!queue)
    return NULL;
 
  // Check if shutdown (atomic read with acquire semantics)
  if (atomic_load_explicit(&queue->shutdown, memory_order_acquire)) {
    return NULL;
  }
 
  // Check if queue is empty (atomic read with acquire semantics)
  size_t current_count = atomic_load_explicit(&queue->count, memory_order_acquire);
  if (current_count == 0) {
    return NULL;
  }
 
  // Remove from head atomically (lock-free dequeue)
  packet_node_t *head = atomic_load_explicit(&queue->head, memory_order_acquire);
  if (!head) {
    return NULL;
  }
 
  // Atomically update head pointer
  packet_node_t *next = atomic_load_explicit(&head->next, memory_order_acquire);
  if (atomic_compare_exchange_weak(&queue->head, &head, next)) {
    // Successfully claimed head node
    if (next == NULL) {
      // Queue became empty, also update tail atomically
      atomic_store_explicit(&queue->tail, (packet_node_t *)NULL, memory_order_release);
    }
 
    // Update counters atomically
    size_t bytes = head->packet.data_len;
    atomic_fetch_sub(&queue->bytes_queued, bytes);
    atomic_fetch_sub(&queue->count, (size_t)1);
    atomic_fetch_add(&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 buffer_pool_free(), and node_pool_put().

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

)

Definition at line 630 of file packet_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;
}

Referenced by packet_queue_enqueue_packet().