lib/evs.c

   1 /*
   2  * vi: set autoindent tabstop=4 shiftwidth=4 :
   3
   4  * Copyright (c) 2004-2005 MontaVista Software, Inc.
   5  * Copyright (c) 2006 Sun Microsystems, Inc.
   6  *
   7  * All rights reserved.
   8  *
   9  * Author: Steven Dake (sdake@mvista.com)
  10  *
  11  * This software licensed under BSD license, the text of which follows:
  12  *
  13  * Redistribution and use in source and binary forms, with or without
  14  * modification, are permitted provided that the following conditions are met:
  15  *
  16  * - Redistributions of source code must retain the above copyright notice,
  17  *   this list of conditions and the following disclaimer.
  18  * - Redistributions in binary form must reproduce the above copyright notice,
  19  *   this list of conditions and the following disclaimer in the documentation
  20  *   and/or other materials provided with the distribution.
  21  * - Neither the name of the MontaVista Software, Inc. nor the names of its
  22  *   contributors may be used to endorse or promote products derived from this
  23  *   software without specific prior written permission.
  24  *
  25  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  26  * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  27  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  28  * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
  29  * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  30  * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  31  * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
  32  * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
  33  * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  34  * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
  35  * THE POSSIBILITY OF SUCH DAMAGE.
  36  */
  37 /*
  38  * Provides an extended virtual synchrony API using the openais executive
  39  */
  40
  41 #include <stdlib.h>
  42 #include <string.h>
  43 #include <unistd.h>
  44 #include <pthread.h>
  45 #include <sys/types.h>
  46 #include <sys/socket.h>
  47 #include <errno.h>
  48
  49 #include <swab.h>
  50 #include "../exec/totem.h"
  51 #include <saAis.h>
  52 #include <evs.h>
  53 #include <ipc_evs.h>
  54 #include <ais_util.h>
  55
  56 struct evs_inst {
  57         int response_fd;
  58         int dispatch_fd;
  59         int finalize;
  60         evs_callbacks_t callbacks;
  61         pthread_mutex_t response_mutex;
  62         pthread_mutex_t dispatch_mutex;
  63 };
  64
  65 struct res_overlay {
  66         mar_res_header_t header __attribute__((aligned(8)));
  67         char data[512000];
  68 };
  69
  70 static void evs_instance_destructor (void *instance);
  71
  72 static struct saHandleDatabase evs_handle_t_db = {
  73         .handleCount                            = 0,
  74         .handles                                        = 0,
  75         .mutex                                          = PTHREAD_MUTEX_INITIALIZER,
  76         .handleInstanceDestructor       = evs_instance_destructor
  77 };
  78
  79 /*
  80  * Clean up function for an evt instance (saEvtInitialize) handle
  81  */
  82 static void evs_instance_destructor (void *instance)
  83 {
  84         struct evs_inst *evs_inst = instance;
  85
  86         pthread_mutex_destroy (&evs_inst->response_mutex);
  87         pthread_mutex_destroy (&evs_inst->dispatch_mutex);
  88 }
  89
  90
  91 /**
  92  * @defgroup evs_openais The extended virtual synchrony passthrough API
  93  * @ingroup openais
  94  *
  95  * @{
  96  */
  97 /**
  98  * test
  99  * @param handle The handle of evs initialize
 100  * @param callbacks The callbacks for evs_initialize
 101  * @returns EVS_OK
 102  */
 103 evs_error_t evs_initialize (
 104         evs_handle_t *handle,
 105         evs_callbacks_t *callbacks)
 106 {
 107         SaAisErrorT error;
 108         struct evs_inst *evs_inst;
 109
 110         error = saHandleCreate (&evs_handle_t_db, sizeof (struct evs_inst), handle);
 111         if (error != SA_AIS_OK) {
 112                 goto error_no_destroy;
 113         }
 114
 115         error = saHandleInstanceGet (&evs_handle_t_db, *handle, (void *)&evs_inst);
 116         if (error != SA_AIS_OK) {
 117                 goto error_destroy;
 118         }
 119
 120         error = saServiceConnect (&evs_inst->response_fd,
 121                 &evs_inst->dispatch_fd,
 122                 EVS_SERVICE);
 123         if (error != SA_AIS_OK) {
 124                 goto error_put_destroy;
 125         }
 126
 127         memcpy (&evs_inst->callbacks, callbacks, sizeof (evs_callbacks_t));
 128
 129         pthread_mutex_init (&evs_inst->response_mutex, NULL);
 130
 131         pthread_mutex_init (&evs_inst->dispatch_mutex, NULL);
 132
 133         saHandleInstancePut (&evs_handle_t_db, *handle);
 134
 135         return (SA_AIS_OK);
 136
 137 error_put_destroy:
 138         saHandleInstancePut (&evs_handle_t_db, *handle);
 139 error_destroy:
 140         saHandleDestroy (&evs_handle_t_db, *handle);
 141 error_no_destroy:
 142         return (error);
 143 }
 144
 145 evs_error_t evs_finalize (
 146         evs_handle_t handle)
 147 {
 148         struct evs_inst *evs_inst;
 149         SaAisErrorT error;
 150
 151         error = saHandleInstanceGet (&evs_handle_t_db, handle, (void *)&evs_inst);
 152         if (error != SA_AIS_OK) {
 153                 return (error);
 154         }
 155 //        TODO is the locking right here
 156         pthread_mutex_lock (&evs_inst->response_mutex);
 157
 158         /*
 159          * Another thread has already started finalizing
 160          */
 161         if (evs_inst->finalize) {
 162                 pthread_mutex_unlock (&evs_inst->response_mutex);
 163                 saHandleInstancePut (&evs_handle_t_db, handle);
 164                 return (EVS_ERR_BAD_HANDLE);
 165         }
 166
 167         evs_inst->finalize = 1;
 168
 169         pthread_mutex_unlock (&evs_inst->response_mutex);
 170
 171         saHandleDestroy (&evs_handle_t_db, handle);
 172     /*
 173      * Disconnect from the server
 174      */
 175     if (evs_inst->response_fd != -1) {
 176         shutdown(evs_inst->response_fd, 0);
 177         close(evs_inst->response_fd);
 178     }
 179     if (evs_inst->dispatch_fd != -1) {
 180         shutdown(evs_inst->dispatch_fd, 0);
 181         close(evs_inst->dispatch_fd);
 182     }
 183         saHandleInstancePut (&evs_handle_t_db, handle);
 184
 185
 186         return (EVS_OK);
 187 }
 188
 189 evs_error_t evs_fd_get (
 190         evs_handle_t handle,
 191         int *fd)
 192 {
 193         SaAisErrorT error;
 194         struct evs_inst *evs_inst;
 195
 196         error = saHandleInstanceGet (&evs_handle_t_db, handle, (void *)&evs_inst);
 197         if (error != SA_AIS_OK) {
 198                 return (error);
 199         }
 200
 201         *fd = evs_inst->dispatch_fd;
 202
 203         saHandleInstancePut (&evs_handle_t_db, handle);
 204
 205         return (SA_AIS_OK);
 206 }
 207
 208 evs_error_t evs_dispatch (
 209         evs_handle_t handle,
 210         evs_dispatch_t dispatch_types)
 211 {
 212         struct pollfd ufds;
 213         int timeout = -1;
 214         SaAisErrorT error;
 215         int cont = 1; /* always continue do loop except when set to 0 */
 216         int dispatch_avail;
 217         struct evs_inst *evs_inst;
 218         struct res_evs_confchg_callback *res_evs_confchg_callback;
 219         struct res_evs_deliver_callback *res_evs_deliver_callback;
 220         evs_callbacks_t callbacks;
 221         struct res_overlay dispatch_data;
 222         int ignore_dispatch = 0;
 223
 224         error = saHandleInstanceGet (&evs_handle_t_db, handle, (void *)&evs_inst);
 225         if (error != SA_AIS_OK) {
 226                 return (error);
 227         }
 228
 229         /*
 230          * Timeout instantly for SA_DISPATCH_ONE or SA_DISPATCH_ALL and
 231          * wait indefinately for SA_DISPATCH_BLOCKING
 232          */
 233         if (dispatch_types == EVS_DISPATCH_ALL) {
 234                 timeout = 0;
 235         }
 236
 237         do {
 238                 ufds.fd = evs_inst->dispatch_fd;
 239                 ufds.events = POLLIN;
 240                 ufds.revents = 0;
 241
 242                 error = saPollRetry (&ufds, 1, timeout);
 243                 if (error != SA_AIS_OK) {
 244                         goto error_nounlock;
 245                 }
 246
 247                 pthread_mutex_lock (&evs_inst->dispatch_mutex);
 248
 249                 /*
 250                  * Regather poll data in case ufds has changed since taking lock
 251                  */
 252                 error = saPollRetry (&ufds, 1, 0);
 253                 if (error != SA_AIS_OK) {
 254                         goto error_nounlock;
 255                 }
 256
 257                 /*
 258                  * Handle has been finalized in another thread
 259                  */
 260                 if (evs_inst->finalize == 1) {
 261                         error = EVS_OK;
 262                         pthread_mutex_unlock (&evs_inst->dispatch_mutex);
 263                         goto error_unlock;
 264                 }
 265
 266                 dispatch_avail = ufds.revents & POLLIN;
 267                 if (dispatch_avail == 0 && dispatch_types == EVS_DISPATCH_ALL) {
 268                         pthread_mutex_unlock (&evs_inst->dispatch_mutex);
 269                         break; /* exit do while cont is 1 loop */
 270                 } else
 271                 if (dispatch_avail == 0) {
 272                         pthread_mutex_unlock (&evs_inst->dispatch_mutex);
 273                         continue; /* next poll */
 274                 }
 275
 276                 if (ufds.revents & POLLIN) {
 277                         /*
 278                          * Queue empty, read response from socket
 279                          */
 280                         error = saRecvRetry (evs_inst->dispatch_fd, &dispatch_data.header,
 281                                 sizeof (mar_res_header_t));
 282                         if (error != SA_AIS_OK) {
 283                                 goto error_unlock;
 284                         }
 285                         if (dispatch_data.header.size > sizeof (mar_res_header_t)) {
 286                                 error = saRecvRetry (evs_inst->dispatch_fd, &dispatch_data.data,
 287                                         dispatch_data.header.size - sizeof (mar_res_header_t));
 288
 289                                 if (error != SA_AIS_OK) {
 290                                         goto error_unlock;
 291                                 }
 292                         }
 293                 } else {
 294                         pthread_mutex_unlock (&evs_inst->dispatch_mutex);
 295                         continue;
 296                 }
 297
 298                 /*
 299                  * Make copy of callbacks, message data, unlock instance, and call callback
 300                  * A risk of this dispatch method is that the callback routines may
 301                  * operate at the same time that evsFinalize has been called.
 302                 */
 303                 memcpy (&callbacks, &evs_inst->callbacks, sizeof (evs_callbacks_t));
 304
 305                 pthread_mutex_unlock (&evs_inst->dispatch_mutex);
 306                 /*
 307                  * Dispatch incoming message
 308                  */
 309                 switch (dispatch_data.header.id) {
 310                 case MESSAGE_RES_EVS_DELIVER_CALLBACK:
 311                         res_evs_deliver_callback = (struct res_evs_deliver_callback *)&dispatch_data;
 312                         callbacks.evs_deliver_fn (
 313                                 res_evs_deliver_callback->local_nodeid,
 314                                 &res_evs_deliver_callback->msg,
 315                                 res_evs_deliver_callback->msglen);
 316                         break;
 317
 318                 case MESSAGE_RES_EVS_CONFCHG_CALLBACK:
 319                         res_evs_confchg_callback = (struct res_evs_confchg_callback *)&dispatch_data;
 320                         callbacks.evs_confchg_fn (
 321                                 res_evs_confchg_callback->member_list,
 322                                 res_evs_confchg_callback->member_list_entries,
 323                                 res_evs_confchg_callback->left_list,
 324                                 res_evs_confchg_callback->left_list_entries,
 325                                 res_evs_confchg_callback->joined_list,
 326                                 res_evs_confchg_callback->joined_list_entries);
 327                         break;
 328
 329                 default:
 330                         error = SA_AIS_ERR_LIBRARY;
 331                         goto error_nounlock;
 332                         break;
 333                 }
 334
 335                 /*
 336                  * Determine if more messages should be processed
 337                  * */
 338                 switch (dispatch_types) {
 339                 case EVS_DISPATCH_ONE:
 340                         if (ignore_dispatch) {
 341                                 ignore_dispatch = 0;
 342                         } else {
 343                                 cont = 0;
 344                         }
 345                         break;
 346                 case EVS_DISPATCH_ALL:
 347                         if (ignore_dispatch) {
 348                                 ignore_dispatch = 0;
 349                         }
 350                         break;
 351                 case EVS_DISPATCH_BLOCKING:
 352                         break;
 353                 }
 354         } while (cont);
 355
 356 error_unlock:
 357         saHandleInstancePut (&evs_handle_t_db, handle);
 358 error_nounlock:
 359         return (error);
 360 }
 361
 362 evs_error_t evs_join (
 363     evs_handle_t handle,
 364     struct evs_group *groups,
 365         int group_entries)
 366 {
 367         evs_error_t error;
 368         struct evs_inst *evs_inst;
 369         struct iovec iov[2];
 370         struct req_lib_evs_join req_lib_evs_join;
 371         struct res_lib_evs_join res_lib_evs_join;
 372
 373         error = saHandleInstanceGet (&evs_handle_t_db, handle, (void *)&evs_inst);
 374         if (error != SA_AIS_OK) {
 375                 return (error);
 376         }
 377
 378         req_lib_evs_join.header.size = sizeof (struct req_lib_evs_join) +
 379                 (group_entries * sizeof (struct evs_group));
 380         req_lib_evs_join.header.id = MESSAGE_REQ_EVS_JOIN;
 381         req_lib_evs_join.group_entries = group_entries;
 382
 383         iov[0].iov_base = (char *)&req_lib_evs_join;
 384         iov[0].iov_len = sizeof (struct req_lib_evs_join);
 385         iov[1].iov_base = (char *)groups;
 386         iov[1].iov_len = (group_entries * sizeof (struct evs_group));
 387
 388         pthread_mutex_lock (&evs_inst->response_mutex);
 389
 390         error = saSendMsgReceiveReply (evs_inst->response_fd, iov, 2,
 391                 &res_lib_evs_join, sizeof (struct res_lib_evs_join));
 392
 393         pthread_mutex_unlock (&evs_inst->response_mutex);
 394
 395         if (error != SA_AIS_OK) {
 396                 goto error_exit;
 397         }
 398
 399         error = res_lib_evs_join.header.error;
 400
 401 error_exit:
 402         saHandleInstancePut (&evs_handle_t_db, handle);
 403
 404         return (error);
 405 }
 406
 407 evs_error_t evs_leave (
 408     evs_handle_t handle,
 409     struct evs_group *groups,
 410         int group_entries)
 411 {
 412         evs_error_t error;
 413         struct evs_inst *evs_inst;
 414         struct iovec iov[2];
 415         struct req_lib_evs_leave req_lib_evs_leave;
 416         struct res_lib_evs_leave res_lib_evs_leave;
 417
 418         error = saHandleInstanceGet (&evs_handle_t_db, handle, (void *)&evs_inst);
 419         if (error != SA_AIS_OK) {
 420                 return (error);
 421         }
 422
 423         req_lib_evs_leave.header.size = sizeof (struct req_lib_evs_leave) +
 424                 (group_entries * sizeof (struct evs_group));
 425         req_lib_evs_leave.header.id = MESSAGE_REQ_EVS_LEAVE;
 426         req_lib_evs_leave.group_entries = group_entries;
 427
 428         iov[0].iov_base = (char *)&req_lib_evs_leave;
 429         iov[0].iov_len = sizeof (struct req_lib_evs_leave);
 430         iov[1].iov_base = (char *)groups;
 431         iov[1].iov_len = (group_entries * sizeof (struct evs_group));
 432
 433         pthread_mutex_lock (&evs_inst->response_mutex);
 434
 435         error = saSendMsgReceiveReply (evs_inst->response_fd, iov, 2,
 436                 &res_lib_evs_leave, sizeof (struct res_lib_evs_leave));
 437
 438         pthread_mutex_unlock (&evs_inst->response_mutex);
 439
 440         if (error != SA_AIS_OK) {
 441                 goto error_exit;
 442         }
 443
 444         error = res_lib_evs_leave.header.error;
 445
 446 error_exit:
 447         saHandleInstancePut (&evs_handle_t_db, handle);
 448
 449         return (error);
 450 }
 451
 452 evs_error_t evs_mcast_joined (
 453         evs_handle_t handle,
 454         evs_guarantee_t guarantee,
 455         struct iovec *iovec,
 456         int iov_len)
 457 {
 458         int i;
 459         evs_error_t error;
 460         struct evs_inst *evs_inst;
 461         struct iovec iov[64];
 462         struct req_lib_evs_mcast_joined req_lib_evs_mcast_joined;
 463         struct res_lib_evs_mcast_joined res_lib_evs_mcast_joined;
 464         int msg_len = 0;
 465
 466         error = saHandleInstanceGet (&evs_handle_t_db, handle, (void *)&evs_inst);
 467         if (error != SA_AIS_OK) {
 468                 return (error);
 469         }
 470
 471         for (i = 0; i < iov_len; i++ ) {
 472                 msg_len += iovec[i].iov_len;
 473         }
 474
 475         req_lib_evs_mcast_joined.header.size = sizeof (struct req_lib_evs_mcast_joined) +
 476                 msg_len;
 477
 478         req_lib_evs_mcast_joined.header.id = MESSAGE_REQ_EVS_MCAST_JOINED;
 479         req_lib_evs_mcast_joined.guarantee = guarantee;
 480         req_lib_evs_mcast_joined.msg_len = msg_len;
 481
 482         iov[0].iov_base = (char *)&req_lib_evs_mcast_joined;
 483         iov[0].iov_len = sizeof (struct req_lib_evs_mcast_joined);
 484         memcpy (&iov[1], iovec, iov_len * sizeof (struct iovec));
 485
 486         pthread_mutex_lock (&evs_inst->response_mutex);
 487
 488         error = saSendMsgReceiveReply (evs_inst->response_fd, iov, iov_len + 1,
 489                 &res_lib_evs_mcast_joined, sizeof (struct res_lib_evs_mcast_joined));
 490
 491         pthread_mutex_unlock (&evs_inst->response_mutex);
 492
 493         if (error != SA_AIS_OK) {
 494                 goto error_exit;
 495         }
 496
 497         error = res_lib_evs_mcast_joined.header.error;
 498
 499 error_exit:
 500         saHandleInstancePut (&evs_handle_t_db, handle);
 501
 502         return (error);
 503 }
 504
 505 evs_error_t evs_mcast_groups (
 506         evs_handle_t handle,
 507         evs_guarantee_t guarantee,
 508         struct evs_group *groups,
 509         int group_entries,
 510         struct iovec *iovec,
 511         int iov_len)
 512 {
 513         int i;
 514         evs_error_t error;
 515         struct evs_inst *evs_inst;
 516         struct iovec iov[64];
 517         struct req_lib_evs_mcast_groups req_lib_evs_mcast_groups;
 518         struct res_lib_evs_mcast_groups res_lib_evs_mcast_groups;
 519         int msg_len = 0;
 520
 521         error = saHandleInstanceGet (&evs_handle_t_db, handle, (void *)&evs_inst);
 522         if (error != SA_AIS_OK) {
 523                 return (error);
 524         }
 525         for (i = 0; i < iov_len; i++) {
 526                 msg_len += iovec[i].iov_len;
 527         }
 528         req_lib_evs_mcast_groups.header.size = sizeof (struct req_lib_evs_mcast_groups) +
 529                 (group_entries * sizeof (struct evs_group)) + msg_len;
 530         req_lib_evs_mcast_groups.header.id = MESSAGE_REQ_EVS_MCAST_GROUPS;
 531         req_lib_evs_mcast_groups.guarantee = guarantee;
 532         req_lib_evs_mcast_groups.msg_len = msg_len;
 533         req_lib_evs_mcast_groups.group_entries = group_entries;
 534
 535         iov[0].iov_base = (char *)&req_lib_evs_mcast_groups;
 536         iov[0].iov_len = sizeof (struct req_lib_evs_mcast_groups);
 537         iov[1].iov_base = (char *)groups;
 538         iov[1].iov_len = (group_entries * sizeof (struct evs_group));
 539         memcpy (&iov[2], iovec, iov_len * sizeof (struct iovec));
 540
 541         pthread_mutex_lock (&evs_inst->response_mutex);
 542
 543         error = saSendMsgReceiveReply (evs_inst->response_fd, iov, iov_len + 2,
 544                 &res_lib_evs_mcast_groups, sizeof (struct res_lib_evs_mcast_groups));
 545
 546         pthread_mutex_unlock (&evs_inst->response_mutex);
 547         if (error != SA_AIS_OK) {
 548                 goto error_exit;
 549         }
 550
 551         error = res_lib_evs_mcast_groups.header.error;
 552
 553 error_exit:
 554         saHandleInstancePut (&evs_handle_t_db, handle);
 555
 556         return (error);
 557 }
 558
 559 evs_error_t evs_membership_get (
 560         evs_handle_t handle,
 561         unsigned int *local_nodeid,
 562         unsigned int *member_list,
 563         unsigned int *member_list_entries)
 564 {
 565         evs_error_t error;
 566         struct evs_inst *evs_inst;
 567         struct iovec iov;
 568         struct req_lib_evs_membership_get req_lib_evs_membership_get;
 569         struct res_lib_evs_membership_get res_lib_evs_membership_get;
 570
 571         error = saHandleInstanceGet (&evs_handle_t_db, handle, (void *)&evs_inst);
 572         if (error != SA_AIS_OK) {
 573                 return (error);
 574         }
 575
 576         req_lib_evs_membership_get.header.size = sizeof (struct req_lib_evs_membership_get);
 577         req_lib_evs_membership_get.header.id = MESSAGE_REQ_EVS_MEMBERSHIP_GET;
 578
 579         iov.iov_base = (char *)&req_lib_evs_membership_get;
 580         iov.iov_len = sizeof (struct req_lib_evs_membership_get);
 581
 582         pthread_mutex_lock (&evs_inst->response_mutex);
 583
 584         error = saSendMsgReceiveReply (evs_inst->response_fd, &iov, 1,
 585                 &res_lib_evs_membership_get, sizeof (struct res_lib_evs_membership_get));
 586
 587         pthread_mutex_unlock (&evs_inst->response_mutex);
 588
 589         if (error != SA_AIS_OK) {
 590                 goto error_exit;
 591         }
 592
 593         error = res_lib_evs_membership_get.header.error;
 594
 595         /*
 596          * Copy results to caller
 597          */
 598         if (local_nodeid) {
 599                 *local_nodeid = res_lib_evs_membership_get.local_nodeid;
 600         }
 601         *member_list_entries = *member_list_entries < res_lib_evs_membership_get.member_list_entries ?
 602                 *member_list_entries : res_lib_evs_membership_get.member_list_entries;
 603         if (member_list) {
 604                 memcpy (member_list, &res_lib_evs_membership_get.member_list,
 605                         *member_list_entries * sizeof (struct in_addr));
 606         }
 607
 608 error_exit:
 609         saHandleInstancePut (&evs_handle_t_db, handle);
 610
 611         return (error);
 612 }
 613
 614 /** @} */