myvtm/libtoolkit/timerqueue.c

#include "precompile.h"
#include "timerqueue.h"
#include "ooputil.h"
#include "memutil.h"
#include "list.h"
#include "array.h"
#include "heaputil.h"
#include "gettimeofday.h"
#include "hash.h"

struct timer_queue_op
{
	int (*cancel)(timer_queue_t *q, timer_entry *entry, int cancel);
	int (*schedule)(timer_queue_t *q, timer_entry *entry, unsigned int delay);
	int (*get_count)(timer_queue_t *q);
	int (*poll_one)(timer_queue_t *q, timer_entry **p_entry, int *next_delay);
	void (*destroy)(timer_queue_t *q);
};

struct timer_queue_s
{
	struct timer_queue_op *op;
};

TOOLKIT_API void timer_queue_destroy(timer_queue_t *q)
{
	q->op->destroy(q);
}

TOOLKIT_API int timer_queue_schedule(timer_queue_t *q, timer_entry *entry, unsigned int delay)
{
	return q->op->schedule(q, entry, delay);
}

TOOLKIT_API int timer_queue_cancel(timer_queue_t *q, timer_entry *entry, int cancel)
{
	return q->op->cancel(q, entry, cancel);
}

TOOLKIT_API int timer_queue_get_count(timer_queue_t *q)
{
	return q->op->get_count(q);
}

TOOLKIT_API int timer_queue_poll(timer_queue_t *q, int *next_delay)
{
	int cnt = 0;
	int t;
	int delay = 0;
	do {
		t = timer_queue_poll_one(q, NULL, &delay);
		cnt += t;
	} while (t && delay == 0);
	if (next_delay)
		*next_delay = delay;
	return cnt;
}

TOOLKIT_API int timer_queue_poll_one(timer_queue_t *q, timer_entry **p_entry, int *next_delay)
{
	return q->op->poll_one(q, p_entry, next_delay);
}

//
// sorted list
//

struct timer_sortedlist_ext
{
	struct list_head entry;
	struct timeval timer_value;
};

struct timer_sortedlist_s
{
	OOP_EXTENDS(timer_queue_s);
	struct list_head node_list;
	int count;
};

static int timer_sortedlist_cancel(timer_queue_t *q, timer_entry *entry, int cancel)
{
	struct timer_sortedlist_s *sortedlist = OOP_DOWNCAST(q, timer_queue_s, timer_sortedlist_s);
	struct timer_sortedlist_ext *ext;
	list_for_each_entry(ext, &sortedlist->node_list, struct timer_sortedlist_ext, entry) {
		timer_entry *te = container_of(ext, timer_entry, _private[0]);
		if (te == entry) {
			sortedlist->count--;
			list_del(&ext->entry);
			if (cancel) {
				te->cb(q, te, -1);
			}
			return 0;
		}
	}
	return -1;
}

static int timer_sortedlist_schedule(timer_queue_t *q, timer_entry *entry, unsigned int delay)
{
	struct timer_sortedlist_s *sortedlist = OOP_DOWNCAST(q, timer_queue_s, timer_sortedlist_s);
	struct timer_sortedlist_ext *ext = (struct timer_sortedlist_ext *)&entry->_private[0];
	struct list_head *ps, *pn;
	assert(entry && entry->cb);
	gettimeofday(&ext->timer_value, NULL);
	timeval_add_msec(&ext->timer_value, delay);
	for (ps = &sortedlist->node_list, pn = ps->next; pn != &sortedlist->node_list; ps = pn, pn = ps->next){
		struct timer_sortedlist_ext *t = slist_entry(pn, struct timer_sortedlist_ext, entry);
		if (timeval_cmp(&ext->timer_value, &t->timer_value) < 0)
			break;
	}
	__list_add(&ext->entry, ps, pn);
	sortedlist->count++;
	return 0;
}

