queue.c File Reference

Go to the source code of this file.

Data Structures
struct	packet_node
struct	packet_queue
	For packet node, this is the live record that ties packet, next together. The owning code creates it, updates it while work is active, and releases its handles and pointed-to storage before freeing the record. More...

Typedefs
typedef struct packet_node	packet_node_t

Functions
packet_queue_t *	packet_queue_create_config (const packet_queue_config_t *config)
packet_queue_t *	packet_queue_create (size_t max_packets)
void	packet_queue_free_packet (queued_packet_t *packet)
asciichat_error_t	packet_queue_push_take (packet_queue_t *queue, packet_type_t type, void **data, size_t length, uint32_t client_id)
asciichat_error_t	packet_queue_push_shared (packet_queue_t *queue, packet_type_t type, const void *data, size_t length, uint32_t client_id)
asciichat_error_t	packet_queue_push_copy (packet_queue_t *queue, packet_type_t type, const void *data, size_t length, uint32_t client_id)
queued_packet_t *	packet_queue_try_dequeue (packet_queue_t *queue)
queued_packet_t *	packet_queue_dequeue (packet_queue_t *queue)
void	packet_queue_clear (packet_queue_t *queue)
void	packet_queue_stop (packet_queue_t *queue)
void	packet_queue_destroy (packet_queue_t *queue)
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_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)

Typedef Documentation

packet_node_t

typedef struct packet_node packet_node_t

Function Documentation

packet_queue_clear()

void packet_queue_clear

(

packet_queue_t *

queue

)

Definition at line 195 of file queue.c.

                                               {
  if (!queue)
    return;
  mutex_lock(&queue->mutex);
  packet_node_t *node;
  while ((node = pop_locked(queue)))
    packet_queue_free_packet(&node->packet);
  mutex_unlock(&queue->mutex);
}

References packet_queue::mutex, packet_node::packet, and packet_queue_free_packet().

Referenced by packet_queue_destroy().

packet_queue_create()

packet_queue_t * packet_queue_create

(

size_t

max_packets

)

Defaults: 16 MiB byte limit and rejection on overflow. Zero packet count disables only the item limit, never the byte limit or memory budget.

Definition at line 59 of file queue.c.

                                                        {
  packet_queue_config_t config = {max_packets, 16 * 1024 * 1024, PACKET_QUEUE_REJECT};
  return packet_queue_create_config(&config);
}

References packet_queue_create_config(), and PACKET_QUEUE_REJECT.

packet_queue_create_config()

packet_queue_t * packet_queue_create_config

(

const packet_queue_config_t *

config

)

Definition at line 37 of file queue.c.

                                                                                {
  if (!config || !config->max_bytes ||
      (config->overflow != PACKET_QUEUE_REJECT && config->overflow != PACKET_QUEUE_DROP_OLDEST)) {
    SET_ERRNO(ERROR_INVALID_PARAM, "Queue needs a byte limit and a valid overflow policy");
    return NULL;
  }
  SAFE_CALLOC_DEFER_FREE(queue, 1, sizeof(packet_queue_t), packet_queue_t *);
  queue->config = *config;
  if (mutex_init(&queue->mutex, "packet_queue") != 0) {
    return NULL;
  }
  defer_named(queue_mutex, mutex_destroy(&queue->mutex));
  buffer_domain_config_t memory = {64 * 1024 * 1024, 8 * 1024 * 1024, 5000000000ULL, false};
  if (buffer_domain_create(&memory, &queue->domain) != ASCIICHAT_OK) {
    return NULL;
  }
  defer_cancel(queue_mutex);
  packet_queue_t *result = queue;
  queue = NULL;
  return result;
}

References ASCIICHAT_OK, buffer_domain_create(), ERROR_INVALID_PARAM, packet_queue_config_t::max_bytes, packet_queue_config_t::overflow, PACKET_QUEUE_DROP_OLDEST, PACKET_QUEUE_REJECT, SAFE_CALLOC_DEFER_FREE, and SET_ERRNO.

