proper thread cancellation when using the charon->interfaces
This commit is contained in:
+60
-1
@@ -184,6 +184,28 @@ static void add_listener(private_bus_t *this, bus_listener_t *listener)
|
||||
pthread_mutex_unlock(&this->mutex);
|
||||
}
|
||||
|
||||
/**
|
||||
* Implementation of bus_t.remove_listener.
|
||||
*/
|
||||
static void remove_listener(private_bus_t *this, bus_listener_t *listener)
|
||||
{
|
||||
iterator_t *iterator;
|
||||
bus_listener_t *current;
|
||||
|
||||
pthread_mutex_lock(&this->mutex);
|
||||
iterator = this->listeners->create_iterator(this->listeners, TRUE);
|
||||
while (iterator->iterate(iterator, (void**)¤t))
|
||||
{
|
||||
if (current == listener)
|
||||
{
|
||||
iterator->remove(iterator);
|
||||
break;
|
||||
}
|
||||
}
|
||||
iterator->destroy(iterator);
|
||||
pthread_mutex_unlock(&this->mutex);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the listener object for the calling thread
|
||||
*/
|
||||
@@ -216,6 +238,32 @@ static active_listener_t *get_active_listener(private_bus_t *this)
|
||||
return found;
|
||||
}
|
||||
|
||||
typedef struct cancel_info_t cancel_info_t;
|
||||
|
||||
/**
|
||||
* cancellation info to cancel a listening operation cleanly
|
||||
*/
|
||||
struct cancel_info_t {
|
||||
/**
|
||||
* mutex to unlock on cancellation
|
||||
*/
|
||||
pthread_mutex_t *mutex;
|
||||
|
||||
/**
|
||||
* listener to unregister
|
||||
*/
|
||||
active_listener_t *listener;
|
||||
};
|
||||
|
||||
/**
|
||||
* disable a listener to cleanly clean up
|
||||
*/
|
||||
static void unregister(cancel_info_t *info)
|
||||
{
|
||||
info->listener->state = UNREGISTERED;
|
||||
pthread_mutex_unlock(info->mutex);
|
||||
}
|
||||
|
||||
/**
|
||||
* Implementation of bus_t.listen.
|
||||
*/
|
||||
@@ -223,14 +271,24 @@ static signal_t listen_(private_bus_t *this, level_t *level, int *thread,
|
||||
ike_sa_t **ike_sa, char** format, va_list* args)
|
||||
{
|
||||
active_listener_t *listener;
|
||||
int oldstate;
|
||||
cancel_info_t info;
|
||||
|
||||
pthread_mutex_lock(&this->mutex);
|
||||
listener = get_active_listener(this);
|
||||
/* go "listening", say hello to a thread which have a signal for us */
|
||||
listener->state = LISTENING;
|
||||
pthread_cond_broadcast(&listener->cond);
|
||||
/* wait until it has us delivered a signal, and go back to "registered" */
|
||||
/* wait until it has us delivered a signal, and go back to "registered".
|
||||
* we allow cancellation here, but must cleanly disable the listener. */
|
||||
info.mutex = &this->mutex;
|
||||
info.listener = listener;
|
||||
pthread_cleanup_push((void*)unregister, &info);
|
||||
pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &oldstate);
|
||||
pthread_cond_wait(&listener->cond, &this->mutex);
|
||||
pthread_setcancelstate(oldstate, NULL);
|
||||
pthread_cleanup_pop(0);
|
||||
|
||||
pthread_mutex_unlock(&this->mutex);
|
||||
|
||||
/* return signal values */
|
||||
@@ -384,6 +442,7 @@ bus_t *bus_create()
|
||||
private_bus_t *this = malloc_thing(private_bus_t);
|
||||
|
||||
this->public.add_listener = (void(*)(bus_t*,bus_listener_t*))add_listener;
|
||||
this->public.remove_listener = (void(*)(bus_t*,bus_listener_t*))remove_listener;
|
||||
this->public.listen = (signal_t(*)(bus_t*,level_t*,int*,ike_sa_t**,char**,va_list*))listen_;
|
||||
this->public.set_listen_state = (void(*)(bus_t*,bool))set_listen_state;
|
||||
this->public.set_sa = (void(*)(bus_t*,ike_sa_t*))set_sa;
|
||||
|
||||
@@ -265,6 +265,14 @@ struct bus_t {
|
||||
*/
|
||||
void (*add_listener) (bus_t *this, bus_listener_t *listener);
|
||||
|
||||
/**
|
||||
* @brief Unregister a listener from the bus.
|
||||
*
|
||||
* @param this bus
|
||||
* @param listener listener to unregister.
|
||||
*/
|
||||
void (*remove_listener) (bus_t *this, bus_listener_t *listener);
|
||||
|
||||
/**
|
||||
* @brief Listen actively on the bus.
|
||||
*
|
||||
@@ -275,6 +283,9 @@ struct bus_t {
|
||||
* it processes a signal, registration is required. This is done through
|
||||
* the set_listen_state() method, see below.
|
||||
*
|
||||
* The listen() function is (has) a thread cancellation point, so might
|
||||
* want to register cleanup handlers.
|
||||
*
|
||||
* @param this bus
|
||||
* @param level verbosity level of the signal
|
||||
* @param thread receives thread number emitted the signal
|
||||
|
||||
@@ -238,6 +238,14 @@ static bool unroute_listener(interface_bus_listener_t *this, signal_t signal,
|
||||
return TRUE;
|
||||
}
|
||||
|
||||
/**
|
||||
* remove a previously registered listener from the bus
|
||||
*/
|
||||
static void remove_listener(interface_bus_listener_t *listener)
|
||||
{
|
||||
charon->bus->remove_listener(charon->bus, &listener->listener);
|
||||
}
|
||||
|
||||
/**
|
||||
* Implementation of interface_manager_t.initiate.
|
||||
*/
|
||||
@@ -259,6 +267,7 @@ static status_t initiate(private_interface_manager_t *this,
|
||||
{
|
||||
ike_sa->set_peer_cfg(ike_sa, peer_cfg);
|
||||
}
|
||||
peer_cfg->destroy(peer_cfg);
|
||||
|
||||
listener.listener.signal = (void*)initiate_listener;
|
||||
listener.callback = callback;
|
||||
@@ -297,8 +306,10 @@ static status_t initiate(private_interface_manager_t *this,
|
||||
retval = NEED_MORE;
|
||||
break;
|
||||
}
|
||||
pthread_cleanup_push((void*)remove_listener, &listener);
|
||||
signal = charon->bus->listen(charon->bus, &level, &thread,
|
||||
¤t, &format, &args);
|
||||
pthread_cleanup_pop(0);
|
||||
/* ike_sa is a valid pointer until we get one of the signals */
|
||||
if (ike_sa == current)
|
||||
{
|
||||
@@ -373,8 +384,10 @@ static status_t terminate_ike(interface_manager_t *this, u_int32_t unique_id,
|
||||
status = NEED_MORE;
|
||||
break;
|
||||
}
|
||||
pthread_cleanup_push((void*)remove_listener, &listener);
|
||||
signal = charon->bus->listen(charon->bus, &level, &thread,
|
||||
¤t, &format, &args);
|
||||
pthread_cleanup_pop(0);
|
||||
|
||||
/* even if we checked in the IKE_SA, the pointer is valid until
|
||||
* we get an IKE_DOWN_... */
|
||||
@@ -475,8 +488,10 @@ static status_t terminate_child(interface_manager_t *this, u_int32_t reqid,
|
||||
status = NEED_MORE;
|
||||
break;
|
||||
}
|
||||
pthread_cleanup_push((void*)remove_listener, &listener);
|
||||
signal = charon->bus->listen(charon->bus, &level, &thread,
|
||||
¤t, &format, &args);
|
||||
pthread_cleanup_pop(0);
|
||||
/* even if we checked in the IKE_SA, the pointer is valid until
|
||||
* we get an IKE_DOWN_... */
|
||||
if (ike_sa == current)
|
||||
|
||||
@@ -84,6 +84,10 @@ struct interface_manager_t {
|
||||
/**
|
||||
* @brief Initiate a CHILD_SA, and if required, an IKE_SA.
|
||||
*
|
||||
* The inititate() function is synchronous and thus blocks until the
|
||||
* IKE_SA is established or failed. Because of this, the initiate() function
|
||||
* contains a thread cancellation point.
|
||||
*
|
||||
* @param this calling object
|
||||
* @param peer_cfg peer_cfg to use for IKE_SA setup
|
||||
* @param child_cfg child_cfg to set up CHILD_SA from
|
||||
@@ -101,6 +105,10 @@ struct interface_manager_t {
|
||||
/**
|
||||
* @brief Terminate an IKE_SA and all of its CHILD_SAs.
|
||||
*
|
||||
* The terminate() function is synchronous and thus blocks until the
|
||||
* IKE_SA is properly deleted, or the delete timed out.
|
||||
* The terminate() function contains a thread cancellation point.
|
||||
*
|
||||
* @param this calling object
|
||||
* @param unique_id unique id of the IKE_SA to terminate.
|
||||
* @param cb logging callback
|
||||
|
||||
@@ -206,9 +206,11 @@ static bool start_connection(private_dbus_interface_t *this, DBusMessage* msg)
|
||||
{
|
||||
status = charon->interfaces->initiate(charon->interfaces, peer_cfg,
|
||||
child_cfg, dbus_log, NULL);
|
||||
peer_cfg->destroy(peer_cfg);
|
||||
}
|
||||
child_cfg->destroy(child_cfg);
|
||||
else
|
||||
{
|
||||
peer_cfg->destroy(peer_cfg);
|
||||
}
|
||||
}
|
||||
reply = dbus_message_new_method_return(msg);
|
||||
dbus_connection_send(this->conn, reply, NULL);
|
||||
|
||||
@@ -784,8 +784,6 @@ static void stroke_initiate(private_stroke_interface_t *this,
|
||||
|
||||
charon->interfaces->initiate(charon->interfaces, peer_cfg, child_cfg,
|
||||
(interface_manager_cb_t)stroke_log, &info);
|
||||
peer_cfg->destroy(peer_cfg);
|
||||
child_cfg->destroy(child_cfg);
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -22,6 +22,7 @@
|
||||
*/
|
||||
|
||||
#include <stdlib.h>
|
||||
#include <unistd.h>
|
||||
#include <pthread.h>
|
||||
|
||||
#include "receiver.h"
|
||||
|
||||
@@ -806,6 +806,7 @@ static status_t initiate(private_ike_sa_t *this, child_cfg_t *child_cfg)
|
||||
}
|
||||
|
||||
task = (task_t*)child_create_create(&this->public, child_cfg);
|
||||
child_cfg->destroy(child_cfg);
|
||||
this->task_manager->queue_task(this->task_manager, task);
|
||||
|
||||
return this->task_manager->initiate(this->task_manager);
|
||||
|
||||
Reference in New Issue
Block a user