static int timer_sortedlist_get_count(timer_queue_t *q)
{
	struct timer_sortedlist_s *sortedlist = OOP_DOWNCAST(q, timer_queue_s, timer_sortedlist_s);
	return sortedlist->count;
}

static int timer_sortedlist_poll_one(timer_queue_t *q, timer_entry **p_entry, int *next_delay)
{
	struct timer_sortedlist_s *sortedlist = OOP_DOWNCAST(q, timer_queue_s, timer_sortedlist_s);
	struct timer_sortedlist_ext *ext;
	struct timeval now;
	int cnt = 0;
	if (sortedlist->count) {
		ext = list_first_entry(&sortedlist->node_list, struct timer_sortedlist_ext, entry);
		gettimeofday(&now, NULL);
		if (timeval_cmp(&ext->timer_value, &now) <= 0) {
			timer_entry *entry = container_of(ext, timer_entry, _private[0]);
			list_del(&ext->entry);
			sortedlist->count--;
			entry->cb(q, entry, 0);
			if (p_entry)
				*p_entry = entry;
			cnt++;
		}
	}
	if (next_delay) {
		if (sortedlist->count) {
			ext = list_first_entry(&sortedlist->node_list, struct timer_sortedlist_ext, entry);
			*next_delay = (int)timeval_sub(&ext->timer_value, &now);
			if (*next_delay < 0)
				*next_delay = 0;
		} else {
			*next_delay = INT_MAX;
		}
	}
	return cnt;
}

static void timer_sortedlist_destroy(timer_queue_t *q)
{
	struct timer_sortedlist_s *sortedlist = OOP_DOWNCAST(q, timer_queue_s, timer_sortedlist_s);
	assert(list_empty(&sortedlist->node_list));
	free(sortedlist);
}

static struct timer_queue_op _sortedlist_ops =
{
	&timer_sortedlist_cancel,
	&timer_sortedlist_schedule,
	&timer_sortedlist_get_count,
	&timer_sortedlist_poll_one,
	&timer_sortedlist_destroy
};

TOOLKIT_API int timer_sortedlist_create(timer_queue_t **p_q)
{
	struct timer_sortedlist_s *sortedlist;
	struct timer_queue_s *q;

	sortedlist = MALLOC_T(struct timer_sortedlist_s);
	if (!sortedlist)
		return -1;
	q = OOP_UPCAST(sortedlist, timer_queue_s, timer_sortedlist_s);
	INIT_LIST_HEAD(&sortedlist->node_list);
	sortedlist->count = 0;
	q->op = &_sortedlist_ops;
	
	*p_q = q;

	return 0;
}

//
// heap
//

struct timer_heap_ext
{
	int timer_id;
	struct timeval timer_value;
};

struct timer_heap_s
{
	OOP_EXTENDS(timer_queue_s);
	array_header_t *arr_heap;
};

static int timer_entry_cmp(timer_entry *x, timer_entry *y)
{
	struct timer_heap_ext *ex = (struct timer_heap_ext *)&x->_private[0];
	struct timer_heap_ext *ey = (struct timer_heap_ext *)&y->_private[0];
	return timeval_cmp(&ex->timer_value, &ey->timer_value);
}

static void timer_entry_swap(struct timer_heap_s *heap, int i, int j)
{
	timer_entry *ti = ARRAY_IDX(heap->arr_heap, i, timer_entry*);
	struct timer_heap_ext *tix = (struct timer_heap_ext *)&ti->_private[0];
	timer_entry *tj = ARRAY_IDX(heap->arr_heap, j, timer_entry*);
	struct timer_heap_ext *tjx = (struct timer_heap_ext *)&tj->_private[0];
	ARRAY_XCHG(heap->arr_heap, i, j, timer_entry*);
	tix->timer_id = j;
	tjx->timer_id = i;
}

