123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234 |
- #include <errno.h>
- #include <stdlib.h>
- #include <string.h>
- #include <assert.h>
- #include <syslog.h>
- #include "pqueue.h"
- #ifdef DEBUG_PQUEUE
- #define DEBUG_ON 1
- #else
- #define DEBUG_ON 0
- #endif
- #define DEBUG_CMD(_a) if (DEBUG_ON) { _a }
- #define MIN_CAPACITY 128 /* min allocated buffer for a packet */
- static int pqueue_alloc (int seq, unsigned char *packet, int packlen, pqueue_t **new);
- int packet_timeout_usecs = DEFAULT_PACKET_TIMEOUT * 1000000;
- static pqueue_t *pq_head = NULL, *pq_tail = NULL;
- /* contains a list of free queue elements.*/
- static pqueue_t *pq_freelist_head = NULL;
- static int pqueue_alloc(int seq, unsigned char *packet, int packlen, pqueue_t **new) {
- pqueue_t *newent;
- DEBUG_CMD(syslog(LOG_DEBUG, "seq=%d, packlen=%d", seq, packlen););
- /* search the freelist for one that has sufficient space */
- if (pq_freelist_head) {
- for (newent = pq_freelist_head; newent; newent = newent->next) {
- if (newent->capacity >= packlen) {
- /* unlink from freelist */
- if (pq_freelist_head == newent)
- pq_freelist_head = newent->next;
- if (newent->prev)
- newent->prev->next = newent->next;
- if (newent->next)
- newent->next->prev = newent->prev;
- if (pq_freelist_head)
- pq_freelist_head->prev = NULL;
- break;
- } /* end if capacity >= packlen */
- } /* end for */
-
- /* nothing found? Take first and reallocate it */
- if (NULL == newent) {
- newent = pq_freelist_head;
- pq_freelist_head = pq_freelist_head->next;
- if (pq_freelist_head)
- pq_freelist_head->prev = NULL;
- DEBUG_CMD(syslog(LOG_DEBUG, "realloc capacity %d to %d",newent->capacity, packlen););
- newent->packet = (unsigned char *)realloc(newent->packet, packlen);
- if (!newent->packet) {
- syslog(LOG_WARNING, "error reallocating packet: %s", strerror(errno));
- return -1;
- }
- newent->capacity = packlen;
- }
-
- DEBUG_CMD(syslog(LOG_DEBUG, "Recycle entry from freelist. Capacity: %d", newent->capacity););
- } else {
- /* allocate a new one */
- newent = (pqueue_t *)malloc( sizeof(pqueue_t) );
- if (!newent) {
- syslog(LOG_WARNING, "error allocating newent: %s", strerror(errno));
- return -1;
- }
- newent->capacity = 0;
- DEBUG_CMD(syslog(LOG_DEBUG, "Alloc new queue entry"););
- }
- if ( ! newent->capacity ) {
- /* a new queue entry was allocated. Allocate the packet buffer */
- int size = packlen < MIN_CAPACITY ? MIN_CAPACITY : packlen;
- /* Allocate at least MIN_CAPACITY */
- DEBUG_CMD(syslog(LOG_DEBUG, "allocating for packet size %d", size););
- newent->packet = (unsigned char *)malloc(size);
- if (!newent->packet) {
- syslog(LOG_WARNING, "error allocating packet: %s", strerror(errno));
- return -1;
- }
- newent->capacity = size;
- } /* endif ! capacity */
- assert( newent->capacity >= packlen );
- /* store the contents into the buffer */
- memcpy(newent->packet, packet, packlen);
- newent->next = newent->prev = NULL;
- newent->seq = seq;
- newent->packlen = packlen;
- gettimeofday(&newent->expires, NULL);
- newent->expires.tv_usec += packet_timeout_usecs;
- newent->expires.tv_sec += (newent->expires.tv_usec / 1000000);
- newent->expires.tv_usec %= 1000000;
- *new = newent;
- return 0;
- }
- int pqueue_add (int seq, unsigned char *packet, int packlen) {
- pqueue_t *newent, *point;
- /* get a new entry */
- if ( 0 != pqueue_alloc(seq, packet, packlen, &newent) ) {
- return -1;
- }
- for (point = pq_head; point != NULL; point = point->next) {
- if (point->seq == seq) {
- // queue already contains this packet
- syslog(LOG_WARNING, "discarding duplicate packet %d", seq);
- pqueue_del(newent);
- return -1;
- }
- if (point->seq > seq) {
- // gone too far: point->seq > seq and point->prev->seq < seq
- if (point->prev) {
- // insert between point->prev and point
- DEBUG_CMD(syslog(LOG_DEBUG, "adding %d between %d and %d",
- seq, point->prev->seq, point->seq););
- point->prev->next = newent;
- } else {
- // insert at head of queue, before point
- DEBUG_CMD(syslog(LOG_DEBUG, "adding %d before %d", seq, point->seq););
- pq_head = newent;
- }
- newent->prev = point->prev; // will be NULL, at head of queue
- newent->next = point;
- point->prev = newent;
- return 0;
- }
- }
- /* We didn't find anywhere to insert the packet,
- * so there are no packets in the queue with higher sequences than this one,
- * so all the packets in the queue have lower sequences,
- * so this packet belongs at the end of the queue (which might be empty)
- */
-
- if (pq_head == NULL) {
- DEBUG_CMD(syslog(LOG_DEBUG, "adding %d to empty queue", seq););
- pq_head = newent;
- } else {
- DEBUG_CMD(syslog(LOG_DEBUG, "adding %d as tail, after %d", seq, pq_tail->seq););
- pq_tail->next = newent;
- }
- newent->prev = pq_tail;
- pq_tail = newent;
- return 0;
- }
- int pqueue_del (pqueue_t *point) {
- DEBUG_CMD(syslog(LOG_DEBUG, "Move seq %d to freelist", point->seq););
- /* unlink from pq */
- if (pq_head == point) pq_head = point->next;
- if (pq_tail == point) pq_tail = point->prev;
- if (point->prev) point->prev->next = point->next;
- if (point->next) point->next->prev = point->prev;
- /* add point to the freelist */
- point->next = pq_freelist_head;
- point->prev = NULL;
- if (point->next)
- point->next->prev = point;
- pq_freelist_head = point;
- DEBUG_CMD(
- int pq_count = 0;
- int pq_freelist_count = 0;
- pqueue_t *point;
- for ( point = pq_head; point ; point = point->next) {
- ++pq_count;
- }
- for ( point = pq_freelist_head; point ; point = point->next) {
- ++pq_freelist_count;
- }
- syslog(LOG_DEBUG, "queue length is %d, freelist length is %d", pq_count, pq_freelist_count);
- );
- return 0;
- }
- pqueue_t *pqueue_head () {
- return pq_head;
- }
- int pqueue_expiry_time (pqueue_t *entry) {
- struct timeval tv;
- int expiry_time;
- gettimeofday(&tv, NULL);
- expiry_time = (entry->expires.tv_sec - tv.tv_sec) * 1000000;
- expiry_time += (entry->expires.tv_usec - tv.tv_usec);
- return expiry_time;
- }
|