lib: use DECLARE_SKIPLIST for timers instead of pqueue

Replaces the use of pqueue_* for the thread_master's timer list with an
instance of DECLARE_SKIPLIST_*.

Signed-off-by: David Lamparter <equinox@diac24.net>
This commit is contained in:
David Lamparter 2019-01-31 02:30:35 +01:00 committed by David Lamparter
parent c284542b66
commit 7c198e4e1a
2 changed files with 35 additions and 66 deletions

View File

@ -27,7 +27,6 @@
#include "memory.h"
#include "log.h"
#include "hash.h"
#include "pqueue.h"
#include "command.h"
#include "sigevent.h"
#include "network.h"
@ -42,6 +41,18 @@ DEFINE_MTYPE_STATIC(LIB, THREAD_STATS, "Thread stats")
DECLARE_LIST(thread_list, struct thread, threaditem)
static int thread_timer_cmp(const struct thread *a, const struct thread *b)
{
if (timercmp(&a->u.sands, &b->u.sands, <))
return -1;
if (timercmp(&a->u.sands, &b->u.sands, >))
return 1;
return 0;
}
DECLARE_SKIPLIST_NONUNIQ(thread_timer_list, struct thread, timeritem,
thread_timer_cmp)
#if defined(__APPLE__)
#include <mach/mach.h>
#include <mach/mach_time.h>
@ -377,25 +388,6 @@ void thread_cmd_init(void)
/* CLI end ------------------------------------------------------------------ */
static int thread_timer_cmp(void *a, void *b)
{
struct thread *thread_a = a;
struct thread *thread_b = b;
if (timercmp(&thread_a->u.sands, &thread_b->u.sands, <))
return -1;
if (timercmp(&thread_a->u.sands, &thread_b->u.sands, >))
return 1;
return 0;
}
static void thread_timer_update(void *node, int actual_position)
{
struct thread *thread = node;
thread->index = actual_position;
}
static void cancelreq_del(void *cr)
{
XFREE(MTYPE_TMP, cr);
@ -440,11 +432,7 @@ struct thread_master *thread_master_create(const char *name)
thread_list_init(&rv->event);
thread_list_init(&rv->ready);
thread_list_init(&rv->unuse);
/* Initialize the timer queues */
rv->timer = pqueue_create();
rv->timer->cmp = thread_timer_cmp;
rv->timer->update = thread_timer_update;
thread_timer_list_init(&rv->timer);
/* Initialize thread_fetch() settings */
rv->spin = true;
@ -542,16 +530,6 @@ static void thread_array_free(struct thread_master *m,
XFREE(MTYPE_THREAD_POLL, thread_array);
}
static void thread_queue_free(struct thread_master *m, struct pqueue *queue)
{
int i;
for (i = 0; i < queue->size; i++)
thread_free(m, queue->array[i]);
pqueue_delete(queue);
}
/*
* thread_master_free_unused
*
@ -574,6 +552,8 @@ void thread_master_free_unused(struct thread_master *m)
/* Stop thread scheduler. */
void thread_master_free(struct thread_master *m)
{
struct thread *t;
pthread_mutex_lock(&masters_mtx);
{
listnode_delete(masters, m);
@ -585,7 +565,8 @@ void thread_master_free(struct thread_master *m)
thread_array_free(m, m->read);
thread_array_free(m, m->write);
thread_queue_free(m, m->timer);
while ((t = thread_timer_list_pop(&m->timer)))
thread_free(m, t);
thread_list_free(m, &m->event);
thread_list_free(m, &m->ready);
thread_list_free(m, &m->unuse);
@ -659,7 +640,6 @@ static struct thread *thread_get(struct thread_master *m, uint8_t type,
thread->add_type = type;
thread->master = m;
thread->arg = arg;
thread->index = -1;
thread->yield = THREAD_YIELD_TIME_SLOT; /* default */
thread->ref = NULL;
@ -818,7 +798,6 @@ funcname_thread_add_timer_timeval(struct thread_master *m,
struct thread **t_ptr, debugargdef)
{
struct thread *thread;
struct pqueue *queue;
assert(m != NULL);
@ -834,7 +813,6 @@ funcname_thread_add_timer_timeval(struct thread_master *m,
return NULL;
}
queue = m->timer;
thread = thread_get(m, type, func, arg, debugargpass);
pthread_mutex_lock(&thread->mtx);
@ -842,7 +820,7 @@ funcname_thread_add_timer_timeval(struct thread_master *m,
monotime(&thread->u.sands);
timeradd(&thread->u.sands, time_relative,
&thread->u.sands);
pqueue_enqueue(thread, queue);
thread_timer_list_add(&m->timer, thread);
if (t_ptr) {
*t_ptr = thread;
thread->ref = t_ptr;
@ -1019,7 +997,6 @@ static void thread_cancel_rw(struct thread_master *master, int fd, short state)
static void do_thread_cancel(struct thread_master *master)
{
struct thread_list_head *list = NULL;
struct pqueue *queue = NULL;
struct thread **thread_array = NULL;
struct thread *thread;
@ -1075,7 +1052,7 @@ static void do_thread_cancel(struct thread_master *master)
thread_array = master->write;
break;
case THREAD_TIMER:
queue = master->timer;
thread_timer_list_del(&master->timer, thread);
break;
case THREAD_EVENT:
list = &master->event;
@ -1088,16 +1065,10 @@ static void do_thread_cancel(struct thread_master *master)
break;
}
if (queue) {
assert(thread->index >= 0);
assert(thread == queue->array[thread->index]);
pqueue_remove_at(thread->index, queue);
} else if (list) {
if (list) {
thread_list_del(list, thread);
} else if (thread_array) {
thread_array[thread->u.fd] = NULL;
} else {
assert(!"Thread should be either in queue or list or array!");
}
if (thread->ref)
@ -1215,15 +1186,15 @@ void thread_cancel_async(struct thread_master *master, struct thread **thread,
}
/* ------------------------------------------------------------------------- */
static struct timeval *thread_timer_wait(struct pqueue *queue,
static struct timeval *thread_timer_wait(struct thread_timer_list_head *timers,
struct timeval *timer_val)
{
if (queue->size) {
struct thread *next_timer = queue->array[0];
monotime_until(&next_timer->u.sands, timer_val);
return timer_val;
}
return NULL;
if (!thread_timer_list_count(timers))
return NULL;
struct thread *next_timer = thread_timer_list_first(timers);
monotime_until(&next_timer->u.sands, timer_val);
return timer_val;
}
static struct thread *thread_run(struct thread_master *m, struct thread *thread,
@ -1315,17 +1286,16 @@ static void thread_process_io(struct thread_master *m, unsigned int num)
}
/* Add all timers that have popped to the ready list. */
static unsigned int thread_process_timers(struct pqueue *queue,
static unsigned int thread_process_timers(struct thread_timer_list_head *timers,
struct timeval *timenow)
{
struct thread *thread;
unsigned int ready = 0;
while (queue->size) {
thread = queue->array[0];
while ((thread = thread_timer_list_first(timers))) {
if (timercmp(timenow, &thread->u.sands, <))
return ready;
pqueue_dequeue(queue);
thread_timer_list_pop(timers);
thread->type = THREAD_READY;
thread_list_add_tail(&thread->master->ready, thread);
ready++;
@ -1407,7 +1377,7 @@ struct thread *thread_fetch(struct thread_master *m, struct thread *fetch)
* once per loop to avoid starvation by events
*/
if (!thread_list_count(&m->ready))
tw = thread_timer_wait(m->timer, &tv);
tw = thread_timer_wait(&m->timer, &tv);
if (thread_list_count(&m->ready) ||
(tw && !timercmp(tw, &zerotime, >)))
@ -1452,7 +1422,7 @@ struct thread *thread_fetch(struct thread_master *m, struct thread *fetch)
/* Post timers to ready queue. */
monotime(&now);
thread_process_timers(m->timer, &now);
thread_process_timers(&m->timer, &now);
/* Post I/O to ready queue. */
if (num > 0)

View File

@ -41,8 +41,7 @@ struct rusage_t {
#define GETRUSAGE(X) thread_getrusage(X)
PREDECL_LIST(thread_list)
struct pqueue;
PREDECL_SKIPLIST_NONUNIQ(thread_timer_list)
struct fd_handler {
/* number of pfd that fit in the allocated space of pfds. This is a
@ -73,7 +72,7 @@ struct thread_master {
struct thread **read;
struct thread **write;
struct pqueue *timer;
struct thread_timer_list_head timer;
struct thread_list_head event, ready, unuse;
struct list *cancel_req;
bool canceled;
@ -95,6 +94,7 @@ struct thread {
uint8_t type; /* thread type */
uint8_t add_type; /* thread type */
struct thread_list_item threaditem;
struct thread_timer_list_item timeritem;
struct thread **ref; /* external reference (if given) */
struct thread_master *master; /* pointer to the struct thread_master */
int (*func)(struct thread *); /* event function */
@ -104,7 +104,6 @@ struct thread {
int fd; /* file descriptor in case of r/w */
struct timeval sands; /* rest of time sands value. */
} u;
int index; /* queue position for timers */
struct timeval real;
struct cpu_thread_history *hist; /* cache pointer to cpu_history */
unsigned long yield; /* yield time in microseconds */