exec/lck.c

   1 /*
   2  * Copyright (c) 2005-2006 MontaVista Software, Inc.
   3  * Copyright (c) 2006 Red Hat, Inc.
   4  * Copyright (c) 2006 Sun Microsystems, Inc.
   5  *
   6  * All rights reserved.
   7  *
   8  * Author: Steven Dake (sdake@mvista.com)
   9  *
  10  * This software licensed under BSD license, the text of which follows:
  11  *
  12  * Redistribution and use in source and binary forms, with or without
  13  * modification, are permitted provided that the following conditions are met:
  14  *
  15  * - Redistributions of source code must retain the above copyright notice,
  16  *   this list of conditions and the following disclaimer.
  17  * - Redistributions in binary form must reproduce the above copyright notice,
  18  *   this list of conditions and the following disclaimer in the documentation
  19  *   and/or other materials provided with the distribution.
  20  * - Neither the name of the MontaVista Software, Inc. nor the names of its
  21  *   contributors may be used to endorse or promote products derived from this
  22  *   software without specific prior written permission.
  23  *
  24  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  25  * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  26  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  27  * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
  28  * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  29  * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  30  * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
  31  * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
  32  * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  33  * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
  34  * THE POSSIBILITY OF SUCH DAMAGE.
  35  */
  36
  37 #include <sys/types.h>
  38 #include <sys/uio.h>
  39 #include <sys/socket.h>
  40 #include <sys/un.h>
  41 #include <sys/time.h>
  42 #include <netinet/in.h>
  43 #include <unistd.h>
  44 #include <fcntl.h>
  45 #include <stdlib.h>
  46 #include <stdio.h>
  47 #include <errno.h>
  48 #include <signal.h>
  49 #include <arpa/inet.h>
  50
  51 #include "swab.h"
  52 #include "objdb.h"
  53 #include "totem.h"
  54 #include "service.h"
  55 #include "../include/saAis.h"
  56 #include "../include/saLck.h"
  57 #include "../include/ipc_lck.h"
  58 #include "../include/list.h"
  59 #include "../include/queue.h"
  60 #include "../lcr/lcr_comp.h"
  61 #include "mempool.h"
  62 #include "util.h"
  63 #include "main.h"
  64 #include "flow.h"
  65 #include "tlist.h"
  66 #include "ipc.h"
  67 #include "totempg.h"
  68 #include "logsys.h"
  69
  70 LOGSYS_DECLARE_SUBSYS ("LCK", LOG_INFO);
  71
  72 enum lck_message_req_types {
  73         MESSAGE_REQ_EXEC_LCK_RESOURCEOPEN = 0,
  74         MESSAGE_REQ_EXEC_LCK_RESOURCECLOSE = 1,
  75         MESSAGE_REQ_EXEC_LCK_RESOURCELOCK = 2,
  76         MESSAGE_REQ_EXEC_LCK_RESOURCEUNLOCK = 3,
  77         MESSAGE_REQ_EXEC_LCK_RESOURCELOCKORPHAN = 4,
  78         MESSAGE_REQ_EXEC_LCK_LOCKPURGE = 5
  79 };
  80
  81 struct resource;
  82 struct resource_lock {
  83         SaLckLockModeT lock_mode;
  84         SaLckLockIdT lock_id;
  85         SaLckLockFlagsT lock_flags;
  86         SaLckWaiterSignalT waiter_signal;
  87         SaLckLockStatusT lock_status;
  88         SaTimeT timeout;
  89         struct resource *resource;
  90         int async_call;
  91         SaInvocationT invocation;
  92         mar_message_source_t callback_source;
  93         mar_message_source_t response_source;
  94         struct list_head list; /* locked resource lock list */
  95         struct list_head resource_list; /* resource locks on a resource */
  96         struct list_head resource_cleanup_list; /* cleanup data for resource locks */
  97 };
  98
  99 struct resource {
 100         mar_name_t name;
 101         int refcount;
 102         struct list_head list;
 103         struct list_head resource_lock_list_head;
 104         struct list_head pr_granted_list_head;
 105         struct list_head pr_pending_list_head;
 106         struct list_head ex_pending_list_head;
 107         struct resource_lock *ex_granted;
 108 };
 109
 110 struct resource_cleanup {
 111         struct resource *resource;
 112         SaLckResourceHandleT resource_handle;
 113         struct list_head resource_lock_list_head;
 114         struct list_head list;
 115 };
 116
 117 DECLARE_LIST_INIT(resource_list_head);
 118
 119 static int lck_exec_init_fn (struct objdb_iface_ver0 *objdb);
 120
 121 static int lck_lib_exit_fn (void *conn);
 122
 123 static int lck_lib_init_fn (void *conn);
 124
 125 static void message_handler_req_exec_lck_resourceopen (
 126         void *message,
 127         unsigned int nodeid);
 128
 129 static void message_handler_req_exec_lck_resourceclose (
 130         void *message,
 131         unsigned int nodeid);
 132
 133 static void message_handler_req_exec_lck_resourcelock (
 134         void *message,
 135         unsigned int nodeid);
 136
 137 static void message_handler_req_exec_lck_resourceunlock (
 138         void *message,
 139         unsigned int nodeid);
 140
 141 static void message_handler_req_exec_lck_resourcelockorphan (
 142         void *message,
 143         unsigned int nodeid);
 144
 145 static void message_handler_req_exec_lck_lockpurge (
 146         void *message,
 147         unsigned int nodeid);
 148
 149 static void message_handler_req_lib_lck_resourceopen (
 150         void *conn,
 151         void *msg);
 152
 153 static void message_handler_req_lib_lck_resourceopenasync (
 154         void *conn,
 155         void *msg);
 156
 157 static void message_handler_req_lib_lck_resourceclose (
 158         void *conn,
 159         void *msg);
 160
 161 static void message_handler_req_lib_lck_resourcelock (
 162         void *conn,
 163         void *msg);
 164
 165 static void message_handler_req_lib_lck_resourcelockasync (
 166         void *conn,
 167         void *msg);
 168
 169 static void message_handler_req_lib_lck_resourceunlock (
 170         void *conn,
 171         void *msg);
 172
 173 static void message_handler_req_lib_lck_resourceunlockasync (
 174         void *conn,
 175         void *msg);
 176
 177 static void message_handler_req_lib_lck_lockpurge (
 178         void *conn,
 179         void *msg);
 180
 181 static void exec_lck_resourceopen_endian_convert (void *msg);
 182
 183 static void exec_lck_resourceclose_endian_convert (void *msg);
 184
 185 static void exec_lck_resourcelock_endian_convert (void *msg);
 186
 187 static void exec_lck_resourceunlock_endian_convert (void *msg);
 188
 189 static void exec_lck_resourcelockorphan_endian_convert (void *msg);
 190
 191 static void exec_lck_lockpurge_endian_convert (void *msg);
 192
 193 #ifdef TODO
 194 static void lck_sync_init (void);
 195 #endif
 196 static int lck_sync_process (void);
 197 static void lck_sync_activate (void);
 198 static void lck_sync_abort (void);
 199
 200 void resource_release (struct resource *resource);
 201
 202 /*
 203 static struct list_head *recovery_lck_next = 0;
 204 static struct list_head *recovery_lck_section_next = 0;
 205 static int recovery_section_data_offset = 0;
 206 static int recovery_section_send_flag = 0;
 207 static int recovery_abort = 0;
 208 //static struct memb_ring_id saved_ring_id;
 209 */
 210
 211
 212 static void lck_confchg_fn (
 213         enum totem_configuration_type configuration_type,
 214         unsigned int *member_list, int member_list_entries,
 215         unsigned int *left_list, int left_list_entries,
 216         unsigned int *joined_list, int joined_list_entries,
 217         struct memb_ring_id *ring_id);
 218
 219 struct lck_pd {
 220         struct list_head resource_list;
 221         struct list_head resource_cleanup_list;
 222 };
 223
 224
 225 /*
 226  * Executive Handler Definition
 227  */
 228 static struct openais_lib_handler lck_lib_service[] =
 229 {
 230         { /* 0 */
 231                 .lib_handler_fn         = message_handler_req_lib_lck_resourceopen,
 232                 .response_size          = sizeof (struct res_lib_lck_resourceopen),
 233                 .response_id            = MESSAGE_RES_LCK_RESOURCEOPEN,
 234                 .flow_control           = OPENAIS_FLOW_CONTROL_REQUIRED
 235         },
 236         { /* 1 */
 237                 .lib_handler_fn         = message_handler_req_lib_lck_resourceopenasync,
 238                 .response_size          = sizeof (struct res_lib_lck_resourceopenasync),
 239                 .response_id            = MESSAGE_RES_LCK_RESOURCEOPENASYNC,
 240                 .flow_control           = OPENAIS_FLOW_CONTROL_REQUIRED
 241         },
 242         { /* 2 */
 243                 .lib_handler_fn         = message_handler_req_lib_lck_resourceclose,
 244                 .response_size          = sizeof (struct res_lib_lck_resourceclose),
 245                 .response_id            = MESSAGE_RES_LCK_RESOURCECLOSE,
 246                 .flow_control           = OPENAIS_FLOW_CONTROL_REQUIRED
 247         },
 248         { /* 3 */
 249                 .lib_handler_fn         = message_handler_req_lib_lck_resourcelock,
 250                 .response_size          = sizeof (struct res_lib_lck_resourcelock),
 251                 .response_id            = MESSAGE_RES_LCK_RESOURCELOCK,
 252                 .flow_control           = OPENAIS_FLOW_CONTROL_REQUIRED
 253         },
 254         { /* 4 */
 255                 .lib_handler_fn         = message_handler_req_lib_lck_resourcelockasync,
 256                 .response_size          = sizeof (struct res_lib_lck_resourcelockasync),
 257                 .response_id            = MESSAGE_RES_LCK_RESOURCELOCKASYNC,
 258                 .flow_control           = OPENAIS_FLOW_CONTROL_REQUIRED
 259         },
 260         { /* 5 */
 261                 .lib_handler_fn         = message_handler_req_lib_lck_resourceunlock,
 262                 .response_size          = sizeof (struct res_lib_lck_resourceunlock),
 263                 .response_id            = MESSAGE_RES_LCK_RESOURCELOCK,
 264                 .flow_control           = OPENAIS_FLOW_CONTROL_REQUIRED
 265         },
 266         { /* 6 */
 267                 .lib_handler_fn         = message_handler_req_lib_lck_resourceunlockasync,
 268                 .response_size          = sizeof (struct res_lib_lck_resourceunlock),
 269                 .response_id            = MESSAGE_RES_LCK_RESOURCEUNLOCKASYNC,
 270                 .flow_control           = OPENAIS_FLOW_CONTROL_REQUIRED
 271         },
 272         { /* 7 */
 273                 .lib_handler_fn         = message_handler_req_lib_lck_lockpurge,
 274                 .response_size          = sizeof (struct res_lib_lck_lockpurge),
 275                 .response_id            = MESSAGE_RES_LCK_LOCKPURGE,
 276                 .flow_control           = OPENAIS_FLOW_CONTROL_REQUIRED
 277         }
 278 };
 279
 280
 281 static struct openais_exec_handler lck_exec_service[] = {
 282         {
 283                 .exec_handler_fn        = message_handler_req_exec_lck_resourceopen,
 284                 .exec_endian_convert_fn = exec_lck_resourceopen_endian_convert
 285         },
 286         {
 287                 .exec_handler_fn        = message_handler_req_exec_lck_resourceclose,
 288                 .exec_endian_convert_fn = exec_lck_resourceclose_endian_convert
 289         },
 290         {
 291                 .exec_handler_fn        = message_handler_req_exec_lck_resourcelock,
 292                 .exec_endian_convert_fn = exec_lck_resourcelock_endian_convert
 293         },
 294         {
 295                 .exec_handler_fn        = message_handler_req_exec_lck_resourceunlock,
 296                 .exec_endian_convert_fn = exec_lck_resourceunlock_endian_convert
 297         },
 298         {
 299                 .exec_handler_fn        = message_handler_req_exec_lck_resourcelockorphan,
 300                 .exec_endian_convert_fn = exec_lck_resourcelockorphan_endian_convert
 301         },
 302         {
 303                 .exec_handler_fn        = message_handler_req_exec_lck_lockpurge,
 304                 .exec_endian_convert_fn = exec_lck_lockpurge_endian_convert
 305         }
 306 };
 307
 308 struct openais_service_handler lck_service_handler = {
 309         .name                           = "openais distributed locking service B.01.01",
 310         .id                             = LCK_SERVICE,
 311         .private_data_size              = sizeof (struct lck_pd),
 312         .flow_control                   = OPENAIS_FLOW_CONTROL_NOT_REQUIRED,
 313         .lib_init_fn                    = lck_lib_init_fn,
 314         .lib_exit_fn                    = lck_lib_exit_fn,
 315         .lib_service                    = lck_lib_service,
 316         .lib_service_count              = sizeof (lck_lib_service) / sizeof (struct openais_lib_handler),
 317         .exec_init_fn                   = lck_exec_init_fn,
 318         .exec_service                   = lck_exec_service,
 319         .exec_service_count             = sizeof (lck_exec_service) / sizeof (struct openais_exec_handler),
 320         .exec_dump_fn                   = NULL,
 321         .confchg_fn                     = lck_confchg_fn,
 322         .sync_init                      = NULL,
 323 //      .sync_init                      = lck_sync_init,
 324         .sync_process                   = lck_sync_process,
 325         .sync_activate                  = lck_sync_activate,
 326         .sync_abort                     = lck_sync_abort,
 327 };
 328
 329 /*
 330  * Dynamic loader definition
 331  */
 332 static struct openais_service_handler *lck_get_handler_ver0 (void);
 333
 334 static struct openais_service_handler_iface_ver0 lck_service_handler_iface = {
 335         .openais_get_service_handler_ver0       = lck_get_handler_ver0
 336 };
 337
 338 static struct lcr_iface openais_lck_ver0[1] = {
 339         {
 340                 .name                           = "openais_lck",
 341                 .version                        = 0,
 342                 .versions_replace               = 0,
 343                 .versions_replace_count         = 0,
 344                 .dependencies                   = 0,
 345                 .dependency_count               = 0,
 346                 .constructor                    = NULL,
 347                 .destructor                     = NULL,
 348                 .interfaces                     = (void **)(void *)&lck_service_handler_iface,
 349         }
 350 };
 351
 352 static struct lcr_comp lck_comp_ver0 = {
 353         .iface_count                            = 1,
 354         .ifaces                                 = openais_lck_ver0
 355 };
 356
 357 static struct openais_service_handler *lck_get_handler_ver0 (void)
 358 {
 359         return (&lck_service_handler);
 360 }
 361
 362 __attribute__ ((constructor)) static void register_this_component (void) {
 363         lcr_interfaces_set (&openais_lck_ver0[0], &lck_service_handler_iface);
 364
 365         lcr_component_register (&lck_comp_ver0);
 366 }
 367
 368 /*
 369  * All data types used for executive messages
 370  */
 371 struct req_exec_lck_resourceopen {
 372         mar_req_header_t header;
 373         mar_message_source_t source;
 374         mar_name_t resource_name;
 375         SaLckResourceHandleT resource_handle;
 376         SaInvocationT invocation;
 377         SaTimeT timeout;
 378         SaLckResourceOpenFlagsT open_flags;
 379         int async_call;
 380         SaAisErrorT fail_with_error;
 381 };
 382
 383 static void exec_lck_resourceopen_endian_convert (void *msg)
 384 {
 385         struct req_exec_lck_resourceopen *to_swab =
 386                 (struct req_exec_lck_resourceopen *)msg;
 387
 388         swab_mar_req_header_t (&to_swab->header);
 389         swab_mar_message_source_t (&to_swab->source);
 390         swab_mar_name_t (&to_swab->resource_name);
 391         to_swab->resource_handle = swab64 (to_swab->resource_handle);
 392         to_swab->invocation = swab64 (to_swab->invocation);
 393         to_swab->timeout = swab64 (to_swab->timeout);
 394         to_swab->open_flags = swab32 (to_swab->open_flags);
 395         to_swab->async_call = swab32 (to_swab->async_call);
 396         to_swab->fail_with_error = swab32 (to_swab->fail_with_error);
 397 }
 398
 399 struct req_exec_lck_resourceclose {
 400         mar_req_header_t header;
 401         mar_message_source_t source;
 402         mar_name_t lockResourceName;
 403         SaLckResourceHandleT resource_handle;
 404 };
 405
 406 static void exec_lck_resourceclose_endian_convert (void *msg)
 407 {
 408         struct req_exec_lck_resourceclose *to_swab =
 409                 (struct req_exec_lck_resourceclose *)msg;
 410
 411         swab_mar_req_header_t (&to_swab->header);
 412         swab_mar_message_source_t (&to_swab->source);
 413         swab_mar_name_t (&to_swab->lockResourceName);
 414         to_swab->resource_handle = swab64 (to_swab->resource_handle);
 415 }
 416
 417 struct req_exec_lck_resourcelock {
 418         mar_req_header_t header;
 419         SaLckResourceHandleT resource_handle;
 420         SaInvocationT invocation;
 421         int async_call;
 422         SaAisErrorT fail_with_error;
 423         mar_message_source_t source;
 424         struct req_lib_lck_resourcelock req_lib_lck_resourcelock;
 425 };
 426
 427 static void exec_lck_resourcelock_endian_convert (void *msg)
 428 {
 429         struct req_exec_lck_resourcelock *to_swab =
 430                 (struct req_exec_lck_resourcelock *)msg;
 431
 432         swab_mar_req_header_t (&to_swab->header);
 433         to_swab->resource_handle = swab64 (to_swab->resource_handle);
 434         to_swab->invocation = swab64 (to_swab->invocation);
 435         to_swab->async_call = swab32 (to_swab->async_call);
 436         to_swab->fail_with_error = swab32 (to_swab->fail_with_error);
 437         swab_mar_message_source_t (&to_swab->source);
 438         swab_req_lib_lck_resourcelock (&to_swab->req_lib_lck_resourcelock);
 439 }
 440
 441 struct req_exec_lck_resourceunlock {
 442         mar_req_header_t header;
 443         mar_message_source_t source;
 444         mar_name_t resource_name;
 445         SaLckLockIdT lock_id;
 446         SaInvocationT invocation;
 447         SaTimeT timeout;
 448         int async_call;
 449 };
 450
 451 static void exec_lck_resourceunlock_endian_convert (void *msg)
 452 {
 453         struct req_exec_lck_resourceunlock *to_swab =
 454                 (struct req_exec_lck_resourceunlock *)msg;
 455
 456         swab_mar_req_header_t (&to_swab->header);
 457         swab_mar_message_source_t (&to_swab->source);
 458         swab_mar_name_t (&to_swab->resource_name);
 459         to_swab->lock_id = swab64 (to_swab->lock_id);
 460         to_swab->invocation = swab64 (to_swab->invocation);
 461         to_swab->timeout = swab64 (to_swab->timeout);
 462         to_swab->async_call = swab32 (to_swab->async_call);
 463 }
 464
 465 struct req_exec_lck_resourcelockorphan {
 466         mar_req_header_t header;
 467         mar_message_source_t source;
 468         mar_name_t resource_name;
 469         SaLckLockIdT lock_id;
 470 };
 471
 472 static void exec_lck_resourcelockorphan_endian_convert (void *msg)
 473 {
 474         struct req_exec_lck_resourcelockorphan *to_swab =
 475                 (struct req_exec_lck_resourcelockorphan *)msg;
 476
 477         swab_mar_req_header_t (&to_swab->header);
 478         swab_mar_message_source_t (&to_swab->source);
 479         swab_mar_name_t (&to_swab->resource_name);
 480         to_swab->lock_id = swab64 (to_swab->lock_id);
 481 }
 482
 483 struct req_exec_lck_lockpurge {
 484         mar_req_header_t header;
 485         mar_message_source_t source;
 486         struct req_lib_lck_lockpurge req_lib_lck_lockpurge;
 487 };
 488
 489 static void exec_lck_lockpurge_endian_convert (void *msg)
 490 {
 491         struct req_exec_lck_lockpurge *to_swab =
 492                 (struct req_exec_lck_lockpurge *)msg;
 493
 494         swab_mar_req_header_t (&to_swab->header);
 495         swab_mar_message_source_t (&to_swab->source);
 496         swab_req_lib_lck_lockpurge (&to_swab->req_lib_lck_lockpurge);
 497 }
 498
 499 #ifdef TODO
 500 static void lck_sync_init (void)
 501 {
 502         return;
 503 }
 504 #endif
 505
 506 static int lck_sync_process (void)
 507 {
 508         return (0);
 509 }
 510
 511 static void lck_sync_activate (void)
 512 {
 513         return;
 514 }
 515
 516 static void lck_sync_abort (void)
 517 {
 518         return;
 519 }
 520
 521 static void lck_confchg_fn (
 522         enum totem_configuration_type configuration_type,
 523         unsigned int *member_list, int member_list_entries,
 524         unsigned int *left_list, int left_list_entries,
 525         unsigned int *joined_list, int joined_list_entries,
 526         struct memb_ring_id *ring_id)
 527 {
 528 }
 529
 530 static struct resource *resource_find (mar_name_t *name)
 531 {
 532         struct list_head *resource_list;
 533         struct resource *resource;
 534
 535    for (resource_list = resource_list_head.next;
 536         resource_list != &resource_list_head;
 537         resource_list = resource_list->next) {
 538
 539         resource = list_entry (resource_list,
 540             struct resource, list);
 541
 542                 if (mar_name_match (name, &resource->name)) {
 543                         return (resource);
 544                 }
 545         }
 546         return (0);
 547 }
 548
 549 static struct resource_lock *resource_lock_find (
 550         struct resource *resource,
 551         mar_message_source_t *source,
 552         SaLckLockIdT lock_id)
 553 {
 554         struct list_head *list;
 555         struct resource_lock *resource_lock;
 556
 557         for (list = resource->resource_lock_list_head.next;
 558                 list != &resource->resource_lock_list_head;
 559                 list = list->next) {
 560
 561                 resource_lock = list_entry (list, struct resource_lock, resource_list);
 562
 563                 if ((memcmp (&resource_lock->callback_source,
 564                         source, sizeof (mar_message_source_t)) == 0) &&
 565                         (lock_id == resource_lock->lock_id)) {
 566
 567                         return (resource_lock);
 568                 }
 569         }
 570         return (0);
 571 }
 572
 573 struct resource_cleanup *lck_resource_cleanup_find (
 574         void *conn,
 575         SaLckResourceHandleT resource_handle)
 576 {
 577         struct list_head *list;
 578         struct resource_cleanup *resource_cleanup;
 579         struct lck_pd *lck_pd = (struct lck_pd *)openais_conn_private_data_get (conn);
 580
 581         for (list = lck_pd->resource_cleanup_list.next;
 582                 list != &lck_pd->resource_cleanup_list; list = list->next) {
 583
 584                 resource_cleanup = list_entry (list, struct resource_cleanup, list);
 585                 if (resource_cleanup->resource_handle == resource_handle) {
 586                         return (resource_cleanup);
 587                 }
 588         }
 589         return (0);
 590 }
 591
 592
 593 int lck_resource_close (struct resource *resource)
 594 {
 595         struct req_exec_lck_resourceclose req_exec_lck_resourceclose;
 596         struct iovec iovec;
 597
 598         req_exec_lck_resourceclose.header.size =
 599                 sizeof (struct req_exec_lck_resourceclose);
 600         req_exec_lck_resourceclose.header.id =
 601                 SERVICE_ID_MAKE (LCK_SERVICE, MESSAGE_REQ_EXEC_LCK_RESOURCECLOSE);
 602
 603         memcpy (&req_exec_lck_resourceclose.lockResourceName,
 604                 &resource->name, sizeof (mar_name_t));
 605
 606         iovec.iov_base = (char *)&req_exec_lck_resourceclose;
 607         iovec.iov_len = sizeof (req_exec_lck_resourceclose);
 608
 609         if (totempg_groups_send_ok_joined (openais_group_handle, &iovec, 1)) {
 610                 assert (totempg_groups_mcast_joined (openais_group_handle, &iovec, 1, TOTEMPG_AGREED) == 0);
 611                 return (0);
 612         }
 613
 614         return (-1);
 615 }
 616
 617 void resource_lock_orphan (struct resource_lock *resource_lock)
 618 {
 619         struct req_exec_lck_resourcelockorphan req_exec_lck_resourcelockorphan;
 620         struct iovec iovec;
 621
 622         req_exec_lck_resourcelockorphan.header.size =
 623                 sizeof (struct req_exec_lck_resourcelockorphan);
 624         req_exec_lck_resourcelockorphan.header.id =
 625                 SERVICE_ID_MAKE (LCK_SERVICE, MESSAGE_REQ_EXEC_LCK_RESOURCELOCKORPHAN);
 626
 627         memcpy (&req_exec_lck_resourcelockorphan.source,
 628                 &resource_lock->callback_source,
 629                 sizeof (mar_message_source_t));
 630
 631         memcpy (&req_exec_lck_resourcelockorphan.resource_name,
 632                 &resource_lock->resource->name,
 633                 sizeof (mar_name_t));
 634
 635         req_exec_lck_resourcelockorphan.lock_id = resource_lock->lock_id;
 636
 637         iovec.iov_base = (char *)&req_exec_lck_resourcelockorphan;
 638         iovec.iov_len = sizeof (req_exec_lck_resourcelockorphan);
 639
 640         assert (totempg_groups_mcast_joined (openais_group_handle, &iovec, 1, TOTEMPG_AGREED) == 0);
 641 }
 642
 643 void lck_resource_cleanup_lock_remove (
 644         struct resource_cleanup *resource_cleanup)
 645 {
 646         struct list_head *list;
 647         struct resource_lock *resource_lock;
 648
 649         for (list = resource_cleanup->resource_lock_list_head.next;
 650                 list != &resource_cleanup->resource_lock_list_head;
 651                 list = list->next) {
 652
 653                 resource_lock = list_entry (list, struct resource_lock, resource_cleanup_list);
 654                 resource_lock_orphan (resource_lock);
 655         }
 656 }
 657
 658 void lck_resource_cleanup_remove (
 659         void *conn,
 660         SaLckResourceHandleT resource_handle)
 661 {
 662
 663         struct list_head *list;
 664         struct resource_cleanup *resource_cleanup;
 665         struct lck_pd *lck_pd = (struct lck_pd *)openais_conn_private_data_get (conn);
 666
 667         for (list = lck_pd->resource_cleanup_list.next;
 668                 list != &lck_pd->resource_cleanup_list;
 669                 list = list->next) {
 670
 671                 resource_cleanup = list_entry (list, struct resource_cleanup, list);
 672                 if (resource_cleanup->resource_handle == resource_handle) {
 673                         list_del (&resource_cleanup->list);
 674                         free (resource_cleanup);
 675                         return;
 676                 }
 677         }
 678 }
 679
 680
 681 static int lck_exec_init_fn (struct objdb_iface_ver0 *objdb)
 682 {
 683         /*
 684          *  Initialize the saved ring ID.
 685          */
 686         return (0);
 687 }
 688
 689 static int lck_lib_exit_fn (void *conn)
 690 {
 691         struct resource_cleanup *resource_cleanup;
 692         struct list_head *cleanup_list;
 693         struct lck_pd *lck_pd = (struct lck_pd *)openais_conn_private_data_get (conn);
 694
 695         log_printf(LOG_LEVEL_NOTICE, "lck_exit_fn conn_info %p\n", conn);
 696
 697         /*
 698          * close all resources opened on this fd
 699          */
 700         cleanup_list = lck_pd->resource_cleanup_list.next;
 701         while (!list_empty(cleanup_list)) {
 702
 703                 resource_cleanup = list_entry (cleanup_list, struct resource_cleanup, list);
 704
 705                 if (resource_cleanup->resource->name.length > 0) {
 706                         lck_resource_cleanup_lock_remove (resource_cleanup);
 707                         lck_resource_close (resource_cleanup->resource);
 708                 }
 709
 710                 list_del (&resource_cleanup->list);
 711                 free (resource_cleanup);
 712
 713                 cleanup_list = lck_pd->resource_cleanup_list.next;
 714         }
 715
 716         return (0);
 717 }
 718
 719 static int lck_lib_init_fn (void *conn)
 720 {
 721     struct lck_pd *lck_pd = (struct lck_pd *)openais_conn_private_data_get (conn);
 722
 723         list_init (&lck_pd->resource_list);
 724         list_init (&lck_pd->resource_cleanup_list);
 725         return (0);
 726 }
 727
 728 static void message_handler_req_exec_lck_resourceopen (
 729         void *message,
 730         unsigned int nodeid)
 731 {
 732         struct req_exec_lck_resourceopen *req_exec_lck_resourceopen = (struct req_exec_lck_resourceopen *)message;
 733         struct res_lib_lck_resourceopen res_lib_lck_resourceopen;
 734         struct res_lib_lck_resourceopenasync res_lib_lck_resourceopenasync;
 735         struct resource *resource;
 736         struct resource_cleanup *resource_cleanup;
 737         SaAisErrorT error = SA_AIS_OK;
 738         struct lck_pd *lck_pd;
 739
 740         log_printf (LOG_LEVEL_NOTICE, "EXEC request: saLckResourceOpen %s\n",
 741                 get_mar_name_t (&req_exec_lck_resourceopen->resource_name));
 742
 743         if (req_exec_lck_resourceopen->fail_with_error != SA_AIS_OK) {
 744                 error = req_exec_lck_resourceopen->fail_with_error;
 745                 goto error_exit;
 746         }
 747
 748         resource = resource_find (&req_exec_lck_resourceopen->resource_name);
 749
 750         /*
 751          * If resource doesn't exist, create one
 752          */
 753         if (resource == 0) {
 754                 if ((req_exec_lck_resourceopen->open_flags & SA_LCK_RESOURCE_CREATE) == 0) {
 755                         error = SA_AIS_ERR_NOT_EXIST;
 756                         goto error_exit;
 757                 }
 758                 resource = malloc (sizeof (struct resource));
 759                 if (resource == 0) {
 760                         error = SA_AIS_ERR_NO_MEMORY;
 761                         goto error_exit;
 762                 }
 763                 memset (resource, 0, sizeof (struct resource));
 764
 765                 memcpy (&resource->name,
 766                         &req_exec_lck_resourceopen->resource_name,
 767                         sizeof (mar_name_t));
 768                 list_init (&resource->list);
 769                 list_init (&resource->resource_lock_list_head);
 770                 list_add (&resource->list, &resource_list_head);
 771                 list_init (&resource->pr_granted_list_head);
 772                 list_init (&resource->pr_pending_list_head);
 773                 list_init (&resource->ex_pending_list_head);
 774                 resource->refcount = 0;
 775                 resource->ex_granted = NULL;
 776         }
 777
 778         /*
 779          * Setup connection information and mark resource as referenced
 780          */
 781         if (message_source_is_local (&req_exec_lck_resourceopen->source)) {
 782                 log_printf (LOG_LEVEL_DEBUG, "Lock resource opened is %p\n", resource);
 783                 resource_cleanup = malloc (sizeof (struct resource_cleanup));
 784                 if (resource_cleanup == 0) {
 785                         free (resource);
 786                         error = SA_AIS_ERR_NO_MEMORY;
 787                 } else {
 788                         lck_pd = (struct lck_pd *)openais_conn_private_data_get (req_exec_lck_resourceopen->source.conn);
 789                         list_init (&resource_cleanup->list);
 790                         list_init (&resource_cleanup->resource_lock_list_head);
 791                         resource_cleanup->resource = resource;
 792                         log_printf (LOG_LEVEL_DEBUG, "resource is %p\n", resource);
 793                         resource_cleanup->resource_handle = req_exec_lck_resourceopen->resource_handle;
 794                         list_add (
 795                                 &resource_cleanup->list,
 796                                 &lck_pd->resource_cleanup_list);
 797                 }
 798                 resource->refcount += 1;
 799         }
 800
 801
 802         /*
 803          * Send error result to LCK library
 804          */
 805 error_exit:
 806         /*
 807          * If this node was the source of the message, respond to this node
 808          */
 809         if (message_source_is_local (&req_exec_lck_resourceopen->source)) {
 810                 /*
 811                  * If its an async call respond with the invocation and handle
 812                  */
 813                 if (req_exec_lck_resourceopen->async_call) {
 814                         res_lib_lck_resourceopenasync.header.size = sizeof (struct res_lib_lck_resourceopenasync);
 815                         res_lib_lck_resourceopenasync.header.id = MESSAGE_RES_LCK_RESOURCEOPENASYNC;
 816                         res_lib_lck_resourceopenasync.header.error = error;
 817                         res_lib_lck_resourceopenasync.resourceHandle = req_exec_lck_resourceopen->resource_handle;
 818                         res_lib_lck_resourceopenasync.invocation = req_exec_lck_resourceopen->invocation;
 819                         memcpy (&res_lib_lck_resourceopenasync.source,
 820                                 &req_exec_lck_resourceopen->source,
 821                                 sizeof (mar_message_source_t));
 822
 823                         openais_conn_send_response (
 824                                 req_exec_lck_resourceopen->source.conn,
 825                                 &res_lib_lck_resourceopenasync,
 826                                 sizeof (struct res_lib_lck_resourceopenasync));
 827                         openais_conn_send_response (
 828                                 openais_conn_partner_get (req_exec_lck_resourceopen->source.conn),
 829                                 &res_lib_lck_resourceopenasync,
 830                                 sizeof (struct res_lib_lck_resourceopenasync));
 831                 } else {
 832                         /*
 833                          * otherwise respond with the normal resourceopen response
 834                          */
 835                         res_lib_lck_resourceopen.header.size = sizeof (struct res_lib_lck_resourceopen);
 836                         res_lib_lck_resourceopen.header.id = MESSAGE_RES_LCK_RESOURCEOPEN;
 837                         res_lib_lck_resourceopen.header.error = error;
 838                         memcpy (&res_lib_lck_resourceopen.source,
 839                                 &req_exec_lck_resourceopen->source,
 840                                 sizeof (mar_message_source_t));
 841
 842                         openais_conn_send_response (req_exec_lck_resourceopen->source.conn,
 843                                 &res_lib_lck_resourceopen,
 844                                 sizeof (struct res_lib_lck_resourceopen));
 845                 }
 846         }
 847 }
 848
 849 static void message_handler_req_exec_lck_resourceclose (
 850         void *message,
 851         unsigned int nodeid)
 852 {
 853         struct req_exec_lck_resourceclose *req_exec_lck_resourceclose = (struct req_exec_lck_resourceclose *)message;
 854         struct res_lib_lck_resourceclose res_lib_lck_resourceclose;
 855         struct resource *resource = 0;
 856         SaAisErrorT error = SA_AIS_OK;
 857
 858         log_printf (LOG_LEVEL_NOTICE, "EXEC request: saLckResourceClose %s\n",
 859                 get_mar_name_t (&req_exec_lck_resourceclose->lockResourceName));
 860
 861         resource = resource_find (&req_exec_lck_resourceclose->lockResourceName);
 862         if (resource == 0) {
 863                 goto error_exit;
 864         }
 865
 866         resource->refcount -= 1;
 867         if (resource->refcount == 0) {
 868         }
 869 error_exit:
 870         if (message_source_is_local(&req_exec_lck_resourceclose->source)) {
 871                 lck_resource_cleanup_remove (
 872                         req_exec_lck_resourceclose->source.conn,
 873                         req_exec_lck_resourceclose->resource_handle);
 874
 875                 res_lib_lck_resourceclose.header.size = sizeof (struct res_lib_lck_resourceclose);
 876                 res_lib_lck_resourceclose.header.id = MESSAGE_RES_LCK_RESOURCECLOSE;
 877                 res_lib_lck_resourceclose.header.error = error;
 878                 openais_conn_send_response (
 879                         req_exec_lck_resourceclose->source.conn,
 880                         &res_lib_lck_resourceclose, sizeof (struct res_lib_lck_resourceclose));
 881         }
 882 }
 883
 884 void waiter_notification_send (struct resource_lock *resource_lock)
 885 {
 886         struct res_lib_lck_lockwaitercallback res_lib_lck_lockwaitercallback;
 887
 888         if (message_source_is_local (&resource_lock->callback_source) == 0) {
 889                 return;
 890         }
 891
 892         res_lib_lck_lockwaitercallback.header.size = sizeof (struct res_lib_lck_lockwaitercallback);
 893         res_lib_lck_lockwaitercallback.header.id = MESSAGE_RES_LCK_LOCKWAITERCALLBACK;
 894         res_lib_lck_lockwaitercallback.header.error = SA_AIS_OK;
 895         res_lib_lck_lockwaitercallback.waiter_signal = resource_lock->waiter_signal;
 896         res_lib_lck_lockwaitercallback.lock_id = resource_lock->lock_id;
 897         res_lib_lck_lockwaitercallback.mode_requested = resource_lock->lock_mode;
 898
 899         if (resource_lock->resource->ex_granted) {
 900                 res_lib_lck_lockwaitercallback.mode_held = SA_LCK_EX_LOCK_MODE;
 901         } else {
 902                 res_lib_lck_lockwaitercallback.mode_held = SA_LCK_PR_LOCK_MODE;
 903         }
 904
 905         openais_conn_send_response (
 906                 openais_conn_partner_get (resource_lock->callback_source.conn),
 907                 &res_lib_lck_lockwaitercallback,
 908                 sizeof (struct res_lib_lck_lockwaitercallback));
 909 }
 910
 911 void waiter_notification_list_send (struct list_head *list_notify_head)
 912 {
 913         struct list_head *list;
 914         struct resource_lock *resource_lock;
 915
 916         for (list = list_notify_head->next;
 917                 list != list_notify_head;
 918                 list = list->next) {
 919
 920                 resource_lock = list_entry (list, struct resource_lock, list);
 921                 waiter_notification_send (resource_lock);
 922         }
 923 }
 924
 925 void resource_lock_async_deliver (
 926         mar_message_source_t *source,
 927         struct resource_lock *resource_lock,
 928         SaAisErrorT error)
 929 {
 930         struct res_lib_lck_resourcelockasync res_lib_lck_resourcelockasync;
 931
 932         if (source && message_source_is_local(source)) {
 933                 if (resource_lock->async_call) {
 934                         res_lib_lck_resourcelockasync.header.size = sizeof (struct res_lib_lck_resourcelockasync);
 935                         res_lib_lck_resourcelockasync.header.id = MESSAGE_RES_LCK_RESOURCELOCKASYNC;
 936                         res_lib_lck_resourcelockasync.header.error = error;
 937                         res_lib_lck_resourcelockasync.resource_lock = (void *)resource_lock;
 938                         res_lib_lck_resourcelockasync.lockStatus = resource_lock->lock_status;
 939                         res_lib_lck_resourcelockasync.invocation = resource_lock->invocation;
 940                         res_lib_lck_resourcelockasync.lockId = resource_lock->lock_id;
 941                         openais_conn_send_response (
 942                                 openais_conn_partner_get (source->conn),
 943                                 &res_lib_lck_resourcelockasync,
 944                                 sizeof (struct res_lib_lck_resourcelockasync));
 945                 }
 946         }
 947 }
 948
 949 void lock_response_deliver (
 950         mar_message_source_t *source,
 951         struct resource_lock *resource_lock,
 952         SaAisErrorT error)
 953 {
 954         struct res_lib_lck_resourcelock res_lib_lck_resourcelock;
 955
 956         if (source && message_source_is_local(source)) {
 957                 if (resource_lock->async_call) {
 958                         resource_lock_async_deliver (&resource_lock->callback_source, resource_lock, error);
 959                 } else {
 960                         res_lib_lck_resourcelock.header.size = sizeof (struct res_lib_lck_resourcelock);
 961                         res_lib_lck_resourcelock.header.id = MESSAGE_RES_LCK_RESOURCELOCK;
 962                         res_lib_lck_resourcelock.header.error = error;
 963                         res_lib_lck_resourcelock.resource_lock = (void *)resource_lock;
 964                         res_lib_lck_resourcelock.lockStatus = resource_lock->lock_status;
 965                         openais_conn_send_response (source->conn,
 966                                 &res_lib_lck_resourcelock,
 967                                 sizeof (struct res_lib_lck_resourcelock));
 968                 }
 969         }
 970 }
 971
 972
 973 /*
 974  * Queue a lock if resource flags allow it
 975  */
 976 void lock_queue (
 977         struct resource *resource,
 978         struct resource_lock *resource_lock)
 979 {
 980         if ((resource_lock->lock_flags & SA_LCK_LOCK_NO_QUEUE) == 0) {
 981                 /*
 982                  * Add lock to the list
 983                  */
 984                 if (resource_lock->lock_mode == SA_LCK_PR_LOCK_MODE) {
 985                         list_add_tail (&resource_lock->list,
 986                                 &resource->pr_pending_list_head);
 987                                 waiter_notification_send (resource->ex_granted);
 988                 } else
 989                 if (resource_lock->lock_mode == SA_LCK_EX_LOCK_MODE) {
 990                         list_add_tail (&resource_lock->list,
 991                                 &resource->ex_pending_list_head);
 992                                 waiter_notification_list_send (&resource->pr_granted_list_head);
 993                 }
 994         } else {
 995                 resource_lock->lock_status = SA_LCK_LOCK_NOT_QUEUED;
 996         }
 997 }
 998
 999 /*
1000 The algorithm:
1001
1002 if ex lock granted
1003         if ex pending list has locks
1004                 send waiter notification to ex lock granted
1005 else
1006         if ex pending list has locks
1007                 if pr granted list has locks
1008                         send waiter notification to all pr granted locks
1009                 else
1010                         grant ex lock from pending to granted
1011         else
1012                 grant all pr pending locks to pr granted list
1013 */
1014 #define SA_LCK_LOCK_NO_STATUS 0
1015 void lock_algorithm (
1016         struct resource *resource,
1017         struct resource_lock *resource_lock)
1018 {
1019         resource_lock->lock_status = SA_LCK_LOCK_NO_STATUS; /* no status */
1020         if (resource->ex_granted) {
1021                 /*
1022                  * Exclusive lock granted
1023                  */
1024                 lock_queue (resource, resource_lock);
1025         } else {
1026                 /*
1027                  * Exclusive lock not granted
1028                  */
1029                 if (resource_lock->lock_mode == SA_LCK_EX_LOCK_MODE) {
1030                         if (list_empty (&resource->pr_granted_list_head) == 0) {
1031                                 lock_queue (resource, resource_lock);
1032                         } else {
1033                                 /*
1034                                  * grant ex lock from pending to granted
1035                                  */
1036                                 resource->ex_granted = resource_lock;
1037                                 resource_lock->lock_status = SA_LCK_LOCK_GRANTED;
1038                         }
1039                 } else {
1040                         /*
1041                          * grant all pr pending locks to pr granted list
1042                          */
1043                         list_add (&resource_lock->list,
1044                                 &resource->pr_granted_list_head);
1045                         resource_lock->lock_status = SA_LCK_LOCK_GRANTED;
1046                 }
1047         }
1048 }
1049
1050 /*
1051  *      if lock in ex, set ex to null
1052  *      delete resource lock from list
1053  *
1054  *       if ex lock not granted
1055  *              if ex pending list has locks
1056  *                      grant first ex pending list lock to ex lock
1057  *      if ex lock not granted
1058  *              if pr pending list has locks
1059  *                      assign all pr pending locks to pr granted lock list
1060  */
1061 void unlock_algorithm (
1062         struct resource *resource,
1063         struct resource_lock *resource_lock)
1064 {
1065         struct resource_lock *resource_lock_grant;
1066         struct list_head *list;
1067         struct list_head *list_p;
1068
1069         /*
1070          * If unlocking the ex lock, reset ex granted
1071          */
1072         if (resource_lock == resource->ex_granted) {
1073                 resource->ex_granted = 0;
1074         }
1075         else {
1076                 /*
1077                  * Delete resource lock from whichever list it is on
1078                  */
1079                 list_del (&resource_lock->list);
1080         }
1081         list_del (&resource_lock->resource_list);
1082
1083         /*
1084          * Check if EX locks are available, if so assign one
1085          */
1086         if (resource->ex_granted == 0) {
1087                 if (list_empty (&resource->ex_pending_list_head) == 0) {
1088                         /*
1089                          * grant first ex pending list lock to ex lock
1090                          */
1091                         resource_lock_grant = list_entry (
1092                                 resource->ex_pending_list_head.next,
1093                                 struct resource_lock, list);
1094
1095                         list_del (&resource_lock_grant->list);
1096                         resource->ex_granted = resource_lock_grant;
1097
1098                         resource_lock_grant->lock_status = SA_LCK_LOCK_GRANTED;
1099                         lock_response_deliver (
1100                                 &resource_lock_grant->response_source,
1101                                 resource_lock_grant,
1102                                 SA_AIS_OK);
1103                 }
1104         }
1105
1106         /*
1107          * Couldn't assign EX lock, so assign any pending PR locks
1108          */
1109         if (resource->ex_granted == 0) {
1110                 if (list_empty (&resource->pr_pending_list_head) == 0) {
1111                         /*
1112                          * assign all pr pending locks to pr granted lock list
1113                          */
1114
1115                    for (list = resource->pr_pending_list_head.next;
1116                                 list != &resource->pr_pending_list_head;
1117                                 list = list->next) {
1118
1119                                 resource_lock_grant = list_entry (list, struct resource_lock, list);
1120                                 resource_lock_grant->lock_status = SA_LCK_LOCK_GRANTED;
1121
1122                                 lock_response_deliver (
1123                                         &resource_lock_grant->response_source,
1124                                         resource_lock_grant,
1125                                         SA_AIS_OK);
1126                         }
1127
1128                         /*
1129                          * Add pending locks to granted list
1130                          */
1131                         list_p = resource->pr_pending_list_head.next;
1132                         list_del (&resource->pr_pending_list_head);
1133                         list_add_tail (list_p,
1134                                 &resource->pr_granted_list_head);
1135                 }
1136         }
1137 }
1138
1139 static void message_handler_req_exec_lck_resourcelock (
1140         void *message,
1141         unsigned int nodeid)
1142 {
1143         struct req_exec_lck_resourcelock *req_exec_lck_resourcelock = (struct req_exec_lck_resourcelock *)message;
1144         struct resource *resource = 0;
1145         struct resource_lock *resource_lock = 0;
1146         struct resource_cleanup *resource_cleanup = 0;
1147
1148         log_printf (LOG_LEVEL_NOTICE, "EXEC request: saLckResourceLock %s\n",
1149                 get_mar_name_t (&req_exec_lck_resourcelock->req_lib_lck_resourcelock.lockResourceName));
1150
1151         resource = resource_find (&req_exec_lck_resourcelock->req_lib_lck_resourcelock.lockResourceName);
1152         if (resource == 0) {
1153                 goto error_exit;
1154         }
1155         resource->refcount += 1;
1156
1157         resource_lock = malloc (sizeof (struct resource_lock));
1158         if (resource_lock == 0) {
1159                 lock_response_deliver (&req_exec_lck_resourcelock->source,
1160                         resource_lock,
1161                         SA_AIS_ERR_NO_MEMORY);
1162                 goto error_exit;
1163         }
1164
1165         /*
1166          * Build resource lock structure
1167          */
1168         memset (resource_lock, 0, sizeof (struct resource_lock));
1169
1170         list_init (&resource_lock->list);
1171         list_init (&resource_lock->resource_list);
1172         list_init (&resource_lock->resource_cleanup_list);
1173
1174         list_add (&resource_lock->resource_list, &resource->resource_lock_list_head);
1175
1176         resource_lock->resource = resource;
1177
1178         resource_lock->lock_mode =
1179                 req_exec_lck_resourcelock->req_lib_lck_resourcelock.lockMode;
1180         resource_lock->lock_flags =
1181                 req_exec_lck_resourcelock->req_lib_lck_resourcelock.lockFlags;
1182         resource_lock->waiter_signal =
1183                 req_exec_lck_resourcelock->req_lib_lck_resourcelock.waiterSignal;
1184         resource_lock->timeout =
1185                 req_exec_lck_resourcelock->req_lib_lck_resourcelock.timeout;
1186         resource_lock->lock_id =
1187                 req_exec_lck_resourcelock->req_lib_lck_resourcelock.lockId;
1188         resource_lock->async_call =
1189                 req_exec_lck_resourcelock->req_lib_lck_resourcelock.async_call;
1190         resource_lock->invocation =
1191                 req_exec_lck_resourcelock->req_lib_lck_resourcelock.invocation;
1192
1193         /*
1194          * Waiter callback source
1195          */
1196         memcpy (&resource_lock->callback_source,
1197                 &req_exec_lck_resourcelock->req_lib_lck_resourcelock.source,
1198                 sizeof (mar_message_source_t));
1199
1200         lock_algorithm (resource, resource_lock);
1201
1202         /*
1203          * Add resource lock to cleanup handler for this api resource instance
1204          */
1205         if (message_source_is_local (&req_exec_lck_resourcelock->source)) {
1206                 resource_cleanup = lck_resource_cleanup_find (
1207                         resource_lock->callback_source.conn,
1208                         req_exec_lck_resourcelock->resource_handle);
1209
1210                 assert (resource_cleanup);
1211
1212                 list_add (&resource_lock->resource_cleanup_list,
1213                         &resource_cleanup->resource_lock_list_head);
1214
1215                 /*
1216                  * If lock queued by lock algorithm, dont send response to library now
1217                  */
1218                 if (resource_lock->lock_status != SA_LCK_LOCK_NO_STATUS) {
1219                         /*
1220                          * If lock granted or denied, deliver callback or
1221                          * response to library for non-async calls
1222                          */
1223                         lock_response_deliver (
1224                                 &req_exec_lck_resourcelock->source,
1225                                 resource_lock,
1226                                 SA_AIS_OK);
1227                 } else {
1228                         memcpy (&resource_lock->response_source,
1229                                 &req_exec_lck_resourcelock->source,
1230                                 sizeof (mar_message_source_t));
1231                 }
1232
1233                 /*
1234                  * Deliver async response to library
1235                  */
1236                 req_exec_lck_resourcelock->source.conn =
1237                         openais_conn_partner_get (req_exec_lck_resourcelock->source.conn);
1238                 resource_lock_async_deliver (
1239                         &req_exec_lck_resourcelock->source,
1240                         resource_lock,
1241                         SA_AIS_OK);
1242 // TODO why is this twice ?
1243                 req_exec_lck_resourcelock->source.conn =
1244                         openais_conn_partner_get (req_exec_lck_resourcelock->source.conn);
1245         }
1246
1247 error_exit:
1248         return;
1249 }
1250
1251 static void message_handler_req_exec_lck_resourceunlock (
1252         void *message,
1253         unsigned int nodeid)
1254 {
1255         struct req_exec_lck_resourceunlock *req_exec_lck_resourceunlock = (struct req_exec_lck_resourceunlock *)message;
1256         struct res_lib_lck_resourceunlock res_lib_lck_resourceunlock;
1257         struct res_lib_lck_resourceunlockasync res_lib_lck_resourceunlockasync;
1258         struct resource *resource = NULL;
1259         struct resource_lock *resource_lock = NULL;
1260         SaAisErrorT error = SA_AIS_OK;
1261
1262         log_printf (LOG_LEVEL_NOTICE, "EXEC request: saLckResourceUnlock %s\n",
1263                 get_mar_name_t (&req_exec_lck_resourceunlock->resource_name));
1264
1265         resource = resource_find (&req_exec_lck_resourceunlock->resource_name);
1266         if (resource == 0) {
1267                 goto error_exit;
1268         }
1269         resource->refcount -= 1;
1270
1271         resource_lock = resource_lock_find (resource,
1272                 &req_exec_lck_resourceunlock->source,
1273                 req_exec_lck_resourceunlock->lock_id);
1274         assert (resource_lock);
1275
1276         list_del (&resource_lock->resource_cleanup_list);
1277         unlock_algorithm (resource, resource_lock);
1278
1279 error_exit:
1280         if (message_source_is_local(&req_exec_lck_resourceunlock->source)) {
1281                 if (req_exec_lck_resourceunlock->async_call) {
1282                         res_lib_lck_resourceunlockasync.header.size = sizeof (struct res_lib_lck_resourceunlockasync);
1283                         res_lib_lck_resourceunlockasync.header.id = MESSAGE_RES_LCK_RESOURCEUNLOCKASYNC;
1284                         res_lib_lck_resourceunlockasync.header.error = error;
1285                         res_lib_lck_resourceunlockasync.invocation =
1286                                 req_exec_lck_resourceunlock->invocation;
1287                         res_lib_lck_resourceunlockasync.lockId = req_exec_lck_resourceunlock->lock_id;
1288
1289                         openais_conn_send_response (
1290                                 openais_conn_partner_get(req_exec_lck_resourceunlock->source.conn),
1291                                 &res_lib_lck_resourceunlockasync,
1292                                 sizeof (struct res_lib_lck_resourceunlockasync));
1293                         openais_conn_send_response (
1294                                 openais_conn_partner_get(resource_lock->callback_source.conn),
1295                                 &res_lib_lck_resourceunlockasync,
1296                                 sizeof (struct res_lib_lck_resourceunlockasync));
1297                 } else {
1298                         res_lib_lck_resourceunlock.header.size = sizeof (struct res_lib_lck_resourceunlock);
1299                         res_lib_lck_resourceunlock.header.id = MESSAGE_RES_LCK_RESOURCEUNLOCK;
1300                         res_lib_lck_resourceunlock.header.error = error;
1301                         openais_conn_send_response (req_exec_lck_resourceunlock->source.conn,
1302                                 &res_lib_lck_resourceunlock, sizeof (struct res_lib_lck_resourceunlock));
1303                 }
1304         }
1305 }
1306
1307 static void message_handler_req_exec_lck_resourcelockorphan (
1308         void *message,
1309         unsigned int nodeid)
1310 {
1311         struct req_exec_lck_resourcelockorphan *req_exec_lck_resourcelockorphan = (struct req_exec_lck_resourcelockorphan *)message;
1312         struct resource *resource = 0;
1313         struct resource_lock *resource_lock = 0;
1314
1315         log_printf (LOG_LEVEL_NOTICE, "EXEC request: Orphan resource locks for resource %s\n",
1316                 get_mar_name_t (&req_exec_lck_resourcelockorphan->resource_name));
1317
1318         resource = resource_find (&req_exec_lck_resourcelockorphan->resource_name);
1319         if (resource == 0) {
1320                 assert (0);
1321         }
1322         resource->refcount -= 1;
1323
1324         resource_lock = resource_lock_find (resource,
1325                 &req_exec_lck_resourcelockorphan->source,
1326                 req_exec_lck_resourcelockorphan->lock_id);
1327         assert (resource_lock);
1328
1329         list_del (&resource_lock->resource_cleanup_list);
1330         unlock_algorithm (resource, resource_lock);
1331 }
1332
1333 static void message_handler_req_exec_lck_lockpurge (
1334         void *msg,
1335         unsigned int nodeid)
1336 {
1337         struct req_exec_lck_lockpurge *req_exec_lck_lockpurge = (struct req_exec_lck_lockpurge *)msg;
1338         struct res_lib_lck_lockpurge res_lib_lck_lockpurge;
1339         struct resource *resource = 0;
1340         SaAisErrorT error = SA_AIS_OK;
1341
1342         log_printf (LOG_LEVEL_DEBUG, "EXEC request: saLckLockPurge %s\n",
1343                 get_mar_name_t (&req_exec_lck_lockpurge->req_lib_lck_lockpurge.lockResourceName));
1344
1345         resource = resource_find (&req_exec_lck_lockpurge->req_lib_lck_lockpurge.lockResourceName);
1346         if (resource == 0) {
1347                 goto error_exit;
1348         }
1349
1350 error_exit:
1351         if (message_source_is_local(&req_exec_lck_lockpurge->source)) {
1352 //              lck_resource_cleanup_remove (req_exec_lck_lockpurge->source.conn,
1353 //                      resource);
1354
1355                 res_lib_lck_lockpurge.header.size = sizeof (struct res_lib_lck_lockpurge);
1356                 res_lib_lck_lockpurge.header.id = MESSAGE_RES_LCK_LOCKPURGE;
1357                 res_lib_lck_lockpurge.header.error = error;
1358                 openais_conn_send_response (req_exec_lck_lockpurge->source.conn,
1359                         &res_lib_lck_lockpurge, sizeof (struct res_lib_lck_lockpurge));
1360         }
1361 }
1362
1363 static void message_handler_req_lib_lck_resourceopen (
1364         void *conn,
1365         void *msg)
1366 {
1367         struct req_lib_lck_resourceopen *req_lib_lck_resourceopen = (struct req_lib_lck_resourceopen *)msg;
1368         struct req_exec_lck_resourceopen req_exec_lck_resourceopen;
1369         struct iovec iovec;
1370
1371         log_printf (LOG_LEVEL_NOTICE, "LIB request: saLckResourceOpen %s\n",
1372                 get_mar_name_t (&req_lib_lck_resourceopen->lockResourceName));
1373
1374         req_exec_lck_resourceopen.header.size =
1375                 sizeof (struct req_exec_lck_resourceopen);
1376         req_exec_lck_resourceopen.header.id =
1377                 SERVICE_ID_MAKE (LCK_SERVICE, MESSAGE_REQ_EXEC_LCK_RESOURCEOPEN);
1378
1379         message_source_set (&req_exec_lck_resourceopen.source, conn);
1380
1381         memcpy (&req_exec_lck_resourceopen.resource_name,
1382                 &req_lib_lck_resourceopen->lockResourceName,
1383                 sizeof (SaNameT));
1384
1385         req_exec_lck_resourceopen.open_flags = req_lib_lck_resourceopen->resourceOpenFlags;
1386         req_exec_lck_resourceopen.async_call = 0;
1387         req_exec_lck_resourceopen.invocation = 0;
1388         req_exec_lck_resourceopen.resource_handle = req_lib_lck_resourceopen->resourceHandle;
1389         req_exec_lck_resourceopen.fail_with_error = SA_AIS_OK;
1390
1391         iovec.iov_base = (char *)&req_exec_lck_resourceopen;
1392         iovec.iov_len = sizeof (req_exec_lck_resourceopen);
1393
1394         assert (totempg_groups_mcast_joined (openais_group_handle, &iovec, 1, TOTEMPG_AGREED) == 0);
1395 }
1396
1397 static void message_handler_req_lib_lck_resourceopenasync (
1398         void *conn,
1399         void *msg)
1400 {
1401         struct req_lib_lck_resourceopen *req_lib_lck_resourceopen = (struct req_lib_lck_resourceopen *)msg;
1402         struct req_exec_lck_resourceopen req_exec_lck_resourceopen;
1403         struct iovec iovec;
1404
1405         log_printf (LOG_LEVEL_NOTICE, "LIB request: saLckResourceOpenAsync %s\n",
1406                 get_mar_name_t (&req_lib_lck_resourceopen->lockResourceName));
1407
1408         req_exec_lck_resourceopen.header.size =
1409                 sizeof (struct req_exec_lck_resourceopen);
1410         req_exec_lck_resourceopen.header.id =
1411                 SERVICE_ID_MAKE (LCK_SERVICE, MESSAGE_REQ_EXEC_LCK_RESOURCEOPEN);
1412
1413         message_source_set (&req_exec_lck_resourceopen.source, conn);
1414
1415         memcpy (&req_exec_lck_resourceopen.resource_name,
1416                 &req_lib_lck_resourceopen->lockResourceName,
1417                 sizeof (mar_name_t));
1418
1419         req_exec_lck_resourceopen.resource_handle = req_lib_lck_resourceopen->resourceHandle;
1420         req_exec_lck_resourceopen.invocation = req_lib_lck_resourceopen->invocation;
1421         req_exec_lck_resourceopen.open_flags = req_lib_lck_resourceopen->resourceOpenFlags;
1422         req_exec_lck_resourceopen.timeout = 0;
1423         req_exec_lck_resourceopen.async_call = 1;
1424
1425         iovec.iov_base = (char *)&req_exec_lck_resourceopen;
1426         iovec.iov_len = sizeof (req_exec_lck_resourceopen);
1427
1428         assert (totempg_groups_mcast_joined (openais_group_handle, &iovec, 1, TOTEMPG_AGREED) == 0);
1429 }
1430
1431 static void message_handler_req_lib_lck_resourceclose (
1432         void *conn,
1433         void *msg)
1434 {
1435         struct req_lib_lck_resourceclose *req_lib_lck_resourceclose = (struct req_lib_lck_resourceclose *)msg;
1436         struct req_exec_lck_resourceclose req_exec_lck_resourceclose;
1437         struct iovec iovecs[2];
1438         struct resource *resource;
1439         struct res_lib_lck_resourceclose res_lib_lck_resourceclose;
1440
1441         log_printf (LOG_LEVEL_NOTICE, "LIB request: saLckResourceClose %s\n",
1442                 get_mar_name_t (&req_lib_lck_resourceclose->lockResourceName));
1443
1444         resource = resource_find (&req_lib_lck_resourceclose->lockResourceName);
1445         if (resource) {
1446                 req_exec_lck_resourceclose.header.size =
1447                         sizeof (struct req_exec_lck_resourceclose);
1448                 req_exec_lck_resourceclose.header.id =
1449                         SERVICE_ID_MAKE (LCK_SERVICE, MESSAGE_REQ_EXEC_LCK_RESOURCECLOSE);
1450
1451                 message_source_set (&req_exec_lck_resourceclose.source, conn);
1452
1453                 memcpy (&req_exec_lck_resourceclose.lockResourceName,
1454                         &req_lib_lck_resourceclose->lockResourceName, sizeof (mar_name_t));
1455
1456                 req_exec_lck_resourceclose.resource_handle = req_lib_lck_resourceclose->resourceHandle;
1457                 iovecs[0].iov_base = (char *)&req_exec_lck_resourceclose;
1458                 iovecs[0].iov_len = sizeof (req_exec_lck_resourceclose);
1459
1460                 if (totempg_groups_send_ok_joined (openais_group_handle, iovecs, 1)) {
1461                         assert (totempg_groups_mcast_joined (openais_group_handle, iovecs, 1, TOTEMPG_AGREED) == 0);
1462                 }
1463         }
1464         else {
1465                 log_printf (LOG_LEVEL_ERROR, "#### LCK: Could Not Find the Checkpoint to close so Returning Error. ####\n");
1466
1467                 res_lib_lck_resourceclose.header.size = sizeof (struct res_lib_lck_resourceclose);
1468                 res_lib_lck_resourceclose.header.id = MESSAGE_RES_LCK_RESOURCECLOSE;
1469                 res_lib_lck_resourceclose.header.error = SA_AIS_ERR_NOT_EXIST;
1470
1471                 openais_conn_send_response (conn,
1472                         &res_lib_lck_resourceclose,
1473                         sizeof (struct res_lib_lck_resourceclose));
1474         }
1475 }
1476
1477 static void message_handler_req_lib_lck_resourcelock (
1478         void *conn,
1479         void *msg)
1480 {
1481         struct req_lib_lck_resourcelock *req_lib_lck_resourcelock = (struct req_lib_lck_resourcelock *)msg;
1482         struct req_exec_lck_resourcelock req_exec_lck_resourcelock;
1483         struct iovec iovecs[2];
1484
1485         log_printf (LOG_LEVEL_NOTICE, "LIB request: saLckResourceLock %s\n",
1486                 get_mar_name_t (&req_lib_lck_resourcelock->lockResourceName));
1487
1488         req_exec_lck_resourcelock.header.size =
1489                 sizeof (struct req_exec_lck_resourcelock);
1490         req_exec_lck_resourcelock.header.id =
1491                 SERVICE_ID_MAKE (LCK_SERVICE, MESSAGE_REQ_EXEC_LCK_RESOURCELOCK);
1492
1493         message_source_set (&req_exec_lck_resourcelock.source, conn);
1494
1495         memcpy (&req_exec_lck_resourcelock.req_lib_lck_resourcelock,
1496                 req_lib_lck_resourcelock,
1497                 sizeof (struct req_lib_lck_resourcelock));
1498
1499         req_exec_lck_resourcelock.resource_handle = req_lib_lck_resourcelock->resourceHandle;
1500         req_exec_lck_resourcelock.async_call = 0;
1501         req_exec_lck_resourcelock.invocation = 0;
1502         req_exec_lck_resourcelock.fail_with_error = SA_AIS_OK;
1503
1504         iovecs[0].iov_base = (char *)&req_exec_lck_resourcelock;
1505         iovecs[0].iov_len = sizeof (req_exec_lck_resourcelock);
1506
1507         assert (totempg_groups_mcast_joined (openais_group_handle, iovecs, 1, TOTEMPG_AGREED) == 0);
1508 }
1509
1510 static void message_handler_req_lib_lck_resourcelockasync (
1511         void *conn,
1512         void *msg)
1513 {
1514         struct req_lib_lck_resourcelock *req_lib_lck_resourcelock = (struct req_lib_lck_resourcelock *)msg;
1515         struct req_exec_lck_resourcelock req_exec_lck_resourcelock;
1516         struct iovec iovecs[2];
1517
1518         log_printf (LOG_LEVEL_NOTICE, "LIB request: saLckResourceLockAsync %s\n",
1519                 get_mar_name_t (&req_lib_lck_resourcelock->lockResourceName));
1520
1521         req_exec_lck_resourcelock.header.size =
1522                 sizeof (struct req_exec_lck_resourcelock);
1523         req_exec_lck_resourcelock.header.id =
1524                 SERVICE_ID_MAKE (LCK_SERVICE, MESSAGE_REQ_EXEC_LCK_RESOURCELOCK);
1525
1526         message_source_set (&req_exec_lck_resourcelock.source, conn);
1527
1528         memcpy (&req_exec_lck_resourcelock.req_lib_lck_resourcelock,
1529                 req_lib_lck_resourcelock,
1530                 sizeof (struct req_lib_lck_resourcelock));
1531
1532         req_exec_lck_resourcelock.resource_handle = req_lib_lck_resourcelock->resourceHandle;
1533         req_exec_lck_resourcelock.async_call = 1;
1534         req_exec_lck_resourcelock.invocation = req_lib_lck_resourcelock->invocation;
1535
1536         iovecs[0].iov_base = (char *)&req_exec_lck_resourcelock;
1537         iovecs[0].iov_len = sizeof (req_exec_lck_resourcelock);
1538
1539         assert (totempg_groups_mcast_joined (openais_group_handle, iovecs, 1, TOTEMPG_AGREED) == 0);
1540 }
1541
1542 static void message_handler_req_lib_lck_resourceunlock (
1543         void *conn,
1544         void *msg)
1545 {
1546         struct req_lib_lck_resourceunlock *req_lib_lck_resourceunlock = (struct req_lib_lck_resourceunlock *)msg;
1547         struct req_exec_lck_resourceunlock req_exec_lck_resourceunlock;
1548         struct iovec iovec;
1549
1550         log_printf (LOG_LEVEL_NOTICE, "LIB request: saLckResourceUnlock %s\n",
1551                 get_mar_name_t (&req_lib_lck_resourceunlock->lockResourceName));
1552
1553         req_exec_lck_resourceunlock.header.size =
1554                 sizeof (struct req_exec_lck_resourceunlock);
1555         req_exec_lck_resourceunlock.header.id =
1556                 SERVICE_ID_MAKE (LCK_SERVICE, MESSAGE_REQ_EXEC_LCK_RESOURCEUNLOCK);
1557
1558         message_source_set (&req_exec_lck_resourceunlock.source, conn);
1559
1560         memcpy (&req_exec_lck_resourceunlock.resource_name,
1561                 &req_lib_lck_resourceunlock->lockResourceName,
1562                 sizeof (mar_name_t));
1563
1564         req_exec_lck_resourceunlock.lock_id = req_lib_lck_resourceunlock->lockId;
1565         req_exec_lck_resourceunlock.async_call = 0;
1566         req_exec_lck_resourceunlock.invocation = 0;
1567
1568         iovec.iov_base = (char *)&req_exec_lck_resourceunlock;
1569         iovec.iov_len = sizeof (req_exec_lck_resourceunlock);
1570
1571         assert (totempg_groups_mcast_joined (openais_group_handle, &iovec, 1, TOTEMPG_AGREED) == 0);
1572 }
1573
1574 static void message_handler_req_lib_lck_resourceunlockasync (
1575         void *conn,
1576         void *msg)
1577 {
1578         struct req_lib_lck_resourceunlock *req_lib_lck_resourceunlock = (struct req_lib_lck_resourceunlock *)msg;
1579         struct req_exec_lck_resourceunlock req_exec_lck_resourceunlock;
1580         struct iovec iovec;
1581
1582         log_printf (LOG_LEVEL_NOTICE, "LIB request: saLckResourceUnlockAsync %s\n",
1583                 get_mar_name_t (&req_lib_lck_resourceunlock->lockResourceName));
1584
1585         req_exec_lck_resourceunlock.header.size =
1586                 sizeof (struct req_exec_lck_resourceunlock);
1587         req_exec_lck_resourceunlock.header.id =
1588                 SERVICE_ID_MAKE (LCK_SERVICE, MESSAGE_REQ_EXEC_LCK_RESOURCEUNLOCK);
1589
1590         message_source_set (&req_exec_lck_resourceunlock.source, conn);
1591
1592         memcpy (&req_exec_lck_resourceunlock.resource_name,
1593                 &req_lib_lck_resourceunlock->lockResourceName,
1594                 sizeof (mar_name_t));
1595
1596         req_exec_lck_resourceunlock.lock_id = req_lib_lck_resourceunlock->lockId;
1597         req_exec_lck_resourceunlock.invocation = req_lib_lck_resourceunlock->invocation;
1598         req_exec_lck_resourceunlock.async_call = 1;
1599
1600         iovec.iov_base = (char *)&req_exec_lck_resourceunlock;
1601         iovec.iov_len = sizeof (req_exec_lck_resourceunlock);
1602
1603         assert (totempg_groups_mcast_joined (openais_group_handle, &iovec, 1, TOTEMPG_AGREED) == 0);
1604 }
1605
1606 static void message_handler_req_lib_lck_lockpurge (
1607         void *conn,
1608         void *msg)
1609 {
1610         struct req_lib_lck_lockpurge *req_lib_lck_lockpurge = (struct req_lib_lck_lockpurge *)msg;
1611         struct req_exec_lck_lockpurge req_exec_lck_lockpurge;
1612         struct iovec iovecs[2];
1613
1614         log_printf (LOG_LEVEL_NOTICE, "LIB request: saLckResourceLockPurge %s\n",
1615                 get_mar_name_t (&req_lib_lck_lockpurge->lockResourceName));
1616
1617         req_exec_lck_lockpurge.header.size =
1618                 sizeof (struct req_exec_lck_lockpurge);
1619         req_exec_lck_lockpurge.header.id =
1620                 SERVICE_ID_MAKE (LCK_SERVICE, MESSAGE_REQ_EXEC_LCK_LOCKPURGE);
1621
1622         message_source_set (&req_exec_lck_lockpurge.source, conn);
1623
1624         memcpy (&req_exec_lck_lockpurge.req_lib_lck_lockpurge,
1625                 req_lib_lck_lockpurge,
1626                 sizeof (struct req_lib_lck_lockpurge));
1627
1628         iovecs[0].iov_base = (char *)&req_exec_lck_lockpurge;
1629         iovecs[0].iov_len = sizeof (req_exec_lck_lockpurge);
1630
1631         assert (totempg_groups_mcast_joined (openais_group_handle, iovecs, 1, TOTEMPG_AGREED) == 0);
1632 }
1633