citadel/database_sleepycat.c

   1 /*
   2  * $Id$
   3  *
   4  * Sleepycat (Berkeley) DB driver for Citadel
   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_DB4_DB_H)
  39 #include <db4/db.h>
  40 #else
  41 #error Neither <db.h> nor <db4/db.h> was found by configure. Install db4-devel.
  42 #endif
  43
  44
  45 #if DB_VERSION_MAJOR < 4 || DB_VERSION_MINOR < 1
  46 #error Citadel requires Berkeley DB v4.1 or newer.  Please upgrade.
  47 #endif
  48
  49
  50 #include <pthread.h>
  51 #include "citadel.h"
  52 #include "server.h"
  53 #include "serv_extensions.h"
  54 #include "citserver.h"
  55 #include "database.h"
  56 #include "msgbase.h"
  57 #include "sysdep_decls.h"
  58 #include "config.h"
  59
  60 static DB *dbp[MAXCDB];         /* One DB handle for each Citadel database */
  61 static DB_ENV *dbenv;           /* The DB environment (global) */
  62
  63 struct cdbtsd {                 /* Thread-specific DB stuff */
  64         DB_TXN *tid;            /* Transaction handle */
  65         DBC *cursors[MAXCDB];   /* Cursors, for traversals... */
  66 };
  67
  68 #ifdef HAVE_ZLIB
  69 #include <zlib.h>
  70 #endif
  71
  72 static pthread_key_t tsdkey;
  73
  74 #define MYCURSORS       (((struct cdbtsd*)pthread_getspecific(tsdkey))->cursors)
  75 #define MYTID           (((struct cdbtsd*)pthread_getspecific(tsdkey))->tid)
  76
  77 /* just a little helper function */
  78 static void txabort(DB_TXN * tid)
  79 {
  80         int ret;
  81
  82         ret = tid->abort(tid);
  83
  84         if (ret) {
  85                 lprintf(CTDL_EMERG, "cdb_*: txn_abort: %s\n",
  86                         db_strerror(ret));
  87                 abort();
  88         }
  89 }
  90
  91 /* this one is even more helpful than the last. */
  92 static void txcommit(DB_TXN * tid)
  93 {
  94         int ret;
  95
  96         ret = tid->commit(tid, 0);
  97
  98         if (ret) {
  99                 lprintf(CTDL_EMERG, "cdb_*: txn_commit: %s\n",
 100                         db_strerror(ret));
 101                 abort();
 102         }
 103 }
 104
 105 /* are you sensing a pattern yet? */
 106 static void txbegin(DB_TXN ** tid)
 107 {
 108         int ret;
 109
 110         ret = dbenv->txn_begin(dbenv, NULL, tid, 0);
 111
 112         if (ret) {
 113                 lprintf(CTDL_EMERG, "cdb_*: txn_begin: %s\n",
 114                         db_strerror(ret));
 115                 abort();
 116         }
 117 }
 118
 119 static void dbpanic(DB_ENV * env, int errval)
 120 {
 121         lprintf(CTDL_EMERG, "cdb_*: Berkeley DB panic: %d\n", errval);
 122 }
 123
 124 static void cclose(DBC * cursor)
 125 {
 126         int ret;
 127
 128         if ((ret = cursor->c_close(cursor))) {
 129                 lprintf(CTDL_EMERG, "cdb_*: c_close: %s\n",
 130                         db_strerror(ret));
 131                 abort();
 132         }
 133 }
 134
 135 static void bailIfCursor(DBC ** cursors, const char *msg)
 136 {
 137         int i;
 138
 139         for (i = 0; i < MAXCDB; i++)
 140                 if (cursors[i] != NULL) {
 141                         lprintf(CTDL_EMERG,
 142                                 "cdb_*: cursor still in progress on cdb %d: %s\n",
 143                                 i, msg);
 144                         abort();
 145                 }
 146 }
 147
 148 static void check_handles(void *arg)
 149 {
 150         if (arg != NULL) {
 151                 struct cdbtsd *tsd = (struct cdbtsd *) arg;
 152
 153                 bailIfCursor(tsd->cursors, "in check_handles");
 154
 155                 if (tsd->tid != NULL) {
 156                         lprintf(CTDL_EMERG,
 157                                 "cdb_*: transaction still in progress!");
 158                         abort();
 159                 }
 160         }
 161 }
 162
 163 static void dest_tsd(void *arg)
 164 {
 165         if (arg != NULL) {
 166                 check_handles(arg);
 167                 free(arg);
 168         }
 169 }
 170
 171 /*
 172  * Ensure that we have a key for thread-specific data.  We don't
 173  * put anything in here that Citadel cares about; this is just database
 174  * related stuff like cursors and transactions.
 175  *
 176  * This should be called immediately after startup by any thread which wants
 177  * to use database calls, except for whatever thread calls open_databases.
 178  */
 179 void cdb_allocate_tsd(void)
 180 {
 181         struct cdbtsd *tsd;
 182
 183         if (pthread_getspecific(tsdkey) != NULL)
 184                 return;
 185
 186         tsd = malloc(sizeof(struct cdbtsd));
 187
 188         tsd->tid = NULL;
 189
 190         memset(tsd->cursors, 0, sizeof tsd->cursors);
 191         pthread_setspecific(tsdkey, tsd);
 192 }
 193
 194 void cdb_free_tsd(void)
 195 {
 196         dest_tsd(pthread_getspecific(tsdkey));
 197         pthread_setspecific(tsdkey, NULL);
 198 }
 199
 200 void cdb_check_handles(void)
 201 {
 202         check_handles(pthread_getspecific(tsdkey));
 203 }
 204
 205
 206 /*
 207  * Reclaim unused space in the databases.  We need to do each one of
 208  * these discretely, rather than in a loop.
 209  *
 210  * This is a stub function in the Sleepycat DB backend, because there is no
 211  * such API call available.
 212  */
 213 void defrag_databases(void)
 214 {
 215         /* do nothing */
 216 }
 217
 218
 219 /*
 220  * Cull the database logs
 221  */
 222 static void cdb_cull_logs(void)
 223 {
 224         u_int32_t flags;
 225         int ret;
 226         char **file, **list;
 227         char errmsg[SIZ];
 228
 229         flags = DB_ARCH_ABS;
 230
 231         /* Get the list of names. */
 232         if ((ret = dbenv->log_archive(dbenv, &list, flags)) != 0) {
 233                 lprintf(CTDL_ERR, "cdb_cull_logs: %s\n", db_strerror(ret));
 234                 return;
 235         }
 236
 237         /* Print the list of names. */
 238         if (list != NULL) {
 239                 for (file = list; *file != NULL; ++file) {
 240                         lprintf(CTDL_DEBUG, "Deleting log: %s\n", *file);
 241                         ret = unlink(*file);
 242                         if (ret != 0) {
 243                                 snprintf(errmsg, sizeof(errmsg),
 244                                          " ** ERROR **\n \n \n "
 245                                          "Citadel was unable to delete the "
 246                                          "database log file '%s' because of the "
 247                                          "following error:\n \n %s\n \n"
 248                                          " This log file is no longer in use "
 249                                          "and may be safely deleted.\n",
 250                                          *file, strerror(errno));
 251                                 aide_message(errmsg);
 252                         }
 253                 }
 254                 free(list);
 255         }
 256 }
 257
 258 /*
 259  * Manually initiate log file cull.
 260  */
 261 void cmd_cull(char *argbuf) {
 262         if (CtdlAccessCheck(ac_internal)) return;
 263         cdb_cull_logs();
 264         cprintf("%d Database log file cull completed.\n", CIT_OK);
 265 }
 266
 267
 268 /*
 269  * Request a checkpoint of the database.
 270  */
 271 static void cdb_checkpoint(void)
 272 {
 273         int ret;
 274         static time_t last_run = 0L;
 275
 276         /* Only do a checkpoint once per minute. */
 277         if ((time(NULL) - last_run) < 60L) {
 278                 return;
 279         }
 280         last_run = time(NULL);
 281
 282         lprintf(CTDL_DEBUG, "-- db checkpoint --\n");
 283         ret = dbenv->txn_checkpoint(dbenv,
 284                                     MAX_CHECKPOINT_KBYTES,
 285                                     MAX_CHECKPOINT_MINUTES, 0);
 286
 287         if (ret != 0) {
 288                 lprintf(CTDL_EMERG, "cdb_checkpoint: txn_checkpoint: %s\n",
 289                         db_strerror(ret));
 290                 abort();
 291         }
 292
 293         /* After a successful checkpoint, we can cull the unused logs */
 294         if (config.c_auto_cull) {
 295                 cdb_cull_logs();
 296         }
 297 }
 298
 299
 300 /*
 301  * Main loop for the checkpoint thread.
 302  */
 303 void *checkpoint_thread(void *arg) {
 304         struct CitContext checkpointCC;
 305
 306         lprintf(CTDL_DEBUG, "checkpoint_thread() initializing\n");
 307
 308         memset(&checkpointCC, 0, sizeof(struct CitContext));
 309         checkpointCC.internal_pgm = 1;
 310         checkpointCC.cs_pid = 0;
 311         pthread_setspecific(MyConKey, (void *)&checkpointCC );
 312
 313         cdb_allocate_tsd();
 314
 315         while (!time_to_die) {
 316                 cdb_checkpoint();
 317                 sleep(1);
 318         }
 319
 320         lprintf(CTDL_DEBUG, "checkpoint_thread() exiting\n");
 321         pthread_exit(NULL);
 322 }
 323
 324 /*
 325  * Open the various databases we'll be using.  Any database which
 326  * does not exist should be created.  Note that we don't need a
 327  * critical section here, because there aren't any active threads
 328  * manipulating the database yet.
 329  */
 330 void open_databases(void)
 331 {
 332         int ret;
 333         int i;
 334         char dbfilename[SIZ];
 335         u_int32_t flags = 0;
 336         char dbdirname[PATH_MAX];
 337         DIR *dp;
 338         struct dirent *d;
 339         char filename[PATH_MAX];
 340
 341
 342         getcwd(dbdirname, sizeof dbdirname);
 343         strcat(dbdirname, "/data");
 344
 345         lprintf(CTDL_DEBUG, "cdb_*: open_databases() starting\n");
 346         lprintf(CTDL_DEBUG, "Compiled db: %s\n", DB_VERSION_STRING);
 347         lprintf(CTDL_INFO, "  Linked db: %s\n",
 348                 db_version(NULL, NULL, NULL));
 349 #ifdef HAVE_ZLIB
 350         lprintf(CTDL_INFO, "Linked zlib: %s\n", zlibVersion());
 351 #endif
 352
 353         /*
 354          * Silently try to create the database subdirectory.  If it's
 355          * already there, no problem.
 356          */
 357         mkdir(dbdirname, 0700);
 358         chmod(dbdirname, 0700);
 359         chown(dbdirname, CTDLUID, (-1));
 360
 361         lprintf(CTDL_DEBUG, "cdb_*: Setting up DB environment\n");
 362         db_env_set_func_yield(sched_yield);
 363         ret = db_env_create(&dbenv, 0);
 364         if (ret) {
 365                 lprintf(CTDL_EMERG, "cdb_*: db_env_create: %s\n",
 366                         db_strerror(ret));
 367                 exit(ret);
 368         }
 369         dbenv->set_errpfx(dbenv, "citserver");
 370         dbenv->set_paniccall(dbenv, dbpanic);
 371
 372         /*
 373          * We want to specify the shared memory buffer pool cachesize,
 374          * but everything else is the default.
 375          */
 376         ret = dbenv->set_cachesize(dbenv, 0, 64 * 1024, 0);
 377         if (ret) {
 378                 lprintf(CTDL_EMERG, "cdb_*: set_cachesize: %s\n",
 379                         db_strerror(ret));
 380                 dbenv->close(dbenv, 0);
 381                 exit(ret);
 382         }
 383
 384         if ((ret = dbenv->set_lk_detect(dbenv, DB_LOCK_DEFAULT))) {
 385                 lprintf(CTDL_EMERG, "cdb_*: set_lk_detect: %s\n",
 386                         db_strerror(ret));
 387                 dbenv->close(dbenv, 0);
 388                 exit(ret);
 389         }
 390
 391         flags =
 392             DB_CREATE | DB_RECOVER | DB_INIT_MPOOL | DB_PRIVATE |
 393             DB_INIT_TXN | DB_INIT_LOCK | DB_THREAD;
 394         lprintf(CTDL_DEBUG, "dbenv->open(dbenv, %s, %d, 0)\n", dbdirname,
 395                 flags);
 396         ret = dbenv->open(dbenv, dbdirname, flags, 0);
 397         if (ret) {
 398                 lprintf(CTDL_DEBUG, "cdb_*: dbenv->open: %s\n",
 399                         db_strerror(ret));
 400                 dbenv->close(dbenv, 0);
 401                 exit(ret);
 402         }
 403
 404         lprintf(CTDL_INFO, "cdb_*: Starting up DB\n");
 405
 406         for (i = 0; i < MAXCDB; ++i) {
 407
 408                 /* Create a database handle */
 409                 ret = db_create(&dbp[i], dbenv, 0);
 410                 if (ret) {
 411                         lprintf(CTDL_DEBUG, "cdb_*: db_create: %s\n",
 412                                 db_strerror(ret));
 413                         exit(ret);
 414                 }
 415
 416
 417                 /* Arbitrary names for our tables -- we reference them by
 418                  * number, so we don't have string names for them.
 419                  */
 420                 snprintf(dbfilename, sizeof dbfilename, "cdb.%02x", i);
 421
 422                 ret = dbp[i]->open(dbp[i],
 423                                    NULL,
 424                                    dbfilename,
 425                                    NULL,
 426                                    DB_BTREE,
 427                                    DB_CREATE | DB_AUTO_COMMIT | DB_THREAD,
 428                                    0600);
 429                 if (ret) {
 430                         lprintf(CTDL_EMERG, "cdb_*: db_open[%d]: %s\n", i,
 431                                 db_strerror(ret));
 432                         exit(ret);
 433                 }
 434         }
 435
 436         if ((ret = pthread_key_create(&tsdkey, dest_tsd))) {
 437                 lprintf(CTDL_EMERG, "cdb_*: pthread_key_create: %s\n",
 438                         strerror(ret));
 439                 exit(1);
 440         }
 441
 442         cdb_allocate_tsd();
 443
 444         /* Now make sure we own all the files, because in a few milliseconds
 445          * we're going to drop root privs.
 446          */
 447         dp = opendir(dbdirname);
 448         if (dp != NULL) {
 449                 while (d = readdir(dp), d != NULL) {
 450                         if (d->d_name[0] != '.') {
 451                                 snprintf(filename, sizeof filename,
 452                                          "%s/%s", dbdirname, d->d_name);
 453                                 chmod(filename, 0600);
 454                                 chown(filename, CTDLUID, (-1));
 455                         }
 456                 }
 457                 closedir(dp);
 458         }
 459
 460         lprintf(CTDL_DEBUG, "cdb_*: open_databases() finished\n");
 461
 462         CtdlRegisterProtoHook(cmd_cull, "CULL", "Cull database logs");
 463 }
 464
 465
 466 /*
 467  * Close all of the db database files we've opened.  This can be done
 468  * in a loop, since it's just a bunch of closes.
 469  */
 470 void close_databases(void)
 471 {
 472         int a;
 473         int ret;
 474
 475         cdb_free_tsd();
 476
 477         if ((ret = dbenv->txn_checkpoint(dbenv, 0, 0, 0))) {
 478                 lprintf(CTDL_EMERG,
 479                         "cdb_*: txn_checkpoint: %s\n", db_strerror(ret));
 480         }
 481
 482         for (a = 0; a < MAXCDB; ++a) {
 483                 lprintf(CTDL_INFO, "cdb_*: Closing database %d\n", a);
 484                 ret = dbp[a]->close(dbp[a], 0);
 485                 if (ret) {
 486                         lprintf(CTDL_EMERG,
 487                                 "cdb_*: db_close: %s\n", db_strerror(ret));
 488                 }
 489
 490         }
 491
 492         /* Close the handle. */
 493         ret = dbenv->close(dbenv, 0);
 494         if (ret) {
 495                 lprintf(CTDL_EMERG,
 496                         "cdb_*: DBENV->close: %s\n", db_strerror(ret));
 497         }
 498 }
 499
 500
 501 /*
 502  * Compression functions only used if we have zlib
 503  */
 504 #ifdef HAVE_ZLIB
 505
 506 void cdb_decompress_if_necessary(struct cdbdata *cdb)
 507 {
 508         static int magic = COMPRESS_MAGIC;
 509         struct CtdlCompressHeader zheader;
 510         char *uncompressed_data;
 511         char *compressed_data;
 512         uLongf destLen, sourceLen;
 513
 514         if (cdb == NULL)
 515                 return;
 516         if (cdb->ptr == NULL)
 517                 return;
 518         if (memcmp(cdb->ptr, &magic, sizeof(magic)))
 519                 return;
 520
 521         /* At this point we know we're looking at a compressed item. */
 522         memcpy(&zheader, cdb->ptr, sizeof(struct CtdlCompressHeader));
 523
 524         compressed_data = cdb->ptr;
 525         compressed_data += sizeof(struct CtdlCompressHeader);
 526
 527         sourceLen = (uLongf) zheader.compressed_len;
 528         destLen = (uLongf) zheader.uncompressed_len;
 529         uncompressed_data = malloc(zheader.uncompressed_len);
 530
 531         if (uncompress((Bytef *) uncompressed_data,
 532                        (uLongf *) & destLen,
 533                        (const Bytef *) compressed_data,
 534                        (uLong) sourceLen) != Z_OK) {
 535                 lprintf(CTDL_EMERG, "uncompress() error\n");
 536                 abort();
 537         }
 538
 539         free(cdb->ptr);
 540         cdb->len = (size_t) destLen;
 541         cdb->ptr = uncompressed_data;
 542 }
 543
 544 #endif                          /* HAVE_ZLIB */
 545
 546
 547 /*
 548  * Store a piece of data.  Returns 0 if the operation was successful.  If a
 549  * key already exists it should be overwritten.
 550  */
 551 int cdb_store(int cdb, void *ckey, int ckeylen, void *cdata, int cdatalen)
 552 {
 553
 554         DBT dkey, ddata;
 555         DB_TXN *tid;
 556         int ret;
 557
 558 #ifdef HAVE_ZLIB
 559         struct CtdlCompressHeader zheader;
 560         char *compressed_data = NULL;
 561         int compressing = 0;
 562         size_t buffer_len;
 563         uLongf destLen;
 564 #endif
 565
 566         memset(&dkey, 0, sizeof(DBT));
 567         memset(&ddata, 0, sizeof(DBT));
 568         dkey.size = ckeylen;
 569         dkey.data = ckey;
 570         ddata.size = cdatalen;
 571         ddata.data = cdata;
 572
 573 #ifdef HAVE_ZLIB
 574         /* Only compress Visit records.  Everything else is uncompressed. */
 575         if (cdb == CDB_VISIT) {
 576                 compressing = 1;
 577                 zheader.magic = COMPRESS_MAGIC;
 578                 zheader.uncompressed_len = cdatalen;
 579                 buffer_len = ((cdatalen * 101) / 100) + 100
 580                     + sizeof(struct CtdlCompressHeader);
 581                 destLen = (uLongf) buffer_len;
 582                 compressed_data = malloc(buffer_len);
 583                 if (compress2((Bytef *) (compressed_data +
 584                                          sizeof(struct
 585                                                 CtdlCompressHeader)),
 586                               &destLen, (Bytef *) cdata, (uLongf) cdatalen,
 587                               1) != Z_OK) {
 588                         lprintf(CTDL_EMERG, "compress2() error\n");
 589                         abort();
 590                 }
 591                 zheader.compressed_len = (size_t) destLen;
 592                 memcpy(compressed_data, &zheader,
 593                        sizeof(struct CtdlCompressHeader));
 594                 ddata.size = (size_t) (sizeof(struct CtdlCompressHeader) +
 595                                        zheader.compressed_len);
 596                 ddata.data = compressed_data;
 597         }
 598 #endif
 599
 600         if (MYTID != NULL) {
 601                 ret = dbp[cdb]->put(dbp[cdb],   /* db */
 602                                     MYTID,      /* transaction ID */
 603                                     &dkey,      /* key */
 604                                     &ddata,     /* data */
 605                                     0); /* flags */
 606                 if (ret) {
 607                         lprintf(CTDL_EMERG, "cdb_store(%d): %s\n", cdb,
 608                                 db_strerror(ret));
 609                         abort();
 610                 }
 611 #ifdef HAVE_ZLIB
 612                 if (compressing)
 613                         free(compressed_data);
 614 #endif
 615                 return ret;
 616
 617         } else {
 618                 bailIfCursor(MYCURSORS,
 619                              "attempt to write during r/o cursor");
 620
 621               retry:
 622                 txbegin(&tid);
 623
 624                 if ((ret = dbp[cdb]->put(dbp[cdb],      /* db */
 625                                          tid,   /* transaction ID */
 626                                          &dkey, /* key */
 627                                          &ddata,        /* data */
 628                                          0))) { /* flags */
 629                         if (ret == DB_LOCK_DEADLOCK) {
 630                                 txabort(tid);
 631                                 goto retry;
 632                         } else {
 633                                 lprintf(CTDL_EMERG, "cdb_store(%d): %s\n",
 634                                         cdb, db_strerror(ret));
 635                                 abort();
 636                         }
 637                 } else {
 638                         txcommit(tid);
 639 #ifdef HAVE_ZLIB
 640                         if (compressing)
 641                                 free(compressed_data);
 642 #endif
 643                         return ret;
 644                 }
 645         }
 646 }
 647
 648
 649 /*
 650  * Delete a piece of data.  Returns 0 if the operation was successful.
 651  */
 652 int cdb_delete(int cdb, void *key, int keylen)
 653 {
 654
 655         DBT dkey;
 656         DB_TXN *tid;
 657         int ret;
 658
 659         memset(&dkey, 0, sizeof dkey);
 660         dkey.size = keylen;
 661         dkey.data = key;
 662
 663         if (MYTID != NULL) {
 664                 ret = dbp[cdb]->del(dbp[cdb], MYTID, &dkey, 0);
 665                 if (ret) {
 666                         lprintf(CTDL_EMERG, "cdb_delete(%d): %s\n", cdb,
 667                                 db_strerror(ret));
 668                         if (ret != DB_NOTFOUND)
 669                                 abort();
 670                 }
 671         } else {
 672                 bailIfCursor(MYCURSORS,
 673                              "attempt to delete during r/o cursor");
 674
 675               retry:
 676                 txbegin(&tid);
 677
 678                 if ((ret = dbp[cdb]->del(dbp[cdb], tid, &dkey, 0))
 679                     && ret != DB_NOTFOUND) {
 680                         if (ret == DB_LOCK_DEADLOCK) {
 681                                 txabort(tid);
 682                                 goto retry;
 683                         } else {
 684                                 lprintf(CTDL_EMERG, "cdb_delete(%d): %s\n",
 685                                         cdb, db_strerror(ret));
 686                                 abort();
 687                         }
 688                 } else {
 689                         txcommit(tid);
 690                 }
 691         }
 692         return ret;
 693 }
 694
 695 static DBC *localcursor(int cdb)
 696 {
 697         int ret;
 698         DBC *curs;
 699
 700         if (MYCURSORS[cdb] == NULL)
 701                 ret = dbp[cdb]->cursor(dbp[cdb], MYTID, &curs, 0);
 702         else
 703                 ret =
 704                     MYCURSORS[cdb]->c_dup(MYCURSORS[cdb], &curs,
 705                                           DB_POSITION);
 706
 707         if (ret) {
 708                 lprintf(CTDL_EMERG, "localcursor: %s\n", db_strerror(ret));
 709                 abort();
 710         }
 711
 712         return curs;
 713 }
 714
 715
 716 /*
 717  * Fetch a piece of data.  If not found, returns NULL.  Otherwise, it returns
 718  * a struct cdbdata which it is the caller's responsibility to free later on
 719  * using the cdb_free() routine.
 720  */
 721 struct cdbdata *cdb_fetch(int cdb, void *key, int keylen)
 722 {
 723
 724         struct cdbdata *tempcdb;
 725         DBT dkey, dret;
 726         int ret;
 727
 728         memset(&dkey, 0, sizeof(DBT));
 729         dkey.size = keylen;
 730         dkey.data = key;
 731
 732         if (MYTID != NULL) {
 733                 memset(&dret, 0, sizeof(DBT));
 734                 dret.flags = DB_DBT_MALLOC;
 735                 ret = dbp[cdb]->get(dbp[cdb], MYTID, &dkey, &dret, 0);
 736         } else {
 737                 DBC *curs;
 738
 739                 do {
 740                         memset(&dret, 0, sizeof(DBT));
 741                         dret.flags = DB_DBT_MALLOC;
 742
 743                         curs = localcursor(cdb);
 744
 745                         ret = curs->c_get(curs, &dkey, &dret, DB_SET);
 746                         cclose(curs);
 747                 }
 748                 while (ret == DB_LOCK_DEADLOCK);
 749
 750         }
 751
 752         if ((ret != 0) && (ret != DB_NOTFOUND)) {
 753                 lprintf(CTDL_EMERG, "cdb_fetch(%d): %s\n", cdb,
 754                         db_strerror(ret));
 755                 abort();
 756         }
 757
 758         if (ret != 0)
 759                 return NULL;
 760         tempcdb = (struct cdbdata *) malloc(sizeof(struct cdbdata));
 761
 762         if (tempcdb == NULL) {
 763                 lprintf(CTDL_EMERG,
 764                         "cdb_fetch: Cannot allocate memory for tempcdb\n");
 765                 abort();
 766         }
 767
 768         tempcdb->len = dret.size;
 769         tempcdb->ptr = dret.data;
 770 #ifdef HAVE_ZLIB
 771         cdb_decompress_if_necessary(tempcdb);
 772 #endif
 773         return (tempcdb);
 774 }
 775
 776
 777 /*
 778  * Free a cdbdata item (ok, this is really no big deal, but we might need to do
 779  * more complex stuff with other database managers in the future).
 780  */
 781 void cdb_free(struct cdbdata *cdb)
 782 {
 783         free(cdb->ptr);
 784         free(cdb);
 785 }
 786
 787 void cdb_close_cursor(int cdb)
 788 {
 789         if (MYCURSORS[cdb] != NULL)
 790                 cclose(MYCURSORS[cdb]);
 791
 792         MYCURSORS[cdb] = NULL;
 793 }
 794
 795 /*
 796  * Prepare for a sequential search of an entire database.
 797  * (There is guaranteed to be no more than one traversal in
 798  * progress per thread at any given time.)
 799  */
 800 void cdb_rewind(int cdb)
 801 {
 802         int ret = 0;
 803
 804         if (MYCURSORS[cdb] != NULL) {
 805                 lprintf(CTDL_EMERG,
 806                         "cdb_rewind: must close cursor on database %d before reopening.\n",
 807                         cdb);
 808                 abort();
 809                 /* cclose(MYCURSORS[cdb]); */
 810         }
 811
 812         /*
 813          * Now initialize the cursor
 814          */
 815         ret = dbp[cdb]->cursor(dbp[cdb], MYTID, &MYCURSORS[cdb], 0);
 816         if (ret) {
 817                 lprintf(CTDL_EMERG, "cdb_rewind: db_cursor: %s\n",
 818                         db_strerror(ret));
 819                 abort();
 820         }
 821 }
 822
 823
 824 /*
 825  * Fetch the next item in a sequential search.  Returns a pointer to a
 826  * cdbdata structure, or NULL if we've hit the end.
 827  */
 828 struct cdbdata *cdb_next_item(int cdb)
 829 {
 830         DBT key, data;
 831         struct cdbdata *cdbret;
 832         int ret = 0;
 833
 834         /* Initialize the key/data pair so the flags aren't set. */
 835         memset(&key, 0, sizeof(key));
 836         memset(&data, 0, sizeof(data));
 837         data.flags = DB_DBT_MALLOC;
 838
 839         ret = MYCURSORS[cdb]->c_get(MYCURSORS[cdb], &key, &data, DB_NEXT);
 840
 841         if (ret) {
 842                 if (ret != DB_NOTFOUND) {
 843                         lprintf(CTDL_EMERG, "cdb_next_item(%d): %s\n",
 844                                 cdb, db_strerror(ret));
 845                         abort();
 846                 }
 847                 cclose(MYCURSORS[cdb]);
 848                 MYCURSORS[cdb] = NULL;
 849                 return NULL;    /* presumably, end of file */
 850         }
 851
 852         cdbret = (struct cdbdata *) malloc(sizeof(struct cdbdata));
 853         cdbret->len = data.size;
 854         cdbret->ptr = data.data;
 855 #ifdef HAVE_ZLIB
 856         cdb_decompress_if_necessary(cdbret);
 857 #endif
 858
 859         return (cdbret);
 860 }
 861
 862
 863
 864 /*
 865  * Transaction-based stuff.  I'm writing this as I bake cookies...
 866  */
 867
 868 void cdb_begin_transaction(void)
 869 {
 870
 871         bailIfCursor(MYCURSORS,
 872                      "can't begin transaction during r/o cursor");
 873
 874         if (MYTID != NULL) {
 875                 lprintf(CTDL_EMERG,
 876                         "cdb_begin_transaction: ERROR: nested transaction\n");
 877                 abort();
 878         }
 879
 880         txbegin(&MYTID);
 881 }
 882
 883 void cdb_end_transaction(void)
 884 {
 885         int i;
 886
 887         for (i = 0; i < MAXCDB; i++)
 888                 if (MYCURSORS[i] != NULL) {
 889                         lprintf(CTDL_WARNING,
 890                                 "cdb_end_transaction: WARNING: cursor %d still open at transaction end\n",
 891                                 i);
 892                         cclose(MYCURSORS[i]);
 893                         MYCURSORS[i] = NULL;
 894                 }
 895
 896         if (MYTID == NULL) {
 897                 lprintf(CTDL_EMERG,
 898                         "cdb_end_transaction: ERROR: txcommit(NULL) !!\n");
 899                 abort();
 900         } else
 901                 txcommit(MYTID);
 902
 903         MYTID = NULL;
 904 }
 905
 906 /*
 907  * Truncate (delete every record)
 908  */
 909 void cdb_trunc(int cdb)
 910 {
 911         /* DB_TXN *tid; */
 912         int ret;
 913         u_int32_t count;
 914
 915         if (MYTID != NULL) {
 916                 lprintf(CTDL_EMERG,
 917                         "cdb_trunc must not be called in a transaction.\n");
 918                 abort();
 919         } else {
 920                 bailIfCursor(MYCURSORS,
 921                              "attempt to write during r/o cursor");
 922
 923               retry:
 924                 /* txbegin(&tid); */
 925
 926                 if ((ret = dbp[cdb]->truncate(dbp[cdb], /* db */
 927                                               NULL,     /* transaction ID */
 928                                               &count,   /* #rows deleted */
 929                                               0))) {    /* flags */
 930                         if (ret == DB_LOCK_DEADLOCK) {
 931                                 /* txabort(tid); */
 932                                 goto retry;
 933                         } else {
 934                                 lprintf(CTDL_EMERG,
 935                                         "cdb_truncate(%d): %s\n", cdb,
 936                                         db_strerror(ret));
 937                                 abort();
 938                         }
 939                 } else {
 940                         /* txcommit(tid); */
 941                 }
 942         }
 943 }