replacing the pthread_mutex in scheduler_t with the wrapped implementation.

added a method to condvar_t which allows to wait for an absolute timeout.
This commit is contained in:
Tobias Brunner
2008-11-25 19:30:02 +00:00
parent ed6146ffbe
commit 6df2731e78
3 changed files with 62 additions and 38 deletions
+18 -22
View File
@@ -25,6 +25,7 @@
#include <daemon.h> #include <daemon.h>
#include <processing/processor.h> #include <processing/processor.h>
#include <processing/jobs/callback_job.h> #include <processing/jobs/callback_job.h>
#include <utils/mutex.h>
typedef struct event_t event_t; typedef struct event_t event_t;
@@ -76,14 +77,12 @@ struct private_scheduler_t {
/** /**
* Exclusive access to list * Exclusive access to list
*/ */
pthread_mutex_t mutex; mutex_t *mutex;
/** /**
* Condvar to wait for next job. * Condvar to wait for next job.
*/ */
pthread_cond_t condvar; condvar_t *condvar;
bool cancelled;
}; };
/** /**
@@ -104,7 +103,6 @@ static long time_difference(timeval_t *end, timeval_t *start)
*/ */
static job_requeue_t schedule(private_scheduler_t * this) static job_requeue_t schedule(private_scheduler_t * this)
{ {
timespec_t timeout;
timeval_t now; timeval_t now;
event_t *event; event_t *event;
long difference; long difference;
@@ -112,7 +110,7 @@ static job_requeue_t schedule(private_scheduler_t * this)
bool timed = FALSE; bool timed = FALSE;
DBG2(DBG_JOB, "waiting for next event..."); DBG2(DBG_JOB, "waiting for next event...");
pthread_mutex_lock(&this->mutex); this->mutex->lock(this->mutex);
gettimeofday(&now, NULL); gettimeofday(&now, NULL);
@@ -122,27 +120,25 @@ static job_requeue_t schedule(private_scheduler_t * this)
difference = time_difference(&now, &event->time); 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); this->list->remove_first(this->list, (void **)&event);
pthread_mutex_unlock(&this->mutex); this->mutex->unlock(this->mutex);
DBG2(DBG_JOB, "got event, queueing job for execution");
charon->processor->queue_job(charon->processor, event->job); charon->processor->queue_job(charon->processor, event->job);
free(event); free(event);
return JOB_REQUEUE_DIRECT; return JOB_REQUEUE_DIRECT;
} }
timeout.tv_sec = event->time.tv_sec;
timeout.tv_nsec = event->time.tv_usec * 1000;
timed = TRUE; 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); pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &oldstate);
if (timed) if (timed)
{ {
pthread_cond_timedwait(&this->condvar, &this->mutex, &timeout); this->condvar->timed_wait_abs(this->condvar, this->mutex, event->time);
} }
else else
{ {
pthread_cond_wait(&this->condvar, &this->mutex); this->condvar->wait(this->condvar, this->mutex);
} }
pthread_setcancelstate(oldstate, NULL); pthread_setcancelstate(oldstate, NULL);
pthread_cleanup_pop(TRUE); pthread_cleanup_pop(TRUE);
@@ -155,9 +151,9 @@ static job_requeue_t schedule(private_scheduler_t * this)
static u_int get_job_load(private_scheduler_t *this) static u_int get_job_load(private_scheduler_t *this)
{ {
int count; int count;
pthread_mutex_lock(&this->mutex); this->mutex->lock(this->mutex);
count = this->list->get_count(this->list); count = this->list->get_count(this->list);
pthread_mutex_unlock(&this->mutex); this->mutex->unlock(this->mutex);
return count; return count;
} }
@@ -182,7 +178,7 @@ 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_usec = (now.tv_usec + us) % 1000000;
event->time.tv_sec = now.tv_sec + (now.tv_usec + us)/1000000 + s; event->time.tv_sec = now.tv_sec + (now.tv_usec + us)/1000000 + s;
pthread_mutex_lock(&this->mutex); this->mutex->lock(this->mutex);
while(TRUE) while(TRUE)
{ {
if (this->list->get_count(this->list) == 0) if (this->list->get_count(this->list) == 0)
@@ -220,8 +216,8 @@ static void schedule_job(private_scheduler_t *this, job_t *job, u_int32_t time)
iterator->destroy(iterator); iterator->destroy(iterator);
break; break;
} }
pthread_cond_signal(&this->condvar); this->condvar->signal(this->condvar);
pthread_mutex_unlock(&this->mutex); this->mutex->unlock(this->mutex);
} }
/** /**
@@ -229,8 +225,9 @@ static void schedule_job(private_scheduler_t *this, job_t *job, u_int32_t time)
*/ */
static void destroy(private_scheduler_t *this) static void destroy(private_scheduler_t *this)
{ {
this->cancelled = TRUE;
this->job->cancel(this->job); this->job->cancel(this->job);
this->condvar->destroy(this->condvar);
this->mutex->destroy(this->mutex);
this->list->destroy_function(this->list, (void*)event_destroy); this->list->destroy_function(this->list, (void*)event_destroy);
free(this); free(this);
} }
@@ -247,9 +244,8 @@ scheduler_t * scheduler_create()
this->public.destroy = (void(*)(scheduler_t*)) destroy; this->public.destroy = (void(*)(scheduler_t*)) destroy;
this->list = linked_list_create(); this->list = linked_list_create();
this->cancelled = FALSE; this->mutex = mutex_create(MUTEX_DEFAULT);
pthread_mutex_init(&this->mutex, NULL); this->condvar = condvar_create(CONDVAR_DEFAULT);
pthread_cond_init(&this->condvar, NULL);
this->job = callback_job_create((callback_job_cb_t)schedule, this, NULL, NULL); this->job = callback_job_create((callback_job_cb_t)schedule, this, NULL, NULL);
charon->processor->queue_job(charon->processor, (job_t*)this->job); charon->processor->queue_job(charon->processor, (job_t*)this->job);
+32 -16
View File
@@ -1,4 +1,5 @@
/* /*
* Copyright (C) 2008 Tobias Brunner
* Copyright (C) 2008 Martin Willi * Copyright (C) 2008 Martin Willi
* Hochschule fuer Technik Rapperswil * Hochschule fuer Technik Rapperswil
* *
@@ -332,28 +333,17 @@ static void wait(private_condvar_t *this, private_mutex_t *mutex)
} }
/** /**
* Implementation of condvar_t.timed_wait. * Implementation of condvar_t.timed_wait_abs.
*/ */
static bool timed_wait(private_condvar_t *this, private_mutex_t *mutex, static bool timed_wait_abs(private_condvar_t *this, private_mutex_t *mutex,
u_int timeout) timeval_t time)
{ {
struct timespec ts; struct timespec ts;
struct timeval tv;
u_int s, ms;
bool timed_out; bool timed_out;
gettimeofday(&tv, NULL); ts.tv_sec = time.tv_sec;
ts.tv_nsec = time.tv_usec * 1000;
s = timeout / 1000;
ms = timeout % 1000;
ts.tv_sec = tv.tv_sec + s;
ts.tv_nsec = tv.tv_usec * 1000 + ms * 1000000;
if (ts.tv_nsec > 1000000000 /* 1s */)
{
ts.tv_nsec -= 1000000000;
ts.tv_sec++;
}
if (mutex->recursive) if (mutex->recursive)
{ {
private_r_mutex_t* recursive = (private_r_mutex_t*)mutex; private_r_mutex_t* recursive = (private_r_mutex_t*)mutex;
@@ -371,6 +361,31 @@ static bool timed_wait(private_condvar_t *this, private_mutex_t *mutex,
return timed_out; return timed_out;
} }
/**
* Implementation of condvar_t.timed_wait.
*/
static bool timed_wait(private_condvar_t *this, private_mutex_t *mutex,
u_int timeout)
{
timeval_t tv;
u_int s, ms;
gettimeofday(&tv, NULL);
s = timeout / 1000;
ms = timeout % 1000;
tv.tv_sec += s;
tv.tv_usec += ms * 1000;
if (tv.tv_usec > 1000000 /* 1s */)
{
tv.tv_usec -= 1000000;
tv.tv_sec++;
}
return timed_wait_abs(this, mutex, tv);
}
/** /**
* Implementation of condvar_t.signal. * Implementation of condvar_t.signal.
*/ */
@@ -410,6 +425,7 @@ condvar_t *condvar_create(condvar_type_t type)
this->public.wait = (void(*)(condvar_t*, mutex_t *mutex))wait; this->public.wait = (void(*)(condvar_t*, mutex_t *mutex))wait;
this->public.timed_wait = (bool(*)(condvar_t*, mutex_t *mutex, u_int timeout))timed_wait; this->public.timed_wait = (bool(*)(condvar_t*, mutex_t *mutex, u_int timeout))timed_wait;
this->public.timed_wait_abs = (bool(*)(condvar_t*, mutex_t *mutex, timeval_t time))timed_wait_abs;
this->public.signal = (void(*)(condvar_t*))signal; this->public.signal = (void(*)(condvar_t*))signal;
this->public.broadcast = (void(*)(condvar_t*))broadcast; this->public.broadcast = (void(*)(condvar_t*))broadcast;
this->public.destroy = (void(*)(condvar_t*))condvar_destroy; this->public.destroy = (void(*)(condvar_t*))condvar_destroy;
+12
View File
@@ -1,4 +1,5 @@
/* /*
* Copyright (C) 2008 Tobias Brunner
* Copyright (C) 2008 Martin Willi * Copyright (C) 2008 Martin Willi
* Hochschule fuer Technik Rapperswil * Hochschule fuer Technik Rapperswil
* *
@@ -11,6 +12,8 @@
* WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
* or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
* for more details. * for more details.
*
* $Id$
*/ */
/** /**
@@ -98,6 +101,15 @@ struct condvar_t {
*/ */
bool (*timed_wait)(condvar_t *this, mutex_t *mutex, u_int timeout); bool (*timed_wait)(condvar_t *this, mutex_t *mutex, u_int timeout);
/**
* Wait on a condvar until it gets signalized, or times out.
*
* @param mutex mutex to release while waiting
* @param time absolute time until timeout
* @return TRUE if timed out, FALSE otherwise
*/
bool (*timed_wait_abs)(condvar_t *this, mutex_t *mutex, timeval_t timeout);
/** /**
* Wake up a single thread in a condvar. * Wake up a single thread in a condvar.
*/ */