fixed callback_job cancellation for threads waiting in the bus
This commit is contained in:
+43
-6
@@ -179,28 +179,65 @@ static void remove_listener(private_bus_t *this, bus_listener_t *listener)
|
|||||||
pthread_mutex_unlock(&this->mutex);
|
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.
|
* Implementation of bus_t.listen.
|
||||||
*/
|
*/
|
||||||
static void listen_(private_bus_t *this, bus_listener_t *listener, job_t *job)
|
static void listen_(private_bus_t *this, bus_listener_t *listener, job_t *job)
|
||||||
{
|
{
|
||||||
entry_t *entry;
|
|
||||||
int old;
|
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);
|
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);
|
charon->processor->queue_job(charon->processor, job);
|
||||||
pthread_cleanup_push((void*)pthread_mutex_unlock, &this->mutex);
|
pthread_cleanup_push((void*)pthread_mutex_unlock, &this->mutex);
|
||||||
|
pthread_cleanup_push((void*)listener_cleanup, &data);
|
||||||
pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &old);
|
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_setcancelstate(old, NULL);
|
||||||
|
pthread_cleanup_pop(FALSE);
|
||||||
|
/* unlock mutex */
|
||||||
pthread_cleanup_pop(TRUE);
|
pthread_cleanup_pop(TRUE);
|
||||||
free(entry);
|
free(data.entry);
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|||||||
@@ -157,6 +157,7 @@ static void execute(private_callback_job_t *this)
|
|||||||
continue;
|
continue;
|
||||||
case JOB_REQUEUE_FAIR:
|
case JOB_REQUEUE_FAIR:
|
||||||
{
|
{
|
||||||
|
this->thread = 0;
|
||||||
charon->processor->queue_job(charon->processor,
|
charon->processor->queue_job(charon->processor,
|
||||||
&this->public.job_interface);
|
&this->public.job_interface);
|
||||||
break;
|
break;
|
||||||
@@ -164,13 +165,13 @@ static void execute(private_callback_job_t *this)
|
|||||||
case JOB_REQUEUE_NONE:
|
case JOB_REQUEUE_NONE:
|
||||||
default:
|
default:
|
||||||
{
|
{
|
||||||
|
this->thread = 0;
|
||||||
cleanup = TRUE;
|
cleanup = TRUE;
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
this->thread = 0;
|
|
||||||
unregister(this);
|
unregister(this);
|
||||||
pthread_cleanup_pop(cleanup);
|
pthread_cleanup_pop(cleanup);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -87,6 +87,8 @@ struct private_scheduler_t {
|
|||||||
* Condvar to wait for next job.
|
* Condvar to wait for next job.
|
||||||
*/
|
*/
|
||||||
pthread_cond_t condvar;
|
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_cond_wait(&this->condvar, &this->mutex);
|
||||||
}
|
}
|
||||||
pthread_setcancelstate(oldstate, NULL);
|
pthread_setcancelstate(oldstate, NULL);
|
||||||
pthread_cleanup_pop(0);
|
pthread_cleanup_pop(TRUE);
|
||||||
|
|
||||||
pthread_mutex_unlock(&this->mutex);
|
|
||||||
return JOB_REQUEUE_DIRECT;
|
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)
|
static void destroy(private_scheduler_t *this)
|
||||||
{
|
{
|
||||||
|
this->cancelled = TRUE;
|
||||||
this->job->cancel(this->job);
|
this->job->cancel(this->job);
|
||||||
this->list->destroy_function(this->list, (void*)event_destroy);
|
this->list->destroy_function(this->list, (void*)event_destroy);
|
||||||
free(this);
|
free(this);
|
||||||
@@ -251,6 +252,7 @@ scheduler_t * scheduler_create()
|
|||||||
this->public.destroy = (void(*)(scheduler_t*)) destroy;
|
this->public.destroy = (void(*)(scheduler_t*)) destroy;
|
||||||
|
|
||||||
this->list = linked_list_create();
|
this->list = linked_list_create();
|
||||||
|
this->cancelled = FALSE;
|
||||||
pthread_mutex_init(&this->mutex, NULL);
|
pthread_mutex_init(&this->mutex, NULL);
|
||||||
pthread_cond_init(&this->condvar, NULL);
|
pthread_cond_init(&this->condvar, NULL);
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user