ctdb/common/ctdb_ltdb.c

   1 /*
   2    ctdb ltdb code
   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 2 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, write to the Free Software
  18    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  19 */
  20
  21 #include "includes.h"
  22 #include "lib/events/events.h"
  23 #include "lib/tdb/include/tdb.h"
  24 #include "system/network.h"
  25 #include "system/filesys.h"
  26 #include "../include/ctdb_private.h"
  27 #include "db_wrap.h"
  28 #include "lib/util/dlinklist.h"
  29
  30 /*
  31   find an attached ctdb_db handle given a name
  32  */
  33 struct ctdb_db_context *ctdb_db_handle(struct ctdb_context *ctdb, const char *name)
  34 {
  35         struct ctdb_db_context *tmp_db;
  36         for (tmp_db=ctdb->db_list;tmp_db;tmp_db=tmp_db->next) {
  37                 if (strcmp(name, tmp_db->db_name) == 0) {
  38                         return tmp_db;
  39                 }
  40         }
  41         return NULL;
  42 }
  43
  44
  45 /*
  46   this is the dummy null procedure that all databases support
  47 */
  48 static int ctdb_null_func(struct ctdb_call_info *call)
  49 {
  50         return 0;
  51 }
  52
  53 /*
  54   this is a plain fetch procedure that all databases support
  55 */
  56 static int ctdb_fetch_func(struct ctdb_call_info *call)
  57 {
  58         call->reply_data = &call->record_data;
  59         return 0;
  60 }
  61
  62
  63 /*
  64   return the lmaster given a key
  65 */
  66 uint32_t ctdb_lmaster(struct ctdb_context *ctdb, const TDB_DATA *key)
  67 {
  68         uint32_t idx, lmaster;
  69
  70         idx = ctdb_hash(key) % ctdb->vnn_map->size;
  71         lmaster = ctdb->vnn_map->map[idx];
  72
  73         return lmaster;
  74 }
  75
  76
  77 /*
  78   construct an initial header for a record with no ltdb header yet
  79 */
  80 static void ltdb_initial_header(struct ctdb_db_context *ctdb_db,
  81                                 TDB_DATA key,
  82                                 struct ctdb_ltdb_header *header)
  83 {
  84         header->rsn = 0;
  85         /* initial dmaster is the lmaster */
  86         header->dmaster = ctdb_lmaster(ctdb_db->ctdb, &key);
  87         header->laccessor = header->dmaster;
  88         header->lacount = 0;
  89 }
  90
  91
  92 /*
  93   fetch a record from the ltdb, separating out the header information
  94   and returning the body of the record. A valid (initial) header is
  95   returned if the record is not present
  96 */
  97 int ctdb_ltdb_fetch(struct ctdb_db_context *ctdb_db,
  98                     TDB_DATA key, struct ctdb_ltdb_header *header,
  99                     TALLOC_CTX *mem_ctx, TDB_DATA *data)
 100 {
 101         TDB_DATA rec;
 102         struct ctdb_context *ctdb = ctdb_db->ctdb;
 103
 104         rec = tdb_fetch(ctdb_db->ltdb->tdb, key);
 105         if (rec.dsize < sizeof(*header)) {
 106                 TDB_DATA d2;
 107                 /* return an initial header */
 108                 if (rec.dptr) free(rec.dptr);
 109                 ltdb_initial_header(ctdb_db, key, header);
 110                 ZERO_STRUCT(d2);
 111                 if (data) {
 112                         *data = d2;
 113                 }
 114                 ctdb_ltdb_store(ctdb_db, key, header, d2);
 115                 return 0;
 116         }
 117
 118         *header = *(struct ctdb_ltdb_header *)rec.dptr;
 119
 120         if (data) {
 121                 data->dsize = rec.dsize - sizeof(struct ctdb_ltdb_header);
 122                 data->dptr = talloc_memdup(mem_ctx,
 123                                            sizeof(struct ctdb_ltdb_header)+rec.dptr,
 124                                            data->dsize);
 125         }
 126
 127         free(rec.dptr);
 128         if (data) {
 129                 CTDB_NO_MEMORY(ctdb, data->dptr);
 130         }
 131
 132         return 0;
 133 }
 134
 135
 136 /*
 137   fetch a record from the ltdb, separating out the header information
 138   and returning the body of the record. A valid (initial) header is
 139   returned if the record is not present
 140 */
 141 int ctdb_ltdb_store(struct ctdb_db_context *ctdb_db, TDB_DATA key,
 142                     struct ctdb_ltdb_header *header, TDB_DATA data)
 143 {
 144         struct ctdb_context *ctdb = ctdb_db->ctdb;
 145         TDB_DATA rec;
 146         int ret;
 147
 148         if (ctdb->flags & CTDB_FLAG_TORTURE) {
 149                 struct ctdb_ltdb_header *h2;
 150                 rec = tdb_fetch(ctdb_db->ltdb->tdb, key);
 151                 h2 = (struct ctdb_ltdb_header *)rec.dptr;
 152                 if (rec.dptr && rec.dsize >= sizeof(h2) && h2->rsn > header->rsn) {
 153                         DEBUG(0,("RSN regression! %llu %llu\n",
 154                                  (unsigned long long)h2->rsn, (unsigned long long)header->rsn));
 155                 }
 156                 if (rec.dptr) free(rec.dptr);
 157         }
 158
 159         rec.dsize = sizeof(*header) + data.dsize;
 160         rec.dptr = talloc_size(ctdb, rec.dsize);
 161         CTDB_NO_MEMORY(ctdb, rec.dptr);
 162
 163         memcpy(rec.dptr, header, sizeof(*header));
 164         memcpy(rec.dptr + sizeof(*header), data.dptr, data.dsize);
 165
 166         ret = tdb_store(ctdb_db->ltdb->tdb, key, rec, TDB_REPLACE);
 167         talloc_free(rec.dptr);
 168
 169         return ret;
 170 }
 171
 172
 173 /*
 174   lock a record in the ltdb, given a key
 175  */
 176 int ctdb_ltdb_lock(struct ctdb_db_context *ctdb_db, TDB_DATA key)
 177 {
 178         return tdb_chainlock(ctdb_db->ltdb->tdb, key);
 179 }
 180
 181 /*
 182   unlock a record in the ltdb, given a key
 183  */
 184 int ctdb_ltdb_unlock(struct ctdb_db_context *ctdb_db, TDB_DATA key)
 185 {
 186         int ret = tdb_chainunlock(ctdb_db->ltdb->tdb, key);
 187         if (ret != 0) {
 188                 DEBUG(0,("tdb_chainunlock failed\n"));
 189         }
 190         return ret;
 191 }
 192
 193 struct lock_fetch_state {
 194         struct ctdb_context *ctdb;
 195         void (*recv_pkt)(void *, uint8_t *, uint32_t);
 196         void *recv_context;
 197         struct ctdb_req_header *hdr;
 198         uint32_t generation;
 199         bool ignore_generation;
 200 };
 201
 202 /*
 203   called when we should retry the operation
 204  */
 205 static void lock_fetch_callback(void *p)
 206 {
 207         struct lock_fetch_state *state = talloc_get_type(p, struct lock_fetch_state);
 208         if (!state->ignore_generation &&
 209             state->generation != state->ctdb->vnn_map->generation) {
 210                 DEBUG(0,("Discarding previous generation lockwait packet\n"));
 211                 talloc_free(state->hdr);
 212                 return;
 213         }
 214         state->recv_pkt(state->recv_context, (uint8_t *)state->hdr, state->hdr->length);
 215         DEBUG(2,(__location__ " PACKET REQUEUED\n"));
 216 }
 217
 218
 219 /*
 220   do a non-blocking ltdb_lock, deferring this ctdb request until we
 221   have the chainlock
 222
 223   It does the following:
 224
 225    1) tries to get the chainlock. If it succeeds, then it returns 0
 226
 227    2) if it fails to get a chainlock immediately then it sets up a
 228    non-blocking chainlock via ctdb_lockwait, and when it gets the
 229    chainlock it re-submits this ctdb request to the main packet
 230    receive function
 231
 232    This effectively queues all ctdb requests that cannot be
 233    immediately satisfied until it can get the lock. This means that
 234    the main ctdb daemon will not block waiting for a chainlock held by
 235    a client
 236
 237    There are 3 possible return values:
 238
 239        0:    means that it got the lock immediately.
 240       -1:    means that it failed to get the lock, and won't retry
 241       -2:    means that it failed to get the lock immediately, but will retry
 242  */
 243 int ctdb_ltdb_lock_requeue(struct ctdb_db_context *ctdb_db,
 244                            TDB_DATA key, struct ctdb_req_header *hdr,
 245                            void (*recv_pkt)(void *, uint8_t *, uint32_t ),
 246                            void *recv_context, bool ignore_generation)
 247 {
 248         int ret;
 249         struct tdb_context *tdb = ctdb_db->ltdb->tdb;
 250         struct lockwait_handle *h;
 251         struct lock_fetch_state *state;
 252
 253         ret = tdb_chainlock_nonblock(tdb, key);
 254
 255         if (ret != 0 &&
 256             !(errno == EACCES || errno == EAGAIN || errno == EDEADLK)) {
 257                 /* a hard failure - don't try again */
 258                 return -1;
 259         }
 260
 261         /* when torturing, ensure we test the contended path */
 262         if ((ctdb_db->ctdb->flags & CTDB_FLAG_TORTURE) &&
 263             random() % 5 == 0) {
 264                 ret = -1;
 265                 tdb_chainunlock(tdb, key);
 266         }
 267
 268         /* first the non-contended path */
 269         if (ret == 0) {
 270                 return 0;
 271         }
 272
 273         state = talloc(hdr, struct lock_fetch_state);
 274         state->ctdb = ctdb_db->ctdb;
 275         state->hdr = hdr;
 276         state->recv_pkt = recv_pkt;
 277         state->recv_context = recv_context;
 278         state->generation = ctdb_db->ctdb->vnn_map->generation;
 279         state->ignore_generation = ignore_generation;
 280
 281         /* now the contended path */
 282         h = ctdb_lockwait(ctdb_db, key, lock_fetch_callback, state);
 283         if (h == NULL) {
 284                 tdb_chainunlock(tdb, key);
 285                 return -1;
 286         }
 287
 288         /* we need to move the packet off the temporary context in ctdb_recv_pkt(),
 289            so it won't be freed yet */
 290         talloc_steal(state, hdr);
 291         talloc_steal(state, h);
 292
 293         /* now tell the caller than we will retry asynchronously */
 294         return -2;
 295 }
 296
 297 /*
 298   a varient of ctdb_ltdb_lock_requeue that also fetches the record
 299  */
 300 int ctdb_ltdb_lock_fetch_requeue(struct ctdb_db_context *ctdb_db,
 301                                  TDB_DATA key, struct ctdb_ltdb_header *header,
 302                                  struct ctdb_req_header *hdr, TDB_DATA *data,
 303                                  void (*recv_pkt)(void *, uint8_t *, uint32_t ),
 304                                  void *recv_context, bool ignore_generation)
 305 {
 306         int ret;
 307
 308         ret = ctdb_ltdb_lock_requeue(ctdb_db, key, hdr, recv_pkt,
 309                                      recv_context, ignore_generation);
 310         if (ret == 0) {
 311                 ret = ctdb_ltdb_fetch(ctdb_db, key, header, hdr, data);
 312                 if (ret != 0) {
 313                         ctdb_ltdb_unlock(ctdb_db, key);
 314                 }
 315         }
 316         return ret;
 317 }
 318
 319
 320 /*
 321   a client has asked to attach a new database
 322  */
 323 int32_t ctdb_control_db_attach(struct ctdb_context *ctdb, TDB_DATA indata,
 324                                TDB_DATA *outdata)
 325 {
 326         const char *db_name = (const char *)indata.dptr;
 327         struct ctdb_db_context *ctdb_db, *tmp_db;
 328         int ret;
 329
 330         /* see if we already have this name */
 331         for (tmp_db=ctdb->db_list;tmp_db;tmp_db=tmp_db->next) {
 332                 if (strcmp(db_name, tmp_db->db_name) == 0) {
 333                         /* this is not an error */
 334                         outdata->dptr  = (uint8_t *)&tmp_db->db_id;
 335                         outdata->dsize = sizeof(tmp_db->db_id);
 336                         return 0;
 337                 }
 338         }
 339
 340         ctdb_db = talloc_zero(ctdb, struct ctdb_db_context);
 341         CTDB_NO_MEMORY(ctdb, ctdb_db);
 342
 343         ctdb_db->ctdb = ctdb;
 344         ctdb_db->db_name = talloc_strdup(ctdb_db, db_name);
 345         CTDB_NO_MEMORY(ctdb, ctdb_db->db_name);
 346
 347         ctdb_db->db_id = ctdb_hash(&indata);
 348
 349         outdata->dptr  = (uint8_t *)&ctdb_db->db_id;
 350         outdata->dsize = sizeof(ctdb_db->db_id);
 351
 352         /* check for hash collisions */
 353         for (tmp_db=ctdb->db_list;tmp_db;tmp_db=tmp_db->next) {
 354                 if (tmp_db->db_id == ctdb_db->db_id) {
 355                         DEBUG(0,("db_id 0x%x hash collision. name1='%s' name2='%s'\n",
 356                                  db_name, tmp_db->db_name));
 357                         talloc_free(ctdb_db);
 358                         return -1;
 359                 }
 360         }
 361
 362         if (ctdb->db_directory == NULL) {
 363                 ctdb->db_directory = VARDIR "/ctdb";
 364         }
 365
 366         /* make sure the db directory exists */
 367         if (mkdir(ctdb->db_directory, 0700) == -1 && errno != EEXIST) {
 368                 DEBUG(0,(__location__ " Unable to create ctdb directory '%s'\n",
 369                          ctdb->db_directory));
 370                 talloc_free(ctdb_db);
 371                 return -1;
 372         }
 373
 374         /* open the database */
 375         ctdb_db->db_path = talloc_asprintf(ctdb_db, "%s/%s.%u",
 376                                            ctdb->db_directory,
 377                                            db_name, ctdb->vnn);
 378
 379         ctdb_db->ltdb = tdb_wrap_open(ctdb, ctdb_db->db_path, 0,
 380                                       TDB_CLEAR_IF_FIRST, O_CREAT|O_RDWR, 0666);
 381         if (ctdb_db->ltdb == NULL) {
 382                 DEBUG(0,("Failed to open tdb '%s'\n", ctdb_db->db_path));
 383                 talloc_free(ctdb_db);
 384                 return -1;
 385         }
 386
 387         DLIST_ADD(ctdb->db_list, ctdb_db);
 388
 389         /*
 390            all databases support the "null" function. we need this in
 391            order to do forced migration of records
 392         */
 393         ret = ctdb_daemon_set_call(ctdb, ctdb_db->db_id, ctdb_null_func, CTDB_NULL_FUNC);
 394         if (ret != 0) {
 395                 DEBUG(0,("Failed to setup null function for '%s'\n", ctdb_db->db_name));
 396                 talloc_free(ctdb_db);
 397                 return -1;
 398         }
 399
 400         /*
 401            all databases support the "fetch" function. we need this
 402            for efficient Samba3 ctdb fetch
 403         */
 404         ret = ctdb_daemon_set_call(ctdb, ctdb_db->db_id, ctdb_fetch_func, CTDB_FETCH_FUNC);
 405         if (ret != 0) {
 406                 DEBUG(0,("Failed to setup fetch function for '%s'\n", ctdb_db->db_name));
 407                 talloc_free(ctdb_db);
 408                 return -1;
 409         }
 410
 411         /* tell all the other nodes about this database */
 412         ctdb_daemon_send_control(ctdb, CTDB_BROADCAST_VNNMAP, 0,
 413                                  CTDB_CONTROL_DB_ATTACH, 0, CTDB_CTRL_FLAG_NOREPLY,
 414                                  indata, NULL, NULL);
 415
 416         DEBUG(1,("Attached to database '%s'\n", ctdb_db->db_path));
 417
 418         /* success */
 419         return 0;
 420 }
 421
 422 /*
 423   called when a broadcast seqnum update comes in
 424  */
 425 int32_t ctdb_ltdb_update_seqnum(struct ctdb_context *ctdb, uint32_t db_id, uint32_t srcnode)
 426 {
 427         struct ctdb_db_context *ctdb_db;
 428         if (srcnode == ctdb->vnn) {
 429                 /* don't update ourselves! */
 430                 return 0;
 431         }
 432
 433         ctdb_db = find_ctdb_db(ctdb, db_id);
 434         if (!ctdb_db) {
 435                 DEBUG(0,("Unknown db_id 0x%x in ctdb_ltdb_update_seqnum\n"));
 436                 return -1;
 437         }
 438
 439         tdb_increment_seqnum_nonblock(ctdb_db->ltdb->tdb);
 440         ctdb_db->seqnum = tdb_get_seqnum(ctdb_db->ltdb->tdb);
 441         return 0;
 442 }
 443
 444 /*
 445   timer to check for seqnum changes in a ltdb and propogate them
 446  */
 447 static void ctdb_ltdb_seqnum_check(struct event_context *ev, struct timed_event *te,
 448                                    struct timeval t, void *p)
 449 {
 450         struct ctdb_db_context *ctdb_db = talloc_get_type(p, struct ctdb_db_context);
 451         struct ctdb_context *ctdb = ctdb_db->ctdb;
 452         uint32_t new_seqnum = tdb_get_seqnum(ctdb_db->ltdb->tdb);
 453         if (new_seqnum != ctdb_db->seqnum) {
 454                 /* something has changed - propogate it */
 455                 TDB_DATA data;
 456                 data.dptr = (uint8_t *)&ctdb_db->db_id;
 457                 data.dsize = sizeof(uint32_t);
 458                 ctdb_daemon_send_control(ctdb, CTDB_BROADCAST_VNNMAP, 0,
 459                                          CTDB_CONTROL_UPDATE_SEQNUM, 0, CTDB_CTRL_FLAG_NOREPLY,
 460                                          data, NULL, NULL);
 461         }
 462         ctdb_db->seqnum = new_seqnum;
 463
 464         /* setup a new timer */
 465         ctdb_db->te = event_add_timed(ctdb->ev, ctdb_db,
 466                                       timeval_current_ofs(ctdb->seqnum_frequency, 0),
 467                                       ctdb_ltdb_seqnum_check, ctdb_db);
 468 }
 469
 470 /*
 471   enable seqnum handling on this db
 472  */
 473 int32_t ctdb_ltdb_enable_seqnum(struct ctdb_context *ctdb, uint32_t db_id)
 474 {
 475         struct ctdb_db_context *ctdb_db;
 476         ctdb_db = find_ctdb_db(ctdb, db_id);
 477         if (!ctdb_db) {
 478                 DEBUG(0,("Unknown db_id 0x%x in ctdb_ltdb_enable_seqnum\n"));
 479                 return -1;
 480         }
 481
 482         if (ctdb_db->te == NULL) {
 483                 ctdb_db->te = event_add_timed(ctdb->ev, ctdb_db,
 484                                               timeval_current_ofs(ctdb->seqnum_frequency, 0),
 485                                               ctdb_ltdb_seqnum_check, ctdb_db);
 486         }
 487
 488         tdb_enable_seqnum(ctdb_db->ltdb->tdb);
 489         ctdb_db->seqnum = tdb_get_seqnum(ctdb_db->ltdb->tdb);
 490         return 0;
 491 }
 492
 493 /*
 494   enable seqnum handling on this db
 495  */
 496 int32_t ctdb_ltdb_set_seqnum_frequency(struct ctdb_context *ctdb, uint32_t frequency)
 497 {
 498         ctdb->seqnum_frequency = frequency;
 499         return 0;
 500 }