queue.h File Reference

Go to the source code of this file.

Data Structures
struct	queued_packet_t
struct	packet_queue_config_t

Typedefs
typedef struct packet_queue	packet_queue_t

Enumerations
enum	packet_queue_overflow_t { PACKET_QUEUE_REJECT , PACKET_QUEUE_DROP_OLDEST }

Functions
packet_queue_t *	packet_queue_create (size_t max_packets)
packet_queue_t *	packet_queue_create_config (const packet_queue_config_t *config)
void	packet_queue_destroy (packet_queue_t *queue)
void	packet_queue_stop (packet_queue_t *queue)
void	packet_queue_clear (packet_queue_t *queue)
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)
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)
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_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_queue_t

typedef struct packet_queue packet_queue_t

Definition at line 12 of file queue.h.

Enumeration Type Documentation

packet_queue_overflow_t

enum packet_queue_overflow_t

Enumerator
PACKET_QUEUE_REJECT
PACKET_QUEUE_DROP_OLDEST

Definition at line 13 of file queue.h.

{ PACKET_QUEUE_REJECT, PACKET_QUEUE_DROP_OLDEST } packet_queue_overflow_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().