citadel/database_sleepycat.c

   1 /*
   2  * $Id$
   3  *
   4  * Sleepycat (Berkeley) DB driver for Citadel/UX
   5  *
   6  */
   7
   8 /*****************************************************************************
   9        Tunable configuration parameters for the Sleepycat DB back end
  10  *****************************************************************************/
  11
  12 /* Citadel will checkpoint the db at the end of every session, but only if
  13  * the specified number of kilobytes has been written, or if the specified
  14  * number of minutes has passed, since the last checkpoint.
  15  */
  16 #define MAX_CHECKPOINT_KBYTES   256
  17 #define MAX_CHECKPOINT_MINUTES  15
  18
  19 /*****************************************************************************/
  20
  21 #ifdef DLL_EXPORT
  22 #define IN_LIBCIT
  23 #endif
  24
  25 #include "sysdep.h"
  26 #include <stdlib.h>
  27 #include <unistd.h>
  28 #include <stdio.h>
  29 #include <ctype.h>
  30 #include <string.h>
  31 #include <errno.h>
  32 #include <sys/types.h>
  33 #include <sys/stat.h>
  34 #include <dirent.h>
  35
  36 #ifdef HAVE_DB_H
  37 #include <db.h>
  38 #elif defined(HAVE_DB3_DB_H)
  39 #include <db3/db.h>
  40 #else
  41 #error Neither <db.h> nor <db3/db.h> was found by configure. Install db3-devel.
  42 #endif
  43
  44 #include <pthread.h>
  45 #include "citadel.h"
  46 #include "server.h"
  47 #include "dynloader.h"
  48 #include "citserver.h"
  49 #include "database.h"
  50 #include "sysdep_decls.h"
  51
  52 static DB *dbp[MAXCDB];         /* One DB handle for each Citadel database */
  53 static DB_ENV *dbenv;           /* The DB environment (global) */
  54
  55 struct cdbtsd {                 /* Thread-specific DB stuff */
  56         DB_TXN *tid;            /* Transaction handle */
  57         DBC *cursors[MAXCDB];   /* Cursors, for traversals... */
  58 };
  59
  60 static pthread_key_t tsdkey;
  61
  62 #define MYCURSORS       (((struct cdbtsd*)pthread_getspecific(tsdkey))->cursors)
  63 #define MYTID           (((struct cdbtsd*)pthread_getspecific(tsdkey))->tid)
  64
  65 /* just a little helper function */
  66 static void txabort(DB_TXN *tid) {
  67         int ret = txn_abort(tid);
  68
  69         if (ret) {
  70                 lprintf(1, "cdb_*: txn_abort: %s\n", db_strerror(ret));
  71                 abort();
  72         }
  73 }
  74
  75 /* this one is even more helpful than the last. */
  76 static void txcommit(DB_TXN *tid) {
  77         int ret = txn_commit(tid, 0);
  78
  79         if (ret) {
  80                 lprintf(1, "cdb_*: txn_commit: %s\n", db_strerror(ret));
  81                 abort();
  82         }
  83 }
  84
  85 /* are you sensing a pattern yet? */
  86 static void txbegin(DB_TXN **tid) {
  87         int ret = txn_begin(dbenv, NULL, tid, 0);
  88
  89         if (ret) {
  90                 lprintf(1, "cdb_*: txn_begin: %s\n", db_strerror(ret));
  91                 abort();
  92         }
  93 }
  94
  95 static void cclose(DBC *cursor) {
  96         int ret;
  97
  98         if ((ret = cursor->c_close(cursor))) {
  99                 lprintf(1, "cdb_*: c_close: %s\n", db_strerror(ret));
 100                 abort();
 101         }
 102 }
 103
 104 static void bailIfCursor(DBC **cursors, const char *msg)
 105 {
 106   int i;
 107
 108   for (i = 0; i < MAXCDB; i++)
 109     if (cursors[i] != NULL)
 110       {
 111         lprintf(1, "cdb_*: cursor still in progress on cdb %d: %s\n", i, msg);
 112         abort();
 113       }
 114 }
 115
 116 static void check_handles(void *arg) {
 117         if (arg != NULL) {
 118                 struct cdbtsd *tsd = (struct cdbtsd *)arg;
 119
 120                 bailIfCursor(tsd->cursors, "in check_handles");
 121
 122                 if (tsd->tid != NULL) {
 123                         lprintf(1, "cdb_*: transaction still in progress!");
 124                         abort();
 125                 }
 126         }
 127 }
 128
 129 static void dest_tsd(void *arg) {
 130         if (arg != NULL) {
 131                 check_handles(arg);
 132                 phree(arg);
 133         }
 134 }
 135
 136 /*
 137  * Ensure that we have a key for thread-specific data.  We don't
 138  * put anything in here that Citadel cares about; this is just database
 139  * related stuff like cursors and transactions.
 140  *
 141  * This should be called immediately after startup by any thread which wants
 142  * to use database calls, except for whatever thread calls open_databases.
 143  */
 144 void cdb_allocate_tsd(void) {
 145         struct cdbtsd *tsd;
 146
 147         if (pthread_getspecific(tsdkey) != NULL)
 148                 return;
 149
 150         tsd = mallok(sizeof *tsd);
 151
 152         tsd->tid = NULL;
 153
 154         memset(tsd->cursors, 0, sizeof tsd->cursors);
 155         pthread_setspecific(tsdkey, tsd);
 156 }
 157
 158 void cdb_free_tsd(void) {
 159         dest_tsd(pthread_getspecific(tsdkey));
 160         pthread_setspecific(tsdkey, NULL);
 161 }
 162
 163 void cdb_check_handles(void) {
 164         check_handles(pthread_getspecific(tsdkey));
 165 }
 166
 167
 168 /*
 169  * Reclaim unused space in the databases.  We need to do each one of
 170  * these discretely, rather than in a loop.
 171  *
 172  * This is a stub function in the Sleepycat DB backend, because there is no
 173  * such API call available.
 174  */
 175 void defrag_databases(void)
 176 {
 177         /* do nothing */
 178 }
 179
 180
 181 /*
 182  * Cull the database logs
 183  */
 184 static void cdb_cull_logs(void) {
 185         u_int32_t flags;
 186         int ret;
 187         char **file, **list;
 188
 189         lprintf(5, "Database log file cull started.\n");
 190
 191         flags = DB_ARCH_ABS;
 192
 193         /* Get the list of names. */
 194 #if DB_VERSION_MAJOR == 3 && DB_VERSION_MINOR < 3
 195         if ((ret = log_archive(dbenv, &list, flags, NULL)) != 0) {
 196 #elif DB_VERSION_MAJOR >= 4
 197         if ((ret = dbenv->log_archive(dbenv, &list, flags)) != 0) {
 198 #else
 199         if ((ret = log_archive(dbenv, &list, flags)) != 0) {
 200 #endif
 201                 lprintf(1, "cdb_cull_logs: %s\n", db_strerror(ret));
 202                 return;
 203         }
 204
 205         /* Print the list of names. */
 206         if (list != NULL) {
 207                 for (file = list; *file != NULL; ++file) {
 208                         lprintf(9, "Deleting log: %s\n", *file);
 209                         unlink(*file);
 210                 }
 211                 free(list);
 212         }
 213
 214         lprintf(5, "Database log file cull ended.\n");
 215 }
 216
 217
 218 /*
 219  * Request a checkpoint of the database.
 220  */
 221 static void cdb_checkpoint(void) {
 222         int ret;
 223         static time_t last_cull = 0L;
 224
 225 #if DB_VERSION_MAJOR >= 4
 226         ret = dbenv->txn_checkpoint(dbenv,
 227 #else
 228         ret = txn_checkpoint(dbenv,
 229 #endif
 230                                 MAX_CHECKPOINT_KBYTES,
 231                                 MAX_CHECKPOINT_MINUTES,
 232                                 0);
 233         if ( (ret != 0) && (ret != DB_INCOMPLETE) ) {
 234                 lprintf(1, "cdb_checkpoint: txn_checkpoint: %s\n", db_strerror(ret));
 235                 abort();
 236         }
 237
 238         if (ret == DB_INCOMPLETE) {
 239                 lprintf(3, "WARNING: txn_checkpoint: %s\n", db_strerror(ret));
 240         }
 241
 242         /* Cull the logs if we haven't done so for 24 hours */
 243         if ((time(NULL) - last_cull) > 86400L) {
 244                 last_cull = time(NULL);
 245                 cdb_cull_logs();
 246         }
 247
 248 }
 249
 250 /*
 251  * Open the various databases we'll be using.  Any database which
 252  * does not exist should be created.  Note that we don't need an S_DATABASE
 253  * critical section here, because there aren't any active threads manipulating
 254  * the database yet -- and besides, it causes problems on BSDI.
 255  */
 256 void open_databases(void)
 257 {
 258         int ret;
 259         int i;
 260         char dbfilename[SIZ];
 261         u_int32_t flags = 0;
 262
 263         lprintf(9, "cdb_*: open_databases() starting\n");
 264         /*
 265          * Silently try to create the database subdirectory.  If it's
 266          * already there, no problem.
 267          */
 268         system("exec mkdir data 2>/dev/null");
 269
 270         lprintf(9, "cdb_*: Setting up DB environment\n");
 271         db_env_set_func_yield(sched_yield);
 272         ret = db_env_create(&dbenv, 0);
 273         if (ret) {
 274                 lprintf(1, "cdb_*: db_env_create: %s\n", db_strerror(ret));
 275                 exit(ret);
 276         }
 277         dbenv->set_errpfx(dbenv, "citserver");
 278
 279         /*
 280          * We want to specify the shared memory buffer pool cachesize,
 281          * but everything else is the default.
 282          */
 283         ret = dbenv->set_cachesize(dbenv, 0, 64 * 1024, 0);
 284         if (ret) {
 285                 lprintf(1, "cdb_*: set_cachesize: %s\n", db_strerror(ret));
 286                 dbenv->close(dbenv, 0);
 287                 exit(ret);
 288         }
 289
 290         if ((ret = dbenv->set_lk_detect(dbenv, DB_LOCK_DEFAULT))) {
 291                 lprintf(1, "cdb_*: set_lk_detect: %s\n", db_strerror(ret));
 292                 dbenv->close(dbenv, 0);
 293                 exit(ret);
 294         }
 295
 296         flags = DB_CREATE|DB_RECOVER|DB_INIT_MPOOL|DB_PRIVATE|DB_INIT_TXN|
 297                 DB_INIT_LOCK|DB_THREAD;
 298         ret = dbenv->open(dbenv, "./data", flags, 0);
 299         if (ret) {
 300                 lprintf(1, "cdb_*: dbenv->open: %s\n", db_strerror(ret));
 301                 dbenv->close(dbenv, 0);
 302                 exit(ret);
 303         }
 304
 305         lprintf(7, "cdb_*: Starting up DB\n");
 306
 307         for (i = 0; i < MAXCDB; ++i) {
 308
 309                 /* Create a database handle */
 310                 ret = db_create(&dbp[i], dbenv, 0);
 311                 if (ret) {
 312                         lprintf(1, "cdb_*: db_create: %s\n", db_strerror(ret));
 313                         exit(ret);
 314                 }
 315
 316
 317                 /* Arbitrary names for our tables -- we reference them by
 318                  * number, so we don't have string names for them.
 319                  */
 320                 sprintf(dbfilename, "cdb.%02x", i);
 321
 322                 ret = dbp[i]->open(dbp[i],
 323                                 dbfilename,
 324                                 NULL,
 325                                 DB_BTREE,
 326                                 DB_CREATE|DB_THREAD,
 327                                 0600);
 328                 if (ret) {
 329                         lprintf(1, "cdb_*: db_open[%d]: %s\n", i, db_strerror(ret));
 330                         exit(ret);
 331                 }
 332         }
 333
 334         if ((ret = pthread_key_create(&tsdkey, dest_tsd))) {
 335                 lprintf(1, "cdb_*: pthread_key_create: %s\n", strerror(ret));
 336                 exit(1);
 337         }
 338
 339         cdb_allocate_tsd();
 340         CtdlRegisterSessionHook(cdb_checkpoint, EVT_TIMER);
 341         lprintf(9, "cdb_*: open_databases() finished\n");
 342 }
 343
 344
 345 /*
 346  * Close all of the db database files we've opened.  This can be done
 347  * in a loop, since it's just a bunch of closes.
 348  */
 349 void close_databases(void)
 350 {
 351         int a;
 352         int ret;
 353
 354         cdb_free_tsd();
 355
 356 #if DB_VERSION_MAJOR >= 4
 357         if ((ret = dbenv->txn_checkpoint(dbenv, 0, 0, 0))) {
 358 #else
 359         if ((ret = txn_checkpoint(dbenv, 0, 0, 0))) {
 360 #endif
 361                 lprintf(1, "cdb_*: txn_checkpoint: %s\n", db_strerror(ret));
 362                 abort();
 363         }
 364
 365         for (a = 0; a < MAXCDB; ++a) {
 366                 lprintf(7, "cdb_*: Closing database %d\n", a);
 367                 ret = dbp[a]->close(dbp[a], 0);
 368                 if (ret) {
 369                         lprintf(1, "cdb_*: db_close: %s\n", db_strerror(ret));
 370                         abort();
 371                 }
 372
 373         }
 374
 375         /* Close the handle. */
 376         ret = dbenv->close(dbenv, 0);
 377         if (ret) {
 378                 lprintf(1, "cdb_*: DBENV->close: %s\n", db_strerror(ret));
 379                 abort();
 380         }
 381 }
 382
 383 /*
 384  * Store a piece of data.  Returns 0 if the operation was successful.  If a
 385  * key already exists it should be overwritten.
 386  */
 387 int cdb_store(int cdb,
 388               void *ckey, int ckeylen,
 389               void *cdata, int cdatalen)
 390 {
 391
 392   DBT dkey, ddata;
 393   DB_TXN *tid;
 394   int ret;
 395
 396   memset(&dkey, 0, sizeof(DBT));
 397   memset(&ddata, 0, sizeof(DBT));
 398   dkey.size = ckeylen;
 399   dkey.data = ckey;
 400   ddata.size = cdatalen;
 401   ddata.data = cdata;
 402
 403   if (MYTID != NULL)
 404     {
 405       ret = dbp[cdb]->put(dbp[cdb],             /* db */
 406                           MYTID,                /* transaction ID */
 407                           &dkey,                /* key */
 408                           &ddata,               /* data */
 409                           0);           /* flags */
 410       if (ret)
 411         {
 412           lprintf(1, "cdb_store(%d): %s\n", cdb,
 413                   db_strerror(ret));
 414           abort();
 415         }
 416       return ret;
 417
 418     }
 419   else
 420     {
 421       bailIfCursor(MYCURSORS, "attempt to write during r/o cursor");
 422
 423     retry:
 424       txbegin(&tid);
 425
 426       if ((ret = dbp[cdb]->put(dbp[cdb],    /* db */
 427                                tid,         /* transaction ID */
 428                                &dkey,       /* key */
 429                                &ddata,      /* data */
 430                                0)))         /* flags */
 431         {
 432           if (ret == DB_LOCK_DEADLOCK)
 433             {
 434               txabort(tid);
 435               goto retry;
 436             }
 437           else
 438             {
 439               lprintf(1, "cdb_store(%d): %s\n", cdb,
 440                       db_strerror(ret));
 441               abort();
 442             }
 443         }
 444       else
 445         {
 446           txcommit(tid);
 447           return ret;
 448         }
 449     }
 450 }
 451
 452
 453 /*
 454  * Delete a piece of data.  Returns 0 if the operation was successful.
 455  */
 456 int cdb_delete(int cdb, void *key, int keylen)
 457 {
 458
 459   DBT dkey;
 460   DB_TXN *tid;
 461   int ret;
 462
 463   memset(&dkey, 0, sizeof dkey);
 464   dkey.size = keylen;
 465   dkey.data = key;
 466
 467   if (MYTID != NULL)
 468     {
 469       ret = dbp[cdb]->del(dbp[cdb], MYTID, &dkey, 0);
 470       if (ret)
 471         {
 472           lprintf(1, "cdb_delete(%d): %s\n", cdb,
 473                   db_strerror(ret));
 474           if (ret != DB_NOTFOUND)
 475             abort();
 476         }
 477     }
 478   else
 479     {
 480       bailIfCursor(MYCURSORS, "attempt to delete during r/o cursor");
 481
 482     retry:
 483       txbegin(&tid);
 484
 485       if ((ret = dbp[cdb]->del(dbp[cdb], tid, &dkey, 0))
 486           && ret != DB_NOTFOUND)
 487         {
 488           if (ret == DB_LOCK_DEADLOCK)
 489             {
 490               txabort(tid);
 491               goto retry;
 492             }
 493           else
 494             {
 495               lprintf(1, "cdb_delete(%d): %s\n", cdb,
 496                       db_strerror(ret));
 497               abort();
 498             }
 499         }
 500       else
 501         {
 502           txcommit(tid);
 503         }
 504     }
 505   return ret;
 506 }
 507
 508 static DBC *localcursor(int cdb)
 509 {
 510   int ret;
 511   DBC *curs;
 512
 513   if (MYCURSORS[cdb] == NULL)
 514     ret = dbp[cdb]->cursor(dbp[cdb], MYTID, &curs, 0);
 515   else
 516     ret = MYCURSORS[cdb]->c_dup(MYCURSORS[cdb], &curs, DB_POSITION);
 517
 518   if (ret)
 519     {
 520       lprintf(1, "localcursor: %s\n", db_strerror(ret));
 521       abort();
 522     }
 523
 524   return curs;
 525 }
 526
 527
 528 /*
 529  * Fetch a piece of data.  If not found, returns NULL.  Otherwise, it returns
 530  * a struct cdbdata which it is the caller's responsibility to free later on
 531  * using the cdb_free() routine.
 532  */
 533 struct cdbdata *cdb_fetch(int cdb, void *key, int keylen)
 534 {
 535
 536   struct cdbdata *tempcdb;
 537   DBT dkey, dret;
 538   int ret;
 539
 540   memset(&dkey, 0, sizeof(DBT));
 541   dkey.size = keylen;
 542   dkey.data = key;
 543
 544   if (MYTID != NULL)
 545     {
 546       memset(&dret, 0, sizeof(DBT));
 547       dret.flags = DB_DBT_MALLOC;
 548       ret = dbp[cdb]->get(dbp[cdb], MYTID, &dkey, &dret, 0);
 549     }
 550   else
 551     {
 552       DBC *curs;
 553
 554       do
 555         {
 556           memset(&dret, 0, sizeof(DBT));
 557           dret.flags = DB_DBT_MALLOC;
 558
 559           curs = localcursor(cdb);
 560
 561           ret = curs->c_get(curs, &dkey, &dret, DB_SET);
 562           cclose(curs);
 563         }
 564       while (ret == DB_LOCK_DEADLOCK);
 565
 566     }
 567
 568   if ((ret != 0) && (ret != DB_NOTFOUND))
 569     {
 570       lprintf(1, "cdb_fetch(%d): %s\n", cdb, db_strerror(ret));
 571       abort();
 572     }
 573
 574   if (ret != 0) return NULL;
 575   tempcdb = (struct cdbdata *) mallok(sizeof(struct cdbdata));
 576
 577   if (tempcdb == NULL)
 578     {
 579       lprintf(2, "cdb_fetch: Cannot allocate memory for tempcdb\n");
 580       abort();
 581     }
 582
 583   tempcdb->len = dret.size;
 584   tempcdb->ptr = dret.data;
 585   return (tempcdb);
 586 }
 587
 588
 589 /*
 590  * Free a cdbdata item (ok, this is really no big deal, but we might need to do
 591  * more complex stuff with other database managers in the future).
 592  */
 593 void cdb_free(struct cdbdata *cdb)
 594 {
 595         phree(cdb->ptr);
 596         phree(cdb);
 597 }
 598
 599 void cdb_close_cursor(int cdb)
 600 {
 601         if (MYCURSORS[cdb] != NULL)
 602                 cclose(MYCURSORS[cdb]);
 603
 604         MYCURSORS[cdb] = NULL;
 605 }
 606
 607 /*
 608  * Prepare for a sequential search of an entire database.
 609  * (There is guaranteed to be no more than one traversal in
 610  * progress per thread at any given time.)
 611  */
 612 void cdb_rewind(int cdb)
 613 {
 614         int ret = 0;
 615
 616         if (MYCURSORS[cdb] != NULL)
 617                 cclose(MYCURSORS[cdb]);
 618
 619         /*
 620          * Now initialize the cursor
 621          */
 622         ret = dbp[cdb]->cursor(dbp[cdb], MYTID, &MYCURSORS[cdb], 0);
 623         if (ret) {
 624                 lprintf(1, "cdb_rewind: db_cursor: %s\n", db_strerror(ret));
 625                 abort();
 626         }
 627 }
 628
 629
 630 /*
 631  * Fetch the next item in a sequential search.  Returns a pointer to a
 632  * cdbdata structure, or NULL if we've hit the end.
 633  */
 634 struct cdbdata *cdb_next_item(int cdb)
 635 {
 636         DBT key, data;
 637         struct cdbdata *cdbret;
 638         int ret = 0;
 639
 640         /* Initialize the key/data pair so the flags aren't set. */
 641         memset(&key, 0, sizeof(key));
 642         memset(&data, 0, sizeof(data));
 643         data.flags = DB_DBT_MALLOC;
 644
 645         ret = MYCURSORS[cdb]->c_get(MYCURSORS[cdb],
 646                 &key, &data, DB_NEXT);
 647
 648         if (ret) {
 649                 if (ret != DB_NOTFOUND) {
 650                         lprintf(1, "cdb_next_item(%d): %s\n",
 651                                 cdb, db_strerror(ret));
 652                         abort();
 653                 }
 654                 cclose(MYCURSORS[cdb]);
 655                 MYCURSORS[cdb] = NULL;
 656                 return NULL;            /* presumably, end of file */
 657         }
 658
 659         cdbret = (struct cdbdata *) mallok(sizeof(struct cdbdata));
 660         cdbret->len = data.size;
 661         cdbret->ptr = data.data;
 662
 663         return (cdbret);
 664 }
 665
 666
 667
 668 /*
 669  * Truncate (delete every record)
 670  */
 671 void cdb_trunc(int cdb) {
 672         /* FIXME this needs to be implemented */
 673 }
 674
 675
 676 /*
 677  * Transaction-based stuff.  I'm writing this as I bake cookies...
 678  */
 679
 680 void cdb_begin_transaction(void) {
 681
 682   bailIfCursor(MYCURSORS, "can't begin transaction during r/o cursor");
 683
 684   if (MYTID != NULL)
 685     {
 686       lprintf(1, "cdb_begin_transaction: ERROR: nested transaction\n");
 687       abort();
 688     }
 689
 690   txbegin(&MYTID);
 691 }
 692
 693 void cdb_end_transaction(void) {
 694   int i;
 695
 696   for (i = 0; i < MAXCDB; i++)
 697     if (MYCURSORS[i] != NULL) {
 698       lprintf(1, "cdb_end_transaction: WARNING: cursor %d still open at transaction end\n", i);
 699       cclose(MYCURSORS[i]);
 700       MYCURSORS[i] = NULL;
 701     }
 702
 703   if (MYTID == NULL)
 704     {
 705       lprintf(1, "cdb_end_transaction: ERROR: txcommit(NULL) !!\n");
 706       abort();
 707     }
 708   else
 709     txcommit(MYTID);
 710
 711   MYTID = NULL;
 712 }
 713