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 /*
 260  * Request a checkpoint of the database.
 261  */
 262 static void cdb_checkpoint(void)
 263 {
 264         int ret;
 265         static time_t last_run = 0L;
 266
 267         /* Only do a checkpoint once per minute. */
 268         if ((time(NULL) - last_run) < 60L) {
 269                 return;
 270         }
 271         last_run = time(NULL);
 272
 273         lprintf(CTDL_DEBUG, "-- db checkpoint --\n");
 274         ret = dbenv->txn_checkpoint(dbenv,
 275                                     MAX_CHECKPOINT_KBYTES,
 276                                     MAX_CHECKPOINT_MINUTES, 0);
 277
 278         if (ret != 0) {
 279                 lprintf(CTDL_EMERG, "cdb_checkpoint: txn_checkpoint: %s\n",
 280                         db_strerror(ret));
 281                 abort();
 282         }
 283
 284         /* After a successful checkpoint, we can cull the unused logs */
 285         cdb_cull_logs();
 286 }
 287
 288
 289 /*
 290  * Main loop for the checkpoint thread.
 291  */
 292 void *checkpoint_thread(void *arg) {
 293         struct CitContext checkpointCC;
 294
 295         lprintf(CTDL_DEBUG, "checkpoint_thread() initializing\n");
 296
 297         memset(&checkpointCC, 0, sizeof(struct CitContext));
 298         checkpointCC.internal_pgm = 1;
 299         checkpointCC.cs_pid = 0;
 300         pthread_setspecific(MyConKey, (void *)&checkpointCC );
 301
 302         cdb_allocate_tsd();
 303
 304         while (!time_to_die) {
 305                 cdb_checkpoint();
 306                 sleep(1);
 307         }
 308
 309         lprintf(CTDL_DEBUG, "checkpoint_thread() exiting\n");
 310         pthread_exit(NULL);
 311 }
 312
 313 /*
 314  * Open the various databases we'll be using.  Any database which
 315  * does not exist should be created.  Note that we don't need a
 316  * critical section here, because there aren't any active threads
 317  * manipulating the database yet.
 318  */
 319 void open_databases(void)
 320 {
 321         int ret;
 322         int i;
 323         char dbfilename[SIZ];
 324         u_int32_t flags = 0;
 325         char dbdirname[PATH_MAX];
 326         DIR *dp;
 327         struct dirent *d;
 328         char filename[PATH_MAX];
 329
 330
 331         getcwd(dbdirname, sizeof dbdirname);
 332         strcat(dbdirname, "/data");
 333
 334         lprintf(CTDL_DEBUG, "cdb_*: open_databases() starting\n");
 335         lprintf(CTDL_DEBUG, "Compiled db: %s\n", DB_VERSION_STRING);
 336         lprintf(CTDL_INFO, "  Linked db: %s\n",
 337                 db_version(NULL, NULL, NULL));
 338 #ifdef HAVE_ZLIB
 339         lprintf(CTDL_INFO, "Linked zlib: %s\n", zlibVersion());
 340 #endif
 341
 342         /*
 343          * Silently try to create the database subdirectory.  If it's
 344          * already there, no problem.
 345          */
 346         mkdir(dbdirname, 0700);
 347         chmod(dbdirname, 0700);
 348         chown(dbdirname, CTDLUID, (-1));
 349
 350         lprintf(CTDL_DEBUG, "cdb_*: Setting up DB environment\n");
 351         db_env_set_func_yield(sched_yield);
 352         ret = db_env_create(&dbenv, 0);
 353         if (ret) {
 354                 lprintf(CTDL_EMERG, "cdb_*: db_env_create: %s\n",
 355                         db_strerror(ret));
 356                 exit(ret);
 357         }
 358         dbenv->set_errpfx(dbenv, "citserver");
 359         dbenv->set_paniccall(dbenv, dbpanic);
 360
 361         /*
 362          * We want to specify the shared memory buffer pool cachesize,
 363          * but everything else is the default.
 364          */
 365         ret = dbenv->set_cachesize(dbenv, 0, 64 * 1024, 0);
 366         if (ret) {
 367                 lprintf(CTDL_EMERG, "cdb_*: set_cachesize: %s\n",
 368                         db_strerror(ret));
 369                 dbenv->close(dbenv, 0);
 370                 exit(ret);
 371         }
 372
 373         if ((ret = dbenv->set_lk_detect(dbenv, DB_LOCK_DEFAULT))) {
 374                 lprintf(CTDL_EMERG, "cdb_*: set_lk_detect: %s\n",
 375                         db_strerror(ret));
 376                 dbenv->close(dbenv, 0);
 377                 exit(ret);
 378         }
 379
 380         flags =
 381             DB_CREATE | DB_RECOVER | DB_INIT_MPOOL | DB_PRIVATE |
 382             DB_INIT_TXN | DB_INIT_LOCK | DB_THREAD;
 383         lprintf(CTDL_DEBUG, "dbenv->open(dbenv, %s, %d, 0)\n", dbdirname,
 384                 flags);
 385         ret = dbenv->open(dbenv, dbdirname, flags, 0);
 386         if (ret) {
 387                 lprintf(CTDL_DEBUG, "cdb_*: dbenv->open: %s\n",
 388                         db_strerror(ret));
 389                 dbenv->close(dbenv, 0);
 390                 exit(ret);
 391         }
 392
 393         lprintf(CTDL_INFO, "cdb_*: Starting up DB\n");
 394
 395         for (i = 0; i < MAXCDB; ++i) {
 396
 397                 /* Create a database handle */
 398                 ret = db_create(&dbp[i], dbenv, 0);
 399                 if (ret) {
 400                         lprintf(CTDL_DEBUG, "cdb_*: db_create: %s\n",
 401                                 db_strerror(ret));
 402                         exit(ret);
 403                 }
 404
 405
 406                 /* Arbitrary names for our tables -- we reference them by
 407                  * number, so we don't have string names for them.
 408                  */
 409                 snprintf(dbfilename, sizeof dbfilename, "cdb.%02x", i);
 410
 411                 ret = dbp[i]->open(dbp[i],
 412                                    NULL,
 413                                    dbfilename,
 414                                    NULL,
 415                                    DB_BTREE,
 416                                    DB_CREATE | DB_AUTO_COMMIT | DB_THREAD,
 417                                    0600);
 418                 if (ret) {
 419                         lprintf(CTDL_EMERG, "cdb_*: db_open[%d]: %s\n", i,
 420                                 db_strerror(ret));
 421                         exit(ret);
 422                 }
 423         }
 424
 425         if ((ret = pthread_key_create(&tsdkey, dest_tsd))) {
 426                 lprintf(CTDL_EMERG, "cdb_*: pthread_key_create: %s\n",
 427                         strerror(ret));
 428                 exit(1);
 429         }
 430
 431         cdb_allocate_tsd();
 432
 433         /* Now make sure we own all the files, because in a few milliseconds
 434          * we're going to drop root privs.
 435          */
 436         dp = opendir(dbdirname);
 437         if (dp != NULL) {
 438                 while (d = readdir(dp), d != NULL) {
 439                         if (d->d_name[0] != '.') {
 440                                 snprintf(filename, sizeof filename,
 441                                          "%s/%s", dbdirname, d->d_name);
 442                                 chmod(filename, 0600);
 443                                 chown(filename, CTDLUID, (-1));
 444                         }
 445                 }
 446                 closedir(dp);
 447         }
 448
 449         lprintf(CTDL_DEBUG, "cdb_*: open_databases() finished\n");
 450 }
 451
 452
 453 /*
 454  * Close all of the db database files we've opened.  This can be done
 455  * in a loop, since it's just a bunch of closes.
 456  */
 457 void close_databases(void)
 458 {
 459         int a;
 460         int ret;
 461
 462         cdb_free_tsd();
 463
 464         if ((ret = dbenv->txn_checkpoint(dbenv, 0, 0, 0))) {
 465                 lprintf(CTDL_EMERG,
 466                         "cdb_*: txn_checkpoint: %s\n", db_strerror(ret));
 467         }
 468
 469         for (a = 0; a < MAXCDB; ++a) {
 470                 lprintf(CTDL_INFO, "cdb_*: Closing database %d\n", a);
 471                 ret = dbp[a]->close(dbp[a], 0);
 472                 if (ret) {
 473                         lprintf(CTDL_EMERG,
 474                                 "cdb_*: db_close: %s\n", db_strerror(ret));
 475                 }
 476
 477         }
 478
 479         /* Close the handle. */
 480         ret = dbenv->close(dbenv, 0);
 481         if (ret) {
 482                 lprintf(CTDL_EMERG,
 483                         "cdb_*: DBENV->close: %s\n", db_strerror(ret));
 484         }
 485 }
 486
 487
 488 /*
 489  * Compression functions only used if we have zlib
 490  */
 491 #ifdef HAVE_ZLIB
 492
 493 void cdb_decompress_if_necessary(struct cdbdata *cdb)
 494 {
 495         static int magic = COMPRESS_MAGIC;
 496         struct CtdlCompressHeader zheader;
 497         char *uncompressed_data;
 498         char *compressed_data;
 499         uLongf destLen, sourceLen;
 500
 501         if (cdb == NULL)
 502                 return;
 503         if (cdb->ptr == NULL)
 504                 return;
 505         if (memcmp(cdb->ptr, &magic, sizeof(magic)))
 506                 return;
 507
 508         /* At this point we know we're looking at a compressed item. */
 509         memcpy(&zheader, cdb->ptr, sizeof(struct CtdlCompressHeader));
 510
 511         compressed_data = cdb->ptr;
 512         compressed_data += sizeof(struct CtdlCompressHeader);
 513
 514         sourceLen = (uLongf) zheader.compressed_len;
 515         destLen = (uLongf) zheader.uncompressed_len;
 516         uncompressed_data = malloc(zheader.uncompressed_len);
 517
 518         if (uncompress((Bytef *) uncompressed_data,
 519                        (uLongf *) & destLen,
 520                        (const Bytef *) compressed_data,
 521                        (uLong) sourceLen) != Z_OK) {
 522                 lprintf(CTDL_EMERG, "uncompress() error\n");
 523                 abort();
 524         }
 525
 526         free(cdb->ptr);
 527         cdb->len = (size_t) destLen;
 528         cdb->ptr = uncompressed_data;
 529 }
 530
 531 #endif                          /* HAVE_ZLIB */
 532
 533
 534 /*
 535  * Store a piece of data.  Returns 0 if the operation was successful.  If a
 536  * key already exists it should be overwritten.
 537  */
 538 int cdb_store(int cdb, void *ckey, int ckeylen, void *cdata, int cdatalen)
 539 {
 540
 541         DBT dkey, ddata;
 542         DB_TXN *tid;
 543         int ret;
 544
 545 #ifdef HAVE_ZLIB
 546         struct CtdlCompressHeader zheader;
 547         char *compressed_data = NULL;
 548         int compressing = 0;
 549         size_t buffer_len;
 550         uLongf destLen;
 551 #endif
 552
 553         memset(&dkey, 0, sizeof(DBT));
 554         memset(&ddata, 0, sizeof(DBT));
 555         dkey.size = ckeylen;
 556         dkey.data = ckey;
 557         ddata.size = cdatalen;
 558         ddata.data = cdata;
 559
 560 #ifdef HAVE_ZLIB
 561         /* Only compress Visit records.  Everything else is uncompressed. */
 562         if (cdb == CDB_VISIT) {
 563                 compressing = 1;
 564                 zheader.magic = COMPRESS_MAGIC;
 565                 zheader.uncompressed_len = cdatalen;
 566                 buffer_len = ((cdatalen * 101) / 100) + 100
 567                     + sizeof(struct CtdlCompressHeader);
 568                 destLen = (uLongf) buffer_len;
 569                 compressed_data = malloc(buffer_len);
 570                 if (compress2((Bytef *) (compressed_data +
 571                                          sizeof(struct
 572                                                 CtdlCompressHeader)),
 573                               &destLen, (Bytef *) cdata, (uLongf) cdatalen,
 574                               1) != Z_OK) {
 575                         lprintf(CTDL_EMERG, "compress2() error\n");
 576                         abort();
 577                 }
 578                 zheader.compressed_len = (size_t) destLen;
 579                 memcpy(compressed_data, &zheader,
 580                        sizeof(struct CtdlCompressHeader));
 581                 ddata.size = (size_t) (sizeof(struct CtdlCompressHeader) +
 582                                        zheader.compressed_len);
 583                 ddata.data = compressed_data;
 584         }
 585 #endif
 586
 587         if (MYTID != NULL) {
 588                 ret = dbp[cdb]->put(dbp[cdb],   /* db */
 589                                     MYTID,      /* transaction ID */
 590                                     &dkey,      /* key */
 591                                     &ddata,     /* data */
 592                                     0); /* flags */
 593                 if (ret) {
 594                         lprintf(CTDL_EMERG, "cdb_store(%d): %s\n", cdb,
 595                                 db_strerror(ret));
 596                         abort();
 597                 }
 598 #ifdef HAVE_ZLIB
 599                 if (compressing)
 600                         free(compressed_data);
 601 #endif
 602                 return ret;
 603
 604         } else {
 605                 bailIfCursor(MYCURSORS,
 606                              "attempt to write during r/o cursor");
 607
 608               retry:
 609                 txbegin(&tid);
 610
 611                 if ((ret = dbp[cdb]->put(dbp[cdb],      /* db */
 612                                          tid,   /* transaction ID */
 613                                          &dkey, /* key */
 614                                          &ddata,        /* data */
 615                                          0))) { /* flags */
 616                         if (ret == DB_LOCK_DEADLOCK) {
 617                                 txabort(tid);
 618                                 goto retry;
 619                         } else {
 620                                 lprintf(CTDL_EMERG, "cdb_store(%d): %s\n",
 621                                         cdb, db_strerror(ret));
 622                                 abort();
 623                         }
 624                 } else {
 625                         txcommit(tid);
 626 #ifdef HAVE_ZLIB
 627                         if (compressing)
 628                                 free(compressed_data);
 629 #endif
 630                         return ret;
 631                 }
 632         }
 633 }
 634
 635
 636 /*
 637  * Delete a piece of data.  Returns 0 if the operation was successful.
 638  */
 639 int cdb_delete(int cdb, void *key, int keylen)
 640 {
 641
 642         DBT dkey;
 643         DB_TXN *tid;
 644         int ret;
 645
 646         memset(&dkey, 0, sizeof dkey);
 647         dkey.size = keylen;
 648         dkey.data = key;
 649
 650         if (MYTID != NULL) {
 651                 ret = dbp[cdb]->del(dbp[cdb], MYTID, &dkey, 0);
 652                 if (ret) {
 653                         lprintf(CTDL_EMERG, "cdb_delete(%d): %s\n", cdb,
 654                                 db_strerror(ret));
 655                         if (ret != DB_NOTFOUND)
 656                                 abort();
 657                 }
 658         } else {
 659                 bailIfCursor(MYCURSORS,
 660                              "attempt to delete during r/o cursor");
 661
 662               retry:
 663                 txbegin(&tid);
 664
 665                 if ((ret = dbp[cdb]->del(dbp[cdb], tid, &dkey, 0))
 666                     && ret != DB_NOTFOUND) {
 667                         if (ret == DB_LOCK_DEADLOCK) {
 668                                 txabort(tid);
 669                                 goto retry;
 670                         } else {
 671                                 lprintf(CTDL_EMERG, "cdb_delete(%d): %s\n",
 672                                         cdb, db_strerror(ret));
 673                                 abort();
 674                         }
 675                 } else {
 676                         txcommit(tid);
 677                 }
 678         }
 679         return ret;
 680 }
 681
 682 static DBC *localcursor(int cdb)
 683 {
 684         int ret;
 685         DBC *curs;
 686
 687         if (MYCURSORS[cdb] == NULL)
 688                 ret = dbp[cdb]->cursor(dbp[cdb], MYTID, &curs, 0);
 689         else
 690                 ret =
 691                     MYCURSORS[cdb]->c_dup(MYCURSORS[cdb], &curs,
 692                                           DB_POSITION);
 693
 694         if (ret) {
 695                 lprintf(CTDL_EMERG, "localcursor: %s\n", db_strerror(ret));
 696                 abort();
 697         }
 698
 699         return curs;
 700 }
 701
 702
 703 /*
 704  * Fetch a piece of data.  If not found, returns NULL.  Otherwise, it returns
 705  * a struct cdbdata which it is the caller's responsibility to free later on
 706  * using the cdb_free() routine.
 707  */
 708 struct cdbdata *cdb_fetch(int cdb, void *key, int keylen)
 709 {
 710
 711         struct cdbdata *tempcdb;
 712         DBT dkey, dret;
 713         int ret;
 714
 715         memset(&dkey, 0, sizeof(DBT));
 716         dkey.size = keylen;
 717         dkey.data = key;
 718
 719         if (MYTID != NULL) {
 720                 memset(&dret, 0, sizeof(DBT));
 721                 dret.flags = DB_DBT_MALLOC;
 722                 ret = dbp[cdb]->get(dbp[cdb], MYTID, &dkey, &dret, 0);
 723         } else {
 724                 DBC *curs;
 725
 726                 do {
 727                         memset(&dret, 0, sizeof(DBT));
 728                         dret.flags = DB_DBT_MALLOC;
 729
 730                         curs = localcursor(cdb);
 731
 732                         ret = curs->c_get(curs, &dkey, &dret, DB_SET);
 733                         cclose(curs);
 734                 }
 735                 while (ret == DB_LOCK_DEADLOCK);
 736
 737         }
 738
 739         if ((ret != 0) && (ret != DB_NOTFOUND)) {
 740                 lprintf(CTDL_EMERG, "cdb_fetch(%d): %s\n", cdb,
 741                         db_strerror(ret));
 742                 abort();
 743         }
 744
 745         if (ret != 0)
 746                 return NULL;
 747         tempcdb = (struct cdbdata *) malloc(sizeof(struct cdbdata));
 748
 749         if (tempcdb == NULL) {
 750                 lprintf(CTDL_EMERG,
 751                         "cdb_fetch: Cannot allocate memory for tempcdb\n");
 752                 abort();
 753         }
 754
 755         tempcdb->len = dret.size;
 756         tempcdb->ptr = dret.data;
 757 #ifdef HAVE_ZLIB
 758         cdb_decompress_if_necessary(tempcdb);
 759 #endif
 760         return (tempcdb);
 761 }
 762
 763
 764 /*
 765  * Free a cdbdata item (ok, this is really no big deal, but we might need to do
 766  * more complex stuff with other database managers in the future).
 767  */
 768 void cdb_free(struct cdbdata *cdb)
 769 {
 770         free(cdb->ptr);
 771         free(cdb);
 772 }
 773
 774 void cdb_close_cursor(int cdb)
 775 {
 776         if (MYCURSORS[cdb] != NULL)
 777                 cclose(MYCURSORS[cdb]);
 778
 779         MYCURSORS[cdb] = NULL;
 780 }
 781
 782 /*
 783  * Prepare for a sequential search of an entire database.
 784  * (There is guaranteed to be no more than one traversal in
 785  * progress per thread at any given time.)
 786  */
 787 void cdb_rewind(int cdb)
 788 {
 789         int ret = 0;
 790
 791         if (MYCURSORS[cdb] != NULL) {
 792                 lprintf(CTDL_EMERG,
 793                         "cdb_rewind: must close cursor on database %d before reopening.\n",
 794                         cdb);
 795                 abort();
 796                 /* cclose(MYCURSORS[cdb]); */
 797         }
 798
 799         /*
 800          * Now initialize the cursor
 801          */
 802         ret = dbp[cdb]->cursor(dbp[cdb], MYTID, &MYCURSORS[cdb], 0);
 803         if (ret) {
 804                 lprintf(CTDL_EMERG, "cdb_rewind: db_cursor: %s\n",
 805                         db_strerror(ret));
 806                 abort();
 807         }
 808 }
 809
 810
 811 /*
 812  * Fetch the next item in a sequential search.  Returns a pointer to a
 813  * cdbdata structure, or NULL if we've hit the end.
 814  */
 815 struct cdbdata *cdb_next_item(int cdb)
 816 {
 817         DBT key, data;
 818         struct cdbdata *cdbret;
 819         int ret = 0;
 820
 821         /* Initialize the key/data pair so the flags aren't set. */
 822         memset(&key, 0, sizeof(key));
 823         memset(&data, 0, sizeof(data));
 824         data.flags = DB_DBT_MALLOC;
 825
 826         ret = MYCURSORS[cdb]->c_get(MYCURSORS[cdb], &key, &data, DB_NEXT);
 827
 828         if (ret) {
 829                 if (ret != DB_NOTFOUND) {
 830                         lprintf(CTDL_EMERG, "cdb_next_item(%d): %s\n",
 831                                 cdb, db_strerror(ret));
 832                         abort();
 833                 }
 834                 cclose(MYCURSORS[cdb]);
 835                 MYCURSORS[cdb] = NULL;
 836                 return NULL;    /* presumably, end of file */
 837         }
 838
 839         cdbret = (struct cdbdata *) malloc(sizeof(struct cdbdata));
 840         cdbret->len = data.size;
 841         cdbret->ptr = data.data;
 842 #ifdef HAVE_ZLIB
 843         cdb_decompress_if_necessary(cdbret);
 844 #endif
 845
 846         return (cdbret);
 847 }
 848
 849
 850
 851 /*
 852  * Transaction-based stuff.  I'm writing this as I bake cookies...
 853  */
 854
 855 void cdb_begin_transaction(void)
 856 {
 857
 858         bailIfCursor(MYCURSORS,
 859                      "can't begin transaction during r/o cursor");
 860
 861         if (MYTID != NULL) {
 862                 lprintf(CTDL_EMERG,
 863                         "cdb_begin_transaction: ERROR: nested transaction\n");
 864                 abort();
 865         }
 866
 867         txbegin(&MYTID);
 868 }
 869
 870 void cdb_end_transaction(void)
 871 {
 872         int i;
 873
 874         for (i = 0; i < MAXCDB; i++)
 875                 if (MYCURSORS[i] != NULL) {
 876                         lprintf(CTDL_WARNING,
 877                                 "cdb_end_transaction: WARNING: cursor %d still open at transaction end\n",
 878                                 i);
 879                         cclose(MYCURSORS[i]);
 880                         MYCURSORS[i] = NULL;
 881                 }
 882
 883         if (MYTID == NULL) {
 884                 lprintf(CTDL_EMERG,
 885                         "cdb_end_transaction: ERROR: txcommit(NULL) !!\n");
 886                 abort();
 887         } else
 888                 txcommit(MYTID);
 889
 890         MYTID = NULL;
 891 }
 892
 893 /*
 894  * Truncate (delete every record)
 895  */
 896 void cdb_trunc(int cdb)
 897 {
 898         /* DB_TXN *tid; */
 899         int ret;
 900         u_int32_t count;
 901
 902         if (MYTID != NULL) {
 903                 lprintf(CTDL_EMERG,
 904                         "cdb_trunc must not be called in a transaction.\n");
 905                 abort();
 906         } else {
 907                 bailIfCursor(MYCURSORS,
 908                              "attempt to write during r/o cursor");
 909
 910               retry:
 911                 /* txbegin(&tid); */
 912
 913                 if ((ret = dbp[cdb]->truncate(dbp[cdb], /* db */
 914                                               NULL,     /* transaction ID */
 915                                               &count,   /* #rows deleted */
 916                                               0))) {    /* flags */
 917                         if (ret == DB_LOCK_DEADLOCK) {
 918                                 /* txabort(tid); */
 919                                 goto retry;
 920                         } else {
 921                                 lprintf(CTDL_EMERG,
 922                                         "cdb_truncate(%d): %s\n", cdb,
 923                                         db_strerror(ret));
 924                                 abort();
 925                         }
 926                 } else {
 927                         /* txcommit(tid); */
 928                 }
 929         }
 930 }