2 * This is a data store backend for the Citadel server which uses Berkeley DB.
4 * Copyright (c) 1987-2009 by the citadel.org team
6 * This program is free software; you can redistribute it and/or modify
7 * it under the terms of the GNU General Public License as published by
8 * the Free Software Foundation; either version 3 of the License, or
9 * (at your option) any later version.
11 * This program 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
14 * GNU General Public License for more details.
16 * You should have received a copy of the GNU General Public License
17 * along with this program; if not, write to the Free Software
18 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
22 /*****************************************************************************
23 Tunable configuration parameters for the Berkeley DB back end
24 *****************************************************************************/
26 /* Citadel will checkpoint the db at the end of every session, but only if
27 * the specified number of kilobytes has been written, or if the specified
28 * number of minutes has passed, since the last checkpoint.
30 #define MAX_CHECKPOINT_KBYTES 256
31 #define MAX_CHECKPOINT_MINUTES 15
33 /*****************************************************************************/
42 #include <sys/types.h>
48 #elif defined(HAVE_DB4_DB_H)
51 #error Neither <db.h> nor <db4/db.h> was found by configure. Install db4-devel.
55 #if DB_VERSION_MAJOR < 4 || DB_VERSION_MINOR < 1
56 #error Citadel requires Berkeley DB v4.1 or newer. Please upgrade.
60 #include <libcitadel.h>
63 #include "citserver.h"
66 #include "sysdep_decls.h"
71 #include "ctdl_module.h"
74 static DB *dbp[MAXCDB]; /* One DB handle for each Citadel database */
75 static DB_ENV *dbenv; /* The DB environment (global) */
83 /* Verbose logging callback */
84 void cdb_verbose_log(const DB_ENV *dbenv, const char *msg)
86 if (!IsEmptyStr(msg)) {
87 CtdlLogPrintf(CTDL_DEBUG, "DB: %s\n", msg);
92 /* Verbose logging callback */
93 void cdb_verbose_err(const DB_ENV *dbenv, const char *errpfx, const char *msg)
95 CtdlLogPrintf(CTDL_ALERT, "DB: %s\n", msg);
99 /* just a little helper function */
100 static void txabort(DB_TXN * tid)
104 ret = tid->abort(tid);
107 CtdlLogPrintf(CTDL_EMERG, "bdb(): txn_abort: %s\n", db_strerror(ret));
112 /* this one is even more helpful than the last. */
113 static void txcommit(DB_TXN * tid)
117 ret = tid->commit(tid, 0);
120 CtdlLogPrintf(CTDL_EMERG, "bdb(): txn_commit: %s\n", db_strerror(ret));
125 /* are you sensing a pattern yet? */
126 static void txbegin(DB_TXN ** tid)
130 ret = dbenv->txn_begin(dbenv, NULL, tid, 0);
133 CtdlLogPrintf(CTDL_EMERG, "bdb(): txn_begin: %s\n", db_strerror(ret));
138 static void dbpanic(DB_ENV * env, int errval)
140 CtdlLogPrintf(CTDL_EMERG, "bdb(): PANIC: %s\n", db_strerror(errval));
143 static void cclose(DBC * cursor)
147 if ((ret = cursor->c_close(cursor))) {
148 CtdlLogPrintf(CTDL_EMERG, "bdb(): c_close: %s\n", db_strerror(ret));
153 static void bailIfCursor(DBC ** cursors, const char *msg)
157 for (i = 0; i < MAXCDB; i++)
158 if (cursors[i] != NULL) {
159 CtdlLogPrintf(CTDL_EMERG,
160 "bdb(): cursor still in progress on cdb %02x: %s\n", i, msg);
165 void check_handles(void *arg)
168 ThreadTSD *tsd = (ThreadTSD *) arg;
170 bailIfCursor(tsd->cursors, "in check_handles");
172 if (tsd->tid != NULL) {
173 CtdlLogPrintf(CTDL_EMERG, "bdb(): transaction still in progress!");
179 void cdb_check_handles(void)
181 check_handles(pthread_getspecific(ThreadKey));
186 * Cull the database logs
188 static void cdb_cull_logs(void)
197 /* Get the list of names. */
198 if ((ret = dbenv->log_archive(dbenv, &list, flags)) != 0) {
199 CtdlLogPrintf(CTDL_ERR, "cdb_cull_logs: %s\n", db_strerror(ret));
203 /* Print the list of names. */
205 for (file = list; *file != NULL; ++file) {
206 CtdlLogPrintf(CTDL_DEBUG, "Deleting log: %s\n", *file);
209 snprintf(errmsg, sizeof(errmsg),
210 " ** ERROR **\n \n \n "
211 "Citadel was unable to delete the "
212 "database log file '%s' because of the "
213 "following error:\n \n %s\n \n"
214 " This log file is no longer in use "
215 "and may be safely deleted.\n",
216 *file, strerror(errno));
217 CtdlAideMessage(errmsg, "Database Warning Message");
225 * Manually initiate log file cull.
227 void cmd_cull(char *argbuf) {
228 if (CtdlAccessCheck(ac_internal)) return;
230 cprintf("%d Database log file cull completed.\n", CIT_OK);
235 * Request a checkpoint of the database. Called once per minute by the thread manager.
237 void cdb_checkpoint(void)
241 CtdlLogPrintf(CTDL_DEBUG, "-- db checkpoint --\n");
242 ret = dbenv->txn_checkpoint(dbenv, MAX_CHECKPOINT_KBYTES, MAX_CHECKPOINT_MINUTES, 0);
245 CtdlLogPrintf(CTDL_EMERG, "cdb_checkpoint: txn_checkpoint: %s\n", db_strerror(ret));
249 /* After a successful checkpoint, we can cull the unused logs */
250 if (config.c_auto_cull) {
258 * Open the various databases we'll be using. Any database which
259 * does not exist should be created. Note that we don't need a
260 * critical section here, because there aren't any active threads
261 * manipulating the database yet.
263 void open_databases(void)
269 int dbversion_major, dbversion_minor, dbversion_patch;
270 int current_dbversion = 0;
272 CtdlLogPrintf(CTDL_DEBUG, "bdb(): open_databases() starting\n");
273 CtdlLogPrintf(CTDL_DEBUG, "Compiled db: %s\n", DB_VERSION_STRING);
274 CtdlLogPrintf(CTDL_INFO, " Linked db: %s\n",
275 db_version(&dbversion_major, &dbversion_minor, &dbversion_patch));
277 current_dbversion = (dbversion_major * 1000000) + (dbversion_minor * 1000) + dbversion_patch;
279 CtdlLogPrintf(CTDL_DEBUG, "Calculated dbversion: %d\n", current_dbversion);
280 CtdlLogPrintf(CTDL_DEBUG, " Previous dbversion: %d\n", CitControl.MMdbversion);
282 if ( (getenv("SUPPRESS_DBVERSION_CHECK") == NULL)
283 && (CitControl.MMdbversion > current_dbversion) ) {
284 CtdlLogPrintf(CTDL_EMERG, "You are attempting to run the Citadel server using a version\n"
285 "of Berkeley DB that is older than that which last created or\n"
286 "updated the database. Because this would probably cause data\n"
287 "corruption or loss, the server is aborting execution now.\n");
291 CitControl.MMdbversion = current_dbversion;
295 CtdlLogPrintf(CTDL_INFO, "Linked zlib: %s\n", zlibVersion());
299 * Silently try to create the database subdirectory. If it's
300 * already there, no problem.
302 if ((mkdir(ctdl_data_dir, 0700) != 0) && (errno != EEXIST)){
303 CtdlLogPrintf(CTDL_EMERG,
304 "unable to create database directory [%s]: %s",
305 ctdl_data_dir, strerror(errno));
307 if (chmod(ctdl_data_dir, 0700) != 0){
308 CtdlLogPrintf(CTDL_EMERG,
309 "unable to set database directory accessrights [%s]: %s",
310 ctdl_data_dir, strerror(errno));
312 if (chown(ctdl_data_dir, CTDLUID, (-1)) != 0){
313 CtdlLogPrintf(CTDL_EMERG,
314 "unable to set the owner for [%s]: %s",
315 ctdl_data_dir, strerror(errno));
317 CtdlLogPrintf(CTDL_DEBUG, "bdb(): Setting up DB environment\n");
318 db_env_set_func_yield((int (*)(u_long, u_long))sched_yield);
319 ret = db_env_create(&dbenv, 0);
321 CtdlLogPrintf(CTDL_EMERG, "bdb(): db_env_create: %s\n", db_strerror(ret));
322 CtdlLogPrintf(CTDL_EMERG, "exit code %d\n", ret);
325 dbenv->set_errpfx(dbenv, "citserver");
326 dbenv->set_paniccall(dbenv, dbpanic);
327 dbenv->set_errcall(dbenv, cdb_verbose_err);
328 dbenv->set_errpfx(dbenv, "ctdl");
329 #if (DB_VERSION_MAJOR == 4) && (DB_VERSION_MINOR >= 3)
330 dbenv->set_msgcall(dbenv, cdb_verbose_log);
332 dbenv->set_verbose(dbenv, DB_VERB_DEADLOCK, 1);
333 dbenv->set_verbose(dbenv, DB_VERB_RECOVERY, 1);
336 * We want to specify the shared memory buffer pool cachesize,
337 * but everything else is the default.
339 ret = dbenv->set_cachesize(dbenv, 0, 64 * 1024, 0);
341 CtdlLogPrintf(CTDL_EMERG, "bdb(): set_cachesize: %s\n", db_strerror(ret));
342 dbenv->close(dbenv, 0);
343 CtdlLogPrintf(CTDL_EMERG, "exit code %d\n", ret);
347 if ((ret = dbenv->set_lk_detect(dbenv, DB_LOCK_DEFAULT))) {
348 CtdlLogPrintf(CTDL_EMERG, "bdb(): set_lk_detect: %s\n", db_strerror(ret));
349 dbenv->close(dbenv, 0);
350 CtdlLogPrintf(CTDL_EMERG, "exit code %d\n", ret);
354 flags = DB_CREATE | DB_INIT_MPOOL | DB_PRIVATE | DB_INIT_TXN | DB_INIT_LOCK | DB_THREAD | DB_RECOVER;
355 CtdlLogPrintf(CTDL_DEBUG, "dbenv->open(dbenv, %s, %d, 0)\n", ctdl_data_dir, flags);
356 ret = dbenv->open(dbenv, ctdl_data_dir, flags, 0);
357 if (ret == DB_RUNRECOVERY) {
358 CtdlLogPrintf(CTDL_ALERT, "dbenv->open: %s\n", db_strerror(ret));
359 CtdlLogPrintf(CTDL_ALERT, "Attempting recovery...\n");
361 ret = dbenv->open(dbenv, ctdl_data_dir, flags, 0);
363 if (ret == DB_RUNRECOVERY) {
364 CtdlLogPrintf(CTDL_ALERT, "dbenv->open: %s\n", db_strerror(ret));
365 CtdlLogPrintf(CTDL_ALERT, "Attempting catastrophic recovery...\n");
366 flags &= ~DB_RECOVER;
367 flags |= DB_RECOVER_FATAL;
368 ret = dbenv->open(dbenv, ctdl_data_dir, flags, 0);
371 CtdlLogPrintf(CTDL_EMERG, "dbenv->open: %s\n", db_strerror(ret));
372 dbenv->close(dbenv, 0);
373 CtdlLogPrintf(CTDL_EMERG, "exit code %d\n", ret);
377 CtdlLogPrintf(CTDL_INFO, "Starting up DB\n");
379 for (i = 0; i < MAXCDB; ++i) {
381 /* Create a database handle */
382 ret = db_create(&dbp[i], dbenv, 0);
384 CtdlLogPrintf(CTDL_EMERG, "db_create: %s\n", db_strerror(ret));
385 CtdlLogPrintf(CTDL_EMERG, "exit code %d\n", ret);
390 /* Arbitrary names for our tables -- we reference them by
391 * number, so we don't have string names for them.
393 snprintf(dbfilename, sizeof dbfilename, "cdb.%02x", i);
395 ret = dbp[i]->open(dbp[i],
400 DB_CREATE | DB_AUTO_COMMIT | DB_THREAD,
404 CtdlLogPrintf(CTDL_EMERG, "db_open[%02x]: %s\n", i, db_strerror(ret));
406 CtdlLogPrintf(CTDL_EMERG, "You may need to tune your database; please read http://www.citadel.org/doku.php/faq:troubleshooting:out_of_lock_entries for more information.\n");
408 CtdlLogPrintf(CTDL_EMERG, "exit code %d\n", ret);
416 /* Make sure we own all the files, because in a few milliseconds
417 * we're going to drop root privs.
419 void cdb_chmod_data(void) {
422 char filename[PATH_MAX];
424 dp = opendir(ctdl_data_dir);
426 while (d = readdir(dp), d != NULL) {
427 if (d->d_name[0] != '.') {
428 snprintf(filename, sizeof filename,
429 "%s/%s", ctdl_data_dir, d->d_name);
430 CtdlLogPrintf(9, "chmod(%s, 0600) returned %d\n",
431 filename, chmod(filename, 0600)
433 CtdlLogPrintf(9, "chown(%s, CTDLUID, -1) returned %d\n",
434 filename, chown(filename, CTDLUID, (-1))
441 CtdlLogPrintf(CTDL_DEBUG, "open_databases() finished\n");
442 CtdlRegisterProtoHook(cmd_cull, "CULL", "Cull database logs");
447 * Close all of the db database files we've opened. This can be done
448 * in a loop, since it's just a bunch of closes.
450 void close_databases(void)
455 ctdl_thread_internal_free_tsd();
457 if ((ret = dbenv->txn_checkpoint(dbenv, 0, 0, 0))) {
458 CtdlLogPrintf(CTDL_EMERG,
459 "txn_checkpoint: %s\n", db_strerror(ret));
462 /* print some statistics... */
464 dbenv->lock_stat_print(dbenv, DB_STAT_ALL);
467 /* close the tables */
468 for (a = 0; a < MAXCDB; ++a) {
469 CtdlLogPrintf(CTDL_INFO, "Closing database %02x\n", a);
470 ret = dbp[a]->close(dbp[a], 0);
472 CtdlLogPrintf(CTDL_EMERG, "db_close: %s\n", db_strerror(ret));
477 /* Close the handle. */
478 ret = dbenv->close(dbenv, 0);
480 CtdlLogPrintf(CTDL_EMERG, "DBENV->close: %s\n", db_strerror(ret));
486 * Compression functions only used if we have zlib
488 void cdb_decompress_if_necessary(struct cdbdata *cdb)
490 static int magic = COMPRESS_MAGIC;
493 (cdb->ptr == NULL) ||
494 (cdb->len < sizeof(magic)) ||
495 (memcmp(cdb->ptr, &magic, sizeof(magic))))
499 /* At this point we know we're looking at a compressed item. */
501 struct CtdlCompressHeader zheader;
502 char *uncompressed_data;
503 char *compressed_data;
504 uLongf destLen, sourceLen;
507 memset(&zheader, 0, sizeof(struct CtdlCompressHeader));
508 cplen = sizeof(struct CtdlCompressHeader);
509 if (sizeof(struct CtdlCompressHeader) > cdb->len)
511 memcpy(&zheader, cdb->ptr, cplen);
513 compressed_data = cdb->ptr;
514 compressed_data += sizeof(struct CtdlCompressHeader);
516 sourceLen = (uLongf) zheader.compressed_len;
517 destLen = (uLongf) zheader.uncompressed_len;
518 uncompressed_data = malloc(zheader.uncompressed_len);
520 if (uncompress((Bytef *) uncompressed_data,
521 (uLongf *) & destLen,
522 (const Bytef *) compressed_data,
523 (uLong) sourceLen) != Z_OK) {
524 CtdlLogPrintf(CTDL_EMERG, "uncompress() error\n");
529 cdb->len = (size_t) destLen;
530 cdb->ptr = uncompressed_data;
531 #else /* HAVE_ZLIB */
532 CtdlLogPrintf(CTDL_EMERG, "Database contains compressed data, but this citserver was built without compression support.\n");
534 #endif /* HAVE_ZLIB */
540 * Store a piece of data. Returns 0 if the operation was successful. If a
541 * key already exists it should be overwritten.
543 int cdb_store(int cdb, const void *ckey, int ckeylen, void *cdata, int cdatalen)
551 struct CtdlCompressHeader zheader;
552 char *compressed_data = NULL;
554 size_t buffer_len = 0;
558 memset(&dkey, 0, sizeof(DBT));
559 memset(&ddata, 0, sizeof(DBT));
562 ddata.size = cdatalen;
566 /* Only compress Visit records. Everything else is uncompressed. */
567 if (cdb == CDB_VISIT) {
569 zheader.magic = COMPRESS_MAGIC;
570 zheader.uncompressed_len = cdatalen;
571 buffer_len = ((cdatalen * 101) / 100) + 100
572 + sizeof(struct CtdlCompressHeader);
573 destLen = (uLongf) buffer_len;
574 compressed_data = malloc(buffer_len);
575 if (compress2((Bytef *) (compressed_data + sizeof(struct CtdlCompressHeader)),
576 &destLen, (Bytef *) cdata, (uLongf) cdatalen, 1) != Z_OK)
578 CtdlLogPrintf(CTDL_EMERG, "compress2() error\n");
581 zheader.compressed_len = (size_t) destLen;
582 memcpy(compressed_data, &zheader, sizeof(struct CtdlCompressHeader));
583 ddata.size = (size_t) (sizeof(struct CtdlCompressHeader) + zheader.compressed_len);
584 ddata.data = compressed_data;
589 ret = dbp[cdb]->put(dbp[cdb], /* db */
590 MYTID, /* transaction ID */
595 CtdlLogPrintf(CTDL_EMERG, "cdb_store(%d): %s\n", cdb, db_strerror(ret));
600 free(compressed_data);
605 bailIfCursor(MYCURSORS, "attempt to write during r/o cursor");
610 if ((ret = dbp[cdb]->put(dbp[cdb], /* db */
611 tid, /* transaction ID */
615 if (ret == DB_LOCK_DEADLOCK) {
619 CtdlLogPrintf(CTDL_EMERG, "cdb_store(%d): %s\n",
620 cdb, db_strerror(ret));
627 free(compressed_data);
637 * Delete a piece of data. Returns 0 if the operation was successful.
639 int cdb_delete(int cdb, void *key, int keylen)
646 memset(&dkey, 0, sizeof dkey);
651 ret = dbp[cdb]->del(dbp[cdb], MYTID, &dkey, 0);
653 CtdlLogPrintf(CTDL_EMERG, "cdb_delete(%d): %s\n", cdb, db_strerror(ret));
654 if (ret != DB_NOTFOUND) {
659 bailIfCursor(MYCURSORS, "attempt to delete during r/o cursor");
664 if ((ret = dbp[cdb]->del(dbp[cdb], tid, &dkey, 0))
665 && ret != DB_NOTFOUND) {
666 if (ret == DB_LOCK_DEADLOCK) {
670 CtdlLogPrintf(CTDL_EMERG, "cdb_delete(%d): %s\n",
671 cdb, db_strerror(ret));
681 static DBC *localcursor(int cdb)
686 if (MYCURSORS[cdb] == NULL)
687 ret = dbp[cdb]->cursor(dbp[cdb], MYTID, &curs, 0);
690 MYCURSORS[cdb]->c_dup(MYCURSORS[cdb], &curs,
694 CtdlLogPrintf(CTDL_EMERG, "localcursor: %s\n", db_strerror(ret));
703 * Fetch a piece of data. If not found, returns NULL. Otherwise, it returns
704 * a struct cdbdata which it is the caller's responsibility to free later on
705 * using the cdb_free() routine.
707 struct cdbdata *cdb_fetch(int cdb, const void *key, int keylen)
710 struct cdbdata *tempcdb;
714 memset(&dkey, 0, sizeof(DBT));
719 memset(&dret, 0, sizeof(DBT));
720 dret.flags = DB_DBT_MALLOC;
721 ret = dbp[cdb]->get(dbp[cdb], MYTID, &dkey, &dret, 0);
726 memset(&dret, 0, sizeof(DBT));
727 dret.flags = DB_DBT_MALLOC;
729 curs = localcursor(cdb);
731 ret = curs->c_get(curs, &dkey, &dret, DB_SET);
734 while (ret == DB_LOCK_DEADLOCK);
738 if ((ret != 0) && (ret != DB_NOTFOUND)) {
739 CtdlLogPrintf(CTDL_EMERG, "cdb_fetch(%d): %s\n", cdb, db_strerror(ret));
745 tempcdb = (struct cdbdata *) malloc(sizeof(struct cdbdata));
747 if (tempcdb == NULL) {
748 CtdlLogPrintf(CTDL_EMERG, "cdb_fetch: Cannot allocate memory for tempcdb\n");
752 tempcdb->len = dret.size;
753 tempcdb->ptr = dret.data;
754 cdb_decompress_if_necessary(tempcdb);
760 * Free a cdbdata item.
762 * Note that we only free the 'ptr' portion if it is not NULL. This allows
763 * other code to assume ownership of that memory simply by storing the
764 * pointer elsewhere and then setting 'ptr' to NULL. cdb_free() will then
767 void cdb_free(struct cdbdata *cdb)
775 void cdb_close_cursor(int cdb)
777 if (MYCURSORS[cdb] != NULL) {
778 cclose(MYCURSORS[cdb]);
781 MYCURSORS[cdb] = NULL;
785 * Prepare for a sequential search of an entire database.
786 * (There is guaranteed to be no more than one traversal in
787 * progress per thread at any given time.)
789 void cdb_rewind(int cdb)
793 if (MYCURSORS[cdb] != NULL) {
794 CtdlLogPrintf(CTDL_EMERG,
795 "cdb_rewind: must close cursor on database %d before reopening.\n", cdb);
797 /* cclose(MYCURSORS[cdb]); */
801 * Now initialize the cursor
803 ret = dbp[cdb]->cursor(dbp[cdb], MYTID, &MYCURSORS[cdb], 0);
805 CtdlLogPrintf(CTDL_EMERG, "cdb_rewind: db_cursor: %s\n", db_strerror(ret));
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.
815 struct cdbdata *cdb_next_item(int cdb)
818 struct cdbdata *cdbret;
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;
826 ret = MYCURSORS[cdb]->c_get(MYCURSORS[cdb], &key, &data, DB_NEXT);
829 if (ret != DB_NOTFOUND) {
830 CtdlLogPrintf(CTDL_EMERG, "cdb_next_item(%d): %s\n", cdb, db_strerror(ret));
833 cclose(MYCURSORS[cdb]);
834 MYCURSORS[cdb] = NULL;
835 return NULL; /* presumably, end of file */
838 cdbret = (struct cdbdata *) malloc(sizeof(struct cdbdata));
839 cdbret->len = data.size;
840 cdbret->ptr = data.data;
841 cdb_decompress_if_necessary(cdbret);
849 * Transaction-based stuff. I'm writing this as I bake cookies...
852 void cdb_begin_transaction(void)
855 bailIfCursor(MYCURSORS, "can't begin transaction during r/o cursor");
858 CtdlLogPrintf(CTDL_EMERG, "cdb_begin_transaction: ERROR: nested transaction\n");
865 void cdb_end_transaction(void)
869 for (i = 0; i < MAXCDB; i++)
870 if (MYCURSORS[i] != NULL) {
871 CtdlLogPrintf(CTDL_WARNING,
872 "cdb_end_transaction: WARNING: cursor %d still open at transaction end\n",
874 cclose(MYCURSORS[i]);
879 CtdlLogPrintf(CTDL_EMERG,
880 "cdb_end_transaction: ERROR: txcommit(NULL) !!\n");
890 * Truncate (delete every record)
892 void cdb_trunc(int cdb)
899 CtdlLogPrintf(CTDL_EMERG,
900 "cdb_trunc must not be called in a transaction.\n");
903 bailIfCursor(MYCURSORS, "attempt to write during r/o cursor");
908 if ((ret = dbp[cdb]->truncate(dbp[cdb], /* db */
909 NULL, /* transaction ID */
910 &count, /* #rows deleted */
912 if (ret == DB_LOCK_DEADLOCK) {
916 CtdlLogPrintf(CTDL_EMERG, "cdb_truncate(%d): %s\n", cdb, db_strerror(ret));
918 CtdlLogPrintf(CTDL_EMERG, "You may need to tune your database; please read http://www.citadel.org/doku.php/faq:troubleshooting:out_of_lock_entries for more information.\n");