proper thread cancellation when using the charon->interfaces

diff --git a/src/charon/bus/bus.c b/src/charon/bus/bus.c
index 84c93f1..5f46cd2 100644 (file)
--- a/src/charon/bus/bus.c
+++ b/src/charon/bus/bus.c
@@ -185,6 +185,28 @@ static void add_listener(private_bus_t *this, bus_listener_t *listener)
 }
 
 /**
 }
 
 /**

+ * 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**)&current))
+       {
+               if (current == listener)
+               {
+                       iterator->remove(iterator);
+                       break;
+               }
+       }
+       iterator->destroy(iterator);
+       pthread_mutex_unlock(&this->mutex);
+}
+
+/**

  * Get the listener object for the calling thread
  */
 static active_listener_t *get_active_listener(private_bus_t *this)
  * Get the listener object for the calling thread
  */
 static active_listener_t *get_active_listener(private_bus_t *this)


@@ -216,6 +238,32 @@ static active_listener_t *get_active_listener(private_bus_t *this)
        return found;
 }
 
        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.
  */
 /**
  * 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;
                                                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);
        
        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_cond_wait(&listener->cond, &this->mutex);

+       pthread_setcancelstate(oldstate, NULL);
+       pthread_cleanup_pop(0);
+       

        pthread_mutex_unlock(&this->mutex);
        
        /* return signal values */
        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;
        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;
        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;

diff --git a/src/charon/bus/bus.h b/src/charon/bus/bus.h
index 200525f..4b46c7e 100644 (file)
--- a/src/charon/bus/bus.h
+++ b/src/charon/bus/bus.h
@@ -266,6 +266,14 @@ struct bus_t {
        void (*add_listener) (bus_t *this, bus_listener_t *listener);
        
        /**
        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.
         *
         * As we are fully multithreaded, we must provide a mechanism
         * @brief Listen actively on the bus.
         *
         * As we are fully multithreaded, we must provide a mechanism


@@ -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.
         *
         * 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
         * @param this          bus
         * @param level         verbosity level of the signal
         * @param thread        receives thread number emitted the signal

diff --git a/src/charon/control/interface_manager.c b/src/charon/control/interface_manager.c
index 8d9bdb6..8cae621 100644 (file)
--- a/src/charon/control/interface_manager.c
+++ b/src/charon/control/interface_manager.c
@@ -239,6 +239,14 @@ static bool unroute_listener(interface_bus_listener_t *this, signal_t signal,
 }
 
 /**
 }
 
 /**

+ * 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.
  */
 static status_t initiate(private_interface_manager_t *this,
  * Implementation of interface_manager_t.initiate.
  */
 static status_t initiate(private_interface_manager_t *this,


@@ -259,6 +267,7 @@ static status_t initiate(private_interface_manager_t *this,
        {
                ike_sa->set_peer_cfg(ike_sa, peer_cfg);
        }
        {
                ike_sa->set_peer_cfg(ike_sa, peer_cfg);
        }

+       peer_cfg->destroy(peer_cfg);

 
        listener.listener.signal = (void*)initiate_listener;
        listener.callback = callback;
 
        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;
                }
                        retval = NEED_MORE;
                        break;
                }

+               pthread_cleanup_push((void*)remove_listener, &listener);

                signal = charon->bus->listen(charon->bus, &level, &thread, 
                                                                         &current, &format, &args);
                signal = charon->bus->listen(charon->bus, &level, &thread, 
                                                                         &current, &format, &args);

+               pthread_cleanup_pop(0);

                /* ike_sa is a valid pointer until we get one of the signals */
                if (ike_sa == current)
                {
                /* 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;
                        }
                                status = NEED_MORE;
                                break;
                        }

+                       pthread_cleanup_push((void*)remove_listener, &listener);

                        signal = charon->bus->listen(charon->bus, &level, &thread, 
                                                                                 &current, &format, &args);
                        signal = charon->bus->listen(charon->bus, &level, &thread, 
                                                                                 &current, &format, &args);

+                       pthread_cleanup_pop(0);

 
                        /* even if we checked in the IKE_SA, the pointer is valid until
                         * we get an IKE_DOWN_... */
 
                        /* 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;
                        }
                                status = NEED_MORE;
                                break;
                        }

+                       pthread_cleanup_push((void*)remove_listener, &listener);

                        signal = charon->bus->listen(charon->bus, &level, &thread, 
                                                                                 &current, &format, &args);
                        signal = charon->bus->listen(charon->bus, &level, &thread, 
                                                                                 &current, &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)
                        /* even if we checked in the IKE_SA, the pointer is valid until
                         * we get an IKE_DOWN_... */
                        if (ike_sa == current)

diff --git a/src/charon/control/interface_manager.h b/src/charon/control/interface_manager.h
index 3c1613a..06a5fe6 100644 (file)
--- a/src/charon/control/interface_manager.h
+++ b/src/charon/control/interface_manager.h
@@ -84,6 +84,10 @@ struct interface_manager_t {
        /**
         * @brief Initiate a CHILD_SA, and if required, an IKE_SA.
         *
        /**
         * @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
         * @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.
         *
        /**
         * @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
         * @param this                  calling object
         * @param unique_id             unique id of the IKE_SA to terminate.
         * @param cb                    logging callback

diff --git a/src/charon/control/interfaces/dbus_interface.c b/src/charon/control/interfaces/dbus_interface.c
index 5159f1d..443df63 100644 (file)
--- a/src/charon/control/interfaces/dbus_interface.c
+++ b/src/charon/control/interfaces/dbus_interface.c
@@ -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);
                {
                        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);
        }
        reply = dbus_message_new_method_return(msg);
        dbus_connection_send(this->conn, reply, NULL);

diff --git a/src/charon/control/interfaces/stroke_interface.c b/src/charon/control/interfaces/stroke_interface.c
index 544ff61..6e3427e 100755 (executable)
--- a/src/charon/control/interfaces/stroke_interface.c
+++ b/src/charon/control/interfaces/stroke_interface.c
@@ -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);
        
        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);

 }
 
 /**
 }
 
 /**

diff --git a/src/charon/network/receiver.c b/src/charon/network/receiver.c
index a472e94..9b4bf71 100644 (file)
--- a/src/charon/network/receiver.c
+++ b/src/charon/network/receiver.c
@@ -22,6 +22,7 @@
  */
 
 #include <stdlib.h>
  */
 
 #include <stdlib.h>

+#include <unistd.h>

 #include <pthread.h>
 
 #include "receiver.h"
 #include <pthread.h>
 
 #include "receiver.h"

diff --git a/src/charon/sa/ike_sa.c b/src/charon/sa/ike_sa.c
index b9b0999..8b4b53e 100644 (file)
--- a/src/charon/sa/ike_sa.c
+++ b/src/charon/sa/ike_sa.c
@@ -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);
        }
        
        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);
        this->task_manager->queue_task(this->task_manager, task);
        
        return this->task_manager->initiate(this->task_manager);