From a6f902baed7abb17a1a9c014e01bb100077f8198 Mon Sep 17 00:00:00 2001 From: Rene Mayrhofer Date: Sun, 1 Mar 2009 10:48:08 +0000 Subject: - Updated to new upstream revision. --- src/charon/processing/scheduler.c | 201 ++++++++++++++++++++++++-------------- 1 file changed, 130 insertions(+), 71 deletions(-) (limited to 'src/charon/processing/scheduler.c') diff --git a/src/charon/processing/scheduler.c b/src/charon/processing/scheduler.c index 42aa2579e..593a51f0b 100644 --- a/src/charon/processing/scheduler.c +++ b/src/charon/processing/scheduler.c @@ -1,4 +1,5 @@ /* + * Copyright (C) 2008 Tobias Brunner * Copyright (C) 2005-2006 Martin Willi * Copyright (C) 2005 Jan Hutter * Hochschule fuer Technik Rapperswil @@ -13,7 +14,7 @@ * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License * for more details. * - * $Id: scheduler.c 3589 2008-03-13 14:14:44Z martin $ + * $Id: scheduler.c 4799 2008-12-12 09:16:31Z martin $ */ #include @@ -25,6 +26,10 @@ #include #include #include +#include + +/* the initial size of the heap */ +#define HEAP_SIZE_DEFAULT 64 typedef struct event_t event_t; @@ -64,26 +69,34 @@ struct private_scheduler_t { scheduler_t public; /** - * Job wich schedules + * Job which queues scheduled jobs to the processor. */ callback_job_t *job; + + /** + * The heap in which the events are stored. + */ + event_t **heap; + + /** + * The size of the heap. + */ + u_int heap_size; /** - * The jobs are scheduled in a list. + * The number of scheduled events. */ - linked_list_t *list; + u_int event_count; /** * Exclusive access to list */ - pthread_mutex_t mutex; + mutex_t *mutex; /** * Condvar to wait for next job. */ - pthread_cond_t condvar; - - bool cancelled; + condvar_t *condvar; }; /** @@ -99,50 +112,103 @@ static long time_difference(timeval_t *end, timeval_t *start) return (s * 1000 + us/1000); } +/** + * Returns the top event without removing it. Returns NULL if the heap is empty. + */ +static event_t *peek_event(private_scheduler_t *this) +{ + return this->event_count > 0 ? this->heap[1] : NULL; +} + +/** + * Removes the top event from the heap and returns it. Returns NULL if the heap + * is empty. + */ +static event_t *remove_event(private_scheduler_t *this) +{ + event_t *event, *top; + if (!this->event_count) + { + return NULL; + } + + /* store the value to return */ + event = this->heap[1]; + /* move the bottom event to the top */ + top = this->heap[1] = this->heap[this->event_count]; + + if (--this->event_count > 1) + { + /* seep down the top event */ + u_int position = 1; + while ((position << 1) <= this->event_count) + { + u_int child = position << 1; + + if ((child + 1) <= this->event_count && + time_difference(&this->heap[child + 1]->time, + &this->heap[child]->time) < 0) + { + /* the "right" child is smaller */ + child++; + } + + if (time_difference(&top->time, &this->heap[child]->time) <= 0) + { + /* the top event fires before the smaller of the two children, stop */ + break; + } + + /* exchange with the smaller child */ + this->heap[position] = this->heap[child]; + position = child; + } + this->heap[position] = top; + } + return event; +} + /** * Get events from the queue and pass it to the processor */ static job_requeue_t schedule(private_scheduler_t * this) { - timespec_t timeout; timeval_t now; event_t *event; long difference; int oldstate; bool timed = FALSE; - DBG2(DBG_JOB, "waiting for next event..."); - pthread_mutex_lock(&this->mutex); + this->mutex->lock(this->mutex); gettimeofday(&now, NULL); - if (this->list->get_count(this->list) > 0) + if ((event = peek_event(this)) != NULL) { - this->list->get_first(this->list, (void **)&event); difference = time_difference(&now, &event->time); - if (difference > 0) + if (difference >= 0) { - DBG2(DBG_JOB, "got event, queueing job for execution"); - this->list->remove_first(this->list, (void **)&event); - pthread_mutex_unlock(&this->mutex); + remove_event(this); + this->mutex->unlock(this->mutex); + DBG2(DBG_JOB, "got event, queuing job for execution"); charon->processor->queue_job(charon->processor, event->job); free(event); return JOB_REQUEUE_DIRECT; } - timeout.tv_sec = event->time.tv_sec; - timeout.tv_nsec = event->time.tv_usec * 1000; + DBG2(DBG_JOB, "next event in %ldms, waiting", -difference); timed = TRUE; } - pthread_cleanup_push((void*)pthread_mutex_unlock, &this->mutex); + pthread_cleanup_push((void*)this->mutex->unlock, this->mutex); pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &oldstate); if (timed) { - pthread_cond_timedwait(&this->condvar, &this->mutex, &timeout); + this->condvar->timed_wait_abs(this->condvar, this->mutex, event->time); } else { - pthread_cond_wait(&this->condvar, &this->mutex); + DBG2(DBG_JOB, "no events, waiting"); + this->condvar->wait(this->condvar, this->mutex); } pthread_setcancelstate(oldstate, NULL); pthread_cleanup_pop(TRUE); @@ -155,9 +221,9 @@ static job_requeue_t schedule(private_scheduler_t * this) static u_int get_job_load(private_scheduler_t *this) { int count; - pthread_mutex_lock(&this->mutex); - count = this->list->get_count(this->list); - pthread_mutex_unlock(&this->mutex); + this->mutex->lock(this->mutex); + count = this->event_count; + this->mutex->unlock(this->mutex); return count; } @@ -167,8 +233,8 @@ static u_int get_job_load(private_scheduler_t *this) static void schedule_job(private_scheduler_t *this, job_t *job, u_int32_t time) { timeval_t now; - event_t *event, *current; - iterator_t *iterator; + event_t *event; + u_int position; time_t s; suseconds_t us; @@ -182,46 +248,30 @@ static void schedule_job(private_scheduler_t *this, job_t *job, u_int32_t time) event->time.tv_usec = (now.tv_usec + us) % 1000000; event->time.tv_sec = now.tv_sec + (now.tv_usec + us)/1000000 + s; - pthread_mutex_lock(&this->mutex); - while(TRUE) + this->mutex->lock(this->mutex); + + this->event_count++; + if (this->event_count > this->heap_size) { - if (this->list->get_count(this->list) == 0) - { - this->list->insert_first(this->list,event); - break; - } - - this->list->get_last(this->list, (void**)¤t); - if (time_difference(&event->time, ¤t->time) >= 0) - { /* new event has to be fired after the last event in list */ - this->list->insert_last(this->list, event); - break; - } - - this->list->get_first(this->list, (void**)¤t); - if (time_difference(&event->time, ¤t->time) < 0) - { /* new event has to be fired before the first event in list */ - this->list->insert_first(this->list, event); - break; - } - - iterator = this->list->create_iterator(this->list, TRUE); - /* first element has not to be checked (already done) */ - iterator->iterate(iterator, (void**)¤t); - while(iterator->iterate(iterator, (void**)¤t)) - { - if (time_difference(&event->time, ¤t->time) <= 0) - { - /* new event has to be fired before the current event in list */ - iterator->insert_before(iterator, event); - break; - } - } - iterator->destroy(iterator); - break; + /* double the size of the heap */ + this->heap_size <<= 1; + this->heap = (event_t**)realloc(this->heap, (this->heap_size + 1) * sizeof(event_t*)); } - pthread_cond_signal(&this->condvar); - pthread_mutex_unlock(&this->mutex); + /* "put" the event to the bottom */ + position = this->event_count; + + /* then bubble it up */ + while (position > 1 && time_difference(&this->heap[position >> 1]->time, + &event->time) > 0) + { + /* parent has to be fired after the new event, move up */ + this->heap[position] = this->heap[position >> 1]; + position >>= 1; + } + this->heap[position] = event; + + this->condvar->signal(this->condvar); + this->mutex->unlock(this->mutex); } /** @@ -229,9 +279,15 @@ static void schedule_job(private_scheduler_t *this, job_t *job, u_int32_t time) */ static void destroy(private_scheduler_t *this) { - this->cancelled = TRUE; + event_t *event; this->job->cancel(this->job); - this->list->destroy_function(this->list, (void*)event_destroy); + this->condvar->destroy(this->condvar); + this->mutex->destroy(this->mutex); + while ((event = remove_event(this)) != NULL) + { + event_destroy(event); + } + free(this->heap); free(this); } @@ -246,10 +302,13 @@ scheduler_t * scheduler_create() this->public.schedule_job = (void (*) (scheduler_t *this, job_t *job, u_int32_t ms)) schedule_job; this->public.destroy = (void(*)(scheduler_t*)) destroy; - this->list = linked_list_create(); - this->cancelled = FALSE; - pthread_mutex_init(&this->mutex, NULL); - pthread_cond_init(&this->condvar, NULL); + /* Note: the root of the heap is at index 1 */ + this->event_count = 0; + this->heap_size = HEAP_SIZE_DEFAULT; + this->heap = (event_t**)calloc(this->heap_size + 1, sizeof(event_t*)); + + this->mutex = mutex_create(MUTEX_DEFAULT); + this->condvar = condvar_create(CONDVAR_DEFAULT); this->job = callback_job_create((callback_job_cb_t)schedule, this, NULL, NULL); charon->processor->queue_job(charon->processor, (job_t*)this->job); -- cgit v1.2.3