diff options
Diffstat (limited to 'src/charon/processing')
27 files changed, 1207 insertions, 1318 deletions
diff --git a/src/charon/processing/event_queue.c b/src/charon/processing/event_queue.c deleted file mode 100644 index 40bcb1ed8..000000000 --- a/src/charon/processing/event_queue.c +++ /dev/null @@ -1,290 +0,0 @@ -/** - * @file event_queue.c - * - * @brief Implementation of event_queue_t - * - */ - -/* - * Copyright (C) 2005-2006 Martin Willi - * Copyright (C) 2005 Jan Hutter - * Hochschule fuer Technik Rapperswil - * - * This program is free software; you can redistribute it and/or modify it - * under the terms of the GNU General Public License as published by the - * Free Software Foundation; either version 2 of the License, or (at your - * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. - * - * This program is distributed in the hope that it will be useful, but - * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY - * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License - * for more details. - */ - -#include <pthread.h> -#include <stdlib.h> - -#include "event_queue.h" - -#include <library.h> -#include <utils/linked_list.h> - - - -typedef struct event_t event_t; - -/** - * Event containing a job and a schedule time - */ -struct event_t { - /** - * Time to fire the event. - */ - timeval_t time; - - /** - * Every event has its assigned job. - */ - job_t * job; -}; - -/** - * destroy an event and its job - */ -static void event_destroy(event_t *event) -{ - event->job->destroy(event->job); - free(event); -} - -typedef struct private_event_queue_t private_event_queue_t; - -/** - * Private Variables and Functions of event_queue_t class. - */ -struct private_event_queue_t { - /** - * Public part. - */ - event_queue_t public; - - /** - * The events are stored in a linked list of type linked_list_t. - */ - linked_list_t *list; - - /** - * Access to linked_list is locked through this mutex. - */ - pthread_mutex_t mutex; - - /** - * If the queue is empty or an event has not to be fired - * a thread has to wait. - * - * This condvar is used to wake up such a thread. - */ - pthread_cond_t condvar; -}; - -/** - * Returns the difference of to timeval structs in milliseconds - */ -static long time_difference(struct timeval *end_time, struct timeval *start_time) -{ - time_t s; - suseconds_t us; - - s = (end_time->tv_sec - start_time->tv_sec); - us = (end_time->tv_usec - start_time->tv_usec); - return ((s * 1000) + us/1000); -} - -/** - * Implements event_queue_t.get_count - */ -static int get_count(private_event_queue_t *this) -{ - int count; - pthread_mutex_lock(&(this->mutex)); - count = this->list->get_count(this->list); - pthread_mutex_unlock(&(this->mutex)); - return count; -} - -/** - * Implements event_queue_t.get - */ -static job_t *get(private_event_queue_t *this) -{ - timespec_t timeout; - timeval_t current_time; - event_t * next_event; - job_t *job; - int oldstate; - - pthread_mutex_lock(&(this->mutex)); - - while (TRUE) - { - while(this->list->get_count(this->list) == 0) - { - /* add mutex unlock handler for cancellation, enable cancellation */ - pthread_cleanup_push((void(*)(void*))pthread_mutex_unlock, (void*)&(this->mutex)); - pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &oldstate); - - pthread_cond_wait( &(this->condvar), &(this->mutex)); - - /* reset cancellation, remove mutex-unlock handler (without executing) */ - pthread_setcancelstate(oldstate, NULL); - pthread_cleanup_pop(0); - } - - this->list->get_first(this->list, (void **)&next_event); - - gettimeofday(¤t_time, NULL); - long difference = time_difference(¤t_time,&(next_event->time)); - if (difference <= 0) - { - timeout.tv_sec = next_event->time.tv_sec; - timeout.tv_nsec = next_event->time.tv_usec * 1000; - - /* add mutex unlock handler for cancellation, enable cancellation */ - pthread_cleanup_push((void(*)(void*))pthread_mutex_unlock, (void*)&(this->mutex)); - pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &oldstate); - - pthread_cond_timedwait(&(this->condvar), &(this->mutex), &timeout); - - /* reset cancellation, remove mutex-unlock handler (without executing) */ - pthread_setcancelstate(oldstate, NULL); - pthread_cleanup_pop(0); - } - else - { - /* event available */ - this->list->remove_first(this->list, (void **)&next_event); - job = next_event->job; - free(next_event); - break; - } - } - pthread_cond_signal( &(this->condvar)); - pthread_mutex_unlock(&(this->mutex)); - - return job; -} - -/** - * Implements function add_absolute of event_queue_t. - * See #event_queue_s.add_absolute for description. - */ -static void add_absolute(private_event_queue_t *this, job_t *job, timeval_t time) -{ - event_t *event; - event_t *current_event; - iterator_t *iterator; - - /* create event */ - event = malloc_thing(event_t); - event->time = time; - event->job = job; - - pthread_mutex_lock(&(this->mutex)); - - /* while just used to break out */ - while(TRUE) - { - if (this->list->get_count(this->list) == 0) - { - this->list->insert_first(this->list,event); - break; - } - - /* check last entry */ - this->list->get_last(this->list,(void **) ¤t_event); - - if (time_difference(&(event->time), &(current_event->time)) >= 0) - { - /* my event has to be fired after the last event in list */ - this->list->insert_last(this->list,event); - break; - } - - /* check first entry */ - this->list->get_first(this->list,(void **) ¤t_event); - - if (time_difference(&(event->time), &(current_event->time)) < 0) - { - /* my 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); - iterator->iterate(iterator, (void**)¤t_event); - /* first element has not to be checked (already done) */ - while(iterator->iterate(iterator, (void**)¤t_event)) - { - if (time_difference(&(event->time), &(current_event->time)) <= 0) - { - /* my event has to be fired before the current event in list */ - iterator->insert_before(iterator,event); - break; - } - } - iterator->destroy(iterator); - break; - } - - pthread_cond_signal( &(this->condvar)); - pthread_mutex_unlock(&(this->mutex)); -} - -/** - * Implements event_queue_t.add_relative. - */ -static void add_relative(event_queue_t *this, job_t *job, u_int32_t ms) -{ - timeval_t current_time; - timeval_t time; - - time_t s = ms / 1000; - suseconds_t us = (ms - s * 1000) * 1000; - - gettimeofday(¤t_time, NULL); - - time.tv_usec = (current_time.tv_usec + us) % 1000000; - time.tv_sec = current_time.tv_sec + (current_time.tv_usec + us)/1000000 + s; - - this->add_absolute(this, job, time); -} - - -/** - * Implements event_queue_t.destroy. - */ -static void event_queue_destroy(private_event_queue_t *this) -{ - this->list->destroy_function(this->list, (void*)event_destroy); - free(this); -} - -/* - * Documented in header - */ -event_queue_t *event_queue_create() -{ - private_event_queue_t *this = malloc_thing(private_event_queue_t); - - this->public.get_count = (int (*) (event_queue_t *event_queue)) get_count; - this->public.get = (job_t *(*) (event_queue_t *event_queue)) get; - this->public.add_absolute = (void (*) (event_queue_t *event_queue, job_t *job, timeval_t time)) add_absolute; - this->public.add_relative = (void (*) (event_queue_t *event_queue, job_t *job, u_int32_t ms)) add_relative; - this->public.destroy = (void (*) (event_queue_t *event_queue)) event_queue_destroy; - - this->list = linked_list_create(); - pthread_mutex_init(&(this->mutex), NULL); - pthread_cond_init(&(this->condvar), NULL); - - return (&this->public); -} diff --git a/src/charon/processing/event_queue.h b/src/charon/processing/event_queue.h deleted file mode 100644 index c85286bf2..000000000 --- a/src/charon/processing/event_queue.h +++ /dev/null @@ -1,118 +0,0 @@ -/** - * @file event_queue.h - * - * @brief Interface of job_queue_t. - * - */ - -/* - * Copyright (C) 2005-2006 Martin Willi - * Copyright (C) 2005 Jan Hutter - * Hochschule fuer Technik Rapperswil - * - * This program is free software; you can redistribute it and/or modify it - * under the terms of the GNU General Public License as published by the - * Free Software Foundation; either version 2 of the License, or (at your - * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. - * - * This program is distributed in the hope that it will be useful, but - * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY - * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License - * for more details. - */ - -#ifndef EVENT_QUEUE_H_ -#define EVENT_QUEUE_H_ - -typedef struct event_queue_t event_queue_t; - -#include <sys/time.h> - -#include <library.h> -#include <processing/jobs/job.h> - -/** - * @brief Event-Queue used to store timed events. - * - * Added events are sorted. The get method blocks until - * the time is elapsed to process the next event. The get - * method is called from the scheduler_t thread, which - * will add the jobs to to job_queue_t for further processing. - * - * Although the event-queue is based on a linked_list_t - * all access functions are thread-save implemented. - * - * @b Constructors: - * - event_queue_create() - * - * @ingroup processing - */ -struct event_queue_t { - - /** - * @brief Returns number of events in queue. - * - * @param event_queue calling object - * @return number of events in queue - */ - int (*get_count) (event_queue_t *event_queue); - - /** - * @brief Get the next job from the event-queue. - * - * If no event is pending, this function blocks until a job can be returned. - * - * @param event_queue calling object - * @param[out] job pointer to a job pointer where to job is returned to - * @return next job - */ - job_t *(*get) (event_queue_t *event_queue); - - /** - * @brief Adds a event to the queue, using a relative time. - * - * This function is non blocking and adds a job_t at a specific time to the list. - * The specific job object has to get destroyed by the thread which - * removes the job. - * - * @param event_queue calling object - * @param[in] job job to add to the queue (job is not copied) - * @param[in] time relative time, when the event has to get fired - */ - void (*add_relative) (event_queue_t *event_queue, job_t *job, u_int32_t ms); - - /** - * @brief Adds a event to the queue, using an absolute time. - * - * This function is non blocking and adds a job_t at a specific time to the list. - * The specific job object has to get destroyed by the thread which - * removes the job. - * - * @param event_queue calling object - * @param[in] job job to add to the queue (job is not copied) - * @param[in] time absolute time, when the event has to get fired - */ - void (*add_absolute) (event_queue_t *event_queue, job_t *job, timeval_t time); - - /** - * @brief Destroys a event_queue object. - * - * @warning The caller of this function has to make sure - * that no thread is going to add or get an event from the event_queue - * after calling this function. - * - * @param event_queue calling object - */ - void (*destroy) (event_queue_t *event_queue); -}; - -/** - * @brief Creates an empty event_queue. - * - * @returns event_queue_t object - * - * @ingroup processing - */ -event_queue_t *event_queue_create(void); - -#endif /*EVENT_QUEUE_H_*/ diff --git a/src/charon/processing/job_queue.c b/src/charon/processing/job_queue.c deleted file mode 100644 index 2310ca6ff..000000000 --- a/src/charon/processing/job_queue.c +++ /dev/null @@ -1,139 +0,0 @@ -/** - * @file job_queue.c - * - * @brief Implementation of job_queue_t - * - */ - -/* - * Copyright (C) 2005-2006 Martin Willi - * Copyright (C) 2005 Jan Hutter - * Hochschule fuer Technik Rapperswil - * - * This program is free software; you can redistribute it and/or modify it - * under the terms of the GNU General Public License as published by the - * Free Software Foundation; either version 2 of the License, or (at your - * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. - * - * This program is distributed in the hope that it will be useful, but - * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY - * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License - * for more details. - */ - -#include <stdlib.h> -#include <pthread.h> - -#include "job_queue.h" - -#include <utils/linked_list.h> - - -typedef struct private_job_queue_t private_job_queue_t; - -/** - * @brief Private Variables and Functions of job_queue class - * - */ -struct private_job_queue_t { - - /** - * public members - */ - job_queue_t public; - - /** - * The jobs are stored in a linked list - */ - linked_list_t *list; - - /** - * access to linked_list is locked through this mutex - */ - pthread_mutex_t mutex; - - /** - * If the queue is empty a thread has to wait - * This condvar is used to wake up such a thread - */ - pthread_cond_t condvar; -}; - - -/** - * implements job_queue_t.get_count - */ -static int get_count(private_job_queue_t *this) -{ - int count; - pthread_mutex_lock(&(this->mutex)); - count = this->list->get_count(this->list); - pthread_mutex_unlock(&(this->mutex)); - return count; -} - -/** - * implements job_queue_t.get - */ -static job_t *get(private_job_queue_t *this) -{ - int oldstate; - job_t *job; - pthread_mutex_lock(&(this->mutex)); - /* go to wait while no jobs available */ - while(this->list->get_count(this->list) == 0) - { - /* add mutex unlock handler for cancellation, enable cancellation */ - pthread_cleanup_push((void(*)(void*))pthread_mutex_unlock, (void*)&(this->mutex)); - pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &oldstate); - - pthread_cond_wait( &(this->condvar), &(this->mutex)); - - /* reset cancellation, remove mutex-unlock handler (without executing) */ - pthread_setcancelstate(oldstate, NULL); - pthread_cleanup_pop(0); - } - this->list->remove_first(this->list, (void **)&job); - pthread_mutex_unlock(&(this->mutex)); - return job; -} - -/** - * implements function job_queue_t.add - */ -static void add(private_job_queue_t *this, job_t *job) -{ - pthread_mutex_lock(&(this->mutex)); - this->list->insert_last(this->list,job); - pthread_cond_signal( &(this->condvar)); - pthread_mutex_unlock(&(this->mutex)); -} - -/** - * implements job_queue_t.destroy - */ -static void job_queue_destroy (private_job_queue_t *this) -{ - this->list->destroy_offset(this->list, offsetof(job_t, destroy)); - free(this); -} - -/* - * - * Documented in header - */ -job_queue_t *job_queue_create(void) -{ - private_job_queue_t *this = malloc_thing(private_job_queue_t); - - this->public.get_count = (int(*)(job_queue_t*))get_count; - this->public.get = (job_t*(*)(job_queue_t*))get; - this->public.add = (void(*)(job_queue_t*, job_t*))add; - this->public.destroy = (void(*)(job_queue_t*))job_queue_destroy; - - this->list = linked_list_create(); - pthread_mutex_init(&(this->mutex), NULL); - pthread_cond_init(&(this->condvar), NULL); - - return (&this->public); -} diff --git a/src/charon/processing/job_queue.h b/src/charon/processing/job_queue.h deleted file mode 100644 index 9b58588ae..000000000 --- a/src/charon/processing/job_queue.h +++ /dev/null @@ -1,100 +0,0 @@ -/** - * @file job_queue.h - * - * @brief Interface of job_queue_t. - * - */ - -/* - * Copyright (C) 2005-2006 Martin Willi - * Copyright (C) 2005 Jan Hutter - * Hochschule fuer Technik Rapperswil - * - * This program is free software; you can redistribute it and/or modify it - * under the terms of the GNU General Public License as published by the - * Free Software Foundation; either version 2 of the License, or (at your - * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. - * - * This program is distributed in the hope that it will be useful, but - * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY - * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License - * for more details. - */ - -#ifndef JOB_QUEUE_H_ -#define JOB_QUEUE_H_ - -typedef struct job_queue_t job_queue_t; - -#include <library.h> -#include <processing/jobs/job.h> - -/** - * @brief The job queue stores jobs, which will be processed by the thread_pool_t. - * - * Jobs are added from various sources, from the threads and - * from the event_queue_t. - * Although the job-queue is based on a linked_list_t - * all access functions are thread-save implemented. - * - * @b Constructors: - * - job_queue_create() - * - * @ingroup processing - */ -struct job_queue_t { - - /** - * @brief Returns number of jobs in queue. - * - * @param job_queue_t calling object - * @returns number of items in queue - */ - int (*get_count) (job_queue_t *job_queue); - - /** - * @brief Get the next job from the queue. - * - * If the queue is empty, this function blocks until a job can be returned. - * After using, the returned job has to get destroyed by the caller. - * - * @param job_queue_t calling object - * @param[out] job pointer to a job pointer where to job is returned to - * @return next job - */ - job_t *(*get) (job_queue_t *job_queue); - - /** - * @brief Adds a job to the queue. - * - * This function is non blocking and adds a job_t to the list. - * The specific job object has to get destroyed by the thread which - * removes the job. - * - * @param job_queue_t calling object - * @param job job to add to the queue (job is not copied) - */ - void (*add) (job_queue_t *job_queue, job_t *job); - - /** - * @brief Destroys a job_queue object. - * - * @warning The caller of this function has to make sure - * that no thread is going to add or get a job from the job_queue - * after calling this function. - * - * @param job_queue_t calling object - */ - void (*destroy) (job_queue_t *job_queue); -}; - -/** - * @brief Creates an empty job_queue. - * - * @return job_queue_t object - * - * @ingroup processing - */ -job_queue_t *job_queue_create(void); - -#endif /*JOB_QUEUE_H_*/ diff --git a/src/charon/processing/jobs/acquire_job.c b/src/charon/processing/jobs/acquire_job.c index b4ffb258d..48a77f558 100644 --- a/src/charon/processing/jobs/acquire_job.c +++ b/src/charon/processing/jobs/acquire_job.c @@ -43,17 +43,17 @@ struct private_acquire_job_t { }; /** - * Implementation of job_t.get_type. + * Implementation of job_t.destroy. */ -static job_type_t get_type(private_acquire_job_t *this) +static void destroy(private_acquire_job_t *this) { - return ACQUIRE; + free(this); } /** * Implementation of job_t.execute. */ -static status_t execute(private_acquire_job_t *this) +static void execute(private_acquire_job_t *this) { ike_sa_t *ike_sa; @@ -63,20 +63,14 @@ static status_t execute(private_acquire_job_t *this) { DBG2(DBG_JOB, "CHILD_SA with reqid %d not found for acquiring", this->reqid); - return DESTROY_ME; } - ike_sa->acquire(ike_sa, this->reqid); - - charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa); - return DESTROY_ME; -} - -/** - * Implementation of job_t.destroy. - */ -static void destroy(private_acquire_job_t *this) -{ - free(this); + else + { + ike_sa->acquire(ike_sa, this->reqid); + + charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa); + } + destroy(this); } /* @@ -87,12 +81,11 @@ acquire_job_t *acquire_job_create(u_int32_t reqid) private_acquire_job_t *this = malloc_thing(private_acquire_job_t); /* interface functions */ - this->public.job_interface.get_type = (job_type_t (*) (job_t *)) get_type; - this->public.job_interface.execute = (status_t (*) (job_t *)) execute; + this->public.job_interface.execute = (void (*) (job_t *)) execute; this->public.job_interface.destroy = (void (*)(job_t*)) destroy; /* private variables */ this->reqid = reqid; - return &(this->public); + return &this->public; } diff --git a/src/charon/processing/jobs/callback_job.c b/src/charon/processing/jobs/callback_job.c new file mode 100644 index 000000000..53e7caa95 --- /dev/null +++ b/src/charon/processing/jobs/callback_job.c @@ -0,0 +1,213 @@ +/** + * @file callback_job.c + * + * @brief Implementation of callback_job_t. + * + */ + +/* + * Copyright (C) 2007 Martin Willi + * Hochschule fuer Technik Rapperswil + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2 of the License, or (at your + * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. + * + * This program is distributed in the hope that it will be useful, but + * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY + * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License + * for more details. + */ + +#include "callback_job.h" + +#include <daemon.h> + +typedef struct private_callback_job_t private_callback_job_t; + +/** + * Private data of an callback_job_t Object. + */ +struct private_callback_job_t { + /** + * Public callback_job_t interface. + */ + callback_job_t public; + + /** + * Callback to call on execution + */ + callback_job_cb_t callback; + + /** + * parameter to supply to callback + */ + void *data; + + /** + * cleanup function for data + */ + callback_job_cleanup_t cleanup; + + /** + * thread ID of the job, if running + */ + pthread_t thread; + + /** + * mutex to access jobs interna + */ + pthread_mutex_t mutex; + + /** + * condvar to synchronize thread startup/cancellation + */ + pthread_cond_t condvar; + + /** + * list of asociated child jobs + */ + linked_list_t *children; + + /** + * parent of this job, or NULL + */ + private_callback_job_t *parent; +}; + +/** + * Implements job_t.destroy. + */ +static void destroy(private_callback_job_t *this) +{ + if (this->cleanup) + { + this->cleanup(this->data); + } + this->children->destroy(this->children); + free(this); +} + +/** + * unregister a child from its parent, if any. + */ +static void unregister(private_callback_job_t *this) +{ + if (this->parent) + { + iterator_t *iterator; + private_callback_job_t *child; + + pthread_mutex_lock(&this->parent->mutex); + iterator = this->parent->children->create_iterator(this->parent->children, TRUE); + while (iterator->iterate(iterator, (void**)&child)) + { + if (child == this) + { + iterator->remove(iterator); + break; + } + } + iterator->destroy(iterator); + pthread_mutex_unlock(&this->parent->mutex); + } +} + +/** + * Implementation of callback_job_t.cancel. + */ +static void cancel(private_callback_job_t *this) +{ + pthread_t thread; + + /* wait until thread has started */ + pthread_mutex_lock(&this->mutex); + while (this->thread == 0) + { + pthread_cond_wait(&this->condvar, &this->mutex); + } + thread = this->thread; + + /* terminate its children */ + this->children->invoke(this->children, offsetof(callback_job_t, cancel)); + pthread_mutex_unlock(&this->mutex); + + /* terminate thread */ + pthread_cancel(thread); + pthread_join(thread, NULL); +} + +/** + * Implementation of job_t.execute. + */ +static void execute(private_callback_job_t *this) +{ + bool cleanup = FALSE; + + pthread_mutex_lock(&this->mutex); + this->thread = pthread_self(); + pthread_cond_signal(&this->condvar); + pthread_mutex_unlock(&this->mutex); + + pthread_cleanup_push((void*)destroy, this); + while (TRUE) + { + switch (this->callback(this->data)) + { + case JOB_REQUEUE_DIRECT: + continue; + case JOB_REQUEUE_FAIR: + { + charon->processor->queue_job(charon->processor, + &this->public.job_interface); + break; + } + case JOB_REQUEUE_NONE: + default: + { + cleanup = TRUE; + break; + } + } + break; + } + unregister(this); + pthread_cleanup_pop(cleanup); +} + +/* + * Described in header. + */ +callback_job_t *callback_job_create(callback_job_cb_t cb, void *data, + callback_job_cleanup_t cleanup, + callback_job_t *parent) +{ + private_callback_job_t *this = malloc_thing(private_callback_job_t); + + /* interface functions */ + this->public.job_interface.execute = (void (*) (job_t *)) execute; + this->public.job_interface.destroy = (void (*) (job_t *)) destroy; + this->public.cancel = (void(*)(callback_job_t*))cancel; + + /* private variables */ + pthread_mutex_init(&this->mutex, NULL); + pthread_cond_init(&this->condvar, NULL); + this->callback = cb; + this->data = data; + this->cleanup = cleanup; + this->thread = 0; + this->children = linked_list_create(); + this->parent = (private_callback_job_t*)parent; + + /* register us at parent */ + if (parent) + { + pthread_mutex_lock(&this->parent->mutex); + this->parent->children->insert_last(this->parent->children, this); + pthread_mutex_unlock(&this->parent->mutex); + } + + return &this->public; +} + diff --git a/src/charon/processing/jobs/callback_job.h b/src/charon/processing/jobs/callback_job.h new file mode 100644 index 000000000..169f2d207 --- /dev/null +++ b/src/charon/processing/jobs/callback_job.h @@ -0,0 +1,135 @@ +/** + * @file callback_job.h + * + * @brief Interface of callback_job_t. + * + */ + +/* + * Copyright (C) 2007 Martin Willi + * Hochschule fuer Technik Rapperswil + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2 of the License, or (at your + * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. + * + * This program is distributed in the hope that it will be useful, but + * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY + * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License + * for more details. + */ + +#ifndef CALLBACK_JOB_H_ +#define CALLBACK_JOB_H_ + +typedef struct callback_job_t callback_job_t; + +#include <library.h> +#include <processing/jobs/job.h> + + +typedef enum job_requeue_t job_requeue_t; + +/** + * @brief Job requeueing policy + * + * The job requeueing policy defines how a job is handled when the callback + * function returns. + * + * @ingroup jobs + */ +enum job_requeue_t { + + /** + * Do not requeue job, destroy it + */ + JOB_REQUEUE_NONE, + + /** + * Reque the job fairly, meaning it has to requeue as any other job + */ + JOB_REQUEUE_FAIR, + + /** + * Reexecute the job directly, without the need of requeueing it + */ + JOB_REQUEUE_DIRECT, +}; + +/** + * @brief The callback function to use for the callback job. + * + * This is the function to use as callback for a callback job. It receives + * a parameter supplied to the callback jobs constructor. + * + * @param data param supplied to job + * @return requeing policy how to requeue the job + * + * @ingroup jobs + */ +typedef job_requeue_t (*callback_job_cb_t)(void *data); + +/** + * @brief Cleanup function to use for data cleanup. + * + * The callback has an optional user argument which receives data. However, + * this data may be cleaned up if it is allocated. This is the function + * to supply to the constructor. + * + * @param data param supplied to job + * @return requeing policy how to requeue the job + * + * @ingroup jobs + */ +typedef void (*callback_job_cleanup_t)(void *data); + +/** + * @brief Class representing an callback Job. + * + * This is a special job which allows a simple callback function to + * be executed by a thread of the thread pool. This allows simple execution + * of asynchronous methods, without to manage threads. + * + * @b Constructors: + * - callback_job_create() + * + * @ingroup jobs + */ +struct callback_job_t { + /** + * The job_t interface. + */ + job_t job_interface; + + /** + * @brief Cancel the jobs thread and wait for its termination. + * + * @param this calling object + */ + void (*cancel)(callback_job_t *this); +}; + +/** + * @brief Creates a callback job. + * + * The cleanup function is called when the job gets destroyed to destroy + * the associated data. + * If parent is not NULL, the specified job gets an association. Whenever + * the parent gets cancelled (or runs out), all of its children are cancelled, + * too. + * + * @param cb callback to call from the processor + * @param data user data to supply to callback + * @param cleanup destructor for data on destruction, or NULL + * @param parent parent of this job + * @return callback_job_t object + * + * @ingroup jobs + */ +callback_job_t *callback_job_create(callback_job_cb_t cb, void *data, + callback_job_cleanup_t cleanup, + callback_job_t *parent); + +#endif /* CALLBACK_JOB_H_ */ + diff --git a/src/charon/processing/jobs/delete_child_sa_job.c b/src/charon/processing/jobs/delete_child_sa_job.c index f694696b0..23f330293 100644 --- a/src/charon/processing/jobs/delete_child_sa_job.c +++ b/src/charon/processing/jobs/delete_child_sa_job.c @@ -54,17 +54,17 @@ struct private_delete_child_sa_job_t { }; /** - * Implementation of job_t.get_type. + * Implementation of job_t.destroy. */ -static job_type_t get_type(private_delete_child_sa_job_t *this) +static void destroy(private_delete_child_sa_job_t *this) { - return DELETE_CHILD_SA; + free(this); } /** * Implementation of job_t.execute. */ -static status_t execute(private_delete_child_sa_job_t *this) +static void execute(private_delete_child_sa_job_t *this) { ike_sa_t *ike_sa; @@ -74,20 +74,14 @@ static status_t execute(private_delete_child_sa_job_t *this) { DBG1(DBG_JOB, "CHILD_SA with reqid %d not found for delete", this->reqid); - return DESTROY_ME; } - ike_sa->delete_child_sa(ike_sa, this->protocol, this->spi); - - charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa); - return DESTROY_ME; -} - -/** - * Implementation of job_t.destroy. - */ -static void destroy(private_delete_child_sa_job_t *this) -{ - free(this); + else + { + ike_sa->delete_child_sa(ike_sa, this->protocol, this->spi); + + charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa); + } + destroy(this); } /* @@ -100,8 +94,7 @@ delete_child_sa_job_t *delete_child_sa_job_create(u_int32_t reqid, private_delete_child_sa_job_t *this = malloc_thing(private_delete_child_sa_job_t); /* interface functions */ - this->public.job_interface.get_type = (job_type_t (*) (job_t *)) get_type; - this->public.job_interface.execute = (status_t (*) (job_t *)) execute; + this->public.job_interface.execute = (void (*) (job_t *)) execute; this->public.job_interface.destroy = (void (*)(job_t*)) destroy; /* private variables */ @@ -109,5 +102,6 @@ delete_child_sa_job_t *delete_child_sa_job_create(u_int32_t reqid, this->protocol = protocol; this->spi = spi; - return &(this->public); + return &this->public; } + diff --git a/src/charon/processing/jobs/delete_ike_sa_job.c b/src/charon/processing/jobs/delete_ike_sa_job.c index 706155aa6..8d8c0cf36 100644 --- a/src/charon/processing/jobs/delete_ike_sa_job.c +++ b/src/charon/processing/jobs/delete_ike_sa_job.c @@ -47,18 +47,20 @@ struct private_delete_ike_sa_job_t { bool delete_if_established; }; + /** - * Implements job_t.get_type. + * Implements job_t.destroy. */ -static job_type_t get_type(private_delete_ike_sa_job_t *this) +static void destroy(private_delete_ike_sa_job_t *this) { - return DELETE_IKE_SA; + this->ike_sa_id->destroy(this->ike_sa_id); + free(this); } /** * Implementation of job_t.execute. */ -static status_t execute(private_delete_ike_sa_job_t *this) +static void execute(private_delete_ike_sa_job_t *this) { ike_sa_t *ike_sa; @@ -93,16 +95,7 @@ static status_t execute(private_delete_ike_sa_job_t *this) } } } - return DESTROY_ME; -} - -/** - * Implements job_t.destroy. - */ -static void destroy(private_delete_ike_sa_job_t *this) -{ - this->ike_sa_id->destroy(this->ike_sa_id); - free(this); + destroy(this); } /* @@ -114,8 +107,7 @@ delete_ike_sa_job_t *delete_ike_sa_job_create(ike_sa_id_t *ike_sa_id, private_delete_ike_sa_job_t *this = malloc_thing(private_delete_ike_sa_job_t); /* interface functions */ - this->public.job_interface.get_type = (job_type_t (*) (job_t *)) get_type; - this->public.job_interface.execute = (status_t (*) (job_t *)) execute; + this->public.job_interface.execute = (void (*) (job_t *)) execute; this->public.job_interface.destroy = (void (*)(job_t *)) destroy;; /* private variables */ diff --git a/src/charon/processing/jobs/job.c b/src/charon/processing/jobs/job.c deleted file mode 100644 index d32d1bc61..000000000 --- a/src/charon/processing/jobs/job.c +++ /dev/null @@ -1,39 +0,0 @@ -/** - * @file job.c - * - * @brief Interface additions to job_t. - * - */ - -/* - * Copyright (C) 2005-2006 Martin Willi - * Copyright (C) 2005 Jan Hutter - * Hochschule fuer Technik Rapperswil - * - * This program is free software; you can redistribute it and/or modify it - * under the terms of the GNU General Public License as published by the - * Free Software Foundation; either version 2 of the License, or (at your - * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. - * - * This program is distributed in the hope that it will be useful, but - * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY - * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License - * for more details. - */ - - -#include "job.h" - -ENUM(job_type_names, PROCESS_MESSAGE, SEND_DPD, - "PROCESS_MESSAGE", - "RETRANSMIT", - "INITIATE", - "ROUTE", - "ACQUIRE", - "DELETE_IKE_SA", - "DELETE_CHILD_SA", - "REKEY_CHILD_SA", - "REKEY_IKE_SA", - "SEND_KEEPALIVE", - "SEND_DPD", -); diff --git a/src/charon/processing/jobs/job.h b/src/charon/processing/jobs/job.h index 28632672d..1826c53b4 100644 --- a/src/charon/processing/jobs/job.h +++ b/src/charon/processing/jobs/job.h @@ -24,108 +24,14 @@ #ifndef JOB_H_ #define JOB_H_ -typedef enum job_type_t job_type_t; typedef struct job_t job_t; #include <library.h> -/** - * @brief Definition of the various job types. - * - * @ingroup jobs - */ -enum job_type_t { - /** - * Process an incoming IKEv2-Message. - * - * Job is implemented in class process_message_job_t - */ - PROCESS_MESSAGE, - - /** - * Retransmit an IKEv2-Message. - * - * Job is implemented in class retransmit_job_t - */ - RETRANSMIT, - - /** - * Set up a CHILD_SA, optional with an IKE_SA. - * - * Job is implemented in class initiate_job_t - */ - INITIATE, - - /** - * Install SPD entries. - * - * Job is implemented in class route_job_t - */ - ROUTE, - - /** - * React on a acquire message from the kernel (e.g. setup CHILD_SA) - * - * Job is implemented in class acquire_job_t - */ - ACQUIRE, - - /** - * Delete an IKE_SA. - * - * Job is implemented in class delete_ike_sa_job_t - */ - DELETE_IKE_SA, - - /** - * Delete a CHILD_SA. - * - * Job is implemented in class delete_child_sa_job_t - */ - DELETE_CHILD_SA, - - /** - * Rekey a CHILD_SA. - * - * Job is implemented in class rekey_child_sa_job_t - */ - REKEY_CHILD_SA, - - /** - * Rekey an IKE_SA. - * - * Job is implemented in class rekey_ike_sa_job_t - */ - REKEY_IKE_SA, - - /** - * Send a keepalive packet. - * - * Job is implemented in class type send_keepalive_job_t - */ - SEND_KEEPALIVE, - - /** - * Send a DPD packet. - * - * Job is implemented in class type send_dpd_job_t - */ - SEND_DPD -}; - -/** - * enum name for job_type_t - * - * @ingroup jobs - */ -extern enum_name_t *job_type_names; - /** * @brief Job-Interface as it is stored in the job queue. * - * A job consists of a job-type and one or more assigned values. - * * @b Constructors: * - None, use specific implementation of the interface. * @@ -134,32 +40,26 @@ extern enum_name_t *job_type_names; struct job_t { /** - * @brief get type of job. - * - * @param this calling object - * @return type of this job - */ - job_type_t (*get_type) (job_t *this); - - /** * @brief Execute a job. * - * Call the internall job routine to process the - * job. If this method returns DESTROY_ME, the job - * must be destroyed by the caller. + * The processing facility executes a job using this method. Jobs are + * one-shot, they destroy themself after execution, so don't use a job + * once it has been executed. * * @param this calling object - * @return status of job execution */ - status_t (*execute) (job_t *this); + void (*execute) (job_t *this); /** - * @brief Destroys a job_t object + * @brief Destroy a job. + * + * Is only called whenever a job was not executed (e.g. due daemon shutdown). + * After execution, jobs destroy themself. * * @param job_t calling object */ void (*destroy) (job_t *job); }; - #endif /* JOB_H_ */ + diff --git a/src/charon/processing/jobs/process_message_job.c b/src/charon/processing/jobs/process_message_job.c index ee7484bbd..6a0921248 100644 --- a/src/charon/processing/jobs/process_message_job.c +++ b/src/charon/processing/jobs/process_message_job.c @@ -44,17 +44,18 @@ struct private_process_message_job_t { }; /** - * Implements job_t.get_type. + * Implements job_t.destroy. */ -static job_type_t get_type(private_process_message_job_t *this) +static void destroy(private_process_message_job_t *this) { - return PROCESS_MESSAGE; + this->message->destroy(this->message); + free(this); } /** * Implementation of job_t.execute. */ -static status_t execute(private_process_message_job_t *this) +static void execute(private_process_message_job_t *this) { ike_sa_t *ike_sa; @@ -75,16 +76,7 @@ static status_t execute(private_process_message_job_t *this) charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa); } } - return DESTROY_ME; -} - -/** - * Implements job_t.destroy. - */ -static void destroy(private_process_message_job_t *this) -{ - this->message->destroy(this->message); - free(this); + destroy(this); } /* @@ -95,8 +87,7 @@ process_message_job_t *process_message_job_create(message_t *message) private_process_message_job_t *this = malloc_thing(private_process_message_job_t); /* interface functions */ - this->public.job_interface.get_type = (job_type_t (*) (job_t *)) get_type; - this->public.job_interface.execute = (status_t (*) (job_t *)) execute; + this->public.job_interface.execute = (void (*) (job_t *)) execute; this->public.job_interface.destroy = (void(*)(job_t*))destroy; /* private variables */ diff --git a/src/charon/processing/jobs/rekey_child_sa_job.c b/src/charon/processing/jobs/rekey_child_sa_job.c index 3422b614d..f754e5a1f 100644 --- a/src/charon/processing/jobs/rekey_child_sa_job.c +++ b/src/charon/processing/jobs/rekey_child_sa_job.c @@ -53,17 +53,17 @@ struct private_rekey_child_sa_job_t { }; /** - * Implementation of job_t.get_type. + * Implementation of job_t.destroy. */ -static job_type_t get_type(private_rekey_child_sa_job_t *this) +static void destroy(private_rekey_child_sa_job_t *this) { - return REKEY_CHILD_SA; + free(this); } /** * Implementation of job_t.execute. */ -static status_t execute(private_rekey_child_sa_job_t *this) +static void execute(private_rekey_child_sa_job_t *this) { ike_sa_t *ike_sa; @@ -73,20 +73,13 @@ static status_t execute(private_rekey_child_sa_job_t *this) { DBG2(DBG_JOB, "CHILD_SA with reqid %d not found for rekeying", this->reqid); - return DESTROY_ME; } - ike_sa->rekey_child_sa(ike_sa, this->protocol, this->spi); - - charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa); - return DESTROY_ME; -} - -/** - * Implementation of job_t.destroy. - */ -static void destroy(private_rekey_child_sa_job_t *this) -{ - free(this); + else + { + ike_sa->rekey_child_sa(ike_sa, this->protocol, this->spi); + charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa); + } + destroy(this); } /* @@ -99,8 +92,7 @@ rekey_child_sa_job_t *rekey_child_sa_job_create(u_int32_t reqid, private_rekey_child_sa_job_t *this = malloc_thing(private_rekey_child_sa_job_t); /* interface functions */ - this->public.job_interface.get_type = (job_type_t (*) (job_t *)) get_type; - this->public.job_interface.execute = (status_t (*) (job_t *)) execute; + this->public.job_interface.execute = (void (*) (job_t *)) execute; this->public.job_interface.destroy = (void (*)(job_t*)) destroy; /* private variables */ @@ -108,5 +100,5 @@ rekey_child_sa_job_t *rekey_child_sa_job_create(u_int32_t reqid, this->protocol = protocol; this->spi = spi; - return &(this->public); + return &this->public; } diff --git a/src/charon/processing/jobs/rekey_ike_sa_job.c b/src/charon/processing/jobs/rekey_ike_sa_job.c index f6c058634..020c3cce8 100644 --- a/src/charon/processing/jobs/rekey_ike_sa_job.c +++ b/src/charon/processing/jobs/rekey_ike_sa_job.c @@ -48,17 +48,18 @@ struct private_rekey_ike_sa_job_t { }; /** - * Implementation of job_t.get_type. + * Implementation of job_t.destroy. */ -static job_type_t get_type(private_rekey_ike_sa_job_t *this) +static void destroy(private_rekey_ike_sa_job_t *this) { - return REKEY_IKE_SA; + this->ike_sa_id->destroy(this->ike_sa_id); + free(this); } /** * Implementation of job_t.execute. */ -static status_t execute(private_rekey_ike_sa_job_t *this) +static void execute(private_rekey_ike_sa_job_t *this) { ike_sa_t *ike_sa; status_t status = SUCCESS; @@ -68,36 +69,28 @@ static status_t execute(private_rekey_ike_sa_job_t *this) if (ike_sa == NULL) { DBG2(DBG_JOB, "IKE_SA to rekey not found"); - return DESTROY_ME; - } - - if (this->reauth) - { - ike_sa->reestablish(ike_sa); } else { - status = ike_sa->rekey(ike_sa); - } - - if (status == DESTROY_ME) - { - charon->ike_sa_manager->checkin_and_destroy(charon->ike_sa_manager, ike_sa); - } - else - { - charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa); + if (this->reauth) + { + status = ike_sa->reestablish(ike_sa); + } + else + { + status = ike_sa->rekey(ike_sa); + } + + if (status == DESTROY_ME) + { + charon->ike_sa_manager->checkin_and_destroy(charon->ike_sa_manager, ike_sa); + } + else + { + charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa); + } } - return DESTROY_ME; -} - -/** - * Implementation of job_t.destroy. - */ -static void destroy(private_rekey_ike_sa_job_t *this) -{ - this->ike_sa_id->destroy(this->ike_sa_id); - free(this); + destroy(this); } /* @@ -108,8 +101,7 @@ rekey_ike_sa_job_t *rekey_ike_sa_job_create(ike_sa_id_t *ike_sa_id, bool reauth) private_rekey_ike_sa_job_t *this = malloc_thing(private_rekey_ike_sa_job_t); /* interface functions */ - this->public.job_interface.get_type = (job_type_t (*) (job_t *)) get_type; - this->public.job_interface.execute = (status_t (*) (job_t *)) execute; + this->public.job_interface.execute = (void (*) (job_t *)) execute; this->public.job_interface.destroy = (void (*)(job_t*)) destroy; /* private variables */ diff --git a/src/charon/processing/jobs/retransmit_job.c b/src/charon/processing/jobs/retransmit_job.c index 5bfa20dfd..8c15aa651 100644 --- a/src/charon/processing/jobs/retransmit_job.c +++ b/src/charon/processing/jobs/retransmit_job.c @@ -48,17 +48,18 @@ struct private_retransmit_job_t { }; /** - * Implements job_t.get_type. + * Implements job_t.destroy. */ -static job_type_t get_type(private_retransmit_job_t *this) +static void destroy(private_retransmit_job_t *this) { - return RETRANSMIT; + this->ike_sa_id->destroy(this->ike_sa_id); + free(this); } /** * Implementation of job_t.execute. */ -static status_t execute(private_retransmit_job_t *this) +static void execute(private_retransmit_job_t *this) { ike_sa_t *ike_sa; @@ -77,16 +78,7 @@ static status_t execute(private_retransmit_job_t *this) charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa); } } - return DESTROY_ME; -} - -/** - * Implements job_t.destroy. - */ -static void destroy(private_retransmit_job_t *this) -{ - this->ike_sa_id->destroy(this->ike_sa_id); - free(this); + destroy(this); } /* @@ -97,8 +89,7 @@ retransmit_job_t *retransmit_job_create(u_int32_t message_id,ike_sa_id_t *ike_sa private_retransmit_job_t *this = malloc_thing(private_retransmit_job_t); /* interface functions */ - this->public.job_interface.get_type = (job_type_t (*) (job_t *)) get_type; - this->public.job_interface.execute = (status_t (*) (job_t *)) execute; + this->public.job_interface.execute = (void (*) (job_t *)) execute; this->public.job_interface.destroy = (void (*) (job_t *)) destroy; /* private variables */ diff --git a/src/charon/processing/jobs/roam_job.c b/src/charon/processing/jobs/roam_job.c new file mode 100644 index 000000000..3b5cd0ed2 --- /dev/null +++ b/src/charon/processing/jobs/roam_job.c @@ -0,0 +1,115 @@ +/** + * @file roam_job.c + * + * @brief Implementation of roam_job_t. + * + */ + +/* + * Copyright (C) 2007 Martin Willi + * Hochschule fuer Technik Rapperswil + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2 of the License, or (at your + * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. + * + * This program is distributed in the hope that it will be useful, but + * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY + * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License + * for more details. + */ + + +#include <stdlib.h> + +#include "roam_job.h" + +#include <sa/ike_sa.h> +#include <daemon.h> + + +typedef struct private_roam_job_t private_roam_job_t; + +/** + * Private data of an roam_job_t Object + */ +struct private_roam_job_t { + /** + * public roam_job_t interface + */ + roam_job_t public; + + /** + * has the address list changed, or the routing only? + */ + bool address; +}; + +/** + * Implements job_t.destroy. + */ +static void destroy(private_roam_job_t *this) +{ + free(this); +} + +/** + * Implementation of job_t.execute. + */ +static void execute(private_roam_job_t *this) +{ + ike_sa_t *ike_sa; + linked_list_t *list; + ike_sa_id_t *id; + iterator_t *iterator; + + /* iterating over all IKE_SAs gives us no way to checkin_and_destroy + * after a DESTROY_ME, so we check out each available IKE_SA by hand. */ + list = linked_list_create(); + iterator = charon->ike_sa_manager->create_iterator(charon->ike_sa_manager); + while (iterator->iterate(iterator, (void**)&ike_sa)) + { + id = ike_sa->get_id(ike_sa); + list->insert_last(list, id->clone(id)); + } + iterator->destroy(iterator); + + while (list->remove_last(list, (void**)&id) == SUCCESS) + { + ike_sa = charon->ike_sa_manager->checkout(charon->ike_sa_manager, id); + if (ike_sa) + { + if (ike_sa->roam(ike_sa, this->address) == DESTROY_ME) + { + charon->ike_sa_manager->checkin_and_destroy( + charon->ike_sa_manager, ike_sa); + } + else + { + charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa); + } + } + id->destroy(id); + } + list->destroy(list); + + destroy(this); +} + +/* + * Described in header + */ +roam_job_t *roam_job_create(bool address) +{ + private_roam_job_t *this = malloc_thing(private_roam_job_t); + + this->public.job_interface.destroy = (void (*) (job_t *)) destroy; + this->public.job_interface.execute = (void (*) (job_t *)) execute; + this->public.job_interface.destroy = (void (*) (job_t *)) destroy; + + this->address = address; + + return &this->public; +} + diff --git a/src/charon/processing/jobs/roam_job.h b/src/charon/processing/jobs/roam_job.h new file mode 100644 index 000000000..293b09f08 --- /dev/null +++ b/src/charon/processing/jobs/roam_job.h @@ -0,0 +1,61 @@ +/** + * @file roam_job.h + * + * @brief Interface of roam_job_t. + */ + +/* + * Copyright (C) 2007 Martin Willi + * Hochschule fuer Technik Rapperswil + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2 of the License, or (at your + * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. + * + * This program is distributed in the hope that it will be useful, but + * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY + * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License + * for more details. + */ + +#ifndef ROAM_JOB_H_ +#define ROAM_JOB_H_ + +typedef struct roam_job_t roam_job_t; + +#include <library.h> +#include <sa/ike_sa_id.h> +#include <processing/jobs/job.h> + +/** + * @brief A job to inform IKE_SAs about changed local address setup. + * + * If a local address appears or disappears, the kernel fires this job to + * update all IKE_SAs. + * + * @b Constructors: + * - roam_job_create() + * + * @ingroup jobs + */ +struct roam_job_t { + + /** + * implements job_t interface + */ + job_t job_interface; +}; + +/** + * @brief Creates a job to inform IKE_SAs about an updated address list. + * + * @param address TRUE if address list changed, FALSE if routing changed + * @return initiate_ike_sa_job_t object + * + * @ingroup jobs + */ +roam_job_t *roam_job_create(bool address); + +#endif /*ROAM_JOB_H_*/ + diff --git a/src/charon/processing/jobs/send_dpd_job.c b/src/charon/processing/jobs/send_dpd_job.c index 7294d78d5..f6786bfb4 100644 --- a/src/charon/processing/jobs/send_dpd_job.c +++ b/src/charon/processing/jobs/send_dpd_job.c @@ -47,45 +47,35 @@ struct private_send_dpd_job_t { }; /** - * Implements send_dpd_job_t.get_type. + * Implements job_t.destroy. */ -static job_type_t get_type(private_send_dpd_job_t *this) +static void destroy(private_send_dpd_job_t *this) { - return SEND_DPD; + this->ike_sa_id->destroy(this->ike_sa_id); + free(this); } /** * Implementation of job_t.execute. */ -static status_t execute(private_send_dpd_job_t *this) +static void execute(private_send_dpd_job_t *this) { ike_sa_t *ike_sa; ike_sa = charon->ike_sa_manager->checkout(charon->ike_sa_manager, this->ike_sa_id); - if (ike_sa == NULL) - { - return DESTROY_ME; - } - - if (ike_sa->send_dpd(ike_sa) == DESTROY_ME) + if (ike_sa) { - charon->ike_sa_manager->checkin_and_destroy(charon->ike_sa_manager, ike_sa); + if (ike_sa->send_dpd(ike_sa) == DESTROY_ME) + { + charon->ike_sa_manager->checkin_and_destroy(charon->ike_sa_manager, ike_sa); + } + else + { + charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa); + } } - else - { - charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa); - } - return DESTROY_ME; -} - -/** - * Implements job_t.destroy. - */ -static void destroy(private_send_dpd_job_t *this) -{ - this->ike_sa_id->destroy(this->ike_sa_id); - free(this); + destroy(this); } /* @@ -96,15 +86,12 @@ send_dpd_job_t *send_dpd_job_create(ike_sa_id_t *ike_sa_id) private_send_dpd_job_t *this = malloc_thing(private_send_dpd_job_t); /* interface functions */ - this->public.job_interface.get_type = (job_type_t (*) (job_t *)) get_type; this->public.job_interface.destroy = (void (*) (job_t *)) destroy; - this->public.job_interface.execute = (status_t (*) (job_t *)) execute; - - /* public functions */ - this->public.destroy = (void (*)(send_dpd_job_t *)) destroy; + this->public.job_interface.execute = (void (*) (job_t *)) execute; + this->public.job_interface.destroy = (void (*) (job_t *)) destroy; /* private variables */ this->ike_sa_id = ike_sa_id->clone(ike_sa_id); - return &(this->public); + return &this->public; } diff --git a/src/charon/processing/jobs/send_dpd_job.h b/src/charon/processing/jobs/send_dpd_job.h index 26c9e2e81..0e4059131 100644 --- a/src/charon/processing/jobs/send_dpd_job.h +++ b/src/charon/processing/jobs/send_dpd_job.h @@ -45,13 +45,6 @@ struct send_dpd_job_t { * implements job_t interface */ job_t job_interface; - - /** - * @brief Destroys an send_dpd_job_t object. - * - * @param this send_dpd_job_t object to destroy - */ - void (*destroy) (send_dpd_job_t *this); }; /** diff --git a/src/charon/processing/jobs/send_keepalive_job.c b/src/charon/processing/jobs/send_keepalive_job.c index 1c1cb288e..8cb51e5dd 100644 --- a/src/charon/processing/jobs/send_keepalive_job.c +++ b/src/charon/processing/jobs/send_keepalive_job.c @@ -47,38 +47,29 @@ struct private_send_keepalive_job_t { }; /** - * Implements send_keepalive_job_t.get_type. + * Implements job_t.destroy. */ -static job_type_t get_type(private_send_keepalive_job_t *this) +static void destroy(private_send_keepalive_job_t *this) { - return SEND_KEEPALIVE; + this->ike_sa_id->destroy(this->ike_sa_id); + free(this); } /** * Implementation of job_t.execute. */ -static status_t execute(private_send_keepalive_job_t *this) +static void execute(private_send_keepalive_job_t *this) { ike_sa_t *ike_sa; ike_sa = charon->ike_sa_manager->checkout(charon->ike_sa_manager, this->ike_sa_id); - if (ike_sa == NULL) + if (ike_sa) { - return DESTROY_ME; + ike_sa->send_keepalive(ike_sa); + charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa); } - ike_sa->send_keepalive(ike_sa); - charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa); - return DESTROY_ME; -} - -/** - * Implements job_t.destroy. - */ -static void destroy(private_send_keepalive_job_t *this) -{ - this->ike_sa_id->destroy(this->ike_sa_id); - free(this); + destroy(this); } /* @@ -89,15 +80,12 @@ send_keepalive_job_t *send_keepalive_job_create(ike_sa_id_t *ike_sa_id) private_send_keepalive_job_t *this = malloc_thing(private_send_keepalive_job_t); /* interface functions */ - this->public.job_interface.get_type = (job_type_t (*) (job_t *)) get_type; this->public.job_interface.destroy = (void (*) (job_t *)) destroy; - this->public.job_interface.execute = (status_t (*) (job_t *)) execute; - - /* public functions */ - this->public.destroy = (void (*)(send_keepalive_job_t *)) destroy; + this->public.job_interface.execute = (void (*) (job_t *)) execute; + this->public.job_interface.destroy = (void (*) (job_t *)) destroy; /* private variables */ this->ike_sa_id = ike_sa_id->clone(ike_sa_id); - return &(this->public); + return &this->public; } diff --git a/src/charon/processing/jobs/send_keepalive_job.h b/src/charon/processing/jobs/send_keepalive_job.h index f7b38337e..e8d214aed 100644 --- a/src/charon/processing/jobs/send_keepalive_job.h +++ b/src/charon/processing/jobs/send_keepalive_job.h @@ -44,13 +44,6 @@ struct send_keepalive_job_t { * implements job_t interface */ job_t job_interface; - - /** - * @brief Destroys an send_keepalive_job_t object. - * - * @param this send_keepalive_job_t object to destroy - */ - void (*destroy) (send_keepalive_job_t *this); }; /** diff --git a/src/charon/processing/processor.c b/src/charon/processing/processor.c new file mode 100644 index 000000000..b3815eeb1 --- /dev/null +++ b/src/charon/processing/processor.c @@ -0,0 +1,233 @@ +/** + * @file processor.c + * + * @brief Implementation of processor_t. + * + */ + +/* + * Copyright (C) 2005-2007 Martin Willi + * Copyright (C) 2005 Jan Hutter + * Hochschule fuer Technik Rapperswil + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2 of the License, or (at your + * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. + * + * This program is distributed in the hope that it will be useful, but + * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY + * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License + * for more details. + */ + +#include <stdlib.h> +#include <pthread.h> +#include <string.h> +#include <errno.h> + +#include "processor.h" + +#include <daemon.h> +#include <utils/linked_list.h> + + +typedef struct private_processor_t private_processor_t; + +/** + * @brief Private data of processor_t class. + */ +struct private_processor_t { + /** + * Public processor_t interface. + */ + processor_t public; + + /** + * Number of running threads + */ + u_int total_threads; + + /** + * Desired number of threads + */ + u_int desired_threads; + + /** + * Number of threads waiting for work + */ + u_int idle_threads; + + /** + * The jobs are stored in a linked list + */ + linked_list_t *list; + + /** + * access to linked_list is locked through this mutex + */ + pthread_mutex_t mutex; + + /** + * Condvar to wait for new jobs + */ + pthread_cond_t condvar; +}; + +static void process_jobs(private_processor_t *this); + +/** + * restart a terminated thread + */ +static void restart(private_processor_t *this) +{ + pthread_t thread; + + if (pthread_create(&thread, NULL, (void*)process_jobs, this) != 0) + { + this->total_threads--; + } +} + +/** + * Process queued jobs, called by the worker threads + */ +static void process_jobs(private_processor_t *this) +{ + int oldstate; + + pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &oldstate); + + DBG2(DBG_JOB, "started worker thread, thread_ID: %06u", (int)pthread_self()); + + pthread_mutex_lock(&this->mutex); + while (this->desired_threads >= this->total_threads) + { + job_t *job; + + if (this->list->get_count(this->list) == 0) + { + this->idle_threads++; + pthread_cond_wait(&this->condvar, &this->mutex); + this->idle_threads--; + continue; + } + this->list->remove_first(this->list, (void**)&job); + pthread_mutex_unlock(&this->mutex); + /* terminated threads are restarted, so we have a constant pool */ + pthread_cleanup_push((void*)restart, this); + job->execute(job); + pthread_cleanup_pop(0); + pthread_mutex_lock(&this->mutex); + } + this->total_threads--; + pthread_cond_broadcast(&this->condvar); + pthread_mutex_unlock(&this->mutex); +} + +/** + * Implementation of processor_t.get_total_threads. + */ +static u_int get_total_threads(private_processor_t *this) +{ + return this->total_threads; +} + +/** + * Implementation of processor_t.get_idle_threads. + */ +static u_int get_idle_threads(private_processor_t *this) +{ + return this->idle_threads; +} + +/** + * implements processor_t.get_job_load + */ +static u_int get_job_load(private_processor_t *this) +{ + u_int load; + pthread_mutex_lock(&this->mutex); + load = this->list->get_count(this->list); + pthread_mutex_unlock(&this->mutex); + return load; +} + +/** + * implements function processor_t.queue_job + */ +static void queue_job(private_processor_t *this, job_t *job) +{ + pthread_mutex_lock(&this->mutex); + this->list->insert_last(this->list, job); + pthread_mutex_unlock(&this->mutex); + pthread_cond_signal(&this->condvar); +} + +/** + * Implementation of processor_t.set_threads. + */ +static void set_threads(private_processor_t *this, u_int count) +{ + pthread_mutex_lock(&this->mutex); + if (count > this->total_threads) + { /* increase thread count */ + int i; + pthread_t current; + + this->desired_threads = count; + DBG1(DBG_JOB, "spawning %d worker threads", count - this->total_threads); + for (i = this->total_threads; i < count; i++) + { + if (pthread_create(¤t, NULL, (void*)process_jobs, this) == 0) + { + this->total_threads++; + } + } + } + else if (count < this->total_threads) + { /* decrease thread count */ + this->desired_threads = count; + } + pthread_mutex_unlock(&this->mutex); +} + +/** + * Implementation of processor_t.destroy. + */ +static void destroy(private_processor_t *this) +{ + set_threads(this, 0); + while (this->total_threads > 0) + { + pthread_cond_broadcast(&this->condvar); + pthread_cond_wait(&this->condvar, &this->mutex); + } + this->list->destroy_offset(this->list, offsetof(job_t, destroy)); + free(this); +} + +/* + * Described in header. + */ +processor_t *processor_create(size_t pool_size) +{ + private_processor_t *this = malloc_thing(private_processor_t); + + this->public.get_total_threads = (u_int(*)(processor_t*))get_total_threads; + this->public.get_idle_threads = (u_int(*)(processor_t*))get_idle_threads; + this->public.get_job_load = (u_int(*)(processor_t*))get_job_load; + this->public.queue_job = (void(*)(processor_t*, job_t*))queue_job; + this->public.set_threads = (void(*)(processor_t*, u_int))set_threads; + this->public.destroy = (void(*)(processor_t*))destroy; + + this->list = linked_list_create(); + pthread_mutex_init(&this->mutex, NULL); + pthread_cond_init(&this->condvar, NULL); + this->total_threads = 0; + this->desired_threads = 0; + this->idle_threads = 0; + + return &this->public; +} + diff --git a/src/charon/processing/processor.h b/src/charon/processing/processor.h new file mode 100644 index 000000000..f12c7f10e --- /dev/null +++ b/src/charon/processing/processor.h @@ -0,0 +1,111 @@ +/** + * @file processor.h + * + * @brief Interface of processor_t. + * + */ + +/* + * Copyright (C) 2005-2007 Martin Willi + * Copyright (C) 2005 Jan Hutter + * Hochschule fuer Technik Rapperswil + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2 of the License, or (at your + * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. + * + * This program is distributed in the hope that it will be useful, but + * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY + * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License + * for more details. + */ + +#ifndef PROCESSOR_H_ +#define PROCESSOR_H_ + +typedef struct processor_t processor_t; + +#include <stdlib.h> + +#include <library.h> +#include <processing/jobs/job.h> + +/** + * @brief The processor uses threads to process queued jobs. + * + * @b Constructors: + * - processor_create() + * + * @ingroup processing + */ +struct processor_t { + + /** + * @brief Get the total number of threads used by the processor. + * + * @param this calling object + * @return size of thread pool + */ + u_int (*get_total_threads) (processor_t *this); + + /** + * @brief Get the number of threads currently waiting. + * + * @param this calling object + * @return number of idle threads + */ + u_int (*get_idle_threads) (processor_t *this); + + /** + * @brief Get the number of queued jobs. + * + * @param this calling object + * @returns number of items in queue + */ + u_int (*get_job_load) (processor_t *this); + + /** + * @brief Adds a job to the queue. + * + * This function is non blocking and adds a job_t to the queue. + * + * @param this calling object + * @param job job to add to the queue + */ + void (*queue_job) (processor_t *this, job_t *job); + + /** + * @brief Set the number of threads to use in the processor. + * + * If the number of threads is smaller than number of currently running + * threads, thread count is decreased. Use 0 to disable the processor. + * This call blocks if it decreases thread count until threads have + * terminated, so make sure there are not too many blocking jobs. + * + * @param this calling object + * @param count number of threads to allocate + */ + void (*set_threads)(processor_t *this, u_int count); + + /** + * @brief Destroy a processor object. + * + * @param processor calling object + */ + void (*destroy) (processor_t *processor); +}; + +/** + * @brief Create the thread pool without any threads. + * + * Use the set_threads method to start processing jobs. + * + * @return processor_t object + * + * @ingroup processing + */ +processor_t *processor_create(); + +#endif /*PROCESSOR_H_*/ + diff --git a/src/charon/processing/scheduler.c b/src/charon/processing/scheduler.c index 7249e43e6..2706585b0 100644 --- a/src/charon/processing/scheduler.c +++ b/src/charon/processing/scheduler.c @@ -23,12 +23,39 @@ #include <stdlib.h> #include <pthread.h> +#include <sys/time.h> #include "scheduler.h" #include <daemon.h> -#include <processing/job_queue.h> +#include <processing/processor.h> +#include <processing/jobs/callback_job.h> +typedef struct event_t event_t; + +/** + * Event containing a job and a schedule time + */ +struct event_t { + /** + * Time to fire the event. + */ + timeval_t time; + + /** + * Every event has its assigned job. + */ + job_t *job; +}; + +/** + * destroy an event and its job + */ +static void event_destroy(event_t *event) +{ + event->job->destroy(event->job); + free(event); +} typedef struct private_scheduler_t private_scheduler_t; @@ -42,36 +69,164 @@ struct private_scheduler_t { scheduler_t public; /** - * Assigned thread. + * Job wich schedules */ - pthread_t assigned_thread; + callback_job_t *job; + + /** + * The jobs are scheduled in a list. + */ + linked_list_t *list; + + /** + * Exclusive access to list + */ + pthread_mutex_t mutex; + + /** + * Condvar to wait for next job. + */ + pthread_cond_t condvar; }; /** - * Implementation of private_scheduler_t.get_events. + * Returns the difference of two timeval structs in milliseconds + */ +static long time_difference(timeval_t *end, timeval_t *start) +{ + time_t s; + suseconds_t us; + + s = end->tv_sec - start->tv_sec; + us = end->tv_usec - start->tv_usec; + return (s * 1000 + us/1000); +} + +/** + * Get events from the queue and pass it to the processor */ -static void get_events(private_scheduler_t * this) +static job_requeue_t schedule(private_scheduler_t * this) { - job_t *current_job; + timespec_t timeout; + timeval_t now; + event_t *event; + long difference; + int oldstate; + bool timed = FALSE; - /* cancellation disabled by default */ - pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL); + DBG2(DBG_JOB, "waiting for next event..."); + pthread_mutex_lock(&this->mutex); - DBG1(DBG_JOB, "scheduler thread running, thread_ID: %06u", - (int)pthread_self()); + gettimeofday(&now, NULL); + + if (this->list->get_count(this->list) > 0) + { + this->list->get_first(this->list, (void **)&event); + difference = time_difference(&now, &event->time); + 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); + 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; + timed = TRUE; + } + pthread_cleanup_push((void*)pthread_mutex_unlock, &this->mutex); + pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &oldstate); + + if (timed) + { + pthread_cond_timedwait(&this->condvar, &this->mutex, &timeout); + } + else + { + pthread_cond_wait(&this->condvar, &this->mutex); + } + pthread_setcancelstate(oldstate, NULL); + pthread_cleanup_pop(0); + + pthread_mutex_unlock(&this->mutex); + return JOB_REQUEUE_DIRECT; +} - charon->drop_capabilities(charon, TRUE); +/** + * Implements scheduler_t.get_job_load + */ +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); + return count; +} - while (TRUE) +/** + * Implements scheduler_t.schedule_job. + */ +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; + time_t s; + suseconds_t us; + + event = malloc_thing(event_t); + event->job = job; + + /* calculate absolute time */ + s = time / 1000; + us = (time - s * 1000) * 1000; + gettimeofday(&now, NULL); + 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) { - DBG2(DBG_JOB, "waiting for next event..."); - /* get a job, this block until one is available */ - current_job = charon->event_queue->get(charon->event_queue); - /* queue the job in the job queue, workers will eat them */ - DBG2(DBG_JOB, "got event, adding job %N to job-queue", - job_type_names, current_job->get_type(current_job)); - charon->job_queue->add(charon->job_queue, current_job); + 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; } + pthread_cond_signal(&this->condvar); + pthread_mutex_unlock(&this->mutex); } /** @@ -79,8 +234,8 @@ static void get_events(private_scheduler_t * this) */ static void destroy(private_scheduler_t *this) { - pthread_cancel(this->assigned_thread); - pthread_join(this->assigned_thread, NULL); + this->job->cancel(this->job); + this->list->destroy_function(this->list, (void*)event_destroy); free(this); } @@ -91,14 +246,17 @@ scheduler_t * scheduler_create() { private_scheduler_t *this = malloc_thing(private_scheduler_t); + this->public.get_job_load = (u_int (*) (scheduler_t *this)) get_job_load; + this->public.schedule_job = (void (*) (scheduler_t *this, job_t *job, u_int32_t ms)) schedule_job; this->public.destroy = (void(*)(scheduler_t*)) destroy; - if (pthread_create(&(this->assigned_thread), NULL, (void*(*)(void*))get_events, this) != 0) - { - /* thread could not be created */ - free(this); - charon->kill(charon, "unable to create scheduler thread"); - } + this->list = linked_list_create(); + pthread_mutex_init(&this->mutex, NULL); + pthread_cond_init(&this->condvar, NULL); + + this->job = callback_job_create((callback_job_cb_t)schedule, this, NULL, NULL); + charon->processor->queue_job(charon->processor, (job_t*)this->job); - return &(this->public); + return &this->public; } + diff --git a/src/charon/processing/scheduler.h b/src/charon/processing/scheduler.h index bea93e7c9..7bde6e638 100644 --- a/src/charon/processing/scheduler.h +++ b/src/charon/processing/scheduler.h @@ -6,7 +6,7 @@ */ /* - * Copyright (C) 2005-2006 Martin Willi + * Copyright (C) 2005-2007 Martin Willi * Copyright (C) 2005 Jan Hutter * Hochschule fuer Technik Rapperswil * @@ -27,14 +27,12 @@ typedef struct scheduler_t scheduler_t; #include <library.h> +#include <processing/jobs/job.h> /** - * @brief The scheduler thread is responsible for timed events. + * @brief The scheduler queues and executes timed events. * - * The scheduler thread takes out jobs from the event-queue and adds them - * to the job-queue. - * - * Starts a thread which does the work, since event-queue is blocking. + * The scheduler stores timed events and passes them to the processor. * * @b Constructors: * - scheduler_create() @@ -44,25 +42,40 @@ typedef struct scheduler_t scheduler_t; struct scheduler_t { /** + * @brief Adds a event to the queue, using a relative time offset. + * + * Schedules a job for execution using a relative time offset. + * + * @param this calling object + * @param job job to schedule + * @param time relative to to schedule job (in ms) + */ + void (*schedule_job) (scheduler_t *this, job_t *job, u_int32_t time); + + /** + * @brief Returns number of jobs scheduled. + * + * @param this calling object + * @return number of scheduled jobs + */ + u_int (*get_job_load) (scheduler_t *this); + + /** * @brief Destroys a scheduler object. * - * @param scheduler calling object + * @param this calling object */ - void (*destroy) (scheduler_t *scheduler); + void (*destroy) (scheduler_t *this); }; /** - * @brief Create a scheduler with its associated thread. - * - * The thread will start to get jobs form the event queue - * and adds them to the job queue. + * @brief Create a scheduler. * - * @return - * - scheduler_t object - * - NULL if thread could not be started + * @return scheduler_t object * * @ingroup processing */ -scheduler_t * scheduler_create(void); +scheduler_t *scheduler_create(void); #endif /*SCHEDULER_H_*/ + diff --git a/src/charon/processing/thread_pool.c b/src/charon/processing/thread_pool.c deleted file mode 100644 index a9891da15..000000000 --- a/src/charon/processing/thread_pool.c +++ /dev/null @@ -1,183 +0,0 @@ -/** - * @file thread_pool.c - * - * @brief Implementation of thread_pool_t. - * - */ - -/* - * Copyright (C) 2005-2006 Martin Willi - * Copyright (C) 2005 Jan Hutter - * Hochschule fuer Technik Rapperswil - * - * This program is free software; you can redistribute it and/or modify it - * under the terms of the GNU General Public License as published by the - * Free Software Foundation; either version 2 of the License, or (at your - * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. - * - * This program is distributed in the hope that it will be useful, but - * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY - * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License - * for more details. - */ - -#include <stdlib.h> -#include <pthread.h> -#include <string.h> -#include <errno.h> - -#include "thread_pool.h" - -#include <daemon.h> -#include <processing/job_queue.h> - - -typedef struct private_thread_pool_t private_thread_pool_t; - -/** - * @brief Private data of thread_pool_t class. - */ -struct private_thread_pool_t { - /** - * Public thread_pool_t interface. - */ - thread_pool_t public; - - /** - * Number of running threads. - */ - u_int pool_size; - - /** - * Number of threads waiting for work - */ - u_int idle_threads; - - /** - * Array of thread ids. - */ - pthread_t *threads; -}; - -/** - * Implementation of private_thread_pool_t.process_jobs. - */ -static void process_jobs(private_thread_pool_t *this) -{ - job_t *job; - status_t status; - - /* cancellation disabled by default */ - pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL); - - DBG1(DBG_JOB, "worker thread running, thread_ID: %06u", - (int)pthread_self()); - - charon->drop_capabilities(charon, TRUE); - - while (TRUE) - { - /* TODO: should be atomic, but is not mission critical */ - this->idle_threads++; - job = charon->job_queue->get(charon->job_queue); - this->idle_threads--; - - status = job->execute(job); - - if (status == DESTROY_ME) - { - job->destroy(job); - } - } -} - -/** - * Implementation of thread_pool_t.get_pool_size. - */ -static u_int get_pool_size(private_thread_pool_t *this) -{ - return this->pool_size; -} - -/** - * Implementation of thread_pool_t.get_idle_threads. - */ -static u_int get_idle_threads(private_thread_pool_t *this) -{ - return this->idle_threads; -} - -/** - * Implementation of thread_pool_t.destroy. - */ -static void destroy(private_thread_pool_t *this) -{ - int current; - /* flag thread for termination */ - for (current = 0; current < this->pool_size; current++) - { - DBG1(DBG_JOB, "cancelling worker thread #%d", current+1); - pthread_cancel(this->threads[current]); - } - - /* wait for all threads */ - for (current = 0; current < this->pool_size; current++) { - if (pthread_join(this->threads[current], NULL) == 0) - { - DBG1(DBG_JOB, "worker thread #%d terminated", current+1); - } - else - { - DBG1(DBG_JOB, "could not terminate worker thread #%d", current+1); - } - } - - /* free mem */ - free(this->threads); - free(this); -} - -/* - * Described in header. - */ -thread_pool_t *thread_pool_create(size_t pool_size) -{ - int current; - private_thread_pool_t *this = malloc_thing(private_thread_pool_t); - - /* fill in public fields */ - this->public.destroy = (void(*)(thread_pool_t*))destroy; - this->public.get_pool_size = (u_int(*)(thread_pool_t*))get_pool_size; - this->public.get_idle_threads = (u_int(*)(thread_pool_t*))get_idle_threads; - - /* initialize member */ - this->pool_size = pool_size; - this->idle_threads = 0; - this->threads = malloc(sizeof(pthread_t) * pool_size); - - /* try to create as many threads as possible, up to pool_size */ - for (current = 0; current < pool_size; current++) - { - if (pthread_create(&(this->threads[current]), NULL, - (void*(*)(void*))process_jobs, this) == 0) - { - DBG1(DBG_JOB, "created worker thread #%d", current+1); - } - else - { - /* creation failed, is it the first one? */ - if (current == 0) - { - free(this->threads); - free(this); - charon->kill(charon, "could not create any worker threads"); - } - /* not all threads could be created, but at least one :-/ */ - DBG1(DBG_JOB, "could only create %d from requested %d threads!", - current, pool_size); - this->pool_size = current; - break; - } - } - return (thread_pool_t*)this; -} diff --git a/src/charon/processing/thread_pool.h b/src/charon/processing/thread_pool.h deleted file mode 100644 index 09a6312a8..000000000 --- a/src/charon/processing/thread_pool.h +++ /dev/null @@ -1,87 +0,0 @@ -/** - * @file thread_pool.h - * - * @brief Interface of thread_pool_t. - * - */ - -/* - * Copyright (C) 2005-2006 Martin Willi - * Copyright (C) 2005 Jan Hutter - * Hochschule fuer Technik Rapperswil - * - * This program is free software; you can redistribute it and/or modify it - * under the terms of the GNU General Public License as published by the - * Free Software Foundation; either version 2 of the License, or (at your - * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. - * - * This program is distributed in the hope that it will be useful, but - * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY - * or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License - * for more details. - */ - -#ifndef THREAD_POOL_H_ -#define THREAD_POOL_H_ - -typedef struct thread_pool_t thread_pool_t; - -#include <stdlib.h> - -#include <library.h> - -/** - * @brief A thread_pool consists of a pool of threads processing jobs from the job queue. - * - * Current implementation uses as many threads as specified in constructor. - * A more improved version would dynamically increase thread count if necessary. - * - * @b Constructors: - * - thread_pool_create() - * - * @todo Add support for dynamic thread handling - * - * @ingroup processing - */ -struct thread_pool_t { - - /** - * @brief Return currently instanciated thread count. - * - * @param thread_pool calling object - * @return size of thread pool - */ - u_int (*get_pool_size) (thread_pool_t *thread_pool); - - /** - * @brief Get the number of threads currently waiting for work. - * - * @param thread_pool calling object - * @return number of idle threads - */ - u_int (*get_idle_threads) (thread_pool_t *thread_pool); - - /** - * @brief Destroy a thread_pool_t object. - * - * Sends cancellation request to all threads and AWAITS their termination. - * - * @param thread_pool calling object - */ - void (*destroy) (thread_pool_t *thread_pool); -}; - -/** - * @brief Create the thread pool using using pool_size of threads. - * - * @param pool_size desired pool size - * @return - * - thread_pool_t object if one ore more threads could be started, or - * - NULL if no threads could be created - * - * @ingroup processing - */ -thread_pool_t *thread_pool_create(size_t pool_size); - - -#endif /*THREAD_POOL_H_*/ |