static void timer_heap_up(struct timer_heap_s *heap, int slot)
{
	int i;
	i = slot;
	while (i) {
		int p = heap_parent(i);
		if (timer_entry_cmp(ARRAY_IDX(heap->arr_heap, i, timer_entry*), ARRAY_IDX(heap->arr_heap, p, timer_entry*)) < 0) {
			timer_entry_swap(heap, i, p);
			i = p;
		} else {
			break;
		}
	}
}

static void timer_heap_down(struct timer_heap_s *heap, int slot)
{
	int i = slot;
	int n = heap->arr_heap->nelts;
	while (i < n) {
		int l = heap_left_child(i);
		int r = heap_right_child(i);
		if (r < n) {
			int min = (timer_entry_cmp(ARRAY_IDX(heap->arr_heap, l, timer_entry*), ARRAY_IDX(heap->arr_heap, r, timer_entry*)) < 0) ? l : r;
			if (timer_entry_cmp(ARRAY_IDX(heap->arr_heap, min, timer_entry*), ARRAY_IDX(heap->arr_heap, i, timer_entry*)) < 0) {
				timer_entry_swap(heap, i, min);
				i = min;
			} else {
				break;
			}
		} else if (l < n) { /* r is out of range */
			if (timer_entry_cmp(ARRAY_IDX(heap->arr_heap, l, timer_entry*), ARRAY_IDX(heap->arr_heap, i, timer_entry*)) < 0) {
				timer_entry_swap(heap, i, l);
				i = l;
			}
			break;
		} else {
			break;
		}
	}
}

static timer_entry *pop_entry(struct timer_heap_s *heap)
{
	timer_entry *entry = ARRAY_IDX(heap->arr_heap, 0, timer_entry*);
	struct timer_heap_ext *ext = (struct timer_heap_ext *)&entry->_private[0];
	int last = --heap->arr_heap->nelts;
	if (last) {
		timer_entry_swap(heap, 0, last);
		ext->timer_id = -1;
		timer_heap_down(heap, 0);
	}
	return entry;
}




static int timer_heap_cancel(timer_queue_t *q, timer_entry *entry, int cancel)
{
	struct timer_heap_s *heap = OOP_DOWNCAST(q, timer_queue_s, timer_heap_s);
	struct timer_heap_ext *ext = (struct timer_heap_ext *)&entry->_private[0];
	int err = -1;
	int slot;
	slot = ext->timer_id;
	if (slot < heap->arr_heap->nelts && slot >= 0) {
		timer_entry *te = ARRAY_IDX(heap->arr_heap, slot, timer_entry*);
		if (te == entry) {
			int last = --heap->arr_heap->nelts;
			if (slot != last) {
				timer_entry_swap(heap, slot, last);
				timer_heap_up(heap, slot);
				timer_heap_down(heap, slot);
			}
			ext->timer_id = -1;
			if (cancel)
				entry->cb(q, entry, -1);
			err = 0;
		}
	}
	return err;
}

static int timer_heap_schedule(timer_queue_t *q, timer_entry *entry, unsigned int delay)
{
	struct timer_heap_s *heap = OOP_DOWNCAST(q, timer_queue_s, timer_heap_s);
	struct timer_heap_ext *ext = (struct timer_heap_ext *)&entry->_private[0];
	gettimeofday(&ext->timer_value, NULL);
	timeval_add_msec(&ext->timer_value, delay);
	ext->timer_id = heap->arr_heap->nelts;
	ARRAY_PUSH(heap->arr_heap, timer_entry*) = entry;
	timer_heap_up(heap, ext->timer_id);
	return 0;
}

static int timer_heap_get_count(timer_queue_t *q)
{
	struct timer_heap_s *heap = OOP_DOWNCAST(q, timer_queue_s, timer_heap_s);
	return heap->arr_heap->nelts;
}

