summaryrefslogtreecommitdiff
path: root/src/libstrongswan/processing/processor.c
diff options
context:
space:
mode:
authorYves-Alexis Perez <corsac@debian.org>2013-01-02 14:18:20 +0100
committerYves-Alexis Perez <corsac@debian.org>2013-01-02 14:18:20 +0100
commitc1343b3278cdf99533b7902744d15969f9d6fdc1 (patch)
treed5ed3dc5677a59260ec41cd39bb284d3e94c91b3 /src/libstrongswan/processing/processor.c
parentb34738ed08c2227300d554b139e2495ca5da97d6 (diff)
downloadvyos-strongswan-c1343b3278cdf99533b7902744d15969f9d6fdc1.tar.gz
vyos-strongswan-c1343b3278cdf99533b7902744d15969f9d6fdc1.zip
Imported Upstream version 5.0.1
Diffstat (limited to 'src/libstrongswan/processing/processor.c')
-rw-r--r--src/libstrongswan/processing/processor.c219
1 files changed, 166 insertions, 53 deletions
diff --git a/src/libstrongswan/processing/processor.c b/src/libstrongswan/processing/processor.c
index 222f1a535..5b7fd467c 100644
--- a/src/libstrongswan/processing/processor.c
+++ b/src/libstrongswan/processing/processor.c
@@ -1,7 +1,7 @@
/*
* Copyright (C) 2005-2011 Martin Willi
* Copyright (C) 2011 revosec AG
- * Copyright (C) 2008-2011 Tobias Brunner
+ * Copyright (C) 2008-2012 Tobias Brunner
* Copyright (C) 2005 Jan Hutter
* Hochschule fuer Technik Rapperswil
*
@@ -58,7 +58,7 @@ struct private_processor_t {
/**
* All threads managed in the pool (including threads that have been
- * cancelled, this allows to join them during destruction)
+ * canceled, this allows to join them later), as worker_thread_t
*/
linked_list_t *threads;
@@ -73,11 +73,6 @@ struct private_processor_t {
int prio_threads[JOB_PRIO_MAX];
/**
- * Priority of the job executed by a thread
- */
- thread_value_t *priority;
-
- /**
* access to job lists is locked through this mutex
*/
mutex_t *mutex;
@@ -93,39 +88,71 @@ struct private_processor_t {
condvar_t *thread_terminated;
};
-static void process_jobs(private_processor_t *this);
+/**
+ * Worker thread
+ */
+typedef struct {
+
+ /**
+ * Reference to the processor
+ */
+ private_processor_t *processor;
+
+ /**
+ * The actual thread
+ */
+ thread_t *thread;
+
+ /**
+ * Job currently being executed by this worker thread
+ */
+ job_t *job;
+
+ /**
+ * Priority of the current job
+ */
+ job_priority_t priority;
+
+} worker_thread_t;
+
+static void process_jobs(worker_thread_t *worker);
/**
* restart a terminated thread
*/
-static void restart(private_processor_t *this)
+static void restart(worker_thread_t *worker)
{
- thread_t *thread;
+ private_processor_t *this = worker->processor;
DBG2(DBG_JOB, "terminated worker thread %.2u", thread_current_id());
- /* respawn thread if required */
this->mutex->lock(this->mutex);
- if (this->desired_threads < this->total_threads ||
- (thread = thread_create((thread_main_t)process_jobs, this)) == NULL)
- {
- this->total_threads--;
- this->thread_terminated->signal(this->thread_terminated);
- }
- else
+ /* cleanup worker thread */
+ this->working_threads[worker->priority]--;
+ worker->job->status = JOB_STATUS_CANCELED;
+ worker->job->destroy(worker->job);
+ worker->job = NULL;
+
+ /* respawn thread if required */
+ if (this->desired_threads >= this->total_threads)
{
- this->threads->insert_last(this->threads, thread);
+ worker_thread_t *new_worker;
+
+ INIT(new_worker,
+ .processor = this,
+ );
+ new_worker->thread = thread_create((thread_main_t)process_jobs,
+ new_worker);
+ if (new_worker->thread)
+ {
+ this->threads->insert_last(this->threads, new_worker);
+ this->mutex->unlock(this->mutex);
+ return;
+ }
+ free(new_worker);
}
- this->mutex->unlock(this->mutex);
-}
-
-/**
- * Decrement working thread count of a priority class
- */
-static void decrement_working_threads(private_processor_t *this)
-{
- this->mutex->lock(this->mutex);
- this->working_threads[(intptr_t)this->priority->get(this->priority)]--;
+ this->total_threads--;
+ this->thread_terminated->signal(this->thread_terminated);
this->mutex->unlock(this->mutex);
}
@@ -147,9 +174,11 @@ static u_int get_idle_threads_nolock(private_processor_t *this)
/**
* Process queued jobs, called by the worker threads
*/
-static void process_jobs(private_processor_t *this)
+static void process_jobs(worker_thread_t *worker)
{
- /* worker threads are not cancellable by default */
+ private_processor_t *this = worker->processor;
+
+ /* worker threads are not cancelable by default */
thread_cancelability(FALSE);
DBG2(DBG_JOB, "started worker thread %.2u", thread_current_id());
@@ -157,7 +186,6 @@ static void process_jobs(private_processor_t *this)
this->mutex->lock(this->mutex);
while (this->desired_threads >= this->total_threads)
{
- job_t *job = NULL;
int i, reserved = 0, idle;
idle = get_idle_threads_nolock(this);
@@ -176,27 +204,80 @@ static void process_jobs(private_processor_t *this)
reserved += this->prio_threads[i] - this->working_threads[i];
}
if (this->jobs[i]->remove_first(this->jobs[i],
- (void**)&job) == SUCCESS)
+ (void**)&worker->job) == SUCCESS)
{
+ job_requeue_t requeue;
+
this->working_threads[i]++;
+ worker->job->status = JOB_STATUS_EXECUTING;
+ worker->priority = i;
this->mutex->unlock(this->mutex);
- this->priority->set(this->priority, (void*)(intptr_t)i);
- /* terminated threads are restarted to get a constant pool */
- thread_cleanup_push((thread_cleanup_t)restart, this);
- thread_cleanup_push((thread_cleanup_t)decrement_working_threads,
- this);
- job->execute(job);
- thread_cleanup_pop(FALSE);
+ /* canceled threads are restarted to get a constant pool */
+ thread_cleanup_push((thread_cleanup_t)restart, worker);
+ while (TRUE)
+ {
+ requeue = worker->job->execute(worker->job);
+ if (requeue.type != JOB_REQUEUE_TYPE_DIRECT)
+ {
+ break;
+ }
+ else if (!worker->job->cancel)
+ { /* only allow cancelable jobs to requeue directly */
+ requeue.type = JOB_REQUEUE_TYPE_FAIR;
+ break;
+ }
+ }
thread_cleanup_pop(FALSE);
this->mutex->lock(this->mutex);
this->working_threads[i]--;
+ if (worker->job->status == JOB_STATUS_CANCELED)
+ { /* job was canceled via a custom cancel() method or did not
+ * use JOB_REQUEUE_TYPE_DIRECT */
+ worker->job->destroy(worker->job);
+ break;
+ }
+ switch (requeue.type)
+ {
+ case JOB_REQUEUE_TYPE_NONE:
+ worker->job->status = JOB_STATUS_DONE;
+ worker->job->destroy(worker->job);
+ break;
+ case JOB_REQUEUE_TYPE_FAIR:
+ worker->job->status = JOB_STATUS_QUEUED;
+ this->jobs[i]->insert_last(this->jobs[i],
+ worker->job);
+ this->job_added->signal(this->job_added);
+ break;
+ case JOB_REQUEUE_TYPE_SCHEDULE:
+ /* scheduler_t does not hold its lock when queeuing jobs
+ * so this should be safe without unlocking our mutex */
+ switch (requeue.schedule)
+ {
+ case JOB_SCHEDULE:
+ lib->scheduler->schedule_job(lib->scheduler,
+ worker->job, requeue.time.rel);
+ break;
+ case JOB_SCHEDULE_MS:
+ lib->scheduler->schedule_job_ms(lib->scheduler,
+ worker->job, requeue.time.rel);
+ break;
+ case JOB_SCHEDULE_TV:
+ lib->scheduler->schedule_job_tv(lib->scheduler,
+ worker->job, requeue.time.abs);
+ break;
+ }
+ break;
+ default:
+ break;
+ }
break;
}
}
- if (!job)
+ if (!worker->job)
{
this->job_added->wait(this->job_added, this->mutex);
}
+ worker->job = NULL;
}
this->total_threads--;
this->thread_terminated->signal(this->thread_terminated);
@@ -266,6 +347,8 @@ METHOD(processor_t, queue_job, void,
job_priority_t prio;
prio = sane_prio(job->get_priority(job));
+ job->status = JOB_STATUS_QUEUED;
+
this->mutex->lock(this->mutex);
this->jobs[prio]->insert_last(this->jobs[prio], job);
this->job_added->signal(this->job_added);
@@ -278,19 +361,26 @@ METHOD(processor_t, set_threads, void,
this->mutex->lock(this->mutex);
if (count > this->total_threads)
{ /* increase thread count */
+ worker_thread_t *worker;
int i;
- thread_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++)
{
- current = thread_create((thread_main_t)process_jobs, this);
- if (current)
+ INIT(worker,
+ .processor = this,
+ );
+ worker->thread = thread_create((thread_main_t)process_jobs, worker);
+ if (worker->thread)
{
- this->threads->insert_last(this->threads, current);
+ this->threads->insert_last(this->threads, worker);
this->total_threads++;
}
+ else
+ {
+ free(worker);
+ }
}
}
else if (count < this->total_threads)
@@ -301,26 +391,49 @@ METHOD(processor_t, set_threads, void,
this->mutex->unlock(this->mutex);
}
-METHOD(processor_t, destroy, void,
+METHOD(processor_t, cancel, void,
private_processor_t *this)
{
- thread_t *current;
- int i;
+ enumerator_t *enumerator;
+ worker_thread_t *worker;
- set_threads(this, 0);
this->mutex->lock(this->mutex);
+ this->desired_threads = 0;
+ /* cancel potentially blocking jobs */
+ enumerator = this->threads->create_enumerator(this->threads);
+ while (enumerator->enumerate(enumerator, (void**)&worker))
+ {
+ if (worker->job && worker->job->cancel)
+ {
+ worker->job->status = JOB_STATUS_CANCELED;
+ if (!worker->job->cancel(worker->job))
+ { /* job requests to be canceled explicitly, otherwise we assume
+ * the thread terminates itself and can be joined */
+ worker->thread->cancel(worker->thread);
+ }
+ }
+ }
+ enumerator->destroy(enumerator);
while (this->total_threads > 0)
{
this->job_added->broadcast(this->job_added);
this->thread_terminated->wait(this->thread_terminated, this->mutex);
}
while (this->threads->remove_first(this->threads,
- (void**)&current) == SUCCESS)
+ (void**)&worker) == SUCCESS)
{
- current->join(current);
+ worker->thread->join(worker->thread);
+ free(worker);
}
this->mutex->unlock(this->mutex);
- this->priority->destroy(this->priority);
+}
+
+METHOD(processor_t, destroy, void,
+ private_processor_t *this)
+{
+ int i;
+
+ cancel(this);
this->thread_terminated->destroy(this->thread_terminated);
this->job_added->destroy(this->job_added);
this->mutex->destroy(this->mutex);
@@ -348,10 +461,10 @@ processor_t *processor_create()
.get_job_load = _get_job_load,
.queue_job = _queue_job,
.set_threads = _set_threads,
+ .cancel = _cancel,
.destroy = _destroy,
},
.threads = linked_list_create(),
- .priority = thread_value_create(NULL),
.mutex = mutex_create(MUTEX_TYPE_DEFAULT),
.job_added = condvar_create(CONDVAR_TYPE_DEFAULT),
.thread_terminated = condvar_create(CONDVAR_TYPE_DEFAULT),