X-Git-Url: https://code.citadel.org/?a=blobdiff_plain;f=citadel%2Fdatabase_sleepycat.c;h=16b98813633c169176901c5063989d03feeef462;hb=fe4a9ff9d1b33197fb5dfc45567b98c63049d7b6;hp=82d886e8989a4fd994a0641585e1b7d333c3751b;hpb=3291d029e1d490e5cf9d5518630d78516752eb35;p=citadel.git

diff --git a/citadel/database_sleepycat.c b/citadel/database_sleepycat.c
index 82d886e89..16b988136 100644
--- a/citadel/database_sleepycat.c
+++ b/citadel/database_sleepycat.c
@@ -51,56 +51,63 @@ static pthread_key_t tsdkey;
 #define MYTID		(((struct cdbtsd*)pthread_getspecific(tsdkey))->tid)
 
 /* just a little helper function */
-static int txabort(DB_TXN *tid) {
+static void txabort(DB_TXN *tid) {
         int ret = txn_abort(tid);
 
-        if (ret)
-                lprintf(1, "txn_abort: %s\n", db_strerror(ret));
-
-        return ret;
+        if (ret) {
+                lprintf(1, "cdb_*: txn_abort: %s\n", db_strerror(ret));
+		abort();
+	}
 }
 
 /* this one is even more helpful than the last. */
-static int txcommit(DB_TXN *tid) {
+static void txcommit(DB_TXN *tid) {
         int ret = txn_commit(tid, 0);
 
-        if (ret)
-                lprintf(1, "txn_commit: %s\n", db_strerror(ret));
-
-        return ret;
+        if (ret) {
+                lprintf(1, "cdb_*: txn_commit: %s\n", db_strerror(ret));
+		abort();
+	}
 }
 
 /* are you sensing a pattern yet? */
-static int txbegin(DB_TXN **tid) {
+static void txbegin(DB_TXN **tid) {
         int ret = txn_begin(dbenv, NULL, tid, 0);
 
-        if (ret)
-                lprintf(1, "txn_begin: %s\n", db_strerror(ret));
+        if (ret) {
+                lprintf(1, "cdb_*: txn_begin: %s\n", db_strerror(ret));
+		abort();
+	}
+}
 
-        return ret;
+static void cclose(DBC *cursor) {
+	int ret;
+
+	if ((ret = cursor->c_close(cursor))) {
+		lprintf(1, "cdb_*: c_close: %s\n", db_strerror(ret));
+		abort();
+	}
 }
 
-static void release_handles(void *arg) {
+static void check_handles(void *arg) {
 	if (arg != NULL) {
 		struct cdbtsd *tsd = (struct cdbtsd *)arg;
 
 		if (tsd->cursor != NULL) {
-			lprintf(1, "WARNING: cursor still in progress; "
-				"closing!\n");
-			tsd->cursor->c_close(tsd->cursor);
+			lprintf(1, "cdb_*: cursor still in progress!");
+			abort();
 		}
 
 		if (tsd->tid != NULL) {
-			lprintf(1, "ERROR: transaction still in progress; "
-				"aborting!\n");
-			txabort(tsd->tid);
+			lprintf(1, "cdb_*: transaction still in progress!");
+			abort();
 		}
 	}
 }
 
 static void dest_tsd(void *arg) {
 	if (arg != NULL) {
-		release_handles(arg);
+		check_handles(arg);
 		phree(arg);
 	}
 }
@@ -114,7 +121,12 @@ static void dest_tsd(void *arg) {
  * to use database calls, except for whatever thread calls open_databases.
  */
 void cdb_allocate_tsd(void) {
-	struct cdbtsd *tsd = mallok(sizeof *tsd);
+	struct cdbtsd *tsd;
+
+	if (pthread_getspecific(tsdkey) != NULL)
+		return;
+
+	tsd = mallok(sizeof *tsd);
 
 	tsd->tid = NULL;
 	tsd->cursor = NULL;
@@ -126,8 +138,8 @@ void cdb_free_tsd(void) {
 	pthread_setspecific(tsdkey, NULL);
 }
 
-void cdb_release_handles(void) {
-	release_handles(pthread_getspecific(tsdkey));
+void cdb_check_handles(void) {
+	check_handles(pthread_getspecific(tsdkey));
 }
 
 
@@ -156,7 +168,8 @@ static void cdb_checkpoint(void) {
 				MAX_CHECKPOINT_MINUTES,
 				0);
 	if (ret) {
-		lprintf(1, "txn_checkpoint: %s\n", db_strerror(ret));
+		lprintf(1, "cdb_checkpoint: txn_checkpoint: %s\n", db_strerror(ret));
+		abort();
 	}
 }
 
@@ -173,18 +186,18 @@ void open_databases(void)
 	char dbfilename[SIZ];
 	u_int32_t flags = 0;
 
-	lprintf(9, "open_databases() starting\n");
+	lprintf(9, "cdb_*: open_databases() starting\n");
         /*
          * Silently try to create the database subdirectory.  If it's
          * already there, no problem.
          */
         system("exec mkdir data 2>/dev/null");
 
-	lprintf(9, "Setting up DB environment\n");
+	lprintf(9, "cdb_*: Setting up DB environment\n");
 	db_env_set_func_yield(sched_yield);
 	ret = db_env_create(&dbenv, 0);
 	if (ret) {
-		lprintf(1, "db_env_create: %s\n", db_strerror(ret));
+		lprintf(1, "cdb_*: db_env_create: %s\n", db_strerror(ret));
 		exit(ret);
 	}
 	dbenv->set_errpfx(dbenv, "citserver");
@@ -195,13 +208,13 @@ void open_databases(void)
          */
         ret = dbenv->set_cachesize(dbenv, 0, 64 * 1024, 0);
 	if (ret) {
-		lprintf(1, "set_cachesize: %s\n", db_strerror(ret));
+		lprintf(1, "cdb_*: set_cachesize: %s\n", db_strerror(ret));
                 dbenv->close(dbenv, 0);
                 exit(ret);
         }
 
 	if ((ret = dbenv->set_lk_detect(dbenv, DB_LOCK_DEFAULT))) {
-		lprintf(1, "set_lk_detect: %s\n", db_strerror(ret));
+		lprintf(1, "cdb_*: set_lk_detect: %s\n", db_strerror(ret));
 		dbenv->close(dbenv, 0);
 		exit(ret);
 	}
@@ -210,19 +223,19 @@ void open_databases(void)
 		DB_INIT_LOCK|DB_THREAD;
         ret = dbenv->open(dbenv, "./data", flags, 0);
 	if (ret) {
-		lprintf(1, "dbenv->open: %s\n", db_strerror(ret));
+		lprintf(1, "cdb_*: dbenv->open: %s\n", db_strerror(ret));
                 dbenv->close(dbenv, 0);
                 exit(ret);
         }
 
-	lprintf(7, "Starting up DB\n");
+	lprintf(7, "cdb_*: Starting up DB\n");
 
 	for (i = 0; i < MAXCDB; ++i) {
 
 		/* Create a database handle */
 		ret = db_create(&dbp[i], dbenv, 0);
 		if (ret) {
-			lprintf(1, "db_create: %s\n", db_strerror(ret));
+			lprintf(1, "cdb_*: db_create: %s\n", db_strerror(ret));
 			exit(ret);
 		}
 
@@ -236,27 +249,27 @@ void open_databases(void)
 				dbfilename,
 				NULL,
 				DB_BTREE,
-				DB_CREATE,
+				DB_CREATE|DB_THREAD,
 				0600);
 		if (ret) {
-			lprintf(1, "db_open[%d]: %s\n", i, db_strerror(ret));
+			lprintf(1, "cdb_*: db_open[%d]: %s\n", i, db_strerror(ret));
 			exit(ret);
 		}
 	}
 
 	if ((ret = pthread_key_create(&tsdkey, dest_tsd))) {
-		lprintf(1, "pthread_key_create: %s\n", strerror(ret));
+		lprintf(1, "cdb_*: pthread_key_create: %s\n", strerror(ret));
 		exit(1);
 	}
 
 	cdb_allocate_tsd();
 	CtdlRegisterSessionHook(cdb_checkpoint, EVT_TIMER);
-	lprintf(9, "open_databases() finished\n");
+	lprintf(9, "cdb_*: open_databases() finished\n");
 }
 
 
 /*
- * Close all of the gdbm database files we've opened.  This can be done
+ * Close all of the db database files we've opened.  This can be done
  * in a loop, since it's just a bunch of closes.
  */
 void close_databases(void)
@@ -267,14 +280,16 @@ void close_databases(void)
 	cdb_free_tsd();
 
 	if ((ret = txn_checkpoint(dbenv, 0, 0, 0))) {
-		lprintf(1, "txn_checkpoint: %s\n", db_strerror(ret));
+		lprintf(1, "cdb_*: txn_checkpoint: %s\n", db_strerror(ret));
+		abort();
 	}
 
 	for (a = 0; a < MAXCDB; ++a) {
-		lprintf(7, "Closing database %d\n", a);
+		lprintf(7, "cdb_*: Closing database %d\n", a);
 		ret = dbp[a]->close(dbp[a], 0);
 		if (ret) {
-			lprintf(1, "db_close: %s\n", db_strerror(ret));
+			lprintf(1, "cdb_*: db_close: %s\n", db_strerror(ret));
+			abort();
 		}
 		
 	}
@@ -282,7 +297,8 @@ void close_databases(void)
         /* Close the handle. */
         ret = dbenv->close(dbenv, 0);
 	if (ret) {
-                lprintf(1, "DBENV->close: %s\n", db_strerror(ret));
+                lprintf(1, "cdb_*: DBENV->close: %s\n", db_strerror(ret));
+		abort();
         }
 }
 
@@ -315,12 +331,12 @@ int cdb_store(int cdb,
 		if (ret) {
 			lprintf(1, "cdb_store(%d): %s\n", cdb,
 				db_strerror(ret));
+			abort();
 		}
 		return ret;
 	} else {
 	    retry:
-		if (txbegin(&tid))
-			return -1;
+		txbegin(&tid);
 
 		if ((ret = dbp[cdb]->put(dbp[cdb],	/* db */
 					 tid,		/* transaction ID */
@@ -328,18 +344,16 @@ int cdb_store(int cdb,
 					 &ddata,	/* data */
 					 0))) {		/* flags */
 			if (ret == DB_LOCK_DEADLOCK) {
-				if (txabort(tid))
-					return ret;
-				else
-					goto retry;
+				txabort(tid);
+				goto retry;
 			} else {
 				lprintf(1, "cdb_store(%d): %s\n", cdb,
 					db_strerror(ret));
-				txabort(tid);
-				return ret;
+				abort();
 			}
 		} else {
-			return txcommit(tid);
+			txcommit(tid);
+			return ret;
 		}
 	}
 }
@@ -355,33 +369,37 @@ int cdb_delete(int cdb, void *key, int keylen)
 	DB_TXN *tid;
 	int ret;
 
+	memset(&dkey, 0, sizeof dkey);
 	dkey.size = keylen;
 	dkey.data = key;
 
 	if (MYTID != NULL) {
 		ret = dbp[cdb]->del(dbp[cdb], MYTID, &dkey, 0);
-		return (ret);
+		if (ret) {
+			lprintf(1, "cdb_delete(%d): %s\n", cdb,
+				db_strerror(ret));
+			if (ret != DB_NOTFOUND)
+				abort();
+		}
 	} else {
 	    retry:
-		if (txbegin(&tid))
-			return -1;
+		txbegin(&tid);
 
-		if ((ret = dbp[cdb]->del(dbp[cdb], tid, &dkey, 0))) {
+		if ((ret = dbp[cdb]->del(dbp[cdb], tid, &dkey, 0))
+		    && ret != DB_NOTFOUND) {
 			if (ret == DB_LOCK_DEADLOCK) {
-					if (txabort(tid))
-						return ret;
-					else
-						goto retry;
+					txabort(tid);
+					goto retry;
 			} else {
-				lprintf(1, "cdb_store(%d): %s\n", cdb,
+				lprintf(1, "cdb_delete(%d): %s\n", cdb,
 					db_strerror(ret));
-				txabort(tid);
-				return ret;
+				abort();
 			}
 		} else {
-			return txcommit(tid);
+			txcommit(tid);
 		}
 	}
+	return ret;
 }
 
 
@@ -410,27 +428,29 @@ struct cdbdata *cdb_fetch(int cdb, void *key, int keylen)
 		ret = dbp[cdb]->get(dbp[cdb], MYTID, &dkey, &dret, 0);
 	} else {
 	    retry:
-		if (txbegin(&tid))
-			return NULL;
+		txbegin(&tid);
 
 		ret = dbp[cdb]->get(dbp[cdb], tid, &dkey, &dret, 0);
 
 		if (ret == DB_LOCK_DEADLOCK) {
-			if (txabort(tid))
-				return NULL;
-			else
-				goto retry;
-		} else if (txcommit(tid))
-			return NULL;
+			txabort(tid);
+			goto retry;
+		}
+		if (ret && ret != DB_NOTFOUND)
+			abort();
+
+		txcommit(tid);
 	}
 
 	if ((ret != 0) && (ret != DB_NOTFOUND)) {
 		lprintf(1, "cdb_fetch: %s\n", db_strerror(ret));
+		abort();
 	}
 	if (ret != 0) return NULL;
 	tempcdb = (struct cdbdata *) mallok(sizeof(struct cdbdata));
 	if (tempcdb == NULL) {
-		lprintf(2, "Cannot allocate memory!\n");
+		lprintf(2, "cdb_fetch: Cannot allocate memory!\n");
+		abort();
 	}
 	tempcdb->len = dret.size;
 	tempcdb->ptr = dret.data;
@@ -459,10 +479,10 @@ void cdb_rewind(int cdb)
 	int ret = 0;
 
 	if (MYCURSOR != NULL)
-		MYCURSOR->c_close(MYCURSOR);
+		cclose(MYCURSOR);
 
 	if (MYTID == NULL) {
-		lprintf(1, "ERROR: cursor use outside transaction\n");
+		lprintf(1, "cdb_rewind: ERROR: cursor use outside transaction\n");
 		abort();
 	}
 
@@ -471,7 +491,8 @@ void cdb_rewind(int cdb)
 	 */
 	ret = dbp[cdb]->cursor(dbp[cdb], MYTID, &MYCURSOR, 0);
 	if (ret) {
-		lprintf(1, "db_cursor: %s\n", db_strerror(ret));
+		lprintf(1, "cdb_rewind: db_cursor: %s\n", db_strerror(ret));
+		abort();
 	}
 }
 
@@ -495,7 +516,12 @@ struct cdbdata *cdb_next_item(int cdb)
 		&key, &data, DB_NEXT);
 	
 	if (ret) {
-		MYCURSOR->c_close(MYCURSOR);
+		if (ret != DB_NOTFOUND) {
+			lprintf(1, "cdb_next_item(%d): %s\n",
+				cdb, db_strerror(ret));
+			abort();
+		}
+		cclose(MYCURSOR);
 		MYCURSOR = NULL;
 		return NULL;		/* presumably, end of file */
 	}
