diff options
| -rw-r--r-- | CHANGES | 51 | ||||
| -rw-r--r-- | apps/app_queue.c | 4 | ||||
| -rw-r--r-- | configs/ais.conf.sample | 76 | ||||
| -rw-r--r-- | doc/distributed_devstate.txt | 310 | ||||
| -rw-r--r-- | main/devicestate.c | 215 | ||||
| -rw-r--r-- | main/pbx.c | 2 | ||||
| -rw-r--r-- | res/Makefile | 5 | ||||
| -rw-r--r-- | res/ais/ais.h | 48 | ||||
| -rw-r--r-- | res/ais/amf.c | 89 | ||||
| -rw-r--r-- | res/ais/ckpt.c | 78 | ||||
| -rw-r--r-- | res/ais/clm.c | 165 | ||||
| -rw-r--r-- | res/ais/evt.c | 588 | ||||
| -rw-r--r-- | res/ais/lck.c | 551 | ||||
| -rw-r--r-- | res/res_ais.c | 193 |
14 files changed, 2346 insertions, 29 deletions
@@ -2,6 +2,17 @@ --- Functionality changes from Asterisk 1.6.0 to Asterisk 1.6.1 ------------- ------------------------------------------------------------------------------ +Device State Handling +--------------------- + * The event infrastructure in Asterisk got another big update to help support + distributed events. It currently supports distributed device state and + distributed Voicemail MWI (Message Waiting Indication). A new module has + been merged, res_ais, which facilitates communicating events between servers. + It uses the SAForum AIS (Service Availability Forum Application Interface + Specification) CLM (Cluster Management) and EVT (Event) services to maintain + a cluster of Asterisk servers, and to share events between them. For more + information on setting this up, see doc/distributed_devstate.txt. + Dialplan Functions ------------------ * Added a new dialplan function, AST_CONFIG(), which allows you to access @@ -17,9 +28,9 @@ Dialplan Functions * TIMEOUT() has been modified to be accurate down to the millisecond. * ENUM*() functions now include the following new options: - 'u' returns the full URI and does not strip off the URI-scheme. - - 's' triggers ISN specific rewriting - - 'i' looks for branches into an Infrastructure ENUM tree - - 'd' for a direct DNS lookup without any flipping of digits. + - 's' triggers ISN specific rewriting + - 'i' looks for branches into an Infrastructure ENUM tree + - 'd' for a direct DNS lookup without any flipping of digits. * TXCIDNAME() has a new zone-suffix parameter (which defaults to 'e164.arpa') * CHANNEL() now has options for the maximum, minimum, and standard or normal deviation of jitter, rtt, and loss for a call using chan_sip. @@ -116,9 +127,9 @@ CLI Changes which shows which configuration files are in use. * New CLI commands, "pri show version" and "ss7 show version" that will display which version of libpri and libss7 are being used, respectively. - A new API call was added so trunk will now have to be compiled against - a versions of libpri and libss7 that have them or it will not know that - these libraries exist. + A new API call was added so trunk will now have to be compiled against + a versions of libpri and libss7 that have them or it will not know that + these libraries exist. DNS manager changes ------------------- @@ -443,10 +454,10 @@ Voicemail Changes a web interface of some kind). * Added the support for marking messages as "urgent." There are two methods to accomplish this. One is to pass the 'U' option to VoiceMail(). Another way to mark a message as urgent - is to specify "review=yes" in voicemail.conf. Doing this will cause allow the user to mark - the message as urgent after he has recorded a voicemail by following the voice instructions. - When listening to voicemails using VoiceMailMain urgent messages will be presented before other - messages + is to specify "review=yes" in voicemail.conf. Doing this will cause allow the user to mark + the message as urgent after he has recorded a voicemail by following the voice instructions. + When listening to voicemails using VoiceMailMain urgent messages will be presented before other + messages Queue changes ------------- @@ -480,18 +491,18 @@ Queue changes device state reported. * New configuration option: randomperiodicannounce. If a list of periodic announcements is specified by the periodic-announce option, then one will be chosen randomly when it is time - to play a periodic announcment + to play a periodic announcment * New configuration options: announce-position now takes two more values in addition to "yes" and "no." Two new options, "limit" and "more," are allowed. These are tied to another option, - announce-position-limit. By setting announce-position to "limit" callers will only have their - position announced if their position is less than what is specified by announce-position-limit. - If announce-position is set to "more" then callers beyond the position specified by announce-position-limit - will be told that their are more than announce-position-limit callers waiting. + announce-position-limit. By setting announce-position to "limit" callers will only have their + position announced if their position is less than what is specified by announce-position-limit. + If announce-position is set to "more" then callers beyond the position specified by announce-position-limit + will be told that their are more than announce-position-limit callers waiting. * Two new queue log events have been added. An ADDMEMBER event will be logged when a realtime queue member is added and a REMOVEMEMBER event will be logged - when a realtime queue member is removed. Since there is no calling channel associated - with these events, the string "REALTIME" is placed where the channel's unique id - is typically placed. + when a realtime queue member is removed. Since there is no calling channel associated + with these events, the string "REALTIME" is placed where the channel's unique id + is typically placed. MeetMe Changes -------------- @@ -761,7 +772,7 @@ Miscellaneous * iLBC source code no longer included (see UPGRADE.txt for details) * If compiled with DETECT_DEADLOCKS enabled and if you have glibc, then if deadlock is detected, a backtrace of the stack which led to the lock calls - will be output to the CLI. + will be output to the CLI. * If compiled with DEBUG_THREADS enabled and if you have glibc, then issuing the "core show locks" CLI command will give lock information output as well - as a backtrace of the stack which led to the lock calls. + as a backtrace of the stack which led to the lock calls. diff --git a/apps/app_queue.c b/apps/app_queue.c index 9d9579739..3f68432a4 100644 --- a/apps/app_queue.c +++ b/apps/app_queue.c @@ -6334,8 +6334,10 @@ static int load_module(void) ast_log(LOG_WARNING, "devicestate taskprocessor reference failed - devicestate notifications will not occur\n"); } - if (!(device_state_sub = ast_event_subscribe(AST_EVENT_DEVICE_STATE, device_state_cb, NULL, AST_EVENT_IE_END))) + if (!(device_state_sub = ast_event_subscribe(AST_EVENT_DEVICE_STATE_CHANGE, device_state_cb, NULL, AST_EVENT_IE_END))) { res = -1; + } + ast_realtime_require_field("queue_members", "paused", RQ_INTEGER1, 1, "uniqueid", RQ_UINTEGER2, 5, NULL); return res ? AST_MODULE_LOAD_DECLINE : 0; diff --git a/configs/ais.conf.sample b/configs/ais.conf.sample new file mode 100644 index 000000000..42991a6ca --- /dev/null +++ b/configs/ais.conf.sample @@ -0,0 +1,76 @@ +; +; Sample configuration file for res_ais +; * SAForum AIS (Application Interface Specification) +; +; More information on the AIS specification is available from the SAForum. +; * http://www.saforum.org/ +; +; A nice open source implementation of AIS is available called openais. Visit +; the openais website for downloads and more information. +; * http://www.openais.org/ +; + +; +; [general] +; The general section is reserved but not currently used. +; + +; +; Event channels are named distributed groups that share events. Each node +; that is the member of the event channel should have an entry in their +; ais.conf file that indicates that they are a member of the event channel. +; Each node's entry for the event channel also indicates which event types +; will be published to other nodes, as well as which event types this node +; will subscribe to from other nodes in the event channel. +; +; The name of the event channel is the name in brackets that begin a section +; in the configuration file. +; [mwi] +; +; To define an event channel, this entry must be in the configuration section: +; type=event_channel +; +; Indicate that a node is capable of publishing events of a certain type by +; using the publish_event directive. +; publish_event=mwi +; +; Indicate that a node is interested in receiving events of a certain type +; from other nodes in the event channel by using the subscribe_event directive. +; subscribe_event=mwi +; +; Supported event types include: mwi, device_state +; + +; +; This example is for a node that can provide MWI state information, but should +; also be listening for MWI state changes from other nodes. Examples of when +; this would be used are when this is both a voicemail server and also has +; phones directly registered to it. +; +; [mwi] +; type=event_channel +; publish_event=mwi +; subscribe_event=mwi +; + +; +; This example would be used for a node that can provide MWI state to other +; nodes, but does not need to know about MWI state changes that happen on +; any other node. This would most likely be a voicemail server where no +; phones are directly registered. +; +; [mwi] +; type=event_channel +; publish_event=mwi +; + +; +; This example would be used for a node that has phones directly registered +; to it, but does not have direct access to voicemail. So, this node wants +; to be informed about MWI state changes on other voicemail server nodes, but +; is not capable of publishing any state changes. +; +; [mwi] +; type=event_channel +; subscribe_event=mwi +; diff --git a/doc/distributed_devstate.txt b/doc/distributed_devstate.txt new file mode 100644 index 000000000..2a685159e --- /dev/null +++ b/doc/distributed_devstate.txt @@ -0,0 +1,310 @@ +=============================================================================== +=== +=== Distributed Device State +=== +=== Copyright (C) 2007-2008, Digium, Inc. +=== Russell Bryant <russell@digium.com> +=== +=============================================================================== + +------------------------------------------------------------------------------- +--- INTRODUCTION +------------------------------------------------------------------------------- + +Various changes have been made related to "event handling" in Asterisk. +One of the most important things included in these changes is the ability +to share certain events between servers. The two types of events that can +currently be shared between servers are: + + 1) MWI - Message Waiting Indication + - This gives you a high performance option for letting servers in a + cluster be aware of changes in the state of a mailbox. Instead of + having each server have to poll an ODBC database, this lets the server + that actually made the change to the mailbox generate an event which + will get distributed to the other servers that have subscribed to this + information. + + 2) Device State + - This lets servers in a local cluster inform each other about changes in + the state of a device on that particular server. When the state of a + device changes on any server, the overall state of that device across + the cluster will get recalculated. So, any subscriptions to the state + of a device, such as hints in the dialplan or an application like + Queue() which reads device state, will then reflect the state of a + device across a cluster. + +------------------------------------------------------------------------------- + +------------------------------------------------------------------------------- +--- OpenAIS Installation +------------------------------------------------------------------------------- + +--- Description --- + +The current solution for providing distributed events with Asterisk is done by +using the AIS (Application Interface Specification), which provides an API for +a distributed event service. While this API is standardized, this code has +been developed exclusively against the open source implementation of AIS called +OpenAIS. + +For more information about OpenAIS, visit their web site: + + http://www.openais.org/ + +--- Download --- + +To quickly downlaod OpenAIS, just check it out of svn: + +$ svn co http://svn.osdl.org/openais/trunk openais-trunk + +--- Compile --- + +$ cd openais-trunk +$ make PREFIX=/usr + +--- Install --- + +By default, the current Makefile installs the libraries into /usr/lib/openais/, +which is a little bit inconvenient. So, open up the Makefile, find the lines +that start with "LIBDIR=" to define the lib installation directory, and remove +the trailing "openais" so it just gets installed in /usr/lib/. + +$ sudo make install PREFIX=/usr + +------------------------------------------------------------------------------- + +------------------------------------------------------------------------------- +--- OpenAIS Configuration +------------------------------------------------------------------------------- + +Basic OpenAIS configuration to get this working is actually pretty easy. When +you install it, it will put some default configuration files into /etc/ais/. +Edit openais.conf ... + +$ ${EDITOR:-vim} /etc/ais/openais.conf + +The only section that you should need to change is the totem - interface +section. + +totem { + ... + interface { + interface { + ringnumber: 0 + bindnetaddr: 10.19.0.0 + mcastaddr: 226.94.1.1 + mcastport: 5405 + } +} + +The default mcastaddr and mcastport is probably fine. But, you need to change +the bindnetaddr to match the network address that the nodes of your cluster +will communicate on. + +The one other thing that you need to do is create a user called "ais". + +$ sudo adduser ais + +See the OpenAIS QUICKSTART file for more information on installing, +configuring, and testing OpenAIS. + +$ cd openais-trunk +$ less QUICKSTART + +------------------------------------------------------------------------------- + +------------------------------------------------------------------------------- +--- Running OpenAIS +------------------------------------------------------------------------------- + +While testing, I would recommend starting the aisexec application in the +foreground so that you can see debug messages that verify that the nodes have +discovered each other and joined the cluster. + +$ sudo aisexec -f + +------------------------------------------------------------------------------- + +------------------------------------------------------------------------------- +--- Installing Asterisk +------------------------------------------------------------------------------- + +Install Asterisk as usual. Just make sure that you run the configure script +after OpenAIS gets installed. That way, it will find the AIS header files and +will let you build the res_ais module. Check menuselect to make sure that +res_ais is going to get compiled and installed. + +$ cd asterisk-events +$ ./configure + +$ make menuselect + ---> Resource Modules + +If you have existing configuration on the system being used for testing, just +be sure to install the addition configuration file needed for res_ais. + +$ sudo cp configs/ais.conf.sample /etc/asterisk/ais.conf + +------------------------------------------------------------------------------- + +------------------------------------------------------------------------------- +--- Configuring Asterisk +------------------------------------------------------------------------------- + +First, ensure that you have a unique "entity ID" set for each server. + +*CLI> core show settings + ... + Entity ID: 01:23:45:67:89:ab + +The code will attempt to generate a unique entity ID for you by reading +MAC addresses off of a network interface. However, you can also set it +manually in the [options] section of asterisk.conf. + +$ sudo ${EDITOR:-vim} /etc/asterisk/asterisk.conf + +[options] +... +entity_id=01:23:45:67:89:ab + + +Edit the Asterisk ais.conf to enable distributed events. For example, if you +would like to enable distributed device state, you should add the following +section to the file: + +$ sudo ${EDITOR:-vim} /etc/asterisk/ais.conf + +[device_state] +type=event_channel +publish_event=device_state +subscribe_event=device_state + +For more information on the contents and available options in this configuration +file, please see the sample configuration file: + +$ cd asterisk-events +$ less configs/ais.conf.sample + +------------------------------------------------------------------------------- + +------------------------------------------------------------------------------- +--- Basic Testing of Asterisk with OpenAIS +------------------------------------------------------------------------------- + +If you have OpenAIS successfully installed and running, as well as Asterisk +with OpenAIS support successfully installed, configured, and running, then you +are ready to test out some of the AIS functionality in Asterisk. + +The first thing to test is to verify that all of the nodes that you think should +be in your cluster are actually there. There is an Asterisk CLI command which +will list the current cluster members using the AIS Cluster Membership Service +(CLM). + +*CLI> ais clm show members + +============================================================= +=== Cluster Members ========================================= +============================================================= +=== +=== --------------------------------------------------------- +=== Node Name: 10.19.2.255 +=== ==> ID: 0xa1302ff +=== ==> Address: 10.19.2.255 +=== ==> Member: Yes +=== --------------------------------------------------------- +=== +=== --------------------------------------------------------- +=== Node Name: 10.19.6.187 +=== ==> ID: 0xa1306bb +=== ==> Address: 10.19.6.187 +=== ==> Member: Yes +=== --------------------------------------------------------- +=== +============================================================= + + +The next thing to do is to verify that you have successfully configured some +event channels in the Asterisk ais.conf file. This command is related to the +event service (EVT), so like the previous command, uses the syntax: +"ais <service name> <command>". + +*CLI> ais evt show event channels + +============================================================= +=== Event Channels ========================================== +============================================================= +=== +=== --------------------------------------------------------- +=== Event Channel Name: mwi +=== ==> Publishing Event Type: mwi +=== ==> Subscribing to Event Type: mwi +=== --------------------------------------------------------- +=== +=== --------------------------------------------------------- +=== Event Channel Name: device_state +=== ==> Publishing Event Type: device_state +=== ==> Subscribing to Event Type: device_state +=== --------------------------------------------------------- +=== +============================================================= + +------------------------------------------------------------------------------- + +------------------------------------------------------------------------------- +--- Testing Distributed Device State +------------------------------------------------------------------------------- + +The easiest way to test distributed device state is to use the DEVICE_STATE() +diaplan function. For example, you could have the following piece of dialplan +on every server: + +[devstate_test] + +exten => 1234,hint,Custom:mystate + +exten => set_inuse,1,Set(DEVICE_STATE(Custom:mystate)=INUSE) +exten => set_not_inuse,1,Set(DEVICE_STATE(Custom:mystate)=NOT_INUSE) + +exten => check,1,NoOp(Custom:mystate is ${DEVICE_STATE(Custom:mystate)}) + + +Now, you can test that the cluster-wide state of "Custom:mystate" is what +you would expect after going to the CLI of each server and adjusting the state. + +server1*CLI> console dial set_inuse@devstate_test + ... + +server2*CLI> console dial check@devstate_test + -- Executing [check@devstate_test:1] NoOp("OSS/dsp", "Custom:mystate is INUSE") in new stack + +Various combinations of setting and checking the state on different servers can +be used to verify that it works as expected. Also, you can see the status of +the hint on each server, as well, to see how extension state would reflect the +state change with distributed device state: + +server2*CLI> core show hints + -= Registered Asterisk Dial Plan Hints =- + 1234@devstate_test : Custom:mystate State:InUse Watchers 0 + + +One other helpful thing here during testing and debugging is to enable debug +logging. To do so, enable debug on the console in /etc/asterisk/logger.conf. +Also, enable debug at the Asterisk CLI. + +*CLI> core set debug 1 + +When you have this debug enabled, you will see output during the processing of +every device state change. The important thing to look for is where the known +state of the device for each server is added together to determine the overall +state. + +------------------------------------------------------------------------------- + + +------------------------------------------------------------------------------- +--- Question, Comments, and Bug Reports +------------------------------------------------------------------------------- + +For now, please direct all feedback to Russell Bryant <russell@digium.com>. + +------------------------------------------------------------------------------- diff --git a/main/devicestate.c b/main/devicestate.c index 85ac6492e..8c20d1703 100644 --- a/main/devicestate.c +++ b/main/devicestate.c @@ -1,9 +1,10 @@ /* * Asterisk -- An open source telephony toolkit. * - * Copyright (C) 1999 - 2007, Digium, Inc. + * Copyright (C) 1999 - 2008, Digium, Inc. * * Mark Spencer <markster@digium.com> + * Russell Bryant <russell@digium.com> * * See http://www.asterisk.org for more information about * the Asterisk project. Please do not directly contact @@ -20,14 +21,18 @@ * * \brief Device state management * - * * \author Mark Spencer <markster@digium.com> + * \author Russell Bryant <russell@digium.com> * * \arg \ref AstExtState */ /*! \page AstExtState Extension and device states in Asterisk * + * (Note that these descriptions of device states and extension + * states have not been updated to the way things work + * in Asterisk 1.6.) + * * Asterisk has an internal system that reports states * for an extension. By using the dialplan priority -1, * also called a \b hint, a connection can be made from an @@ -171,6 +176,23 @@ enum devstate_cache { CACHE_OFF, }; +struct devstate_change { + AST_LIST_ENTRY(devstate_change) entry; + uint32_t state; + struct ast_eid eid; + char device[1]; +}; + +struct { + pthread_t thread; + struct ast_event_sub *event_sub; + ast_cond_t cond; + ast_mutex_t lock; + AST_LIST_HEAD_NOLOCK(, devstate_change) devstate_change_q; +} devstate_collector = { + .thread = AST_PTHREADT_NULL, +}; + /* Forward declarations */ static int getproviderstate(const char *provider, const char *address); @@ -271,7 +293,7 @@ static enum ast_device_state devstate_cached(const char *device) enum ast_device_state res = AST_DEVICE_UNKNOWN; struct ast_event *event; - event = ast_event_get_cached(AST_EVENT_DEVICE_STATE, + event = ast_event_get_cached(AST_EVENT_DEVICE_STATE_CHANGE, AST_EVENT_IE_DEVICE, AST_EVENT_IE_PLTYPE_STR, device, AST_EVENT_IE_END); @@ -383,7 +405,6 @@ static int getproviderstate(const char *provider, const char *address) struct devstate_prov *devprov; int res = AST_DEVICE_INVALID; - AST_RWLIST_RDLOCK(&devstate_provs); AST_RWLIST_TRAVERSE(&devstate_provs, devprov, list) { ast_debug(5, "Checking provider %s with %s\n", devprov->label, provider); @@ -394,6 +415,7 @@ static int getproviderstate(const char *provider, const char *address) } } AST_RWLIST_UNLOCK(&devstate_provs); + return res; } @@ -401,7 +423,9 @@ static void devstate_event(const char *device, enum ast_device_state state, enum { struct ast_event *event; - if (!(event = ast_event_new(AST_EVENT_DEVICE_STATE, + ast_debug(1, "device '%s' state '%d'\n", device, state); + + if (!(event = ast_event_new(AST_EVENT_DEVICE_STATE_CHANGE, AST_EVENT_IE_DEVICE, AST_EVENT_IE_PLTYPE_STR, device, AST_EVENT_IE_STATE, AST_EVENT_IE_PLTYPE_UINT, state, AST_EVENT_IE_END))) { @@ -413,6 +437,7 @@ static void devstate_event(const char *device, enum ast_device_state state, enum * device name if it exists. */ ast_event_queue_and_cache(event, AST_EVENT_IE_DEVICE, AST_EVENT_IE_PLTYPE_STR, + AST_EVENT_IE_EID, AST_EVENT_IE_PLTYPE_RAW, sizeof(struct ast_eid), AST_EVENT_IE_END); } else { ast_event_queue(event); @@ -540,9 +565,189 @@ static void *do_devstate_changes(void *data) return NULL; } +static void destroy_devstate_change(struct devstate_change *sc) +{ + ast_free(sc); +} + +#define MAX_SERVERS 64 +struct change_collection { + struct devstate_change states[MAX_SERVERS]; + size_t num_states; +}; + +static void devstate_cache_cb(const struct ast_event *event, void *data) +{ + struct change_collection *collection = data; + int i; + const struct ast_eid *eid; + + if (collection->num_states == ARRAY_LEN(collection->states)) { + ast_log(LOG_ERROR, "More per-server state values than we have room for (MAX_SERVERS is %d)\n", + MAX_SERVERS); + return; + } + + if (!(eid = ast_event_get_ie_raw(event, AST_EVENT_IE_EID))) { + ast_log(LOG_ERROR, "Device state change event with no EID\n"); + return; + } + + i = collection->num_states; + + collection->states[i].state = ast_event_get_ie_uint(event, AST_EVENT_IE_STATE); + collection->states[i].eid = *eid; + + collection->num_states++; +} + +static void process_collection(const char *device, struct change_collection *collection) +{ + int i; + struct ast_devstate_aggregate agg; + enum ast_device_state state; + struct ast_event *event; + + ast_devstate_aggregate_init(&agg); + + for (i = 0; i < collection->num_states; i++) { + ast_debug(1, "Adding per-server state of '%s' for '%s'\n", + devstate2str(collection->states[i].state), device); + ast_devstate_aggregate_add(&agg, collection->states[i].state); + } + + state = ast_devstate_aggregate_result(&agg); + + ast_debug(1, "Aggregate devstate result is %d\n", state); + + event = ast_event_get_cached(AST_EVENT_DEVICE_STATE, + AST_EVENT_IE_DEVICE, AST_EVENT_IE_PLTYPE_STR, device, + AST_EVENT_IE_END); + + if (event) { + enum ast_device_state old_state; + + old_state = ast_event_get_ie_uint(event, AST_EVENT_IE_STATE); + + ast_event_destroy(event); + + if (state == old_state) { + /* No change since last reported device state */ + ast_debug(1, "Aggregate state for device '%s' has not changed from '%s'\n", + device, devstate2str(state)); + return; + } + } + + ast_debug(1, "Aggregate state for device '%s' has changed to '%s'\n", + device, devstate2str(state)); + + 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); + + if (!event) + return; + + ast_event_queue_and_cache(event, + AST_EVENT_IE_DEVICE, AST_EVENT_IE_PLTYPE_STR, + AST_EVENT_IE_END); +} + +static void handle_devstate_change(struct devstate_change *sc) +{ + struct ast_event_sub *tmp_sub; + struct change_collection collection = { + .num_states = 0, + }; + + ast_debug(1, "Processing device state change for '%s'\n", sc->device); + + if (!(tmp_sub = ast_event_subscribe_new(AST_EVENT_DEVICE_STATE_CHANGE, devstate_cache_cb, &collection))) { + ast_log(LOG_ERROR, "Failed to create subscription\n"); + return; + } + + if (ast_event_sub_append_ie_str(tmp_sub, AST_EVENT_IE_DEVICE, sc->device)) { + ast_log(LOG_ERROR, "Failed to append device IE\n"); + ast_event_sub_destroy(tmp_sub); + return; + } + + /* Populate the collection of device states from the cache */ + ast_event_dump_cache(tmp_sub); + + process_collection(sc->device, &collection); + + ast_event_sub_destroy(tmp_sub); +} + +static void *run_devstate_collector(void *data) +{ + for (;;) { + struct devstate_change *sc; + + ast_mutex_lock(&devstate_collector.lock); + while (!(sc = AST_LIST_REMOVE_HEAD(&devstate_collector.devstate_change_q, entry))) + ast_cond_wait(&devstate_collector.cond, &devstate_collector.lock); + ast_mutex_unlock(&devstate_collector.lock); + + handle_devstate_change(sc); + + destroy_devstate_change(sc); + } + + return NULL; +} + +static void devstate_change_collector_cb(const struct ast_event *event, void *data) +{ + struct devstate_change *sc; + const char *device; + const struct ast_eid *eid; + uint32_t state; + + device = ast_event_get_ie_str(event, AST_EVENT_IE_DEVICE); + eid = ast_event_get_ie_raw(event, AST_EVENT_IE_EID); + state = ast_event_get_ie_uint(event, AST_EVENT_IE_STATE); + + if (ast_strlen_zero(device) || !eid) { + ast_log(LOG_ERROR, "Invalid device state change event received\n"); + return; + } + + if (!(sc = ast_calloc(1, sizeof(*sc) + strlen(device)))) + return; + + strcpy(sc->device, device); + sc->eid = *eid; + sc->state = state; + + ast_mutex_lock(&devstate_collector.lock); + AST_LIST_INSERT_TAIL(&devstate_collector.devstate_change_q, sc, entry); + ast_cond_signal(&devstate_collector.cond); + ast_mutex_unlock(&devstate_collector.lock); +} + /*! \brief Initialize the device state engine in separate thread */ int ast_device_state_engine_init(void) { + devstate_collector.event_sub = ast_event_subscribe(AST_EVENT_DEVICE_STATE_CHANGE, + devstate_change_collector_cb, NULL, AST_EVENT_IE_END); + + if (!devstate_collector.event_sub) { + ast_log(LOG_ERROR, "Failed to create subscription for the device state change collector\n"); + return -1; + } + + ast_mutex_init(&devstate_collector.lock); + ast_cond_init(&devstate_collector.cond, NULL); + if (ast_pthread_create_background(&devstate_collector.thread, NULL, run_devstate_collector, NULL) < 0) { + ast_log(LOG_ERROR, "Unable to start device state collector thread.\n"); + return -1; + } + ast_cond_init(&change_pending, NULL); if (ast_pthread_create_background(&change_thread, NULL, do_devstate_changes, NULL) < 0) { ast_log(LOG_ERROR, "Unable to start device state change thread.\n"); diff --git a/main/pbx.c b/main/pbx.c index fcc3fb3f0..6bb8a14f6 100644 --- a/main/pbx.c +++ b/main/pbx.c @@ -8070,7 +8070,7 @@ int load_pbx(void) /* Register manager application */ ast_manager_register2("ShowDialPlan", EVENT_FLAG_CONFIG | EVENT_FLAG_REPORTING, manager_show_dialplan, "List dialplan", mandescr_show_dialplan); - if (!(device_state_sub = ast_event_subscribe(AST_EVENT_DEVICE_STATE, device_state_cb, NULL, + if (!(device_state_sub = ast_event_subscribe(AST_EVENT_DEVICE_STATE_CHANGE, device_state_cb, NULL, AST_EVENT_IE_END))) { return -1; } diff --git a/res/Makefile b/res/Makefile index 92f0f336e..a7ba0833f 100644 --- a/res/Makefile +++ b/res/Makefile @@ -35,6 +35,8 @@ ael/ael_lex.o: ASTCFLAGS+=-I. -Iael -Wno-unused ael/ael.tab.o: ael/ael.tab.c ael/ael.tab.h ../include/asterisk/ael_structs.h ael/ael.tab.o: ASTCFLAGS+=-I. -Iael -DYYENABLE_NLS=0 +$(if $(filter res_ais,$(EMBEDDED_MODS)),modules.link,res_ais.so): ais/clm.o ais/evt.o + $(if $(filter res_snmp,$(EMBEDDED_MODS)),modules.link,res_snmp.so): snmp/agent.o $(if $(filter res_ael_share,$(EMBEDDED_MODS)),modules.link,res_ael_share.so): ael/ael_lex.o ael/ael.tab.o ael/pval.o @@ -49,5 +51,4 @@ ael/ael.tab.c ael/ael.tab.h: ael/pval.o: ael/pval.c clean:: - rm -f snmp/*.o - rm -f ael/*.o + rm -f snmp/*.o ael/*.o ais/*.o diff --git a/res/ais/ais.h b/res/ais/ais.h new file mode 100644 index 000000000..2c4c18a87 --- /dev/null +++ b/res/ais/ais.h @@ -0,0 +1,48 @@ +/* + * Asterisk -- An open source telephony toolkit. + * + * Copyright (C) 2007, Digium, Inc. + * + * Russell Bryant <russell@digium.com> + * + * See http://www.asterisk.org for more information about + * the Asterisk project. Please do not directly contact + * any of the maintainers of this project for assistance; + * the project provides a web site, mailing lists and IRC + * channels for your use. + * + * This program is free software, distributed under the terms of + * the GNU General Public License Version 2. See the LICENSE file + * at the top of the source tree. + */ + +/*! + * \file + * \author Russell Bryant <russell@digium.com> + * + * \brief Usage of the SAForum AIS (Application Interface Specification) + * + * \arg http://www.openais.org/ + */ + +#ifndef AST_AIS_H +#define AST_AIS_H + +#include <openais/saAis.h> +#include <openais/saClm.h> +#include <openais/saEvt.h> + +extern SaVersionT ais_version; + +extern SaClmHandleT clm_handle; +extern SaEvtHandleT evt_handle; + +int ast_ais_clm_load_module(void); +int ast_ais_clm_unload_module(void); + +int ast_ais_evt_load_module(void); +int ast_ais_evt_unload_module(void); + +const char *ais_err2str(SaAisErrorT error); + +#endif /* AST_AIS_H */ diff --git a/res/ais/amf.c b/res/ais/amf.c new file mode 100644 index 000000000..fec9af6e0 --- /dev/null +++ b/res/ais/amf.c @@ -0,0 +1,89 @@ +/* + * Asterisk -- An open source telephony toolkit. + * + * Copyright (C) 2007, Digium, Inc. + * + * Russell Bryant <russell@digium.com> + * + * See http://www.asterisk.org for more information about + * the Asterisk project. Please do not directly contact + * any of the maintainers of this project for assistance; + * the project provides a web site, mailing lists and IRC + * channels for your use. + * + * This program is free software, distributed under the terms of + * the GNU General Public License Version 2. See the LICENSE file + * at the top of the source tree. + */ + +/*! + * \file + * \author Russell Bryant <russell@digium.com> + * + * \brief Usage of the SAForum AIS (Application Interface Specification) + * + * \arg http://www.openais.org/ + * + * This file contains the code specific to the use of the AMF (Application + * Management Framework). + */ + +#include "asterisk.h" + +ASTERISK_FILE_VERSION(__FILE__, "$Revision$"); + +#include <stdlib.h> +#include <stdio.h> +#include <string.h> +#include <unistd.h> +#include <errno.h> + +#include "ais.h" + +#include "asterisk/module.h" +#include "asterisk/utils.h" +#include "asterisk/cli.h" +#include "asterisk/logger.h" + +SaAmfHandleT amf_handle; + +static const SaAmfCallbacksT amf_callbacks = { + .saAmfHealthcheckCallback = NULL, + .saAmfComponentTerminateCallback = NULL, + .saAmfCSISetCallback = NULL, + .saAmfProtectionGroupTrackCallback = NULL, +#if 0 + /*! XXX \todo These appear to be define in the B.02.01 spec, but this won't + * compile with them in there. Look into it some more ... */ + .saAmfProxiedComponentInstantiateCallback = NULL, + .saAmfProxiedComponentCleanupCallback = NULL, +#endif +}; + +int ast_ais_amf_load_module(void) +{ + SaAisErrorT ais_res; + + ais_res = saAmfInitialize(&amf_handle, &amf_callbacks, &ais_version); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Could not initialize AMF: %s\n", + ais_err2str(ais_res)); + return -1; + } + + return 0; +} + +int ast_ais_amf_unload_module(void) +{ + SaAisErrorT ais_res; + + ais_res = saAmfFinalize(amf_handle); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Problem stopping AMF: %s\n", + ais_err2str(ais_res)); + return -1; + } + + return 0; +} diff --git a/res/ais/ckpt.c b/res/ais/ckpt.c new file mode 100644 index 000000000..bdf4b312c --- /dev/null +++ b/res/ais/ckpt.c @@ -0,0 +1,78 @@ +/* + * Asterisk -- An open source telephony toolkit. + * + * Copyright (C) 2007, Digium, Inc. + * + * Russell Bryant <russell@digium.com> + * + * See http://www.asterisk.org for more information about + * the Asterisk project. Please do not directly contact + * any of the maintainers of this project for assistance; + * the project provides a web site, mailing lists and IRC + * channels for your use. + * + * This program is free software, distributed under the terms of + * the GNU General Public License Version 2. See the LICENSE file + * at the top of the source tree. + */ + +/*! + * \file + * \author Russell Bryant <russell@digium.com> + * + * \brief Usage of the SAForum AIS (Application Interface Specification) + * + * \arg http://www.openais.org/ + * + * This file contains the code specific to the use of the CKPT (Checkpoint) + * service. + */ + +#include "asterisk.h" + +ASTERISK_FILE_VERSION(__FILE__, "$Revision$"); + +#include <stdlib.h> +#include <stdio.h> +#include <string.h> +#include <unistd.h> +#include <errno.h> + +#include "ais.h" + +#include "asterisk/module.h" +#include "asterisk/utils.h" +#include "asterisk/cli.h" +#include "asterisk/logger.h" + +SaCkptHandleT ckpt_handle; + +static const SaCkptCallbacksT ckpt_callbacks; + +int ast_ais_ckpt_load_module(void) +{ + SaAisErrorT ais_res; + + ais_res = saCkptInitialize(&ckpt_handle, &ckpt_callbacks, &ais_version); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Could not initialize CKPT service: %s\n", + ais_err2str(ais_res)); + return -1; + } + + return 0; +} + +int ast_ais_ckpt_unload_module(void) +{ + SaAisErrorT ais_res; + + ais_res = saCkptFinalize(amf_handle); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Problem stopping CKPT service: %s\n", + ais_err2str(ais_res)); + return -1; + } + + return 0; +} diff --git a/res/ais/clm.c b/res/ais/clm.c new file mode 100644 index 000000000..5d7a356be --- /dev/null +++ b/res/ais/clm.c @@ -0,0 +1,165 @@ +/* + * Asterisk -- An open source telephony toolkit. + * + * Copyright (C) 2007, Digium, Inc. + * + * Russell Bryant <russell@digium.com> + * + * See http://www.asterisk.org for more information about + * the Asterisk project. Please do not directly contact + * any of the maintainers of this project for assistance; + * the project provides a web site, mailing lists and IRC + * channels for your use. + * + * This program is free software, distributed under the terms of + * the GNU General Public License Version 2. See the LICENSE file + * at the top of the source tree. + */ + +/*! + * \file + * \author Russell Bryant <russell@digium.com> + * + * \brief Usage of the SAForum AIS (Application Interface Specification) + * + * \arg http://www.openais.org/ + * + * This file contains the code specific to the use of the CLM + * (Cluster Membership) Service. + */ + +#include "asterisk.h" + +ASTERISK_FILE_VERSION(__FILE__, "$Revision$"); + +#include <stdlib.h> +#include <stdio.h> +#include <string.h> +#include <unistd.h> +#include <errno.h> + +#include "ais.h" + +#include "asterisk/module.h" +#include "asterisk/utils.h" +#include "asterisk/cli.h" +#include "asterisk/logger.h" + +SaClmHandleT clm_handle; + +static void clm_node_get_cb(SaInvocationT invocation, + const SaClmClusterNodeT *cluster_node, SaAisErrorT error); +static void clm_track_cb(const SaClmClusterNotificationBufferT *notif_buffer, + SaUint32T num_members, SaAisErrorT error); + +static const SaClmCallbacksT clm_callbacks = { + .saClmClusterNodeGetCallback = clm_node_get_cb, + .saClmClusterTrackCallback = clm_track_cb, +}; + +static void clm_node_get_cb(SaInvocationT invocation, + const SaClmClusterNodeT *cluster_node, SaAisErrorT error) +{ + +} + +static void clm_track_cb(const SaClmClusterNotificationBufferT *notif_buffer, + SaUint32T num_members, SaAisErrorT error) +{ + +} + +static char *ais_clm_show_members(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a) +{ + int i; + SaClmClusterNotificationBufferT buf; + SaClmClusterNotificationT notif[64]; + SaAisErrorT ais_res; + + switch (cmd) { + case CLI_INIT: + e->command = "ais clm show members"; + e->usage = + "Usage: ais clm show members\n" + " List members of the cluster using the CLM (Cluster Membership) service.\n"; + return NULL; + + case CLI_GENERATE: + return NULL; /* no completion */ + } + + if (a->argc != e->args) + return CLI_SHOWUSAGE; + + buf.notification = notif; + buf.numberOfItems = ARRAY_LEN(notif); + + ais_res = saClmClusterTrack(clm_handle, SA_TRACK_CURRENT, &buf); + if (ais_res != SA_AIS_OK) { + ast_cli(a->fd, "Error retrieving current cluster members.\n"); + return CLI_FAILURE; + } + + ast_cli(a->fd, "\n" + "=============================================================\n" + "=== Cluster Members =========================================\n" + "=============================================================\n" + "===\n"); + + for (i = 0; i < buf.numberOfItems; i++) { + SaClmClusterNodeT *node = &buf.notification[i].clusterNode; + + ast_cli(a->fd, "=== ---------------------------------------------------------\n" + "=== Node Name: %s\n" + "=== ==> ID: 0x%x\n" + "=== ==> Address: %s\n" + "=== ==> Member: %s\n", + (char *) node->nodeName.value, (int) node->nodeId, + (char *) node->nodeAddress.value, + node->member ? "Yes" : "No"); + + ast_cli(a->fd, "=== ---------------------------------------------------------\n" + "===\n"); + } + + ast_cli(a->fd, "=============================================================\n" + "\n"); + + return CLI_SUCCESS; +} + +static struct ast_cli_entry ais_cli[] = { + AST_CLI_DEFINE(ais_clm_show_members, "List current members of the cluster"), +}; + +int ast_ais_clm_load_module(void) +{ + SaAisErrorT ais_res; + + ais_res = saClmInitialize(&clm_handle, &clm_callbacks, &ais_version); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Could not initialize cluster membership service: %s\n", + ais_err2str(ais_res)); + return -1; + } + + ast_cli_register_multiple(ais_cli, ARRAY_LEN(ais_cli)); + + return 0; +} + +int ast_ais_clm_unload_module(void) +{ + SaAisErrorT ais_res; + + ast_cli_unregister_multiple(ais_cli, ARRAY_LEN(ais_cli)); + + ais_res = saClmFinalize(clm_handle); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Problem stopping cluster membership service: %s\n", + ais_err2str(ais_res)); + return -1; + } + + return 0; +} diff --git a/res/ais/evt.c b/res/ais/evt.c new file mode 100644 index 000000000..0057f0481 --- /dev/null +++ b/res/ais/evt.c @@ -0,0 +1,588 @@ +/* + * Asterisk -- An open source telephony toolkit. + * + * Copyright (C) 2007, Digium, Inc. + * + * Russell Bryant <russell@digium.com> + * + * See http://www.asterisk.org for more information about + * the Asterisk project. Please do not directly contact + * any of the maintainers of this project for assistance; + * the project provides a web site, mailing lists and IRC + * channels for your use. + * + * This program is free software, distributed under the terms of + * the GNU General Public License Version 2. See the LICENSE file + * at the top of the source tree. + */ + +/*! + * \file + * \author Russell Bryant <russell@digium.com> + * + * \brief Usage of the SAForum AIS (Application Interface Specification) + * + * \arg http://www.openais.org/ + * + * This file contains the code specific to the use of the EVT + * (Event) Service. + */ + +#include "asterisk.h" + +ASTERISK_FILE_VERSION(__FILE__, "$Revision$"); + +#include <stdlib.h> +#include <stdio.h> +#include <string.h> +#include <unistd.h> +#include <errno.h> + +#include "ais.h" + +#include "asterisk/module.h" +#include "asterisk/utils.h" +#include "asterisk/cli.h" +#include "asterisk/logger.h" +#include "asterisk/event.h" +#include "asterisk/config.h" +#include "asterisk/linkedlists.h" + +#ifndef AST_MODULE +/* XXX HACK */ +#define AST_MODULE "res_ais" +#endif + +SaEvtHandleT evt_handle; + +void evt_channel_open_cb(SaInvocationT invocation, SaEvtChannelHandleT channel_handle, + SaAisErrorT error); +void evt_event_deliver_cb(SaEvtSubscriptionIdT subscription_id, + const SaEvtEventHandleT event_handle, const SaSizeT event_datalen); + +static const SaEvtCallbacksT evt_callbacks = { + .saEvtChannelOpenCallback = evt_channel_open_cb, + .saEvtEventDeliverCallback = evt_event_deliver_cb, +}; + +static const struct { + const char *str; + enum ast_event_type type; +} supported_event_types[] = { + { "mwi", AST_EVENT_MWI }, + { "device_state", AST_EVENT_DEVICE_STATE_CHANGE }, +}; + +/*! Used to provide unique id's to egress subscriptions */ +static int unique_id; + +struct subscribe_event { + AST_LIST_ENTRY(subscribe_event) entry; + /*! This is a unique identifier to identify this subscription in the event + * channel through the different API calls, subscribe, unsubscribe, and + * the event deliver callback. */ + SaEvtSubscriptionIdT id; + enum ast_event_type type; +}; + +struct publish_event { + AST_LIST_ENTRY(publish_event) entry; + /*! We subscribe to events internally so that we can publish them + * on this event channel. */ + struct ast_event_sub *sub; + enum ast_event_type type; +}; + +struct event_channel { + AST_RWLIST_ENTRY(event_channel) entry; + AST_LIST_HEAD_NOLOCK(, subscribe_event) subscribe_events; + AST_LIST_HEAD_NOLOCK(, publish_event) publish_events; + SaEvtChannelHandleT handle; + char name[1]; +}; + +static AST_RWLIST_HEAD_STATIC(event_channels, event_channel); + +void evt_channel_open_cb(SaInvocationT invocation, SaEvtChannelHandleT channel_handle, + SaAisErrorT error) +{ + +} + +static void queue_event(struct ast_event *ast_event) +{ + enum ast_event_type type; + + /*! + * \todo This hack macks me sad. I need to come up with a better way to + * figure out whether an event should be cached or not, and what + * parameters to cache on. + * + * As long as the types of events that are supported is limited, + * this isn't *terrible*, I guess. Perhaps we should just define + * caching rules in the core, and make them configurable, and not + * have it be the job of the event publishers. + */ + + type = ast_event_get_type(ast_event); + + if (type == AST_EVENT_MWI) { + ast_event_queue_and_cache(ast_event, + AST_EVENT_IE_MAILBOX, AST_EVENT_IE_PLTYPE_STR, + AST_EVENT_IE_CONTEXT, AST_EVENT_IE_PLTYPE_STR, + AST_EVENT_IE_END); + } else if (type == AST_EVENT_DEVICE_STATE_CHANGE) { + ast_event_queue_and_cache(ast_event, + AST_EVENT_IE_DEVICE, AST_EVENT_IE_PLTYPE_STR, + AST_EVENT_IE_EID, AST_EVENT_IE_PLTYPE_RAW, sizeof(struct ast_eid), + AST_EVENT_IE_END); + } else { + ast_event_queue(ast_event); + } +} + +void evt_event_deliver_cb(SaEvtSubscriptionIdT sub_id, + const SaEvtEventHandleT event_handle, const SaSizeT event_datalen) +{ + /* It is important to note that this works because we *know* that this + * function will only be called by a single thread, the dispatch_thread. + * If this module gets changed such that this is no longer the case, this + * should get changed to a thread-local buffer, instead. */ + static unsigned char buf[4096]; + struct ast_event *event_dup, *event = (void *) buf; + SaAisErrorT ais_res; + SaSizeT len = sizeof(buf); + + if (event_datalen > len) { + ast_log(LOG_ERROR, "Event received with size %u, which is too big\n" + "for the allocated size %u. Change the code to increase the size.\n", + (unsigned int) event_datalen, (unsigned int) len); + return; + } + + ais_res = saEvtEventDataGet(event_handle, event, &len); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Error retrieving event payload: %s\n", + ais_err2str(ais_res)); + return; + } + + if (!ast_eid_cmp(&g_eid, ast_event_get_ie_raw(event, AST_EVENT_IE_EID))) { + /* Don't feed events back in that originated locally. */ + return; + } + + if (!(event_dup = ast_malloc(len))) + return; + + memcpy(event_dup, event, len); + + queue_event(event_dup); +} + +static const char *type_to_filter_str(enum ast_event_type type) +{ + const char *filter_str = NULL; + int i; + + for (i = 0; i < ARRAY_LEN(supported_event_types); i++) { + if (supported_event_types[i].type == type) { + filter_str = supported_event_types[i].str; + break; + } + } + + return filter_str; +} + +static void ast_event_cb(const struct ast_event *ast_event, void *data) +{ + SaEvtEventHandleT event_handle; + SaAisErrorT ais_res; + struct event_channel *event_channel = data; + SaClmClusterNodeT local_node; + SaEvtEventPatternArrayT pattern_array; + SaEvtEventPatternT pattern; + SaSizeT len; + const char *filter_str; + SaEvtEventIdT event_id; + + ast_log(LOG_DEBUG, "Got an event to forward\n"); + + if (ast_eid_cmp(&g_eid, ast_event_get_ie_raw(ast_event, AST_EVENT_IE_EID))) { + /* If the event didn't originate from this server, don't send it back out. */ + ast_log(LOG_DEBUG, "Returning here\n"); + return; + } + + ais_res = saEvtEventAllocate(event_channel->handle, &event_handle); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Error allocating event: %s\n", ais_err2str(ais_res)); + ast_log(LOG_DEBUG, "Returning here\n"); + return; + } + + ais_res = saClmClusterNodeGet(clm_handle, SA_CLM_LOCAL_NODE_ID, + SA_TIME_ONE_SECOND, &local_node); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Error getting local node name: %s\n", ais_err2str(ais_res)); + goto return_event_free; + } + + filter_str = type_to_filter_str(ast_event_get_type(ast_event)); + len = strlen(filter_str) + 1; + pattern.pattern = (SaUint8T *) filter_str; + pattern.patternSize = len; + pattern.allocatedSize = len; + + pattern_array.allocatedNumber = 1; + pattern_array.patternsNumber = 1; + pattern_array.patterns = &pattern; + + /*! + * /todo Make retention time configurable + * /todo Make event priorities configurable + */ + ais_res = saEvtEventAttributesSet(event_handle, &pattern_array, + SA_EVT_LOWEST_PRIORITY, SA_TIME_ONE_MINUTE, &local_node.nodeName); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Error setting event attributes: %s\n", ais_err2str(ais_res)); + goto return_event_free; + } + + ais_res = saEvtEventPublish(event_handle, + ast_event, ast_event_get_size(ast_event), &event_id); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Error publishing event: %s\n", ais_err2str(ais_res)); + goto return_event_free; + } + +return_event_free: + ais_res = saEvtEventFree(event_handle); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Error freeing allocated event: %s\n", ais_err2str(ais_res)); + } + ast_log(LOG_DEBUG, "Returning here (event_free)\n"); +} + +static char *ais_evt_show_event_channels(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a) +{ + struct event_channel *event_channel; + + switch (cmd) { + case CLI_INIT: + e->command = "ais evt show event channels"; + e->usage = + "Usage: ais evt show event channels\n" + " List configured event channels for the (EVT) Eventing service.\n"; + return NULL; + + case CLI_GENERATE: + return NULL; /* no completion */ + } + + if (a->argc != e->args) + return CLI_SHOWUSAGE; + + ast_cli(a->fd, "\n" + "=============================================================\n" + "=== Event Channels ==========================================\n" + "=============================================================\n" + "===\n"); + + AST_RWLIST_RDLOCK(&event_channels); + AST_RWLIST_TRAVERSE(&event_channels, event_channel, entry) { + struct publish_event *publish_event; + struct subscribe_event *subscribe_event; + + ast_cli(a->fd, "=== ---------------------------------------------------------\n" + "=== Event Channel Name: %s\n", event_channel->name); + + AST_LIST_TRAVERSE(&event_channel->publish_events, publish_event, entry) { + ast_cli(a->fd, "=== ==> Publishing Event Type: %s\n", + type_to_filter_str(publish_event->type)); + } + + AST_LIST_TRAVERSE(&event_channel->subscribe_events, subscribe_event, entry) { + ast_cli(a->fd, "=== ==> Subscribing to Event Type: %s\n", + type_to_filter_str(subscribe_event->type)); + } + + ast_cli(a->fd, "=== ---------------------------------------------------------\n" + "===\n"); + } + AST_RWLIST_UNLOCK(&event_channels); + + ast_cli(a->fd, "=============================================================\n" + "\n"); + + return CLI_SUCCESS; +} + +static struct ast_cli_entry ais_cli[] = { + AST_CLI_DEFINE(ais_evt_show_event_channels, "Show configured event channels"), +}; + +static void add_publish_event(struct event_channel *event_channel, const char *event_type) +{ + int i; + enum ast_event_type type = -1; + struct publish_event *publish_event; + + for (i = 0; i < ARRAY_LEN(supported_event_types); i++) { + if (!strcasecmp(event_type, supported_event_types[i].str)) { + type = supported_event_types[i].type; + break; + } + } + + if (type == -1) { + ast_log(LOG_WARNING, "publish_event option given with invalid value '%s'\n", event_type); + return; + } + + if (!(publish_event = ast_calloc(1, sizeof(*publish_event)))) + return; + + publish_event->type = type; + ast_log(LOG_DEBUG, "Subscribing to event type %d\n", type); + publish_event->sub = ast_event_subscribe(type, ast_event_cb, event_channel, + AST_EVENT_IE_END); + ast_event_dump_cache(publish_event->sub); + + AST_LIST_INSERT_TAIL(&event_channel->publish_events, publish_event, entry); +} + +static SaAisErrorT set_egress_subscription(struct event_channel *event_channel, + struct subscribe_event *subscribe_event) +{ + SaAisErrorT ais_res; + SaEvtEventFilterArrayT filter_array; + SaEvtEventFilterT filter; + const char *filter_str = NULL; + SaSizeT len; + + /* We know it's going to be valid. It was checked earlier. */ + filter_str = type_to_filter_str(subscribe_event->type); + + filter.filterType = SA_EVT_EXACT_FILTER; + len = strlen(filter_str) + 1; + filter.filter.allocatedSize = len; + filter.filter.patternSize = len; + filter.filter.pattern = (SaUint8T *) filter_str; + + filter_array.filtersNumber = 1; + filter_array.filters = &filter; + + ais_res = saEvtEventSubscribe(event_channel->handle, &filter_array, + subscribe_event->id); + + return ais_res; +} + +static void add_subscribe_event(struct event_channel *event_channel, const char *event_type) +{ + int i; + enum ast_event_type type = -1; + struct subscribe_event *subscribe_event; + SaAisErrorT ais_res; + + for (i = 0; i < ARRAY_LEN(supported_event_types); i++) { + if (!strcasecmp(event_type, supported_event_types[i].str)) { + type = supported_event_types[i].type; + break; + } + } + + if (type == -1) { + ast_log(LOG_WARNING, "subscribe_event option given with invalid value '%s'\n", event_type); + return; + } + + if (!(subscribe_event = ast_calloc(1, sizeof(*subscribe_event)))) + return; + + subscribe_event->type = type; + subscribe_event->id = ast_atomic_fetchadd_int(&unique_id, +1); + + ais_res = set_egress_subscription(event_channel, subscribe_event); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Error setting up egress subscription: %s\n", + ais_err2str(ais_res)); + free(subscribe_event); + return; + } + + AST_LIST_INSERT_TAIL(&event_channel->subscribe_events, subscribe_event, entry); +} + +static void build_event_channel(struct ast_config *cfg, const char *cat) +{ + struct ast_variable *var; + struct event_channel *event_channel; + SaAisErrorT ais_res; + SaNameT sa_name = { 0, }; + + AST_RWLIST_WRLOCK(&event_channels); + AST_RWLIST_TRAVERSE(&event_channels, event_channel, entry) { + if (!strcasecmp(event_channel->name, cat)) + break; + } + AST_RWLIST_UNLOCK(&event_channels); + if (event_channel) { + ast_log(LOG_WARNING, "Event channel '%s' was specified twice in " + "configuration. Second instance ignored.\n", cat); + return; + } + + if (!(event_channel = ast_calloc(1, sizeof(*event_channel) + strlen(cat)))) + return; + + strcpy(event_channel->name, cat); + ast_copy_string((char *) sa_name.value, cat, sizeof(sa_name.value)); + sa_name.length = strlen((char *) sa_name.value); + ais_res = saEvtChannelOpen(evt_handle, &sa_name, + SA_EVT_CHANNEL_PUBLISHER | SA_EVT_CHANNEL_SUBSCRIBER | SA_EVT_CHANNEL_CREATE, + SA_TIME_MAX, &event_channel->handle); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Error opening event channel: %s\n", ais_err2str(ais_res)); + free(event_channel); + return; + } + + for (var = ast_variable_browse(cfg, cat); var; var = var->next) { + if (!strcasecmp(var->name, "type")) { + continue; + } else if (!strcasecmp(var->name, "publish_event")) { + add_publish_event(event_channel, var->value); + } else if (!strcasecmp(var->name, "subscribe_event")) { + add_subscribe_event(event_channel, var->value); + } else { + ast_log(LOG_WARNING, "Event channel '%s' contains invalid option '%s'\n", + event_channel->name, var->name); + } + } + + AST_RWLIST_WRLOCK(&event_channels); + AST_RWLIST_INSERT_TAIL(&event_channels, event_channel, entry); + AST_RWLIST_UNLOCK(&event_channels); +} + +static void load_config(void) +{ + static const char filename[] = "ais.conf"; + struct ast_config *cfg; + const char *cat = NULL; + struct ast_flags config_flags = { 0 }; + + if (!(cfg = ast_config_load(filename, config_flags))) + return; + + while ((cat = ast_category_browse(cfg, cat))) { + const char *type; + + if (!strcasecmp(cat, "general")) + continue; + + if (!(type = ast_variable_retrieve(cfg, cat, "type"))) { + ast_log(LOG_WARNING, "Invalid entry in %s defined with no type!\n", + filename); + continue; + } + + if (!strcasecmp(type, "event_channel")) { + build_event_channel(cfg, cat); + } else { + ast_log(LOG_WARNING, "Entry in %s defined with invalid type '%s'\n", + filename, type); + } + } + + ast_config_destroy(cfg); +} + +static void publish_event_destroy(struct publish_event *publish_event) +{ + ast_event_unsubscribe(publish_event->sub); + + free(publish_event); +} + +static void subscribe_event_destroy(const struct event_channel *event_channel, + struct subscribe_event *subscribe_event) +{ + SaAisErrorT ais_res; + + /* saEvtChannelClose() will actually do this automatically, but it just + * feels cleaner to go ahead and do it manually ... */ + ais_res = saEvtEventUnsubscribe(event_channel->handle, subscribe_event->id); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Error unsubscribing: %s\n", ais_err2str(ais_res)); + } + + free(subscribe_event); +} + +static void event_channel_destroy(struct event_channel *event_channel) +{ + struct publish_event *publish_event; + struct subscribe_event *subscribe_event; + SaAisErrorT ais_res; + + while ((publish_event = AST_LIST_REMOVE_HEAD(&event_channel->publish_events, entry))) + publish_event_destroy(publish_event); + while ((subscribe_event = AST_LIST_REMOVE_HEAD(&event_channel->subscribe_events, entry))) + subscribe_event_destroy(event_channel, subscribe_event); + + ais_res = saEvtChannelClose(event_channel->handle); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Error closing event channel '%s': %s\n", + event_channel->name, ais_err2str(ais_res)); + } + + free(event_channel); +} + +static void destroy_event_channels(void) +{ + struct event_channel *event_channel; + + AST_RWLIST_WRLOCK(&event_channels); + while ((event_channel = AST_RWLIST_REMOVE_HEAD(&event_channels, entry))) + event_channel_destroy(event_channel); + AST_RWLIST_UNLOCK(&event_channels); +} + +int ast_ais_evt_load_module(void) +{ + SaAisErrorT ais_res; + + ais_res = saEvtInitialize(&evt_handle, &evt_callbacks, &ais_version); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Could not initialize eventing service: %s\n", + ais_err2str(ais_res)); + return -1; + } + + load_config(); + + ast_cli_register_multiple(ais_cli, ARRAY_LEN(ais_cli)); + + return 0; +} + +int ast_ais_evt_unload_module(void) +{ + SaAisErrorT ais_res; + + destroy_event_channels(); + + ais_res = saEvtFinalize(evt_handle); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Problem stopping eventing service: %s\n", + ais_err2str(ais_res)); + return -1; + } + + return 0; +} diff --git a/res/ais/lck.c b/res/ais/lck.c new file mode 100644 index 000000000..7e7533dc6 --- /dev/null +++ b/res/ais/lck.c @@ -0,0 +1,551 @@ +/* + * Asterisk -- An open source telephony toolkit. + * + * Copyright (C) 2007, Digium, Inc. + * + * Russell Bryant <russell@digium.com> + * + * See http://www.asterisk.org for more information about + * the Asterisk project. Please do not directly contact + * any of the maintainers of this project for assistance; + * the project provides a web site, mailing lists and IRC + * channels for your use. + * + * This program is free software, distributed under the terms of + * the GNU General Public License Version 2. See the LICENSE file + * at the top of the source tree. + */ + +/*! + * \file + * \author Russell Bryant <russell@digium.com> + * + * \brief Usage of the SAForum AIS (Application Interface Specification) + * + * \arg http://www.openais.org/ + * + * This file contains the code specific to the use of the LCK + * (Distributed Locks) Service. + */ + +#include "asterisk.h" + +ASTERISK_FILE_VERSION(__FILE__, "$Revision$"); + +#include <stdlib.h> +#include <stdio.h> +#include <string.h> +#include <unistd.h> +#include <errno.h> + +#include "ais.h" + +#include "asterisk/module.h" +#include "asterisk/utils.h" +#include "asterisk/cli.h" +#include "asterisk/logger.h" +#include "asterisk/pbx.h" +#include "asterisk/app.h" +#include "asterisk/astobj2.h" +#include "asterisk/strings.h" +#include "asterisk/channel.h" + +SaLckHandleT lck_handle; + +/*! + * \brief Callbacks available in the Lock Service + * + * None of these are actually required if only synchronous locking is used. + * However, some of them must be implemented should the asynchronous locks + * be used. + */ +static SaLckCallbacksT lck_callbacks = { + /*! Get notified when a cluster-wide lock gets created */ + .saLckResourceOpenCallback = NULL, + /*! Get notified when an asynchronous lock request gets granted */ + .saLckLockGrantCallback = NULL, + /*! Be informed when a currently held lock is blocking another node */ + .saLckLockWaiterCallback = NULL, + /*! Get notified when an asynchronous unlock request is done */ + .saLckResourceUnlockCallback = NULL, +}; + +enum lock_type { + RDLOCK, + WRLOCK, + TRY_RDLOCK, + TRY_WRLOCK, +}; + +#define LOCK_BUCKETS 101 + +/*! + * Every thread that wants to use a distributed lock must open its own handle + * to the lock. So, a thread-local container of opened locks is used to keep + * track of what locks have been opened. + * + * \todo It would be nice to be able to have a thread-local container, instead + * of using a thread-local wrapper like this. + */ +struct lock_resources { + struct ao2_container *locks; +}; + +static int lock_resources_init(void *); +static void lock_resources_destroy(void *); + +AST_THREADSTORAGE_CUSTOM(locks_ts_key, + lock_resources_init, lock_resources_destroy); + +struct lock_resource { + SaLckResourceHandleT handle; + SaLckLockIdT id; + SaNameT ais_name; + const char *name; +}; + +static int lock_hash_cb(const void *obj, int flags) +{ + const struct lock_resource *lock = obj; + + return ast_str_hash(lock->name); +} + +static int lock_cmp_cb(void *obj, void *arg, int flags) +{ + struct lock_resource *lock1 = obj, *lock2 = arg; + + return !strcasecmp(lock1->name, lock2->name) ? CMP_MATCH : 0; +} + +static int lock_resources_init(void *data) +{ + struct lock_resources *lock_resources = data; + + if (!(lock_resources->locks = ao2_container_alloc(LOCK_BUCKETS, + lock_hash_cb, lock_cmp_cb))) { + return -1; + } + + return 0; + +} + +static void lock_resources_destroy(void *data) +{ + struct lock_resources *lock_resources = data; + + ao2_ref(lock_resources->locks, -1); + + ast_free(lock_resources); +} + +static void lock_destructor(void *obj) +{ + struct lock_resource *lock = obj; + + if (lock->name) + ast_free((void *) lock->name); +} + +static inline struct lock_resource *lock_ref(struct lock_resource *lock) +{ + ao2_ref(lock, +1); + return lock; +} + +static inline struct lock_resource *lock_unref(struct lock_resource *lock) +{ + ao2_ref(lock, -1); + return NULL; +} + +static void lock_datastore_destroy(void *data) +{ + struct lock_resource *lock = data; + SaAisErrorT ais_res; + + ais_res = saLckResourceUnlock(lock->id, SA_TIME_ONE_SECOND * 3); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Error unlocking '%s': %s\n", lock->name, + ais_err2str(ais_res)); + } + + lock_unref(lock); +} + +static struct lock_resource *find_lock(const char *name) +{ + struct lock_resource *lock, tmp_lock = { + .name = name, + }; + SaAisErrorT ais_res; + struct lock_resources *lock_resources; + + if (!(lock_resources = ast_threadstorage_get(&locks_ts_key, + sizeof(*lock_resources)))) { + return NULL; + } + + /* Return the lock if it has already been opened by this thread */ + if ((lock = ao2_find(lock_resources->locks, &tmp_lock, OBJ_POINTER))) + return lock; + + /* Allocate and open the lock */ + if (!(lock = ao2_alloc(sizeof(*lock), lock_destructor))) + return NULL; + + if (!(lock->name = ast_strdup(name))) + return lock_unref(lock); + + /* Map the name into the SaNameT for convenience */ + ast_copy_string((char *) lock->ais_name.value, lock->name, + sizeof(lock->ais_name.value)); + lock->ais_name.length = strlen(lock->name); + + ais_res = saLckResourceOpen(lck_handle, &lock->ais_name, + SA_LCK_RESOURCE_CREATE, SA_TIME_ONE_SECOND * 3, &lock->handle); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Failed to open lock: %s\n", ais_err2str(ais_res)); + return lock_unref(lock); + } + + return lock; +} + +const struct ast_datastore_info dlock_datastore_info = { + .type = "DLOCK", + .destroy = lock_datastore_destroy, +}; + +static void add_lock_to_chan(struct ast_channel *chan, struct lock_resource *lock, + enum lock_type lock_type, double timeout, char *buf, size_t len) +{ + struct ast_datastore *datastore; + SaAisErrorT ais_res; + SaLckLockModeT mode = 0; + SaLckLockFlagsT flags = 0; + SaLckLockStatusT status; + + ast_channel_lock(chan); + datastore = ast_channel_datastore_find(chan, &dlock_datastore_info, lock->name); + + if (datastore) { + ast_log(LOG_ERROR, "The DLOCK '%s' is already locked by channel '%s'\n", + lock->name, chan->name); + ast_channel_unlock(chan); + ast_copy_string(buf, "FAILURE", len); + return; + } + ast_channel_unlock(chan); + + switch (lock_type) { + case TRY_RDLOCK: + flags = SA_LCK_LOCK_NO_QUEUE; + mode = SA_LCK_PR_LOCK_MODE; + break; + case RDLOCK: + flags = SA_LCK_LOCK_NO_QUEUE; + mode = SA_LCK_PR_LOCK_MODE; + break; + case TRY_WRLOCK: + flags = SA_LCK_LOCK_NO_QUEUE; + mode = SA_LCK_EX_LOCK_MODE; + break; + case WRLOCK: + flags = SA_LCK_LOCK_NO_QUEUE; + mode = SA_LCK_EX_LOCK_MODE; + break; + } + + /* Actually acquire the lock now */ + ais_res = saLckResourceLock(lock->handle, &lock->id, mode, flags, 0, + (SaTimeT) timeout * SA_TIME_ONE_SECOND, &status); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Problem acquiring lock '%s': %s\n", + lock->name, ais_err2str(ais_res)); + ast_copy_string(buf, (ais_res == SA_AIS_ERR_TIMEOUT) ? "TIMEOUT" : + "FAILURE", len); + return; + } + + switch (status) { + case SA_LCK_LOCK_GRANTED: + ast_copy_string(buf, "SUCCESS", len); + break; + case SA_LCK_LOCK_DEADLOCK: + ast_copy_string(buf, "DEADLOCK", len); + return; + /*! XXX \todo Need to look at handling these other cases in a different way */ + case SA_LCK_LOCK_NOT_QUEUED: + case SA_LCK_LOCK_ORPHANED: + case SA_LCK_LOCK_NO_MORE: + case SA_LCK_LOCK_DUPLICATE_EX: + ast_copy_string(buf, "FAILURE", len); + return; + } + + if (!(datastore = ast_channel_datastore_alloc(&dlock_datastore_info, + lock->name))) { + ast_copy_string(buf, "FAILURE", len); + return; + } + + datastore->data = lock_ref(lock); + + ast_channel_lock(chan); + ast_channel_datastore_add(chan, datastore); + ast_channel_unlock(chan); +} + +static int handle_lock(struct ast_channel *chan, enum lock_type lock_type, + char *data, char *buf, size_t len) +{ + AST_DECLARE_APP_ARGS(args, + AST_APP_ARG(name); + AST_APP_ARG(timeout); + ); + int res = 0; + double timeout = 3; + struct lock_resource *lock = NULL; + + ast_autoservice_start(chan); + + AST_STANDARD_APP_ARGS(args, data); + if (ast_strlen_zero(args.name)) { + ast_log(LOG_ERROR, "The DLOCK functions require a lock name\n"); + res = -1; + goto return_cleanup; + } + switch (lock_type) { + case RDLOCK: + case WRLOCK: + if (!ast_strlen_zero(args.timeout) && ((timeout = atof(args.timeout)) < 0)) { + ast_log(LOG_ERROR, "Timeout value '%s' not valid\n", args.timeout); + res = -1; + goto return_cleanup; + } + break; + case TRY_RDLOCK: + case TRY_WRLOCK: + if (!ast_strlen_zero(args.timeout)) { + ast_log(LOG_ERROR, "The trylock functions only take one argument\n"); + res = -1; + goto return_cleanup; + } + } + + if (!(lock = find_lock(args.name))) { + ast_copy_string(buf, "FAILURE", len); + res = -1; + goto return_cleanup; + } + + add_lock_to_chan(chan, lock, lock_type, timeout, buf, len); + + lock = lock_unref(lock); + +return_cleanup: + ast_autoservice_stop(chan); + + return res; +} + +static int handle_rdlock(struct ast_channel *chan, const char *cmd, char *data, + char *buf, size_t len) +{ + return handle_lock(chan, RDLOCK, data, buf, len); +} + +static int handle_wrlock(struct ast_channel *chan, const char *cmd, char *data, + char *buf, size_t len) +{ + return handle_lock(chan, WRLOCK, data, buf, len); +} + +static int handle_tryrdlock(struct ast_channel *chan, const char *cmd, char *data, + char *buf, size_t len) +{ + return handle_lock(chan, TRY_RDLOCK, data, buf, len); +} + +static int handle_trywrlock(struct ast_channel *chan, const char *cmd, char *data, + char *buf, size_t len) +{ + return handle_lock(chan, TRY_WRLOCK, data, buf, len); +} + +static int handle_unlock(struct ast_channel *chan, const char *cmd, char *data, + char *buf, size_t len) +{ + struct ast_datastore *datastore; + struct lock_resource *lock; + SaAisErrorT ais_res; + int res = 0; + + if (ast_strlen_zero(data)) { + ast_log(LOG_ERROR, "DLOCK_UNLOCK requires a lock name\n"); + ast_copy_string(buf, "FAILURE", len); + return -1; + } + + ast_autoservice_start(chan); + + ast_channel_lock(chan); + datastore = ast_channel_datastore_find(chan, &dlock_datastore_info, data); + if (!datastore) { + ast_log(LOG_ERROR, "The DLOCK '%s' is not locked by channel '%s'\n", + data, chan->name); + ast_channel_unlock(chan); + ast_copy_string(buf, "FAILURE", len); + return -1; + } + ast_channel_datastore_remove(chan, datastore); + ast_channel_unlock(chan); + + lock = datastore->data; + ais_res = saLckResourceUnlock(lock->id, SA_TIME_ONE_SECOND * 3); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Error unlocking '%s': %s\n", lock->name, + ais_err2str(ais_res)); + res = -1; + ast_copy_string(buf, (ais_res == SA_AIS_ERR_TIMEOUT) ? "TIMEOUT" : + "FAILURE", len); + } else { + ast_copy_string(buf, "SUCCESS", len); + } + + datastore->data = lock_unref(lock); + ast_channel_datastore_free(datastore); + + ast_autoservice_stop(chan); + + return res; +} + +#define LOCK_DESC_COMMON \ +" The name of the lock can be anything. The first time a named lock gets\n" \ +"used, it will be automatically created and maintained amongst the cluster.\n" \ +" The result of this function will be one of the following:\n" \ +" SUCCESS | TIMEOUT | FAILURE | DEADLOCK\n" DEADLOCK_DESC + +#define DEADLOCK_DESC \ +" The result, DEADLOCK, can only be provided if the AIS implementation in\n" \ +"use provides the optional feature of deadlock detection. If the lock fails\n" \ +"with the result of DEADLOCK, it means that the AIS implementation has\n" \ +"determined that if this lock were acquired, it would cause a deadlock.\n" + +static struct ast_custom_function dlock_rdlock = { + .name = "DLOCK_RDLOCK", + .synopsis = "Read-lock a distributed lock", + .desc = +" This function will read-lock a distributed lock provided by the locking\n" +"service of AIS. This is a blocking operation. However, a timeout can be\n" +"specified to avoid deadlocks. The default timeout used if one is not\n" +"provided as an argument is 3 seconds.\n" +LOCK_DESC_COMMON +"", + .syntax = "DLOCK_RDLOCK(<lock_name>,[timeout])", + .read = handle_rdlock, +}; + +static struct ast_custom_function dlock_wrlock = { + .name = "DLOCK_WRLOCK", + .synopsis = "Write-lock a distributed lock", + .desc = +" This function will write-lock a distributed lock provided by the locking\n" +"service of AIS. This is a blocking operation. However, a timeout can be\n" +"specified to avoid deadlocks. The default timeout used if one is not\n" +"provided as an argument is 3 seconds.\n" +LOCK_DESC_COMMON +"", + .syntax = "DLOCK_WRLOCK(<lock_name>,[timeout])", + .read = handle_wrlock, +}; + +static struct ast_custom_function dlock_tryrdlock = { + .name = "DLOCK_TRYRDLOCK", + .synopsis = "Try to read-lock a distributed lock", + .desc = +" This function will attempt to read-lock a distributed lock provided by the\n" +"locking service of AIS. This is a non-blocking operation.\n" +" The name of the lock can be anything. The first time a named lock gets\n" +"used, it will be automatically created and maintained amongst the cluster.\n" +" The result of this function will be one of the following:\n" +" SUCCESS | FAILURE | DEADLOCK\n" +DEADLOCK_DESC +"", + .syntax = "DLOCK_TRYRDLOCK(<lock_name>)", + .read = handle_tryrdlock, +}; + +static struct ast_custom_function dlock_trywrlock = { + .name = "DLOCK_TRYWRLOCK", + .synopsis = "Try to write-lock a distributed lock", + .desc = +" This function will attempt to write-lock a distributed lock provided by\n" +"the locking service of AIS. This is a non-blocking operation.\n" +" The name of the lock can be anything. The first time a named lock gets\n" +"used, it will be automatically created and maintained amongst the cluster.\n" +" The result of this function will be one of the following:\n" +" SUCCESS | FAILURE | DEADLOCK\n" +DEADLOCK_DESC +"", + .syntax = "DLOCK_TRYWRLOCK(<lock_name>)", + .read = handle_trywrlock, +}; + +static struct ast_custom_function dlock_unlock = { + .name = "DLOCK_UNLOCK", + .synopsis = "Unlock a distributed lock", + .desc = +" This function will unlock a currently held distributed lock. This should\n" +"be used regardless of the lock was read or write locked. The result of\n" +"this funtion will be one of the following:\n" +" SUCCESS | TIMEOUT | FAILURE\n" +"", + .syntax = "DLOCK_UNLOCK(<lock_name>)", + .read = handle_unlock, +}; + +int ast_ais_lck_load_module(void) +{ + SaAisErrorT ais_res; + int res; + + ais_res = saLckInitialize(&lck_handle, &lck_callbacks, &ais_version); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Could not initialize distributed locking service: %s\n", + ais_err2str(ais_res)); + return -1; + } + + res = ast_custom_function_register(&dlock_rdlock); + res |= ast_custom_function_register(&dlock_wrlock); + res |= ast_custom_function_register(&dlock_tryrdlock); + res |= ast_custom_function_register(&dlock_trywrlock); + res |= ast_custom_function_register(&dlock_unlock); + + return res; +} + +int ast_ais_lck_unload_module(void) +{ + SaAisErrorT ais_res; + int res = 0; + + ast_custom_function_unregister(&dlock_rdlock); + ast_custom_function_unregister(&dlock_wrlock); + ast_custom_function_unregister(&dlock_tryrdlock); + ast_custom_function_unregister(&dlock_trywrlock); + ast_custom_function_unregister(&dlock_unlock); + + ais_res = saLckFinalize(lck_handle); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Problem stopping distributed locking service: %s\n", + ais_err2str(ais_res)); + res = -1; + } + + return res; +} diff --git a/res/res_ais.c b/res/res_ais.c new file mode 100644 index 000000000..884eb0d23 --- /dev/null +++ b/res/res_ais.c @@ -0,0 +1,193 @@ +/* + * Asterisk -- An open source telephony toolkit. + * + * Copyright (C) 2007 - 2008, Digium, Inc. + * + * Russell Bryant <russell@digium.com> + * + * See http://www.asterisk.org for more information about + * the Asterisk project. Please do not directly contact + * any of the maintainers of this project for assistance; + * the project provides a web site, mailing lists and IRC + * channels for your use. + * + * This program is free software, distributed under the terms of + * the GNU General Public License Version 2. See the LICENSE file + * at the top of the source tree. + */ + +/*! + * \file + * \author Russell Bryant <russell@digium.com> + * + * \brief Usage of the SAForum AIS (Application Interface Specification) + * + * \arg http://www.openais.org/ + * + * This file contains the common code between the uses of the different AIS + * services. + */ + +/*** MODULEINFO + <depend>SaClm</depend> + <depend>SaEvt</depend> + ***/ + +#include "asterisk.h" + +ASTERISK_FILE_VERSION(__FILE__, "$Revision$"); + +#include <stdlib.h> +#include <stdio.h> +#include <string.h> +#include <unistd.h> +#include <errno.h> +#include <signal.h> +#include <pthread.h> + +#include "ais/ais.h" + +#include "asterisk/module.h" +#include "asterisk/options.h" +#include "asterisk/logger.h" +#include "asterisk/channel.h" +#include "asterisk/utils.h" +#include "asterisk/cli.h" + +static struct { + pthread_t id; + unsigned int stop:1; +} dispatch_thread = { + .id = AST_PTHREADT_NULL, +}; + +SaVersionT ais_version = { 'B', 1, 1 }; + +static const struct ais_error { + SaAisErrorT error; + const char *desc; +} ais_errors[] = { + { SA_AIS_OK, "OK" }, + { SA_AIS_ERR_LIBRARY, "Library Error" }, + { SA_AIS_ERR_VERSION, "Version Not Compatible" }, + { SA_AIS_ERR_INIT, "Callback Not Registered" }, + { SA_AIS_ERR_TIMEOUT, "Timeout" }, + { SA_AIS_ERR_TRY_AGAIN , "Try Again" }, + { SA_AIS_ERR_INVALID_PARAM, "Invalid Parameter" }, + { SA_AIS_ERR_NO_MEMORY, "No Memory" }, + { SA_AIS_ERR_BAD_HANDLE, "Invalid Handle" }, + { SA_AIS_ERR_BUSY, "Resource Already In Use" }, + { SA_AIS_ERR_ACCESS, "Access Denied" }, + { SA_AIS_ERR_NOT_EXIST, "Does Not Exist" }, + { SA_AIS_ERR_NAME_TOO_LONG, "Name Too Long" }, + { SA_AIS_ERR_EXIST, "Already Exists" }, + { SA_AIS_ERR_NO_SPACE, "Buffer Too Small" }, + { SA_AIS_ERR_INTERRUPT, "Request Interrupted" }, + { SA_AIS_ERR_NAME_NOT_FOUND, "Name Not Found" }, + { SA_AIS_ERR_NO_RESOURCES, "Not Enough Resources" }, + { SA_AIS_ERR_NOT_SUPPORTED, "Requested Function Not Supported" }, + { SA_AIS_ERR_BAD_OPERATION, "Operation Not Allowed" }, + { SA_AIS_ERR_FAILED_OPERATION, "Operation Failed" }, + { SA_AIS_ERR_MESSAGE_ERROR, "Communication Error" }, + { SA_AIS_ERR_QUEUE_FULL, "Destination Queue Full" }, + { SA_AIS_ERR_QUEUE_NOT_AVAILABLE, "Destination Queue Not Available" }, + { SA_AIS_ERR_BAD_FLAGS, "Invalid Flags" }, + { SA_AIS_ERR_TOO_BIG, "Value Too Large" }, + { SA_AIS_ERR_NO_SECTIONS, "No More Sections to Initialize" }, +}; + +const char *ais_err2str(SaAisErrorT error) +{ + int x; + + for (x = 0; x < ARRAY_LEN(ais_errors); x++) { + if (ais_errors[x].error == error) + return ais_errors[x].desc; + } + + return "Unknown"; +} + +static void *dispatch_thread_handler(void *data) +{ + SaSelectionObjectT clm_fd, evt_fd, max_fd; + int res; + fd_set read_fds; + SaAisErrorT ais_res; + + ais_res = saClmSelectionObjectGet(clm_handle, &clm_fd); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Failed to retrieve select fd for CLM service. " + "This module will not operate.\n"); + return NULL; + } + + ais_res = saEvtSelectionObjectGet(evt_handle, &evt_fd); + if (ais_res != SA_AIS_OK) { + ast_log(LOG_ERROR, "Failed to retrieve select fd for EVT service. " + "This module will not operate.\n"); + return NULL; + } + + max_fd = clm_fd > evt_fd ? clm_fd : evt_fd; + + while (!dispatch_thread.stop) { + FD_ZERO(&read_fds); + FD_SET(clm_fd, &read_fds); + FD_SET(evt_fd, &read_fds); + + res = ast_select(max_fd + 1, &read_fds, NULL, NULL, NULL); + if (res == -1 && errno != EINTR && errno != EAGAIN) { + ast_log(LOG_ERROR, "Select error (%s) dispatch thread going away now, " + "and the module will no longer operate.\n", strerror(errno)); + break; + } + + if (FD_ISSET(clm_fd, &read_fds)) + saClmDispatch(clm_handle, SA_DISPATCH_ALL); + if (FD_ISSET(evt_fd, &read_fds)) + saEvtDispatch(evt_handle, SA_DISPATCH_ALL); + } + + return NULL; +} + +static int load_module(void) +{ + if (ast_ais_clm_load_module()) + goto return_error; + + if (ast_ais_evt_load_module()) + goto evt_failed; + + if (ast_pthread_create_background(&dispatch_thread.id, NULL, + dispatch_thread_handler, NULL)) { + ast_log(LOG_ERROR, "Error starting AIS dispatch thread.\n"); + goto dispatch_failed; + } + + return AST_MODULE_LOAD_SUCCESS; + +dispatch_failed: + ast_ais_evt_unload_module(); +evt_failed: + ast_ais_clm_unload_module(); +return_error: + return AST_MODULE_LOAD_DECLINE; +} + +static int unload_module(void) +{ + ast_ais_clm_unload_module(); + ast_ais_evt_unload_module(); + + if (dispatch_thread.id != AST_PTHREADT_NULL) { + dispatch_thread.stop = 1; + pthread_kill(dispatch_thread.id, SIGURG); /* poke! */ + pthread_join(dispatch_thread.id, NULL); + } + + return 0; +} + +AST_MODULE_INFO_STANDARD(ASTERISK_GPL_KEY, "SAForum AIS"); |