static int timer_heap_poll_one(timer_queue_t *q, timer_entry **p_entry, int *next_delay)
{
	struct timer_heap_s *heap = OOP_DOWNCAST(q, timer_queue_s, timer_heap_s);
	struct timeval now;
	int cnt = 0;
	if (heap->arr_heap->nelts) {
		timer_entry *entry = ARRAY_IDX(heap->arr_heap, 0, timer_entry*);
		struct timer_heap_ext *ext = (struct timer_heap_ext *)&entry->_private[0];
		gettimeofday(&now, NULL);
		if (timeval_cmp(&ext->timer_value, &now) <= 0) {
			pop_entry(heap);			
			entry->cb(q, entry, 0);
			if (p_entry)
				*p_entry = entry;
			cnt++;
		}
	}
	if (next_delay) {
		if (heap->arr_heap->nelts) {
			timer_entry *entry = ARRAY_IDX(heap->arr_heap, 0, timer_entry*);
			struct timer_heap_ext *ext = (struct timer_heap_ext *)&entry->_private[0];

			// 解决向前调整系统日期导致返回值超过0x7FFFFFFF返回负数，导致next_delay始终返回为0，引发CPU100%问题
			*next_delay = (int)timeval_sub(&ext->timer_value, &now);
			if (*next_delay < 0)
			{
				if (timeval_cmp(&ext->timer_value, &now) <= 0)
					*next_delay = 0;
				else
					*next_delay = INT_MAX;
			}
		} else {
			*next_delay = INT_MAX;
		}
	}
	return cnt;
}

static void timer_heap_destroy(timer_queue_t *q)
{
	struct timer_heap_s *heap = OOP_DOWNCAST(q, timer_queue_s, timer_heap_s);
//	assert(heap->arr_heap->nelts == 0);
	array_free(heap->arr_heap);
	free(heap);
}

static struct timer_queue_op _heap_ops =
{
	&timer_heap_cancel,
	&timer_heap_schedule,
	&timer_heap_get_count,
	&timer_heap_poll_one,
	&timer_heap_destroy
};

TOOLKIT_API int timer_heap_create(timer_queue_t **p_q)
{
	struct timer_heap_s *heap;
	struct timer_queue_s *q;

	heap = MALLOC_T(struct timer_heap_s);
	if (!heap)
		return -1;
	q = OOP_UPCAST(heap, timer_queue_s, timer_heap_s);
	heap->arr_heap = array_make(0, sizeof(timer_entry*));
	if (!heap->arr_heap) {
		free(heap);
		return -1;
	}

	// 指定定时器实现回调
	q->op = &_heap_ops;
	*p_q = q;
	return 0;
}

//
// wheel
//

struct timer_wheel_ext
{
	struct hlist_node hentry;
	int expire_circle;
	int expire_slot;
};

struct timer_wheel_s
{
	OOP_EXTENDS(timer_queue_s);
	struct hlist_head *buckets;
	int count;
	int num_wheel;
	int wheel_time;
	int cursor_circle;
	int cursor_slot;
};

static int timer_wheel_cancel(timer_queue_t *q, timer_entry *entry, int cancel)
{
	struct timer_wheel_s *wheel = OOP_DOWNCAST(q, timer_queue_s, timer_wheel_s);
	struct timer_wheel_ext *ext = (struct timer_wheel_ext *)&entry->_private[0];
	struct timer_wheel_ext *tpos;
	struct hlist_node *pos;
	hlist_for_each_entry(tpos, pos, &wheel->buckets[ext->expire_slot], struct timer_wheel_ext, hentry) {
		if (ext == tpos) {
			hlist_del(pos);
			if (cancel)
				entry->cb(q, entry, -1);
			return 0;
		}
	}
	return -1;
}

