Merge major changes to the way device state is passed around Asterisk. The two

places that cared about device states were app_queue and the hint code in pbx.c.
The changes include converting it to use the Asterisk event system, as well as
other efficiency improvements.
 * app_queue: This module used to register a callback into devicestate.c to
   monitor device state changes.  Now, it is just a subscriber to Asterisk
   events with the type, device state.
 * pbx.c hints: Previously, the device state processing thread in devicestate.c
   would call ast_hint_state_changed() each time the state of a device changed.
   Then, that code would go looking for all the hints that monitor that device,
   and call their callbacks.  All of this blocked the device state processing
   thread.  Now, the hint code is a subscriber of Asterisk events with the
   type, device state.  Furthermore, when this code receives a device state
   change event, it queues it up to be processed by another thread so that it
   doesn't block one of the event processing threads.


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

8 files changed, 174 insertions, 90 deletions

diff --git a/apps/app_queue.c b/apps/app_queue.c
index dad3ac33b..7d06d2df2 100644
--- a/apps/app_queue.c
+++ b/apps/app_queue.c
@@ -92,6 +92,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 #include "asterisk/astdb.h"
 #include "asterisk/devicestate.h"
 #include "asterisk/stringfields.h"
+#include "asterisk/event.h"
 
 enum {
 	QUEUE_STRATEGY_RINGALL = 0,
@@ -257,6 +258,9 @@ static int autofill_default = 0;
 /*! \brief queues.conf [general] option */
 static int montype_default = 0;
 
+/*! \brief Subscription to device state change events */
+static struct ast_event_sub *device_state_sub;
+
 enum queue_result {
 	QUEUE_UNKNOWN = 0,
 	QUEUE_TIMEOUT = 1,
@@ -656,10 +660,8 @@ static void *device_state_thread(void *data)
 	return NULL;
 }
 
-static int statechange_queue(const char *dev, enum ast_device_state state, void *data)
+static int statechange_queue(const char *dev, enum ast_device_state state)
 {
-	/* Avoid potential for deadlocks by spawning a new thread to handle
-	   the event */
 	struct statechange *sc;
 
 	if (!(sc = ast_calloc(1, sizeof(*sc) + strlen(dev) + 1)))
@@ -676,6 +678,22 @@ static int statechange_queue(const char *dev, enum ast_device_state state, void
 	return 0;
 }
 
+static void device_state_cb(const struct ast_event *event, void *unused)
+{
+	enum ast_device_state state;
+	const char *device;
+
+	state = ast_event_get_ie_uint(event, AST_EVENT_IE_STATE);
+	device = ast_event_get_ie_str(event, AST_EVENT_IE_DEVICE);
+
+	if (ast_strlen_zero(device)) {
+		ast_log(LOG_ERROR, "Received invalid event that had no device IE\n");
+		return;
+	}
+
+	statechange_queue(device, state);
+}
+
 static struct member *create_queue_member(const char *interface, const char *membername, int penalty, int paused)
 {
 	struct member *cur;
@@ -4747,7 +4765,9 @@ static int unload_module(void)
 	res |= ast_custom_function_unregister(&queuemembercount_function);
 	res |= ast_custom_function_unregister(&queuememberlist_function);
 	res |= ast_custom_function_unregister(&queuewaitingcount_function);
-	ast_devstate_del(statechange_queue, NULL);
+
+	if (device_state_sub)
+		ast_event_unsubscribe(device_state_sub);
 
 	ast_module_user_hangup_all();
 
@@ -4788,7 +4808,9 @@ static int load_module(void)
 	res |= ast_custom_function_register(&queuemembercount_function);
 	res |= ast_custom_function_register(&queuememberlist_function);
 	res |= ast_custom_function_register(&queuewaitingcount_function);
-	res |= ast_devstate_add(statechange_queue, NULL);
+
+	if (!(device_state_sub = ast_event_subscribe(AST_EVENT_DEVICE_STATE, device_state_cb, NULL, AST_EVENT_IE_END)))
+		res = -1;
 
 	return res;
 }
diff --git a/include/asterisk.h b/include/asterisk.h
index bdd59402e..6e20c16a9 100644
--- a/include/asterisk.h
+++ b/include/asterisk.h
@@ -87,6 +87,7 @@ void dnsmgr_start_refresh(void);		/*!< Provided by dnsmgr.c */
 int dnsmgr_reload(void);			/*!< Provided by dnsmgr.c */
 void threadstorage_init(void);			/*!< Provided by threadstorage.c */
 void ast_event_init(void);          /*!< Provided by event.c */
+int ast_device_state_engine_init(void); /*!< Provided by devicestate.c */
 
 /* Many headers need 'ast_channel' to be defined */
 struct ast_channel;
diff --git a/include/asterisk/devicestate.h b/include/asterisk/devicestate.h
index dd7b939be..b13547a16 100644
--- a/include/asterisk/devicestate.h
+++ b/include/asterisk/devicestate.h
@@ -19,6 +19,18 @@
 /*! \file
  * \brief Device state management
  *
+ * To subscribe to device state changes, use the generic ast_event_subscribe
+ * method.  For an example, see apps/app_queue.c.
+ *
+ * \todo Currently, when the state of a device changes, the device state provider
+ * calls one of the functions defined here to queue an object to say that the 
+ * state of a device has changed.  However, this does not include the new state.
+ * Another thread processes these device state change objects and calls the
+ * device state provider's callback to figure out what the new state is.  It
+ * would make a lot more sense for the new state to be included in the original
+ * function call that says the state of a device has changed.  However, it
+ * will take a lot of work to change this.
+ *
  * \arg See \ref AstExtState
  */
 
@@ -29,7 +41,11 @@
 extern "C" {
 #endif
 
-/*! Device States */
+/*! Device States 
+ *  \note The order of these states may not change because they are included
+ *        in Asterisk events which may be transmitted across the network to
+ *        other servers.
+ */
 enum ast_device_state {
 	AST_DEVICE_UNKNOWN,      /*!< Device is valid but channel didn't know state */
 	AST_DEVICE_NOT_INUSE,    /*!< Device is not used */
@@ -42,9 +58,6 @@ enum ast_device_state {
 	AST_DEVICE_ONHOLD,       /*!< Device is on hold */
 };
 
-/*! \brief Devicestate watcher call back */
-typedef int (*ast_devstate_cb_type)(const char *dev, enum ast_device_state state, void *data);
-
 /*!  \brief Devicestate provider call back */
 typedef enum ast_device_state (*ast_devstate_prov_cb_type)(const char *data);
 
@@ -93,7 +106,6 @@ enum ast_device_state ast_device_state(const char *device);
 int ast_device_state_changed(const char *fmt, ...)
 	__attribute__ ((format (printf, 1, 2)));
 
-
 /*! \brief Tells Asterisk the State for Device is changed 
  * \param device devicename like a dialstring
  * Asterisk polls the new extensionstates and calls the registered
@@ -102,22 +114,6 @@ int ast_device_state_changed(const char *fmt, ...)
  */
 int ast_device_state_changed_literal(const char *device);
 
-/*! \brief Registers a device state change callback 
- * \param callback Callback
- * \param data to pass to callback
- * The callback is called if the state for extension is changed
- * Return -1 on failure, ID on success
- */ 
-int ast_devstate_add(ast_devstate_cb_type callback, void *data);
-
-/*! \brief Unregisters a device state change callback 
- * \param callback Callback
- * \param data to pass to callback
- * The callback is called if the state for extension is changed
- * Return -1 on failure, ID on success
- */ 
-void ast_devstate_del(ast_devstate_cb_type callback, void *data);
-
 /*! \brief Add device state provider 
  * \param label to use in hint, like label:object
  * \param callback Callback
@@ -132,8 +128,6 @@ int ast_devstate_prov_add(const char *label, ast_devstate_prov_cb_type callback)
  */ 
 int ast_devstate_prov_del(const char *label);
 
-int ast_device_state_engine_init(void);
-
 #if defined(__cplusplus) || defined(c_plusplus)
 }
 #endif
diff --git a/include/asterisk/event_defs.h b/include/asterisk/event_defs.h
index b1e27c981..2627bae8d 100644
--- a/include/asterisk/event_defs.h
+++ b/include/asterisk/event_defs.h
@@ -37,13 +37,15 @@ enum ast_event_type {
 	    unique to the event itself, not necessarily across all events. */
 	AST_EVENT_CUSTOM = 0x01,
 	/*! Voicemail message waiting indication */
-	AST_EVENT_MWI    = 0x02,
+	AST_EVENT_MWI          = 0x02,
 	/*! Someone has subscribed to events */
-	AST_EVENT_SUB    = 0x03,
+	AST_EVENT_SUB          = 0x03,
 	/*! Someone has unsubscribed from events */
-	AST_EVENT_UNSUB  = 0x04,
+	AST_EVENT_UNSUB        = 0x04,
+	/*! The state of a device has changed */
+	AST_EVENT_DEVICE_STATE = 0x05,
 	/*! Number of event types.  This should be the last event type + 1 */
-	AST_EVENT_TOTAL  = 0x05,
+	AST_EVENT_TOTAL        = 0x06,
 };
 
 /*! \brief Event Information Element types */
@@ -82,11 +84,25 @@ enum ast_event_ie_type {
 	 */
 	AST_EVENT_IE_EVENTTYPE = 0x05,
 	/*!
-	 * \brief Hint that someone cares than an IE exists
+	 * \brief Hint that someone cares that an IE exists
 	 * Used by: AST_EVENT_SUB
 	 * Payload type: UINT (ast_event_ie_type)
 	 */
 	AST_EVENT_IE_EXISTS    = 0x06,
+	/*!
+	 * \brief Device Name
+	 * Used by AST_EVENT_DEVICE_STATE
+	 * Payload type: STR
+	 */
+	AST_EVENT_IE_DEVICE    = 0x07,
+	/*!
+	 * \brief Generic State IE
+	 * Used by AST_EVENT_DEVICE_STATE
+	 * Payload type: UINT
+	 * The actual state values depend on the event which
+	 * this IE is a part of.
+	 */
+	 AST_EVENT_IE_STATE    = 0x08,
 };
 
 /*!
diff --git a/include/asterisk/pbx.h b/include/asterisk/pbx.h
index 2759b36d1..e7468c6ef 100644
--- a/include/asterisk/pbx.h
+++ b/include/asterisk/pbx.h
@@ -877,8 +877,6 @@ int ast_func_read(struct ast_channel *chan, const char *function, char *workspac
  */
 int ast_func_write(struct ast_channel *chan, const char *function, const char *value);
 
-void ast_hint_state_changed(const char *device);
-
 #if defined(__cplusplus) || defined(c_plusplus)
 }
 #endif
diff --git a/main/asterisk.c b/main/asterisk.c
index a2f9efb33..efb6403d9 100644
--- a/main/asterisk.c
+++ b/main/asterisk.c
@@ -2930,7 +2930,7 @@ int main(int argc, char *argv[])
 
 	threadstorage_init();
 
-	if (load_modules(1)) {		/* Load modules */
+	if (load_modules(1)) {		/* Load modules, pre-load only */
 		printf(term_quit());
 		exit(1);
 	}
diff --git a/main/devicestate.c b/main/devicestate.c
index 567a52a36..a1b4c4ca3 100644
--- a/main/devicestate.c
+++ b/main/devicestate.c
@@ -1,7 +1,7 @@
 /*
  * Asterisk -- An open source telephony toolkit.
  *
- * Copyright (C) 1999 - 2006, Digium, Inc.
+ * Copyright (C) 1999 - 2007, Digium, Inc.
  *
  * Mark Spencer <markster@digium.com>
  *
@@ -126,6 +126,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 #include "asterisk/pbx.h"
 #include "asterisk/app.h"
 #include "asterisk/options.h"
+#include "asterisk/event.h"
 
 /*! \brief Device state strings for printing */
 static const char *devstatestring[] = {
@@ -150,16 +151,6 @@ struct devstate_prov {
 /*! \brief A list of providers */
 static AST_RWLIST_HEAD_STATIC(devstate_provs, devstate_prov);
 
-/*! \brief  A device state watcher (callback) */
-struct devstate_cb {
-	void *data;
-	ast_devstate_cb_type callback;	/*!< Where to report when state changes */
-	AST_RWLIST_ENTRY(devstate_cb) list;
-};
-
-/*! \brief A device state watcher list */
-static AST_RWLIST_HEAD_STATIC(devstate_cbs, devstate_cb);
-
 struct state_change {
 	AST_LIST_ENTRY(state_change) list;
 	char device[1];
@@ -380,59 +371,28 @@ static int getproviderstate(const char *provider, const char *address)
 	return res;
 }
 
-/*! \brief Add device state watcher */
-int ast_devstate_add(ast_devstate_cb_type callback, void *data)
-{
-	struct devstate_cb *devcb;
-
-	if (!callback || !(devcb = ast_calloc(1, sizeof(*devcb))))
-		return -1;
-
-	devcb->data = data;
-	devcb->callback = callback;
-
-	AST_RWLIST_WRLOCK(&devstate_cbs);
-	AST_RWLIST_INSERT_HEAD(&devstate_cbs, devcb, list);
-	AST_RWLIST_UNLOCK(&devstate_cbs);
-
-	return 0;
-}
-
-/*! \brief Remove device state watcher */
-void ast_devstate_del(ast_devstate_cb_type callback, void *data)
-{
-	struct devstate_cb *devcb;
-
-	AST_RWLIST_WRLOCK(&devstate_cbs);
-	AST_RWLIST_TRAVERSE_SAFE_BEGIN(&devstate_cbs, devcb, list) {
-		if ((devcb->callback == callback) && (devcb->data == data)) {
-			AST_RWLIST_REMOVE_CURRENT(&devstate_cbs, list);
-			free(devcb);
-			break;
-		}
-	}
-	AST_RWLIST_TRAVERSE_SAFE_END;
-	AST_RWLIST_UNLOCK(&devstate_cbs);
-}
-
 /*! \brief Notify callback watchers of change, and notify PBX core for hint updates
 	Normally executed within a separate thread
 */
 static void do_state_change(const char *device)
 {
-	int state;
-	struct devstate_cb *devcb;
+	enum ast_device_state state;
+	struct ast_event *event;
 
 	state = ast_device_state(device);
 	if (option_debug > 2)
 		ast_log(LOG_DEBUG, "Changing state for %s - state %d (%s)\n", device, state, devstate2str(state));
 
-	AST_RWLIST_RDLOCK(&devstate_cbs);
-	AST_RWLIST_TRAVERSE(&devstate_cbs, devcb, list)
-		devcb->callback(device, state, devcb->data);
-	AST_RWLIST_UNLOCK(&devstate_cbs);
+	if (!(event = ast_event_new(AST_EVENT_DEVICE_STATE,
+			AST_EVENT_IE_DEVICE, AST_EVENT_IE_PLTYPE_STR, device,
+			AST_EVENT_IE_STATE, AST_EVENT_IE_PLTYPE_UINT, state,
+			AST_EVENT_IE_END))) {
+		return;
+	}
 
-	ast_hint_state_changed(device);
+	ast_event_queue_and_cache(event,
+		AST_EVENT_IE_DEVICE, AST_EVENT_IE_PLTYPE_STR,
+		AST_EVENT_IE_END);
 }
 
 static int __ast_device_state_changed_literal(char *buf)
diff --git a/main/pbx.c b/main/pbx.c
index cf27f905b..49767206c 100644
--- a/main/pbx.c
+++ b/main/pbx.c
@@ -63,6 +63,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 #include "asterisk/app.h"
 #include "asterisk/devicestate.h"
 #include "asterisk/stringfields.h"
+#include "asterisk/event.h"
 
 /*!
  * \note I M P O R T A N T :
@@ -220,6 +221,29 @@ static const struct cfextension_states {
 	{ AST_EXTENSION_INUSE | AST_EXTENSION_ONHOLD,  "InUse&Hold" }
 };
 
+struct statechange {
+	AST_LIST_ENTRY(statechange) entry;
+	char dev[0];
+};
+
+/*!
+ * \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 int pbx_builtin_answer(struct ast_channel *, void *);
 static int pbx_builtin_goto(struct ast_channel *, void *);
 static int pbx_builtin_hangup(struct ast_channel *, void *);
@@ -248,6 +272,9 @@ static struct varshead globals = AST_LIST_HEAD_NOLOCK_INIT_VALUE;
 
 static int autofallthrough = 1;
 
+/*! \brief Subscription for device state change events */
+static struct ast_event_sub *device_state_sub;
+
 AST_MUTEX_DEFINE_STATIC(maxcalllock);
 static int countcalls;
 
@@ -1917,7 +1944,7 @@ int ast_extension_state(struct ast_channel *c, const char *context, const char *
 	return ast_extension_state2(e);    		/* Check all devices in the hint */
 }
 
-void ast_hint_state_changed(const char *device)
+static void handle_statechange(const char *device)
 {
 	struct ast_hint *hint;
 
@@ -1960,6 +1987,49 @@ void ast_hint_state_changed(const char *device)
 	AST_RWLIST_UNLOCK(&hints);
 }
 
+static int statechange_queue(const char *dev)
+{
+	/* Avoid potential for deadlocks by spawning a new thread to handle
+	   the event */
+	struct statechange *sc;
+
+	if (!(sc = ast_calloc(1, sizeof(*sc) + strlen(dev) + 1)))
+		return 0;
+
+	strcpy(sc->dev, dev);
+
+	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;
+}
+
+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);
+			/* Check to see if we were woken up to see the request to stop */
+			if (device_state.stop) {
+				ast_mutex_unlock(&device_state.lock);
+				return NULL;
+			}
+		}
+		ast_mutex_unlock(&device_state.lock);
+
+		handle_statechange(sc->dev);
+
+		free(sc);
+	}
+
+	return NULL;
+}
+
 /*! \brief  ast_extension_state_add: Add watcher for extension states */
 int ast_extension_state_add(const char *context, const char *exten,
 			    ast_state_cb_type callback, void *data)
@@ -6032,6 +6102,19 @@ static int pbx_builtin_sayphonetic(struct ast_channel *chan, void *data)
 	return res;
 }
 
+static void device_state_cb(const struct ast_event *event, void *unused)
+{
+	const char *device;
+
+	device = ast_event_get_ie_str(event, AST_EVENT_IE_DEVICE);
+	if (ast_strlen_zero(device)) {
+		ast_log(LOG_ERROR, "Received invalid event that had no device IE\n");
+		return;
+	}
+
+	statechange_queue(device);
+}
+
 int load_pbx(void)
 {
 	int x;
@@ -6055,6 +6138,16 @@ int load_pbx(void)
 	
 	/* Register manager application */
 	ast_manager_register2("ShowDialPlan", EVENT_FLAG_CONFIG, manager_show_dialplan, "List dialplan", mandescr_show_dialplan);
+
+	ast_mutex_init(&device_state.lock);
+	ast_cond_init(&device_state.cond, NULL);
+	ast_pthread_create(&device_state.thread, NULL, device_state_thread, NULL);
+
+	if (!(device_state_sub = ast_event_subscribe(AST_EVENT_DEVICE_STATE, device_state_cb, NULL,
+			AST_EVENT_IE_END))) {
+		return -1;
+	}
+
 	return 0;
 }