Referenced by packet_queue_create().

packet_queue_dequeue()

queued_packet_t * packet_queue_dequeue

(

packet_queue_t *

queue

)

Definition at line 191 of file queue.c.

                                                             {
  return packet_queue_try_dequeue(queue);
}

References packet_queue_try_dequeue().

packet_queue_destroy()

void packet_queue_destroy

(

packet_queue_t *

queue

)

Join queue users before destruction. Dequeued packets may outlive the queue.

Definition at line 213 of file queue.c.

                                                 {
  if (!queue)
    return;
  packet_queue_stop(queue);
  packet_queue_clear(queue);
  buffer_domain_close(&queue->domain);
  mutex_destroy(&queue->mutex);
  SAFE_FREE(queue);
}

References buffer_domain_close(), packet_queue::domain, packet_queue::mutex, packet_queue_clear(), packet_queue_stop(), and SAFE_FREE.

packet_queue_free_packet()

void packet_queue_free_packet

(

queued_packet_t *

packet

)

Definition at line 78 of file queue.c.

                                                       {
  if (!packet)
    return;
  buffer_release(packet->data);
  /* packet is the first member of the owning node allocation. */
  buffer_release(packet);
}

References buffer_release, and queued_packet_t::data.

Referenced by packet_queue_clear(), and packet_queue_try_dequeue().

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 246 of file queue.c.

                                                                                                              {
  if (!queue)
    return;
  mutex_lock(&queue->mutex);
  if (enqueued)
    *enqueued = queue->enqueued;
  if (dequeued)
    *dequeued = queue->dequeued;
  if (dropped)
    *dropped = queue->dropped;
  mutex_unlock(&queue->mutex);
}

References packet_queue::dequeued, packet_queue::dropped, packet_queue::enqueued, and packet_queue::mutex.

packet_queue_is_empty()

bool packet_queue_is_empty

(

packet_queue_t *

queue

)

Definition at line 232 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

)

Definition at line 236 of file queue.c.

                                                 {
  if (!queue)
    return false;
  mutex_lock(&queue->mutex);
  bool full = (queue->config.max_packets && queue->count >= queue->config.max_packets) ||
              queue->bytes >= queue->config.max_bytes;
  mutex_unlock(&queue->mutex);
  return full;
}

References packet_queue::bytes, packet_queue::config, packet_queue::count, packet_queue_config_t::max_bytes, packet_queue_config_t::max_packets, and packet_queue::mutex.

packet_queue_push_copy()

asciichat_error_t packet_queue_push_copy	(	packet_queue_t *	queue,
		packet_type_t	type,
		const void *	data,
		size_t	length,
		uint32_t	client_id )

Copy a borrowed view into queue-owned storage.

Definition at line 160 of file queue.c.

                                                             {
  if (!queue || (length && !data) || length > queue->config.max_bytes)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid packet copy");
  void *copy = NULL;
  if (length) {
    copy = buffer_alloc(queue->domain, length);
    if (!copy)
      return SET_ERRNO(ERROR_MEMORY, "Cannot copy packet within memory budget");
    memcpy(copy, data, length);
  }
  asciichat_error_t result = packet_queue_push_take(queue, type, &copy, length, client_id);
  buffer_release(copy);
  return result;
}

References buffer_alloc(), buffer_release, packet_queue::config, packet_queue::domain, ERROR_INVALID_PARAM, ERROR_MEMORY, packet_queue_config_t::max_bytes, packet_queue_push_take(), and SET_ERRNO.

packet_queue_push_shared()

asciichat_error_t packet_queue_push_shared	(	packet_queue_t *	queue,
		packet_type_t	type,
		const void *	data,
		size_t	length,
		uint32_t	client_id )

Queue a shared immutable buffer, retaining an independent reference.