static int timer_wheel_schedule(timer_queue_t *q, timer_entry *entry, unsigned int delay)
{
	struct timer_wheel_s *wheel = OOP_DOWNCAST(q, timer_queue_s, timer_wheel_s);
	struct timer_wheel_ext *ext = (struct timer_wheel_ext *)&entry->_private[0];
	int slot = (delay + wheel->wheel_time -1) / wheel->wheel_time + wheel->cursor_slot;
	ext->expire_circle = slot / wheel->num_wheel;
	slot = slot % wheel->num_wheel;
	ext->expire_slot = slot;
	if (hlist_empty(&wheel->buckets[slot])) {
		hlist_add_head(&ext->hentry, &wheel->buckets[slot]);
	} else {
		struct timer_wheel_ext *tpos, *last;
		struct hlist_node *pos;
		hlist_for_each_entry(tpos, pos, &wheel->buckets[slot], struct timer_wheel_ext, hentry) {
			if (ext->expire_circle < tpos->expire_circle) {
				hlist_add_before(&ext->hentry, pos);
				return 0;
			}
			last = tpos;
		}
		hlist_add_after(&last->hentry, &ext->hentry);
	}
	return 0;	
}

static int timer_wheel_get_count(timer_queue_t *q)
{
	struct timer_wheel_s *wheel = OOP_DOWNCAST(q, timer_queue_s, timer_wheel_s);
	return wheel->count;
}

static int timer_wheel_poll_one(timer_queue_t *q, timer_entry **p_entry, int *next_delay)
{
	struct timer_wheel_s *wheel = OOP_DOWNCAST(q, timer_queue_s, timer_wheel_s);
	int cnt = 0, t;

	do {
		struct timer_wheel_ext *tpos;
		struct hlist_node *pos, *n;
		t = 0;
		hlist_for_each_entry_safe(tpos, pos, n, &wheel->buckets[wheel->cursor_slot], struct timer_wheel_ext, hentry) {
			if (wheel->cursor_circle >= tpos->expire_circle) {
				timer_entry *entry = container_of((void*)tpos, timer_entry, _private[0]);
				hlist_del(pos);
				entry->cb(q, entry, 0);
				if (p_entry)
					*p_entry = entry;
				t++;
			} else {
				break;
			}
		}
		cnt += t;
	} while (t > 0);

	wheel->cursor_slot ++;
	if (wheel->cursor_slot == wheel->num_wheel) {
		wheel->cursor_circle ++;
		wheel->cursor_slot = 0;
	}

	if (next_delay) {
		*next_delay = wheel->wheel_time;
	}

	return cnt;
}

static void timer_wheel_destroy(timer_queue_t *q)
{
	struct timer_wheel_s *wheel = OOP_DOWNCAST(q, timer_queue_s, timer_wheel_s);
	assert(wheel->count == 0);
	free(wheel->buckets);
	free(wheel);
}

static struct timer_queue_op _wheel_ops =
{
	&timer_wheel_cancel,
	&timer_wheel_schedule,
	&timer_wheel_get_count,
	&timer_wheel_poll_one,
	&timer_wheel_destroy
};

TOOLKIT_API int timer_wheel_create(int num_wheel, int wheel_time, timer_queue_t **p_q)
{
	struct timer_wheel_s *wheel;
	struct timer_queue_s *q;

	assert(num_wheel > 0);
	assert(wheel_time > 0);

	wheel = MALLOC_T(struct timer_wheel_s);
	if (!wheel) {
		return -1;
	}
	q = OOP_UPCAST(wheel, timer_queue_s, timer_wheel_s);
	wheel->num_wheel = num_wheel;
	wheel->wheel_time = wheel_time;
	wheel->count = 0;
	wheel->cursor_slot = 0;
	wheel->cursor_circle = 0;
	wheel->buckets = (struct hlist_head*)calloc(num_wheel, sizeof(struct hlist_head));
	if (!wheel->buckets) {
		free(wheel);
		return -1;
	} else {
		int i;
		for (i = 0; i < num_wheel; ++i) {
			INIT_HLIST_HEAD(&wheel->buckets[i]);
		}
	}
	q->op = &_wheel_ops;

	*p_q = q;

	return 0;
}