I know we have talked about rewriting app_queue for Asterisk 1.6, but once I
authorRussell Bryant <russell@russellbryant.com>
Mon, 21 May 2007 06:56:21 +0000 (06:56 +0000)
committerRussell Bryant <russell@russellbryant.com>
Mon, 21 May 2007 06:56:21 +0000 (06:56 +0000)
saw this, I couldn't help myself from changing it.  Previously, for *every*
device state change, app_queue would spawn a thread to handle it.  Now, the
device state callback just puts the state change in a queue and it gets
handled by a single state change processing thread.

git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@65298 65c4cc65-6c06-0410-ace0-fbb531ad65f3

apps/app_queue.c

index 63722a6..5e7e34d 100644 (file)
@@ -533,14 +533,14 @@ static enum queue_member_status get_member_status(struct call_queue *q, int max_
 }
 
 struct statechange {
+       AST_LIST_ENTRY(statechange) entry;
        int state;
        char dev[0];
 };
 
-static void *changethread(void *data)
+static void *handle_statechange(struct statechange *sc)
 {
        struct call_queue *q;
-       struct statechange *sc = data;
        struct member *cur;
        struct member_interface *curint;
        char *loc;
@@ -618,26 +618,62 @@ static void *changethread(void *data)
        return NULL;
 }
 
-static int statechange_queue(const char *dev, enum ast_device_state state, void *ign)
+/*!
+ * \brief Data used by the device state thread
+ */
+static struct {
+       /*! Set to 1 to stop the thread */
+       unsigned int stop:1;
+       /*! The device state monitoring thread */
+       pthread_t thread;
+       /*! Lock for the state change queue */
+       ast_mutex_t lock;
+       /*! Condition for the state change queue */
+       ast_cond_t cond;
+       /*! Queue of state changes */
+       AST_LIST_HEAD_NOLOCK(, statechange) state_change_q;
+} device_state = {
+       .thread = AST_PTHREADT_NULL,
+};
+
+static void *device_state_thread(void *data)
+{
+       struct statechange *sc;
+
+       while (!device_state.stop) {
+               ast_mutex_lock(&device_state.lock);
+               while (!(sc = AST_LIST_REMOVE_HEAD(&device_state.state_change_q, entry)))
+                       ast_cond_wait(&device_state.cond, &device_state.lock);
+               ast_mutex_unlock(&device_state.lock);
+
+               /* Check to see if we were woken up to see the request to stop */
+               if (device_state.stop)
+                       return NULL;
+
+               handle_statechange(sc);
+
+               free(sc);
+       }
+
+       return NULL;
+}
+
+static int statechange_queue(const char *dev, enum ast_device_state state, void *data)
 {
        /* Avoid potential for deadlocks by spawning a new thread to handle
           the event */
        struct statechange *sc;
-       pthread_t t;
-       pthread_attr_t attr;
 
        if (!(sc = ast_calloc(1, sizeof(*sc) + strlen(dev) + 1)))
                return 0;
 
        sc->state = state;
        strcpy(sc->dev, dev);
-       pthread_attr_init(&attr);
-       pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
-       if (ast_pthread_create_background(&t, &attr, changethread, sc)) {
-               ast_log(LOG_WARNING, "Failed to create update thread!\n");
-               free(sc);
-       }
-       pthread_attr_destroy(&attr);
+
+       ast_mutex_lock(&device_state.lock);
+       AST_LIST_INSERT_TAIL(&device_state.state_change_q, sc, entry);
+       ast_cond_signal(&device_state.cond);
+       ast_mutex_unlock(&device_state.lock);
 
        return 0;
 }
@@ -4686,6 +4722,14 @@ static int unload_module(void)
 {
        int res;
 
+       if (device_state.thread != AST_PTHREADT_NULL) {
+               device_state.stop = 1;
+               ast_mutex_lock(&device_state.lock);
+               ast_cond_signal(&device_state.cond);
+               ast_mutex_unlock(&device_state.lock);
+               pthread_join(device_state.thread, NULL);
+       }
+
        ast_cli_unregister_multiple(cli_queue, sizeof(cli_queue) / sizeof(struct ast_cli_entry));
        res = ast_manager_unregister("QueueStatus");
        res |= ast_manager_unregister("Queues");
@@ -4717,10 +4761,17 @@ static int unload_module(void)
 static int load_module(void)
 {
        int res;
-       if(!reload_queues())
+
+       if (!reload_queues())
                return AST_MODULE_LOAD_DECLINE;
+
        if (queue_persistent_members)
                reload_queue_members();
+
+       ast_mutex_init(&device_state.lock);
+       ast_cond_init(&device_state.cond, NULL);
+       ast_pthread_create(&device_state.thread, NULL, device_state_thread, NULL);
+
        ast_cli_register_multiple(cli_queue, sizeof(cli_queue) / sizeof(struct ast_cli_entry));
        res = ast_register_application(app, queue_exec, synopsis, descrip);
        res |= ast_register_application(app_aqm, aqm_exec, app_aqm_synopsis, app_aqm_descrip);