Definition at line 147 of file queue.c.

                                                               {
  if (length > buffer_capacity(data))
    return SET_ERRNO(ERROR_INVALID_PARAM, "Shared packet exceeds buffer capacity");
  const void *shared = buffer_share(data);
  if (data && !shared)
    return SET_ERRNO(ERROR_INVALID_STATE, "Cannot share packet buffer");
  asciichat_error_t result = push(queue, type, shared, length, client_id);
  if (result != ASCIICHAT_OK)
    buffer_release(shared);
  return result;
}

References ASCIICHAT_OK, buffer_capacity(), buffer_release, buffer_share(), ERROR_INVALID_PARAM, ERROR_INVALID_STATE, and SET_ERRNO.

packet_queue_push_take()

asciichat_error_t packet_queue_push_take	(	packet_queue_t *	queue,
		packet_type_t	type,
		void **	data,
		size_t	length,
		uint32_t	client_id )

On success, transfers the owning buffer pointer and clears *data. On any error, the caller retains ownership and *data is unchanged.

Definition at line 137 of file queue.c.

                                                             {
  if (!data || (length && !*data) || length > buffer_capacity(*data))
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid owned packet buffer");
  asciichat_error_t result = push(queue, type, *data, length, client_id);
  if (result == ASCIICHAT_OK)
    *data = NULL;
  return result;
}

References ASCIICHAT_OK, buffer_capacity(), ERROR_INVALID_PARAM, and SET_ERRNO.

Referenced by packet_queue_push_copy().

packet_queue_size()

size_t packet_queue_size

(

packet_queue_t *

queue

)

Definition at line 223 of file queue.c.

                                                {
  if (!queue)
    return 0;
  mutex_lock(&queue->mutex);
  size_t count = queue->count;
  mutex_unlock(&queue->mutex);
  return count;
}

References packet_queue::count, and packet_queue::mutex.

Referenced by packet_queue_is_empty().

packet_queue_stop()

void packet_queue_stop

(

packet_queue_t *

queue

)

Definition at line 205 of file queue.c.

                                              {
  if (!queue)
    return;
  mutex_lock(&queue->mutex);
  queue->stopped = true;
  mutex_unlock(&queue->mutex);
}

References packet_queue::mutex, and packet_queue::stopped.

Referenced by packet_queue_destroy().

packet_queue_try_dequeue()

queued_packet_t * packet_queue_try_dequeue

(

packet_queue_t *

queue

)

Definition at line 176 of file queue.c.

                                                                 {
  if (!queue)
    return NULL;
  mutex_lock(&queue->mutex);
  packet_node_t *node = queue->stopped ? NULL : pop_locked(queue);
  if (node)
    queue->dequeued++;
  mutex_unlock(&queue->mutex);
  if (node && !packet_queue_validate_packet(&node->packet)) {
    packet_queue_free_packet(&node->packet);
    return NULL;
  }
  return node ? &node->packet : NULL;
}

References packet_queue::dequeued, packet_queue::mutex, packet_node::packet, packet_queue_free_packet(), packet_queue_validate_packet(), and packet_queue::stopped.

Referenced by packet_queue_dequeue().

packet_queue_validate_packet()

bool packet_queue_validate_packet

(

const queued_packet_t *

packet

)

Definition at line 259 of file queue.c.

                                                                 {
  return packet && NET_TO_HOST_U64(packet->header.magic) == PACKET_MAGIC && NET_TO_HOST_U16(packet->header.type) > 0 &&
         NET_TO_HOST_U16(packet->header.type) <= 10000 && NET_TO_HOST_U32(packet->header.length) == packet->data_len &&
         (!packet->data_len ||
          (packet->data && NET_TO_HOST_U32(packet->header.crc32) == asciichat_crc32(packet->data, packet->data_len)));
}

References asciichat_crc32, packet_header_t::crc32, queued_packet_t::data, queued_packet_t::data_len, 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, and packet_header_t::type.

Referenced by packet_queue_try_dequeue().