From: Art Cancro <ajc@citadel.org>
Date: Thu, 24 Aug 2023 23:16:20 +0000 (-0900)
Subject: ctdlload: use the backend API instead of BDB calls
X-Git-Tag: v989~46
X-Git-Url: https://code.citadel.org/?a=commitdiff_plain;h=311527f49f3cb0e718e9a6a6392e27910407b284;p=citadel.git

ctdlload: use the backend API instead of BDB calls

It works, but it is several orders of magnitude slower.  Why?
---

diff --git a/citadel/Makefile b/citadel/Makefile
index cf13c6eca..58425abe4 100644
--- a/citadel/Makefile
+++ b/citadel/Makefile
@@ -56,8 +56,8 @@ chkpwd: utils/chkpwd.c utils/auth.c utils/*.h server/*.h
 ctdldump: utils/ctdldump.c utils/*.h server/*.h ${BACKEND_OBJECTS}
 	cc ${CFLAGS} ${LDFLAGS} utils/ctdldump.c ${BACKEND_OBJECTS} -lcitadel -lz ${BACKEND_LDFLAGS} -lpthread -o ctdldump
 
-ctdlload: utils/ctdlload.c server/makeuserkey.c utils/*.h server/*.h
-	cc ${CFLAGS} ${LDFLAGS} utils/ctdlload.c server/makeuserkey.c -lcitadel -lz ${BACKEND_LDFLAGS} -lpthread -o ctdlload
+ctdlload: utils/ctdlload.c server/makeuserkey.c utils/*.h server/*.h ${BACKEND_OBJECTS}
+	cc ${CFLAGS} ${LDFLAGS} utils/ctdlload.c server/makeuserkey.c ${BACKEND_OBJECTS} -lcitadel -lz ${BACKEND_LDFLAGS} -lpthread -o ctdlload
 
 config.mk: configure
 	./configure
diff --git a/citadel/utils/ctdlload.c b/citadel/utils/ctdlload.c
index 3a5f64a6c..45e68536e 100644
--- a/citadel/utils/ctdlload.c
+++ b/citadel/utils/ctdlload.c
@@ -19,14 +19,15 @@
 #include <errno.h>
 #include <stdarg.h>
 #include <limits.h>
+#include <syslog.h>
 #include <libcitadel.h>
 #include <zlib.h>
-#include <db.h>
 #include "../server/sysdep.h"
 #include "../server/citadel_defs.h"
 #include "../server/server.h"
 #include "../server/makeuserkey.h"
 #include "../server/citadel_dirs.h"
+#include "../server/database.h"
 
 uid_t ctdluid = 0;
 
@@ -41,80 +42,9 @@ void *reallok(void *ptr, size_t size) {
 }
 
 
-// Open a database environment
-DB_ENV *open_dbenv(char *dirname) {
-
-	DB_ENV *dbenv = NULL;
-
-	int ret;
-	int i;
-	u_int32_t flags = 0;
-	int dbversion_major, dbversion_minor, dbversion_patch;
- 	db_version(&dbversion_major, &dbversion_minor, &dbversion_patch);
-
-	// Create synthetic integer version numbers and compare them.
-	// Never run with a libdb other than the one with which it was compiled.
-	int compiled_db_version = ( (DB_VERSION_MAJOR * 1000000) + (DB_VERSION_MINOR * 1000) + (DB_VERSION_PATCH) );
-	int linked_db_version = ( (dbversion_major * 1000000) + (dbversion_minor * 1000) + (dbversion_patch) );
-	if (compiled_db_version != linked_db_version) {
-		fprintf(stderr,
-			"db: ctdlload is running with a version of libdb other than the one with which it was compiled.\n"
-			"db: compiled: %d\n"
-			"db:   linked: %d\n"
-			"db: This is an invalid configuration.  ctdlload will now exit to prevent data loss.",
-			compiled_db_version,
-			linked_db_version
-		);
-		exit(CTDLEXIT_DB);
-	}
-
-	ret = db_env_create(&dbenv, 0);
-	if (ret) {
-		fprintf(stderr,"db: db_env_create: %s\n", db_strerror(ret));
-		fprintf(stderr,"db: exit code %d\n", ret);
-		exit(CTDLEXIT_DB);
-	}
-
-	// We want to specify the shared memory buffer pool cachesize, but everything else is the default.
-	ret = dbenv->set_cachesize(dbenv, 0, 64 * 1024, 0);
-	if (ret) {
-		fprintf(stderr,"db: set_cachesize: %s\n", db_strerror(ret));
-		dbenv->close(dbenv, 0);
-		fprintf(stderr,"db: exit code %d\n", ret);
-		exit(CTDLEXIT_DB);
-	}
-
-	if ((ret = dbenv->set_lk_detect(dbenv, DB_LOCK_DEFAULT))) {
-		fprintf(stderr,"db: set_lk_detect: %s\n", db_strerror(ret));
-		dbenv->close(dbenv, 0);
-		fprintf(stderr,"db: exit code %d\n", ret);
-		exit(CTDLEXIT_DB);
-	}
-
-	flags = DB_CREATE | DB_INIT_MPOOL | DB_PRIVATE | DB_INIT_LOG;
-	ret = dbenv->open(dbenv, dirname, flags, 0);
-	if (ret) {
-		fprintf(stderr,"db: dbenv->open: %s\n", db_strerror(ret));
-		dbenv->close(dbenv, 0);
-		fprintf(stderr,"db: exit code %d\n", ret);
-		exit(CTDLEXIT_DB);
-	}
-
-	return(dbenv);
-}
-
-
-void close_dbenv(DB_ENV *dbenv) {
-	int ret = dbenv->close(dbenv, 0);
-	if (ret) {
-		fprintf(stderr,"db: dbenv->close: %s\n", db_strerror(ret));
-	}
-}
-
-
 // Convert a "msgtext" record to a message on disk.   NOT THREADSAFE
 // This also works for "bigmsg" records.
-int import_msgtext(char *line, DBT *out_key, DBT *out_data) {
+int import_msgtext(char *line, struct cdbkeyval *kv) {
 
 	static char *b64_decoded_msg = NULL;
 	static size_t b64_decoded_alloc = 0;
@@ -126,20 +56,20 @@ int import_msgtext(char *line, DBT *out_key, DBT *out_data) {
 	token = strtok(NULL, "|");
 
 	// The record key will be the message number
-	out_key->size = sizeof(long);
-	out_key->data = reallok(out_key->data, out_key->size);
-	memcpy(out_key->data, &msgnum, out_key->size);
+	kv->key.len = sizeof(long);
+	kv->key.ptr = reallok(kv->key.ptr, kv->key.len);
+	memcpy(kv->key.ptr, &msgnum, kv->key.len);
 
 	// The record data will be the decoded message text.
 	// We are allocating more memory than we need, but BDB will only write the number of bytes we tell it to.
-	out_data->data = reallok(out_data->data, strlen(token));
-	out_data->size = CtdlDecodeBase64(out_data->data, token, strlen(token));
+	kv->val.ptr = (char *) reallok(kv->val.ptr, strlen(token));
+	kv->val.len = CtdlDecodeBase64(kv->val.ptr, token, strlen(token));
 	return(1);
 }
 
 
 // Convert a "msgmeta" record to a message metadata record on disk.  NOT THREADSAFE
-int import_msgmeta(char *line, DBT *out_key, DBT *out_data) {
+int import_msgmeta(char *line, struct cdbkeyval *kv) {
 	char *token;
 	struct MetaData *m = malloc(sizeof(struct MetaData));
 
@@ -166,20 +96,20 @@ int import_msgmeta(char *line, DBT *out_key, DBT *out_data) {
 	// metadata records are stored in the CDB_MSGMAIN table,
 	// but with the index being the *negative* of the message number.
 	long index = 0 - m->meta_msgnum;
-	out_key->size = sizeof(long);
-	out_key->data = reallok(NULL, out_key->size);
-	memcpy(out_key->data, &index, out_key->size);
+	kv->key.len = sizeof(long);
+	kv->key.ptr = reallok(NULL, kv->key.len);
+	memcpy(kv->key.ptr, &index, kv->key.len);
 
 	// data
-	out_data->size = sizeof(struct MetaData);
-	out_data->data = m;				// out_data owns this memory now
+	kv->val.len = sizeof(struct MetaData);
+	kv->val.ptr = (char *) m;				// out_data owns this memory now
 
 	return(1);
 }
 
 
 // Convert a "user" record to a record on disk.  (Source string is unusable after this function is called.)
-int import_user(char *line, DBT *out_key, DBT *out_data) {
+int import_user(char *line, struct cdbkeyval *kv) {
 	char userkey[USERNAME_SIZE];
 	char *token;
 	struct ctdluser *u = malloc(sizeof(struct ctdluser));
@@ -235,16 +165,16 @@ int import_user(char *line, DBT *out_key, DBT *out_data) {
 	}
 	
 	makeuserkey(userkey, u->fullname);
-	out_key->size = strlen(userkey);
-	out_key->data = strdup(userkey);
-	out_data->size = sizeof(struct ctdluser);
-	out_data->data = u;
+	kv->key.len = strlen(userkey);
+	kv->key.ptr = strdup(userkey);
+	kv->val.len = sizeof(struct ctdluser);
+	kv->val.ptr = (char *) u;
 	return(1);
 }
 
 
 // Convert a "room" record to a record on disk.  (Source string is unusable after this function is called.)
-int import_room(char *line, DBT *out_key, DBT *out_data) {
+int import_room(char *line, struct cdbkeyval *kv) {
 	char *token;
 	struct ctdlroom *r = malloc(sizeof(struct ctdlroom));
 
@@ -308,21 +238,21 @@ int import_room(char *line, DBT *out_key, DBT *out_data) {
 	}
 
 	// The key is the room name in all lower case
-	out_key->size = strlen(r->QRname);
-	out_key->data = strdup(r->QRname);
-	char *k = (char *)out_key->data;
-	for (int i=0; i<=out_key->size; ++i) {
+	kv->key.len = strlen(r->QRname);
+	kv->key.ptr = strdup(r->QRname);
+	char *k = (char *)kv->key.ptr;
+	for (int i=0; i<=kv->key.len; ++i) {
 		k[i] = tolower(k[i]);
 	}
 
-	out_data->size = sizeof(struct ctdlroom);
-	out_data->data = r;
+	kv->val.len = sizeof(struct ctdlroom);
+	kv->val.ptr = (char *) r;
 	return(1);
 }
 
 
 // Convert a floor record to a record on disk.
-int import_floor(char *line, DBT *out_key, DBT *out_data) {
+int import_floor(char *line, struct cdbkeyval *kv) {
 	char *token;
 	struct floor *f = malloc(sizeof(struct floor));
 	int floor_num;
@@ -353,19 +283,19 @@ int import_floor(char *line, DBT *out_key, DBT *out_data) {
 		}
 	}
 
-	out_key->size = sizeof(int);
-	out_key->data = malloc(out_key->size);
-	memcpy(out_key->data, &floor_num, out_key->size);
+	kv->key.len = sizeof(int);
+	kv->key.ptr = malloc(kv->key.len);
+	memcpy(kv->key.ptr, &floor_num, kv->key.len);
 
-	out_data->size = sizeof(struct floor);
-	out_data->data = f;
+	kv->val.len = sizeof(struct floor);
+	kv->val.ptr = (char *) f;
 	return(1);
 }
 
 
 // Import a msglist record
 // msglist|26|32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51|
-int import_msglist(char *line, DBT *out_key, DBT *out_data) {
+int import_msglist(char *line, struct cdbkeyval *kv) {
 	long roomnum;
 	char *token, *mtoken;
 	char *p = line;
@@ -388,19 +318,19 @@ int import_msglist(char *line, DBT *out_key, DBT *out_data) {
 		}
 	}
 
-	out_key->size = sizeof(long);
-	out_key->data = malloc(out_key->size);
-	memcpy(out_key->data, &roomnum, out_key->size);
+	kv->key.len = sizeof(long);
+	kv->key.ptr = malloc(kv->key.len);
+	memcpy(kv->key.ptr, &roomnum, kv->key.len);
 
-	out_data->size = num_msgs * sizeof(long);
-	out_data->data = msglist;
+	kv->val.len = num_msgs * sizeof(long);
+	kv->val.ptr = (char *) msglist;
 
 	return(1);
 }
 
 
 // Convert a "visit" record to a record on disk.
-int import_visit(char *line, DBT *out_key, DBT *out_data) {
+int import_visit(char *line, struct cdbkeyval *kv) {
 	char *token;
 	struct visit *v = malloc(sizeof(struct visit));
 
@@ -437,18 +367,18 @@ int import_visit(char *line, DBT *out_key, DBT *out_data) {
 	}
 
 	// The key is the same as the first three data members (3 x long)
-	out_key->size = sizeof(long) * 3;
-	out_key->data = reallok(NULL, out_key->size);
-	memcpy(out_key->data, v, out_key->size);
+	kv->key.len = sizeof(long) * 3;
+	kv->key.ptr = reallok(NULL, kv->key.len);
+	memcpy(kv->key.ptr, v, kv->key.len);
 
-	out_data->size = sizeof(struct visit);
-	out_data->data = v;
+	kv->val.len = sizeof(struct visit);
+	kv->val.ptr = (char *) v;
 	return(1);
 }
 
 
 // Convert a "dir" record to a record on disk.
-int import_dir(char *line, DBT *out_key, DBT *out_data) {
+int import_dir(char *line, struct cdbkeyval *kv) {
 	char dirkey[SIZ];
 	char username[USERNAME_SIZE];
 	char *token;
@@ -465,19 +395,19 @@ int import_dir(char *line, DBT *out_key, DBT *out_data) {
 		}
 	}
 
-	out_key->size = strlen(dirkey);
-	out_key->data = reallok(NULL, out_key->size);
-	memcpy(out_key->data, dirkey, strlen(dirkey));
+	kv->key.len = strlen(dirkey);
+	kv->key.ptr = reallok(NULL, kv->key.len);
+	memcpy(kv->key.ptr, dirkey, strlen(dirkey));
 
-	out_data->size = strlen(username) + 1;
-	out_data->data = strdup(username);
+	kv->val.len = strlen(username) + 1;
+	kv->val.ptr = strdup(username);
 
 	return(1);
 }
 
 
 // Convert a "usetable" record to a record on disk.
-int import_usetable(char *line, DBT *out_key, DBT *out_data) {
+int import_usetable(char *line, struct cdbkeyval *kv) {
 	char *token;
 	struct UseTable *u = malloc(sizeof(struct UseTable));
 
@@ -496,19 +426,19 @@ int import_usetable(char *line, DBT *out_key, DBT *out_data) {
 	}
 
 	// the key is just an int (the hash)
-	out_key->size = sizeof(int);
-	out_key->data = reallok(NULL, out_key->size);
-	memcpy(out_key->data, &u->hash, out_key->size);
+	kv->key.len = sizeof(int);
+	kv->key.ptr = reallok(NULL, kv->key.len);
+	memcpy(kv->key.ptr, &u->hash, kv->key.len);
 
-	out_data->size = sizeof(struct UseTable);
-	out_data->data = u;
+	kv->val.len = sizeof(struct UseTable);
+	kv->val.ptr = (char *) u;
 	return(1);
 }
 
 
 // Import a full text search index record.
 // It's just like a msglists record: a key and a list of message numbers, but the key is "int" instead of "long"
-int import_fulltext(char *line, DBT *out_key, DBT *out_data) {
+int import_fulltext(char *line, struct cdbkeyval *kv) {
 	int indexnum;
 	char *token, *mtoken;
 	char *p = line;
@@ -531,12 +461,12 @@ int import_fulltext(char *line, DBT *out_key, DBT *out_data) {
 		}
 	}
 
-	out_key->size = sizeof(int);
-	out_key->data = malloc(out_key->size);
-	memcpy(out_key->data, &indexnum, out_key->size);
+	kv->key.len = sizeof(int);
+	kv->key.ptr = malloc(kv->key.len);
+	memcpy(kv->key.ptr, &indexnum, kv->key.len);
 
-	out_data->size = num_msgs * sizeof(long);
-	out_data->data = msglist;
+	kv->val.len = num_msgs * sizeof(long);
+	kv->val.ptr = (char *) msglist;
 
 	return(1);
 }
@@ -544,7 +474,7 @@ int import_fulltext(char *line, DBT *out_key, DBT *out_data) {
 
 // Import an EUID Index record
 // euidindex|msgnum|roomnum|euid|
-int import_euidindex(char *line, DBT *out_key, DBT *out_data) {
+int import_euidindex(char *line, struct cdbkeyval *kv) {
 	char euid[SIZ];
 	long msgnum;
 	long roomnum;
@@ -568,19 +498,19 @@ int import_euidindex(char *line, DBT *out_key, DBT *out_data) {
 	// The structure of an euidindex record *key* is:
 	// |----room_number----|----------EUID-------------|
 	//    (sizeof long)       (actual length of euid)
-	out_key->size = sizeof(long) + strlen(euid) + 1;
-	out_key->data = reallok(NULL, out_key->size);
-	memcpy(out_key->data, &roomnum, sizeof(long));
-	strcpy(out_key->data + sizeof(long), euid);
+	kv->key.len = sizeof(long) + strlen(euid) + 1;
+	kv->key.ptr = reallok(NULL, kv->key.len);
+	memcpy(kv->key.ptr, &roomnum, sizeof(long));
+	strcpy(kv->key.ptr + sizeof(long), euid);
 
 	// The structure of an euidindex record *value* is:
 	// |-----msg_number----|----room_number----|----------EUID-------------|
 	//    (sizeof long)       (sizeof long)       (actual length of euid)
-	out_data->size = sizeof(long) + sizeof(long) + strlen(euid) + 1;
-	out_data->data = reallok(NULL, out_data->size);
-	memcpy(out_data->data, &msgnum, sizeof(long));
-	memcpy(out_data->data + sizeof(long), &roomnum, sizeof(long));
-	strcpy(out_data->data + sizeof(long) + sizeof(long), euid);
+	kv->val.len = sizeof(long) + sizeof(long) + strlen(euid) + 1;
+	kv->val.ptr = (char *) reallok(NULL, kv->val.len);
+	memcpy(kv->val.ptr, &msgnum, sizeof(long));
+	memcpy(kv->val.ptr + sizeof(long), &roomnum, sizeof(long));
+	strcpy(kv->val.ptr + sizeof(long) + sizeof(long), euid);
 
 	return(1);
 }
@@ -588,7 +518,7 @@ int import_euidindex(char *line, DBT *out_key, DBT *out_data) {
 
 // Import a "users by number" (secondary index) record
 // The key is a "long"
-int import_usersbynumber(char *line, DBT *out_key, DBT *out_data) {
+int import_usersbynumber(char *line, struct cdbkeyval *kv) {
 	char *token;
 	long usernum;
 
@@ -599,11 +529,11 @@ int import_usersbynumber(char *line, DBT *out_key, DBT *out_data) {
 				usernum = atol(token);
 				break;
 			case 2:
-				out_key->size = sizeof(long);
-				out_key->data = reallok(NULL, sizeof(long));
-				memcpy(out_key->data, &usernum, out_key->size);
-				out_data->data = strdup(token);
-				out_data->size = strlen(out_data->data) + 1;
+				kv->key.len = sizeof(long);
+				kv->key.ptr = reallok(NULL, sizeof(long));
+				memcpy(kv->key.ptr, &usernum, kv->key.len);
+				kv->val.ptr = (char *) strdup(token);
+				kv->val.len = strlen(kv->val.ptr) + 1;
 				return(1);
 		}
 	}
@@ -615,7 +545,7 @@ int import_usersbynumber(char *line, DBT *out_key, DBT *out_data) {
 // Import a config record
 // The key is the config key
 // The data is the config key, a null, the value, and another null
-int import_config(char *line, DBT *out_key, DBT *out_data) {
+int import_config(char *line, struct cdbkeyval *kv) {
 	char *token;
 	char *p = line;
 	char *k, *v;
@@ -624,15 +554,15 @@ int import_config(char *line, DBT *out_key, DBT *out_data) {
 		switch(i) {
 			case 1:
 				k = token;
-				out_key->size = strlen(token);
-				out_key->data = strdup(token);
+				kv->key.len = strlen(token);
+				kv->key.ptr = strdup(token);
 				break;
 			case 2:
 				v = token;
-				out_data->size = strlen(k) + strlen(v) + 2;
-				out_data->data = reallok(NULL, out_data->size);
-				strcpy(out_data->data, k);
-				strcpy(out_data->data + strlen(k) + 1, v);
+				kv->val.len = strlen(k) + strlen(v) + 2;
+				kv->val.ptr = (char *) reallok(NULL, kv->val.len);
+				strcpy(kv->val.ptr, k);
+				strcpy(kv->val.ptr + strlen(k) + 1, v);
 				break;
 		}
 	}
@@ -642,30 +572,17 @@ int import_config(char *line, DBT *out_key, DBT *out_data) {
 
 
 // Ingest one line of dump data.  NOT REENTRANT
-void ingest_one(char *line, DB_ENV *dst_dbenv) {
+void ingest_one(char *line, struct cdbkeyval *kv) {
 
 	static int good_rows = 0;
 	static int bad_rows = 0;
-	static int previous_cdb = -1 ;
 	static int current_cdb = -1 ;
-	static DB *dst_dbp;
 	char record_type[32];
-	int ret;
-	char dbfilename[32];
 	int row_was_good;
-	DBT out_key, out_data;
-
-	// We are assuming that the lines of the dump file will generally be sorted by table.
-	// By remembering the last table we worked with, we can do close/open if the table changes.
-
-	if (current_cdb >= 0) {
-		fprintf(stderr, "   \033[33m%02x \033[32m%9d \033[31m%8d\033[0m\r", current_cdb, good_rows, bad_rows);
-		fflush(stderr);
-	}
+	static time_t last_diag = 0;
 
 	// Clear out our record buffer
-	memset(&out_key, 0, sizeof(DBT));
-	memset(&out_data, 0, sizeof(DBT));
+	memset(kv, 0, sizeof(struct cdbkeyval));
 	row_was_good = 0;
 
 	// Identify the record type we are currently working with,
@@ -673,130 +590,93 @@ void ingest_one(char *line, DB_ENV *dst_dbenv) {
 	extract_token(record_type, line, 0, '|', sizeof record_type);
 	if (!strcasecmp(record_type, "msgtext")) {
 		current_cdb = CDB_MSGMAIN;
-		row_was_good = import_msgtext(line, &out_key, &out_data);
+		row_was_good = import_msgtext(line, kv);
 	}
 	else if (!strcasecmp(record_type, "msgmeta")) {
 		current_cdb = CDB_MSGMAIN;
-		row_was_good = import_msgmeta(line, &out_key, &out_data);
+		row_was_good = import_msgmeta(line, kv);
 	}
 	else if (!strcasecmp(record_type, "user")) {
 		current_cdb = CDB_USERS;
-		row_was_good = import_user(line, &out_key, &out_data);
+		row_was_good = import_user(line, kv);
 	}
 	else if (!strcasecmp(record_type, "room")) {
 		current_cdb = CDB_ROOMS;
-		row_was_good = import_room(line, &out_key, &out_data);
+		row_was_good = import_room(line, kv);
 	}
 	else if (!strcasecmp(record_type, "floor")) {
 		current_cdb = CDB_FLOORTAB;
-		row_was_good = import_floor(line, &out_key, &out_data);
+		row_was_good = import_floor(line, kv);
 	}
 	else if (!strcasecmp(record_type, "msglist")) {
 		current_cdb = CDB_MSGLISTS;
-		row_was_good = import_msglist(line, &out_key, &out_data);
+		row_was_good = import_msglist(line, kv);
 	}
 	else if (!strcasecmp(record_type, "visit")) {
 		current_cdb = CDB_VISIT;
-		row_was_good = import_visit(line, &out_key, &out_data);
+		row_was_good = import_visit(line, kv);
 	}
 	else if (!strcasecmp(record_type, "dir")) {
 		current_cdb = CDB_DIRECTORY;
-		row_was_good = import_dir(line, &out_key, &out_data);
+		row_was_good = import_dir(line, kv);
 	}
 	else if (!strcasecmp(record_type, "use")) {
 		current_cdb = CDB_USETABLE;
-		row_was_good = import_usetable(line, &out_key, &out_data);
+		row_was_good = import_usetable(line, kv);
 	}
 	else if (!strcasecmp(record_type, "bigmsg")) {
 		current_cdb = CDB_BIGMSGS;
-		row_was_good = import_msgtext(line, &out_key, &out_data);
+		row_was_good = import_msgtext(line, kv);
 	}
 	else if (!strcasecmp(record_type, "fulltext")) {
 		current_cdb = CDB_FULLTEXT;
-		row_was_good = import_fulltext(line, &out_key, &out_data);
+		row_was_good = import_fulltext(line, kv);
 	}
 	else if (!strcasecmp(record_type, "euidindex")) {
 		current_cdb = CDB_EUIDINDEX;
-		row_was_good = import_euidindex(line, &out_key, &out_data);
+		row_was_good = import_euidindex(line, kv);
 	}
 	else if (!strcasecmp(record_type, "usersbynumber")) {
 		current_cdb = CDB_USERSBYNUMBER;
-		row_was_good = import_usersbynumber(line, &out_key, &out_data);
+		row_was_good = import_usersbynumber(line, kv);
 	}
 	else if (!strcasecmp(record_type, "config")) {
 		current_cdb = CDB_CONFIG;
-		row_was_good = import_config(line, &out_key, &out_data);
+		row_was_good = import_config(line, kv);
 	}
 	else {
 		current_cdb = -1 ;
 	}
 
-	// If the current record is a different table than the previous record,
-	// then close the other table and open the correct one.
-	if (current_cdb != previous_cdb) {
-		if (previous_cdb >= 0) {
-			fprintf(stderr, "\n");
-		}
-		if (previous_cdb >= 0) {
-			ret = dst_dbp->close(dst_dbp, 0);
-			if (ret) {
-				fprintf(stderr, "db: db_close: %s\n", db_strerror(ret));
-			}
-		}
-
-		if (current_cdb >= 0) {
-			good_rows = 0;
-			bad_rows = 0;
-			snprintf(dbfilename, sizeof dbfilename, "cdb.%02x", current_cdb);
-
-			// create a database handle for the destination table
-			ret = db_create(&dst_dbp, dst_dbenv, 0);
-			if (ret) {
-				fprintf(stderr, "db: db_create: %s\n", db_strerror(ret));
-				fprintf(stderr, "db: exit code %d\n", ret);
-				exit(CTDLEXIT_DB);
-			}
-		
-			// open the file containing the destination table
-			ret = dst_dbp->open(dst_dbp, NULL, dbfilename, NULL, DB_BTREE, (DB_CREATE | DB_TRUNCATE), 0600);
-			if (ret) {
-				fprintf(stderr, "db: db_open(%s): %s\n", dbfilename, db_strerror(ret));
-				fprintf(stderr, "db: exit code %d\n", ret);
-				exit(CTDLEXIT_DB);
-			}
-		}
-
-		previous_cdb = current_cdb;
-	}
-
 	// If the conversion function was successful, write the record to the database.
-	if (row_was_good) {
+	if ( (row_was_good) && (current_cdb >= 0) ) {
+		cdb_store(current_cdb, kv->key.ptr, kv->key.len, kv->val.ptr, kv->val.len);
 		++good_rows;
-		ret = dst_dbp->put(dst_dbp, NULL, &out_key, &out_data, 0);
-		if (ret) {
-			fprintf(stderr, "db: cdb_put(%x): %s", current_cdb, db_strerror(ret));
-			exit(CTDLEXIT_DB);
-		}
 	}
 	else {
 		++bad_rows;
 	}
+	if (time(NULL) > last_diag) {
+		fprintf(stderr, "   \033[32m%9d\033[0m / \033[31m%8d\033[0m\r", good_rows, bad_rows);
+		fflush(stderr);
+		last_diag = time(NULL);
+	}
 
-	free(out_key.data);
-	free(out_data.data);
 }
 
 
 // This is the loop that loads the dump data.  NOT REENTRANT
-void ingest(DB_ENV *dst_dbenv) {
+void ingest(void) {
 	static size_t line_alloc = 1;
 	static char *line;
 	static size_t line_len = 0;
 	int ch;
 	int begin_found = 0;
 	int end_found = 0;
+	struct cdbkeyval kv;
+
+	memset(&kv, 0, sizeof(struct cdbkeyval));
 
-	fprintf(stderr, "\033[7mtable\033[0m \033[7mgood_rows\033[0m \033[7mbad_rows\033[0m\n");
 	line = reallok(NULL, line_alloc);
 
 	do {
@@ -816,9 +696,17 @@ void ingest(DB_ENV *dst_dbenv) {
 		}
 
 		if (line_len > 0) {
-			ingest_one(line, dst_dbenv);
-			if (!strncasecmp(line, HKEY("begin|"))) begin_found = 1;
-			if (!strncasecmp(line, HKEY("end|"))) end_found = 1;
+			if ( (begin_found) && (!end_found) ) {
+				ingest_one(line, &kv);
+			}
+			if (!strncasecmp(line, HKEY("begin|"))) {
+				begin_found = 1;
+				fprintf(stderr, "   good rows / bad rows:\n");
+			}
+			if (!strncasecmp(line, HKEY("end|"))) {
+				fprintf(stderr, "\n");
+				end_found = 1;
+			}
 		}
 
 	} while (ch >= 0);
@@ -836,7 +724,6 @@ void ingest(DB_ENV *dst_dbenv) {
 int main(int argc, char **argv) {
 	char dst_dir[PATH_MAX];
 	int confirmed = 0;
-	static DB_ENV *dst_dbenv;		// Target DB environment
 
 	// display the greeting
 	fprintf(stderr,	"\033[44m\033[33m\033[1m \033[K\033[0m\n"
@@ -880,14 +767,22 @@ int main(int argc, char **argv) {
 		exit(1);
 	}
 
+	// backend modules use syslog -- redirect to stderr
+	openlog("ctdlload", LOG_PERROR , LOG_DAEMON);
+
 	// Remove any database that is already in the target directory (yes, delete it, be careful)
 	char cmd[PATH_MAX];
 	snprintf(cmd, sizeof cmd, "rm -fv %s/cdb.* %s/log.*", dst_dir, dst_dir);
 	system(cmd);
 
-	dst_dbenv = open_dbenv(dst_dir);
-	ingest(dst_dbenv);
-	close_dbenv(dst_dbenv);
+	// initialize the database backend
+	cdb_init_backends();
+	cdb_open_databases();
+
+	ingest();
+
+	// close databases
+	cdb_close_databases();
 
 	fprintf(stderr, "ctdlload: \033[32m\033[1mfinished\033[0m\n");
 	exit(0);