fixed callback_job cancellation for threads waiting in the bus

3 files changed, 50 insertions, 10 deletions

diff --git a/src/charon/bus/bus.c b/src/charon/bus/bus.c
index 62e57ae3b..e53ac43ce 100644
--- a/src/charon/bus/bus.c
+++ b/src/charon/bus/bus.c
@@ -179,28 +179,65 @@ static void remove_listener(private_bus_t *this, bus_listener_t *listener)
 	pthread_mutex_unlock(&this->mutex);
 }
 
+typedef struct cleanup_data_t cleanup_data_t;
+
+/**
+ * data to remove a listener using pthread_cleanup handler
+ */
+struct cleanup_data_t {
+	/** bus instance */
+	private_bus_t *this;
+	/** listener entry */
+	entry_t *entry;
+};
+
+/**
+ * pthread_cleanup handler to remove a listener
+ */
+static void listener_cleanup(cleanup_data_t *data)
+{
+	iterator_t *iterator;
+	entry_t *entry;
+
+	iterator = data->this->listeners->create_iterator(data->this->listeners, TRUE);
+	while (iterator->iterate(iterator, (void**)&entry))
+	{
+		if (entry == data->entry)
+		{
+			iterator->remove(iterator);
+			free(entry);
+			break;
+		}
+	}
+	iterator->destroy(iterator);
+}
+
 /**
  * Implementation of bus_t.listen.
  */
 static void listen_(private_bus_t *this, bus_listener_t *listener, job_t *job)
 {
-	entry_t *entry;
 	int old;
+	cleanup_data_t data;
 	
-	entry = entry_create(listener, TRUE);
+	data.this = this;
+	data.entry = entry_create(listener, TRUE);
 
 	pthread_mutex_lock(&this->mutex);
-	this->listeners->insert_last(this->listeners, entry);
+	this->listeners->insert_last(this->listeners, data.entry);
 	charon->processor->queue_job(charon->processor, job);
 	pthread_cleanup_push((void*)pthread_mutex_unlock, &this->mutex);
+	pthread_cleanup_push((void*)listener_cleanup, &data);
 	pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &old);
-	while (entry->blocker)
+	while (data.entry->blocker)
 	{
-		pthread_cond_wait(&entry->cond, &this->mutex);
+		pthread_cond_wait(&data.entry->cond, &this->mutex);
 	}
 	pthread_setcancelstate(old, NULL);
+	pthread_cleanup_pop(FALSE);
+	/* unlock mutex */
 	pthread_cleanup_pop(TRUE);
-	free(entry);
+	free(data.entry);
 }
 
 /**
diff --git a/src/charon/processing/jobs/callback_job.c b/src/charon/processing/jobs/callback_job.c
index 924af900c..53297916e 100644
--- a/src/charon/processing/jobs/callback_job.c
+++ b/src/charon/processing/jobs/callback_job.c
@@ -157,6 +157,7 @@ static void execute(private_callback_job_t *this)
 				continue;
 			case JOB_REQUEUE_FAIR:
 			{
+				this->thread = 0;
 				charon->processor->queue_job(charon->processor,
 											 &this->public.job_interface);
 				break;
@@ -164,13 +165,13 @@ static void execute(private_callback_job_t *this)
 			case JOB_REQUEUE_NONE:
 			default:
 			{
+				this->thread = 0;
 				cleanup = TRUE;
 				break;
 			}
 		}
 		break;
 	}
-	this->thread = 0;
 	unregister(this);
 	pthread_cleanup_pop(cleanup);
 }
diff --git a/src/charon/processing/scheduler.c b/src/charon/processing/scheduler.c
index 2706585b0..ededb479a 100644
--- a/src/charon/processing/scheduler.c
+++ b/src/charon/processing/scheduler.c
@@ -87,6 +87,8 @@ struct private_scheduler_t {
 	 * Condvar to wait for next job.
 	 */
 	pthread_cond_t condvar;
+	
+	bool cancelled;
 };
 
 /**
@@ -148,9 +150,7 @@ static job_requeue_t schedule(private_scheduler_t * this)
 		pthread_cond_wait(&this->condvar, &this->mutex);
 	}
 	pthread_setcancelstate(oldstate, NULL);
-	pthread_cleanup_pop(0);
-		
-	pthread_mutex_unlock(&this->mutex);
+	pthread_cleanup_pop(TRUE);
 	return JOB_REQUEUE_DIRECT;
 }
 
@@ -234,6 +234,7 @@ static void schedule_job(private_scheduler_t *this, job_t *job, u_int32_t time)
  */
 static void destroy(private_scheduler_t *this)
 {
+	this->cancelled = TRUE;
 	this->job->cancel(this->job);
 	this->list->destroy_function(this->list, (void*)event_destroy);
 	free(this);
@@ -251,6 +252,7 @@ scheduler_t * scheduler_create()
 	this->public.destroy = (void(*)(scheduler_t*)) destroy;
 	
 	this->list = linked_list_create();
+	this->cancelled = FALSE;
 	pthread_mutex_init(&this->mutex, NULL);
 	pthread_cond_init(&this->condvar, NULL);