@@ -515,7 +541,7 @@ struct cdbdata *cdb_next_item(int cdb)
 void cdb_begin_transaction(void) {
 
 	if (MYTID != NULL) {	/* FIXME this slows it down, take it out */
-		lprintf(1, "ERROR: opening a new transaction with one already open!\n");
+		lprintf(1, "cdb_begin_transaction: ERROR: opening a new transaction with one already open!\n");
 		abort();
 	}
 	else {
@@ -525,12 +551,16 @@ void cdb_begin_transaction(void) {
 
 void cdb_end_transaction(void) {
 	if (MYCURSOR != NULL) {
-		lprintf(1, "WARNING: cursor still open at transaction end\n");
-		MYCURSOR->c_close(MYCURSOR);
+		lprintf(1, "cdb_end_transaction: WARNING: cursor still open at transaction end\n");
+		cclose(MYCURSOR);
 		MYCURSOR = NULL;
 	}
-	if (MYTID == NULL) lprintf(1, "ERROR: txcommit(NULL) !!\n");
-	else txcommit(MYTID);
+	if (MYTID == NULL) {
+		lprintf(1, "cdb_end_transaction: ERROR: txcommit(NULL) !!\n");
+		abort();
+	} else
+		txcommit(MYTID);
+
 	MYTID = NULL;
 }