diff --git a/exec/coropoll.c b/exec/coropoll.c index c791f8fa..a0aa8c3a 100644 --- a/exec/coropoll.c +++ b/exec/coropoll.c @@ -1,449 +1,442 @@ /* * Copyright (c) 2003-2004 MontaVista Software, Inc. * Copyright (c) 2006-2008 Red Hat, Inc. * * All rights reserved. * * Author: Steven Dake (sdake@redhat.com) * * 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 MontaVista Software, Inc. 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 #include "tlist.h" typedef int (*dispatch_fn_t) (hdb_handle_t hdb_handle_t, int fd, int revents, void *data); struct poll_entry { struct pollfd ufd; dispatch_fn_t dispatch_fn; void *data; }; struct poll_instance { struct poll_entry *poll_entries; struct pollfd *ufds; int poll_entry_count; struct timerlist timerlist; int stop_requested; }; -/* - * All instances in one database - */ -static struct hdb_handle_database poll_instance_database = { - .handle_count = 0, - .handles = 0, - .iterator = 0 -}; +DECLARE_HDB_DATABASE (poll_instance_database); hdb_handle_t poll_create (void) { hdb_handle_t handle; struct poll_instance *poll_instance; unsigned int res; res = hdb_handle_create (&poll_instance_database, sizeof (struct poll_instance), &handle); if (res != 0) { goto error_exit; } res = hdb_handle_get (&poll_instance_database, handle, (void *)&poll_instance); if (res != 0) { goto error_destroy; } poll_instance->poll_entries = 0; poll_instance->ufds = 0; poll_instance->poll_entry_count = 0; poll_instance->stop_requested = 0; timerlist_init (&poll_instance->timerlist); return (handle); error_destroy: hdb_handle_destroy (&poll_instance_database, handle); error_exit: return (-1); } int poll_destroy (hdb_handle_t handle) { struct poll_instance *poll_instance; int res = 0; res = hdb_handle_get (&poll_instance_database, handle, (void *)&poll_instance); if (res != 0) { res = -ENOENT; goto error_exit; } if (poll_instance->poll_entries) { free (poll_instance->poll_entries); } if (poll_instance->ufds) { free (poll_instance->ufds); } hdb_handle_destroy (&poll_instance_database, handle); hdb_handle_put (&poll_instance_database, handle); error_exit: return (res); } int poll_dispatch_add ( hdb_handle_t handle, int fd, int events, void *data, int (*dispatch_fn) ( hdb_handle_t hdb_handle_t, int fd, int revents, void *data)) { struct poll_instance *poll_instance; struct poll_entry *poll_entries; struct pollfd *ufds; int found = 0; int install_pos; int res = 0; res = hdb_handle_get (&poll_instance_database, handle, (void *)&poll_instance); if (res != 0) { res = -ENOENT; goto error_exit; } for (found = 0, install_pos = 0; install_pos < poll_instance->poll_entry_count; install_pos++) { if (poll_instance->poll_entries[install_pos].ufd.fd == -1) { found = 1; break; } } if (found == 0) { /* * Grow pollfd list */ poll_entries = (struct poll_entry *)realloc (poll_instance->poll_entries, (poll_instance->poll_entry_count + 1) * sizeof (struct poll_entry)); if (poll_entries == NULL) { res = -ENOMEM; goto error_put; } poll_instance->poll_entries = poll_entries; ufds = (struct pollfd *)realloc (poll_instance->ufds, (poll_instance->poll_entry_count + 1) * sizeof (struct pollfd)); if (ufds == NULL) { res = -ENOMEM; goto error_put; } poll_instance->ufds = ufds; poll_instance->poll_entry_count += 1; install_pos = poll_instance->poll_entry_count - 1; } /* * Install new dispatch handler */ poll_instance->poll_entries[install_pos].ufd.fd = fd; poll_instance->poll_entries[install_pos].ufd.events = events; poll_instance->poll_entries[install_pos].ufd.revents = 0; poll_instance->poll_entries[install_pos].dispatch_fn = dispatch_fn; poll_instance->poll_entries[install_pos].data = data; error_put: hdb_handle_put (&poll_instance_database, handle); error_exit: return (res); } int poll_dispatch_modify ( hdb_handle_t handle, int fd, int events, int (*dispatch_fn) ( hdb_handle_t hdb_handle_t, int fd, int revents, void *data)) { struct poll_instance *poll_instance; int i; int res = 0; res = hdb_handle_get (&poll_instance_database, handle, (void *)&poll_instance); if (res != 0) { res = -ENOENT; goto error_exit; } /* * Find file descriptor to modify events and dispatch function */ for (i = 0; i < poll_instance->poll_entry_count; i++) { if (poll_instance->poll_entries[i].ufd.fd == fd) { poll_instance->poll_entries[i].ufd.events = events; poll_instance->poll_entries[i].dispatch_fn = dispatch_fn; goto error_put; } } res = -EBADF; error_put: hdb_handle_put (&poll_instance_database, handle); error_exit: return (res); } int poll_dispatch_delete ( hdb_handle_t handle, int fd) { struct poll_instance *poll_instance; int i; int res = 0; res = hdb_handle_get (&poll_instance_database, handle, (void *)&poll_instance); if (res != 0) { res = -ENOENT; goto error_exit; } /* * Find dispatch fd to delete */ res = -EBADF; for (i = 0; i < poll_instance->poll_entry_count; i++) { if (poll_instance->poll_entries[i].ufd.fd == fd) { poll_instance->poll_entries[i].ufd.fd = -1; poll_instance->poll_entries[i].ufd.revents = 0; break; } } hdb_handle_put (&poll_instance_database, handle); error_exit: return (res); } int poll_timer_add ( hdb_handle_t handle, int msec_duration, void *data, void (*timer_fn) (void *data), poll_timer_handle *timer_handle_out) { struct poll_instance *poll_instance; int res = 0; if (timer_handle_out == NULL) { res = -ENOENT; goto error_exit; } res = hdb_handle_get (&poll_instance_database, handle, (void *)&poll_instance); if (res != 0) { res = -ENOENT; goto error_exit; } timerlist_add_duration (&poll_instance->timerlist, timer_fn, data, ((unsigned long long)msec_duration) * 1000000ULL, timer_handle_out); hdb_handle_put (&poll_instance_database, handle); error_exit: return (res); } int poll_timer_delete ( hdb_handle_t handle, poll_timer_handle timer_handle) { struct poll_instance *poll_instance; int res = 0; if (timer_handle == 0) { return (0); } res = hdb_handle_get (&poll_instance_database, handle, (void *)&poll_instance); if (res != 0) { res = -ENOENT; goto error_exit; } timerlist_del (&poll_instance->timerlist, (void *)timer_handle); hdb_handle_put (&poll_instance_database, handle); error_exit: return (res); } int poll_stop ( hdb_handle_t handle) { struct poll_instance *poll_instance; unsigned int res; res = hdb_handle_get (&poll_instance_database, handle, (void *)&poll_instance); if (res != 0) { res = -ENOENT; goto error_exit; } poll_instance->stop_requested = 1; hdb_handle_put (&poll_instance_database, handle); error_exit: return (res); } int poll_run ( hdb_handle_t handle) { struct poll_instance *poll_instance; int i; unsigned long long expire_timeout_msec = -1; int res; int poll_entry_count; res = hdb_handle_get (&poll_instance_database, handle, (void *)&poll_instance); if (res != 0) { goto error_exit; } for (;;) { for (i = 0; i < poll_instance->poll_entry_count; i++) { memcpy (&poll_instance->ufds[i], &poll_instance->poll_entries[i].ufd, sizeof (struct pollfd)); } expire_timeout_msec = timerlist_msec_duration_to_expire (&poll_instance->timerlist); if (expire_timeout_msec != -1 && expire_timeout_msec > 0xFFFFFFFF) { expire_timeout_msec = 0xFFFFFFFE; } retry_poll: res = poll (poll_instance->ufds, poll_instance->poll_entry_count, expire_timeout_msec); if (poll_instance->stop_requested) { return (0); } if (errno == EINTR && res == -1) { goto retry_poll; } else if (res == -1) { goto error_exit; } poll_entry_count = poll_instance->poll_entry_count; for (i = 0; i < poll_entry_count; i++) { if (poll_instance->ufds[i].fd != -1 && poll_instance->ufds[i].revents) { res = poll_instance->poll_entries[i].dispatch_fn (handle, poll_instance->ufds[i].fd, poll_instance->ufds[i].revents, poll_instance->poll_entries[i].data); /* * Remove dispatch functions that return -1 */ if (res == -1) { poll_instance->poll_entries[i].ufd.fd = -1; /* empty entry */ } } } timerlist_expire (&poll_instance->timerlist); } /* for (;;) */ hdb_handle_put (&poll_instance_database, handle); error_exit: return (-1); } #ifdef COMPILE_OUT void poll_print_state ( hdb_handle_t handle, int fd) { struct poll_instance *poll_instance; int i; int res = 0; res = hdb_handle_get (&poll_instance_database, handle, (void *)&poll_instance); if (res != 0) { res = -ENOENT; exit (1); } for (i = 0; i < poll_instance->poll_entry_count; i++) { if (poll_instance->poll_entries[i].ufd.fd == fd) { printf ("fd %d\n", poll_instance->poll_entries[i].ufd.fd); printf ("events %d\n", poll_instance->poll_entries[i].ufd.events); printf ("dispatch_fn %p\n", poll_instance->poll_entries[i].dispatch_fn); } } } #endif diff --git a/exec/objdb.c b/exec/objdb.c index d04f9595..1f34f7db 100644 --- a/exec/objdb.c +++ b/exec/objdb.c @@ -1,1619 +1,1608 @@ /* * Copyright (c) 2006 MontaVista Software, Inc. * Copyright (c) 2007-2009 Red Hat, Inc. * * All rights reserved. * * Author: Steven Dake (sdake@redhat.com) * * 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 MontaVista Software, Inc. 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 "main.h" struct object_key { void *key_name; size_t key_len; void *value; size_t value_len; struct list_head list; }; struct object_tracker { hdb_handle_t object_handle; void * data_pt; object_track_depth_t depth; object_key_change_notify_fn_t key_change_notify_fn; object_create_notify_fn_t object_create_notify_fn; object_destroy_notify_fn_t object_destroy_notify_fn; object_reload_notify_fn_t object_reload_notify_fn; struct list_head tracker_list; struct list_head object_list; }; struct object_instance { void *object_name; size_t object_name_len; hdb_handle_t object_handle; hdb_handle_t parent_handle; struct list_head key_head; struct list_head child_head; struct list_head child_list; struct list_head *find_child_list; struct list_head *iter_key_list; struct list_head *iter_list; void *priv; struct object_valid *object_valid_list; int object_valid_list_entries; struct object_key_valid *object_key_valid_list; int object_key_valid_list_entries; struct list_head track_head; }; struct object_find_instance { struct list_head *find_child_list; struct list_head *child_head; void *object_name; size_t object_len; }; struct objdb_iface_ver0 objdb_iface; struct list_head objdb_trackers_head; static pthread_rwlock_t reload_lock; static pthread_t lock_thread; static pthread_mutex_t meta_lock; -static struct hdb_handle_database object_instance_database = { - .handle_count = 0, - .handles = 0, - .iterator = 0, - .mutex = PTHREAD_MUTEX_INITIALIZER -}; - -static struct hdb_handle_database object_find_instance_database = { - .handle_count = 0, - .handles = 0, - .iterator = 0, - .mutex = PTHREAD_MUTEX_INITIALIZER -}; +DECLARE_HDB_DATABASE (object_instance_database); +DECLARE_HDB_DATABASE (object_find_instance_database); static void objdb_wrlock(void) { pthread_mutex_lock(&meta_lock); pthread_rwlock_wrlock(&reload_lock); lock_thread = pthread_self(); pthread_mutex_unlock(&meta_lock); } static void objdb_rdlock(void) { pthread_mutex_lock(&meta_lock); if (lock_thread != pthread_self()) pthread_rwlock_rdlock(&reload_lock); pthread_mutex_unlock(&meta_lock); } static void objdb_rdunlock(void) { pthread_mutex_lock(&meta_lock); if (lock_thread != pthread_self()) pthread_rwlock_unlock(&reload_lock); pthread_mutex_unlock(&meta_lock); } static void objdb_wrunlock(void) { pthread_mutex_lock(&meta_lock); pthread_rwlock_unlock(&reload_lock); lock_thread = 0; pthread_mutex_unlock(&meta_lock); } static int objdb_init (void) { hdb_handle_t handle; struct object_instance *instance; unsigned int res; res = hdb_handle_create (&object_instance_database, sizeof (struct object_instance), &handle); if (res != 0) { goto error_exit; } res = hdb_handle_get (&object_instance_database, handle, (void *)&instance); if (res != 0) { goto error_destroy; } instance->find_child_list = &instance->child_head; instance->object_name = "parent"; instance->object_name_len = strlen ("parent"); instance->object_handle = handle; instance->parent_handle = OBJECT_PARENT_HANDLE; instance->priv = NULL; instance->object_valid_list = NULL; instance->object_valid_list_entries = 0; list_init (&instance->key_head); list_init (&instance->child_head); list_init (&instance->child_list); list_init (&instance->track_head); list_init (&objdb_trackers_head); pthread_rwlock_init(&reload_lock, NULL); pthread_mutex_init(&meta_lock, NULL); hdb_handle_put (&object_instance_database, handle); return (0); error_destroy: hdb_handle_destroy (&object_instance_database, handle); error_exit: return (-1); } static int _object_notify_deleted_children(struct object_instance *parent_pt) { struct list_head *list; struct list_head *notify_list; int res; struct object_instance *obj_pt = NULL; struct object_tracker * tracker_pt; for (list = parent_pt->child_head.next; list != &parent_pt->child_head; list = list->next) { obj_pt = list_entry(list, struct object_instance, child_list); res = _object_notify_deleted_children(obj_pt); if (res) return res; for (notify_list = obj_pt->track_head.next; notify_list != &obj_pt->track_head; notify_list = notify_list->next) { tracker_pt = list_entry (notify_list, struct object_tracker, object_list); if ((tracker_pt != NULL) && (tracker_pt->object_destroy_notify_fn != NULL)) tracker_pt->object_destroy_notify_fn(parent_pt->object_handle, obj_pt->object_name, obj_pt->object_name_len, tracker_pt->data_pt); } } return 0; } static void object_created_notification( hdb_handle_t object_handle, hdb_handle_t parent_object_handle, const void *name_pt, size_t name_len) { struct list_head * list; struct object_instance * obj_pt; struct object_tracker * tracker_pt; hdb_handle_t obj_handle = object_handle; unsigned int res; do { res = hdb_handle_get (&object_instance_database, obj_handle, (void *)&obj_pt); for (list = obj_pt->track_head.next; list != &obj_pt->track_head; list = list->next) { tracker_pt = list_entry (list, struct object_tracker, object_list); if (((obj_handle == parent_object_handle) || (tracker_pt->depth == OBJECT_TRACK_DEPTH_RECURSIVE)) && (tracker_pt->object_create_notify_fn != NULL)) { tracker_pt->object_create_notify_fn(object_handle, parent_object_handle, name_pt, name_len, tracker_pt->data_pt); } } hdb_handle_put (&object_instance_database, obj_handle); obj_handle = obj_pt->parent_handle; } while (obj_handle != OBJECT_PARENT_HANDLE); } static void object_pre_deletion_notification(hdb_handle_t object_handle, hdb_handle_t parent_object_handle, const void *name_pt, size_t name_len) { struct list_head * list; struct object_instance * obj_pt; struct object_tracker * tracker_pt; hdb_handle_t obj_handle = object_handle; unsigned int res; do { res = hdb_handle_get (&object_instance_database, obj_handle, (void *)&obj_pt); for (list = obj_pt->track_head.next; list != &obj_pt->track_head; list = list->next) { tracker_pt = list_entry (list, struct object_tracker, object_list); if (((obj_handle == parent_object_handle) || (tracker_pt->depth == OBJECT_TRACK_DEPTH_RECURSIVE)) && (tracker_pt->object_destroy_notify_fn != NULL)) { tracker_pt->object_destroy_notify_fn( parent_object_handle, name_pt, name_len, tracker_pt->data_pt); } } /* notify child object listeners */ if (obj_handle == object_handle) _object_notify_deleted_children(obj_pt); obj_handle = obj_pt->parent_handle; hdb_handle_put (&object_instance_database, obj_pt->object_handle); } while (obj_handle != OBJECT_PARENT_HANDLE); } static void object_key_changed_notification(hdb_handle_t object_handle, const void *name_pt, size_t name_len, const void *value_pt, size_t value_len, object_change_type_t type) { struct list_head * list; struct object_instance * obj_pt; struct object_instance * owner_pt = NULL; struct object_tracker * tracker_pt; hdb_handle_t obj_handle = object_handle; unsigned int res; do { res = hdb_handle_get (&object_instance_database, obj_handle, (void *)&obj_pt); if (owner_pt == NULL) owner_pt = obj_pt; for (list = obj_pt->track_head.next; list != &obj_pt->track_head; list = list->next) { tracker_pt = list_entry (list, struct object_tracker, object_list); if (((obj_handle == object_handle) || (tracker_pt->depth == OBJECT_TRACK_DEPTH_RECURSIVE)) && (tracker_pt->key_change_notify_fn != NULL)) tracker_pt->key_change_notify_fn(type, obj_pt->parent_handle, object_handle, owner_pt->object_name, owner_pt->object_name_len, name_pt, name_len, value_pt, value_len, tracker_pt->data_pt); } obj_handle = obj_pt->parent_handle; hdb_handle_put (&object_instance_database, obj_pt->object_handle); } while (obj_handle != OBJECT_PARENT_HANDLE); } static void object_reload_notification(int startstop, int flush) { struct list_head * list; struct object_instance * obj_pt; struct object_tracker * tracker_pt; unsigned int res; res = hdb_handle_get (&object_instance_database, OBJECT_PARENT_HANDLE, (void *)&obj_pt); for (list = obj_pt->track_head.next; list != &obj_pt->track_head; list = list->next) { tracker_pt = list_entry (list, struct object_tracker, object_list); if (tracker_pt->object_reload_notify_fn != NULL) { tracker_pt->object_reload_notify_fn(startstop, flush, tracker_pt->data_pt); } } hdb_handle_put (&object_instance_database, OBJECT_PARENT_HANDLE); } /* * object db create/destroy/set */ static int object_create ( hdb_handle_t parent_object_handle, hdb_handle_t *object_handle, const void *object_name, size_t object_name_len) { struct object_instance *object_instance; struct object_instance *parent_instance; unsigned int res; int found = 0; int i; objdb_rdlock(); res = hdb_handle_get (&object_instance_database, parent_object_handle, (void *)&parent_instance); if (res != 0) { goto error_exit; } /* * Do validation check if validation is configured for the parent object */ if (parent_instance->object_valid_list_entries) { for (i = 0; i < parent_instance->object_valid_list_entries; i++) { if ((object_name_len == parent_instance->object_valid_list[i].object_len) && (memcmp (object_name, parent_instance->object_valid_list[i].object_name, object_name_len) == 0)) { found = 1; break; } } /* * Item not found in validation list */ if (found == 0) { goto error_object_put; } } res = hdb_handle_create (&object_instance_database, sizeof (struct object_instance), object_handle); if (res != 0) { goto error_object_put; } res = hdb_handle_get (&object_instance_database, *object_handle, (void *)&object_instance); if (res != 0) { goto error_destroy; } list_init (&object_instance->key_head); list_init (&object_instance->child_head); list_init (&object_instance->child_list); list_init (&object_instance->track_head); object_instance->object_name = malloc (object_name_len); if (object_instance->object_name == 0) { goto error_put_destroy; } memcpy (object_instance->object_name, object_name, object_name_len); object_instance->object_name_len = object_name_len; list_add_tail (&object_instance->child_list, &parent_instance->child_head); object_instance->object_handle = *object_handle; object_instance->find_child_list = &object_instance->child_head; object_instance->iter_key_list = &object_instance->key_head; object_instance->iter_list = &object_instance->child_head; object_instance->priv = NULL; object_instance->object_valid_list = NULL; object_instance->object_valid_list_entries = 0; object_instance->parent_handle = parent_object_handle; hdb_handle_put (&object_instance_database, *object_handle); hdb_handle_put (&object_instance_database, parent_object_handle); object_created_notification(object_instance->object_handle, object_instance->parent_handle, object_instance->object_name, object_instance->object_name_len); objdb_rdunlock(); return (0); error_put_destroy: hdb_handle_put (&object_instance_database, *object_handle); error_destroy: hdb_handle_destroy (&object_instance_database, *object_handle); error_object_put: hdb_handle_put (&object_instance_database, parent_object_handle); error_exit: objdb_rdunlock(); return (-1); } static int object_priv_set ( hdb_handle_t object_handle, void *priv) { int res; struct object_instance *object_instance; objdb_rdlock(); res = hdb_handle_get (&object_instance_database, object_handle, (void *)&object_instance); if (res != 0) { goto error_exit; } object_instance->priv = priv; hdb_handle_put (&object_instance_database, object_handle); objdb_rdunlock(); return (0); error_exit: objdb_rdunlock(); return (-1); } static int object_key_create ( hdb_handle_t object_handle, const void *key_name, size_t key_len, const void *value, size_t value_len) { struct object_instance *instance; struct object_key *object_key; unsigned int res; struct list_head *list; int found = 0; int i; objdb_rdlock(); res = hdb_handle_get (&object_instance_database, object_handle, (void *)&instance); if (res != 0) { goto error_exit; } /* * Do validation check if validation is configured for the parent object */ if (instance->object_key_valid_list_entries) { for (i = 0; i < instance->object_key_valid_list_entries; i++) { if ((key_len == instance->object_key_valid_list[i].key_len) && (memcmp (key_name, instance->object_key_valid_list[i].key_name, key_len) == 0)) { found = 1; break; } } /* * Item not found in validation list */ if (found == 0) { goto error_put; } else { if (instance->object_key_valid_list[i].validate_callback) { res = instance->object_key_valid_list[i].validate_callback ( key_name, key_len, value, value_len); if (res != 0) { goto error_put; } } } } /* See if it already exists */ found = 0; for (list = instance->key_head.next; list != &instance->key_head; list = list->next) { object_key = list_entry (list, struct object_key, list); if ((object_key->key_len == key_len) && (memcmp (object_key->key_name, key_name, key_len) == 0)) { found = 1; break; } } if (found) { free(object_key->value); } else { object_key = malloc (sizeof (struct object_key)); if (object_key == 0) { goto error_put; } object_key->key_name = malloc (key_len); if (object_key->key_name == 0) { goto error_put_object; } memcpy (object_key->key_name, key_name, key_len); list_init (&object_key->list); list_add_tail (&object_key->list, &instance->key_head); } object_key->value = malloc (value_len); if (object_key->value == 0) { goto error_put_key; } memcpy (object_key->value, value, value_len); object_key->key_len = key_len; object_key->value_len = value_len; object_key_changed_notification(object_handle, key_name, key_len, value, value_len, OBJECT_KEY_CREATED); objdb_rdunlock(); return (0); error_put_key: free (object_key->key_name); error_put_object: free (object_key); error_put: hdb_handle_put (&object_instance_database, object_handle); error_exit: objdb_rdunlock(); return (-1); } static int _clear_object(struct object_instance *instance) { struct list_head *list; int res; struct object_instance *find_instance = NULL; struct object_key *object_key = NULL; for (list = instance->key_head.next; list != &instance->key_head; ) { object_key = list_entry (list, struct object_key, list); list = list->next; list_del(&object_key->list); free(object_key->key_name); free(object_key->value); } for (list = instance->child_head.next; list != &instance->child_head; ) { find_instance = list_entry (list, struct object_instance, child_list); res = _clear_object(find_instance); if (res) return res; list = list->next; list_del(&find_instance->child_list); free(find_instance->object_name); free(find_instance); } return 0; } static int object_destroy ( hdb_handle_t object_handle) { struct object_instance *instance; unsigned int res; objdb_rdlock(); res = hdb_handle_get (&object_instance_database, object_handle, (void *)&instance); if (res != 0) { objdb_rdunlock(); return (res); } object_pre_deletion_notification(object_handle, instance->parent_handle, instance->object_name, instance->object_name_len); /* Recursively clear sub-objects & keys */ res = _clear_object(instance); list_del(&instance->child_list); free(instance->object_name); free(instance); objdb_rdunlock(); return (res); } static int object_valid_set ( hdb_handle_t object_handle, struct object_valid *object_valid_list, size_t object_valid_list_entries) { struct object_instance *instance; unsigned int res; objdb_rdlock(); res = hdb_handle_get (&object_instance_database, object_handle, (void *)&instance); if (res != 0) { goto error_exit; } instance->object_valid_list = object_valid_list; instance->object_valid_list_entries = object_valid_list_entries; hdb_handle_put (&object_instance_database, object_handle); objdb_rdunlock(); return (0); error_exit: objdb_rdunlock(); return (-1); } static int object_key_valid_set ( hdb_handle_t object_handle, struct object_key_valid *object_key_valid_list, size_t object_key_valid_list_entries) { struct object_instance *instance; unsigned int res; objdb_rdlock(); res = hdb_handle_get (&object_instance_database, object_handle, (void *)&instance); if (res != 0) { goto error_exit; } instance->object_key_valid_list = object_key_valid_list; instance->object_key_valid_list_entries = object_key_valid_list_entries; hdb_handle_put (&object_instance_database, object_handle); objdb_rdunlock(); return (0); error_exit: objdb_rdunlock(); return (-1); } /* * object db reading */ static int object_find_create ( hdb_handle_t object_handle, const void *object_name, size_t object_len, hdb_handle_t *object_find_handle) { unsigned int res; struct object_instance *object_instance; struct object_find_instance *object_find_instance; objdb_rdlock(); res = hdb_handle_get (&object_instance_database, object_handle, (void *)&object_instance); if (res != 0) { goto error_exit; } res = hdb_handle_create (&object_find_instance_database, sizeof (struct object_find_instance), object_find_handle); if (res != 0) { goto error_put; } res = hdb_handle_get (&object_find_instance_database, *object_find_handle, (void *)&object_find_instance); if (res != 0) { goto error_destroy; } object_find_instance->find_child_list = &object_instance->child_head; object_find_instance->child_head = &object_instance->child_head; object_find_instance->object_name = object_name; object_find_instance->object_len = object_len; hdb_handle_put (&object_instance_database, object_handle); hdb_handle_put (&object_find_instance_database, *object_find_handle); objdb_rdunlock(); return (0); error_destroy: hdb_handle_destroy (&object_instance_database, *object_find_handle); error_put: hdb_handle_put (&object_instance_database, object_handle); error_exit: objdb_rdunlock(); return (-1); } static int object_find_next ( hdb_handle_t object_find_handle, hdb_handle_t *object_handle) { unsigned int res; struct object_find_instance *object_find_instance; struct object_instance *object_instance = NULL; struct list_head *list; unsigned int found = 0; objdb_rdlock(); res = hdb_handle_get (&object_find_instance_database, object_find_handle, (void *)&object_find_instance); if (res != 0) { goto error_exit; } res = -1; for (list = object_find_instance->find_child_list->next; list != object_find_instance->child_head; list = list->next) { object_instance = list_entry (list, struct object_instance, child_list); if (object_find_instance->object_len == 0 || ((object_instance->object_name_len == object_find_instance->object_len) && (memcmp (object_instance->object_name, object_find_instance->object_name, object_find_instance->object_len) == 0))) { found = 1; break; } } object_find_instance->find_child_list = list; hdb_handle_put (&object_find_instance_database, object_find_handle); if (found) { *object_handle = object_instance->object_handle; res = 0; } objdb_rdunlock(); return (res); error_exit: objdb_rdunlock(); return (-1); } static int object_find_destroy ( hdb_handle_t object_find_handle) { struct object_find_instance *object_find_instance; unsigned int res; objdb_rdlock(); res = hdb_handle_get (&object_find_instance_database, object_find_handle, (void *)&object_find_instance); if (res != 0) { goto error_exit; } hdb_handle_put(&object_find_instance_database, object_find_handle); hdb_handle_destroy(&object_find_instance_database, object_find_handle); objdb_rdunlock(); return (0); error_exit: objdb_rdunlock(); return (-1); } static int object_key_get ( hdb_handle_t object_handle, const void *key_name, size_t key_len, void **value, size_t *value_len) { unsigned int res = 0; struct object_instance *instance; struct object_key *object_key = NULL; struct list_head *list; int found = 0; objdb_rdlock(); res = hdb_handle_get (&object_instance_database, object_handle, (void *)&instance); if (res != 0) { goto error_exit; } for (list = instance->key_head.next; list != &instance->key_head; list = list->next) { object_key = list_entry (list, struct object_key, list); if ((object_key->key_len == key_len) && (memcmp (object_key->key_name, key_name, key_len) == 0)) { found = 1; break; } } if (found) { *value = object_key->value; if (value_len) { *value_len = object_key->value_len; } } else { res = -1; } hdb_handle_put (&object_instance_database, object_handle); objdb_rdunlock(); return (res); error_exit: objdb_rdunlock(); return (-1); } static int object_key_increment ( hdb_handle_t object_handle, const void *key_name, size_t key_len, unsigned int *value) { unsigned int res = 0; struct object_instance *instance; struct object_key *object_key = NULL; struct list_head *list; int found = 0; objdb_rdlock(); res = hdb_handle_get (&object_instance_database, object_handle, (void *)&instance); if (res != 0) { goto error_exit; } for (list = instance->key_head.next; list != &instance->key_head; list = list->next) { object_key = list_entry (list, struct object_key, list); if ((object_key->key_len == key_len) && (memcmp (object_key->key_name, key_name, key_len) == 0)) { found = 1; break; } } if (found && object_key->value_len == sizeof(int)) { (*(int *)object_key->value)++; *value = *(int *)object_key->value; } else { res = -1; } hdb_handle_put (&object_instance_database, object_handle); objdb_rdunlock(); return (res); error_exit: objdb_rdunlock(); return (-1); } static int object_key_decrement ( hdb_handle_t object_handle, const void *key_name, size_t key_len, unsigned int *value) { unsigned int res = 0; struct object_instance *instance; struct object_key *object_key = NULL; struct list_head *list; int found = 0; objdb_rdlock(); res = hdb_handle_get (&object_instance_database, object_handle, (void *)&instance); if (res != 0) { goto error_exit; } for (list = instance->key_head.next; list != &instance->key_head; list = list->next) { object_key = list_entry (list, struct object_key, list); if ((object_key->key_len == key_len) && (memcmp (object_key->key_name, key_name, key_len) == 0)) { found = 1; break; } } if (found && object_key->value_len == sizeof(int)) { (*(int *)object_key->value)--; *value = *(int *)object_key->value; } else { res = -1; } hdb_handle_put (&object_instance_database, object_handle); objdb_rdunlock(); return (res); error_exit: objdb_rdunlock(); return (-1); } static int object_key_delete ( hdb_handle_t object_handle, const void *key_name, size_t key_len) { unsigned int res; int ret = 0; struct object_instance *instance; struct object_key *object_key = NULL; struct list_head *list; int found = 0; objdb_rdlock(); res = hdb_handle_get (&object_instance_database, object_handle, (void *)&instance); if (res != 0) { goto error_exit; } for (list = instance->key_head.next; list != &instance->key_head; list = list->next) { object_key = list_entry (list, struct object_key, list); if ((object_key->key_len == key_len) && (memcmp (object_key->key_name, key_name, key_len) == 0)) { found = 1; break; } } if (found) { list_del(&object_key->list); free(object_key->key_name); free(object_key->value); free(object_key); } else { ret = -1; errno = ENOENT; } hdb_handle_put (&object_instance_database, object_handle); if (ret == 0) object_key_changed_notification(object_handle, key_name, key_len, NULL, 0, OBJECT_KEY_DELETED); objdb_rdunlock(); return (ret); error_exit: objdb_rdunlock(); return (-1); } static int object_key_replace ( hdb_handle_t object_handle, const void *key_name, size_t key_len, const void *new_value, size_t new_value_len) { unsigned int res; int ret = 0; struct object_instance *instance; struct object_key *object_key = NULL; struct list_head *list; int found = 0; objdb_rdlock(); res = hdb_handle_get (&object_instance_database, object_handle, (void *)&instance); if (res != 0) { goto error_exit; } for (list = instance->key_head.next; list != &instance->key_head; list = list->next) { object_key = list_entry (list, struct object_key, list); if ((object_key->key_len == key_len) && (memcmp (object_key->key_name, key_name, key_len) == 0)) { found = 1; break; } } if (found) { int i; int found_validator = 0; /* * Do validation check if validation is configured for the parent object */ if (instance->object_key_valid_list_entries) { for (i = 0; i < instance->object_key_valid_list_entries; i++) { if ((key_len == instance->object_key_valid_list[i].key_len) && (memcmp (key_name, instance->object_key_valid_list[i].key_name, key_len) == 0)) { found_validator = 1; break; } } /* * Item not found in validation list */ if (found_validator == 0) { goto error_put; } else { if (instance->object_key_valid_list[i].validate_callback) { res = instance->object_key_valid_list[i].validate_callback ( key_name, key_len, new_value, new_value_len); if (res != 0) { goto error_put; } } } } if (new_value_len != object_key->value_len) { void *replacement_value; replacement_value = malloc(new_value_len); if (!replacement_value) goto error_exit; free(object_key->value); object_key->value = replacement_value; } memcpy(object_key->value, new_value, new_value_len); object_key->value_len = new_value_len; } else { ret = -1; errno = ENOENT; } hdb_handle_put (&object_instance_database, object_handle); if (ret == 0) object_key_changed_notification(object_handle, key_name, key_len, new_value, new_value_len, OBJECT_KEY_REPLACED); objdb_rdunlock(); return (ret); error_put: hdb_handle_put (&object_instance_database, object_handle); error_exit: objdb_rdunlock(); return (-1); } static int object_priv_get ( hdb_handle_t object_handle, void **priv) { int res; struct object_instance *object_instance; objdb_rdunlock(); res = hdb_handle_get (&object_instance_database, object_handle, (void *)&object_instance); if (res != 0) { goto error_exit; } *priv = object_instance->priv; hdb_handle_put (&object_instance_database, object_handle); objdb_rdunlock(); return (0); error_exit: objdb_rdunlock(); return (-1); } static int _dump_object(struct object_instance *instance, FILE *file, int depth) { struct list_head *list; int res; int i; struct object_instance *find_instance = NULL; struct object_key *object_key = NULL; char stringbuf1[1024]; char stringbuf2[1024]; memcpy(stringbuf1, instance->object_name, instance->object_name_len); stringbuf1[instance->object_name_len] = '\0'; for (i=0; iobject_handle != OBJECT_PARENT_HANDLE) fprintf(file, "%s {\n", stringbuf1); for (list = instance->key_head.next; list != &instance->key_head; list = list->next) { object_key = list_entry (list, struct object_key, list); memcpy(stringbuf1, object_key->key_name, object_key->key_len); stringbuf1[object_key->key_len] = '\0'; memcpy(stringbuf2, object_key->value, object_key->value_len); stringbuf2[object_key->value_len] = '\0'; for (i=0; ichild_head.next; list != &instance->child_head; list = list->next) { find_instance = list_entry (list, struct object_instance, child_list); res = _dump_object(find_instance, file, depth+1); if (res) return res; } for (i=0; iobject_handle != OBJECT_PARENT_HANDLE) fprintf(file, "}\n"); return 0; } static int object_key_iter_reset(hdb_handle_t object_handle) { unsigned int res; struct object_instance *instance; objdb_rdlock(); res = hdb_handle_get (&object_instance_database, object_handle, (void *)&instance); if (res != 0) { goto error_exit; } instance->iter_key_list = &instance->key_head; hdb_handle_put (&object_instance_database, object_handle); objdb_rdunlock(); return (0); error_exit: objdb_rdunlock(); return (-1); } static int object_key_iter(hdb_handle_t parent_object_handle, void **key_name, size_t *key_len, void **value, size_t *value_len) { unsigned int res; struct object_instance *instance; struct object_key *find_key = NULL; struct list_head *list; unsigned int found = 0; objdb_rdlock(); res = hdb_handle_get (&object_instance_database, parent_object_handle, (void *)&instance); if (res != 0) { goto error_exit; } res = -ENOENT; list = instance->iter_key_list->next; if (list != &instance->key_head) { find_key = list_entry (list, struct object_key, list); found = 1; } instance->iter_key_list = list; if (found) { *key_name = find_key->key_name; if (key_len) *key_len = find_key->key_len; *value = find_key->value; if (value_len) *value_len = find_key->value_len; res = 0; } else { res = -1; } hdb_handle_put (&object_instance_database, parent_object_handle); objdb_rdunlock(); return (res); error_exit: objdb_rdunlock(); return (-1); } static int object_key_iter_from(hdb_handle_t parent_object_handle, hdb_handle_t start_pos, void **key_name, size_t *key_len, void **value, size_t *value_len) { unsigned int pos = 0; unsigned int res; struct object_instance *instance; struct object_key *find_key = NULL; struct list_head *list; unsigned int found = 0; objdb_rdlock(); res = hdb_handle_get (&object_instance_database, parent_object_handle, (void *)&instance); if (res != 0) { goto error_exit; } res = -ENOENT; for (list = instance->key_head.next; list != &instance->key_head; list = list->next) { find_key = list_entry (list, struct object_key, list); if (pos++ == start_pos) { found = 1; break; } } if (found) { *key_name = find_key->key_name; if (key_len) *key_len = find_key->key_len; *value = find_key->value; if (value_len) *value_len = find_key->value_len; res = 0; } else { res = -1; } hdb_handle_put (&object_instance_database, parent_object_handle); objdb_rdunlock(); return (res); error_exit: objdb_rdunlock(); return (-1); } static int object_parent_get(hdb_handle_t object_handle, hdb_handle_t *parent_handle) { struct object_instance *instance; unsigned int res; objdb_rdlock(); res = hdb_handle_get (&object_instance_database, object_handle, (void *)&instance); if (res != 0) { objdb_rdunlock(); return (res); } if (object_handle == OBJECT_PARENT_HANDLE) *parent_handle = 0; else *parent_handle = instance->parent_handle; hdb_handle_put (&object_instance_database, object_handle); objdb_rdunlock(); return (0); } static int object_name_get(hdb_handle_t object_handle, char *object_name, size_t *object_name_len) { struct object_instance *instance; unsigned int res; objdb_rdlock(); res = hdb_handle_get (&object_instance_database, object_handle, (void *)&instance); if (res != 0) { objdb_rdunlock(); return (res); } memcpy(object_name, instance->object_name, instance->object_name_len); *object_name_len = instance->object_name_len; hdb_handle_put (&object_instance_database, object_handle); objdb_rdunlock(); return (0); } static int object_track_start(hdb_handle_t object_handle, object_track_depth_t depth, object_key_change_notify_fn_t key_change_notify_fn, object_create_notify_fn_t object_create_notify_fn, object_destroy_notify_fn_t object_destroy_notify_fn, object_reload_notify_fn_t object_reload_notify_fn, void * priv_data_pt) { struct object_instance *instance; unsigned int res; struct object_tracker * tracker_pt; res = hdb_handle_get (&object_instance_database, object_handle, (void *)&instance); if (res != 0) { return (res); } tracker_pt = malloc(sizeof(struct object_tracker)); tracker_pt->depth = depth; tracker_pt->object_handle = object_handle; tracker_pt->key_change_notify_fn = key_change_notify_fn; tracker_pt->object_create_notify_fn = object_create_notify_fn; tracker_pt->object_destroy_notify_fn = object_destroy_notify_fn; tracker_pt->object_reload_notify_fn = object_reload_notify_fn; tracker_pt->data_pt = priv_data_pt; list_init(&tracker_pt->object_list); list_init(&tracker_pt->tracker_list); list_add(&tracker_pt->object_list, &instance->track_head); list_add(&tracker_pt->tracker_list, &objdb_trackers_head); hdb_handle_put (&object_instance_database, object_handle); return (res); } static void object_track_stop(object_key_change_notify_fn_t key_change_notify_fn, object_create_notify_fn_t object_create_notify_fn, object_destroy_notify_fn_t object_destroy_notify_fn, object_reload_notify_fn_t object_reload_notify_fn, void * priv_data_pt) { struct object_instance *instance; struct object_tracker * tracker_pt = NULL; struct object_tracker * obj_tracker_pt = NULL; struct list_head *list, *tmp_list; struct list_head *obj_list, *tmp_obj_list; unsigned int res; /* go through the global list and find all the trackers to stop */ for (list = objdb_trackers_head.next, tmp_list = list->next; list != &objdb_trackers_head; list = tmp_list, tmp_list = tmp_list->next) { tracker_pt = list_entry (list, struct object_tracker, tracker_list); if (tracker_pt && (tracker_pt->data_pt == priv_data_pt) && (tracker_pt->object_create_notify_fn == object_create_notify_fn) && (tracker_pt->object_destroy_notify_fn == object_destroy_notify_fn) && (tracker_pt->object_reload_notify_fn == object_reload_notify_fn) && (tracker_pt->key_change_notify_fn == key_change_notify_fn)) { /* get the object & take this tracker off of it's list. */ res = hdb_handle_get (&object_instance_database, tracker_pt->object_handle, (void *)&instance); if (res != 0) continue; for (obj_list = instance->track_head.next, tmp_obj_list = obj_list->next; obj_list != &instance->track_head; obj_list = tmp_obj_list, tmp_obj_list = tmp_obj_list->next) { obj_tracker_pt = list_entry (obj_list, struct object_tracker, object_list); if (obj_tracker_pt == tracker_pt) { /* this is the tracker we are after. */ list_del(obj_list); } } hdb_handle_put (&object_instance_database, tracker_pt->object_handle); /* remove the tracker off of the global list */ list_del(list); free(tracker_pt); } } } static int object_dump(hdb_handle_t object_handle, FILE *file) { struct object_instance *instance; unsigned int res; objdb_rdlock(); res = hdb_handle_get (&object_instance_database, object_handle, (void *)&instance); if (res != 0) { objdb_rdunlock(); return (res); } res = _dump_object(instance, file, -1); hdb_handle_put (&object_instance_database, object_handle); objdb_rdunlock(); return (res); } static int object_write_config(const char **error_string) { struct config_iface_ver0 **modules; int num_modules; int i; int res; main_get_config_modules(&modules, &num_modules); objdb_wrlock(); for (i=0; iconfig_writeconfig) { res = modules[i]->config_writeconfig(&objdb_iface, error_string); if (res) { objdb_wrunlock(); return res; } } } objdb_wrunlock(); return 0; } static int object_reload_config(int flush, const char **error_string) { struct config_iface_ver0 **modules; int num_modules; int i; int res; main_get_config_modules(&modules, &num_modules); object_reload_notification(OBJDB_RELOAD_NOTIFY_START, flush); objdb_wrlock(); for (i=0; iconfig_reloadconfig) { res = modules[i]->config_reloadconfig(&objdb_iface, flush, error_string); if (res) { object_reload_notification(OBJDB_RELOAD_NOTIFY_FAILED, flush); objdb_wrunlock(); return res; } } } objdb_wrunlock(); object_reload_notification(OBJDB_RELOAD_NOTIFY_END, flush); return 0; } struct objdb_iface_ver0 objdb_iface = { .objdb_init = objdb_init, .object_create = object_create, .object_priv_set = object_priv_set, .object_key_create = object_key_create, .object_key_delete = object_key_delete, .object_key_replace = object_key_replace, .object_destroy = object_destroy, .object_valid_set = object_valid_set, .object_key_valid_set = object_key_valid_set, .object_find_create = object_find_create, .object_find_next = object_find_next, .object_find_destroy = object_find_destroy, .object_key_get = object_key_get, .object_key_iter_reset = object_key_iter_reset, .object_key_iter = object_key_iter, .object_key_iter_from = object_key_iter_from, .object_priv_get = object_priv_get, .object_parent_get = object_parent_get, .object_name_get = object_name_get, .object_track_start = object_track_start, .object_track_stop = object_track_stop, .object_dump = object_dump, .object_write_config = object_write_config, .object_reload_config = object_reload_config, .object_key_increment = object_key_increment, .object_key_decrement = object_key_decrement, }; struct lcr_iface objdb_iface_ver0[1] = { { .name = "objdb", .version = 0, .versions_replace = 0, .versions_replace_count = 0, .dependencies = 0, .dependency_count = 0, .constructor = NULL, .destructor = NULL, .interfaces = NULL, } }; struct lcr_comp objdb_comp_ver0 = { .iface_count = 1, .ifaces = objdb_iface_ver0 }; __attribute__ ((constructor)) static void objdb_comp_register (void) { lcr_interfaces_set (&objdb_iface_ver0[0], &objdb_iface); lcr_component_register (&objdb_comp_ver0); } diff --git a/exec/totemnet.c b/exec/totemnet.c index b50752ad..01338ed7 100644 --- a/exec/totemnet.c +++ b/exec/totemnet.c @@ -1,1512 +1,1504 @@ /* * Copyright (c) 2005 MontaVista Software, Inc. * Copyright (c) 2006-2008 Red Hat, Inc. * * All rights reserved. * * Author: Steven Dake (sdake@redhat.com) * 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 MontaVista Software, Inc. 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 #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include "totemnet.h" #include "wthread.h" #include "crypto.h" #define MCAST_SOCKET_BUFFER_SIZE (TRANSMITS_ALLOWED * FRAME_SIZE_MAX) #define NETIF_STATE_REPORT_UP 1 #define NETIF_STATE_REPORT_DOWN 2 #define BIND_STATE_UNBOUND 0 #define BIND_STATE_REGULAR 1 #define BIND_STATE_LOOPBACK 2 #define HMAC_HASH_SIZE 20 struct security_header { unsigned char hash_digest[HMAC_HASH_SIZE]; /* The hash *MUST* be first in the data structure */ unsigned char salt[16]; /* random number */ char msg[0]; } __attribute__((packed)); struct totemnet_mcast_thread_state { unsigned char iobuf[FRAME_SIZE_MAX]; prng_state prng_state; }; struct totemnet_socket { int mcast_recv; int mcast_send; int token; }; struct totemnet_instance { hmac_state totemnet_hmac_state; prng_state totemnet_prng_state; unsigned char totemnet_private_key[1024]; unsigned int totemnet_private_key_len; hdb_handle_t totemnet_poll_handle; struct totem_interface *totem_interface; int netif_state_report; int netif_bind_state; struct worker_thread_group worker_thread_group; void *context; void (*totemnet_deliver_fn) ( void *context, void *msg, int msg_len); void (*totemnet_iface_change_fn) ( void *context, struct totem_ip_address *iface_address); /* * Function and data used to log messages */ int totemnet_log_level_security; int totemnet_log_level_error; int totemnet_log_level_warning; int totemnet_log_level_notice; int totemnet_log_level_debug; int totemnet_subsys_id; void (*totemnet_log_printf) (int subsys, const char *function, const char *file, int line, unsigned int level, const char *format, ...)__attribute__((format(printf, 6, 7))); hdb_handle_t handle; char iov_buffer[FRAME_SIZE_MAX]; char iov_buffer_flush[FRAME_SIZE_MAX]; struct iovec totemnet_iov_recv; struct iovec totemnet_iov_recv_flush; struct totemnet_socket totemnet_sockets; struct totem_ip_address mcast_address; int stats_sent; int stats_recv; int stats_delv; int stats_remcasts; int stats_orf_token; struct timeval stats_tv_start; struct totem_ip_address my_id; int firstrun; poll_timer_handle timer_netif_check_timeout; unsigned int my_memb_entries; int flushing; struct totem_config *totem_config; struct totem_ip_address token_target; }; struct work_item { struct iovec iovec[20]; unsigned int iov_len; struct totemnet_instance *instance; }; static void netif_down_check (struct totemnet_instance *instance); static int totemnet_build_sockets ( struct totemnet_instance *instance, struct totem_ip_address *bindnet_address, struct totem_ip_address *mcastaddress, struct totemnet_socket *sockets, struct totem_ip_address *bound_to); static struct totem_ip_address localhost; -/* - * All instances in one database - */ -static struct hdb_handle_database totemnet_instance_database = { - .handle_count = 0, - .handles = 0, - .iterator = 0, - .mutex = PTHREAD_MUTEX_INITIALIZER -}; +DECLARE_HDB_DATABASE (totemnet_instance_database); static void totemnet_instance_initialize (struct totemnet_instance *instance) { memset (instance, 0, sizeof (struct totemnet_instance)); instance->netif_state_report = NETIF_STATE_REPORT_UP | NETIF_STATE_REPORT_DOWN; instance->totemnet_iov_recv.iov_base = instance->iov_buffer; instance->totemnet_iov_recv.iov_len = FRAME_SIZE_MAX; //sizeof (instance->iov_buffer); instance->totemnet_iov_recv_flush.iov_base = instance->iov_buffer_flush; instance->totemnet_iov_recv_flush.iov_len = FRAME_SIZE_MAX; //sizeof (instance->iov_buffer); /* * There is always atleast 1 processor */ instance->my_memb_entries = 1; } #define log_printf(level, format, args...) \ do { \ instance->totemnet_log_printf (instance->totemnet_subsys_id, \ __FUNCTION__, __FILE__, __LINE__, \ level, (const char *)format, ##args); \ } while (0); static int authenticate_and_decrypt ( struct totemnet_instance *instance, struct iovec *iov) { unsigned char keys[48]; struct security_header *header = iov[0].iov_base; prng_state keygen_prng_state; prng_state stream_prng_state; unsigned char *hmac_key = &keys[32]; unsigned char *cipher_key = &keys[16]; unsigned char *initial_vector = &keys[0]; unsigned char digest_comparison[HMAC_HASH_SIZE]; unsigned long len; /* * Generate MAC, CIPHER, IV keys from private key */ memset (keys, 0, sizeof (keys)); sober128_start (&keygen_prng_state); sober128_add_entropy (instance->totemnet_private_key, instance->totemnet_private_key_len, &keygen_prng_state); sober128_add_entropy (header->salt, sizeof (header->salt), &keygen_prng_state); sober128_read (keys, sizeof (keys), &keygen_prng_state); /* * Setup stream cipher */ sober128_start (&stream_prng_state); sober128_add_entropy (cipher_key, 16, &stream_prng_state); sober128_add_entropy (initial_vector, 16, &stream_prng_state); /* * Authenticate contents of message */ hmac_init (&instance->totemnet_hmac_state, DIGEST_SHA1, hmac_key, 16); hmac_process (&instance->totemnet_hmac_state, (unsigned char *)iov->iov_base + HMAC_HASH_SIZE, iov->iov_len - HMAC_HASH_SIZE); len = hash_descriptor[DIGEST_SHA1]->hashsize; assert (HMAC_HASH_SIZE >= len); hmac_done (&instance->totemnet_hmac_state, digest_comparison, &len); if (memcmp (digest_comparison, header->hash_digest, len) != 0) { log_printf (instance->totemnet_log_level_security, "Received message has invalid digest... ignoring.\n"); return (-1); } /* * Decrypt the contents of the message with the cipher key */ sober128_read ((unsigned char*)iov->iov_base + sizeof (struct security_header), iov->iov_len - sizeof (struct security_header), &stream_prng_state); return (0); } static void encrypt_and_sign_worker ( struct totemnet_instance *instance, unsigned char *buf, int *buf_len, struct iovec *iovec, unsigned int iov_len, prng_state *prng_state_in) { int i; unsigned char *addr; unsigned char keys[48]; struct security_header *header; unsigned char *hmac_key = &keys[32]; unsigned char *cipher_key = &keys[16]; unsigned char *initial_vector = &keys[0]; unsigned long len; int outlen = 0; hmac_state hmac_state; prng_state keygen_prng_state; prng_state stream_prng_state; header = (struct security_header *)buf; addr = buf + sizeof (struct security_header); memset (keys, 0, sizeof (keys)); memset (header->salt, 0, sizeof (header->salt)); /* * Generate MAC, CIPHER, IV keys from private key */ sober128_read (header->salt, sizeof (header->salt), prng_state_in); sober128_start (&keygen_prng_state); sober128_add_entropy (instance->totemnet_private_key, instance->totemnet_private_key_len, &keygen_prng_state); sober128_add_entropy (header->salt, sizeof (header->salt), &keygen_prng_state); sober128_read (keys, sizeof (keys), &keygen_prng_state); /* * Setup stream cipher */ sober128_start (&stream_prng_state); sober128_add_entropy (cipher_key, 16, &stream_prng_state); sober128_add_entropy (initial_vector, 16, &stream_prng_state); outlen = sizeof (struct security_header); /* * Copy remainder of message, then encrypt it */ for (i = 1; i < iov_len; i++) { memcpy (addr, iovec[i].iov_base, iovec[i].iov_len); addr += iovec[i].iov_len; outlen += iovec[i].iov_len; } /* * Encrypt message by XORing stream cipher data */ sober128_read (buf + sizeof (struct security_header), outlen - sizeof (struct security_header), &stream_prng_state); memset (&hmac_state, 0, sizeof (hmac_state)); /* * Sign the contents of the message with the hmac key and store signature in message */ hmac_init (&hmac_state, DIGEST_SHA1, hmac_key, 16); hmac_process (&hmac_state, buf + HMAC_HASH_SIZE, outlen - HMAC_HASH_SIZE); len = hash_descriptor[DIGEST_SHA1]->hashsize; hmac_done (&hmac_state, header->hash_digest, &len); *buf_len = outlen; } static inline void ucast_sendmsg ( struct totemnet_instance *instance, struct totem_ip_address *system_to, struct iovec *iovec_in, unsigned int iov_len_in) { struct msghdr msg_ucast; int res = 0; int buf_len; unsigned char sheader[sizeof (struct security_header)]; unsigned char encrypt_data[FRAME_SIZE_MAX]; struct iovec iovec_encrypt[20]; struct iovec *iovec_sendmsg; struct sockaddr_storage sockaddr; unsigned int iov_len; int addrlen; if (instance->totem_config->secauth == 1) { iovec_encrypt[0].iov_base = sheader; iovec_encrypt[0].iov_len = sizeof (struct security_header); memcpy (&iovec_encrypt[1], &iovec_in[0], sizeof (struct iovec) * iov_len_in); /* * Encrypt and digest the message */ encrypt_and_sign_worker ( instance, encrypt_data, &buf_len, iovec_encrypt, iov_len_in + 1, &instance->totemnet_prng_state); iovec_encrypt[0].iov_base = encrypt_data; iovec_encrypt[0].iov_len = buf_len; iovec_sendmsg = &iovec_encrypt[0]; iov_len = 1; } else { iovec_sendmsg = iovec_in; iov_len = iov_len_in; } /* * Build unicast message */ totemip_totemip_to_sockaddr_convert(system_to, instance->totem_interface->ip_port, &sockaddr, &addrlen); msg_ucast.msg_name = &sockaddr; msg_ucast.msg_namelen = addrlen; msg_ucast.msg_iov = iovec_sendmsg; msg_ucast.msg_iovlen = iov_len; msg_ucast.msg_control = 0; msg_ucast.msg_controllen = 0; msg_ucast.msg_flags = 0; /* * Transmit multicast message * An error here is recovered by totemsrp */ res = sendmsg (instance->totemnet_sockets.mcast_send, &msg_ucast, MSG_NOSIGNAL); } static inline void mcast_sendmsg ( struct totemnet_instance *instance, struct iovec *iovec_in, unsigned int iov_len_in) { struct msghdr msg_mcast; int res = 0; int buf_len; unsigned char sheader[sizeof (struct security_header)]; unsigned char encrypt_data[FRAME_SIZE_MAX]; struct iovec iovec_encrypt[20]; struct iovec *iovec_sendmsg; struct sockaddr_storage sockaddr; unsigned int iov_len; int addrlen; if (instance->totem_config->secauth == 1) { iovec_encrypt[0].iov_base = sheader; iovec_encrypt[0].iov_len = sizeof (struct security_header); memcpy (&iovec_encrypt[1], &iovec_in[0], sizeof (struct iovec) * iov_len_in); /* * Encrypt and digest the message */ encrypt_and_sign_worker ( instance, encrypt_data, &buf_len, iovec_encrypt, iov_len_in + 1, &instance->totemnet_prng_state); iovec_encrypt[0].iov_base = encrypt_data; iovec_encrypt[0].iov_len = buf_len; iovec_sendmsg = &iovec_encrypt[0]; iov_len = 1; } else { iovec_sendmsg = iovec_in; iov_len = iov_len_in; } /* * Build multicast message */ totemip_totemip_to_sockaddr_convert(&instance->mcast_address, instance->totem_interface->ip_port, &sockaddr, &addrlen); msg_mcast.msg_name = &sockaddr; msg_mcast.msg_namelen = addrlen; msg_mcast.msg_iov = iovec_sendmsg; msg_mcast.msg_iovlen = iov_len; msg_mcast.msg_control = 0; msg_mcast.msg_controllen = 0; msg_mcast.msg_flags = 0; /* * Transmit multicast message * An error here is recovered by totemsrp */ res = sendmsg (instance->totemnet_sockets.mcast_send, &msg_mcast, MSG_NOSIGNAL); } static void totemnet_mcast_thread_state_constructor ( void *totemnet_mcast_thread_state_in) { struct totemnet_mcast_thread_state *totemnet_mcast_thread_state = (struct totemnet_mcast_thread_state *)totemnet_mcast_thread_state_in; memset (totemnet_mcast_thread_state, 0, sizeof (totemnet_mcast_thread_state)); rng_make_prng (128, PRNG_SOBER, &totemnet_mcast_thread_state->prng_state, NULL); } static void totemnet_mcast_worker_fn (void *thread_state, void *work_item_in) { struct work_item *work_item = (struct work_item *)work_item_in; struct totemnet_mcast_thread_state *totemnet_mcast_thread_state = (struct totemnet_mcast_thread_state *)thread_state; struct totemnet_instance *instance = work_item->instance; struct msghdr msg_mcast; unsigned char sheader[sizeof (struct security_header)]; int res = 0; int buf_len; struct iovec iovec_encrypted; struct iovec *iovec_sendmsg; struct sockaddr_storage sockaddr; unsigned int iovs; int addrlen; if (instance->totem_config->secauth == 1) { memmove (&work_item->iovec[1], &work_item->iovec[0], work_item->iov_len * sizeof (struct iovec)); work_item->iovec[0].iov_base = sheader; work_item->iovec[0].iov_len = sizeof (struct security_header); /* * Encrypt and digest the message */ encrypt_and_sign_worker ( instance, totemnet_mcast_thread_state->iobuf, &buf_len, work_item->iovec, work_item->iov_len + 1, &totemnet_mcast_thread_state->prng_state); iovec_sendmsg = &iovec_encrypted; iovec_sendmsg->iov_base = totemnet_mcast_thread_state->iobuf; iovec_sendmsg->iov_len = buf_len; iovs = 1; } else { iovec_sendmsg = work_item->iovec; iovs = work_item->iov_len; } totemip_totemip_to_sockaddr_convert(&instance->mcast_address, instance->totem_interface->ip_port, &sockaddr, &addrlen); msg_mcast.msg_name = &sockaddr; msg_mcast.msg_namelen = addrlen; msg_mcast.msg_iov = iovec_sendmsg; msg_mcast.msg_iovlen = iovs; msg_mcast.msg_control = 0; msg_mcast.msg_controllen = 0; msg_mcast.msg_flags = 0; /* * Transmit multicast message * An error here is recovered by totemnet */ res = sendmsg (instance->totemnet_sockets.mcast_send, &msg_mcast, MSG_NOSIGNAL); if (res > 0) { instance->stats_sent += res; } } int totemnet_finalize ( hdb_handle_t handle) { struct totemnet_instance *instance; int res = 0; res = hdb_handle_get (&totemnet_instance_database, handle, (void *)&instance); if (res != 0) { res = ENOENT; goto error_exit; } worker_thread_group_exit (&instance->worker_thread_group); hdb_handle_put (&totemnet_instance_database, handle); error_exit: return (res); } /* * Only designed to work with a message with one iov */ static int net_deliver_fn ( hdb_handle_t handle, int fd, int revents, void *data) { struct totemnet_instance *instance = (struct totemnet_instance *)data; struct msghdr msg_recv; struct iovec *iovec; struct security_header *security_header; struct sockaddr_storage system_from; int bytes_received; int res = 0; unsigned char *msg_offset; unsigned int size_delv; if (instance->flushing == 1) { iovec = &instance->totemnet_iov_recv_flush; } else { iovec = &instance->totemnet_iov_recv; } /* * Receive datagram */ msg_recv.msg_name = &system_from; msg_recv.msg_namelen = sizeof (struct sockaddr_storage); msg_recv.msg_iov = iovec; msg_recv.msg_iovlen = 1; msg_recv.msg_control = 0; msg_recv.msg_controllen = 0; msg_recv.msg_flags = 0; bytes_received = recvmsg (fd, &msg_recv, MSG_NOSIGNAL | MSG_DONTWAIT); if (bytes_received == -1) { return (0); } else { instance->stats_recv += bytes_received; } if ((instance->totem_config->secauth == 1) && (bytes_received < sizeof (struct security_header))) { log_printf (instance->totemnet_log_level_security, "Received message is too short... ignoring %d.\n", bytes_received); return (0); } security_header = (struct security_header *)iovec->iov_base; iovec->iov_len = bytes_received; if (instance->totem_config->secauth == 1) { /* * Authenticate and if authenticated, decrypt datagram */ res = authenticate_and_decrypt (instance, iovec); if (res == -1) { log_printf (instance->totemnet_log_level_security, "Invalid packet data\n"); iovec->iov_len = FRAME_SIZE_MAX; return 0; } msg_offset = (unsigned char *)iovec->iov_base + sizeof (struct security_header); size_delv = bytes_received - sizeof (struct security_header); } else { msg_offset = iovec->iov_base; size_delv = bytes_received; } /* * Handle incoming message */ instance->totemnet_deliver_fn ( instance->context, msg_offset, size_delv); iovec->iov_len = FRAME_SIZE_MAX; return (0); } static int netif_determine ( struct totemnet_instance *instance, struct totem_ip_address *bindnet, struct totem_ip_address *bound_to, int *interface_up, int *interface_num) { int res; res = totemip_iface_check (bindnet, bound_to, interface_up, interface_num, + 0); // TODO andrew can address this instance->totem_config->clear_node_high_bit); return (res); } /* * If the interface is up, the sockets for totem are built. If the interface is down * this function is requeued in the timer list to retry building the sockets later. */ static void timer_function_netif_check_timeout ( void *data) { struct totemnet_instance *instance = (struct totemnet_instance *)data; int res; int interface_up; int interface_num; struct totem_ip_address *bind_address; /* * Build sockets for every interface */ netif_determine (instance, &instance->totem_interface->bindnet, &instance->totem_interface->boundto, &interface_up, &interface_num); /* * If the network interface isn't back up and we are already * in loopback mode, add timer to check again and return */ if ((instance->netif_bind_state == BIND_STATE_LOOPBACK && interface_up == 0) || (instance->my_memb_entries == 1 && instance->netif_bind_state == BIND_STATE_REGULAR && interface_up == 1)) { poll_timer_add (instance->totemnet_poll_handle, instance->totem_config->downcheck_timeout, (void *)instance, timer_function_netif_check_timeout, &instance->timer_netif_check_timeout); /* * Add a timer to check for a downed regular interface */ return; } if (instance->totemnet_sockets.mcast_recv > 0) { close (instance->totemnet_sockets.mcast_recv); poll_dispatch_delete (instance->totemnet_poll_handle, instance->totemnet_sockets.mcast_recv); } if (instance->totemnet_sockets.mcast_send > 0) { close (instance->totemnet_sockets.mcast_send); } if (instance->totemnet_sockets.token > 0) { close (instance->totemnet_sockets.token); poll_dispatch_delete (instance->totemnet_poll_handle, instance->totemnet_sockets.token); } if (interface_up == 0) { /* * Interface is not up */ instance->netif_bind_state = BIND_STATE_LOOPBACK; bind_address = &localhost; /* * Add a timer to retry building interfaces and request memb_gather_enter */ poll_timer_add (instance->totemnet_poll_handle, instance->totem_config->downcheck_timeout, (void *)instance, timer_function_netif_check_timeout, &instance->timer_netif_check_timeout); } else { /* * Interface is up */ instance->netif_bind_state = BIND_STATE_REGULAR; bind_address = &instance->totem_interface->bindnet; } /* * Create and bind the multicast and unicast sockets */ res = totemnet_build_sockets (instance, &instance->mcast_address, bind_address, &instance->totemnet_sockets, &instance->totem_interface->boundto); poll_dispatch_add ( instance->totemnet_poll_handle, instance->totemnet_sockets.mcast_recv, POLLIN, instance, net_deliver_fn); poll_dispatch_add ( instance->totemnet_poll_handle, instance->totemnet_sockets.token, POLLIN, instance, net_deliver_fn); totemip_copy (&instance->my_id, &instance->totem_interface->boundto); /* * This reports changes in the interface to the user and totemsrp */ if (instance->netif_bind_state == BIND_STATE_REGULAR) { if (instance->netif_state_report & NETIF_STATE_REPORT_UP) { log_printf (instance->totemnet_log_level_notice, "The network interface [%s] is now up.\n", totemip_print (&instance->totem_interface->boundto)); instance->netif_state_report = NETIF_STATE_REPORT_DOWN; instance->totemnet_iface_change_fn (instance->context, &instance->my_id); } /* * Add a timer to check for interface going down in single membership */ if (instance->my_memb_entries == 1) { poll_timer_add (instance->totemnet_poll_handle, instance->totem_config->downcheck_timeout, (void *)instance, timer_function_netif_check_timeout, &instance->timer_netif_check_timeout); } } else { if (instance->netif_state_report & NETIF_STATE_REPORT_DOWN) { log_printf (instance->totemnet_log_level_notice, "The network interface is down.\n"); instance->totemnet_iface_change_fn (instance->context, &instance->my_id); } instance->netif_state_report = NETIF_STATE_REPORT_UP; } } /* * Check if an interface is down and reconfigure * totemnet waiting for it to come back up */ static void netif_down_check (struct totemnet_instance *instance) { timer_function_netif_check_timeout (instance); } /* Set the socket priority to INTERACTIVE to ensure that our messages don't get queued behind anything else */ static void totemnet_traffic_control_set(struct totemnet_instance *instance, int sock) { #ifdef SO_PRIORITY int prio = 6; /* TC_PRIO_INTERACTIVE */ if (setsockopt(sock, SOL_SOCKET, SO_PRIORITY, &prio, sizeof(int))) log_printf (instance->totemnet_log_level_warning, "Could not set traffic priority. (%s)\n", strerror (errno)); #endif } static int totemnet_build_sockets_ip ( struct totemnet_instance *instance, struct totem_ip_address *mcast_address, struct totem_ip_address *bindnet_address, struct totemnet_socket *sockets, struct totem_ip_address *bound_to, int interface_num) { struct sockaddr_storage sockaddr; struct ipv6_mreq mreq6; struct ip_mreq mreq; struct sockaddr_storage mcast_ss, boundto_ss; struct sockaddr_in6 *mcast_sin6 = (struct sockaddr_in6 *)&mcast_ss; struct sockaddr_in *mcast_sin = (struct sockaddr_in *)&mcast_ss; struct sockaddr_in *boundto_sin = (struct sockaddr_in *)&boundto_ss; unsigned int sendbuf_size; unsigned int recvbuf_size; unsigned int optlen = sizeof (sendbuf_size); int addrlen; int res; int flag; /* * Create multicast recv socket */ sockets->mcast_recv = socket (bindnet_address->family, SOCK_DGRAM, 0); if (sockets->mcast_recv == -1) { perror ("socket"); return (-1); } totemip_nosigpipe (sockets->mcast_recv); res = fcntl (sockets->mcast_recv, F_SETFL, O_NONBLOCK); if (res == -1) { log_printf (instance->totemnet_log_level_warning, "Could not set non-blocking operation on multicast socket: %s\n", strerror (errno)); return (-1); } /* * Force reuse */ flag = 1; if ( setsockopt(sockets->mcast_recv, SOL_SOCKET, SO_REUSEADDR, (char *)&flag, sizeof (flag)) < 0) { perror("setsockopt reuseaddr"); return (-1); } /* * Bind to multicast socket used for multicast receives */ totemip_totemip_to_sockaddr_convert(mcast_address, instance->totem_interface->ip_port, &sockaddr, &addrlen); res = bind (sockets->mcast_recv, (struct sockaddr *)&sockaddr, addrlen); if (res == -1) { perror ("bind mcast recv socket failed"); return (-1); } /* * Setup mcast send socket */ sockets->mcast_send = socket (bindnet_address->family, SOCK_DGRAM, 0); if (sockets->mcast_send == -1) { perror ("socket"); return (-1); } totemip_nosigpipe (sockets->mcast_send); res = fcntl (sockets->mcast_send, F_SETFL, O_NONBLOCK); if (res == -1) { log_printf (instance->totemnet_log_level_warning, "Could not set non-blocking operation on multicast socket: %s\n", strerror (errno)); return (-1); } /* * Force reuse */ flag = 1; if ( setsockopt(sockets->mcast_send, SOL_SOCKET, SO_REUSEADDR, (char *)&flag, sizeof (flag)) < 0) { perror("setsockopt reuseaddr"); return (-1); } totemip_totemip_to_sockaddr_convert(bound_to, instance->totem_interface->ip_port - 1, &sockaddr, &addrlen); res = bind (sockets->mcast_send, (struct sockaddr *)&sockaddr, addrlen); if (res == -1) { perror ("bind mcast send socket failed"); return (-1); } /* * Setup unicast socket */ sockets->token = socket (bindnet_address->family, SOCK_DGRAM, 0); if (sockets->token == -1) { perror ("socket2"); return (-1); } totemip_nosigpipe (sockets->token); res = fcntl (sockets->token, F_SETFL, O_NONBLOCK); if (res == -1) { log_printf (instance->totemnet_log_level_warning, "Could not set non-blocking operation on token socket: %s\n", strerror (errno)); return (-1); } /* * Force reuse */ flag = 1; if ( setsockopt(sockets->token, SOL_SOCKET, SO_REUSEADDR, (char *)&flag, sizeof (flag)) < 0) { perror("setsockopt reuseaddr"); return (-1); } /* * Bind to unicast socket used for token send/receives * This has the side effect of binding to the correct interface */ totemip_totemip_to_sockaddr_convert(bound_to, instance->totem_interface->ip_port, &sockaddr, &addrlen); res = bind (sockets->token, (struct sockaddr *)&sockaddr, addrlen); if (res == -1) { perror ("bind token socket failed"); return (-1); } recvbuf_size = MCAST_SOCKET_BUFFER_SIZE; sendbuf_size = MCAST_SOCKET_BUFFER_SIZE; /* * Set buffer sizes to avoid overruns */ res = setsockopt (sockets->mcast_recv, SOL_SOCKET, SO_RCVBUF, &recvbuf_size, optlen); res = setsockopt (sockets->mcast_send, SOL_SOCKET, SO_SNDBUF, &sendbuf_size, optlen); res = getsockopt (sockets->mcast_recv, SOL_SOCKET, SO_RCVBUF, &recvbuf_size, &optlen); if (res == 0) { log_printf (instance->totemnet_log_level_notice, "Receive multicast socket recv buffer size (%d bytes).\n", recvbuf_size); } res = getsockopt (sockets->mcast_send, SOL_SOCKET, SO_SNDBUF, &sendbuf_size, &optlen); if (res == 0) { log_printf (instance->totemnet_log_level_notice, "Transmit multicast socket send buffer size (%d bytes).\n", sendbuf_size); } /* * Join group membership on socket */ totemip_totemip_to_sockaddr_convert(mcast_address, instance->totem_interface->ip_port, &mcast_ss, &addrlen); totemip_totemip_to_sockaddr_convert(bound_to, instance->totem_interface->ip_port, &boundto_ss, &addrlen); switch ( bindnet_address->family ) { case AF_INET: memset(&mreq, 0, sizeof(mreq)); mreq.imr_multiaddr.s_addr = mcast_sin->sin_addr.s_addr; mreq.imr_interface.s_addr = boundto_sin->sin_addr.s_addr; res = setsockopt (sockets->mcast_recv, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof (mreq)); if (res == -1) { perror ("join ipv4 multicast group failed"); return (-1); } break; case AF_INET6: memset(&mreq6, 0, sizeof(mreq6)); memcpy(&mreq6.ipv6mr_multiaddr, &mcast_sin6->sin6_addr, sizeof(struct in6_addr)); mreq6.ipv6mr_interface = interface_num; res = setsockopt (sockets->mcast_recv, IPPROTO_IPV6, IPV6_JOIN_GROUP, &mreq6, sizeof (mreq6)); if (res == -1) { perror ("join ipv6 multicast group failed"); return (-1); } break; } /* * Turn on multicast loopback */ flag = 1; switch ( bindnet_address->family ) { case AF_INET: res = setsockopt (sockets->mcast_send, IPPROTO_IP, IP_MULTICAST_LOOP, &flag, sizeof (flag)); break; case AF_INET6: res = setsockopt (sockets->mcast_send, IPPROTO_IPV6, IPV6_MULTICAST_LOOP, &flag, sizeof (flag)); } if (res == -1) { perror ("turn off loopback"); return (-1); } /* * Set multicast packets TTL */ if ( bindnet_address->family == AF_INET6 ) { flag = 255; res = setsockopt (sockets->mcast_send, IPPROTO_IPV6, IPV6_MULTICAST_HOPS, &flag, sizeof (flag)); if (res == -1) { perror ("setp mcast hops"); return (-1); } } /* * Bind to a specific interface for multicast send and receive */ switch ( bindnet_address->family ) { case AF_INET: if (setsockopt (sockets->mcast_send, IPPROTO_IP, IP_MULTICAST_IF, &boundto_sin->sin_addr, sizeof (boundto_sin->sin_addr)) < 0) { perror ("cannot select interface"); return (-1); } if (setsockopt (sockets->mcast_recv, IPPROTO_IP, IP_MULTICAST_IF, &boundto_sin->sin_addr, sizeof (boundto_sin->sin_addr)) < 0) { perror ("cannot select interface"); return (-1); } break; case AF_INET6: if (setsockopt (sockets->mcast_send, IPPROTO_IPV6, IPV6_MULTICAST_IF, &interface_num, sizeof (interface_num)) < 0) { perror ("cannot select interface"); return (-1); } if (setsockopt (sockets->mcast_recv, IPPROTO_IPV6, IPV6_MULTICAST_IF, &interface_num, sizeof (interface_num)) < 0) { perror ("cannot select interface"); return (-1); } break; } return 0; } static int totemnet_build_sockets ( struct totemnet_instance *instance, struct totem_ip_address *mcast_address, struct totem_ip_address *bindnet_address, struct totemnet_socket *sockets, struct totem_ip_address *bound_to) { int interface_num; int interface_up; int res; /* * Determine the ip address bound to and the interface name */ res = netif_determine (instance, bindnet_address, bound_to, &interface_up, &interface_num); if (res == -1) { return (-1); } totemip_copy(&instance->my_id, bound_to); res = totemnet_build_sockets_ip (instance, mcast_address, bindnet_address, sockets, bound_to, interface_num); /* We only send out of the token socket */ totemnet_traffic_control_set(instance, sockets->token); return res; } /* * Totem Network interface - also does encryption/decryption * depends on poll abstraction, POSIX, IPV4 */ /* * Create an instance */ int totemnet_initialize ( hdb_handle_t poll_handle, hdb_handle_t *handle, struct totem_config *totem_config, int interface_no, void *context, void (*deliver_fn) ( void *context, void *msg, int msg_len), void (*iface_change_fn) ( void *context, struct totem_ip_address *iface_address)) { struct totemnet_instance *instance; unsigned int res; res = hdb_handle_create (&totemnet_instance_database, sizeof (struct totemnet_instance), handle); if (res != 0) { goto error_exit; } res = hdb_handle_get (&totemnet_instance_database, *handle, (void *)&instance); if (res != 0) { goto error_destroy; } totemnet_instance_initialize (instance); instance->totem_config = totem_config; /* * Configure logging */ instance->totemnet_log_level_security = 1; //totem_config->totem_logging_configuration.log_level_security; instance->totemnet_log_level_error = totem_config->totem_logging_configuration.log_level_error; instance->totemnet_log_level_warning = totem_config->totem_logging_configuration.log_level_warning; instance->totemnet_log_level_notice = totem_config->totem_logging_configuration.log_level_notice; instance->totemnet_log_level_debug = totem_config->totem_logging_configuration.log_level_debug; instance->totemnet_subsys_id = totem_config->totem_logging_configuration.log_subsys_id; instance->totemnet_log_printf = totem_config->totem_logging_configuration.log_printf; /* * Initialize random number generator for later use to generate salt */ memcpy (instance->totemnet_private_key, totem_config->private_key, totem_config->private_key_len); instance->totemnet_private_key_len = totem_config->private_key_len; rng_make_prng (128, PRNG_SOBER, &instance->totemnet_prng_state, NULL); /* * Initialize local variables for totemnet */ instance->totem_interface = &totem_config->interfaces[interface_no]; totemip_copy (&instance->mcast_address, &instance->totem_interface->mcast_addr); memset (instance->iov_buffer, 0, FRAME_SIZE_MAX); /* * If threaded send requested, initialize thread group data structure */ if (totem_config->threads) { worker_thread_group_init ( &instance->worker_thread_group, totem_config->threads, 128, sizeof (struct work_item), sizeof (struct totemnet_mcast_thread_state), totemnet_mcast_thread_state_constructor, totemnet_mcast_worker_fn); } instance->totemnet_poll_handle = poll_handle; instance->totem_interface->bindnet.nodeid = instance->totem_config->node_id; instance->context = context; instance->totemnet_deliver_fn = deliver_fn; instance->totemnet_iface_change_fn = iface_change_fn; instance->handle = *handle; rng_make_prng (128, PRNG_SOBER, &instance->totemnet_prng_state, NULL); totemip_localhost (instance->mcast_address.family, &localhost); netif_down_check (instance); error_exit: hdb_handle_put (&totemnet_instance_database, *handle); return (0); error_destroy: hdb_handle_destroy (&totemnet_instance_database, *handle); return (-1); } int totemnet_processor_count_set ( hdb_handle_t handle, int processor_count) { struct totemnet_instance *instance; int res = 0; res = hdb_handle_get (&totemnet_instance_database, handle, (void *)&instance); if (res != 0) { res = ENOENT; goto error_exit; } instance->my_memb_entries = processor_count; poll_timer_delete (instance->totemnet_poll_handle, instance->timer_netif_check_timeout); if (processor_count == 1) { poll_timer_add (instance->totemnet_poll_handle, instance->totem_config->downcheck_timeout, (void *)instance, timer_function_netif_check_timeout, &instance->timer_netif_check_timeout); } hdb_handle_put (&totemnet_instance_database, handle); error_exit: return (res); } int totemnet_recv_flush (hdb_handle_t handle) { struct totemnet_instance *instance; struct pollfd ufd; int nfds; int res = 0; res = hdb_handle_get (&totemnet_instance_database, handle, (void *)&instance); if (res != 0) { res = ENOENT; goto error_exit; } instance->flushing = 1; do { ufd.fd = instance->totemnet_sockets.mcast_recv; ufd.events = POLLIN; nfds = poll (&ufd, 1, 0); if (nfds == 1 && ufd.revents & POLLIN) { net_deliver_fn (0, instance->totemnet_sockets.mcast_recv, ufd.revents, instance); } } while (nfds == 1); instance->flushing = 0; hdb_handle_put (&totemnet_instance_database, handle); error_exit: return (res); } int totemnet_send_flush (hdb_handle_t handle) { struct totemnet_instance *instance; int res = 0; res = hdb_handle_get (&totemnet_instance_database, handle, (void *)&instance); if (res != 0) { res = ENOENT; goto error_exit; } worker_thread_group_wait (&instance->worker_thread_group); hdb_handle_put (&totemnet_instance_database, handle); error_exit: return (res); } int totemnet_token_send ( hdb_handle_t handle, struct iovec *iovec, unsigned int iov_len) { struct totemnet_instance *instance; int res = 0; res = hdb_handle_get (&totemnet_instance_database, handle, (void *)&instance); if (res != 0) { res = ENOENT; goto error_exit; } ucast_sendmsg (instance, &instance->token_target, iovec, iov_len); hdb_handle_put (&totemnet_instance_database, handle); error_exit: return (res); } int totemnet_mcast_flush_send ( hdb_handle_t handle, struct iovec *iovec, unsigned int iov_len) { struct totemnet_instance *instance; int res = 0; res = hdb_handle_get (&totemnet_instance_database, handle, (void *)&instance); if (res != 0) { res = ENOENT; goto error_exit; } mcast_sendmsg (instance, iovec, iov_len); hdb_handle_put (&totemnet_instance_database, handle); error_exit: return (res); } int totemnet_mcast_noflush_send ( hdb_handle_t handle, struct iovec *iovec, unsigned int iov_len) { struct totemnet_instance *instance; struct work_item work_item; int res = 0; res = hdb_handle_get (&totemnet_instance_database, handle, (void *)&instance); if (res != 0) { res = ENOENT; goto error_exit; } if (instance->totem_config->threads) { memcpy (&work_item.iovec[0], iovec, iov_len * sizeof (struct iovec)); work_item.iov_len = iov_len; work_item.instance = instance; worker_thread_group_work_add (&instance->worker_thread_group, &work_item); } else { mcast_sendmsg (instance, iovec, iov_len); } hdb_handle_put (&totemnet_instance_database, handle); error_exit: return (res); } extern int totemnet_iface_check (hdb_handle_t handle) { struct totemnet_instance *instance; int res = 0; res = hdb_handle_get (&totemnet_instance_database, handle, (void *)&instance); if (res != 0) { res = ENOENT; goto error_exit; } timer_function_netif_check_timeout (instance); hdb_handle_put (&totemnet_instance_database, handle); error_exit: return (res); } extern void totemnet_net_mtu_adjust (struct totem_config *totem_config) { #define UDPIP_HEADER_SIZE (20 + 8) /* 20 bytes for ip 8 bytes for udp */ if (totem_config->secauth == 1) { totem_config->net_mtu -= sizeof (struct security_header) + UDPIP_HEADER_SIZE; } else { totem_config->net_mtu -= UDPIP_HEADER_SIZE; } } const char *totemnet_iface_print (hdb_handle_t handle) { struct totemnet_instance *instance; int res = 0; const char *ret_char; res = hdb_handle_get (&totemnet_instance_database, handle, (void *)&instance); if (res != 0) { ret_char = "Invalid totemnet handle"; goto error_exit; } ret_char = totemip_print (&instance->my_id); hdb_handle_put (&totemnet_instance_database, handle); error_exit: return (ret_char); } int totemnet_iface_get ( hdb_handle_t handle, struct totem_ip_address *addr) { struct totemnet_instance *instance; unsigned int res; res = hdb_handle_get (&totemnet_instance_database, handle, (void *)&instance); if (res != 0) { goto error_exit; } memcpy (addr, &instance->my_id, sizeof (struct totem_ip_address)); hdb_handle_put (&totemnet_instance_database, handle); error_exit: return (res); } int totemnet_token_target_set ( hdb_handle_t handle, struct totem_ip_address *token_target) { struct totemnet_instance *instance; unsigned int res; res = hdb_handle_get (&totemnet_instance_database, handle, (void *)&instance); if (res != 0) { goto error_exit; } memcpy (&instance->token_target, token_target, sizeof (struct totem_ip_address)); hdb_handle_put (&totemnet_instance_database, handle); error_exit: return (res); } diff --git a/exec/totempg.c b/exec/totempg.c index f723decf..8ba64cef 100644 --- a/exec/totempg.c +++ b/exec/totempg.c @@ -1,1338 +1,1333 @@ /* * Copyright (c) 2003-2005 MontaVista Software, Inc. * Copyright (c) 2005 OSDL. * Copyright (c) 2006-2009 Red Hat, Inc. * * All rights reserved. * * Author: Steven Dake (sdake@redhat.com) * Author: Mark Haverkamp (markh@osdl.org) * * 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 MontaVista Software, Inc. 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. */ /* * FRAGMENTATION AND PACKING ALGORITHM: * * Assemble the entire message into one buffer * if full fragment * store fragment into lengths list * for each full fragment * multicast fragment * set length and fragment fields of pg mesage * store remaining multicast into head of fragmentation data and set lens field * * If a message exceeds the maximum packet size allowed by the totem * single ring protocol, the protocol could lose forward progress. * Statically calculating the allowed data amount doesn't work because * the amount of data allowed depends on the number of fragments in * each message. In this implementation, the maximum fragment size * is dynamically calculated for each fragment added to the message. * It is possible for a message to be two bytes short of the maximum * packet size. This occurs when a message or collection of * messages + the mcast header + the lens are two bytes short of the * end of the packet. Since another len field consumes two bytes, the * len field would consume the rest of the packet without room for data. * * One optimization would be to forgo the final len field and determine * it from the size of the udp datagram. Then this condition would no * longer occur. */ /* * ASSEMBLY AND UNPACKING ALGORITHM: * * copy incoming packet into assembly data buffer indexed by current * location of end of fragment * * if not fragmented * deliver all messages in assembly data buffer * else * if msg_count > 1 and fragmented * deliver all messages except last message in assembly data buffer * copy last fragmented section to start of assembly data buffer * else * if msg_count = 1 and fragmented * do nothing * */ #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include "totemmrp.h" #include "totemsrp.h" #define min(a,b) ((a) < (b)) ? a : b struct totempg_mcast_header { short version; short type; }; /* * totempg_mcast structure * * header: Identify the mcast. * fragmented: Set if this message continues into next message * continuation: Set if this message is a continuation from last message * msg_count Indicates how many packed messages are contained * in the mcast. * Also, the size of each packed message and the messages themselves are * appended to the end of this structure when sent. */ struct totempg_mcast { struct totempg_mcast_header header; unsigned char fragmented; unsigned char continuation; unsigned short msg_count; /* * short msg_len[msg_count]; */ /* * data for messages */ }; /* * Maximum packet size for totem pg messages */ #define TOTEMPG_PACKET_SIZE (totempg_totem_config->net_mtu - \ sizeof (struct totempg_mcast)) /* * Local variables used for packing small messages */ static unsigned short mcast_packed_msg_lens[FRAME_SIZE_MAX]; static int mcast_packed_msg_count = 0; static int totempg_reserved = 0; /* * Function and data used to log messages */ static int totempg_log_level_security; static int totempg_log_level_error; static int totempg_log_level_warning; static int totempg_log_level_notice; static int totempg_log_level_debug; static int totempg_subsys_id; static void (*totempg_log_printf) (int subsys_id, const char *function, const char *file, int line, unsigned int level, const char *format, ...) __attribute__((format(printf, 6, 7))); struct totem_config *totempg_totem_config; struct assembly { unsigned int nodeid; unsigned char data[MESSAGE_SIZE_MAX]; int index; unsigned char last_frag_num; struct list_head list; }; static void assembly_deref (struct assembly *assembly); static int callback_token_received_fn (enum totem_callback_token_type type, const void *data); enum throw_away_mode_t { THROW_AWAY_INACTIVE, THROW_AWAY_ACTIVE }; static enum throw_away_mode_t throw_away_mode = THROW_AWAY_INACTIVE; DECLARE_LIST_INIT(assembly_list_inuse); DECLARE_LIST_INIT(assembly_list_free); /* * Staging buffer for packed messages. Messages are staged in this buffer * before sending. Multiple messages may fit which cuts down on the * number of mcasts sent. If a message doesn't completely fit, then * the mcast header has a fragment bit set that says that there are more * data to follow. fragment_size is an index into the buffer. It indicates * the size of message data and where to place new message data. * fragment_contuation indicates whether the first packed message in * the buffer is a continuation of a previously packed fragment. */ static unsigned char *fragmentation_data; static int fragment_size = 0; static int fragment_continuation = 0; static struct iovec iov_delv; static unsigned int totempg_max_handle = 0; struct totempg_group_instance { void (*deliver_fn) ( unsigned int nodeid, struct iovec *iovec, unsigned int iov_len, int endian_conversion_required); void (*confchg_fn) ( enum totem_configuration_type configuration_type, const unsigned int *member_list, size_t member_list_entries, const unsigned int *left_list, size_t left_list_entries, const unsigned int *joined_list, size_t joined_list_entries, const struct memb_ring_id *ring_id); struct totempg_group *groups; int groups_cnt; }; -static struct hdb_handle_database totempg_groups_instance_database = { - .handle_count = 0, - .handles = 0, - .iterator = 0, - .mutex = PTHREAD_MUTEX_INITIALIZER -}; +DECLARE_HDB_DATABASE (totempg_groups_instance_database); static unsigned char next_fragment = 1; static pthread_mutex_t totempg_mutex = PTHREAD_MUTEX_INITIALIZER; static pthread_mutex_t callback_token_mutex = PTHREAD_MUTEX_INITIALIZER; static pthread_mutex_t mcast_msg_mutex = PTHREAD_MUTEX_INITIALIZER; #define log_printf(level, format, args...) \ do { \ totempg_log_printf (totempg_subsys_id, __FUNCTION__, \ __FILE__, __LINE__, level, format, ##args); \ } while (0); static int msg_count_send_ok (int msg_count); static int byte_count_send_ok (int byte_count); static struct assembly *assembly_ref (unsigned int nodeid) { struct assembly *assembly; struct list_head *list; /* * Search inuse list for node id and return assembly buffer if found */ for (list = assembly_list_inuse.next; list != &assembly_list_inuse; list = list->next) { assembly = list_entry (list, struct assembly, list); if (nodeid == assembly->nodeid) { return (assembly); } } /* * Nothing found in inuse list get one from free list if available */ if (list_empty (&assembly_list_free) == 0) { assembly = list_entry (assembly_list_free.next, struct assembly, list); list_del (&assembly->list); list_add (&assembly->list, &assembly_list_inuse); assembly->nodeid = nodeid; return (assembly); } /* * Nothing available in inuse or free list, so allocate a new one */ assembly = malloc (sizeof (struct assembly)); memset (assembly, 0, sizeof (struct assembly)); /* * TODO handle memory allocation failure here */ assert (assembly); assembly->nodeid = nodeid; list_init (&assembly->list); list_add (&assembly->list, &assembly_list_inuse); return (assembly); } static void assembly_deref (struct assembly *assembly) { list_del (&assembly->list); list_add (&assembly->list, &assembly_list_free); } static inline void app_confchg_fn ( enum totem_configuration_type configuration_type, const unsigned int *member_list, size_t member_list_entries, const unsigned int *left_list, size_t left_list_entries, const unsigned int *joined_list, size_t joined_list_entries, const struct memb_ring_id *ring_id) { int i; struct totempg_group_instance *instance; unsigned int res; for (i = 0; i <= totempg_max_handle; i++) { res = hdb_handle_get (&totempg_groups_instance_database, hdb_nocheck_convert (i), (void *)&instance); if (res == 0) { if (instance->confchg_fn) { instance->confchg_fn ( configuration_type, member_list, member_list_entries, left_list, left_list_entries, joined_list, joined_list_entries, ring_id); } hdb_handle_put (&totempg_groups_instance_database, hdb_nocheck_convert (i)); } } } static inline void group_endian_convert ( struct iovec *iovec) { unsigned short *group_len; int i; struct iovec iovec_aligned = { NULL, 0 }; struct iovec *iovec_swab; /* * Align data structure for sparc and ia64 */ if ((size_t)iovec->iov_base % 4 != 0) { iovec_aligned.iov_base = alloca(iovec->iov_len); memcpy(iovec_aligned.iov_base, iovec->iov_base, iovec->iov_len); iovec_aligned.iov_len = iovec->iov_len; iovec_swab = &iovec_aligned; } else { iovec_swab = iovec; } group_len = (unsigned short *)iovec_swab->iov_base; group_len[0] = swab16(group_len[0]); for (i = 1; i < group_len[0] + 1; i++) { group_len[i] = swab16(group_len[i]); } if (iovec_swab == &iovec_aligned) { memcpy(iovec->iov_base, iovec_aligned.iov_base, iovec->iov_len); } } static inline int group_matches ( struct iovec *iovec, unsigned int iov_len, struct totempg_group *groups_b, unsigned int group_b_cnt, unsigned int *adjust_iovec) { unsigned short *group_len; char *group_name; int i; int j; struct iovec iovec_aligned = { NULL, 0 }; assert (iov_len == 1); /* * Align data structure for sparc and ia64 */ if ((size_t)iovec->iov_base % 4 != 0) { iovec_aligned.iov_base = alloca(iovec->iov_len); memcpy(iovec_aligned.iov_base, iovec->iov_base, iovec->iov_len); iovec_aligned.iov_len = iovec->iov_len; iovec = &iovec_aligned; } group_len = (unsigned short *)iovec->iov_base; group_name = ((char *)iovec->iov_base) + sizeof (unsigned short) * (group_len[0] + 1); /* * Calculate amount to adjust the iovec by before delivering to app */ *adjust_iovec = sizeof (unsigned short) * (group_len[0] + 1); for (i = 1; i < group_len[0] + 1; i++) { *adjust_iovec += group_len[i]; } /* * Determine if this message should be delivered to this instance */ for (i = 1; i < group_len[0] + 1; i++) { for (j = 0; j < group_b_cnt; j++) { if ((group_len[i] == groups_b[j].group_len) && (memcmp (groups_b[j].group, group_name, group_len[i]) == 0)) { return (1); } } group_name += group_len[i]; } return (0); } static inline void app_deliver_fn ( unsigned int nodeid, struct iovec *iovec, unsigned int iov_len, int endian_conversion_required) { int i; struct totempg_group_instance *instance; struct iovec stripped_iovec; unsigned int adjust_iovec; unsigned int res; struct iovec aligned_iovec = { NULL, 0 }; if (endian_conversion_required) { group_endian_convert (iovec); } /* * Align data structure for sparc and ia64 */ aligned_iovec.iov_base = alloca(iovec->iov_len); aligned_iovec.iov_len = iovec->iov_len; memcpy(aligned_iovec.iov_base, iovec->iov_base, iovec->iov_len); iovec = &aligned_iovec; for (i = 0; i <= totempg_max_handle; i++) { res = hdb_handle_get (&totempg_groups_instance_database, hdb_nocheck_convert (i), (void *)&instance); if (res == 0) { assert (iov_len == 1); if (group_matches (iovec, iov_len, instance->groups, instance->groups_cnt, &adjust_iovec)) { stripped_iovec.iov_len = iovec->iov_len - adjust_iovec; // stripped_iovec.iov_base = (char *)iovec->iov_base + adjust_iovec; /* * Align data structure for sparc and ia64 */ if ((char *)iovec->iov_base + adjust_iovec % 4 != 0) { /* * Deal with misalignment */ stripped_iovec.iov_base = alloca (stripped_iovec.iov_len); memcpy (stripped_iovec.iov_base, (char *)iovec->iov_base + adjust_iovec, stripped_iovec.iov_len); } instance->deliver_fn ( nodeid, &stripped_iovec, iov_len, endian_conversion_required); } hdb_handle_put (&totempg_groups_instance_database, hdb_nocheck_convert(i)); } } } static void totempg_confchg_fn ( enum totem_configuration_type configuration_type, const unsigned int *member_list, size_t member_list_entries, const unsigned int *left_list, size_t left_list_entries, const unsigned int *joined_list, size_t joined_list_entries, const struct memb_ring_id *ring_id) { // TODO optimize this app_confchg_fn (configuration_type, member_list, member_list_entries, left_list, left_list_entries, joined_list, joined_list_entries, ring_id); } static void totempg_deliver_fn ( unsigned int nodeid, struct iovec *iovec, unsigned int iov_len, int endian_conversion_required) { struct totempg_mcast *mcast; unsigned short *msg_lens; int i; struct assembly *assembly; char header[FRAME_SIZE_MAX]; int h_index; int a_i = 0; int msg_count; int continuation; int start; assembly = assembly_ref (nodeid); assert (assembly); /* * Assemble the header into one block of data and * assemble the packet contents into one block of data to simplify delivery */ if (iov_len == 1) { /* * This message originated from external processor * because there is only one iovec for the full msg. */ char *data; int datasize; mcast = (struct totempg_mcast *)iovec[0].iov_base; if (endian_conversion_required) { mcast->msg_count = swab16 (mcast->msg_count); } msg_count = mcast->msg_count; datasize = sizeof (struct totempg_mcast) + msg_count * sizeof (unsigned short); memcpy (header, iovec[0].iov_base, datasize); assert(iovec); data = iovec[0].iov_base; msg_lens = (unsigned short *) (header + sizeof (struct totempg_mcast)); if (endian_conversion_required) { for (i = 0; i < mcast->msg_count; i++) { msg_lens[i] = swab16 (msg_lens[i]); } } memcpy (&assembly->data[assembly->index], &data[datasize], iovec[0].iov_len - datasize); } else { /* * The message originated from local processor * becasue there is greater than one iovec for then full msg. */ h_index = 0; for (i = 0; i < 2; i++) { memcpy (&header[h_index], iovec[i].iov_base, iovec[i].iov_len); h_index += iovec[i].iov_len; } mcast = (struct totempg_mcast *)header; // TODO make sure we are using a copy of mcast not the actual data itself msg_lens = (unsigned short *) (header + sizeof (struct totempg_mcast)); for (i = 2; i < iov_len; i++) { a_i = assembly->index; assert (iovec[i].iov_len + a_i <= MESSAGE_SIZE_MAX); memcpy (&assembly->data[a_i], iovec[i].iov_base, iovec[i].iov_len); a_i += msg_lens[i - 2]; } iov_len -= 2; } /* * If the last message in the buffer is a fragment, then we * can't deliver it. We'll first deliver the full messages * then adjust the assembly buffer so we can add the rest of the * fragment when it arrives. */ msg_count = mcast->fragmented ? mcast->msg_count - 1 : mcast->msg_count; continuation = mcast->continuation; iov_delv.iov_base = &assembly->data[0]; iov_delv.iov_len = assembly->index + msg_lens[0]; /* * Make sure that if this message is a continuation, that it * matches the sequence number of the previous fragment. * Also, if the first packed message is a continuation * of a previous message, but the assembly buffer * is empty, then we need to discard it since we can't * assemble a complete message. Likewise, if this message isn't a * continuation and the assembly buffer is empty, we have to discard * the continued message. */ start = 0; if (throw_away_mode == THROW_AWAY_ACTIVE) { /* Throw away the first msg block */ if (mcast->fragmented == 0 || mcast->fragmented == 1) { throw_away_mode = THROW_AWAY_INACTIVE; assembly->index += msg_lens[0]; iov_delv.iov_base = &assembly->data[assembly->index]; iov_delv.iov_len = msg_lens[1]; start = 1; } } else if (throw_away_mode == THROW_AWAY_INACTIVE) { if (continuation == assembly->last_frag_num) { assembly->last_frag_num = mcast->fragmented; for (i = start; i < msg_count; i++) { app_deliver_fn(nodeid, &iov_delv, 1, endian_conversion_required); assembly->index += msg_lens[i]; iov_delv.iov_base = &assembly->data[assembly->index]; if (i < (msg_count - 1)) { iov_delv.iov_len = msg_lens[i + 1]; } } } else { throw_away_mode = THROW_AWAY_ACTIVE; } } if (mcast->fragmented == 0) { /* * End of messages, dereference assembly struct */ assembly->last_frag_num = 0; assembly->index = 0; assembly_deref (assembly); } else { /* * Message is fragmented, keep around assembly list */ if (mcast->msg_count > 1) { memmove (&assembly->data[0], &assembly->data[assembly->index], msg_lens[msg_count]); assembly->index = 0; } assembly->index += msg_lens[msg_count]; } } /* * Totem Process Group Abstraction * depends on poll abstraction, POSIX, IPV4 */ void *callback_token_received_handle; int callback_token_received_fn (enum totem_callback_token_type type, const void *data) { struct totempg_mcast mcast; struct iovec iovecs[3]; int res; pthread_mutex_lock (&mcast_msg_mutex); if (mcast_packed_msg_count == 0) { pthread_mutex_unlock (&mcast_msg_mutex); return (0); } if (totemmrp_avail() == 0) { pthread_mutex_unlock (&mcast_msg_mutex); return (0); } mcast.fragmented = 0; /* * Was the first message in this buffer a continuation of a * fragmented message? */ mcast.continuation = fragment_continuation; fragment_continuation = 0; mcast.msg_count = mcast_packed_msg_count; iovecs[0].iov_base = &mcast; iovecs[0].iov_len = sizeof (struct totempg_mcast); iovecs[1].iov_base = mcast_packed_msg_lens; iovecs[1].iov_len = mcast_packed_msg_count * sizeof (unsigned short); iovecs[2].iov_base = &fragmentation_data[0]; iovecs[2].iov_len = fragment_size; res = totemmrp_mcast (iovecs, 3, 0); mcast_packed_msg_count = 0; fragment_size = 0; pthread_mutex_unlock (&mcast_msg_mutex); return (0); } /* * Initialize the totem process group abstraction */ int totempg_initialize ( hdb_handle_t poll_handle, struct totem_config *totem_config) { int res; totempg_totem_config = totem_config; totempg_log_level_security = totem_config->totem_logging_configuration.log_level_security; totempg_log_level_error = totem_config->totem_logging_configuration.log_level_error; totempg_log_level_warning = totem_config->totem_logging_configuration.log_level_warning; totempg_log_level_notice = totem_config->totem_logging_configuration.log_level_notice; totempg_log_level_debug = totem_config->totem_logging_configuration.log_level_debug; totempg_log_printf = totem_config->totem_logging_configuration.log_printf; totempg_subsys_id = totem_config->totem_logging_configuration.log_subsys_id; fragmentation_data = malloc (TOTEMPG_PACKET_SIZE); if (fragmentation_data == 0) { return (-1); } res = totemmrp_initialize ( poll_handle, totem_config, totempg_deliver_fn, totempg_confchg_fn); totemmrp_callback_token_create ( &callback_token_received_handle, TOTEM_CALLBACK_TOKEN_RECEIVED, 0, callback_token_received_fn, 0); totemsrp_net_mtu_adjust (totem_config); return (res); } void totempg_finalize (void) { pthread_mutex_lock (&totempg_mutex); totemmrp_finalize (); pthread_mutex_unlock (&totempg_mutex); } /* * Multicast a message */ static int mcast_msg ( struct iovec *iovec_in, unsigned int iov_len, int guarantee) { int res = 0; struct totempg_mcast mcast; struct iovec iovecs[3]; struct iovec iovec[64]; int i; int dest, src; int max_packet_size = 0; int copy_len = 0; int copy_base = 0; int total_size = 0; pthread_mutex_lock (&mcast_msg_mutex); totemmrp_new_msg_signal (); /* * Remove zero length iovectors from the list */ assert (iov_len < 64); for (dest = 0, src = 0; src < iov_len; src++) { if (iovec_in[src].iov_len) { memcpy (&iovec[dest++], &iovec_in[src], sizeof (struct iovec)); } } iov_len = dest; max_packet_size = TOTEMPG_PACKET_SIZE - (sizeof (unsigned short) * (mcast_packed_msg_count + 1)); mcast_packed_msg_lens[mcast_packed_msg_count] = 0; /* * Check if we would overwrite new message queue */ for (i = 0; i < iov_len; i++) { total_size += iovec[i].iov_len; } if (byte_count_send_ok (total_size + sizeof(unsigned short) * (mcast_packed_msg_count+1)) == 0) { pthread_mutex_unlock (&mcast_msg_mutex); return(-1); } for (i = 0; i < iov_len; ) { mcast.fragmented = 0; mcast.continuation = fragment_continuation; copy_len = iovec[i].iov_len - copy_base; /* * If it all fits with room left over, copy it in. * We need to leave at least sizeof(short) + 1 bytes in the * fragment_buffer on exit so that max_packet_size + fragment_size * doesn't exceed the size of the fragment_buffer on the next call. */ if ((copy_len + fragment_size) < (max_packet_size - sizeof (unsigned short))) { memcpy (&fragmentation_data[fragment_size], (char *)iovec[i].iov_base + copy_base, copy_len); fragment_size += copy_len; mcast_packed_msg_lens[mcast_packed_msg_count] += copy_len; next_fragment = 1; copy_len = 0; copy_base = 0; i++; continue; /* * If it just fits or is too big, then send out what fits. */ } else { unsigned char *data_ptr; copy_len = min(copy_len, max_packet_size - fragment_size); if( copy_len == max_packet_size ) data_ptr = (unsigned char *)iovec[i].iov_base + copy_base; else { data_ptr = fragmentation_data; memcpy (&fragmentation_data[fragment_size], (unsigned char *)iovec[i].iov_base + copy_base, copy_len); } memcpy (&fragmentation_data[fragment_size], (unsigned char *)iovec[i].iov_base + copy_base, copy_len); mcast_packed_msg_lens[mcast_packed_msg_count] += copy_len; /* * if we're not on the last iovec or the iovec is too large to * fit, then indicate a fragment. This also means that the next * message will have the continuation of this one. */ if ((i < (iov_len - 1)) || ((copy_base + copy_len) < iovec[i].iov_len)) { if (!next_fragment) { next_fragment++; } fragment_continuation = next_fragment; mcast.fragmented = next_fragment++; assert(fragment_continuation != 0); assert(mcast.fragmented != 0); } else { fragment_continuation = 0; } /* * assemble the message and send it */ mcast.msg_count = ++mcast_packed_msg_count; iovecs[0].iov_base = &mcast; iovecs[0].iov_len = sizeof(struct totempg_mcast); iovecs[1].iov_base = mcast_packed_msg_lens; iovecs[1].iov_len = mcast_packed_msg_count * sizeof(unsigned short); iovecs[2].iov_base = data_ptr; iovecs[2].iov_len = max_packet_size; assert (totemmrp_avail() > 0); res = totemmrp_mcast (iovecs, 3, guarantee); /* * Recalculate counts and indexes for the next. */ mcast_packed_msg_lens[0] = 0; mcast_packed_msg_count = 0; fragment_size = 0; max_packet_size = TOTEMPG_PACKET_SIZE - (sizeof(unsigned short)); /* * If the iovec all fit, go to the next iovec */ if ((copy_base + copy_len) == iovec[i].iov_len) { copy_len = 0; copy_base = 0; i++; /* * Continue with the rest of the current iovec. */ } else { copy_base += copy_len; } } } /* * Bump only if we added message data. This may be zero if * the last buffer just fit into the fragmentation_data buffer * and we were at the last iovec. */ if (mcast_packed_msg_lens[mcast_packed_msg_count]) { mcast_packed_msg_count++; } pthread_mutex_unlock (&mcast_msg_mutex); return (res); } /* * Determine if a message of msg_size could be queued */ static int msg_count_send_ok ( int msg_count) { int avail = 0; avail = totemmrp_avail () - totempg_reserved - 1; return (avail > msg_count); } static int byte_count_send_ok ( int byte_count) { unsigned int msg_count = 0; int avail = 0; avail = totemmrp_avail () - 1; msg_count = (byte_count / (totempg_totem_config->net_mtu - 25)) + 1; return (avail > msg_count); } static int send_reserve ( int msg_size) { unsigned int msg_count = 0; msg_count = (msg_size / (totempg_totem_config->net_mtu - 25)) + 1; totempg_reserved += msg_count; return (msg_count); } static void send_release ( int msg_count) { totempg_reserved -= msg_count; } int totempg_callback_token_create ( void **handle_out, enum totem_callback_token_type type, int delete, int (*callback_fn) (enum totem_callback_token_type type, const void *), const void *data) { unsigned int res; pthread_mutex_lock (&callback_token_mutex); res = totemmrp_callback_token_create (handle_out, type, delete, callback_fn, data); pthread_mutex_unlock (&callback_token_mutex); return (res); } void totempg_callback_token_destroy ( void *handle_out) { pthread_mutex_lock (&callback_token_mutex); totemmrp_callback_token_destroy (handle_out); pthread_mutex_unlock (&callback_token_mutex); } /* * vi: set autoindent tabstop=4 shiftwidth=4 : */ int totempg_groups_initialize ( hdb_handle_t *handle, void (*deliver_fn) ( unsigned int nodeid, struct iovec *iovec, unsigned int iov_len, int endian_conversion_required), void (*confchg_fn) ( enum totem_configuration_type configuration_type, const unsigned int *member_list, size_t member_list_entries, const unsigned int *left_list, size_t left_list_entries, const unsigned int *joined_list, size_t joined_list_entries, const struct memb_ring_id *ring_id)) { struct totempg_group_instance *instance; unsigned int res; pthread_mutex_lock (&totempg_mutex); res = hdb_handle_create (&totempg_groups_instance_database, sizeof (struct totempg_group_instance), handle); if (res != 0) { goto error_exit; } if (*handle > totempg_max_handle) { totempg_max_handle = *handle; } res = hdb_handle_get (&totempg_groups_instance_database, *handle, (void *)&instance); if (res != 0) { goto error_destroy; } instance->deliver_fn = deliver_fn; instance->confchg_fn = confchg_fn; instance->groups = 0; instance->groups_cnt = 0; hdb_handle_put (&totempg_groups_instance_database, *handle); pthread_mutex_unlock (&totempg_mutex); return (0); error_destroy: hdb_handle_destroy (&totempg_groups_instance_database, *handle); error_exit: pthread_mutex_unlock (&totempg_mutex); return (-1); } int totempg_groups_join ( hdb_handle_t handle, const struct totempg_group *groups, size_t group_cnt) { struct totempg_group_instance *instance; struct totempg_group *new_groups; unsigned int res; pthread_mutex_lock (&totempg_mutex); res = hdb_handle_get (&totempg_groups_instance_database, handle, (void *)&instance); if (res != 0) { goto error_exit; } new_groups = realloc (instance->groups, sizeof (struct totempg_group) * (instance->groups_cnt + group_cnt)); if (new_groups == 0) { res = ENOMEM; goto error_exit; } memcpy (&new_groups[instance->groups_cnt], groups, group_cnt * sizeof (struct totempg_group)); instance->groups = new_groups; instance->groups_cnt = instance->groups_cnt = group_cnt; hdb_handle_put (&totempg_groups_instance_database, handle); error_exit: pthread_mutex_unlock (&totempg_mutex); return (res); } int totempg_groups_leave ( hdb_handle_t handle, const struct totempg_group *groups, size_t group_cnt) { struct totempg_group_instance *instance; unsigned int res; pthread_mutex_lock (&totempg_mutex); res = hdb_handle_get (&totempg_groups_instance_database, handle, (void *)&instance); if (res != 0) { goto error_exit; } hdb_handle_put (&totempg_groups_instance_database, handle); error_exit: pthread_mutex_unlock (&totempg_mutex); return (res); } #define MAX_IOVECS_FROM_APP 32 #define MAX_GROUPS_PER_MSG 32 int totempg_groups_mcast_joined ( hdb_handle_t handle, const struct iovec *iovec, unsigned int iov_len, int guarantee) { struct totempg_group_instance *instance; unsigned short group_len[MAX_GROUPS_PER_MSG + 1]; struct iovec iovec_mcast[MAX_GROUPS_PER_MSG + 1 + MAX_IOVECS_FROM_APP]; int i; unsigned int res; pthread_mutex_lock (&totempg_mutex); res = hdb_handle_get (&totempg_groups_instance_database, handle, (void *)&instance); if (res != 0) { goto error_exit; } /* * Build group_len structure and the iovec_mcast structure */ group_len[0] = instance->groups_cnt; for (i = 0; i < instance->groups_cnt; i++) { group_len[i + 1] = instance->groups[i].group_len; iovec_mcast[i + 1].iov_len = instance->groups[i].group_len; iovec_mcast[i + 1].iov_base = (void *) instance->groups[i].group; } iovec_mcast[0].iov_len = (instance->groups_cnt + 1) * sizeof (unsigned short); iovec_mcast[0].iov_base = group_len; for (i = 0; i < iov_len; i++) { iovec_mcast[i + instance->groups_cnt + 1].iov_len = iovec[i].iov_len; iovec_mcast[i + instance->groups_cnt + 1].iov_base = iovec[i].iov_base; } res = mcast_msg (iovec_mcast, iov_len + instance->groups_cnt + 1, guarantee); hdb_handle_put (&totempg_groups_instance_database, handle); error_exit: pthread_mutex_unlock (&totempg_mutex); return (res); } int totempg_groups_joined_reserve ( hdb_handle_t handle, const struct iovec *iovec, unsigned int iov_len) { struct totempg_group_instance *instance; unsigned int size = 0; unsigned int i; unsigned int res; unsigned int reserved = 0; pthread_mutex_lock (&totempg_mutex); pthread_mutex_lock (&mcast_msg_mutex); res = hdb_handle_get (&totempg_groups_instance_database, handle, (void *)&instance); if (res != 0) { goto error_exit; } for (i = 0; i < instance->groups_cnt; i++) { size += instance->groups[i].group_len; } for (i = 0; i < iov_len; i++) { size += iovec[i].iov_len; } reserved = send_reserve (size); if (msg_count_send_ok (reserved) == 0) { send_release (reserved); reserved = 0; } hdb_handle_put (&totempg_groups_instance_database, handle); error_exit: pthread_mutex_unlock (&mcast_msg_mutex); pthread_mutex_unlock (&totempg_mutex); return (reserved); } int totempg_groups_joined_release (int msg_count) { pthread_mutex_lock (&totempg_mutex); pthread_mutex_lock (&mcast_msg_mutex); send_release (msg_count); pthread_mutex_unlock (&mcast_msg_mutex); pthread_mutex_unlock (&totempg_mutex); return 0; } int totempg_groups_mcast_groups ( hdb_handle_t handle, int guarantee, const struct totempg_group *groups, size_t groups_cnt, const struct iovec *iovec, unsigned int iov_len) { struct totempg_group_instance *instance; unsigned short group_len[MAX_GROUPS_PER_MSG + 1]; struct iovec iovec_mcast[MAX_GROUPS_PER_MSG + 1 + MAX_IOVECS_FROM_APP]; int i; unsigned int res; pthread_mutex_lock (&totempg_mutex); res = hdb_handle_get (&totempg_groups_instance_database, handle, (void *)&instance); if (res != 0) { goto error_exit; } /* * Build group_len structure and the iovec_mcast structure */ group_len[0] = groups_cnt; for (i = 0; i < groups_cnt; i++) { group_len[i + 1] = groups[i].group_len; iovec_mcast[i + 1].iov_len = groups[i].group_len; iovec_mcast[i + 1].iov_base = (void *) groups[i].group; } iovec_mcast[0].iov_len = (groups_cnt + 1) * sizeof (unsigned short); iovec_mcast[0].iov_base = group_len; for (i = 0; i < iov_len; i++) { iovec_mcast[i + groups_cnt + 1].iov_len = iovec[i].iov_len; iovec_mcast[i + groups_cnt + 1].iov_base = iovec[i].iov_base; } res = mcast_msg (iovec_mcast, iov_len + groups_cnt + 1, guarantee); hdb_handle_put (&totempg_groups_instance_database, handle); error_exit: pthread_mutex_unlock (&totempg_mutex); return (res); } /* * Returns -1 if error, 0 if can't send, 1 if can send the message */ int totempg_groups_send_ok_groups ( hdb_handle_t handle, const struct totempg_group *groups, size_t groups_cnt, const struct iovec *iovec, unsigned int iov_len) { struct totempg_group_instance *instance; unsigned int size = 0; unsigned int i; unsigned int res; pthread_mutex_lock (&totempg_mutex); res = hdb_handle_get (&totempg_groups_instance_database, handle, (void *)&instance); if (res != 0) { goto error_exit; } for (i = 0; i < groups_cnt; i++) { size += groups[i].group_len; } for (i = 0; i < iov_len; i++) { size += iovec[i].iov_len; } res = msg_count_send_ok (size); hdb_handle_put (&totempg_groups_instance_database, handle); error_exit: pthread_mutex_unlock (&totempg_mutex); return (res); } int totempg_ifaces_get ( unsigned int nodeid, struct totem_ip_address *interfaces, char ***status, unsigned int *iface_count) { int res; res = totemmrp_ifaces_get ( nodeid, interfaces, status, iface_count); return (res); } int totempg_ring_reenable (void) { int res; res = totemmrp_ring_reenable (); return (res); } const char *totempg_ifaces_print (unsigned int nodeid) { static char iface_string[256 * INTERFACE_MAX]; char one_iface[64]; struct totem_ip_address interfaces[INTERFACE_MAX]; char **status; unsigned int iface_count; unsigned int i; int res; iface_string[0] = '\0'; res = totempg_ifaces_get (nodeid, interfaces, &status, &iface_count); if (res == -1) { return ("no interface found for nodeid"); } for (i = 0; i < iface_count; i++) { sprintf (one_iface, "r(%d) ip(%s) ", i, totemip_print (&interfaces[i])); strcat (iface_string, one_iface); } return (iface_string); } unsigned int totempg_my_nodeid_get (void) { return (totemmrp_my_nodeid_get()); } int totempg_my_family_get (void) { return (totemmrp_my_family_get()); } diff --git a/exec/totemrrp.c b/exec/totemrrp.c index d6354df3..ef56df01 100644 --- a/exec/totemrrp.c +++ b/exec/totemrrp.c @@ -1,1744 +1,1738 @@ /* * Copyright (c) 2005 MontaVista Software, Inc. * Copyright (c) 2006-2008 Red Hat, Inc. * * All rights reserved. * * Author: Steven Dake (sdake@redhat.com) * * 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 MontaVista Software, Inc. 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 #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include "totemnet.h" #include "totemrrp.h" void rrp_deliver_fn ( void *context, void *msg, int msg_len); void rrp_iface_change_fn ( void *context, struct totem_ip_address *iface_addr); struct totemrrp_instance; struct passive_instance { struct totemrrp_instance *rrp_instance; unsigned int *faulty; unsigned int *token_recv_count; unsigned int *mcast_recv_count; unsigned char token[15000]; unsigned int token_len; poll_timer_handle timer_expired_token; poll_timer_handle timer_problem_decrementer; void *totemrrp_context; unsigned int token_xmit_iface; unsigned int msg_xmit_iface; }; struct active_instance { struct totemrrp_instance *rrp_instance; unsigned int *faulty; unsigned int *last_token_recv; unsigned int *counter_problems; unsigned char token[15000]; unsigned int token_len; unsigned int last_token_seq; poll_timer_handle timer_expired_token; poll_timer_handle timer_problem_decrementer; void *totemrrp_context; }; struct rrp_algo { const char *name; void * (*initialize) ( struct totemrrp_instance *rrp_instance, int interface_count); void (*mcast_recv) ( struct totemrrp_instance *instance, unsigned int iface_no, void *context, void *msg, unsigned int msg_len); void (*mcast_noflush_send) ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len); void (*mcast_flush_send) ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len); void (*token_recv) ( struct totemrrp_instance *instance, unsigned int iface_no, void *context, void *msg, unsigned int msg_len, unsigned int token_seqid); void (*token_send) ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len); void (*recv_flush) ( struct totemrrp_instance *instance); void (*send_flush) ( struct totemrrp_instance *instance); void (*iface_check) ( struct totemrrp_instance *instance); void (*processor_count_set) ( struct totemrrp_instance *instance, unsigned int processor_count); void (*token_target_set) ( struct totemrrp_instance *instance, struct totem_ip_address *token_target, unsigned int iface_no); void (*ring_reenable) ( struct totemrrp_instance *instance); }; struct totemrrp_instance { hdb_handle_t totemrrp_poll_handle; struct totem_interface *interfaces; struct rrp_algo *rrp_algo; void *context; char *status[INTERFACE_MAX]; void (*totemrrp_deliver_fn) ( void *context, void *msg, int msg_len); void (*totemrrp_iface_change_fn) ( void *context, struct totem_ip_address *iface_addr, unsigned int iface_no); void (*totemrrp_token_seqid_get) ( void *msg, unsigned int *seqid, unsigned int *token_is); unsigned int (*totemrrp_msgs_missing) (void); /* * Function and data used to log messages */ int totemrrp_log_level_security; int totemrrp_log_level_error; int totemrrp_log_level_warning; int totemrrp_log_level_notice; int totemrrp_log_level_debug; int totemrrp_subsys_id; void (*totemrrp_log_printf) (int subsys, const char *function, const char *file, int line, unsigned int level, const char *format, ...)__attribute__((format(printf, 6, 7))); hdb_handle_t handle; hdb_handle_t *net_handles; void *rrp_algo_instance; int interface_count; hdb_handle_t poll_handle; int processor_count; struct totem_config *totem_config; }; /* * None Replication Forward Declerations */ static void none_mcast_recv ( struct totemrrp_instance *instance, unsigned int iface_no, void *context, void *msg, unsigned int msg_len); static void none_mcast_noflush_send ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len); static void none_mcast_flush_send ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len); static void none_token_recv ( struct totemrrp_instance *instance, unsigned int iface_no, void *context, void *msg, unsigned int msg_len, unsigned int token_seqid); static void none_token_send ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len); static void none_recv_flush ( struct totemrrp_instance *instance); static void none_send_flush ( struct totemrrp_instance *instance); static void none_iface_check ( struct totemrrp_instance *instance); static void none_processor_count_set ( struct totemrrp_instance *instance, unsigned int processor_count_set); static void none_token_target_set ( struct totemrrp_instance *instance, struct totem_ip_address *token_target, unsigned int iface_no); static void none_ring_reenable ( struct totemrrp_instance *instance); /* * Passive Replication Forward Declerations */ static void *passive_instance_initialize ( struct totemrrp_instance *rrp_instance, int interface_count); static void passive_mcast_recv ( struct totemrrp_instance *instance, unsigned int iface_no, void *context, void *msg, unsigned int msg_len); static void passive_mcast_noflush_send ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len); static void passive_mcast_flush_send ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len); static void passive_token_recv ( struct totemrrp_instance *instance, unsigned int iface_no, void *context, void *msg, unsigned int msg_len, unsigned int token_seqid); static void passive_token_send ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len); static void passive_recv_flush ( struct totemrrp_instance *instance); static void passive_send_flush ( struct totemrrp_instance *instance); static void passive_iface_check ( struct totemrrp_instance *instance); static void passive_processor_count_set ( struct totemrrp_instance *instance, unsigned int processor_count_set); static void passive_token_target_set ( struct totemrrp_instance *instance, struct totem_ip_address *token_target, unsigned int iface_no); static void passive_ring_reenable ( struct totemrrp_instance *instance); /* * Active Replication Forward Definitions */ static void *active_instance_initialize ( struct totemrrp_instance *rrp_instance, int interface_count); static void active_mcast_recv ( struct totemrrp_instance *instance, unsigned int iface_no, void *context, void *msg, unsigned int msg_len); static void active_mcast_noflush_send ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len); static void active_mcast_flush_send ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len); static void active_token_recv ( struct totemrrp_instance *instance, unsigned int iface_no, void *context, void *msg, unsigned int msg_len, unsigned int token_seqid); static void active_token_send ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len); static void active_recv_flush ( struct totemrrp_instance *instance); static void active_send_flush ( struct totemrrp_instance *instance); static void active_iface_check ( struct totemrrp_instance *instance); static void active_processor_count_set ( struct totemrrp_instance *instance, unsigned int processor_count_set); static void active_token_target_set ( struct totemrrp_instance *instance, struct totem_ip_address *token_target, unsigned int iface_no); static void active_ring_reenable ( struct totemrrp_instance *instance); static void active_timer_expired_token_start ( struct active_instance *active_instance); static void active_timer_expired_token_cancel ( struct active_instance *active_instance); static void active_timer_problem_decrementer_start ( struct active_instance *active_instance); static void active_timer_problem_decrementer_cancel ( struct active_instance *active_instance); struct rrp_algo none_algo = { .name = "none", .initialize = NULL, .mcast_recv = none_mcast_recv, .mcast_noflush_send = none_mcast_noflush_send, .mcast_flush_send = none_mcast_flush_send, .token_recv = none_token_recv, .token_send = none_token_send, .recv_flush = none_recv_flush, .send_flush = none_send_flush, .iface_check = none_iface_check, .processor_count_set = none_processor_count_set, .token_target_set = none_token_target_set, .ring_reenable = none_ring_reenable }; struct rrp_algo passive_algo = { .name = "passive", .initialize = passive_instance_initialize, .mcast_recv = passive_mcast_recv, .mcast_noflush_send = passive_mcast_noflush_send, .mcast_flush_send = passive_mcast_flush_send, .token_recv = passive_token_recv, .token_send = passive_token_send, .recv_flush = passive_recv_flush, .send_flush = passive_send_flush, .iface_check = passive_iface_check, .processor_count_set = passive_processor_count_set, .token_target_set = passive_token_target_set, .ring_reenable = passive_ring_reenable }; struct rrp_algo active_algo = { .name = "active", .initialize = active_instance_initialize, .mcast_recv = active_mcast_recv, .mcast_noflush_send = active_mcast_noflush_send, .mcast_flush_send = active_mcast_flush_send, .token_recv = active_token_recv, .token_send = active_token_send, .recv_flush = active_recv_flush, .send_flush = active_send_flush, .iface_check = active_iface_check, .processor_count_set = active_processor_count_set, .token_target_set = active_token_target_set, .ring_reenable = active_ring_reenable }; struct rrp_algo *rrp_algos[] = { &none_algo, &passive_algo, &active_algo }; #define RRP_ALGOS_COUNT 3 /* * All instances in one database */ -static struct hdb_handle_database totemrrp_instance_database = { - .handle_count = 0, - .handles = 0, - .iterator = 0, - .mutex = PTHREAD_MUTEX_INITIALIZER -}; - +DECLARE_HDB_DATABASE (totemrrp_instance_database); #define log_printf(level, format, args...) \ do { \ rrp_instance->totemrrp_log_printf ( \ rrp_instance->totemrrp_subsys_id, \ __FUNCTION__, __FILE__, __LINE__, level, \ format, ##args); \ } while (0); /* * None Replication Implementation */ static void none_mcast_recv ( struct totemrrp_instance *rrp_instance, unsigned int iface_no, void *context, void *msg, unsigned int msg_len) { rrp_instance->totemrrp_deliver_fn ( context, msg, msg_len); } static void none_mcast_flush_send ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len) { totemnet_mcast_flush_send (instance->net_handles[0], iovec, iov_len); } static void none_mcast_noflush_send ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len) { totemnet_mcast_noflush_send (instance->net_handles[0], iovec, iov_len); } static void none_token_recv ( struct totemrrp_instance *rrp_instance, unsigned int iface_no, void *context, void *msg, unsigned int msg_len, unsigned int token_seq) { rrp_instance->totemrrp_deliver_fn ( context, msg, msg_len); } static void none_token_send ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len) { totemnet_token_send ( instance->net_handles[0], iovec, iov_len); } static void none_recv_flush (struct totemrrp_instance *instance) { totemnet_recv_flush (instance->net_handles[0]); } static void none_send_flush (struct totemrrp_instance *instance) { totemnet_send_flush (instance->net_handles[0]); } static void none_iface_check (struct totemrrp_instance *instance) { totemnet_iface_check (instance->net_handles[0]); } static void none_processor_count_set ( struct totemrrp_instance *instance, unsigned int processor_count) { totemnet_processor_count_set (instance->net_handles[0], processor_count); } static void none_token_target_set ( struct totemrrp_instance *instance, struct totem_ip_address *token_target, unsigned int iface_no) { totemnet_token_target_set (instance->net_handles[0], token_target); } static void none_ring_reenable ( struct totemrrp_instance *instance) { /* * No operation */ } /* * Passive Replication Implementation */ void *passive_instance_initialize ( struct totemrrp_instance *rrp_instance, int interface_count) { struct passive_instance *instance; instance = malloc (sizeof (struct passive_instance)); if (instance == 0) { goto error_exit; } memset (instance, 0, sizeof (struct passive_instance)); instance->faulty = malloc (sizeof (int) * interface_count); if (instance->faulty == 0) { free (instance); instance = 0; goto error_exit; } memset (instance->faulty, 0, sizeof (int) * interface_count); instance->token_recv_count = malloc (sizeof (int) * interface_count); if (instance->token_recv_count == 0) { free (instance->faulty); free (instance); instance = 0; goto error_exit; } memset (instance->token_recv_count, 0, sizeof (int) * interface_count); instance->mcast_recv_count = malloc (sizeof (int) * interface_count); if (instance->mcast_recv_count == 0) { free (instance->token_recv_count); free (instance->faulty); free (instance); instance = 0; goto error_exit; } memset (instance->mcast_recv_count, 0, sizeof (int) * interface_count); error_exit: return ((void *)instance); } static void timer_function_passive_token_expired (void *context) { struct passive_instance *passive_instance = (struct passive_instance *)context; struct totemrrp_instance *rrp_instance = passive_instance->rrp_instance; rrp_instance->totemrrp_deliver_fn ( passive_instance->totemrrp_context, passive_instance->token, passive_instance->token_len); } /* TODO static void timer_function_passive_problem_decrementer (void *context) { // struct passive_instance *passive_instance = (struct passive_instance *)context; // struct totemrrp_instance *rrp_instance = passive_instance->rrp_instance; } */ static void passive_timer_expired_token_start ( struct passive_instance *passive_instance) { poll_timer_add ( passive_instance->rrp_instance->poll_handle, passive_instance->rrp_instance->totem_config->rrp_token_expired_timeout, (void *)passive_instance, timer_function_passive_token_expired, &passive_instance->timer_expired_token); } static void passive_timer_expired_token_cancel ( struct passive_instance *passive_instance) { poll_timer_delete ( passive_instance->rrp_instance->poll_handle, passive_instance->timer_expired_token); } /* static void passive_timer_problem_decrementer_start ( struct passive_instance *passive_instance) { poll_timer_add ( passive_instance->rrp_instance->poll_handle, passive_instance->rrp_instance->totem_config->rrp_problem_count_timeout, (void *)passive_instance, timer_function_passive_problem_decrementer, &passive_instance->timer_problem_decrementer); } static void passive_timer_problem_decrementer_cancel ( struct passive_instance *passive_instance) { poll_timer_delete ( passive_instance->rrp_instance->poll_handle, passive_instance->timer_problem_decrementer); } */ static void passive_mcast_recv ( struct totemrrp_instance *rrp_instance, unsigned int iface_no, void *context, void *msg, unsigned int msg_len) { struct passive_instance *passive_instance = (struct passive_instance *)rrp_instance->rrp_algo_instance; unsigned int max; unsigned int i; rrp_instance->totemrrp_deliver_fn ( context, msg, msg_len); if (rrp_instance->totemrrp_msgs_missing() == 0 && passive_instance->timer_expired_token) { /* * Delivers the last token */ rrp_instance->totemrrp_deliver_fn ( passive_instance->totemrrp_context, passive_instance->token, passive_instance->token_len); passive_timer_expired_token_cancel (passive_instance); } /* * Monitor for failures * TODO doesn't handle wrap-around of the mcast recv count */ passive_instance->mcast_recv_count[iface_no] += 1; max = 0; for (i = 0; i < rrp_instance->interface_count; i++) { if (max < passive_instance->mcast_recv_count[i]) { max = passive_instance->mcast_recv_count[i]; } } for (i = 0; i < rrp_instance->interface_count; i++) { if ((passive_instance->faulty[i] == 0) && (max - passive_instance->mcast_recv_count[i] > rrp_instance->totem_config->rrp_problem_count_threshold)) { passive_instance->faulty[i] = 1; sprintf (rrp_instance->status[i], "Marking ringid %u interface %s FAULTY - adminisrtative intervention required.", i, totemnet_iface_print (rrp_instance->net_handles[i])); log_printf ( rrp_instance->totemrrp_log_level_error, "%s", rrp_instance->status[i]); } } } static void passive_mcast_flush_send ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len) { struct passive_instance *passive_instance = (struct passive_instance *)instance->rrp_algo_instance; do { passive_instance->msg_xmit_iface = (passive_instance->msg_xmit_iface + 1) % instance->interface_count; } while (passive_instance->faulty[passive_instance->msg_xmit_iface] == 1); totemnet_mcast_flush_send (instance->net_handles[passive_instance->msg_xmit_iface], iovec, iov_len); } static void passive_mcast_noflush_send ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len) { struct passive_instance *passive_instance = (struct passive_instance *)instance->rrp_algo_instance; do { passive_instance->msg_xmit_iface = (passive_instance->msg_xmit_iface + 1) % instance->interface_count; } while (passive_instance->faulty[passive_instance->msg_xmit_iface] == 1); totemnet_mcast_noflush_send (instance->net_handles[passive_instance->msg_xmit_iface], iovec, iov_len); } static void passive_token_recv ( struct totemrrp_instance *rrp_instance, unsigned int iface_no, void *context, void *msg, unsigned int msg_len, unsigned int token_seq) { struct passive_instance *passive_instance = (struct passive_instance *)rrp_instance->rrp_algo_instance; unsigned int max; unsigned int i; passive_instance->totemrrp_context = context; // this should be in totemrrp_instance ? TODO if (rrp_instance->totemrrp_msgs_missing() == 0) { rrp_instance->totemrrp_deliver_fn ( context, msg, msg_len); } else { memcpy (passive_instance->token, msg, msg_len); passive_timer_expired_token_start (passive_instance); } /* * Monitor for failures * TODO doesn't handle wrap-around of the token */ passive_instance->token_recv_count[iface_no] += 1; max = 0; for (i = 0; i < rrp_instance->interface_count; i++) { if (max < passive_instance->token_recv_count[i]) { max = passive_instance->token_recv_count[i]; } } for (i = 0; i < rrp_instance->interface_count; i++) { if ((passive_instance->faulty[i] == 0) && (max - passive_instance->token_recv_count[i] > rrp_instance->totem_config->rrp_problem_count_threshold)) { passive_instance->faulty[i] = 1; sprintf (rrp_instance->status[i], "Marking seqid %d ringid %u interface %s FAULTY - adminisrtative intervention required.", token_seq, i, totemnet_iface_print (rrp_instance->net_handles[i])); log_printf ( rrp_instance->totemrrp_log_level_error, "%s", rrp_instance->status[i]); } } } static void passive_token_send ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len) { struct passive_instance *passive_instance = (struct passive_instance *)instance->rrp_algo_instance; do { passive_instance->token_xmit_iface = (passive_instance->token_xmit_iface + 1) % instance->interface_count; } while (passive_instance->faulty[passive_instance->token_xmit_iface] == 1); totemnet_token_send ( instance->net_handles[passive_instance->token_xmit_iface], iovec, iov_len); } static void passive_recv_flush (struct totemrrp_instance *instance) { struct passive_instance *rrp_algo_instance = (struct passive_instance *)instance->rrp_algo_instance; unsigned int i; for (i = 0; i < instance->interface_count; i++) { if (rrp_algo_instance->faulty[i] == 0) { totemnet_recv_flush (instance->net_handles[i]); } } } static void passive_send_flush (struct totemrrp_instance *instance) { struct passive_instance *rrp_algo_instance = (struct passive_instance *)instance->rrp_algo_instance; unsigned int i; for (i = 0; i < instance->interface_count; i++) { if (rrp_algo_instance->faulty[i] == 0) { totemnet_send_flush (instance->net_handles[i]); } } } static void passive_iface_check (struct totemrrp_instance *instance) { struct passive_instance *rrp_algo_instance = (struct passive_instance *)instance->rrp_algo_instance; unsigned int i; for (i = 0; i < instance->interface_count; i++) { if (rrp_algo_instance->faulty[i] == 0) { totemnet_iface_check (instance->net_handles[i]); } } } static void passive_processor_count_set ( struct totemrrp_instance *instance, unsigned int processor_count) { struct passive_instance *rrp_algo_instance = (struct passive_instance *)instance->rrp_algo_instance; unsigned int i; for (i = 0; i < instance->interface_count; i++) { if (rrp_algo_instance->faulty[i] == 0) { totemnet_processor_count_set (instance->net_handles[i], processor_count); } } } static void passive_token_target_set ( struct totemrrp_instance *instance, struct totem_ip_address *token_target, unsigned int iface_no) { totemnet_token_target_set (instance->net_handles[iface_no], token_target); } static void passive_ring_reenable ( struct totemrrp_instance *instance) { struct passive_instance *rrp_algo_instance = (struct passive_instance *)instance->rrp_algo_instance; memset (rrp_algo_instance->mcast_recv_count, 0, sizeof (unsigned int) * instance->interface_count); memset (rrp_algo_instance->token_recv_count, 0, sizeof (unsigned int) * instance->interface_count); memset (rrp_algo_instance->faulty, 0, sizeof (unsigned int) * instance->interface_count); } /* * Active Replication Implementation */ void *active_instance_initialize ( struct totemrrp_instance *rrp_instance, int interface_count) { struct active_instance *instance; instance = malloc (sizeof (struct active_instance)); if (instance == 0) { goto error_exit; } memset (instance, 0, sizeof (struct active_instance)); instance->faulty = malloc (sizeof (int) * interface_count); if (instance->faulty == 0) { free (instance); instance = 0; goto error_exit; } memset (instance->faulty, 0, sizeof (unsigned int) * interface_count); instance->last_token_recv = malloc (sizeof (int) * interface_count); if (instance->last_token_recv == 0) { free (instance->faulty); free (instance); instance = 0; goto error_exit; } memset (instance->last_token_recv, 0, sizeof (unsigned int) * interface_count); instance->counter_problems = malloc (sizeof (int) * interface_count); if (instance->counter_problems == 0) { free (instance->last_token_recv); free (instance->faulty); free (instance); instance = 0; goto error_exit; } memset (instance->counter_problems, 0, sizeof (unsigned int) * interface_count); instance->timer_expired_token = 0; instance->timer_problem_decrementer = 0; instance->rrp_instance = rrp_instance; error_exit: return ((void *)instance); } static void timer_function_active_problem_decrementer (void *context) { struct active_instance *active_instance = (struct active_instance *)context; struct totemrrp_instance *rrp_instance = active_instance->rrp_instance; unsigned int problem_found = 0; unsigned int i; for (i = 0; i < rrp_instance->interface_count; i++) { if (active_instance->counter_problems[i] > 0) { problem_found = 1; active_instance->counter_problems[i] -= 1; if (active_instance->counter_problems[i] == 0) { sprintf (rrp_instance->status[i], "ring %d active with no faults", i); } else { sprintf (rrp_instance->status[i], "Decrementing problem counter for iface %s to [%d of %d]", totemnet_iface_print (rrp_instance->net_handles[i]), active_instance->counter_problems[i], rrp_instance->totem_config->rrp_problem_count_threshold); } log_printf ( rrp_instance->totemrrp_log_level_warning, "%s", rrp_instance->status[i]); } } if (problem_found) { active_timer_problem_decrementer_start (active_instance); } else { active_instance->timer_problem_decrementer = 0; } } static void timer_function_active_token_expired (void *context) { struct active_instance *active_instance = (struct active_instance *)context; struct totemrrp_instance *rrp_instance = active_instance->rrp_instance; unsigned int i; for (i = 0; i < rrp_instance->interface_count; i++) { if (active_instance->last_token_recv[i] == 0) { active_instance->counter_problems[i] += 1; if (active_instance->timer_problem_decrementer == 0) { active_timer_problem_decrementer_start (active_instance); } sprintf (rrp_instance->status[i], "Incrementing problem counter for seqid %d iface %s to [%d of %d]", active_instance->last_token_seq, totemnet_iface_print (rrp_instance->net_handles[i]), active_instance->counter_problems[i], rrp_instance->totem_config->rrp_problem_count_threshold); log_printf ( rrp_instance->totemrrp_log_level_warning, "%s", rrp_instance->status[i]); } } for (i = 0; i < rrp_instance->interface_count; i++) { if (active_instance->counter_problems[i] >= rrp_instance->totem_config->rrp_problem_count_threshold) { active_instance->faulty[i] = 1; sprintf (rrp_instance->status[i], "Marking seqid %d ringid %u interface %s FAULTY - adminisrtative intervention required.", active_instance->last_token_seq, i, totemnet_iface_print (rrp_instance->net_handles[i])); log_printf ( rrp_instance->totemrrp_log_level_error, "%s", rrp_instance->status[i]); active_timer_problem_decrementer_cancel (active_instance); } } rrp_instance->totemrrp_deliver_fn ( active_instance->totemrrp_context, active_instance->token, active_instance->token_len); } static void active_timer_expired_token_start ( struct active_instance *active_instance) { poll_timer_add ( active_instance->rrp_instance->poll_handle, active_instance->rrp_instance->totem_config->rrp_token_expired_timeout, (void *)active_instance, timer_function_active_token_expired, &active_instance->timer_expired_token); } static void active_timer_expired_token_cancel ( struct active_instance *active_instance) { poll_timer_delete ( active_instance->rrp_instance->poll_handle, active_instance->timer_expired_token); } static void active_timer_problem_decrementer_start ( struct active_instance *active_instance) { poll_timer_add ( active_instance->rrp_instance->poll_handle, active_instance->rrp_instance->totem_config->rrp_problem_count_timeout, (void *)active_instance, timer_function_active_problem_decrementer, &active_instance->timer_problem_decrementer); } static void active_timer_problem_decrementer_cancel ( struct active_instance *active_instance) { poll_timer_delete ( active_instance->rrp_instance->poll_handle, active_instance->timer_problem_decrementer); } /* * active replication */ static void active_mcast_recv ( struct totemrrp_instance *instance, unsigned int iface_no, void *context, void *msg, unsigned int msg_len) { instance->totemrrp_deliver_fn ( context, msg, msg_len); } static void active_mcast_flush_send ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len) { int i; struct active_instance *rrp_algo_instance = (struct active_instance *)instance->rrp_algo_instance; for (i = 0; i < instance->interface_count; i++) { if (rrp_algo_instance->faulty[i] == 0) { totemnet_mcast_flush_send (instance->net_handles[i], iovec, iov_len); } } } static void active_mcast_noflush_send ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len) { int i; struct active_instance *rrp_algo_instance = (struct active_instance *)instance->rrp_algo_instance; for (i = 0; i < instance->interface_count; i++) { if (rrp_algo_instance->faulty[i] == 0) { totemnet_mcast_noflush_send (instance->net_handles[i], iovec, iov_len); } } } static void active_token_recv ( struct totemrrp_instance *instance, unsigned int iface_no, void *context, void *msg, unsigned int msg_len, unsigned int token_seq) { int i; struct active_instance *active_instance = (struct active_instance *)instance->rrp_algo_instance; active_instance->totemrrp_context = context; // this should be in totemrrp_instance ? if (token_seq > active_instance->last_token_seq) { memcpy (active_instance->token, msg, msg_len); active_instance->token_len = msg_len; for (i = 0; i < instance->interface_count; i++) { active_instance->last_token_recv[i] = 0; } active_instance->last_token_recv[iface_no] = 1; active_timer_expired_token_start (active_instance); } active_instance->last_token_seq = token_seq; if (token_seq == active_instance->last_token_seq) { active_instance->last_token_recv[iface_no] = 1; for (i = 0; i < instance->interface_count; i++) { if ((active_instance->last_token_recv[i] == 0) && active_instance->faulty[i] == 0) { return; /* don't deliver token */ } } active_timer_expired_token_cancel (active_instance); instance->totemrrp_deliver_fn ( context, msg, msg_len); } } static void active_token_send ( struct totemrrp_instance *instance, struct iovec *iovec, unsigned int iov_len) { struct active_instance *rrp_algo_instance = (struct active_instance *)instance->rrp_algo_instance; int i; for (i = 0; i < instance->interface_count; i++) { if (rrp_algo_instance->faulty[i] == 0) { totemnet_token_send ( instance->net_handles[i], iovec, iov_len); } } } static void active_recv_flush (struct totemrrp_instance *instance) { struct active_instance *rrp_algo_instance = (struct active_instance *)instance->rrp_algo_instance; unsigned int i; for (i = 0; i < instance->interface_count; i++) { if (rrp_algo_instance->faulty[i] == 0) { totemnet_recv_flush (instance->net_handles[i]); } } } static void active_send_flush (struct totemrrp_instance *instance) { struct active_instance *rrp_algo_instance = (struct active_instance *)instance->rrp_algo_instance; unsigned int i; for (i = 0; i < instance->interface_count; i++) { if (rrp_algo_instance->faulty[i] == 0) { totemnet_send_flush (instance->net_handles[i]); } } } static void active_iface_check (struct totemrrp_instance *instance) { struct active_instance *rrp_algo_instance = (struct active_instance *)instance->rrp_algo_instance; unsigned int i; for (i = 0; i < instance->interface_count; i++) { if (rrp_algo_instance->faulty[i] == 0) { totemnet_iface_check (instance->net_handles[i]); } } } static void active_processor_count_set ( struct totemrrp_instance *instance, unsigned int processor_count) { struct active_instance *rrp_algo_instance = (struct active_instance *)instance->rrp_algo_instance; unsigned int i; for (i = 0; i < instance->interface_count; i++) { if (rrp_algo_instance->faulty[i] == 0) { totemnet_processor_count_set (instance->net_handles[i], processor_count); } } } static void active_token_target_set ( struct totemrrp_instance *instance, struct totem_ip_address *token_target, unsigned int iface_no) { totemnet_token_target_set (instance->net_handles[iface_no], token_target); } static void active_ring_reenable ( struct totemrrp_instance *instance) { struct active_instance *rrp_algo_instance = (struct active_instance *)instance->rrp_algo_instance; memset (rrp_algo_instance->last_token_recv, 0, sizeof (unsigned int) * instance->interface_count); memset (rrp_algo_instance->faulty, 0, sizeof (unsigned int) * instance->interface_count); memset (rrp_algo_instance->counter_problems, 0, sizeof (unsigned int) * instance->interface_count); } struct deliver_fn_context { struct totemrrp_instance *instance; void *context; int iface_no; }; static void totemrrp_instance_initialize (struct totemrrp_instance *instance) { memset (instance, 0, sizeof (struct totemrrp_instance)); } static int totemrrp_algorithm_set ( struct totem_config *totem_config, struct totemrrp_instance *instance) { unsigned int res = -1; unsigned int i; for (i = 0; i < RRP_ALGOS_COUNT; i++) { if (strcmp (totem_config->rrp_mode, rrp_algos[i]->name) == 0) { instance->rrp_algo = rrp_algos[i]; if (rrp_algos[i]->initialize) { instance->rrp_algo_instance = rrp_algos[i]->initialize ( instance, totem_config->interface_count); } res = 0; break; } } for (i = 0; i < totem_config->interface_count; i++) { instance->status[i] = malloc (1024); sprintf (instance->status[i], "ring %d active with no faults", i); } return (res); } void rrp_deliver_fn ( void *context, void *msg, int msg_len) { unsigned int token_seqid; unsigned int token_is; struct deliver_fn_context *deliver_fn_context = (struct deliver_fn_context *)context; deliver_fn_context->instance->totemrrp_token_seqid_get ( msg, &token_seqid, &token_is); if (token_is) { /* * Deliver to the token receiver for this rrp algorithm */ deliver_fn_context->instance->rrp_algo->token_recv ( deliver_fn_context->instance, deliver_fn_context->iface_no, deliver_fn_context->context, msg, msg_len, token_seqid); } else { /* * Deliver to the mcast receiver for this rrp algorithm */ deliver_fn_context->instance->rrp_algo->mcast_recv ( deliver_fn_context->instance, deliver_fn_context->iface_no, deliver_fn_context->context, msg, msg_len); } } void rrp_iface_change_fn ( void *context, struct totem_ip_address *iface_addr) { struct deliver_fn_context *deliver_fn_context = (struct deliver_fn_context *)context; deliver_fn_context->instance->totemrrp_iface_change_fn ( deliver_fn_context->context, iface_addr, deliver_fn_context->iface_no); } int totemrrp_finalize ( hdb_handle_t handle) { struct totemrrp_instance *instance; int res = 0; int i; res = hdb_handle_get (&totemrrp_instance_database, handle, (void *)&instance); if (res != 0) { res = ENOENT; goto error_exit; } for (i = 0; i < instance->interface_count; i++) { totemnet_finalize (instance->net_handles[i]); } hdb_handle_put (&totemrrp_instance_database, handle); error_exit: return (res); } /* * Totem Redundant Ring interface * depends on poll abstraction, POSIX, IPV4 */ /* * Create an instance */ int totemrrp_initialize ( hdb_handle_t poll_handle, hdb_handle_t *handle, struct totem_config *totem_config, void *context, void (*deliver_fn) ( void *context, void *msg, int msg_len), void (*iface_change_fn) ( void *context, struct totem_ip_address *iface_addr, unsigned int iface_no), void (*token_seqid_get) ( void *msg, unsigned int *seqid, unsigned int *token_is), unsigned int (*msgs_missing) (void)) { struct totemrrp_instance *instance; unsigned int res; int i; res = hdb_handle_create (&totemrrp_instance_database, sizeof (struct totemrrp_instance), handle); if (res != 0) { goto error_exit; } res = hdb_handle_get (&totemrrp_instance_database, *handle, (void *)&instance); if (res != 0) { goto error_destroy; } totemrrp_instance_initialize (instance); instance->totem_config = totem_config; res = totemrrp_algorithm_set ( instance->totem_config, instance); if (res == -1) { goto error_put; } /* * Configure logging */ instance->totemrrp_log_level_security = totem_config->totem_logging_configuration.log_level_security; instance->totemrrp_log_level_error = totem_config->totem_logging_configuration.log_level_error; instance->totemrrp_log_level_warning = totem_config->totem_logging_configuration.log_level_warning; instance->totemrrp_log_level_notice = totem_config->totem_logging_configuration.log_level_notice; instance->totemrrp_log_level_debug = totem_config->totem_logging_configuration.log_level_debug; instance->totemrrp_subsys_id = totem_config->totem_logging_configuration.log_subsys_id; instance->totemrrp_log_printf = totem_config->totem_logging_configuration.log_printf; instance->interfaces = totem_config->interfaces; instance->totemrrp_poll_handle = poll_handle; instance->totemrrp_deliver_fn = deliver_fn; instance->totemrrp_iface_change_fn = iface_change_fn; instance->totemrrp_token_seqid_get = token_seqid_get; instance->totemrrp_msgs_missing = msgs_missing; instance->interface_count = totem_config->interface_count; instance->net_handles = malloc (sizeof (hdb_handle_t) * totem_config->interface_count); instance->context = context; instance->poll_handle = poll_handle; for (i = 0; i < totem_config->interface_count; i++) { struct deliver_fn_context *deliver_fn_context; deliver_fn_context = malloc (sizeof (struct deliver_fn_context)); assert (deliver_fn_context); deliver_fn_context->instance = instance; deliver_fn_context->context = context; deliver_fn_context->iface_no = i; totemnet_initialize ( poll_handle, &instance->net_handles[i], totem_config, i, (void *)deliver_fn_context, rrp_deliver_fn, rrp_iface_change_fn); } totemnet_net_mtu_adjust (totem_config); error_exit: hdb_handle_put (&totemrrp_instance_database, *handle); return (0); error_put: hdb_handle_put (&totemrrp_instance_database, *handle); error_destroy: hdb_handle_destroy (&totemrrp_instance_database, *handle); return (res); } int totemrrp_processor_count_set ( hdb_handle_t handle, unsigned int processor_count) { struct totemrrp_instance *instance; int res = 0; res = hdb_handle_get (&totemrrp_instance_database, handle, (void *)&instance); if (res != 0) { res = ENOENT; goto error_exit; } instance->rrp_algo->processor_count_set (instance, processor_count); instance->processor_count = processor_count; hdb_handle_put (&totemrrp_instance_database, handle); error_exit: return (res); } int totemrrp_token_target_set ( hdb_handle_t handle, struct totem_ip_address *addr, unsigned int iface_no) { struct totemrrp_instance *instance; int res = 0; res = hdb_handle_get (&totemrrp_instance_database, handle, (void *)&instance); if (res != 0) { res = ENOENT; goto error_exit; } instance->rrp_algo->token_target_set (instance, addr, iface_no); hdb_handle_put (&totemrrp_instance_database, handle); error_exit: return (res); } int totemrrp_recv_flush (hdb_handle_t handle) { struct totemrrp_instance *instance; int res = 0; res = hdb_handle_get (&totemrrp_instance_database, handle, (void *)&instance); if (res != 0) { res = ENOENT; goto error_exit; } instance->rrp_algo->recv_flush (instance); hdb_handle_put (&totemrrp_instance_database, handle); error_exit: return (res); } int totemrrp_send_flush (hdb_handle_t handle) { struct totemrrp_instance *instance; int res = 0; res = hdb_handle_get (&totemrrp_instance_database, handle, (void *)&instance); if (res != 0) { res = ENOENT; goto error_exit; } instance->rrp_algo->send_flush (instance); hdb_handle_put (&totemrrp_instance_database, handle); error_exit: return (res); } int totemrrp_token_send ( hdb_handle_t handle, struct iovec *iovec, unsigned int iov_len) { struct totemrrp_instance *instance; int res = 0; res = hdb_handle_get (&totemrrp_instance_database, handle, (void *)&instance); if (res != 0) { res = ENOENT; goto error_exit; } instance->rrp_algo->token_send (instance, iovec, iov_len); hdb_handle_put (&totemrrp_instance_database, handle); error_exit: return (res); } int totemrrp_mcast_flush_send ( hdb_handle_t handle, struct iovec *iovec, unsigned int iov_len) { struct totemrrp_instance *instance; int res = 0; res = hdb_handle_get (&totemrrp_instance_database, handle, (void *)&instance); if (res != 0) { res = ENOENT; goto error_exit; } // TODO this needs to return the result instance->rrp_algo->mcast_flush_send (instance, iovec, iov_len); hdb_handle_put (&totemrrp_instance_database, handle); error_exit: return (res); } int totemrrp_mcast_noflush_send ( hdb_handle_t handle, struct iovec *iovec, unsigned int iov_len) { struct totemrrp_instance *instance; int res = 0; res = hdb_handle_get (&totemrrp_instance_database, handle, (void *)&instance); if (res != 0) { res = ENOENT; goto error_exit; } /* * merge detects go out through mcast_flush_send so it is safe to * flush these messages if we are only one processor. This avoids * an encryption/hmac and decryption/hmac */ if (instance->processor_count > 1) { // TODO this needs to return the result instance->rrp_algo->mcast_noflush_send (instance, iovec, iov_len); } hdb_handle_put (&totemrrp_instance_database, handle); error_exit: return (res); } int totemrrp_iface_check (hdb_handle_t handle) { struct totemrrp_instance *instance; int res = 0; res = hdb_handle_get (&totemrrp_instance_database, handle, (void *)&instance); if (res != 0) { res = ENOENT; goto error_exit; } instance->rrp_algo->iface_check (instance); hdb_handle_put (&totemrrp_instance_database, handle); error_exit: return (res); } int totemrrp_ifaces_get ( hdb_handle_t handle, char ***status, unsigned int *iface_count) { struct totemrrp_instance *instance; int res = 0; res = hdb_handle_get (&totemrrp_instance_database, handle, (void *)&instance); if (res != 0) { res = ENOENT; goto error_exit; } *status = instance->status; if (iface_count) { *iface_count = instance->interface_count; } hdb_handle_put (&totemrrp_instance_database, handle); error_exit: return (res); } int totemrrp_ring_reenable ( hdb_handle_t handle) { struct totemrrp_instance *instance; int res = 0; unsigned int i; printf ("totemrrp ring reenable\n"); res = hdb_handle_get (&totemrrp_instance_database, handle, (void *)&instance); if (res != 0) { res = ENOENT; goto error_exit; } instance->rrp_algo->ring_reenable (instance); for (i = 0; i < instance->interface_count; i++) { sprintf (instance->status[i], "ring %d active with no faults", i); } hdb_handle_put (&totemrrp_instance_database, handle); error_exit: return (res); } diff --git a/exec/totemsrp.c b/exec/totemsrp.c index db08edba..d10dec0b 100644 --- a/exec/totemsrp.c +++ b/exec/totemsrp.c @@ -1,4210 +1,4206 @@ /* * Copyright (c) 2003-2006 MontaVista Software, Inc. * Copyright (c) 2006-2009 Red Hat, Inc. * * All rights reserved. * * Author: Steven Dake (sdake@redhat.com) * * 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 MontaVista Software, Inc. 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. */ /* * The first version of this code was based upon Yair Amir's PhD thesis: * http://www.cs.jhu.edu/~yairamir/phd.ps) (ch4,5). * * The current version of totemsrp implements the Totem protocol specified in: * http://citeseer.ist.psu.edu/amir95totem.html * * The deviations from the above published protocols are: * - encryption of message contents with SOBER128 * - authentication of meessage contents with SHA1/HMAC * - token hold mode where token doesn't rotate on unused ring - reduces cpu * usage on 1.6ghz xeon from 35% to less then .1 % as measured by top */ #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include "totemsrp.h" #include "totemrrp.h" #include "wthread.h" #include "crypto.h" #define LOCALHOST_IP inet_addr("127.0.0.1") #define QUEUE_RTR_ITEMS_SIZE_MAX 256 /* allow 256 retransmit items */ #define RETRANS_MESSAGE_QUEUE_SIZE_MAX 500 /* allow 500 messages to be queued */ #define RECEIVED_MESSAGE_QUEUE_SIZE_MAX 500 /* allow 500 messages to be queued */ #define MAXIOVS 5 #define RETRANSMIT_ENTRIES_MAX 30 #define TOKEN_SIZE_MAX 64000 /* bytes */ /* * Rollover handling: * SEQNO_START_MSG is the starting sequence number after a new configuration * This should remain zero, unless testing overflow in which case * 0x7ffff000 and 0xfffff000 are good starting values. * * SEQNO_START_TOKEN is the starting sequence number after a new configuration * for a token. This should remain zero, unless testing overflow in which * case 07fffff00 or 0xffffff00 are good starting values. * * SEQNO_START_MSG is the starting sequence number after a new configuration * This should remain zero, unless testing overflow in which case * 0x7ffff000 and 0xfffff000 are good values to start with */ #define SEQNO_START_MSG 0x0 #define SEQNO_START_TOKEN 0x0 /* * These can be used ot test different rollover points * #define SEQNO_START_MSG 0xfffffe00 * #define SEQNO_START_TOKEN 0xfffffe00 */ /* * These can be used to test the error recovery algorithms * #define TEST_DROP_ORF_TOKEN_PERCENTAGE 30 * #define TEST_DROP_COMMIT_TOKEN_PERCENTAGE 30 * #define TEST_DROP_MCAST_PERCENTAGE 50 * #define TEST_RECOVERY_MSG_COUNT 300 */ /* * we compare incoming messages to determine if their endian is * different - if so convert them * * do not change */ #define ENDIAN_LOCAL 0xff22 enum message_type { MESSAGE_TYPE_ORF_TOKEN = 0, /* Ordering, Reliability, Flow (ORF) control Token */ MESSAGE_TYPE_MCAST = 1, /* ring ordered multicast message */ MESSAGE_TYPE_MEMB_MERGE_DETECT = 2, /* merge rings if there are available rings */ MESSAGE_TYPE_MEMB_JOIN = 3, /* membership join message */ MESSAGE_TYPE_MEMB_COMMIT_TOKEN = 4, /* membership commit token */ MESSAGE_TYPE_TOKEN_HOLD_CANCEL = 5, /* cancel the holding of the token */ }; enum encapsulation_type { MESSAGE_ENCAPSULATED = 1, MESSAGE_NOT_ENCAPSULATED = 2 }; /* * New membership algorithm local variables */ struct srp_addr { struct totem_ip_address addr[INTERFACE_MAX]; }; struct consensus_list_item { struct srp_addr addr; int set; }; struct token_callback_instance { struct list_head list; int (*callback_fn) (enum totem_callback_token_type type, const void *); enum totem_callback_token_type callback_type; int delete; void *data; }; struct totemsrp_socket { int mcast; int token; }; struct message_header { char type; char encapsulated; unsigned short endian_detector; unsigned int nodeid; } __attribute__((packed)); struct mcast { struct message_header header; struct srp_addr system_from; unsigned int seq; int this_seqno; struct memb_ring_id ring_id; unsigned int node_id; int guarantee; } __attribute__((packed)); /* * MTU - multicast message header - IP header - UDP header * * On lossy switches, making use of the DF UDP flag can lead to loss of * forward progress. So the packets must be fragmented by a higher layer * * This layer can only handle packets of MTU size. */ #define FRAGMENT_SIZE (FRAME_SIZE_MAX - sizeof (struct mcast) - 20 - 8) struct rtr_item { struct memb_ring_id ring_id; unsigned int seq; }__attribute__((packed)); struct orf_token { struct message_header header; unsigned int seq; unsigned int token_seq; unsigned int aru; unsigned int aru_addr; struct memb_ring_id ring_id; unsigned int backlog; unsigned int fcc; int retrans_flg; int rtr_list_entries; struct rtr_item rtr_list[0]; }__attribute__((packed)); struct memb_join { struct message_header header; struct srp_addr system_from; unsigned int proc_list_entries; unsigned int failed_list_entries; unsigned long long ring_seq; unsigned char end_of_memb_join[0]; /* * These parts of the data structure are dynamic: * struct srp_addr proc_list[]; * struct srp_addr failed_list[]; */ } __attribute__((packed)); struct memb_merge_detect { struct message_header header; struct srp_addr system_from; struct memb_ring_id ring_id; } __attribute__((packed)); struct token_hold_cancel { struct message_header header; struct memb_ring_id ring_id; } __attribute__((packed)); struct memb_commit_token_memb_entry { struct memb_ring_id ring_id; unsigned int aru; unsigned int high_delivered; unsigned int received_flg; }__attribute__((packed)); struct memb_commit_token { struct message_header header; unsigned int token_seq; struct memb_ring_id ring_id; unsigned int retrans_flg; int memb_index; int addr_entries; unsigned char end_of_commit_token[0]; /* * These parts of the data structure are dynamic: * * struct srp_addr addr[PROCESSOR_COUNT_MAX]; * struct memb_commit_token_memb_entry memb_list[PROCESSOR_COUNT_MAX]; */ }__attribute__((packed)); struct message_item { struct mcast *mcast; struct iovec iovec[MAXIOVS]; unsigned int iov_len; }; struct sort_queue_item { struct iovec iovec[MAXIOVS]; unsigned int iov_len; }; struct orf_token_mcast_thread_state { char iobuf[9000]; prng_state prng_state; }; enum memb_state { MEMB_STATE_OPERATIONAL = 1, MEMB_STATE_GATHER = 2, MEMB_STATE_COMMIT = 3, MEMB_STATE_RECOVERY = 4 }; struct totemsrp_instance { int iface_changes; /* * Flow control mcasts and remcasts on last and current orf_token */ int fcc_remcast_last; int fcc_mcast_last; int fcc_remcast_current; struct consensus_list_item consensus_list[PROCESSOR_COUNT_MAX]; int consensus_list_entries; struct srp_addr my_id; struct srp_addr my_proc_list[PROCESSOR_COUNT_MAX]; struct srp_addr my_failed_list[PROCESSOR_COUNT_MAX]; struct srp_addr my_new_memb_list[PROCESSOR_COUNT_MAX]; struct srp_addr my_trans_memb_list[PROCESSOR_COUNT_MAX]; struct srp_addr my_memb_list[PROCESSOR_COUNT_MAX]; struct srp_addr my_deliver_memb_list[PROCESSOR_COUNT_MAX]; struct srp_addr my_left_memb_list[PROCESSOR_COUNT_MAX]; int my_proc_list_entries; int my_failed_list_entries; int my_new_memb_entries; int my_trans_memb_entries; int my_memb_entries; int my_deliver_memb_entries; int my_left_memb_entries; struct memb_ring_id my_ring_id; struct memb_ring_id my_old_ring_id; int my_aru_count; int my_merge_detect_timeout_outstanding; unsigned int my_last_aru; int my_seq_unchanged; int my_received_flg; unsigned int my_high_seq_received; unsigned int my_install_seq; int my_rotation_counter; int my_set_retrans_flg; int my_retrans_flg_count; unsigned int my_high_ring_delivered; int heartbeat_timeout; /* * Queues used to order, deliver, and recover messages */ struct queue new_message_queue; struct queue retrans_message_queue; struct sq regular_sort_queue; struct sq recovery_sort_queue; /* * Received up to and including */ unsigned int my_aru; unsigned int my_high_delivered; struct list_head token_callback_received_listhead; struct list_head token_callback_sent_listhead; char *orf_token_retransmit[TOKEN_SIZE_MAX]; int orf_token_retransmit_size; unsigned int my_token_seq; /* * Timers */ poll_timer_handle timer_orf_token_timeout; poll_timer_handle timer_orf_token_retransmit_timeout; poll_timer_handle timer_orf_token_hold_retransmit_timeout; poll_timer_handle timer_merge_detect_timeout; poll_timer_handle memb_timer_state_gather_join_timeout; poll_timer_handle memb_timer_state_gather_consensus_timeout; poll_timer_handle memb_timer_state_commit_timeout; poll_timer_handle timer_heartbeat_timeout; /* * Function and data used to log messages */ int totemsrp_log_level_security; int totemsrp_log_level_error; int totemsrp_log_level_warning; int totemsrp_log_level_notice; int totemsrp_log_level_debug; int totemsrp_subsys_id; void (*totemsrp_log_printf) (int subsys, const char *function, const char *file, int line, unsigned int level, const char *format, ...)__attribute__((format(printf, 6, 7)));; enum memb_state memb_state; //TODO struct srp_addr next_memb; char iov_buffer[FRAME_SIZE_MAX]; struct iovec totemsrp_iov_recv; hdb_handle_t totemsrp_poll_handle; /* * Function called when new message received */ int (*totemsrp_recv) (char *group, struct iovec *iovec, unsigned int iov_len); struct totem_ip_address mcast_address; void (*totemsrp_deliver_fn) ( unsigned int nodeid, struct iovec *iovec, unsigned int iov_len, int endian_conversion_required); void (*totemsrp_confchg_fn) ( enum totem_configuration_type configuration_type, const unsigned int *member_list, size_t member_list_entries, const unsigned int *left_list, size_t left_list_entries, const unsigned int *joined_list, size_t joined_list_entries, const struct memb_ring_id *ring_id); int global_seqno; int my_token_held; unsigned long long token_ring_id_seq; unsigned int last_released; unsigned int set_aru; int old_ring_state_saved; int old_ring_state_aru; unsigned int old_ring_state_high_seq_received; int ring_saved; unsigned int my_last_seq; struct timeval tv_old; hdb_handle_t totemrrp_handle; struct totem_config *totem_config; unsigned int use_heartbeat; unsigned int my_trc; unsigned int my_pbl; unsigned int my_cbl; }; struct message_handlers { int count; int (*handler_functions[6]) ( struct totemsrp_instance *instance, void *msg, int msg_len, int endian_conversion_needed); }; /* * forward decls */ static int message_handler_orf_token ( struct totemsrp_instance *instance, void *msg, int msg_len, int endian_conversion_needed); static int message_handler_mcast ( struct totemsrp_instance *instance, void *msg, int msg_len, int endian_conversion_needed); static int message_handler_memb_merge_detect ( struct totemsrp_instance *instance, void *msg, int msg_len, int endian_conversion_needed); static int message_handler_memb_join ( struct totemsrp_instance *instance, void *msg, int msg_len, int endian_conversion_needed); static int message_handler_memb_commit_token ( struct totemsrp_instance *instance, void *msg, int msg_len, int endian_conversion_needed); static int message_handler_token_hold_cancel ( struct totemsrp_instance *instance, void *msg, int msg_len, int endian_conversion_needed); static void totemsrp_instance_initialize (struct totemsrp_instance *instance); static unsigned int main_msgs_missing (void); static void main_token_seqid_get ( void *msg, unsigned int *seqid, unsigned int *token_is); static void srp_addr_copy (struct srp_addr *dest, struct srp_addr *src); static void srp_addr_to_nodeid ( unsigned int *nodeid_out, struct srp_addr *srp_addr_in, unsigned int entries); static int srp_addr_equal (struct srp_addr *a, struct srp_addr *b); static void memb_ring_id_create_or_load (struct totemsrp_instance *, struct memb_ring_id *); static void token_callbacks_execute (struct totemsrp_instance *instance, enum totem_callback_token_type type); static void memb_state_gather_enter (struct totemsrp_instance *instance, int gather_from); static void messages_deliver_to_app (struct totemsrp_instance *instance, int skip, unsigned int end_point); static int orf_token_mcast (struct totemsrp_instance *instance, struct orf_token *oken, int fcc_mcasts_allowed); static void messages_free (struct totemsrp_instance *instance, unsigned int token_aru); static void memb_ring_id_set_and_store (struct totemsrp_instance *instance, struct memb_ring_id *ring_id); static void memb_state_commit_token_update (struct totemsrp_instance *instance, struct memb_commit_token *commit_token); static void memb_state_commit_token_target_set (struct totemsrp_instance *instance, struct memb_commit_token *commit_token); static int memb_state_commit_token_send (struct totemsrp_instance *instance, struct memb_commit_token *memb_commit_token); static void memb_state_commit_token_create (struct totemsrp_instance *instance, struct memb_commit_token *commit_token); static int token_hold_cancel_send (struct totemsrp_instance *instance); static void orf_token_endian_convert (struct orf_token *in, struct orf_token *out); static void memb_commit_token_endian_convert (struct memb_commit_token *in, struct memb_commit_token *out); static void memb_join_endian_convert (struct memb_join *in, struct memb_join *out); static void mcast_endian_convert (struct mcast *in, struct mcast *out); static void memb_merge_detect_endian_convert ( struct memb_merge_detect *in, struct memb_merge_detect *out); static void srp_addr_copy_endian_convert (struct srp_addr *out, struct srp_addr *in); static void timer_function_orf_token_timeout (void *data); static void timer_function_heartbeat_timeout (void *data); static void timer_function_token_retransmit_timeout (void *data); static void timer_function_token_hold_retransmit_timeout (void *data); static void timer_function_merge_detect_timeout (void *data); void main_deliver_fn ( void *context, void *msg, int msg_len); void main_iface_change_fn ( void *context, struct totem_ip_address *iface_address, unsigned int iface_no); /* * All instances in one database */ -static struct hdb_handle_database totemsrp_instance_database = { - .handle_count = 0, - .handles = 0, - .iterator = 0, - .mutex = PTHREAD_MUTEX_INITIALIZER -}; +DECLARE_HDB_DATABASE (totemsrp_instance_database); + struct message_handlers totemsrp_message_handlers = { 6, { message_handler_orf_token, message_handler_mcast, message_handler_memb_merge_detect, message_handler_memb_join, message_handler_memb_commit_token, message_handler_token_hold_cancel } }; static const char *rundir = NULL; #define log_printf(level, format, args...) \ do { \ instance->totemsrp_log_printf (instance->totemsrp_subsys_id, \ __FUNCTION__, __FILE__, __LINE__, level, \ format, ##args); \ } while (0); static void totemsrp_instance_initialize (struct totemsrp_instance *instance) { memset (instance, 0, sizeof (struct totemsrp_instance)); list_init (&instance->token_callback_received_listhead); list_init (&instance->token_callback_sent_listhead); instance->my_received_flg = 1; instance->my_token_seq = SEQNO_START_TOKEN - 1; instance->memb_state = MEMB_STATE_OPERATIONAL; instance->set_aru = -1; instance->my_aru = SEQNO_START_MSG; instance->my_high_seq_received = SEQNO_START_MSG; instance->my_high_delivered = SEQNO_START_MSG; } static void main_token_seqid_get ( void *msg, unsigned int *seqid, unsigned int *token_is) { struct orf_token *token = (struct orf_token *)msg; *seqid = 0; *token_is = 0; if (token->header.type == MESSAGE_TYPE_ORF_TOKEN) { *seqid = token->token_seq; *token_is = 1; } } static unsigned int main_msgs_missing (void) { // TODO return (0); } /* * Exported interfaces */ int totemsrp_initialize ( hdb_handle_t poll_handle, hdb_handle_t *handle, struct totem_config *totem_config, void (*deliver_fn) ( unsigned int nodeid, struct iovec *iovec, unsigned int iov_len, int endian_conversion_required), void (*confchg_fn) ( enum totem_configuration_type configuration_type, const unsigned int *member_list, size_t member_list_entries, const unsigned int *left_list, size_t left_list_entries, const unsigned int *joined_list, size_t joined_list_entries, const struct memb_ring_id *ring_id)) { struct totemsrp_instance *instance; unsigned int res; res = hdb_handle_create (&totemsrp_instance_database, sizeof (struct totemsrp_instance), handle); if (res != 0) { goto error_exit; } res = hdb_handle_get (&totemsrp_instance_database, *handle, (void *)&instance); if (res != 0) { goto error_destroy; } rundir = getenv ("COROSYNC_RUN_DIR"); if (rundir == NULL) { rundir = LOCALSTATEDIR "/lib/corosync"; } res = mkdir (rundir, 0700); if (res == -1 && errno != EEXIST) { goto error_put; } res = chdir (rundir); if (res == -1) { goto error_put; } totemsrp_instance_initialize (instance); instance->totem_config = totem_config; /* * Configure logging */ instance->totemsrp_log_level_security = totem_config->totem_logging_configuration.log_level_security; instance->totemsrp_log_level_error = totem_config->totem_logging_configuration.log_level_error; instance->totemsrp_log_level_warning = totem_config->totem_logging_configuration.log_level_warning; instance->totemsrp_log_level_notice = totem_config->totem_logging_configuration.log_level_notice; instance->totemsrp_log_level_debug = totem_config->totem_logging_configuration.log_level_debug; instance->totemsrp_subsys_id = totem_config->totem_logging_configuration.log_subsys_id; instance->totemsrp_log_printf = totem_config->totem_logging_configuration.log_printf; /* * Initialize local variables for totemsrp */ totemip_copy (&instance->mcast_address, &totem_config->interfaces[0].mcast_addr); memset (instance->iov_buffer, 0, FRAME_SIZE_MAX); /* * Display totem configuration */ log_printf (instance->totemsrp_log_level_notice, "Token Timeout (%d ms) retransmit timeout (%d ms)\n", totem_config->token_timeout, totem_config->token_retransmit_timeout); log_printf (instance->totemsrp_log_level_notice, "token hold (%d ms) retransmits before loss (%d retrans)\n", totem_config->token_hold_timeout, totem_config->token_retransmits_before_loss_const); log_printf (instance->totemsrp_log_level_notice, "join (%d ms) send_join (%d ms) consensus (%d ms) merge (%d ms)\n", totem_config->join_timeout, totem_config->send_join_timeout, totem_config->consensus_timeout, totem_config->merge_timeout); log_printf (instance->totemsrp_log_level_notice, "downcheck (%d ms) fail to recv const (%d msgs)\n", totem_config->downcheck_timeout, totem_config->fail_to_recv_const); log_printf (instance->totemsrp_log_level_notice, "seqno unchanged const (%d rotations) Maximum network MTU %d\n", totem_config->seqno_unchanged_const, totem_config->net_mtu); log_printf (instance->totemsrp_log_level_notice, "window size per rotation (%d messages) maximum messages per rotation (%d messages)\n", totem_config->window_size, totem_config->max_messages); log_printf (instance->totemsrp_log_level_notice, "send threads (%d threads)\n", totem_config->threads); log_printf (instance->totemsrp_log_level_notice, "RRP token expired timeout (%d ms)\n", totem_config->rrp_token_expired_timeout); log_printf (instance->totemsrp_log_level_notice, "RRP token problem counter (%d ms)\n", totem_config->rrp_problem_count_timeout); log_printf (instance->totemsrp_log_level_notice, "RRP threshold (%d problem count)\n", totem_config->rrp_problem_count_threshold); log_printf (instance->totemsrp_log_level_notice, "RRP mode set to %s.\n", instance->totem_config->rrp_mode); log_printf (instance->totemsrp_log_level_notice, "heartbeat_failures_allowed (%d)\n", totem_config->heartbeat_failures_allowed); log_printf (instance->totemsrp_log_level_notice, "max_network_delay (%d ms)\n", totem_config->max_network_delay); queue_init (&instance->retrans_message_queue, RETRANS_MESSAGE_QUEUE_SIZE_MAX, sizeof (struct message_item)); sq_init (&instance->regular_sort_queue, QUEUE_RTR_ITEMS_SIZE_MAX, sizeof (struct sort_queue_item), 0); sq_init (&instance->recovery_sort_queue, QUEUE_RTR_ITEMS_SIZE_MAX, sizeof (struct sort_queue_item), 0); instance->totemsrp_poll_handle = poll_handle; instance->totemsrp_deliver_fn = deliver_fn; instance->totemsrp_confchg_fn = confchg_fn; instance->use_heartbeat = 1; if ( totem_config->heartbeat_failures_allowed == 0 ) { log_printf (instance->totemsrp_log_level_notice, "HeartBeat is Disabled. To enable set heartbeat_failures_allowed > 0\n"); instance->use_heartbeat = 0; } if (instance->use_heartbeat) { instance->heartbeat_timeout = (totem_config->heartbeat_failures_allowed) * totem_config->token_retransmit_timeout + totem_config->max_network_delay; if (instance->heartbeat_timeout >= totem_config->token_timeout) { log_printf (instance->totemsrp_log_level_notice, "total heartbeat_timeout (%d ms) is not less than token timeout (%d ms)\n", instance->heartbeat_timeout, totem_config->token_timeout); log_printf (instance->totemsrp_log_level_notice, "heartbeat_timeout = heartbeat_failures_allowed * token_retransmit_timeout + max_network_delay\n"); log_printf (instance->totemsrp_log_level_notice, "heartbeat timeout should be less than the token timeout. HeartBeat is Diabled !!\n"); instance->use_heartbeat = 0; } else { log_printf (instance->totemsrp_log_level_notice, "total heartbeat_timeout (%d ms)\n", instance->heartbeat_timeout); } } totemrrp_initialize ( poll_handle, &instance->totemrrp_handle, totem_config, instance, main_deliver_fn, main_iface_change_fn, main_token_seqid_get, main_msgs_missing); /* * Must have net_mtu adjusted by totemrrp_initialize first */ queue_init (&instance->new_message_queue, MESSAGE_QUEUE_MAX, sizeof (struct message_item)); hdb_handle_put (&totemsrp_instance_database, *handle); return (0); error_put: hdb_handle_put (&totemsrp_instance_database, *handle); error_destroy: hdb_handle_destroy (&totemsrp_instance_database, *handle); error_exit: return (-1); } void totemsrp_finalize ( hdb_handle_t handle) { struct totemsrp_instance *instance; unsigned int res; res = hdb_handle_get (&totemsrp_instance_database, handle, (void *)&instance); if (res != 0) { return; } hdb_handle_put (&totemsrp_instance_database, handle); } int totemsrp_ifaces_get ( hdb_handle_t handle, unsigned int nodeid, struct totem_ip_address *interfaces, char ***status, unsigned int *iface_count) { struct totemsrp_instance *instance; int res; unsigned int found = 0; unsigned int i; res = hdb_handle_get (&totemsrp_instance_database, handle, (void *)&instance); if (res != 0) { goto error_exit; } for (i = 0; i < instance->my_memb_entries; i++) { if (instance->my_memb_list[i].addr[0].nodeid == nodeid) { found = 1; break; } } if (found) { memcpy (interfaces, &instance->my_memb_list[i], sizeof (struct srp_addr)); *iface_count = instance->totem_config->interface_count; goto finish; } for (i = 0; i < instance->my_left_memb_entries; i++) { if (instance->my_left_memb_list[i].addr[0].nodeid == nodeid) { found = 1; break; } } if (found) { memcpy (interfaces, &instance->my_left_memb_list[i], sizeof (struct srp_addr)); *iface_count = instance->totem_config->interface_count; } else { res = -1; } finish: totemrrp_ifaces_get (instance->totemrrp_handle, status, NULL); hdb_handle_put (&totemsrp_instance_database, handle); error_exit: return (res); } unsigned int totemsrp_my_nodeid_get ( hdb_handle_t handle) { struct totemsrp_instance *instance; unsigned int res; res = hdb_handle_get (&totemsrp_instance_database, handle, (void *)&instance); if (res != 0) { return (0); } res = instance->totem_config->interfaces[0].boundto.nodeid; hdb_handle_put (&totemsrp_instance_database, handle); return (res); } int totemsrp_my_family_get ( hdb_handle_t handle) { struct totemsrp_instance *instance; int res; res = hdb_handle_get (&totemsrp_instance_database, handle, (void *)&instance); if (res != 0) { return (0); } res = instance->totem_config->interfaces[0].boundto.family; hdb_handle_put (&totemsrp_instance_database, handle); return (res); } int totemsrp_ring_reenable ( hdb_handle_t handle) { struct totemsrp_instance *instance; int res; res = hdb_handle_get (&totemsrp_instance_database, handle, (void *)&instance); if (res != 0) { goto error_exit; } totemrrp_ring_reenable (instance->totemrrp_handle); hdb_handle_put (&totemsrp_instance_database, handle); error_exit: return (res); } /* * Set operations for use by the membership algorithm */ static int srp_addr_equal (struct srp_addr *a, struct srp_addr *b) { unsigned int i; unsigned int res; for (i = 0; i < 1; i++) { res = totemip_equal (&a->addr[i], &b->addr[i]); if (res == 0) { return (0); } } return (1); } static void srp_addr_copy (struct srp_addr *dest, struct srp_addr *src) { unsigned int i; for (i = 0; i < INTERFACE_MAX; i++) { totemip_copy (&dest->addr[i], &src->addr[i]); } } static void srp_addr_to_nodeid ( unsigned int *nodeid_out, struct srp_addr *srp_addr_in, unsigned int entries) { unsigned int i; for (i = 0; i < entries; i++) { nodeid_out[i] = srp_addr_in[i].addr[0].nodeid; } } static void srp_addr_copy_endian_convert (struct srp_addr *out, struct srp_addr *in) { int i; for (i = 0; i < INTERFACE_MAX; i++) { totemip_copy_endian_convert (&out->addr[i], &in->addr[i]); } } static void memb_consensus_reset (struct totemsrp_instance *instance) { instance->consensus_list_entries = 0; } static void memb_set_subtract ( struct srp_addr *out_list, int *out_list_entries, struct srp_addr *one_list, int one_list_entries, struct srp_addr *two_list, int two_list_entries) { int found = 0; int i; int j; *out_list_entries = 0; for (i = 0; i < one_list_entries; i++) { for (j = 0; j < two_list_entries; j++) { if (srp_addr_equal (&one_list[i], &two_list[j])) { found = 1; break; } } if (found == 0) { srp_addr_copy (&out_list[*out_list_entries], &one_list[i]); *out_list_entries = *out_list_entries + 1; } found = 0; } } /* * Set consensus for a specific processor */ static void memb_consensus_set ( struct totemsrp_instance *instance, struct srp_addr *addr) { int found = 0; int i; for (i = 0; i < instance->consensus_list_entries; i++) { if (srp_addr_equal(addr, &instance->consensus_list[i].addr)) { found = 1; break; /* found entry */ } } srp_addr_copy (&instance->consensus_list[i].addr, addr); instance->consensus_list[i].set = 1; if (found == 0) { instance->consensus_list_entries++; } return; } /* * Is consensus set for a specific processor */ static int memb_consensus_isset ( struct totemsrp_instance *instance, struct srp_addr *addr) { int i; for (i = 0; i < instance->consensus_list_entries; i++) { if (srp_addr_equal (addr, &instance->consensus_list[i].addr)) { return (instance->consensus_list[i].set); } } return (0); } /* * Is consensus agreed upon based upon consensus database */ static int memb_consensus_agreed ( struct totemsrp_instance *instance) { struct srp_addr token_memb[PROCESSOR_COUNT_MAX]; int token_memb_entries = 0; int agreed = 1; int i; memb_set_subtract (token_memb, &token_memb_entries, instance->my_proc_list, instance->my_proc_list_entries, instance->my_failed_list, instance->my_failed_list_entries); for (i = 0; i < token_memb_entries; i++) { if (memb_consensus_isset (instance, &token_memb[i]) == 0) { agreed = 0; break; } } assert (token_memb_entries >= 1); return (agreed); } static void memb_consensus_notset ( struct totemsrp_instance *instance, struct srp_addr *no_consensus_list, int *no_consensus_list_entries, struct srp_addr *comparison_list, int comparison_list_entries) { int i; *no_consensus_list_entries = 0; for (i = 0; i < instance->my_proc_list_entries; i++) { if (memb_consensus_isset (instance, &instance->my_proc_list[i]) == 0) { srp_addr_copy (&no_consensus_list[*no_consensus_list_entries], &instance->my_proc_list[i]); *no_consensus_list_entries = *no_consensus_list_entries + 1; } } } /* * Is set1 equal to set2 Entries can be in different orders */ static int memb_set_equal ( struct srp_addr *set1, int set1_entries, struct srp_addr *set2, int set2_entries) { int i; int j; int found = 0; if (set1_entries != set2_entries) { return (0); } for (i = 0; i < set2_entries; i++) { for (j = 0; j < set1_entries; j++) { if (srp_addr_equal (&set1[j], &set2[i])) { found = 1; break; } } if (found == 0) { return (0); } found = 0; } return (1); } /* * Is subset fully contained in fullset */ static int memb_set_subset ( struct srp_addr *subset, int subset_entries, struct srp_addr *fullset, int fullset_entries) { int i; int j; int found = 0; if (subset_entries > fullset_entries) { return (0); } for (i = 0; i < subset_entries; i++) { for (j = 0; j < fullset_entries; j++) { if (srp_addr_equal (&subset[i], &fullset[j])) { found = 1; } } if (found == 0) { return (0); } found = 0; } return (1); } /* * merge subset into fullset taking care not to add duplicates */ static void memb_set_merge ( struct srp_addr *subset, int subset_entries, struct srp_addr *fullset, int *fullset_entries) { int found = 0; int i; int j; for (i = 0; i < subset_entries; i++) { for (j = 0; j < *fullset_entries; j++) { if (srp_addr_equal (&fullset[j], &subset[i])) { found = 1; break; } } if (found == 0) { srp_addr_copy (&fullset[*fullset_entries], &subset[i]); *fullset_entries = *fullset_entries + 1; } found = 0; } return; } static void memb_set_and ( struct srp_addr *set1, int set1_entries, struct srp_addr *set2, int set2_entries, struct srp_addr *and, int *and_entries) { int i; int j; int found = 0; *and_entries = 0; for (i = 0; i < set2_entries; i++) { for (j = 0; j < set1_entries; j++) { if (srp_addr_equal (&set1[j], &set2[i])) { found = 1; break; } } if (found) { srp_addr_copy (&and[*and_entries], &set1[j]); *and_entries = *and_entries + 1; } found = 0; } return; } #ifdef CODE_COVERAGE static void memb_set_print ( char *string, struct srp_addr *list, int list_entries) { int i; int j; printf ("List '%s' contains %d entries:\n", string, list_entries); for (i = 0; i < list_entries; i++) { for (j = 0; j < INTERFACE_MAX; j++) { printf ("Address %d\n", i); printf ("\tiface %d %s\n", j, totemip_print (&list[i].addr[j])); printf ("family %d\n", list[i].addr[j].family); } } } #endif static void reset_token_retransmit_timeout (struct totemsrp_instance *instance) { poll_timer_delete (instance->totemsrp_poll_handle, instance->timer_orf_token_retransmit_timeout); poll_timer_add (instance->totemsrp_poll_handle, instance->totem_config->token_retransmit_timeout, (void *)instance, timer_function_token_retransmit_timeout, &instance->timer_orf_token_retransmit_timeout); } static void start_merge_detect_timeout (struct totemsrp_instance *instance) { if (instance->my_merge_detect_timeout_outstanding == 0) { poll_timer_add (instance->totemsrp_poll_handle, instance->totem_config->merge_timeout, (void *)instance, timer_function_merge_detect_timeout, &instance->timer_merge_detect_timeout); instance->my_merge_detect_timeout_outstanding = 1; } } static void cancel_merge_detect_timeout (struct totemsrp_instance *instance) { poll_timer_delete (instance->totemsrp_poll_handle, instance->timer_merge_detect_timeout); instance->my_merge_detect_timeout_outstanding = 0; } /* * ring_state_* is used to save and restore the sort queue * state when a recovery operation fails (and enters gather) */ static void old_ring_state_save (struct totemsrp_instance *instance) { if (instance->old_ring_state_saved == 0) { instance->old_ring_state_saved = 1; instance->old_ring_state_aru = instance->my_aru; instance->old_ring_state_high_seq_received = instance->my_high_seq_received; log_printf (instance->totemsrp_log_level_notice, "Saving state aru %x high seq received %x\n", instance->my_aru, instance->my_high_seq_received); } } static void ring_save (struct totemsrp_instance *instance) { if (instance->ring_saved == 0) { instance->ring_saved = 1; memcpy (&instance->my_old_ring_id, &instance->my_ring_id, sizeof (struct memb_ring_id)); } } static void ring_reset (struct totemsrp_instance *instance) { instance->ring_saved = 0; } static void ring_state_restore (struct totemsrp_instance *instance) { if (instance->old_ring_state_saved) { totemip_zero_set(&instance->my_ring_id.rep); instance->my_aru = instance->old_ring_state_aru; instance->my_high_seq_received = instance->old_ring_state_high_seq_received; log_printf (instance->totemsrp_log_level_notice, "Restoring instance->my_aru %x my high seq received %x\n", instance->my_aru, instance->my_high_seq_received); } } static void old_ring_state_reset (struct totemsrp_instance *instance) { instance->old_ring_state_saved = 0; } static void reset_token_timeout (struct totemsrp_instance *instance) { poll_timer_delete (instance->totemsrp_poll_handle, instance->timer_orf_token_timeout); poll_timer_add (instance->totemsrp_poll_handle, instance->totem_config->token_timeout, (void *)instance, timer_function_orf_token_timeout, &instance->timer_orf_token_timeout); } static void reset_heartbeat_timeout (struct totemsrp_instance *instance) { poll_timer_delete (instance->totemsrp_poll_handle, instance->timer_heartbeat_timeout); poll_timer_add (instance->totemsrp_poll_handle, instance->heartbeat_timeout, (void *)instance, timer_function_heartbeat_timeout, &instance->timer_heartbeat_timeout); } static void cancel_token_timeout (struct totemsrp_instance *instance) { poll_timer_delete (instance->totemsrp_poll_handle, instance->timer_orf_token_timeout); } static void cancel_heartbeat_timeout (struct totemsrp_instance *instance) { poll_timer_delete (instance->totemsrp_poll_handle, instance->timer_heartbeat_timeout); } static void cancel_token_retransmit_timeout (struct totemsrp_instance *instance) { poll_timer_delete (instance->totemsrp_poll_handle, instance->timer_orf_token_retransmit_timeout); } static void start_token_hold_retransmit_timeout (struct totemsrp_instance *instance) { poll_timer_add (instance->totemsrp_poll_handle, instance->totem_config->token_hold_timeout, (void *)instance, timer_function_token_hold_retransmit_timeout, &instance->timer_orf_token_hold_retransmit_timeout); } static void cancel_token_hold_retransmit_timeout (struct totemsrp_instance *instance) { poll_timer_delete (instance->totemsrp_poll_handle, instance->timer_orf_token_hold_retransmit_timeout); } static void memb_state_consensus_timeout_expired ( struct totemsrp_instance *instance) { struct srp_addr no_consensus_list[PROCESSOR_COUNT_MAX]; int no_consensus_list_entries; if (memb_consensus_agreed (instance)) { memb_consensus_reset (instance); memb_consensus_set (instance, &instance->my_id); reset_token_timeout (instance); // REVIEWED } else { memb_consensus_notset ( instance, no_consensus_list, &no_consensus_list_entries, instance->my_proc_list, instance->my_proc_list_entries); memb_set_merge (no_consensus_list, no_consensus_list_entries, instance->my_failed_list, &instance->my_failed_list_entries); memb_state_gather_enter (instance, 0); } } static void memb_join_message_send (struct totemsrp_instance *instance); static void memb_merge_detect_transmit (struct totemsrp_instance *instance); /* * Timers used for various states of the membership algorithm */ static void timer_function_orf_token_timeout (void *data) { struct totemsrp_instance *instance = (struct totemsrp_instance *)data; switch (instance->memb_state) { case MEMB_STATE_OPERATIONAL: log_printf (instance->totemsrp_log_level_notice, "The token was lost in the OPERATIONAL state.\n"); totemrrp_iface_check (instance->totemrrp_handle); memb_state_gather_enter (instance, 2); break; case MEMB_STATE_GATHER: log_printf (instance->totemsrp_log_level_notice, "The consensus timeout expired.\n"); memb_state_consensus_timeout_expired (instance); memb_state_gather_enter (instance, 3); break; case MEMB_STATE_COMMIT: log_printf (instance->totemsrp_log_level_notice, "The token was lost in the COMMIT state.\n"); memb_state_gather_enter (instance, 4); break; case MEMB_STATE_RECOVERY: log_printf (instance->totemsrp_log_level_notice, "The token was lost in the RECOVERY state.\n"); ring_state_restore (instance); memb_state_gather_enter (instance, 5); break; } } static void timer_function_heartbeat_timeout (void *data) { struct totemsrp_instance *instance = (struct totemsrp_instance *)data; log_printf (instance->totemsrp_log_level_notice, "HeartBeat Timer expired Invoking token loss mechanism in state %d \n", instance->memb_state); timer_function_orf_token_timeout(data); } static void memb_timer_function_state_gather (void *data) { struct totemsrp_instance *instance = (struct totemsrp_instance *)data; switch (instance->memb_state) { case MEMB_STATE_OPERATIONAL: case MEMB_STATE_RECOVERY: assert (0); /* this should never happen */ break; case MEMB_STATE_GATHER: case MEMB_STATE_COMMIT: memb_join_message_send (instance); /* * Restart the join timeout `*/ poll_timer_delete (instance->totemsrp_poll_handle, instance->memb_timer_state_gather_join_timeout); poll_timer_add (instance->totemsrp_poll_handle, instance->totem_config->join_timeout, (void *)instance, memb_timer_function_state_gather, &instance->memb_timer_state_gather_join_timeout); break; } } static void memb_timer_function_gather_consensus_timeout (void *data) { struct totemsrp_instance *instance = (struct totemsrp_instance *)data; memb_state_consensus_timeout_expired (instance); } static void deliver_messages_from_recovery_to_regular (struct totemsrp_instance *instance) { unsigned int i; struct sort_queue_item *recovery_message_item; struct sort_queue_item regular_message_item; unsigned int range = 0; int res; void *ptr; struct mcast *mcast; log_printf (instance->totemsrp_log_level_debug, "recovery to regular %x-%x\n", SEQNO_START_MSG + 1, instance->my_aru); range = instance->my_aru - SEQNO_START_MSG; /* * Move messages from recovery to regular sort queue */ // todo should i be initialized to 0 or 1 ? for (i = 1; i <= range; i++) { res = sq_item_get (&instance->recovery_sort_queue, i + SEQNO_START_MSG, &ptr); if (res != 0) { continue; } recovery_message_item = (struct sort_queue_item *)ptr; /* * Convert recovery message into regular message */ if (recovery_message_item->iov_len > 1) { mcast = recovery_message_item->iovec[1].iov_base; memcpy (®ular_message_item.iovec[0], &recovery_message_item->iovec[1], sizeof (struct iovec) * recovery_message_item->iov_len); } else { mcast = recovery_message_item->iovec[0].iov_base; if (mcast->header.encapsulated == MESSAGE_ENCAPSULATED) { /* * Message is a recovery message encapsulated * in a new ring message */ regular_message_item.iovec[0].iov_base = (char *)recovery_message_item->iovec[0].iov_base + sizeof (struct mcast); regular_message_item.iovec[0].iov_len = recovery_message_item->iovec[0].iov_len - sizeof (struct mcast); regular_message_item.iov_len = 1; mcast = regular_message_item.iovec[0].iov_base; } else { continue; /* TODO this case shouldn't happen */ /* * Message is originated on new ring and not * encapsulated */ regular_message_item.iovec[0].iov_base = recovery_message_item->iovec[0].iov_base; regular_message_item.iovec[0].iov_len = recovery_message_item->iovec[0].iov_len; } } log_printf (instance->totemsrp_log_level_debug, "comparing if ring id is for this processors old ring seqno %d\n", mcast->seq); /* * Only add this message to the regular sort * queue if it was originated with the same ring * id as the previous ring */ if (memcmp (&instance->my_old_ring_id, &mcast->ring_id, sizeof (struct memb_ring_id)) == 0) { regular_message_item.iov_len = recovery_message_item->iov_len; res = sq_item_inuse (&instance->regular_sort_queue, mcast->seq); if (res == 0) { sq_item_add (&instance->regular_sort_queue, ®ular_message_item, mcast->seq); if (sq_lt_compare (instance->old_ring_state_high_seq_received, mcast->seq)) { instance->old_ring_state_high_seq_received = mcast->seq; } } } else { log_printf (instance->totemsrp_log_level_notice, "-not adding msg with seq no %x\n", mcast->seq); } } } /* * Change states in the state machine of the membership algorithm */ static void memb_state_operational_enter (struct totemsrp_instance *instance) { struct srp_addr joined_list[PROCESSOR_COUNT_MAX]; int joined_list_entries = 0; unsigned int aru_save; unsigned int joined_list_totemip[PROCESSOR_COUNT_MAX]; unsigned int trans_memb_list_totemip[PROCESSOR_COUNT_MAX]; unsigned int new_memb_list_totemip[PROCESSOR_COUNT_MAX]; unsigned int left_list[PROCESSOR_COUNT_MAX]; memb_consensus_reset (instance); old_ring_state_reset (instance); ring_reset (instance); deliver_messages_from_recovery_to_regular (instance); log_printf (instance->totemsrp_log_level_debug, "Delivering to app %x to %x\n", instance->my_high_delivered + 1, instance->old_ring_state_high_seq_received); aru_save = instance->my_aru; instance->my_aru = instance->old_ring_state_aru; messages_deliver_to_app (instance, 0, instance->old_ring_state_high_seq_received); /* * Calculate joined and left list */ memb_set_subtract (instance->my_left_memb_list, &instance->my_left_memb_entries, instance->my_memb_list, instance->my_memb_entries, instance->my_trans_memb_list, instance->my_trans_memb_entries); memb_set_subtract (joined_list, &joined_list_entries, instance->my_new_memb_list, instance->my_new_memb_entries, instance->my_trans_memb_list, instance->my_trans_memb_entries); /* * Install new membership */ instance->my_memb_entries = instance->my_new_memb_entries; memcpy (&instance->my_memb_list, instance->my_new_memb_list, sizeof (struct srp_addr) * instance->my_memb_entries); instance->last_released = 0; instance->my_set_retrans_flg = 0; /* * Deliver transitional configuration to application */ srp_addr_to_nodeid (left_list, instance->my_left_memb_list, instance->my_left_memb_entries); srp_addr_to_nodeid (trans_memb_list_totemip, instance->my_trans_memb_list, instance->my_trans_memb_entries); instance->totemsrp_confchg_fn (TOTEM_CONFIGURATION_TRANSITIONAL, trans_memb_list_totemip, instance->my_trans_memb_entries, left_list, instance->my_left_memb_entries, 0, 0, &instance->my_ring_id); // TODO we need to filter to ensure we only deliver those // messages which are part of instance->my_deliver_memb messages_deliver_to_app (instance, 1, instance->old_ring_state_high_seq_received); instance->my_aru = aru_save; /* * Deliver regular configuration to application */ srp_addr_to_nodeid (new_memb_list_totemip, instance->my_new_memb_list, instance->my_new_memb_entries); srp_addr_to_nodeid (joined_list_totemip, joined_list, joined_list_entries); instance->totemsrp_confchg_fn (TOTEM_CONFIGURATION_REGULAR, new_memb_list_totemip, instance->my_new_memb_entries, 0, 0, joined_list_totemip, joined_list_entries, &instance->my_ring_id); /* * The recovery sort queue now becomes the regular * sort queue. It is necessary to copy the state * into the regular sort queue. */ sq_copy (&instance->regular_sort_queue, &instance->recovery_sort_queue); instance->my_last_aru = SEQNO_START_MSG; sq_items_release (&instance->regular_sort_queue, SEQNO_START_MSG - 1); /* When making my_proc_list smaller, ensure that the * now non-used entries are zero-ed out. There are some suspect * assert's that assume that there is always 2 entries in the list. * These fail when my_proc_list is reduced to 1 entry (and the * valid [0] entry is the same as the 'unused' [1] entry). */ memset(instance->my_proc_list, 0, sizeof (struct srp_addr) * instance->my_proc_list_entries); instance->my_proc_list_entries = instance->my_new_memb_entries; memcpy (instance->my_proc_list, instance->my_new_memb_list, sizeof (struct srp_addr) * instance->my_memb_entries); instance->my_failed_list_entries = 0; instance->my_high_delivered = instance->my_aru; // TODO the recovery messages are leaked log_printf (instance->totemsrp_log_level_notice, "entering OPERATIONAL state.\n"); instance->memb_state = MEMB_STATE_OPERATIONAL; instance->my_received_flg = 1; return; } static void memb_state_gather_enter ( struct totemsrp_instance *instance, int gather_from) { memb_set_merge ( &instance->my_id, 1, instance->my_proc_list, &instance->my_proc_list_entries); assert (srp_addr_equal (&instance->my_proc_list[0], &instance->my_proc_list[1]) == 0); memb_join_message_send (instance); /* * Restart the join timeout */ poll_timer_delete (instance->totemsrp_poll_handle, instance->memb_timer_state_gather_join_timeout); poll_timer_add (instance->totemsrp_poll_handle, instance->totem_config->join_timeout, (void *)instance, memb_timer_function_state_gather, &instance->memb_timer_state_gather_join_timeout); /* * Restart the consensus timeout */ poll_timer_delete (instance->totemsrp_poll_handle, instance->memb_timer_state_gather_consensus_timeout); poll_timer_add (instance->totemsrp_poll_handle, instance->totem_config->consensus_timeout, (void *)instance, memb_timer_function_gather_consensus_timeout, &instance->memb_timer_state_gather_consensus_timeout); /* * Cancel the token loss and token retransmission timeouts */ cancel_token_retransmit_timeout (instance); // REVIEWED cancel_token_timeout (instance); // REVIEWED cancel_merge_detect_timeout (instance); memb_consensus_reset (instance); memb_consensus_set (instance, &instance->my_id); log_printf (instance->totemsrp_log_level_notice, "entering GATHER state from %d.\n", gather_from); instance->memb_state = MEMB_STATE_GATHER; return; } static void timer_function_token_retransmit_timeout (void *data); static void memb_state_commit_enter ( struct totemsrp_instance *instance, struct memb_commit_token *commit_token) { ring_save (instance); old_ring_state_save (instance); memb_state_commit_token_update (instance, commit_token); memb_state_commit_token_target_set (instance, commit_token); memb_ring_id_set_and_store (instance, &commit_token->ring_id); memb_state_commit_token_send (instance, commit_token); instance->token_ring_id_seq = instance->my_ring_id.seq; poll_timer_delete (instance->totemsrp_poll_handle, instance->memb_timer_state_gather_join_timeout); instance->memb_timer_state_gather_join_timeout = 0; poll_timer_delete (instance->totemsrp_poll_handle, instance->memb_timer_state_gather_consensus_timeout); instance->memb_timer_state_gather_consensus_timeout = 0; reset_token_timeout (instance); // REVIEWED reset_token_retransmit_timeout (instance); // REVIEWED log_printf (instance->totemsrp_log_level_notice, "entering COMMIT state.\n"); instance->memb_state = MEMB_STATE_COMMIT; /* * reset all flow control variables since we are starting a new ring */ instance->my_trc = 0; instance->my_pbl = 0; instance->my_cbl = 0; return; } static void memb_state_recovery_enter ( struct totemsrp_instance *instance, struct memb_commit_token *commit_token) { int i; int local_received_flg = 1; unsigned int low_ring_aru; unsigned int range = 0; unsigned int messages_originated = 0; char is_originated[4096]; char not_originated[4096]; char seqno_string_hex[10]; struct srp_addr *addr; struct memb_commit_token_memb_entry *memb_list; addr = (struct srp_addr *)commit_token->end_of_commit_token; memb_list = (struct memb_commit_token_memb_entry *)(addr + commit_token->addr_entries); log_printf (instance->totemsrp_log_level_notice, "entering RECOVERY state.\n"); instance->my_high_ring_delivered = 0; sq_reinit (&instance->recovery_sort_queue, SEQNO_START_MSG); queue_reinit (&instance->retrans_message_queue); low_ring_aru = instance->old_ring_state_high_seq_received; memb_state_commit_token_send (instance, commit_token); instance->my_token_seq = SEQNO_START_TOKEN - 1; /* * Build regular configuration */ totemrrp_processor_count_set ( instance->totemrrp_handle, commit_token->addr_entries); /* * Build transitional configuration */ memb_set_and (instance->my_new_memb_list, instance->my_new_memb_entries, instance->my_memb_list, instance->my_memb_entries, instance->my_trans_memb_list, &instance->my_trans_memb_entries); for (i = 0; i < instance->my_new_memb_entries; i++) { log_printf (instance->totemsrp_log_level_notice, "position [%d] member %s:\n", i, totemip_print (&addr[i].addr[0])); log_printf (instance->totemsrp_log_level_notice, "previous ring seq %lld rep %s\n", memb_list[i].ring_id.seq, totemip_print (&memb_list[i].ring_id.rep)); log_printf (instance->totemsrp_log_level_notice, "aru %x high delivered %x received flag %d\n", memb_list[i].aru, memb_list[i].high_delivered, memb_list[i].received_flg); // assert (totemip_print (&memb_list[i].ring_id.rep) != 0); } /* * Determine if any received flag is false */ for (i = 0; i < commit_token->addr_entries; i++) { if (memb_set_subset (&instance->my_new_memb_list[i], 1, instance->my_trans_memb_list, instance->my_trans_memb_entries) && memb_list[i].received_flg == 0) { instance->my_deliver_memb_entries = instance->my_trans_memb_entries; memcpy (instance->my_deliver_memb_list, instance->my_trans_memb_list, sizeof (struct srp_addr) * instance->my_trans_memb_entries); local_received_flg = 0; break; } } if (local_received_flg == 1) { goto no_originate; } /* Else originate messages if we should */ /* * Calculate my_low_ring_aru, instance->my_high_ring_delivered for the transitional membership */ for (i = 0; i < commit_token->addr_entries; i++) { if (memb_set_subset (&instance->my_new_memb_list[i], 1, instance->my_deliver_memb_list, instance->my_deliver_memb_entries) && memcmp (&instance->my_old_ring_id, &memb_list[i].ring_id, sizeof (struct memb_ring_id)) == 0) { if (sq_lt_compare (memb_list[i].aru, low_ring_aru)) { low_ring_aru = memb_list[i].aru; } if (sq_lt_compare (instance->my_high_ring_delivered, memb_list[i].high_delivered)) { instance->my_high_ring_delivered = memb_list[i].high_delivered; } } } /* * Copy all old ring messages to instance->retrans_message_queue */ range = instance->old_ring_state_high_seq_received - low_ring_aru; if (range == 0) { /* * No messages to copy */ goto no_originate; } assert (range < 1024); log_printf (instance->totemsrp_log_level_notice, "copying all old ring messages from %x-%x.\n", low_ring_aru + 1, instance->old_ring_state_high_seq_received); strcpy (not_originated, "Not Originated for recovery: "); strcpy (is_originated, "Originated for recovery: "); for (i = 1; i <= range; i++) { struct sort_queue_item *sort_queue_item; struct message_item message_item; void *ptr; int res; sprintf (seqno_string_hex, "%x ", low_ring_aru + i); res = sq_item_get (&instance->regular_sort_queue, low_ring_aru + i, &ptr); if (res != 0) { strcat (not_originated, seqno_string_hex); continue; } strcat (is_originated, seqno_string_hex); sort_queue_item = ptr; assert (sort_queue_item->iov_len > 0); assert (sort_queue_item->iov_len <= MAXIOVS); messages_originated++; memset (&message_item, 0, sizeof (struct message_item)); // TODO LEAK message_item.mcast = malloc (sizeof (struct mcast)); assert (message_item.mcast); message_item.mcast->header.type = MESSAGE_TYPE_MCAST; srp_addr_copy (&message_item.mcast->system_from, &instance->my_id); message_item.mcast->header.encapsulated = MESSAGE_ENCAPSULATED; message_item.mcast->header.nodeid = instance->my_id.addr[0].nodeid; assert (message_item.mcast->header.nodeid); message_item.mcast->header.endian_detector = ENDIAN_LOCAL; memcpy (&message_item.mcast->ring_id, &instance->my_ring_id, sizeof (struct memb_ring_id)); message_item.iov_len = sort_queue_item->iov_len; memcpy (&message_item.iovec, &sort_queue_item->iovec, sizeof (struct iovec) * sort_queue_item->iov_len); queue_item_add (&instance->retrans_message_queue, &message_item); } log_printf (instance->totemsrp_log_level_notice, "Originated %d messages in RECOVERY.\n", messages_originated); strcat (not_originated, "\n"); strcat (is_originated, "\n"); log_printf (instance->totemsrp_log_level_notice, "%s", is_originated); log_printf (instance->totemsrp_log_level_notice, "%s", not_originated); goto originated; no_originate: log_printf (instance->totemsrp_log_level_notice, "Did not need to originate any messages in recovery.\n"); originated: instance->my_aru = SEQNO_START_MSG; instance->my_aru_count = 0; instance->my_seq_unchanged = 0; instance->my_high_seq_received = SEQNO_START_MSG; instance->my_install_seq = SEQNO_START_MSG; instance->last_released = SEQNO_START_MSG; reset_token_timeout (instance); // REVIEWED reset_token_retransmit_timeout (instance); // REVIEWED instance->memb_state = MEMB_STATE_RECOVERY; return; } int totemsrp_new_msg_signal (hdb_handle_t handle) { struct totemsrp_instance *instance; unsigned int res; res = hdb_handle_get (&totemsrp_instance_database, handle, (void *)&instance); if (res != 0) { goto error_exit; } token_hold_cancel_send (instance); hdb_handle_put (&totemsrp_instance_database, handle); return (0); error_exit: return (-1); } int totemsrp_mcast ( hdb_handle_t handle, struct iovec *iovec, unsigned int iov_len, int guarantee) { int i; int j; struct message_item message_item; struct totemsrp_instance *instance; unsigned int res; res = hdb_handle_get (&totemsrp_instance_database, handle, (void *)&instance); if (res != 0) { goto error_exit; } if (queue_is_full (&instance->new_message_queue)) { log_printf (instance->totemsrp_log_level_warning, "queue full\n"); return (-1); } for (j = 0, i = 0; i < iov_len; i++) { j+= iovec[i].iov_len; } memset (&message_item, 0, sizeof (struct message_item)); /* * Allocate pending item */ // TODO LEAK message_item.mcast = malloc (sizeof (struct mcast)); if (message_item.mcast == 0) { goto error_mcast; } /* * Set mcast header */ message_item.mcast->header.type = MESSAGE_TYPE_MCAST; message_item.mcast->header.endian_detector = ENDIAN_LOCAL; message_item.mcast->header.encapsulated = MESSAGE_NOT_ENCAPSULATED; message_item.mcast->header.nodeid = instance->my_id.addr[0].nodeid; assert (message_item.mcast->header.nodeid); message_item.mcast->guarantee = guarantee; srp_addr_copy (&message_item.mcast->system_from, &instance->my_id); for (i = 0; i < iov_len; i++) { // TODO LEAK message_item.iovec[i].iov_base = malloc (iovec[i].iov_len); if (message_item.iovec[i].iov_base == 0) { goto error_iovec; } memcpy (message_item.iovec[i].iov_base, iovec[i].iov_base, iovec[i].iov_len); message_item.iovec[i].iov_len = iovec[i].iov_len; } message_item.iov_len = iov_len; log_printf (instance->totemsrp_log_level_debug, "mcasted message added to pending queue\n"); queue_item_add (&instance->new_message_queue, &message_item); hdb_handle_put (&totemsrp_instance_database, handle); return (0); error_iovec: for (j = 0; j < i; j++) { free (message_item.iovec[j].iov_base); } free(message_item.mcast); error_mcast: hdb_handle_put (&totemsrp_instance_database, handle); error_exit: return (-1); } /* * Determine if there is room to queue a new message */ int totemsrp_avail (hdb_handle_t handle) { int avail; struct totemsrp_instance *instance; unsigned int res; res = hdb_handle_get (&totemsrp_instance_database, handle, (void *)&instance); if (res != 0) { goto error_exit; } queue_avail (&instance->new_message_queue, &avail); hdb_handle_put (&totemsrp_instance_database, handle); return (avail); error_exit: return (0); } /* * ORF Token Management */ /* * Recast message to mcast group if it is available */ static int orf_token_remcast ( struct totemsrp_instance *instance, int seq) { struct sort_queue_item *sort_queue_item; int res; void *ptr; struct sq *sort_queue; if (instance->memb_state == MEMB_STATE_RECOVERY) { sort_queue = &instance->recovery_sort_queue; } else { sort_queue = &instance->regular_sort_queue; } res = sq_in_range (sort_queue, seq); if (res == 0) { log_printf (instance->totemsrp_log_level_debug, "sq not in range\n"); return (-1); } /* * Get RTR item at seq, if not available, return */ res = sq_item_get (sort_queue, seq, &ptr); if (res != 0) { return -1; } sort_queue_item = ptr; totemrrp_mcast_noflush_send (instance->totemrrp_handle, sort_queue_item->iovec, sort_queue_item->iov_len); return (0); } /* * Free all freeable messages from ring */ static void messages_free ( struct totemsrp_instance *instance, unsigned int token_aru) { struct sort_queue_item *regular_message; unsigned int i, j; int res; int log_release = 0; unsigned int release_to; unsigned int range = 0; release_to = token_aru; if (sq_lt_compare (instance->my_last_aru, release_to)) { release_to = instance->my_last_aru; } if (sq_lt_compare (instance->my_high_delivered, release_to)) { release_to = instance->my_high_delivered; } /* * Ensure we dont try release before an already released point */ if (sq_lt_compare (release_to, instance->last_released)) { return; } range = release_to - instance->last_released; assert (range < 1024); /* * Release retransmit list items if group aru indicates they are transmitted */ for (i = 1; i <= range; i++) { void *ptr; res = sq_item_get (&instance->regular_sort_queue, instance->last_released + i, &ptr); if (res == 0) { regular_message = ptr; for (j = 0; j < regular_message->iov_len; j++) { free (regular_message->iovec[j].iov_base); } } sq_items_release (&instance->regular_sort_queue, instance->last_released + i); log_release = 1; } instance->last_released += range; if (log_release) { log_printf (instance->totemsrp_log_level_debug, "releasing messages up to and including %x\n", release_to); } } static void update_aru ( struct totemsrp_instance *instance) { unsigned int i; int res; struct sq *sort_queue; unsigned int range; unsigned int my_aru_saved = 0; if (instance->memb_state == MEMB_STATE_RECOVERY) { sort_queue = &instance->recovery_sort_queue; } else { sort_queue = &instance->regular_sort_queue; } range = instance->my_high_seq_received - instance->my_aru; if (range > 1024) { return; } my_aru_saved = instance->my_aru; for (i = 1; i <= range; i++) { void *ptr; res = sq_item_get (sort_queue, my_aru_saved + i, &ptr); /* * If hole, stop updating aru */ if (res != 0) { break; } } instance->my_aru += i - 1; } /* * Multicasts pending messages onto the ring (requires orf_token possession) */ static int orf_token_mcast ( struct totemsrp_instance *instance, struct orf_token *token, int fcc_mcasts_allowed) { struct message_item *message_item = 0; struct queue *mcast_queue; struct sq *sort_queue; struct sort_queue_item sort_queue_item; struct sort_queue_item *sort_queue_item_ptr; struct mcast *mcast; unsigned int fcc_mcast_current; if (instance->memb_state == MEMB_STATE_RECOVERY) { mcast_queue = &instance->retrans_message_queue; sort_queue = &instance->recovery_sort_queue; reset_token_retransmit_timeout (instance); // REVIEWED } else { mcast_queue = &instance->new_message_queue; sort_queue = &instance->regular_sort_queue; } for (fcc_mcast_current = 0; fcc_mcast_current < fcc_mcasts_allowed; fcc_mcast_current++) { if (queue_is_empty (mcast_queue)) { break; } message_item = (struct message_item *)queue_item_get (mcast_queue); /* preincrement required by algo */ if (instance->old_ring_state_saved && (instance->memb_state == MEMB_STATE_GATHER || instance->memb_state == MEMB_STATE_COMMIT)) { log_printf (instance->totemsrp_log_level_debug, "not multicasting at seqno is %d\n", token->seq); return (0); } message_item->mcast->seq = ++token->seq; message_item->mcast->this_seqno = instance->global_seqno++; /* * Build IO vector */ memset (&sort_queue_item, 0, sizeof (struct sort_queue_item)); sort_queue_item.iovec[0].iov_base = message_item->mcast; sort_queue_item.iovec[0].iov_len = sizeof (struct mcast); mcast = sort_queue_item.iovec[0].iov_base; memcpy (&sort_queue_item.iovec[1], message_item->iovec, message_item->iov_len * sizeof (struct iovec)); memcpy (&mcast->ring_id, &instance->my_ring_id, sizeof (struct memb_ring_id)); sort_queue_item.iov_len = message_item->iov_len + 1; assert (sort_queue_item.iov_len < 16); /* * Add message to retransmit queue */ sort_queue_item_ptr = sq_item_add (sort_queue, &sort_queue_item, message_item->mcast->seq); totemrrp_mcast_noflush_send (instance->totemrrp_handle, sort_queue_item_ptr->iovec, sort_queue_item_ptr->iov_len); /* * Delete item from pending queue */ queue_item_remove (mcast_queue); /* * If messages mcasted, deliver any new messages to totempg */ instance->my_high_seq_received = token->seq; } update_aru (instance); /* * Return 1 if more messages are available for single node clusters */ return (fcc_mcast_current); } /* * Remulticasts messages in orf_token's retransmit list (requires orf_token) * Modify's orf_token's rtr to include retransmits required by this process */ static int orf_token_rtr ( struct totemsrp_instance *instance, struct orf_token *orf_token, unsigned int *fcc_allowed) { unsigned int res; unsigned int i, j; unsigned int found; unsigned int total_entries; struct sq *sort_queue; struct rtr_item *rtr_list; unsigned int range = 0; char retransmit_msg[1024]; char value[64]; if (instance->memb_state == MEMB_STATE_RECOVERY) { sort_queue = &instance->recovery_sort_queue; } else { sort_queue = &instance->regular_sort_queue; } rtr_list = &orf_token->rtr_list[0]; strcpy (retransmit_msg, "Retransmit List: "); if (orf_token->rtr_list_entries) { log_printf (instance->totemsrp_log_level_debug, "Retransmit List %d\n", orf_token->rtr_list_entries); for (i = 0; i < orf_token->rtr_list_entries; i++) { sprintf (value, "%x ", rtr_list[i].seq); strcat (retransmit_msg, value); } strcat (retransmit_msg, "\n"); log_printf (instance->totemsrp_log_level_notice, "%s", retransmit_msg); } total_entries = orf_token->rtr_list_entries; /* * Retransmit messages on orf_token's RTR list from RTR queue */ for (instance->fcc_remcast_current = 0, i = 0; instance->fcc_remcast_current < *fcc_allowed && i < orf_token->rtr_list_entries;) { /* * If this retransmit request isn't from this configuration, * try next rtr entry */ if (memcmp (&rtr_list[i].ring_id, &instance->my_ring_id, sizeof (struct memb_ring_id)) != 0) { i += 1; continue; } res = orf_token_remcast (instance, rtr_list[i].seq); if (res == 0) { /* * Multicasted message, so no need to copy to new retransmit list */ orf_token->rtr_list_entries -= 1; assert (orf_token->rtr_list_entries >= 0); memmove (&rtr_list[i], &rtr_list[i + 1], sizeof (struct rtr_item) * (orf_token->rtr_list_entries)); instance->fcc_remcast_current++; } else { i += 1; } } *fcc_allowed = *fcc_allowed - instance->fcc_remcast_current; /* * Add messages to retransmit to RTR list * but only retry if there is room in the retransmit list */ range = instance->my_high_seq_received - instance->my_aru; assert (range < 100000); for (i = 1; (orf_token->rtr_list_entries < RETRANSMIT_ENTRIES_MAX) && (i <= range); i++) { /* * Ensure message is within the sort queue range */ res = sq_in_range (sort_queue, instance->my_aru + i); if (res == 0) { break; } /* * Find if a message is missing from this processor */ res = sq_item_inuse (sort_queue, instance->my_aru + i); if (res == 0) { /* * Determine if missing message is already in retransmit list */ found = 0; for (j = 0; j < orf_token->rtr_list_entries; j++) { if (instance->my_aru + i == rtr_list[j].seq) { found = 1; } } if (found == 0) { /* * Missing message not found in current retransmit list so add it */ memcpy (&rtr_list[orf_token->rtr_list_entries].ring_id, &instance->my_ring_id, sizeof (struct memb_ring_id)); rtr_list[orf_token->rtr_list_entries].seq = instance->my_aru + i; orf_token->rtr_list_entries++; } } } return (instance->fcc_remcast_current); } static void token_retransmit (struct totemsrp_instance *instance) { struct iovec iovec; iovec.iov_base = instance->orf_token_retransmit; iovec.iov_len = instance->orf_token_retransmit_size; totemrrp_token_send (instance->totemrrp_handle, &iovec, 1); } /* * Retransmit the regular token if no mcast or token has * been received in retransmit token period retransmit * the token to the next processor */ static void timer_function_token_retransmit_timeout (void *data) { struct totemsrp_instance *instance = (struct totemsrp_instance *)data; switch (instance->memb_state) { case MEMB_STATE_GATHER: break; case MEMB_STATE_COMMIT: case MEMB_STATE_OPERATIONAL: case MEMB_STATE_RECOVERY: token_retransmit (instance); reset_token_retransmit_timeout (instance); // REVIEWED break; } } static void timer_function_token_hold_retransmit_timeout (void *data) { struct totemsrp_instance *instance = (struct totemsrp_instance *)data; switch (instance->memb_state) { case MEMB_STATE_GATHER: break; case MEMB_STATE_COMMIT: break; case MEMB_STATE_OPERATIONAL: case MEMB_STATE_RECOVERY: token_retransmit (instance); break; } } static void timer_function_merge_detect_timeout(void *data) { struct totemsrp_instance *instance = (struct totemsrp_instance *)data; instance->my_merge_detect_timeout_outstanding = 0; switch (instance->memb_state) { case MEMB_STATE_OPERATIONAL: if (totemip_equal(&instance->my_ring_id.rep, &instance->my_id.addr[0])) { memb_merge_detect_transmit (instance); } break; case MEMB_STATE_GATHER: case MEMB_STATE_COMMIT: case MEMB_STATE_RECOVERY: break; } } /* * Send orf_token to next member (requires orf_token) */ static int token_send ( struct totemsrp_instance *instance, struct orf_token *orf_token, int forward_token) { struct iovec iovec; int res = 0; unsigned int iov_len = sizeof (struct orf_token) + (orf_token->rtr_list_entries * sizeof (struct rtr_item)); memcpy (instance->orf_token_retransmit, orf_token, iov_len); instance->orf_token_retransmit_size = iov_len; orf_token->header.nodeid = instance->my_id.addr[0].nodeid; assert (orf_token->header.nodeid); if (forward_token == 0) { return (0); } iovec.iov_base = orf_token; iovec.iov_len = iov_len; totemrrp_token_send (instance->totemrrp_handle, &iovec, 1); return (res); } static int token_hold_cancel_send (struct totemsrp_instance *instance) { struct token_hold_cancel token_hold_cancel; struct iovec iovec[2]; /* * Only cancel if the token is currently held */ if (instance->my_token_held == 0) { return (0); } instance->my_token_held = 0; /* * Build message */ token_hold_cancel.header.type = MESSAGE_TYPE_TOKEN_HOLD_CANCEL; token_hold_cancel.header.endian_detector = ENDIAN_LOCAL; token_hold_cancel.header.nodeid = instance->my_id.addr[0].nodeid; assert (token_hold_cancel.header.nodeid); iovec[0].iov_base = &token_hold_cancel; iovec[0].iov_len = sizeof (struct token_hold_cancel) - sizeof (struct memb_ring_id); iovec[1].iov_base = &instance->my_ring_id; iovec[1].iov_len = sizeof (struct memb_ring_id); totemrrp_mcast_flush_send (instance->totemrrp_handle, iovec, 2); return (0); } //AAA static int orf_token_send_initial (struct totemsrp_instance *instance) { struct orf_token orf_token; int res; orf_token.header.type = MESSAGE_TYPE_ORF_TOKEN; orf_token.header.endian_detector = ENDIAN_LOCAL; orf_token.header.encapsulated = 0; orf_token.header.nodeid = instance->my_id.addr[0].nodeid; assert (orf_token.header.nodeid); orf_token.seq = SEQNO_START_MSG; orf_token.token_seq = SEQNO_START_TOKEN; orf_token.retrans_flg = 1; instance->my_set_retrans_flg = 1; if (queue_is_empty (&instance->retrans_message_queue) == 1) { orf_token.retrans_flg = 0; instance->my_set_retrans_flg = 0; } else { orf_token.retrans_flg = 1; instance->my_set_retrans_flg = 1; } orf_token.aru = 0; orf_token.aru = SEQNO_START_MSG - 1; orf_token.aru_addr = instance->my_id.addr[0].nodeid; memcpy (&orf_token.ring_id, &instance->my_ring_id, sizeof (struct memb_ring_id)); orf_token.fcc = 0; orf_token.backlog = 0; orf_token.rtr_list_entries = 0; res = token_send (instance, &orf_token, 1); return (res); } static void memb_state_commit_token_update ( struct totemsrp_instance *instance, struct memb_commit_token *commit_token) { struct srp_addr *addr; struct memb_commit_token_memb_entry *memb_list; unsigned int high_aru; unsigned int i; addr = (struct srp_addr *)commit_token->end_of_commit_token; memb_list = (struct memb_commit_token_memb_entry *)(addr + commit_token->addr_entries); memcpy (instance->my_new_memb_list, addr, sizeof (struct srp_addr) * commit_token->addr_entries); instance->my_new_memb_entries = commit_token->addr_entries; memcpy (&memb_list[commit_token->memb_index].ring_id, &instance->my_old_ring_id, sizeof (struct memb_ring_id)); assert (!totemip_zero_check(&instance->my_old_ring_id.rep)); memb_list[commit_token->memb_index].aru = instance->old_ring_state_aru; /* * TODO high delivered is really instance->my_aru, but with safe this * could change? */ instance->my_received_flg = (instance->my_aru == instance->my_high_seq_received); memb_list[commit_token->memb_index].received_flg = instance->my_received_flg; memb_list[commit_token->memb_index].high_delivered = instance->my_high_delivered; /* * find high aru up to current memb_index for all matching ring ids * if any ring id matching memb_index has aru less then high aru set * received flag for that entry to false */ high_aru = memb_list[commit_token->memb_index].aru; for (i = 0; i <= commit_token->memb_index; i++) { if (memcmp (&memb_list[commit_token->memb_index].ring_id, &memb_list[i].ring_id, sizeof (struct memb_ring_id)) == 0) { if (sq_lt_compare (high_aru, memb_list[i].aru)) { high_aru = memb_list[i].aru; } } } for (i = 0; i <= commit_token->memb_index; i++) { if (memcmp (&memb_list[commit_token->memb_index].ring_id, &memb_list[i].ring_id, sizeof (struct memb_ring_id)) == 0) { if (sq_lt_compare (memb_list[i].aru, high_aru)) { memb_list[i].received_flg = 0; if (i == commit_token->memb_index) { instance->my_received_flg = 0; } } } } commit_token->header.nodeid = instance->my_id.addr[0].nodeid; commit_token->memb_index += 1; assert (commit_token->memb_index <= commit_token->addr_entries); assert (commit_token->header.nodeid); } static void memb_state_commit_token_target_set ( struct totemsrp_instance *instance, struct memb_commit_token *commit_token) { struct srp_addr *addr; unsigned int i; addr = (struct srp_addr *)commit_token->end_of_commit_token; for (i = 0; i < instance->totem_config->interface_count; i++) { totemrrp_token_target_set ( instance->totemrrp_handle, &addr[commit_token->memb_index % commit_token->addr_entries].addr[i], i); } } static int memb_state_commit_token_send ( struct totemsrp_instance *instance, struct memb_commit_token *commit_token) { struct iovec iovec; struct srp_addr *addr; struct memb_commit_token_memb_entry *memb_list; addr = (struct srp_addr *)commit_token->end_of_commit_token; memb_list = (struct memb_commit_token_memb_entry *)(addr + commit_token->addr_entries); commit_token->token_seq++; iovec.iov_base = commit_token; iovec.iov_len = sizeof (struct memb_commit_token) + ((sizeof (struct srp_addr) + sizeof (struct memb_commit_token_memb_entry)) * commit_token->addr_entries); /* * Make a copy for retransmission if necessary */ memcpy (instance->orf_token_retransmit, commit_token, iovec.iov_len); instance->orf_token_retransmit_size = iovec.iov_len; totemrrp_token_send (instance->totemrrp_handle, &iovec, 1); /* * Request retransmission of the commit token in case it is lost */ reset_token_retransmit_timeout (instance); return (0); } static int memb_lowest_in_config (struct totemsrp_instance *instance) { struct srp_addr token_memb[PROCESSOR_COUNT_MAX]; int token_memb_entries = 0; int i; struct totem_ip_address *lowest_addr; memb_set_subtract (token_memb, &token_memb_entries, instance->my_proc_list, instance->my_proc_list_entries, instance->my_failed_list, instance->my_failed_list_entries); /* * find representative by searching for smallest identifier */ lowest_addr = &token_memb[0].addr[0]; for (i = 1; i < token_memb_entries; i++) { if (totemip_compare(lowest_addr, &token_memb[i].addr[0]) > 0) { totemip_copy (lowest_addr, &token_memb[i].addr[0]); } } return (totemip_compare (lowest_addr, &instance->my_id.addr[0]) == 0); } static int srp_addr_compare (const void *a, const void *b) { const struct srp_addr *srp_a = (const struct srp_addr *)a; const struct srp_addr *srp_b = (const struct srp_addr *)b; return (totemip_compare (&srp_a->addr[0], &srp_b->addr[0])); } static void memb_state_commit_token_create ( struct totemsrp_instance *instance, struct memb_commit_token *commit_token) { struct srp_addr token_memb[PROCESSOR_COUNT_MAX]; struct srp_addr *addr; struct memb_commit_token_memb_entry *memb_list; int token_memb_entries = 0; log_printf (instance->totemsrp_log_level_notice, "Creating commit token because I am the rep.\n"); memb_set_subtract (token_memb, &token_memb_entries, instance->my_proc_list, instance->my_proc_list_entries, instance->my_failed_list, instance->my_failed_list_entries); memset (commit_token, 0, sizeof (struct memb_commit_token)); commit_token->header.type = MESSAGE_TYPE_MEMB_COMMIT_TOKEN; commit_token->header.endian_detector = ENDIAN_LOCAL; commit_token->header.encapsulated = 0; commit_token->header.nodeid = instance->my_id.addr[0].nodeid; assert (commit_token->header.nodeid); totemip_copy(&commit_token->ring_id.rep, &instance->my_id.addr[0]); commit_token->ring_id.seq = instance->token_ring_id_seq + 4; /* * This qsort is necessary to ensure the commit token traverses * the ring in the proper order */ qsort (token_memb, token_memb_entries, sizeof (struct srp_addr), srp_addr_compare); commit_token->memb_index = 0; commit_token->addr_entries = token_memb_entries; addr = (struct srp_addr *)commit_token->end_of_commit_token; memb_list = (struct memb_commit_token_memb_entry *)(addr + commit_token->addr_entries); memcpy (addr, token_memb, token_memb_entries * sizeof (struct srp_addr)); memset (memb_list, 0, sizeof (struct memb_commit_token_memb_entry) * token_memb_entries); } static void memb_join_message_send (struct totemsrp_instance *instance) { struct memb_join memb_join; struct iovec iovec[3]; unsigned int iovs; memb_join.header.type = MESSAGE_TYPE_MEMB_JOIN; memb_join.header.endian_detector = ENDIAN_LOCAL; memb_join.header.encapsulated = 0; memb_join.header.nodeid = instance->my_id.addr[0].nodeid; assert (memb_join.header.nodeid); assert (srp_addr_equal (&instance->my_proc_list[0], &instance->my_proc_list[1]) == 0); memb_join.ring_seq = instance->my_ring_id.seq; memb_join.proc_list_entries = instance->my_proc_list_entries; memb_join.failed_list_entries = instance->my_failed_list_entries; srp_addr_copy (&memb_join.system_from, &instance->my_id); iovec[0].iov_base = &memb_join; iovec[0].iov_len = sizeof (struct memb_join); iovec[1].iov_base = &instance->my_proc_list; iovec[1].iov_len = instance->my_proc_list_entries * sizeof (struct srp_addr); if (instance->my_failed_list_entries == 0) { iovs = 2; } else { iovs = 3; iovec[2].iov_base = instance->my_failed_list; iovec[2].iov_len = instance->my_failed_list_entries * sizeof (struct srp_addr); } if (instance->totem_config->send_join_timeout) { usleep (random() % (instance->totem_config->send_join_timeout * 1000)); } totemrrp_mcast_flush_send ( instance->totemrrp_handle, iovec, iovs); } static void memb_merge_detect_transmit (struct totemsrp_instance *instance) { struct memb_merge_detect memb_merge_detect; struct iovec iovec[2]; memb_merge_detect.header.type = MESSAGE_TYPE_MEMB_MERGE_DETECT; memb_merge_detect.header.endian_detector = ENDIAN_LOCAL; memb_merge_detect.header.encapsulated = 0; memb_merge_detect.header.nodeid = instance->my_id.addr[0].nodeid; srp_addr_copy (&memb_merge_detect.system_from, &instance->my_id); assert (memb_merge_detect.header.nodeid); iovec[0].iov_base = &memb_merge_detect; iovec[0].iov_len = sizeof (struct memb_merge_detect) - sizeof (struct memb_ring_id); iovec[1].iov_base = &instance->my_ring_id; iovec[1].iov_len = sizeof (struct memb_ring_id); totemrrp_mcast_flush_send (instance->totemrrp_handle, iovec, 2); } static void memb_ring_id_create_or_load ( struct totemsrp_instance *instance, struct memb_ring_id *memb_ring_id) { int fd; int res; char filename[256]; snprintf (filename, sizeof(filename), "%s/ringid_%s", rundir, totemip_print (&instance->my_id.addr[0])); fd = open (filename, O_RDONLY, 0700); if (fd > 0) { res = read (fd, &memb_ring_id->seq, sizeof (unsigned long long)); assert (res == sizeof (unsigned long long)); close (fd); } else if (fd == -1 && errno == ENOENT) { memb_ring_id->seq = 0; umask(0); fd = open (filename, O_CREAT|O_RDWR, 0700); if (fd == -1) { log_printf (instance->totemsrp_log_level_warning, "Couldn't create %s %s\n", filename, strerror (errno)); } res = write (fd, &memb_ring_id->seq, sizeof (unsigned long long)); assert (res == sizeof (unsigned long long)); close (fd); } else { log_printf (instance->totemsrp_log_level_warning, "Couldn't open %s %s\n", filename, strerror (errno)); } totemip_copy(&memb_ring_id->rep, &instance->my_id.addr[0]); assert (!totemip_zero_check(&memb_ring_id->rep)); instance->token_ring_id_seq = memb_ring_id->seq; } static void memb_ring_id_set_and_store ( struct totemsrp_instance *instance, struct memb_ring_id *ring_id) { char filename[256]; int fd; int res; memcpy (&instance->my_ring_id, ring_id, sizeof (struct memb_ring_id)); snprintf (filename, sizeof(filename), "%s/ringid_%s", rundir, totemip_print (&instance->my_id.addr[0])); fd = open (filename, O_WRONLY, 0777); if (fd == -1) { fd = open (filename, O_CREAT|O_RDWR, 0777); } if (fd == -1) { log_printf (instance->totemsrp_log_level_warning, "Couldn't store new ring id %llx to stable storage (%s)\n", instance->my_ring_id.seq, strerror (errno)); assert (0); return; } log_printf (instance->totemsrp_log_level_notice, "Storing new sequence id for ring %llx\n", instance->my_ring_id.seq); //assert (fd > 0); res = write (fd, &instance->my_ring_id.seq, sizeof (unsigned long long)); assert (res == sizeof (unsigned long long)); close (fd); } int totemsrp_callback_token_create ( hdb_handle_t handle, void **handle_out, enum totem_callback_token_type type, int delete, int (*callback_fn) (enum totem_callback_token_type type, const void *), const void *data) { struct token_callback_instance *callback_handle; struct totemsrp_instance *instance; unsigned int res; res = hdb_handle_get (&totemsrp_instance_database, handle, (void *)&instance); if (res != 0) { goto error_exit; } token_hold_cancel_send (instance); callback_handle = malloc (sizeof (struct token_callback_instance)); if (callback_handle == 0) { return (-1); } *handle_out = (void *)callback_handle; list_init (&callback_handle->list); callback_handle->callback_fn = callback_fn; callback_handle->data = (void *) data; callback_handle->callback_type = type; callback_handle->delete = delete; switch (type) { case TOTEM_CALLBACK_TOKEN_RECEIVED: list_add (&callback_handle->list, &instance->token_callback_received_listhead); break; case TOTEM_CALLBACK_TOKEN_SENT: list_add (&callback_handle->list, &instance->token_callback_sent_listhead); break; } hdb_handle_put (&totemsrp_instance_database, handle); error_exit: return (0); } void totemsrp_callback_token_destroy (hdb_handle_t handle, void **handle_out) { struct token_callback_instance *h; if (*handle_out) { h = (struct token_callback_instance *)*handle_out; list_del (&h->list); free (h); h = NULL; *handle_out = 0; } } static void token_callbacks_execute ( struct totemsrp_instance *instance, enum totem_callback_token_type type) { struct list_head *list; struct list_head *list_next; struct list_head *callback_listhead = 0; struct token_callback_instance *token_callback_instance; int res; int del; switch (type) { case TOTEM_CALLBACK_TOKEN_RECEIVED: callback_listhead = &instance->token_callback_received_listhead; break; case TOTEM_CALLBACK_TOKEN_SENT: callback_listhead = &instance->token_callback_sent_listhead; break; default: assert (0); } for (list = callback_listhead->next; list != callback_listhead; list = list_next) { token_callback_instance = list_entry (list, struct token_callback_instance, list); list_next = list->next; del = token_callback_instance->delete; if (del == 1) { list_del (list); } res = token_callback_instance->callback_fn ( token_callback_instance->callback_type, token_callback_instance->data); /* * This callback failed to execute, try it again on the next token */ if (res == -1 && del == 1) { list_add (list, callback_listhead); } else if (del) { free (token_callback_instance); } } } /* * Flow control functions */ static unsigned int backlog_get (struct totemsrp_instance *instance) { unsigned int backlog = 0; if (instance->memb_state == MEMB_STATE_OPERATIONAL) { backlog = queue_used (&instance->new_message_queue); } else if (instance->memb_state == MEMB_STATE_RECOVERY) { backlog = queue_used (&instance->retrans_message_queue); } return (backlog); } static int fcc_calculate ( struct totemsrp_instance *instance, struct orf_token *token) { unsigned int transmits_allowed; unsigned int backlog_calc; transmits_allowed = instance->totem_config->max_messages; if (transmits_allowed > instance->totem_config->window_size - token->fcc) { transmits_allowed = instance->totem_config->window_size - token->fcc; } instance->my_cbl = backlog_get (instance); /* * Only do backlog calculation if there is a backlog otherwise * we would result in div by zero */ if (token->backlog + instance->my_cbl - instance->my_pbl) { backlog_calc = (instance->totem_config->window_size * instance->my_pbl) / (token->backlog + instance->my_cbl - instance->my_pbl); if (backlog_calc > 0 && transmits_allowed > backlog_calc) { transmits_allowed = backlog_calc; } } return (transmits_allowed); } /* * don't overflow the RTR sort queue */ static void fcc_rtr_limit ( struct totemsrp_instance *instance, struct orf_token *token, unsigned int *transmits_allowed) { assert ((QUEUE_RTR_ITEMS_SIZE_MAX - *transmits_allowed - instance->totem_config->window_size) >= 0); if (sq_lt_compare (instance->last_released + QUEUE_RTR_ITEMS_SIZE_MAX - *transmits_allowed - instance->totem_config->window_size, token->seq)) { *transmits_allowed = 0; } } static void fcc_token_update ( struct totemsrp_instance *instance, struct orf_token *token, unsigned int msgs_transmitted) { token->fcc += msgs_transmitted - instance->my_trc; token->backlog += instance->my_cbl - instance->my_pbl; assert (token->backlog >= 0); instance->my_trc = msgs_transmitted; instance->my_pbl = instance->my_cbl; } /* * Message Handlers */ struct timeval tv_old; /* * message handler called when TOKEN message type received */ static int message_handler_orf_token ( struct totemsrp_instance *instance, void *msg, int msg_len, int endian_conversion_needed) { char token_storage[1500]; char token_convert[1500]; struct orf_token *token = NULL; int forward_token; unsigned int transmits_allowed; unsigned int mcasted_retransmit; unsigned int mcasted_regular; unsigned int last_aru; #ifdef GIVEINFO struct timeval tv_current; struct timeval tv_diff; gettimeofday (&tv_current, NULL); timersub (&tv_current, &tv_old, &tv_diff); memcpy (&tv_old, &tv_current, sizeof (struct timeval)); log_printf (instance->totemsrp_log_level_notice, "Time since last token %0.4f ms\n", (((float)tv_diff.tv_sec) * 1000) + ((float)tv_diff.tv_usec) / 1000.0); #endif #ifdef TEST_DROP_ORF_TOKEN_PERCENTAGE if (random()%100 < TEST_DROP_ORF_TOKEN_PERCENTAGE) { return (0); } #endif if (endian_conversion_needed) { orf_token_endian_convert ((struct orf_token *)msg, (struct orf_token *)token_convert); msg = (struct orf_token *)token_convert; } /* * Make copy of token and retransmit list in case we have * to flush incoming messages from the kernel queue */ token = (struct orf_token *)token_storage; memcpy (token, msg, sizeof (struct orf_token)); memcpy (&token->rtr_list[0], (char *)msg + sizeof (struct orf_token), sizeof (struct rtr_item) * RETRANSMIT_ENTRIES_MAX); /* * Handle merge detection timeout */ if (token->seq == instance->my_last_seq) { start_merge_detect_timeout (instance); instance->my_seq_unchanged += 1; } else { cancel_merge_detect_timeout (instance); cancel_token_hold_retransmit_timeout (instance); instance->my_seq_unchanged = 0; } instance->my_last_seq = token->seq; #ifdef TEST_RECOVERY_MSG_COUNT if (instance->memb_state == MEMB_STATE_OPERATIONAL && token->seq > TEST_RECOVERY_MSG_COUNT) { return (0); } #endif totemrrp_recv_flush (instance->totemrrp_handle); /* * Determine if we should hold (in reality drop) the token */ instance->my_token_held = 0; if (totemip_equal(&instance->my_ring_id.rep, &instance->my_id.addr[0]) && instance->my_seq_unchanged > instance->totem_config->seqno_unchanged_const) { instance->my_token_held = 1; } else if (!totemip_equal(&instance->my_ring_id.rep, &instance->my_id.addr[0]) && instance->my_seq_unchanged >= instance->totem_config->seqno_unchanged_const) { instance->my_token_held = 1; } /* * Hold onto token when there is no activity on ring and * this processor is the ring rep */ forward_token = 1; if (totemip_equal(&instance->my_ring_id.rep, &instance->my_id.addr[0])) { if (instance->my_token_held) { forward_token = 0; } } token_callbacks_execute (instance, TOTEM_CALLBACK_TOKEN_RECEIVED); switch (instance->memb_state) { case MEMB_STATE_COMMIT: /* Discard token */ break; case MEMB_STATE_OPERATIONAL: messages_free (instance, token->aru); case MEMB_STATE_GATHER: /* * DO NOT add break, we use different free mechanism in recovery state */ case MEMB_STATE_RECOVERY: last_aru = instance->my_last_aru; instance->my_last_aru = token->aru; /* * Discard tokens from another configuration */ if (memcmp (&token->ring_id, &instance->my_ring_id, sizeof (struct memb_ring_id)) != 0) { if ((forward_token) && instance->use_heartbeat) { reset_heartbeat_timeout(instance); } else { cancel_heartbeat_timeout(instance); } return (0); /* discard token */ } /* * Discard retransmitted tokens */ if (sq_lte_compare (token->token_seq, instance->my_token_seq)) { /* * If this processor receives a retransmitted token, it is sure * the previous processor is still alive. As a result, it can * reset its token timeout. If some processor previous to that * has failed, it will eventually not execute a reset of the * token timeout, and will cause a reconfiguration to occur. */ reset_token_timeout (instance); if ((forward_token) && instance->use_heartbeat) { reset_heartbeat_timeout(instance); } else { cancel_heartbeat_timeout(instance); } return (0); /* discard token */ } transmits_allowed = fcc_calculate (instance, token); mcasted_retransmit = orf_token_rtr (instance, token, &transmits_allowed); fcc_rtr_limit (instance, token, &transmits_allowed); mcasted_regular = orf_token_mcast (instance, token, transmits_allowed); fcc_token_update (instance, token, mcasted_retransmit + mcasted_regular); if (sq_lt_compare (instance->my_aru, token->aru) || instance->my_id.addr[0].nodeid == token->aru_addr || token->aru_addr == 0) { token->aru = instance->my_aru; if (token->aru == token->seq) { token->aru_addr = 0; } else { token->aru_addr = instance->my_id.addr[0].nodeid; } } if (token->aru == last_aru && token->aru_addr != 0) { instance->my_aru_count += 1; } else { instance->my_aru_count = 0; } if (instance->my_aru_count > instance->totem_config->fail_to_recv_const && token->aru_addr != instance->my_id.addr[0].nodeid) { log_printf (instance->totemsrp_log_level_error, "FAILED TO RECEIVE\n"); // TODO if we fail to receive, it may be possible to end with a gather // state of proc == failed = 0 entries /* THIS IS A BIG TODO memb_set_merge (&token->aru_addr, 1, instance->my_failed_list, &instance->my_failed_list_entries); */ ring_state_restore (instance); memb_state_gather_enter (instance, 6); } else { instance->my_token_seq = token->token_seq; token->token_seq += 1; if (instance->memb_state == MEMB_STATE_RECOVERY) { /* * instance->my_aru == instance->my_high_seq_received means this processor * has recovered all messages it can recover * (ie: its retrans queue is empty) */ if (queue_is_empty (&instance->retrans_message_queue) == 0) { if (token->retrans_flg == 0) { token->retrans_flg = 1; instance->my_set_retrans_flg = 1; } } else if (token->retrans_flg == 1 && instance->my_set_retrans_flg) { token->retrans_flg = 0; } log_printf (instance->totemsrp_log_level_debug, "token retrans flag is %d my set retrans flag%d retrans queue empty %d count %d, aru %x\n", token->retrans_flg, instance->my_set_retrans_flg, queue_is_empty (&instance->retrans_message_queue), instance->my_retrans_flg_count, token->aru); if (token->retrans_flg == 0) { instance->my_retrans_flg_count += 1; } else { instance->my_retrans_flg_count = 0; } if (instance->my_retrans_flg_count == 2) { instance->my_install_seq = token->seq; } log_printf (instance->totemsrp_log_level_debug, "install seq %x aru %x high seq received %x\n", instance->my_install_seq, instance->my_aru, instance->my_high_seq_received); if (instance->my_retrans_flg_count >= 2 && instance->my_received_flg == 0 && sq_lte_compare (instance->my_install_seq, instance->my_aru)) { instance->my_received_flg = 1; instance->my_deliver_memb_entries = instance->my_trans_memb_entries; memcpy (instance->my_deliver_memb_list, instance->my_trans_memb_list, sizeof (struct totem_ip_address) * instance->my_trans_memb_entries); } if (instance->my_retrans_flg_count >= 3 && sq_lte_compare (instance->my_install_seq, token->aru)) { instance->my_rotation_counter += 1; } else { instance->my_rotation_counter = 0; } if (instance->my_rotation_counter == 2) { log_printf (instance->totemsrp_log_level_debug, "retrans flag count %x token aru %x install seq %x aru %x %x\n", instance->my_retrans_flg_count, token->aru, instance->my_install_seq, instance->my_aru, token->seq); memb_state_operational_enter (instance); instance->my_rotation_counter = 0; instance->my_retrans_flg_count = 0; } } totemrrp_send_flush (instance->totemrrp_handle); token_send (instance, token, forward_token); #ifdef GIVEINFO gettimeofday (&tv_current, NULL); timersub (&tv_current, &tv_old, &tv_diff); memcpy (&tv_old, &tv_current, sizeof (struct timeval)); log_printf (instance->totemsrp_log_level_notice, "I held %0.4f ms\n", ((float)tv_diff.tv_usec) / 1000.0); #endif if (instance->memb_state == MEMB_STATE_OPERATIONAL) { messages_deliver_to_app (instance, 0, instance->my_high_seq_received); } /* * Deliver messages after token has been transmitted * to improve performance */ reset_token_timeout (instance); // REVIEWED reset_token_retransmit_timeout (instance); // REVIEWED if (totemip_equal(&instance->my_id.addr[0], &instance->my_ring_id.rep) && instance->my_token_held == 1) { start_token_hold_retransmit_timeout (instance); } token_callbacks_execute (instance, TOTEM_CALLBACK_TOKEN_SENT); } break; } if ((forward_token) && instance->use_heartbeat) { reset_heartbeat_timeout(instance); } else { cancel_heartbeat_timeout(instance); } return (0); } static void messages_deliver_to_app ( struct totemsrp_instance *instance, int skip, unsigned int end_point) { struct sort_queue_item *sort_queue_item_p; unsigned int i; int res; struct mcast *mcast_in; struct mcast mcast_header; unsigned int range = 0; int endian_conversion_required; unsigned int my_high_delivered_stored = 0; range = end_point - instance->my_high_delivered; if (range) { log_printf (instance->totemsrp_log_level_debug, "Delivering %x to %x\n", instance->my_high_delivered, end_point); } assert (range < 10240); my_high_delivered_stored = instance->my_high_delivered; /* * Deliver messages in order from rtr queue to pending delivery queue */ for (i = 1; i <= range; i++) { void *ptr = 0; /* * If out of range of sort queue, stop assembly */ res = sq_in_range (&instance->regular_sort_queue, my_high_delivered_stored + i); if (res == 0) { break; } res = sq_item_get (&instance->regular_sort_queue, my_high_delivered_stored + i, &ptr); /* * If hole, stop assembly */ if (res != 0 && skip == 0) { break; } instance->my_high_delivered = my_high_delivered_stored + i; if (res != 0) { continue; } sort_queue_item_p = ptr; mcast_in = sort_queue_item_p->iovec[0].iov_base; assert (mcast_in != (struct mcast *)0xdeadbeef); endian_conversion_required = 0; if (mcast_in->header.endian_detector != ENDIAN_LOCAL) { endian_conversion_required = 1; mcast_endian_convert (mcast_in, &mcast_header); } else { memcpy (&mcast_header, mcast_in, sizeof (struct mcast)); } /* * Skip messages not originated in instance->my_deliver_memb */ if (skip && memb_set_subset (&mcast_header.system_from, 1, instance->my_deliver_memb_list, instance->my_deliver_memb_entries) == 0) { instance->my_high_delivered = my_high_delivered_stored + i; continue; } /* * Message found */ log_printf (instance->totemsrp_log_level_debug, "Delivering MCAST message with seq %x to pending delivery queue\n", mcast_header.seq); /* * Message is locally originated multicast */ if (sort_queue_item_p->iov_len > 1 && sort_queue_item_p->iovec[0].iov_len == sizeof (struct mcast)) { instance->totemsrp_deliver_fn ( mcast_header.header.nodeid, &sort_queue_item_p->iovec[1], sort_queue_item_p->iov_len - 1, endian_conversion_required); } else { sort_queue_item_p->iovec[0].iov_len -= sizeof (struct mcast); sort_queue_item_p->iovec[0].iov_base = (char *)sort_queue_item_p->iovec[0].iov_base + sizeof (struct mcast); instance->totemsrp_deliver_fn ( mcast_header.header.nodeid, sort_queue_item_p->iovec, sort_queue_item_p->iov_len, endian_conversion_required); sort_queue_item_p->iovec[0].iov_len += sizeof (struct mcast); sort_queue_item_p->iovec[0].iov_base = (char *)sort_queue_item_p->iovec[0].iov_base - sizeof (struct mcast); } //TODO instance->stats_delv += 1; } } /* * recv message handler called when MCAST message type received */ static int message_handler_mcast ( struct totemsrp_instance *instance, void *msg, int msg_len, int endian_conversion_needed) { struct sort_queue_item sort_queue_item; struct sq *sort_queue; struct mcast mcast_header; if (endian_conversion_needed) { mcast_endian_convert (msg, &mcast_header); } else { memcpy (&mcast_header, msg, sizeof (struct mcast)); } if (mcast_header.header.encapsulated == MESSAGE_ENCAPSULATED) { sort_queue = &instance->recovery_sort_queue; } else { sort_queue = &instance->regular_sort_queue; } assert (msg_len < FRAME_SIZE_MAX); #ifdef TEST_DROP_MCAST_PERCENTAGE if (random()%100 < TEST_DROP_MCAST_PERCENTAGE) { printf ("dropping message %d\n", mcast_header.seq); return (0); } else { printf ("accepting message %d\n", mcast_header.seq); } #endif if (srp_addr_equal (&mcast_header.system_from, &instance->my_id) == 0) { cancel_token_retransmit_timeout (instance); } /* * If the message is foreign execute the switch below */ if (memcmp (&instance->my_ring_id, &mcast_header.ring_id, sizeof (struct memb_ring_id)) != 0) { switch (instance->memb_state) { case MEMB_STATE_OPERATIONAL: memb_set_merge ( &mcast_header.system_from, 1, instance->my_proc_list, &instance->my_proc_list_entries); memb_state_gather_enter (instance, 7); break; case MEMB_STATE_GATHER: if (!memb_set_subset ( &mcast_header.system_from, 1, instance->my_proc_list, instance->my_proc_list_entries)) { memb_set_merge (&mcast_header.system_from, 1, instance->my_proc_list, &instance->my_proc_list_entries); memb_state_gather_enter (instance, 8); return (0); } break; case MEMB_STATE_COMMIT: /* discard message */ break; case MEMB_STATE_RECOVERY: /* discard message */ break; } return (0); } log_printf (instance->totemsrp_log_level_debug, "Received ringid(%s:%lld) seq %x\n", totemip_print (&mcast_header.ring_id.rep), mcast_header.ring_id.seq, mcast_header.seq); /* * Add mcast message to rtr queue if not already in rtr queue * otherwise free io vectors */ if (msg_len > 0 && msg_len < FRAME_SIZE_MAX && sq_in_range (sort_queue, mcast_header.seq) && sq_item_inuse (sort_queue, mcast_header.seq) == 0) { /* * Allocate new multicast memory block */ // TODO LEAK sort_queue_item.iovec[0].iov_base = malloc (msg_len); if (sort_queue_item.iovec[0].iov_base == 0) { return (-1); /* error here is corrected by the algorithm */ } memcpy (sort_queue_item.iovec[0].iov_base, msg, msg_len); sort_queue_item.iovec[0].iov_len = msg_len; assert (sort_queue_item.iovec[0].iov_len > 0); assert (sort_queue_item.iovec[0].iov_len < FRAME_SIZE_MAX); sort_queue_item.iov_len = 1; if (sq_lt_compare (instance->my_high_seq_received, mcast_header.seq)) { instance->my_high_seq_received = mcast_header.seq; } sq_item_add (sort_queue, &sort_queue_item, mcast_header.seq); } update_aru (instance); if (instance->memb_state == MEMB_STATE_OPERATIONAL) { messages_deliver_to_app (instance, 0, instance->my_high_seq_received); } /* TODO remove from retrans message queue for old ring in recovery state */ return (0); } static int message_handler_memb_merge_detect ( struct totemsrp_instance *instance, void *msg, int msg_len, int endian_conversion_needed) { struct memb_merge_detect *memb_merge_detect = (struct memb_merge_detect *)msg; if (endian_conversion_needed) { memb_merge_detect_endian_convert (msg, msg); } /* * do nothing if this is a merge detect from this configuration */ if (memcmp (&instance->my_ring_id, &memb_merge_detect->ring_id, sizeof (struct memb_ring_id)) == 0) { return (0); } /* * Execute merge operation */ switch (instance->memb_state) { case MEMB_STATE_OPERATIONAL: memb_set_merge (&memb_merge_detect->system_from, 1, instance->my_proc_list, &instance->my_proc_list_entries); memb_state_gather_enter (instance, 9); break; case MEMB_STATE_GATHER: if (!memb_set_subset ( &memb_merge_detect->system_from, 1, instance->my_proc_list, instance->my_proc_list_entries)) { memb_set_merge (&memb_merge_detect->system_from, 1, instance->my_proc_list, &instance->my_proc_list_entries); memb_state_gather_enter (instance, 10); return (0); } break; case MEMB_STATE_COMMIT: /* do nothing in commit */ break; case MEMB_STATE_RECOVERY: /* do nothing in recovery */ break; } return (0); } static int memb_join_process ( struct totemsrp_instance *instance, struct memb_join *memb_join) { unsigned char *commit_token_storage[TOKEN_SIZE_MAX]; struct memb_commit_token *my_commit_token = (struct memb_commit_token *)commit_token_storage; struct srp_addr *proc_list; struct srp_addr *failed_list; proc_list = (struct srp_addr *)memb_join->end_of_memb_join; failed_list = proc_list + memb_join->proc_list_entries; if (memb_set_equal (proc_list, memb_join->proc_list_entries, instance->my_proc_list, instance->my_proc_list_entries) && memb_set_equal (failed_list, memb_join->failed_list_entries, instance->my_failed_list, instance->my_failed_list_entries)) { memb_consensus_set (instance, &memb_join->system_from); if (memb_consensus_agreed (instance) && memb_lowest_in_config (instance)) { memb_state_commit_token_create (instance, my_commit_token); memb_state_commit_enter (instance, my_commit_token); } else { return (0); } } else if (memb_set_subset (proc_list, memb_join->proc_list_entries, instance->my_proc_list, instance->my_proc_list_entries) && memb_set_subset (failed_list, memb_join->failed_list_entries, instance->my_failed_list, instance->my_failed_list_entries)) { return (0); } else if (memb_set_subset (&memb_join->system_from, 1, instance->my_failed_list, instance->my_failed_list_entries)) { return (0); } else { memb_set_merge (proc_list, memb_join->proc_list_entries, instance->my_proc_list, &instance->my_proc_list_entries); if (memb_set_subset ( &instance->my_id, 1, failed_list, memb_join->failed_list_entries)) { memb_set_merge ( &memb_join->system_from, 1, instance->my_failed_list, &instance->my_failed_list_entries); } else { memb_set_merge (failed_list, memb_join->failed_list_entries, instance->my_failed_list, &instance->my_failed_list_entries); } memb_state_gather_enter (instance, 11); return (1); /* gather entered */ } return (0); /* gather not entered */ } static void memb_join_endian_convert (struct memb_join *in, struct memb_join *out) { int i; struct srp_addr *in_proc_list; struct srp_addr *in_failed_list; struct srp_addr *out_proc_list; struct srp_addr *out_failed_list; out->header.type = in->header.type; out->header.endian_detector = ENDIAN_LOCAL; out->header.nodeid = swab32 (in->header.nodeid); srp_addr_copy_endian_convert (&out->system_from, &in->system_from); out->proc_list_entries = swab32 (in->proc_list_entries); out->failed_list_entries = swab32 (in->failed_list_entries); out->ring_seq = swab64 (in->ring_seq); in_proc_list = (struct srp_addr *)in->end_of_memb_join; in_failed_list = in_proc_list + out->proc_list_entries; out_proc_list = (struct srp_addr *)out->end_of_memb_join; out_failed_list = out_proc_list + out->proc_list_entries; for (i = 0; i < out->proc_list_entries; i++) { srp_addr_copy_endian_convert (&out_proc_list[i], &in_proc_list[i]); } for (i = 0; i < out->failed_list_entries; i++) { srp_addr_copy_endian_convert (&out_failed_list[i], &in_failed_list[i]); } } static void memb_commit_token_endian_convert (struct memb_commit_token *in, struct memb_commit_token *out) { int i; struct srp_addr *in_addr = (struct srp_addr *)in->end_of_commit_token; struct srp_addr *out_addr = (struct srp_addr *)out->end_of_commit_token; struct memb_commit_token_memb_entry *in_memb_list; struct memb_commit_token_memb_entry *out_memb_list; out->header.type = in->header.type; out->header.endian_detector = ENDIAN_LOCAL; out->header.nodeid = swab32 (in->header.nodeid); out->token_seq = swab32 (in->token_seq); totemip_copy_endian_convert(&out->ring_id.rep, &in->ring_id.rep); out->ring_id.seq = swab64 (in->ring_id.seq); out->retrans_flg = swab32 (in->retrans_flg); out->memb_index = swab32 (in->memb_index); out->addr_entries = swab32 (in->addr_entries); in_memb_list = (struct memb_commit_token_memb_entry *)(in_addr + out->addr_entries); out_memb_list = (struct memb_commit_token_memb_entry *)(out_addr + out->addr_entries); for (i = 0; i < out->addr_entries; i++) { srp_addr_copy_endian_convert (&out_addr[i], &in_addr[i]); /* * Only convert the memb entry if it has been set */ if (in_memb_list[i].ring_id.rep.family != 0) { totemip_copy_endian_convert (&out_memb_list[i].ring_id.rep, &in_memb_list[i].ring_id.rep); out_memb_list[i].ring_id.seq = swab64 (in_memb_list[i].ring_id.seq); out_memb_list[i].aru = swab32 (in_memb_list[i].aru); out_memb_list[i].high_delivered = swab32 (in_memb_list[i].high_delivered); out_memb_list[i].received_flg = swab32 (in_memb_list[i].received_flg); } } } static void orf_token_endian_convert (struct orf_token *in, struct orf_token *out) { int i; out->header.type = in->header.type; out->header.endian_detector = ENDIAN_LOCAL; out->header.nodeid = swab32 (in->header.nodeid); out->seq = swab32 (in->seq); out->token_seq = swab32 (in->token_seq); out->aru = swab32 (in->aru); totemip_copy_endian_convert(&out->ring_id.rep, &in->ring_id.rep); out->aru_addr = swab32(in->aru_addr); out->ring_id.seq = swab64 (in->ring_id.seq); out->fcc = swab32 (in->fcc); out->backlog = swab32 (in->backlog); out->retrans_flg = swab32 (in->retrans_flg); out->rtr_list_entries = swab32 (in->rtr_list_entries); for (i = 0; i < out->rtr_list_entries; i++) { totemip_copy_endian_convert(&out->rtr_list[i].ring_id.rep, &in->rtr_list[i].ring_id.rep); out->rtr_list[i].ring_id.seq = swab64 (in->rtr_list[i].ring_id.seq); out->rtr_list[i].seq = swab32 (in->rtr_list[i].seq); } } static void mcast_endian_convert (struct mcast *in, struct mcast *out) { out->header.type = in->header.type; out->header.endian_detector = ENDIAN_LOCAL; out->header.nodeid = swab32 (in->header.nodeid); out->header.encapsulated = in->header.encapsulated; out->seq = swab32 (in->seq); out->this_seqno = swab32 (in->this_seqno); totemip_copy_endian_convert(&out->ring_id.rep, &in->ring_id.rep); out->ring_id.seq = swab64 (in->ring_id.seq); out->node_id = swab32 (in->node_id); out->guarantee = swab32 (in->guarantee); srp_addr_copy_endian_convert (&out->system_from, &in->system_from); } static void memb_merge_detect_endian_convert ( struct memb_merge_detect *in, struct memb_merge_detect *out) { out->header.type = in->header.type; out->header.endian_detector = ENDIAN_LOCAL; out->header.nodeid = swab32 (in->header.nodeid); totemip_copy_endian_convert(&out->ring_id.rep, &in->ring_id.rep); out->ring_id.seq = swab64 (in->ring_id.seq); srp_addr_copy_endian_convert (&out->system_from, &in->system_from); } static int message_handler_memb_join ( struct totemsrp_instance *instance, void *msg, int msg_len, int endian_conversion_needed) { struct memb_join *memb_join; struct memb_join *memb_join_convert = alloca (msg_len); int gather_entered; if (endian_conversion_needed) { memb_join = memb_join_convert; memb_join_endian_convert (msg, memb_join_convert); } else { memb_join = (struct memb_join *)msg; } if (instance->token_ring_id_seq < memb_join->ring_seq) { instance->token_ring_id_seq = memb_join->ring_seq; } switch (instance->memb_state) { case MEMB_STATE_OPERATIONAL: gather_entered = memb_join_process (instance, memb_join); if (gather_entered == 0) { memb_state_gather_enter (instance, 12); } break; case MEMB_STATE_GATHER: memb_join_process (instance, memb_join); break; case MEMB_STATE_COMMIT: if (memb_set_subset (&memb_join->system_from, 1, instance->my_new_memb_list, instance->my_new_memb_entries) && memb_join->ring_seq >= instance->my_ring_id.seq) { memb_join_process (instance, memb_join); memb_state_gather_enter (instance, 13); } break; case MEMB_STATE_RECOVERY: if (memb_set_subset (&memb_join->system_from, 1, instance->my_new_memb_list, instance->my_new_memb_entries) && memb_join->ring_seq >= instance->my_ring_id.seq) { ring_state_restore (instance); memb_join_process (instance, memb_join); memb_state_gather_enter (instance, 14); } break; } return (0); } static int message_handler_memb_commit_token ( struct totemsrp_instance *instance, void *msg, int msg_len, int endian_conversion_needed) { struct memb_commit_token *memb_commit_token_convert = alloca (msg_len); struct memb_commit_token *memb_commit_token; struct srp_addr sub[PROCESSOR_COUNT_MAX]; int sub_entries; struct srp_addr *addr; struct memb_commit_token_memb_entry *memb_list; log_printf (instance->totemsrp_log_level_debug, "got commit token\n"); if (endian_conversion_needed) { memb_commit_token = memb_commit_token_convert; memb_commit_token_endian_convert (msg, memb_commit_token); } else { memb_commit_token = (struct memb_commit_token *)msg; } addr = (struct srp_addr *)memb_commit_token->end_of_commit_token; memb_list = (struct memb_commit_token_memb_entry *)(addr + memb_commit_token->addr_entries); #ifdef TEST_DROP_COMMIT_TOKEN_PERCENTAGE if (random()%100 < TEST_DROP_COMMIT_TOKEN_PERCENTAGE) { return (0); } #endif switch (instance->memb_state) { case MEMB_STATE_OPERATIONAL: /* discard token */ break; case MEMB_STATE_GATHER: memb_set_subtract (sub, &sub_entries, instance->my_proc_list, instance->my_proc_list_entries, instance->my_failed_list, instance->my_failed_list_entries); if (memb_set_equal (addr, memb_commit_token->addr_entries, sub, sub_entries) && memb_commit_token->ring_id.seq > instance->my_ring_id.seq) { memb_state_commit_enter (instance, memb_commit_token); } break; case MEMB_STATE_COMMIT: /* * If retransmitted commit tokens are sent on this ring * filter them out and only enter recovery once the * commit token has traversed the array. This is * determined by : * memb_commit_token->memb_index == memb_commit_token->addr_entries) { */ if (memb_commit_token->ring_id.seq == instance->my_ring_id.seq && memb_commit_token->memb_index == memb_commit_token->addr_entries) { memb_state_recovery_enter (instance, memb_commit_token); } break; case MEMB_STATE_RECOVERY: if (totemip_equal (&instance->my_id.addr[0], &instance->my_ring_id.rep)) { log_printf (instance->totemsrp_log_level_notice, "Sending initial ORF token\n"); // TODO convert instead of initiate orf_token_send_initial (instance); reset_token_timeout (instance); // REVIEWED reset_token_retransmit_timeout (instance); // REVIEWED } break; } return (0); } static int message_handler_token_hold_cancel ( struct totemsrp_instance *instance, void *msg, int msg_len, int endian_conversion_needed) { struct token_hold_cancel *token_hold_cancel = (struct token_hold_cancel *)msg; if (memcmp (&token_hold_cancel->ring_id, &instance->my_ring_id, sizeof (struct memb_ring_id)) == 0) { instance->my_seq_unchanged = 0; if (totemip_equal(&instance->my_ring_id.rep, &instance->my_id.addr[0])) { timer_function_token_retransmit_timeout (instance); } } return (0); } void main_deliver_fn ( void *context, void *msg, int msg_len) { struct totemsrp_instance *instance = (struct totemsrp_instance *)context; struct message_header *message_header = (struct message_header *)msg; if (msg_len < sizeof (struct message_header)) { log_printf (instance->totemsrp_log_level_security, "Received message is too short... ignoring %d.\n", msg_len); return; } if ((int)message_header->type >= totemsrp_message_handlers.count) { log_printf (instance->totemsrp_log_level_security, "Type of received message is wrong... ignoring %d.\n", (int)message_header->type); return; } /* * Handle incoming message */ totemsrp_message_handlers.handler_functions[(int)message_header->type] ( instance, msg, msg_len, message_header->endian_detector != ENDIAN_LOCAL); } void main_iface_change_fn ( void *context, struct totem_ip_address *iface_addr, unsigned int iface_no) { struct totemsrp_instance *instance = (struct totemsrp_instance *)context; totemip_copy (&instance->my_id.addr[iface_no], iface_addr); assert (instance->my_id.addr[iface_no].nodeid); totemip_copy (&instance->my_memb_list[0].addr[iface_no], iface_addr); if (instance->iface_changes++ == 0) { memb_ring_id_create_or_load (instance, &instance->my_ring_id); log_printf ( instance->totemsrp_log_level_notice, "Created or loaded sequence id %lld.%s for this ring.\n", instance->my_ring_id.seq, totemip_print (&instance->my_ring_id.rep)); } if (instance->iface_changes >= instance->totem_config->interface_count) { memb_state_gather_enter (instance, 15); } } void totemsrp_net_mtu_adjust (struct totem_config *totem_config) { totem_config->net_mtu -= sizeof (struct mcast); } diff --git a/include/corosync/hdb.h b/include/corosync/hdb.h index c61490fe..acca50c7 100644 --- a/include/corosync/hdb.h +++ b/include/corosync/hdb.h @@ -1,276 +1,346 @@ /* * Copyright (c) 2002-2006 MontaVista Software, Inc. * Copyright (c) 2006-2009 Red Hat, Inc. * * All rights reserved. * * Author: Steven Dake (sdake@redhat.com) * * 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 MontaVista Software, Inc. 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. */ #ifndef HDB_H_DEFINED #define HDB_H_DEFINED #include #include #include #include typedef unsigned long long hdb_handle_t; enum HDB_HANDLE_STATE { HDB_HANDLE_STATE_EMPTY, HDB_HANDLE_STATE_PENDINGREMOVAL, HDB_HANDLE_STATE_ACTIVE }; struct hdb_handle { int state; void *instance; int check; int ref_count; }; struct hdb_handle_database { unsigned int handle_count; struct hdb_handle *handles; unsigned int iterator; - pthread_mutex_t mutex; +#if defined(HAVE_PTHREAD_SPIN_LOCK) + pthread_spinlock_t lock; +#else + pthread_mutex_t lock; +#endif + unsigned int first_run; }; +#if defined(HAVE_PTHREAD_SPIN_LOCK) +static inline void hdb_database_lock (pthread_spinlock_t *spinlock) +{ + pthread_spin_lock (spinlock); +} + +static inline void hdb_database_unlock (pthread_spinlock_t *spinlock) +{ + pthread_spin_unlock (spinlock); +} +static inline void hdb_database_lock_init (pthread_spinlock_t *spinlock) +{ + pthread_spin_init (spinlock, 0); +} + +static inline void hdb_database_lock_destroy (pthread_spinlock_t *spinlock) +{ + pthread_spin_destroy (spinlock); +} + +#else +static inline void hdb_database_lock (pthread_mutex_t *mutex) +{ + pthread_mutex_lock (mutex); +} + +static inline void hdb_database_unlock (pthread_mutex_t *mutex) +{ + pthread_mutex_unlock (mutex); +} +static inline void hdb_database_lock_init (pthread_mutex_t *mutex) +{ + pthread_mutex_init (mutex, NULL); +} + +static inline void hdb_database_lock_destroy (pthread_mutex_t *mutex) +{ + pthread_mutex_destroy (mutex); +} +#endif + +#define DECLARE_HDB_DATABASE(database_name) \ +static struct hdb_handle_database (database_name); \ +static void database_name##_init(void)__attribute__((constructor)); \ +static void database_name##_init(void) \ +{ \ + memset (&(database_name), 0, sizeof (struct hdb_handle_database));\ + hdb_database_lock_init (&(database_name).lock); \ +} + +#define DECLARE_HDB_DATABASE_FIRSTRUN(database_name) \ +static struct hdb_handle_database (database_name) = { \ + .first_run = 1, \ +}; \ +static void database_name##_init(void)__attribute__((constructor)); \ +static void database_name##_init(void) \ +{ \ + memset (&(database_name), 0, sizeof (struct hdb_handle_database));\ + hdb_database_lock_init (&(database_name).lock); \ +} + static inline void hdb_create ( struct hdb_handle_database *handle_database) { memset (handle_database, 0, sizeof (struct hdb_handle_database)); - pthread_mutex_init (&handle_database->mutex, NULL); + hdb_database_lock_init (&handle_database->lock); } static inline void hdb_destroy ( struct hdb_handle_database *handle_database) { if (handle_database->handles) { free (handle_database->handles); } - pthread_mutex_destroy (&handle_database->mutex); + hdb_database_lock_destroy (&handle_database->lock); memset (handle_database, 0, sizeof (struct hdb_handle_database)); } static inline int hdb_handle_create ( struct hdb_handle_database *handle_database, int instance_size, hdb_handle_t *handle_id_out) { int handle; unsigned int check; void *new_handles; int found = 0; void *instance; int i; - pthread_mutex_lock (&handle_database->mutex); + if (handle_database->first_run == 1) { + memset (handle_database, 0, sizeof (struct hdb_handle_database)); + hdb_database_lock_init (&handle_database->lock); + } + hdb_database_lock (&handle_database->lock); for (handle = 0; handle < handle_database->handle_count; handle++) { if (handle_database->handles[handle].state == HDB_HANDLE_STATE_EMPTY) { found = 1; break; } } if (found == 0) { handle_database->handle_count += 1; new_handles = (struct hdb_handle *)realloc (handle_database->handles, sizeof (struct hdb_handle) * handle_database->handle_count); if (new_handles == NULL) { - pthread_mutex_unlock (&handle_database->mutex); + hdb_database_unlock (&handle_database->lock); return (-1); } handle_database->handles = new_handles; } instance = (void *)malloc (instance_size); if (instance == 0) { return (-1); } /* * This code makes sure the random number isn't zero * We use 0 to specify an invalid handle out of the 1^64 address space * If we get 0 200 times in a row, the RNG may be broken */ for (i = 0; i < 200; i++) { check = random(); if (check != 0 && check != 0xffffffff) { break; } } memset (instance, 0, instance_size); handle_database->handles[handle].state = HDB_HANDLE_STATE_ACTIVE; handle_database->handles[handle].instance = instance; handle_database->handles[handle].ref_count = 1; handle_database->handles[handle].check = check; *handle_id_out = (((unsigned long long)(check)) << 32) | handle; - pthread_mutex_unlock (&handle_database->mutex); + hdb_database_unlock (&handle_database->lock); return (0); } static inline int hdb_handle_get ( struct hdb_handle_database *handle_database, hdb_handle_t handle_in, void **instance) { unsigned int check = ((unsigned int)(((unsigned long long)handle_in) >> 32)); unsigned int handle = handle_in & 0xffffffff; - pthread_mutex_lock (&handle_database->mutex); + hdb_database_lock (&handle_database->lock); if (check != 0xffffffff && check != handle_database->handles[handle].check) { - pthread_mutex_unlock (&handle_database->mutex); + hdb_database_unlock (&handle_database->lock); return (-1); } *instance = NULL; if (handle >= handle_database->handle_count) { - pthread_mutex_unlock (&handle_database->mutex); + hdb_database_unlock (&handle_database->lock); return (-1); } if (handle_database->handles[handle].state != HDB_HANDLE_STATE_ACTIVE) { - pthread_mutex_unlock (&handle_database->mutex); + hdb_database_unlock (&handle_database->lock); return (-1); } *instance = handle_database->handles[handle].instance; handle_database->handles[handle].ref_count += 1; - pthread_mutex_unlock (&handle_database->mutex); + hdb_database_unlock (&handle_database->lock); return (0); } static inline int hdb_handle_put ( struct hdb_handle_database *handle_database, hdb_handle_t handle_in) { unsigned int check = ((unsigned int)(((unsigned long long)handle_in) >> 32)); unsigned int handle = handle_in & 0xffffffff; - pthread_mutex_lock (&handle_database->mutex); + hdb_database_lock (&handle_database->lock); if (check != 0xffffffff && check != handle_database->handles[handle].check) { - pthread_mutex_unlock (&handle_database->mutex); + hdb_database_unlock (&handle_database->lock); return (-1); } handle_database->handles[handle].ref_count -= 1; assert (handle_database->handles[handle].ref_count >= 0); if (handle_database->handles[handle].ref_count == 0) { free (handle_database->handles[handle].instance); memset (&handle_database->handles[handle], 0, sizeof (struct hdb_handle)); } - pthread_mutex_unlock (&handle_database->mutex); + hdb_database_unlock (&handle_database->lock); return (0); } static inline int hdb_handle_destroy ( struct hdb_handle_database *handle_database, hdb_handle_t handle_in) { unsigned int check = ((unsigned int)(((unsigned long long)handle_in) >> 32)); unsigned int handle = handle_in & 0xffffffff; int res; - pthread_mutex_lock (&handle_database->mutex); + hdb_database_lock (&handle_database->lock); if (check != 0xffffffff && check != handle_database->handles[handle].check) { - pthread_mutex_unlock (&handle_database->mutex); + hdb_database_unlock (&handle_database->lock); return (-1); } handle_database->handles[handle].state = HDB_HANDLE_STATE_PENDINGREMOVAL; - pthread_mutex_unlock (&handle_database->mutex); + hdb_database_unlock (&handle_database->lock); res = hdb_handle_put (handle_database, handle); return (res); } static inline void hdb_iterator_reset ( struct hdb_handle_database *handle_database) { handle_database->iterator = 0; } static inline int hdb_iterator_next ( struct hdb_handle_database *handle_database, void **instance, hdb_handle_t *handle) { int res = -1; while (handle_database->iterator < handle_database->handle_count) { *handle = ((unsigned long long)(handle_database->handles[handle_database->iterator].check) << 32) | handle_database->iterator; res = hdb_handle_get ( handle_database, *handle, instance); handle_database->iterator += 1; if (res == 0) { break; } } return (res); } static inline unsigned int hdb_base_convert (hdb_handle_t handle) { return (handle & 0xffffffff); } static inline unsigned long long hdb_nocheck_convert (unsigned int handle) { unsigned long long retvalue = 0xffffffffULL << 32 | handle; return (retvalue); } #endif /* HDB_H_DEFINED */ diff --git a/lcr/lcr_ifact.c b/lcr/lcr_ifact.c index 4e53e140..bcbc2004 100644 --- a/lcr/lcr_ifact.c +++ b/lcr/lcr_ifact.c @@ -1,541 +1,547 @@ /* * Copyright (C) 2006 Steven Dake (sdake@redhat.com) * * 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 MontaVista Software, Inc. 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 #ifdef COROSYNC_SOLARIS #include #endif #include #include #include #include #include struct lcr_component_instance { struct lcr_iface *ifaces; int iface_count; hdb_handle_t comp_handle; void *dl_handle; int refcount; char library_name[256]; }; struct lcr_iface_instance { hdb_handle_t component_handle; void *context; void (*destructor) (void *context); }; +DECLARE_HDB_DATABASE_FIRSTRUN (lcr_component_instance_database); + +DECLARE_HDB_DATABASE_FIRSTRUN (lcr_iface_instance_database); + +/* static struct hdb_handle_database lcr_component_instance_database = { .handle_count = 0, .handles = 0, .iterator = 0 }; static struct hdb_handle_database lcr_iface_instance_database = { .handle_count = 0, .handles = 0, .iterator = 0 }; +*/ static hdb_handle_t g_component_handle = 0xFFFFFFFF; #if defined(COROSYNC_LINUX) || defined(COROSYNC_SOLARIS) static int lcr_select_so (const struct dirent *dirent) #else static int lcr_select_so (struct dirent *dirent) #endif { unsigned int len; len = strlen (dirent->d_name); if (len > 6) { if (strcmp (".lcrso", dirent->d_name + len - 6) == 0) { return (1); } } return (0); } #ifndef COROSYNC_SOLARIS #ifdef COROSYNC_LINUX static int pathlist_select (const struct dirent *dirent) #else static int pathlist_select (struct dirent *dirent) #endif { if (fnmatch ("*.conf", dirent->d_name, 0) == 0) { return (1); } return (0); } #endif static inline struct lcr_component_instance *lcr_comp_find ( const char *iface_name, unsigned int version, unsigned int *iface_number) { struct lcr_component_instance *instance; void *instance_p = NULL; hdb_handle_t component_handle = 0; int i; /* * Try to find interface in already loaded component */ hdb_iterator_reset (&lcr_component_instance_database); while (hdb_iterator_next (&lcr_component_instance_database, &instance_p, &component_handle) == 0) { instance = (struct lcr_component_instance *)instance_p; for (i = 0; i < instance->iface_count; i++) { if ((strcmp (instance->ifaces[i].name, iface_name) == 0) && instance->ifaces[i].version == version) { *iface_number = i; return (instance); } } hdb_handle_put (&lcr_component_instance_database, component_handle); } return (NULL); } static inline int lcr_lib_loaded ( char *library_name) { struct lcr_component_instance *instance; void *instance_p = NULL; hdb_handle_t component_handle = 0; /* * Try to find interface in already loaded component */ hdb_iterator_reset (&lcr_component_instance_database); while (hdb_iterator_next (&lcr_component_instance_database, (void *)&instance_p, &component_handle) == 0) { instance = (struct lcr_component_instance *)instance_p; if (strcmp (instance->library_name, library_name) == 0) { return (1); } hdb_handle_put (&lcr_component_instance_database, component_handle); } return (0); } enum { PATH_LIST_SIZE = 128 }; const char *path_list[PATH_LIST_SIZE]; unsigned int path_list_entries = 0; static void defaults_path_build (void) { char cwd[1024]; char *res; res = getcwd (cwd, sizeof (cwd)); if (res != NULL && (path_list[0] = strdup (cwd)) != NULL) { path_list_entries++; } path_list[path_list_entries++] = LCRSODIR; } static void ld_library_path_build (void) { char *ld_library_path; char *my_ld_library_path; char *p_s, *ptrptr; ld_library_path = getenv ("LD_LIBRARY_PATH"); if (ld_library_path == NULL) { return; } my_ld_library_path = strdup (ld_library_path); if (my_ld_library_path == NULL) { return; } p_s = strtok_r (my_ld_library_path, ":", &ptrptr); while (p_s != NULL) { char *p = strdup (p_s); if (p && path_list_entries < PATH_LIST_SIZE) { path_list[path_list_entries++] = p; } p_s = strtok_r (NULL, ":", &ptrptr); } free (my_ld_library_path); } static int ldso_path_build (const char *path, const char *filename) { #ifndef COROSYNC_SOLARIS FILE *fp; char string[1024]; char filename_cat[1024]; char newpath[1024]; char *newpath_tmp; char *new_filename; int j; struct dirent **scandir_list; unsigned int scandir_entries; snprintf (filename_cat, sizeof(filename_cat), "%s/%s", path, filename); if (filename[0] == '*') { scandir_entries = scandir ( path, &scandir_list, pathlist_select, alphasort); if (scandir_entries == 0) { return 0; } else if (scandir_entries == -1) { return -1; } else { for (j = 0; j < scandir_entries; j++) { ldso_path_build (path, scandir_list[j]->d_name); } } } fp = fopen (filename_cat, "r"); if (fp == NULL) { return (-1); } while (fgets (string, sizeof (string), fp)) { char *p; if (strlen(string) > 0) string[strlen(string) - 1] = '\0'; if (strncmp (string, "include", strlen ("include")) == 0) { newpath_tmp = string + strlen ("include") + 1; for (j = strlen (string); string[j] != ' ' && string[j] != '/' && j > 0; j--) { } string[j] = '\0'; new_filename = &string[j] + 1; strcpy (newpath, path); strcat (newpath, "/"); strcat (newpath, newpath_tmp); ldso_path_build (newpath, new_filename); continue; } p = strdup (string); if (p && path_list_entries < PATH_LIST_SIZE) { path_list[path_list_entries++] = p; } } fclose(fp); #endif return (0); } #if defined (COROSYNC_SOLARIS) && !defined(HAVE_SCANDIR) static int scandir ( const char *dir, struct dirent ***namelist, int (*filter)(const struct dirent *), int (*compar)(const struct dirent **, const struct dirent **)) { DIR *d; struct dirent *entry, **names = NULL; int namelist_items = 0, namelist_size = 0; d = opendir(dir); if (d == NULL) return -1; names = NULL; while ((entry = readdir (d)) != NULL) { struct dirent *tmpentry; if ((filter != NULL) && ((*filter)(entry) == 0)) { continue; } if (namelist_items >= namelist_size) { struct dirent **tmp; namelist_size += 512; if ((unsigned long)namelist_size > INT_MAX) { errno = EOVERFLOW; goto fail; } tmp = realloc (names, namelist_size * sizeof(struct dirent *)); if (tmp == NULL) { goto fail; } names = tmp; } tmpentry = malloc (entry->d_reclen); if (tmpentry == NULL) { goto fail; } (void) memcpy (tmpentry, entry, entry->d_reclen); names[namelist_items++] = tmpentry; } (void) closedir (d); if ((namelist_items > 1) && (compar != NULL)) { qsort (names, namelist_items, sizeof (struct dirent *), (int (*)(const void *, const void *))compar); } *namelist = names; return namelist_items; fail: { int err = errno; (void) closedir (d); while (namelist_items != 0) { namelist_items--; free (*namelist[namelist_items]); } if (names != NULL) { free (names); } *namelist = NULL; errno = err; return -1; } } #endif #if defined (COROSYNC_SOLARIS) && !defined(HAVE_ALPHASORT) static int alphasort (const struct dirent **a, const struct dirent **b) { return strcmp ((*a)->d_name, (*b)->d_name); } #endif static int interface_find_and_load ( const char *path, const char *iface_name, int version, struct lcr_component_instance **instance_ret, unsigned int *iface_number) { struct lcr_component_instance *instance; void *dl_handle; struct dirent **scandir_list; int scandir_entries; unsigned int libs_to_scan; char dl_name[1024]; scandir_entries = scandir (path, &scandir_list, lcr_select_so, alphasort); if (scandir_entries > 0) /* * no error so load the object */ for (libs_to_scan = 0; libs_to_scan < scandir_entries; libs_to_scan++) { /* * Load objects, scan them, unload them if they are not a match */ snprintf (dl_name, sizeof(dl_name), "%s/%s", path, scandir_list[libs_to_scan]->d_name); /* * Don't reload already loaded libraries */ if (lcr_lib_loaded (dl_name)) { continue; } dl_handle = dlopen (dl_name, RTLD_LAZY); if (dl_handle == NULL) { fprintf(stderr, "%s: open failed: %s\n", dl_name, dlerror()); continue; } instance = lcr_comp_find (iface_name, version, iface_number); if (instance) { instance->dl_handle = dl_handle; strcpy (instance->library_name, dl_name); goto found; } /* * No matching interfaces found, try next shared object */ if (g_component_handle != 0xFFFFFFFF) { hdb_handle_destroy (&lcr_component_instance_database, g_component_handle); g_component_handle = 0xFFFFFFFF; } dlclose (dl_handle); } /* scanning for lcrso loop */ if (scandir_entries > 0) { int i; for (i = 0; i < scandir_entries; i++) { free (scandir_list[i]); } free (scandir_list); } return -1; found: *instance_ret = instance; if (scandir_entries > 0) { int i; for (i = 0; i < scandir_entries; i++) { free (scandir_list[i]); } free (scandir_list); } return 0; } static unsigned int lcr_initialized = 0; int lcr_ifact_reference ( hdb_handle_t *iface_handle, const char *iface_name, int version, void **iface, void *context) { struct lcr_iface_instance *iface_instance; struct lcr_component_instance *instance; unsigned int iface_number; unsigned int res; unsigned int i; /* * Determine if the component is already loaded */ instance = lcr_comp_find (iface_name, version, &iface_number); if (instance) { goto found; } if (lcr_initialized == 0) { lcr_initialized = 1; defaults_path_build (); ld_library_path_build (); ldso_path_build ("/etc", "ld.so.conf"); } // TODO error checking in this code is weak /* * Search through all lcrso files for desired interface */ for (i = 0; i < path_list_entries; i++) { res = interface_find_and_load ( path_list[i], iface_name, version, &instance, &iface_number); if (res == 0) { goto found; } } /* * No matching interfaces found in all shared objects */ return (-1); found: *iface = instance->ifaces[iface_number].interfaces; if (instance->ifaces[iface_number].constructor) { instance->ifaces[iface_number].constructor (context); } hdb_handle_create (&lcr_iface_instance_database, sizeof (struct lcr_iface_instance), iface_handle); hdb_handle_get (&lcr_iface_instance_database, *iface_handle, (void *)&iface_instance); iface_instance->component_handle = instance->comp_handle; iface_instance->context = context; iface_instance->destructor = instance->ifaces[iface_number].destructor; hdb_handle_put (&lcr_iface_instance_database, *iface_handle); return (0); } int lcr_ifact_release (hdb_handle_t handle) { struct lcr_iface_instance *iface_instance; int res = 0; res = hdb_handle_get (&lcr_iface_instance_database, handle, (void *)&iface_instance); if (iface_instance->destructor) { iface_instance->destructor (iface_instance->context); } hdb_handle_put (&lcr_component_instance_database, iface_instance->component_handle); hdb_handle_put (&lcr_iface_instance_database, handle); hdb_handle_destroy (&lcr_iface_instance_database, handle); return (res); } void lcr_component_register (struct lcr_comp *comp) { struct lcr_component_instance *instance; static hdb_handle_t comp_handle; hdb_handle_create (&lcr_component_instance_database, sizeof (struct lcr_component_instance), &comp_handle); hdb_handle_get (&lcr_component_instance_database, comp_handle, (void *)&instance); instance->ifaces = comp->ifaces; instance->iface_count = comp->iface_count; instance->comp_handle = comp_handle; instance->dl_handle = NULL; hdb_handle_put (&lcr_component_instance_database, comp_handle); g_component_handle = comp_handle; }