/* * Copyright (c) 2004-2005 Mark Haverkamp * Copyright (c) 2004-2005 Open Source Development Lab * * All rights reserved. * * This software licensed under BSD license, the text of which follows: * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions are met: * * - Redistributions of source code must retain the above copyright notice, * this list of conditions and the following disclaimer. * - Redistributions in binary form must reproduce the above copyright notice, * this list of conditions and the following disclaimer in the documentation * and/or other materials provided with the distribution. * - Neither the name of the Open Source Developement Lab nor the names of its * contributors may be used to endorse or promote products derived from this * software without specific prior written permission. * * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF * THE POSSIBILITY OF SUCH DAMAGE. */ #include #include #include #include #include #include #include #include #include "../include/ipc_evt.h" #include "util.h" #include "../exec/totempg.h" #include "../include/list.h" static void evtHandleInstanceDestructor(void *instance); static void chanHandleInstanceDestructor(void *instance); static void eventHandleInstanceDestructor(void *instance); /* * Versions of the SAF AIS specification supported by this library */ static SaVersionT supported_versions[] = { {'B', 0x01, 0x01} }; static struct saVersionDatabase evt_version_database = { sizeof(supported_versions) / sizeof(SaVersionT), supported_versions }; /* * Event instance data */ struct saHandleDatabase evt_instance_handle_db = { .handleCount = 0, .handles = 0, .mutex = PTHREAD_MUTEX_INITIALIZER, .handleInstanceDestructor = evtHandleInstanceDestructor }; /* * Channel instance data */ struct saHandleDatabase channel_handle_db = { .handleCount = 0, .handles = 0, .mutex = PTHREAD_MUTEX_INITIALIZER, .handleInstanceDestructor = chanHandleInstanceDestructor }; /* * Event instance data */ struct saHandleDatabase event_handle_db = { .handleCount = 0, .handles = 0, .mutex = PTHREAD_MUTEX_INITIALIZER, .handleInstanceDestructor = eventHandleInstanceDestructor }; struct res_overlay { struct res_header header; char data[MESSAGE_SIZE_MAX]; }; struct handle_list { SaUint64T hl_handle; struct list_head hl_entry; }; /* * data required to support events for a given initialization * * ei_dispatch_fd: fd used for getting callback data e.g. async event data * ei_response_fd: fd used for everything else (i.e. evt sync api commands). * ei_callback: callback function. * ei_version: version sent to the evtInitialize call. * ei_node_id: our node id. * ei_node_name: our node name. * ei_finalize: instance in finalize flag * ei_dispatch_mutex: mutex for dispatch fd * ei_response_mutex: mutex for response fd * ei_channel_list: list of associated channels (struct handle_list) * */ struct event_instance { int ei_dispatch_fd; int ei_response_fd; SaEvtCallbacksT ei_callback; SaVersionT ei_version; SaClmNodeIdT ei_node_id; SaNameT ei_node_name; int ei_finalize; pthread_mutex_t ei_dispatch_mutex; pthread_mutex_t ei_response_mutex; struct list_head ei_channel_list; }; /* * Data associated with an opened channel * * eci_channel_name: name of channel * eci_open_flags: channel open flags * eci_svr_channel_handle: channel handle returned from server * eci_closing: channel in process of being closed * eci_mutex: channel mutex * eci_event_list: events associated with this * channel (struct handle_list) * eci_hl: pointer to event instance handle struct * for this channel. */ struct event_channel_instance { SaNameT eci_channel_name; SaEvtChannelOpenFlagsT eci_open_flags; uint32_t eci_svr_channel_handle; SaEvtHandleT eci_instance_handle; int eci_closing; pthread_mutex_t eci_mutex; struct list_head eci_event_list; struct handle_list *eci_hl; }; /* * Event data. * * Store event data from saEvtEventAllocate function. * Store event data from received events. * * edi_channel_handle: handle (local) of assocated channel * edi_patterns: event patterns * edi_priority: event priority * edi_retention_time: event's retention time * edi_pub_name: event's publisher name * edi_pub_node: event's publisher node * edi_pub_time: event's publish time * edi_event_id: event's Id * edi_event_data: event's data * edi_event_data_size: size of edi_event_data * edi_freeing: event is being freed * edi_mutex: event data mutex * edi_hl: pointer to channel's handle * struct for this event. */ struct event_data_instance { SaEvtChannelHandleT edi_channel_handle; SaEvtEventPatternArrayT edi_patterns; SaUint8T edi_priority; SaTimeT edi_retention_time; SaNameT edi_pub_name; SaClmNodeIdT edi_pub_node; SaTimeT edi_pub_time; SaEvtEventIdT edi_event_id; void *edi_event_data; SaSizeT edi_event_data_size; int edi_freeing; pthread_mutex_t edi_mutex; struct handle_list *edi_hl; }; #define min(a,b) ((a) < (b) ? (a) : (b)) /* * Clean up function for an evt instance (saEvtInitialize) handle * Not to be confused with event data. */ static void evtHandleInstanceDestructor(void *instance) { struct event_instance *evti = instance; struct event_channel_instance *eci; struct handle_list *hl; struct list_head *l, *nxt; uint64_t handle; SaAisErrorT error; /* * Free up any channel data */ for (l = evti->ei_channel_list.next; l != &evti->ei_channel_list; l = nxt) { nxt = l->next; hl = list_entry(l, struct handle_list, hl_entry); handle = hl->hl_handle; error = saHandleInstanceGet(&channel_handle_db, hl->hl_handle, (void*)&eci); if (error != SA_AIS_OK) { /* * already gone */ continue; } saHandleDestroy(&channel_handle_db, handle); saHandleInstancePut(&channel_handle_db, handle); } } /* * Clean up function for an open channel handle */ static void chanHandleInstanceDestructor(void *instance) { struct event_channel_instance *eci = instance; struct list_head *l, *nxt; struct handle_list *hl; uint64_t handle; if (eci->eci_hl) { list_del(&eci->eci_hl->hl_entry); free(eci->eci_hl); eci->eci_hl = 0; } /* * Free up any channel associated events */ for (l = eci->eci_event_list.next; l != &eci->eci_event_list; l = nxt) { nxt = l->next; hl = list_entry(l, struct handle_list, hl_entry); handle = hl->hl_handle; saEvtEventFree(handle); } } /* * Clean up function for an event handle */ static void eventHandleInstanceDestructor(void *instance) { struct event_data_instance *edi = instance; int i; if (edi->edi_hl) { list_del(&edi->edi_hl->hl_entry); free(edi->edi_hl); edi->edi_hl = 0; } for (i = 0; i < edi->edi_patterns.patternsNumber; i++) { free(edi->edi_patterns.patterns[i].pattern); } if (edi->edi_patterns.patterns) { free(edi->edi_patterns.patterns); } if (edi->edi_event_data) { free(edi->edi_event_data); } } static SaErrorT evt_recv_event(int fd, struct lib_event_data **msg) { SaErrorT error; struct res_header hdr; void *data; error = saRecvRetry(fd, &hdr, sizeof(hdr), MSG_WAITALL | MSG_NOSIGNAL); if (error != SA_AIS_OK) { goto msg_out; } *msg = malloc(hdr.size); if (!*msg) { error = SA_AIS_ERR_LIBRARY; goto msg_out; } data = (void *)((unsigned long)*msg) + sizeof(hdr); memcpy(*msg, &hdr, sizeof(hdr)); if (hdr.size > sizeof(hdr)) { error = saRecvRetry(fd, data, hdr.size - sizeof(hdr), MSG_WAITALL | MSG_NOSIGNAL); if (error != SA_AIS_OK) { goto msg_out; } } msg_out: return error; } /* * The saEvtInitialize() function initializes the Event Service for the * invoking process. A user of the Event Service must invoke this function * before it invokes any other function of the Event Service API. Each * initialization returns a different callback handle that the process * can use to communicate with that library instance. */ SaAisErrorT saEvtInitialize( SaEvtHandleT *evtHandle, const SaEvtCallbacksT *callbacks, SaVersionT *version) { SaAisErrorT error = SA_AIS_OK; struct event_instance *evti; if (!version || !evtHandle) { error = SA_AIS_ERR_INVALID_PARAM; goto error_nofree; } /* * validate the requested version with what we support */ error = saVersionVerify(&evt_version_database, version); if (error != SA_AIS_OK) { goto error_nofree; } /* * Allocate instance data, allocate unique handle for instance, * assign instance data to unique handle */ error = saHandleCreate(&evt_instance_handle_db, sizeof(*evti), (void*)evtHandle); if (error != SA_AIS_OK) { goto error_nofree; } error = saHandleInstanceGet(&evt_instance_handle_db, *evtHandle, (void*)&evti); if (error != SA_AIS_OK) { if (error == SA_AIS_ERR_BAD_HANDLE) { error = SA_AIS_ERR_LIBRARY; } goto error_handle_free; } memset(evti, 0, sizeof(*evti)); list_init(&evti->ei_channel_list); /* * Save the version so we can check with the event server * and see if it supports this version. */ evti->ei_version = *version; /* * Set up communication with the event server */ error = saServiceConnectTwo(&evti->ei_response_fd, &evti->ei_dispatch_fd, EVT_SERVICE); if (error != SA_AIS_OK) { goto error_handle_put; } /* * The callback function is saved in the event instance for * saEvtDispatch() to use. */ if (callbacks) { memcpy(&evti->ei_callback, callbacks, sizeof(evti->ei_callback)); } pthread_mutex_init(&evti->ei_dispatch_mutex, NULL); pthread_mutex_init(&evti->ei_response_mutex, NULL); saHandleInstancePut(&evt_instance_handle_db, *evtHandle); return SA_AIS_OK; error_handle_put: saHandleInstancePut(&evt_instance_handle_db, *evtHandle); error_handle_free: (void)saHandleDestroy(&evt_instance_handle_db, *evtHandle); error_nofree: return error; } /* * The saEvtSelectionObjectGet() function returns the operating system * handle selectionObject, associated with the handle evtHandle, allowing * the invoking process to ascertain when callbacks are pending. This * function allows a process to avoid repeated invoking saEvtDispatch() to * see if there is a new event, thus, needlessly consuming CPU time. In a * POSIX environment the system handle could be a file descriptor that is * used with the poll() or select() system calls to detect incoming callbacks. */ SaAisErrorT saEvtSelectionObjectGet( SaEvtHandleT evtHandle, SaSelectionObjectT *selectionObject) { struct event_instance *evti; SaAisErrorT error; if (!selectionObject) { return SA_AIS_ERR_INVALID_PARAM; } error = saHandleInstanceGet(&evt_instance_handle_db, evtHandle, (void *)&evti); if (error != SA_AIS_OK) { return error; } *selectionObject = evti->ei_dispatch_fd; saHandleInstancePut(&evt_instance_handle_db, evtHandle); return SA_AIS_OK; } /* * Alocate an event data structure and associated handle to be * used to supply event data to a call back function. */ static SaAisErrorT make_event(SaEvtEventHandleT *event_handle, struct lib_event_data *evt) { struct event_data_instance *edi; struct event_channel_instance *eci; SaEvtEventPatternT *pat; SaUint8T *str; SaAisErrorT error; struct handle_list *hl; int i; error = saHandleCreate(&event_handle_db, sizeof(*edi), (void*)event_handle); if (error != SA_AIS_OK) { if (error == SA_AIS_ERR_NO_MEMORY) { error = SA_AIS_ERR_LIBRARY; } goto make_evt_done; } error = saHandleInstanceGet(&event_handle_db, *event_handle, (void*)&edi); if (error != SA_AIS_OK) { goto make_evt_done; } error = saHandleInstanceGet(&channel_handle_db, evt->led_lib_channel_handle, (void*)&eci); if (error != SA_AIS_OK) { goto make_evt_done_put; } memset(edi, 0, sizeof(*edi)); pthread_mutex_init(&edi->edi_mutex, NULL); edi->edi_freeing = 0; edi->edi_channel_handle = evt->led_lib_channel_handle; edi->edi_priority = evt->led_priority; edi->edi_retention_time = evt->led_retention_time; edi->edi_pub_node = evt->led_publisher_node_id; edi->edi_pub_time = evt->led_publish_time; edi->edi_event_data_size = evt->led_user_data_size; edi->edi_event_id = evt->led_event_id; edi->edi_pub_name = evt->led_publisher_name; if (edi->edi_event_data_size) { edi->edi_event_data = malloc(edi->edi_event_data_size); memcpy(edi->edi_event_data, evt->led_body + evt->led_user_data_offset, edi->edi_event_data_size); } /* * Move the pattern bits into the SaEvtEventPatternArrayT */ edi->edi_patterns.patternsNumber = evt->led_patterns_number; edi->edi_patterns.allocatedNumber = evt->led_patterns_number; edi->edi_patterns.patterns = malloc(sizeof(SaEvtEventPatternT) * edi->edi_patterns.patternsNumber); pat = (SaEvtEventPatternT *)evt->led_body; str = evt->led_body + sizeof(SaEvtEventPatternT) * edi->edi_patterns.patternsNumber; for (i = 0; i < evt->led_patterns_number; i++) { edi->edi_patterns.patterns[i].patternSize = pat->patternSize; edi->edi_patterns.patterns[i].allocatedSize = pat->patternSize; edi->edi_patterns.patterns[i].pattern = malloc(pat->patternSize); if (!edi->edi_patterns.patterns[i].pattern) { printf("make_event: couldn't alloc %lld bytes\n", pat->patternSize); error = SA_AIS_ERR_LIBRARY; break; } memcpy(edi->edi_patterns.patterns[i].pattern, str, pat->patternSize); str += pat->patternSize; pat++; } hl = malloc(sizeof(*hl)); edi->edi_hl = hl; hl->hl_handle = *event_handle; list_init(&hl->hl_entry); list_add(&hl->hl_entry, &eci->eci_event_list); saHandleInstancePut (&channel_handle_db, evt->led_lib_channel_handle); make_evt_done_put: saHandleInstancePut (&event_handle_db, *event_handle); make_evt_done: return error; } /* * The saEvtDispatch() function invokes, in the context of the calling * thread, one or all of the pending callbacks for the handle evtHandle. */ SaAisErrorT saEvtDispatch( SaEvtHandleT evtHandle, SaDispatchFlagsT dispatchFlags) { struct pollfd ufds; int timeout = -1; SaAisErrorT error; int dispatch_avail; struct event_instance *evti; SaEvtEventHandleT event_handle; SaEvtCallbacksT callbacks; int ignore_dispatch = 0; int cont = 1; /* always continue do loop except when set to 0 */ int poll_fd; struct res_overlay dispatch_data; struct lib_event_data *evt = 0; struct res_evt_event_data res; if (dispatchFlags < SA_DISPATCH_ONE || dispatchFlags > SA_DISPATCH_BLOCKING) { return SA_AIS_ERR_INVALID_PARAM; } error = saHandleInstanceGet(&evt_instance_handle_db, evtHandle, (void *)&evti); if (error != SA_AIS_OK) { return error; } /* * Timeout instantly for SA_DISPATCH_ALL */ if (dispatchFlags == SA_DISPATCH_ALL || dispatchFlags == SA_DISPATCH_ONE) { timeout = 0; } do { poll_fd = evti->ei_dispatch_fd; ufds.fd = poll_fd; ufds.events = POLLIN; ufds.revents = 0; error = saPollRetry(&ufds, 1, timeout); if (error != SA_AIS_OK) { goto dispatch_unlock; } pthread_mutex_lock(&evti->ei_dispatch_mutex); /* * Check the poll data in case the fd status has changed * since taking the lock */ error = saPollRetry(&ufds, 1, 0); if (error != SA_AIS_OK) { goto dispatch_unlock; } /* * Handle has been finalized in another thread */ if (evti->ei_finalize == 1) { error = SA_AIS_OK; goto dispatch_unlock; } if ((ufds.revents & (POLLERR|POLLHUP|POLLNVAL)) != 0) { error = SA_AIS_ERR_BAD_HANDLE; goto dispatch_unlock; } dispatch_avail = ufds.revents & POLLIN; if (dispatch_avail == 0 && (dispatchFlags == SA_DISPATCH_ALL || dispatchFlags == SA_DISPATCH_ONE)) { pthread_mutex_unlock(&evti->ei_dispatch_mutex); break; /* exit do while cont is 1 loop */ } else if (dispatch_avail == 0) { pthread_mutex_unlock(&evti->ei_dispatch_mutex); continue; /* next poll */ } if (ufds.revents & POLLIN) { error = saRecvRetry (evti->ei_dispatch_fd, &dispatch_data.header, sizeof (struct res_header), MSG_WAITALL | MSG_NOSIGNAL); if (error != SA_AIS_OK) { goto dispatch_unlock; } if (dispatch_data.header.size > sizeof (struct res_header)) { error = saRecvRetry (evti->ei_dispatch_fd, &dispatch_data.data, dispatch_data.header.size - sizeof (struct res_header), MSG_WAITALL | MSG_NOSIGNAL); if (error != SA_AIS_OK) { goto dispatch_unlock; } } } else { pthread_mutex_unlock(&evti->ei_dispatch_mutex); continue; } /* * Make copy of callbacks, message data, unlock instance, * and call callback. A risk of this dispatch method is that * the callback routines may operate at the same time that * EvtFinalize has been called in another thread. */ memcpy(&callbacks, &evti->ei_callback, sizeof(evti->ei_callback)); pthread_mutex_unlock(&evti->ei_dispatch_mutex); /* * Dispatch incoming response */ switch (dispatch_data.header.id) { case MESSAGE_RES_EVT_AVAILABLE: /* * There are events available. Send a request for one and then * dispatch it. */ res.evd_head.id = MESSAGE_REQ_EVT_EVENT_DATA; res.evd_head.size = sizeof(res); pthread_mutex_lock(&evti->ei_response_mutex); error = saSendRetry(evti->ei_response_fd, &res, sizeof(res), MSG_NOSIGNAL); if (error != SA_AIS_OK) { printf("MESSAGE_RES_EVT_AVAILABLE: send failed: %d\n", error); pthread_mutex_unlock(&evti->ei_response_mutex); break; } error = evt_recv_event(evti->ei_response_fd, &evt); pthread_mutex_unlock(&evti->ei_response_mutex); if (error != SA_AIS_OK) { printf("MESSAGE_RES_EVT_AVAILABLE: receive failed: %d\n", error); break; } /* * No data available. This is OK, another thread may have * grabbed it. */ if (evt->led_head.error == SA_AIS_ERR_NOT_EXIST) { // printf("MESSAGE_RES_EVT_AVAILABLE: No event data\n"); error = SA_AIS_OK; break; } if (evt->led_head.error != SA_AIS_OK) { error = evt->led_head.error; printf("MESSAGE_RES_EVT_AVAILABLE: Error returned: %d\n", error); break; } error = make_event(&event_handle, evt); if (error != SA_AIS_OK) { break; } /* * Only call if there was a function registered */ if (callbacks.saEvtEventDeliverCallback) { callbacks.saEvtEventDeliverCallback(evt->led_sub_id, event_handle, evt->led_user_data_size); } break; case MESSAGE_RES_EVT_CHAN_OPEN_CALLBACK: { struct res_evt_open_chan_async *resa = (struct res_evt_open_chan_async *)&dispatch_data; struct event_channel_instance *eci; /* * Check for errors. If there are none, then * look up the local channel via the handle that we * got from the callback request. All we need to do * is place in the handle from the server side and then * we can call the callback. */ error = resa->ica_head.error; if (error == SA_AIS_OK) { error = saHandleInstanceGet(&channel_handle_db, resa->ica_c_handle, (void*)&eci); if (error == SA_AIS_OK) { eci->eci_svr_channel_handle = resa->ica_channel_handle; saHandleInstancePut (&channel_handle_db, resa->ica_c_handle); } } /* * Only call if there was a function registered */ if (callbacks.saEvtChannelOpenCallback) { callbacks.saEvtChannelOpenCallback(resa->ica_invocation, resa->ica_c_handle, error); } } break; default: printf("Dispatch: Bad message type 0x%x\n", dispatch_data.header.id); error = SA_AIS_ERR_LIBRARY; goto dispatch_put; } /* * If empty is zero it means the we got the * message from the queue and we are responsible * for freeing it. */ if (evt) { free(evt); evt = 0; } /* * Determine if more messages should be processed */ switch (dispatchFlags) { case SA_DISPATCH_ONE: if (ignore_dispatch) { ignore_dispatch = 0; } else { cont = 0; } break; case SA_DISPATCH_ALL: if (ignore_dispatch) { ignore_dispatch = 0; } break; case SA_DISPATCH_BLOCKING: break; } } while (cont); goto dispatch_put; dispatch_unlock: pthread_mutex_unlock(&evti->ei_dispatch_mutex); dispatch_put: saHandleInstancePut(&evt_instance_handle_db, evtHandle); return error; } /* * The saEvtFinalize() function closes the association, represented by the * evtHandle parameter, between the process and the Event Service. It may * free up resources. * This function cannot be invoked before the process has invoked the * corresponding saEvtInitialize() function for the Event Service. After * this function is invoked, the selection object is no longer valid. * Moreover, the Event Service is unavailable for further use unless it is * reinitialized using the saEvtInitialize() function. */ SaAisErrorT saEvtFinalize(SaEvtHandleT evtHandle) { struct event_instance *evti; SaAisErrorT error; error = saHandleInstanceGet(&evt_instance_handle_db, evtHandle, (void *)&evti); if (error != SA_AIS_OK) { return error; } pthread_mutex_lock(&evti->ei_response_mutex); /* * Another thread has already started finalizing */ if (evti->ei_finalize) { pthread_mutex_unlock(&evti->ei_response_mutex); saHandleInstancePut(&evt_instance_handle_db, evtHandle); return SA_AIS_ERR_BAD_HANDLE; } evti->ei_finalize = 1; pthread_mutex_unlock(&evti->ei_response_mutex); saHandleDestroy(&evt_instance_handle_db, evtHandle); /* * Disconnect from the server */ if (evti->ei_response_fd != -1) { shutdown(evti->ei_response_fd, 0); close(evti->ei_response_fd); } if (evti->ei_dispatch_fd != -1) { shutdown(evti->ei_dispatch_fd, 0); close(evti->ei_dispatch_fd); } saHandleInstancePut(&evt_instance_handle_db, evtHandle); return error; } /* * The saEvtChannelOpen() function creates a new event channel or open an * existing channel. The saEvtChannelOpen() function is a blocking operation * and returns a new event channel handle. An event channel may be opened * multiple times by the same or different processes for publishing, and * subscribing to, events. If a process opens an event channel multiple * times, it is possible to receive the same event multiple times. However, * a process shall never receive an event more than once on a particular * event channel handle. If a process opens a channel twice and an event is * matched on both open channels, the Event Service performs two * callbacks -- one for each opened channel. */ SaAisErrorT saEvtChannelOpen( SaEvtHandleT evtHandle, const SaNameT *channelName, SaEvtChannelOpenFlagsT channelOpenFlags, SaTimeT timeout, SaEvtChannelHandleT *channelHandle) { struct event_instance *evti; struct req_evt_channel_open req; struct res_evt_channel_open res; struct event_channel_instance *eci; struct handle_list *hl; SaAisErrorT error; struct iovec iov; if (!channelHandle || !channelName) { return SA_AIS_ERR_INVALID_PARAM; } if ((channelOpenFlags & ~(SA_EVT_CHANNEL_CREATE|SA_EVT_CHANNEL_PUBLISHER| SA_EVT_CHANNEL_SUBSCRIBER)) != 0) { return SA_AIS_ERR_BAD_FLAGS; } error = saHandleInstanceGet(&evt_instance_handle_db, evtHandle, (void*)&evti); if (error != SA_AIS_OK) { goto chan_open_done; } /* * create a handle for this open channel */ error = saHandleCreate(&channel_handle_db, sizeof(*eci), (void*)channelHandle); if (error != SA_AIS_OK) { goto chan_open_put; } error = saHandleInstanceGet(&channel_handle_db, *channelHandle, (void*)&eci); if (error != SA_AIS_OK) { saHandleDestroy(&channel_handle_db, *channelHandle); goto chan_open_put; } list_init(&eci->eci_event_list); /* * Send the request to the server and wait for a response */ req.ico_head.size = sizeof(req); req.ico_head.id = MESSAGE_REQ_EVT_OPEN_CHANNEL; req.ico_c_handle = *channelHandle; req.ico_timeout = timeout; req.ico_open_flag = channelOpenFlags; req.ico_channel_name = *channelName; iov.iov_base = &req; iov.iov_len = sizeof(req); pthread_mutex_lock(&evti->ei_response_mutex); error = saSendMsgReceiveReply(evti->ei_response_fd, &iov, 1, &res, sizeof(res)); pthread_mutex_unlock (&evti->ei_response_mutex); if (error != SA_AIS_OK) { goto chan_open_free; } if (res.ico_head.id != MESSAGE_RES_EVT_OPEN_CHANNEL) { error = SA_AIS_ERR_LIBRARY; goto chan_open_free; } error = res.ico_head.error; if (error != SA_AIS_OK) { goto chan_open_free; } eci->eci_svr_channel_handle = res.ico_channel_handle; eci->eci_channel_name = *channelName; eci->eci_open_flags = channelOpenFlags; eci->eci_instance_handle = evtHandle; eci->eci_closing = 0; hl = malloc(sizeof(*hl)); eci->eci_hl = hl; hl->hl_handle = *channelHandle; list_init(&hl->hl_entry); list_add(&hl->hl_entry, &evti->ei_channel_list); pthread_mutex_init(&eci->eci_mutex, NULL); saHandleInstancePut (&evt_instance_handle_db, evtHandle); saHandleInstancePut (&channel_handle_db, *channelHandle); return SA_AIS_OK; chan_open_free: saHandleDestroy(&channel_handle_db, *channelHandle); saHandleInstancePut (&channel_handle_db, *channelHandle); chan_open_put: saHandleInstancePut (&evt_instance_handle_db, evtHandle); chan_open_done: return error; } /* * The saEvtChannelClose() function closes an event channel and frees * resources allocated for that event channel in the invoking process. */ SaAisErrorT saEvtChannelClose(SaEvtChannelHandleT channelHandle) { SaAisErrorT error; struct event_instance *evti; struct event_channel_instance *eci; struct req_evt_channel_close req; struct res_evt_channel_close res; struct iovec iov; error = saHandleInstanceGet(&channel_handle_db, channelHandle, (void*)&eci); if (error != SA_AIS_OK) { goto chan_close_done; } /* * get the evt handle for the fd */ error = saHandleInstanceGet(&evt_instance_handle_db, eci->eci_instance_handle, (void*)&evti); if (error != SA_AIS_OK) { goto chan_close_put1; } /* * Make sure that the channel isn't being closed elsewhere */ pthread_mutex_lock(&eci->eci_mutex); if (eci->eci_closing) { pthread_mutex_unlock(&eci->eci_mutex); saHandleInstancePut(&channel_handle_db, channelHandle); return SA_AIS_ERR_BAD_HANDLE; } eci->eci_closing = 1; pthread_mutex_unlock(&eci->eci_mutex); /* * Send the request to the server and wait for a response */ req.icc_head.size = sizeof(req); req.icc_head.id = MESSAGE_REQ_EVT_CLOSE_CHANNEL; req.icc_channel_handle = eci->eci_svr_channel_handle; iov.iov_base = &req; iov.iov_len = sizeof (req); pthread_mutex_lock(&evti->ei_response_mutex); error = saSendMsgReceiveReply (evti->ei_response_fd, &iov, 1, &res, sizeof (res)); pthread_mutex_unlock(&evti->ei_response_mutex); if (error != SA_AIS_OK) { eci->eci_closing = 0; goto chan_close_put2; } if (res.icc_head.id != MESSAGE_RES_EVT_CLOSE_CHANNEL) { error = SA_AIS_ERR_LIBRARY; eci->eci_closing = 0; goto chan_close_put2; } error = res.icc_head.error; saHandleInstancePut(&evt_instance_handle_db, eci->eci_instance_handle); saHandleDestroy(&channel_handle_db, channelHandle); saHandleInstancePut(&channel_handle_db, channelHandle); return error; chan_close_put2: saHandleInstancePut(&evt_instance_handle_db, eci->eci_instance_handle); chan_close_put1: saHandleInstancePut(&channel_handle_db, channelHandle); chan_close_done: return error; } /* * The saEvtChannelOpenAsync() function creates a new event channel or open an * existing channel. The saEvtChannelOpenAsync() function is a non-blocking * operation. A new event channel handle is returned in the channel open * callback function (SaEvtChannelOpenCallbackT). */ SaAisErrorT saEvtChannelOpenAsync(SaEvtHandleT evtHandle, SaInvocationT invocation, const SaNameT *channelName, SaEvtChannelOpenFlagsT channelOpenFlags) { struct event_instance *evti; struct req_evt_channel_open req; struct res_evt_channel_open res; struct event_channel_instance *eci; SaEvtChannelHandleT channel_handle; SaAisErrorT error; struct handle_list *hl; struct iovec iov; if (!channelName) { return SA_AIS_ERR_INVALID_PARAM; } if ((channelOpenFlags & ~(SA_EVT_CHANNEL_CREATE|SA_EVT_CHANNEL_PUBLISHER| SA_EVT_CHANNEL_SUBSCRIBER)) != 0) { return SA_AIS_ERR_BAD_FLAGS; } error = saHandleInstanceGet(&evt_instance_handle_db, evtHandle, (void*)&evti); if (error != SA_AIS_OK) { goto chan_open_done; } /* * Make sure that an open channel callback has been * registered before allowing the open to continue. */ if (!evti->ei_callback.saEvtChannelOpenCallback) { error = SA_AIS_ERR_INIT; goto chan_open_put; } /* * create a handle for this open channel */ error = saHandleCreate(&channel_handle_db, sizeof(*eci), &channel_handle); if (error != SA_AIS_OK) { goto chan_open_put; } error = saHandleInstanceGet(&channel_handle_db, channel_handle, (void*)&eci); if (error != SA_AIS_OK) { saHandleDestroy(&channel_handle_db, channel_handle); goto chan_open_put; } list_init(&eci->eci_event_list); /* * Send the request to the server. The response isn't the open channel, * just an ack. The open channel will be returned when the channel open * callback is called. */ req.ico_head.size = sizeof(req); req.ico_head.id = MESSAGE_REQ_EVT_OPEN_CHANNEL_ASYNC; req.ico_c_handle = channel_handle; req.ico_timeout = 0; req.ico_invocation = invocation; req.ico_open_flag = channelOpenFlags; req.ico_channel_name = *channelName; iov.iov_base = &req; iov.iov_len = sizeof(req); pthread_mutex_lock(&evti->ei_response_mutex); error = saSendMsgReceiveReply (evti->ei_response_fd, &iov, 1, &res, sizeof (res)); pthread_mutex_unlock(&evti->ei_response_mutex); if (error != SA_AIS_OK) { goto chan_open_free; } if (res.ico_head.id != MESSAGE_RES_EVT_OPEN_CHANNEL) { error = SA_AIS_ERR_LIBRARY; goto chan_open_free; } error = res.ico_head.error; if (error != SA_AIS_OK) { goto chan_open_free; } eci->eci_svr_channel_handle = 0; /* filled in by callback */ eci->eci_channel_name = *channelName; eci->eci_open_flags = channelOpenFlags; eci->eci_instance_handle = evtHandle; eci->eci_closing = 0; list_init(&eci->eci_event_list); hl = malloc(sizeof(*hl)); eci->eci_hl = hl; hl->hl_handle = channel_handle; list_init(&hl->hl_entry); list_add(&hl->hl_entry, &evti->ei_channel_list); pthread_mutex_init(&eci->eci_mutex, NULL); saHandleInstancePut (&evt_instance_handle_db, evtHandle); saHandleInstancePut (&channel_handle_db, channel_handle); return SA_AIS_OK; chan_open_free: saHandleDestroy(&channel_handle_db, channel_handle); saHandleInstancePut (&channel_handle_db, channel_handle); chan_open_put: saHandleInstancePut (&evt_instance_handle_db, evtHandle); chan_open_done: return error; } /* * The SaEvtChannelUnlink function deletes an existing event channel * from the cluster. * * After completion of the invocation: * - An open of the channel name without a create will fail. * - The channel remains available to any already opened instances. * - If an open/create is performed on this channel name a new instance * is created. * - The ulinked channel's resources will be deleted when the last open * instance is closed. * * Note that an event channel is only deleted from the cluster * namespace when saEvtChannelUnlink() is invoked on it. The deletion * of an event channel frees all resources allocated by the Event * Service for it, such as published events with non-zero retention * time. */ SaAisErrorT saEvtChannelUnlink( SaEvtHandleT evtHandle, const SaNameT *channelName) { struct event_instance *evti; struct req_evt_channel_unlink req; struct res_evt_channel_unlink res; struct iovec iov; SaAisErrorT error; if (!channelName) { return SA_AIS_ERR_INVALID_PARAM; } error = saHandleInstanceGet(&evt_instance_handle_db, evtHandle, (void*)&evti); if (error != SA_AIS_OK) { goto chan_unlink_done; } /* * Send the request to the server and wait for a response */ req.iuc_head.size = sizeof(req); req.iuc_head.id = MESSAGE_REQ_EVT_UNLINK_CHANNEL; req.iuc_channel_name = *channelName; iov.iov_base = &req; iov.iov_len = sizeof(req); pthread_mutex_lock(&evti->ei_response_mutex); error = saSendMsgReceiveReply (evti->ei_response_fd, &iov, 1, &res, sizeof (res)); pthread_mutex_unlock(&evti->ei_response_mutex); if (error != SA_AIS_OK) { goto chan_unlink_put; } if (res.iuc_head.id != MESSAGE_RES_EVT_UNLINK_CHANNEL) { error = SA_AIS_ERR_LIBRARY; goto chan_unlink_put; } error = res.iuc_head.error; chan_unlink_put: saHandleInstancePut (&evt_instance_handle_db, evtHandle); chan_unlink_done: return error; } /* * The saEvtEventAllocate() function allocates memory for the event, but not * for the eventHandle, and initializes all event attributes to default values. * The event allocated by saEvtEventAllocate() is freed by invoking * saEvtEventFree(). * The memory associated with the eventHandle is not deallocated by the * saEvtEventAllocate() function or the saEvtEventFree() function. It is the * responsibility of the invoking process to deallocate the memory for the * eventHandle when the process has published the event and has freed the * event by invoking saEvtEventFree(). */ SaAisErrorT saEvtEventAllocate( const SaEvtChannelHandleT channelHandle, SaEvtEventHandleT *eventHandle) { SaAisErrorT error; struct event_data_instance *edi; struct event_instance *evti; struct event_channel_instance *eci; struct handle_list *hl; if (!eventHandle) { return SA_AIS_ERR_INVALID_PARAM; } error = saHandleInstanceGet(&channel_handle_db, channelHandle, (void*)&eci); if (error != SA_AIS_OK) { goto alloc_done; } error = saHandleInstanceGet(&evt_instance_handle_db, eci->eci_instance_handle, (void*)&evti); if (error != SA_AIS_OK) { goto alloc_put1; } error = saHandleCreate(&event_handle_db, sizeof(*edi), (void*)eventHandle); if (error != SA_AIS_OK) { goto alloc_put2; } error = saHandleInstanceGet(&event_handle_db, *eventHandle, (void*)&edi); if (error != SA_AIS_OK) { goto alloc_put2; } memset(edi, 0, sizeof(*edi)); pthread_mutex_init(&edi->edi_mutex, NULL); edi->edi_freeing = 0; edi->edi_channel_handle = channelHandle; edi->edi_pub_node = evti->ei_node_id; edi->edi_priority = SA_EVT_LOWEST_PRIORITY; edi->edi_event_id = SA_EVT_EVENTID_NONE; edi->edi_pub_time = SA_TIME_UNKNOWN; edi->edi_retention_time = 0; hl = malloc(sizeof(*hl)); edi->edi_hl = hl; hl->hl_handle = *eventHandle; list_init(&hl->hl_entry); list_add(&hl->hl_entry, &eci->eci_event_list); saHandleInstancePut (&event_handle_db, *eventHandle); alloc_put2: saHandleInstancePut (&evt_instance_handle_db, eci->eci_instance_handle); alloc_put1: saHandleInstancePut (&channel_handle_db, edi->edi_channel_handle); alloc_done: return error; } /* * The saEvtEventFree() function gives the Event Service premission to * deallocate the memory that contains the attributes of the event that is * associated with eventHandle. The function is used to free events allocated * by saEvtEventAllocate() and by saEvtEventDeliverCallback(). */ SaAisErrorT saEvtEventFree(SaEvtEventHandleT eventHandle) { SaAisErrorT error; struct event_data_instance *edi; error = saHandleInstanceGet(&event_handle_db, eventHandle, (void*)&edi); if (error != SA_AIS_OK) { goto evt_free_done; } /* * Make sure that the event isn't being freed elsewhere */ pthread_mutex_lock(&edi->edi_mutex); if (edi->edi_freeing) { pthread_mutex_unlock(&edi->edi_mutex); saHandleInstancePut(&event_handle_db, eventHandle); return SA_AIS_ERR_BAD_HANDLE; } edi->edi_freeing = 1; pthread_mutex_unlock(&edi->edi_mutex); saHandleDestroy(&event_handle_db, eventHandle); saHandleInstancePut(&event_handle_db, eventHandle); evt_free_done: return error; } /* * This function may be used to assign writeable event attributes. It takes * as arguments an event handle eventHandle and the attribute to be set in * the event structure of the event with that handle. Note: The only * attributes that a process can set are the patternArray, priority, * retentionTime and publisherName attributes. */ SaAisErrorT saEvtEventAttributesSet( const SaEvtEventHandleT eventHandle, const SaEvtEventPatternArrayT *patternArray, SaEvtEventPriorityT priority, SaTimeT retentionTime, const SaNameT *publisherName) { SaEvtEventPatternT *oldpatterns; SaSizeT oldnumber; SaAisErrorT error; struct event_data_instance *edi; int i; error = saHandleInstanceGet(&event_handle_db, eventHandle, (void*)&edi); if (error != SA_AIS_OK) { goto attr_set_done; } pthread_mutex_lock(&edi->edi_mutex); edi->edi_priority = priority; edi->edi_retention_time = retentionTime; if (publisherName) { edi->edi_pub_name = *publisherName; } if (!patternArray) { goto attr_set_unlock; } oldpatterns = edi->edi_patterns.patterns; oldnumber = edi->edi_patterns.patternsNumber; edi->edi_patterns.patterns = 0; edi->edi_patterns.patterns = malloc(sizeof(SaEvtEventPatternT) * patternArray->patternsNumber); if (!edi->edi_patterns.patterns) { error = SA_AIS_ERR_NO_MEMORY; goto attr_set_done_reset; } edi->edi_patterns.patternsNumber = patternArray->patternsNumber; edi->edi_patterns.allocatedNumber = patternArray->patternsNumber; /* * copy the patterns from the caller. allocating memory for * of all the strings. */ for (i = 0; i < patternArray->patternsNumber; i++) { edi->edi_patterns.patterns[i].pattern = malloc(patternArray->patterns[i].patternSize); if (!edi->edi_patterns.patterns[i].pattern) { int j; for (j = 0; j < i; j++) { free(edi->edi_patterns.patterns[j].pattern); } free(edi->edi_patterns.patterns); error = SA_AIS_ERR_NO_MEMORY; goto attr_set_done_reset; } memcpy(edi->edi_patterns.patterns[i].pattern, patternArray->patterns[i].pattern, patternArray->patterns[i].patternSize); edi->edi_patterns.patterns[i].patternSize = patternArray->patterns[i].patternSize; edi->edi_patterns.patterns[i].allocatedSize = patternArray->patterns[i].patternSize; } /* * free up the old pattern memory */ if (oldpatterns) { for (i = 0; i < oldnumber; i++) { if (oldpatterns[i].pattern) { free(oldpatterns[i].pattern); } } free (oldpatterns); } goto attr_set_unlock; attr_set_done_reset: edi->edi_patterns.patterns = oldpatterns; edi->edi_patterns.patternsNumber = oldnumber; attr_set_unlock: pthread_mutex_unlock(&edi->edi_mutex); saHandleInstancePut(&event_handle_db, eventHandle); attr_set_done: return error; } /* * This function takes as parameters an event handle eventHandle and the * attributes of the event with that handle. The function retrieves the * value of the attributes for the event and stores them at the address * provided for the out parameters. * It is the responsibility of the invoking process to allocate memory for * the out parameters before it invokes this function. The Event Service * assigns the out values into that memory. For each of the out, or in/out, * parameters, if the invoking process provides a NULL reference, the * Availability Management Framework does not return the out value. * Similarly, it is the responsibility of the invoking process to allocate * memory for the patternArray. */ SaAisErrorT saEvtEventAttributesGet( SaEvtEventHandleT eventHandle, SaEvtEventPatternArrayT *patternArray, SaEvtEventPriorityT *priority, SaTimeT *retentionTime, SaNameT *publisherName, SaTimeT *publishTime, SaEvtEventIdT *eventId) { SaAisErrorT error; struct event_data_instance *edi; SaSizeT npats; int i; error = saHandleInstanceGet(&event_handle_db, eventHandle, (void*)&edi); if (error != SA_AIS_OK) { goto attr_get_done; } pthread_mutex_lock(&edi->edi_mutex); /* * Go through the args and send back information if the pointer * isn't NULL */ if (eventId) { *eventId = edi->edi_event_id; } if (publishTime) { *publishTime = edi->edi_pub_time; } if (publisherName) { *publisherName = edi->edi_pub_name; } if (retentionTime) { *retentionTime = edi->edi_retention_time; } if (priority) { *priority = edi->edi_priority; } if (!patternArray) { goto attr_get_unlock; } /* * The spec says that if the called passes in a NULL patterns array, * then we allocate the required data and the caller is responsible * for dealocating later. Otherwise, we copy to pre-allocated space. * If there are more patterns than allcated space, we set the return * code to SA_AIS_ERR_NO_SPACE and copy as much as will fit. We will * return the total number of patterns available in the patternsNumber * regardless of how much was allocated. * */ if (patternArray->patterns == NULL) { npats = edi->edi_patterns.patternsNumber; patternArray->allocatedNumber = edi->edi_patterns.patternsNumber; patternArray->patternsNumber = edi->edi_patterns.patternsNumber; patternArray->patterns = malloc(sizeof(*patternArray->patterns) * edi->edi_patterns.patternsNumber); for (i = 0; i < edi->edi_patterns.patternsNumber; i++) { patternArray->patterns[i].allocatedSize = edi->edi_patterns.patterns[i].allocatedSize; patternArray->patterns[i].patternSize = edi->edi_patterns.patterns[i].patternSize; patternArray->patterns[i].pattern = malloc(edi->edi_patterns.patterns[i].patternSize); } } else { if (patternArray->allocatedNumber < edi->edi_patterns.allocatedNumber) { error = SA_AIS_ERR_NO_SPACE; npats = patternArray->allocatedNumber; } else { npats = edi->edi_patterns.patternsNumber; } } patternArray->patternsNumber = edi->edi_patterns.patternsNumber; /* * copy the patterns to the callers structure. If we have pre-allocated * data, the patterns may not fit in the supplied space. In that case we * return NO_SPACE. */ for (i = 0; i < npats; i++) { memcpy(patternArray->patterns[i].pattern, edi->edi_patterns.patterns[i].pattern, min(patternArray->patterns[i].allocatedSize, edi->edi_patterns.patterns[i].patternSize)); if (patternArray->patterns[i].allocatedSize < edi->edi_patterns.patterns[i].patternSize) { error = SA_AIS_ERR_NO_SPACE; } patternArray->patterns[i].patternSize = edi->edi_patterns.patterns[i].patternSize; } attr_get_unlock: pthread_mutex_unlock(&edi->edi_mutex); saHandleInstancePut(&event_handle_db, eventHandle); attr_get_done: return error; } /* * The saEvtEventDataGet() function allows a process to retrieve the data * associated with an event previously delivered by * saEvtEventDeliverCallback(). */ SaAisErrorT saEvtEventDataGet( const SaEvtEventHandleT eventHandle, void *eventData, SaSizeT *eventDataSize) { SaAisErrorT error = SA_AIS_OK; struct event_data_instance *edi; SaSizeT xfsize; if (!eventData || !eventDataSize) { goto data_get_done; } error = saHandleInstanceGet(&event_handle_db, eventHandle, (void*)&edi); if (error != SA_AIS_OK) { goto data_get_done; } pthread_mutex_lock(&edi->edi_mutex); if (edi->edi_event_data && edi->edi_event_data_size) { xfsize = min(*eventDataSize, edi->edi_event_data_size); *eventDataSize = edi->edi_event_data_size; if (*eventDataSize < edi->edi_event_data_size) { error = SA_AIS_ERR_NO_SPACE; } memcpy(eventData, edi->edi_event_data, xfsize); } else { *eventDataSize = 0; } pthread_mutex_unlock(&edi->edi_mutex); saHandleInstancePut(&event_handle_db, eventHandle); data_get_done: return error; } /* * Calculate the size in bytes for patterns */ static size_t patt_size(const SaEvtEventPatternArrayT *patterns) { int i; size_t size = sizeof(SaEvtEventPatternArrayT); for (i = 0; i < patterns->patternsNumber; i++) { size += sizeof(SaEvtEventPatternT); size += patterns->patterns[i].patternSize; } return size; } /* * copy patterns to a form for sending to the server */ static uint32_t aispatt_to_evt_patt(const SaEvtEventPatternArrayT *patterns, void *data) { int i; SaEvtEventPatternT *pats = data; SaUint8T *str = (SaUint8T *)pats + (patterns->patternsNumber * sizeof(*pats)); /* * Pointers are replaced with offsets into the data array. These * will be later converted back into pointers when received as events. */ for (i = 0; i < patterns->patternsNumber; i++) { memcpy(str, patterns->patterns[i].pattern, patterns->patterns[i].patternSize); pats->patternSize = patterns->patterns[i].patternSize; pats->pattern = (SaUint8T *)((void *)str - data); str += patterns->patterns[i].patternSize; pats++; } return patterns->patternsNumber; } /* * Calculate the size in bytes for filters */ static size_t filt_size(const SaEvtEventFilterArrayT *filters) { int i; size_t size = sizeof(SaEvtEventFilterArrayT); for (i = 0; i < filters->filtersNumber; i++) { size += sizeof(SaEvtEventFilterT); size += filters->filters[i].filter.patternSize; } return size; } /* * Convert the Sa filters to a form that can be sent over the network * i.e. replace pointers with offsets. The pointers will be reconstituted * by the receiver. */ static uint32_t aisfilt_to_evt_filt(const SaEvtEventFilterArrayT *filters, void *data) { int i; SaEvtEventFilterArrayT *filta = data; SaEvtEventFilterT *filts = data + sizeof(SaEvtEventFilterArrayT); SaUint8T *str = (SaUint8T *)filts + (filters->filtersNumber * sizeof(*filts)); /* * Pointers are replaced with offsets into the data array. These * will be later converted back into pointers by the evt server. */ filta->filters = (SaEvtEventFilterT *)((void *)filts - data); filta->filtersNumber = filters->filtersNumber; for (i = 0; i < filters->filtersNumber; i++) { filts->filterType = filters->filters[i].filterType; filts->filter.patternSize = filters->filters[i].filter.patternSize; memcpy(str, filters->filters[i].filter.pattern, filters->filters[i].filter.patternSize); filts->filter.pattern = (SaUint8T *)((void *)str - data); str += filters->filters[i].filter.patternSize; filts++; } return filters->filtersNumber; } /* * The saEvtEventPublish() function publishes an event on the associated * channel. The event to be published consists of a * standard set of attributes (the event header) and an optional data part. * Before an event can be published, the publisher process must invoke the * saEvtEventPatternArraySet() function to set the event patterns. The event * is delivered to subscribers whose subscription filter matches the event * patterns. * When the Event Service publishes an event, it automatically sets * the following readonly event attributes: * - Event attribute time * - Event publisher identifier * - Event publisher node identifier * - Event identifier * In addition to the event attributes, a process can supply values for the * eventData and eventDataSize parameters for publication as part of the * event. The data portion of the event may be at most SA_EVT_DATA_MAX_LEN * bytes in length. * The process may assume that the invocation of saEvtEventPublish() copies * all pertinent parameters, including the memory associated with the * eventHandle and eventData parameters, to its own local memory. However, * the invocation does not automatically deallocate memory associated with * the eventHandle and eventData parameters. It is the responsibility of the * invoking process to deallocate the memory for those parameters after * saEvtEventPublish() returns. */ SaAisErrorT saEvtEventPublish( const SaEvtEventHandleT eventHandle, const void *eventData, SaSizeT eventDataSize, SaEvtEventIdT *eventId) { SaAisErrorT error; struct event_data_instance *edi; struct event_instance *evti; struct event_channel_instance *eci; struct lib_event_data *req; struct res_evt_event_publish res; size_t pattern_size; struct event_pattern *patterns; void *data_start; struct iovec iov; if (!eventId) { return SA_AIS_ERR_INVALID_PARAM; } if (eventDataSize > SA_EVT_DATA_MAX_LEN) { error = SA_AIS_ERR_INVALID_PARAM; goto pub_done; } error = saHandleInstanceGet(&event_handle_db, eventHandle, (void*)&edi); if (error != SA_AIS_OK) { goto pub_done; } pthread_mutex_lock(&edi->edi_mutex); error = saHandleInstanceGet(&channel_handle_db, edi->edi_channel_handle, (void*)&eci); if (error != SA_AIS_OK) { goto pub_put1; } /* * See if we can publish to this channel */ if (!(eci->eci_open_flags & SA_EVT_CHANNEL_PUBLISHER)) { error = SA_AIS_ERR_ACCESS; goto pub_put2; } error = saHandleInstanceGet(&evt_instance_handle_db, eci->eci_instance_handle, (void*)&evti); if (error != SA_AIS_OK) { goto pub_put2; } /* * Figure out how much memory we need for the patterns and data */ pattern_size = patt_size(&edi->edi_patterns); req = malloc(sizeof(*req) + eventDataSize + pattern_size); patterns = (struct event_pattern *)req->led_body; data_start = (void *)req->led_body + pattern_size; if (!req) { error = SA_AIS_ERR_NO_MEMORY; goto pub_put3; } /* * copy everything to the request structure */ aispatt_to_evt_patt(&edi->edi_patterns, patterns); req->led_patterns_number = edi->edi_patterns.patternsNumber; req->led_user_data_offset = pattern_size; if (eventData && eventDataSize) { memcpy(data_start, eventData, eventDataSize); req->led_user_data_size = eventDataSize; } else { req->led_user_data_size = 0; } req->led_head.id = MESSAGE_REQ_EVT_PUBLISH; req->led_head.size = sizeof(*req) + pattern_size + eventDataSize; req->led_svr_channel_handle = eci->eci_svr_channel_handle; req->led_retention_time = edi->edi_retention_time; req->led_publish_time = clustTimeNow(); req->led_priority = edi->edi_priority; req->led_publisher_name = edi->edi_pub_name; iov.iov_base = req; iov.iov_len = req->led_head.size; pthread_mutex_lock(&evti->ei_response_mutex); error = saSendMsgReceiveReply(evti->ei_response_fd, &iov, 1, &res, sizeof(res)); pthread_mutex_unlock (&evti->ei_response_mutex); free(req); if (error != SA_AIS_OK) { pthread_mutex_unlock (&evti->ei_response_mutex); goto pub_put3; } error = res.iep_head.error; if (error == SA_AIS_OK) { *eventId = res.iep_event_id; } pub_put3: saHandleInstancePut (&evt_instance_handle_db, eci->eci_instance_handle); pub_put2: saHandleInstancePut (&channel_handle_db, edi->edi_channel_handle); pub_put1: pthread_mutex_unlock(&edi->edi_mutex); saHandleInstancePut(&event_handle_db, eventHandle); pub_done: return error; } /* * The saEvtEventSubscribe() function enables a process to subscribe for * events on an event channel by registering one or more filters on that * event channel. The process must have opened the event channel, designated * by channelHandle, with the SA_EVT_CHANNEL_SUBSCRIBER flag set for an * invocation of this function to be successful. * The memory associated with the filters is not deallocated by the * saEvtEventSubscribe() function. It is the responsibility of the invoking * process to deallocate the memory when the saEvtEventSubscribe() function * returns. * For a given subscription, the filters parameter cannot be modified. To * change the filters parameter without losing events, a process must * establish a new subscription with the new filters parameter. After the new * subscription is established, the old subscription can be removed by * invoking the saEvtEventUnsubscribe() function. */ SaAisErrorT saEvtEventSubscribe( const SaEvtChannelHandleT channelHandle, const SaEvtEventFilterArrayT *filters, SaEvtSubscriptionIdT subscriptionId) { SaAisErrorT error; struct event_instance *evti; struct event_channel_instance *eci; struct req_evt_event_subscribe *req; struct res_evt_event_subscribe res; int sz; struct iovec iov; if (!filters) { return SA_AIS_ERR_INVALID_PARAM; } error = saHandleInstanceGet(&channel_handle_db, channelHandle, (void*)&eci); if (error != SA_AIS_OK) { goto subscribe_done; } /* * get the evt handle for the fd */ error = saHandleInstanceGet(&evt_instance_handle_db, eci->eci_instance_handle, (void*)&evti); if (error != SA_AIS_OK) { goto subscribe_put1; } /* * Make sure that a deliver callback has been * registered before allowing the subscribe to continue. */ if (!evti->ei_callback.saEvtEventDeliverCallback) { error = SA_AIS_ERR_INIT; goto subscribe_put2; } /* * See if we can subscribe to this channel */ if (!(eci->eci_open_flags & SA_EVT_CHANNEL_SUBSCRIBER)) { error = SA_AIS_ERR_ACCESS; goto subscribe_put2; } /* * calculate size needed to store the filters */ sz = filt_size(filters); req = malloc(sizeof(*req) + sz); if (!req) { error = SA_AIS_ERR_NO_MEMORY; goto subscribe_put2; } /* * Copy the supplied filters to the request */ req->ics_filter_count = aisfilt_to_evt_filt(filters, req->ics_filter_data); req->ics_head.id = MESSAGE_REQ_EVT_SUBSCRIBE; req->ics_head.size = sizeof(*req) + sz; req->ics_channel_handle = eci->eci_svr_channel_handle; req->ics_sub_id = subscriptionId; req->ics_filter_size = sz; iov.iov_base = req; iov.iov_len = req->ics_head.size; pthread_mutex_lock(&evti->ei_response_mutex); error = saSendMsgReceiveReply(evti->ei_response_fd, &iov, 1, &res, sizeof(res)); pthread_mutex_unlock (&evti->ei_response_mutex); free(req); if (res.ics_head.id != MESSAGE_RES_EVT_SUBSCRIBE) { goto subscribe_put2; } error = res.ics_head.error; subscribe_put2: saHandleInstancePut(&evt_instance_handle_db, eci->eci_instance_handle); subscribe_put1: saHandleInstancePut(&channel_handle_db, channelHandle); subscribe_done: return error; } /* * The saEvtEventUnsubscribe() function allows a process to stop receiving * events for a particular subscription on an event channel by removing the * subscription. The saEvtEventUnsubscribe() operation is successful if the * subscriptionId parameter matches a previously registered subscription. * Pending events that no longer match any subscription, because the * saEvtEventUnsubscribe() operation had been invoked, are purged. a process * that wishes to modify a subscription without losing any events must * establish the new subscription before removing the existing subscription. */ SaAisErrorT saEvtEventUnsubscribe( const SaEvtChannelHandleT channelHandle, SaEvtSubscriptionIdT subscriptionId) { SaAisErrorT error; struct event_instance *evti; struct event_channel_instance *eci; struct req_evt_event_unsubscribe req; struct res_evt_event_unsubscribe res; struct iovec iov; error = saHandleInstanceGet(&channel_handle_db, channelHandle, (void*)&eci); if (error != SA_AIS_OK) { goto unsubscribe_done; } error = saHandleInstanceGet(&evt_instance_handle_db, eci->eci_instance_handle, (void*)&evti); if (error != SA_AIS_OK) { goto unsubscribe_put1; } req.icu_head.id = MESSAGE_REQ_EVT_UNSUBSCRIBE; req.icu_head.size = sizeof(req); req.icu_channel_handle = eci->eci_svr_channel_handle; req.icu_sub_id = subscriptionId; iov.iov_base = &req; iov.iov_len = sizeof(req); pthread_mutex_lock(&evti->ei_response_mutex); error = saSendMsgReceiveReply(evti->ei_response_fd, &iov, 1, &res, sizeof(res)); pthread_mutex_unlock (&evti->ei_response_mutex); if (error != SA_AIS_OK) { goto unsubscribe_put2; } if (res.icu_head.id != MESSAGE_RES_EVT_UNSUBSCRIBE) { error = SA_AIS_ERR_LIBRARY; goto unsubscribe_put2; } error = res.icu_head.error; unsubscribe_put2: saHandleInstancePut(&evt_instance_handle_db, eci->eci_instance_handle); unsubscribe_put1: saHandleInstancePut(&channel_handle_db, channelHandle); unsubscribe_done: return error; } /* * The saEvtEventRetentionTimeClear() function is used to clear the retention * time of a published event. It indicates to the Event Service that it does * not need to keep the event any longer for potential new subscribers. Once * the value of the retention time is reset to 0, the event is no longer * available for new subscribers. The event is held until all old subscribers * in the system process the event and free the event using saEvtEventFree(). */ SaAisErrorT saEvtEventRetentionTimeClear( const SaEvtChannelHandleT channelHandle, const SaEvtEventIdT eventId) { SaAisErrorT error; struct event_instance *evti; struct event_channel_instance *eci; struct req_evt_event_clear_retentiontime req; struct res_evt_event_clear_retentiontime res; struct iovec iov; error = saHandleInstanceGet(&channel_handle_db, channelHandle, (void*)&eci); if (error != SA_AIS_OK) { goto ret_time_done; } error = saHandleInstanceGet(&evt_instance_handle_db, eci->eci_instance_handle, (void*)&evti); if (error != SA_AIS_OK) { goto ret_time_put1; } req.iec_head.id = MESSAGE_REQ_EVT_CLEAR_RETENTIONTIME; req.iec_head.size = sizeof(req); req.iec_channel_handle = eci->eci_svr_channel_handle; req.iec_event_id = eventId; iov.iov_base = &req; iov.iov_len = sizeof(req); pthread_mutex_lock(&evti->ei_response_mutex); error = saSendMsgReceiveReply(evti->ei_response_fd, &iov, 1, &res, sizeof(res)); pthread_mutex_unlock (&evti->ei_response_mutex); if (error != SA_AIS_OK) { goto ret_time_put2; } if (res.iec_head.id != MESSAGE_RES_EVT_CLEAR_RETENTIONTIME) { error = SA_AIS_ERR_LIBRARY; goto ret_time_put2; } error = res.iec_head.error; ret_time_put2: saHandleInstancePut(&evt_instance_handle_db, eci->eci_instance_handle); ret_time_put1: saHandleInstancePut(&channel_handle_db, channelHandle); ret_time_done: return error; } /* * vi: set autoindent tabstop=4 shiftwidth=4 : */