source/cluster/ctdb/common/ctdb_lockwait.c

   1 /*
   2    wait for a tdb chain lock
   3
   4    Copyright (C) Andrew Tridgell  2006
   5
   6    This library is free software; you can redistribute it and/or
   7    modify it under the terms of the GNU Lesser General Public
   8    License as published by the Free Software Foundation; either
   9    version 3 of the License, or (at your option) any later version.
  10
  11    This library is distributed in the hope that it will be useful,
  12    but WITHOUT ANY WARRANTY; without even the implied warranty of
  13    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14    Lesser General Public License for more details.
  15
  16    You should have received a copy of the GNU Lesser General Public
  17    License along with this library; if not, see <http://www.gnu.org/licenses/>.
  18 */
  19
  20 #include "includes.h"
  21 #include "lib/events/events.h"
  22 #include "system/filesys.h"
  23 #include "system/wait.h"
  24 #include "db_wrap.h"
  25 #include "lib/tdb/include/tdb.h"
  26 #include "../include/ctdb_private.h"
  27
  28
  29 struct lockwait_handle {
  30         struct ctdb_context *ctdb;
  31         struct fd_event *fde;
  32         int fd[2];
  33         pid_t child;
  34         void *private_data;
  35         void (*callback)(void *);
  36         struct timeval start_time;
  37 };
  38
  39 static void lockwait_handler(struct event_context *ev, struct fd_event *fde,
  40                              uint16_t flags, void *private_data)
  41 {
  42         struct lockwait_handle *h = talloc_get_type(private_data,
  43                                                      struct lockwait_handle);
  44         void (*callback)(void *) = h->callback;
  45         void *p = h->private_data;
  46         pid_t child = h->child;
  47         talloc_set_destructor(h, NULL);
  48         close(h->fd[0]);
  49         ctdb_latency(&h->ctdb->status.max_lockwait_latency, h->start_time);
  50         h->ctdb->status.pending_lockwait_calls--;
  51         talloc_free(h);
  52         callback(p);
  53         waitpid(child, NULL, 0);
  54 }
  55
  56 static int lockwait_destructor(struct lockwait_handle *h)
  57 {
  58         h->ctdb->status.pending_lockwait_calls--;
  59         close(h->fd[0]);
  60         kill(h->child, SIGKILL);
  61         waitpid(h->child, NULL, 0);
  62         return 0;
  63 }
  64
  65 /*
  66   setup a non-blocking chainlock on a tdb record. If this function
  67   returns NULL then it could not get the chainlock. Otherwise it
  68   returns a opaque handle, and will call callback() once it has
  69   managed to get the chainlock. You can cancel it by using talloc_free
  70   on the returned handle.
  71
  72   It is the callers responsibility to unlock the chainlock once
  73   acquired
  74  */
  75 struct lockwait_handle *ctdb_lockwait(struct ctdb_db_context *ctdb_db,
  76                                       TDB_DATA key,
  77                                       void (*callback)(void *private_data),
  78                                       void *private_data)
  79 {
  80         struct lockwait_handle *result;
  81         int ret;
  82
  83         ctdb_db->ctdb->status.lockwait_calls++;
  84         ctdb_db->ctdb->status.pending_lockwait_calls++;
  85
  86         if (!(result = talloc_zero(ctdb_db, struct lockwait_handle))) {
  87                 ctdb_db->ctdb->status.pending_lockwait_calls--;
  88                 return NULL;
  89         }
  90
  91         ret = pipe(result->fd);
  92
  93         if (ret != 0) {
  94                 talloc_free(result);
  95                 ctdb_db->ctdb->status.pending_lockwait_calls--;
  96                 return NULL;
  97         }
  98
  99         result->child = fork();
 100
 101         if (result->child == (pid_t)-1) {
 102                 close(result->fd[0]);
 103                 close(result->fd[1]);
 104                 talloc_free(result);
 105                 ctdb_db->ctdb->status.pending_lockwait_calls--;
 106                 return NULL;
 107         }
 108
 109         result->callback = callback;
 110         result->private_data = private_data;
 111         result->ctdb = ctdb_db->ctdb;
 112
 113         if (result->child == 0) {
 114                 close(result->fd[0]);
 115                 /*
 116                  * Do we need a tdb_reopen here?
 117                  */
 118                 tdb_chainlock(ctdb_db->ltdb->tdb, key);
 119                 _exit(0);
 120         }
 121
 122         close(result->fd[1]);
 123         talloc_set_destructor(result, lockwait_destructor);
 124
 125         result->fde = event_add_fd(ctdb_db->ctdb->ev, result, result->fd[0],
 126                                    EVENT_FD_READ, lockwait_handler,
 127                                    (void *)result);
 128         if (result->fde == NULL) {
 129                 talloc_free(result);
 130                 ctdb_db->ctdb->status.pending_lockwait_calls--;
 131                 return NULL;
 132         }
 133
 134         result->start_time = timeval_current();
 135
 136         return result;
 137 }