tdb: handle processes dying during transaction commit.

tdb transactions were designed to be robust against the machine
powering off, but interestingly were never designed to handle the case
where an administrator kill -9's a process during commit.  Because
recovery is only done on tdb_open, processes with the tdb already
mapped will simply use it despite it being corrupt and needing
recovery.

The solution to this is to check for recovery every time we grab a
data lock: we could have gained the lock because a process just died.
This has no measurable cost: here is the time for tdbtorture -s 0 -n 1
-l 10000:

Before:
	2.75 2.50 2.81 3.19 2.91 2.53 2.72 2.50 2.78 2.77 = Avg 2.75

After:
	2.81 2.57 3.42 2.49 3.02 2.49 2.84 2.48 2.80 2.43 = Avg 2.74

Signed-off-by: Rusty Russell <rusty@rustcorp.com.au>

diff --git a/ccan/tdb/lock.c b/ccan/tdb/lock.c
index fd451e58cd314b54167b476e8a74890aa8e23581..2a6f1230c9d7308b83592771ec03de5d7f5a95f1 100644 (file)
--- a/ccan/tdb/lock.c
+++ b/ccan/tdb/lock.c
@@ -309,10 +309,44 @@ int tdb_nest_lock(struct tdb_context *tdb, uint32_t offset, int ltype,
        return 0;
 }
 
+static int tdb_lock_and_recover(struct tdb_context *tdb)
+{
+       int ret;
+
+       /* We need to match locking order in transaction commit. */
+       if (tdb_brlock(tdb, F_WRLCK, FREELIST_TOP, 0, TDB_LOCK_WAIT)) {
+               return -1;
+       }
+
+       if (tdb_brlock(tdb, F_WRLCK, OPEN_LOCK, 1, TDB_LOCK_WAIT)) {
+               tdb_brunlock(tdb, F_WRLCK, FREELIST_TOP, 0);
+               return -1;
+       }
+
+       ret = tdb_transaction_recover(tdb);
+
+       tdb_brunlock(tdb, F_WRLCK, OPEN_LOCK, 1);
+       tdb_brunlock(tdb, F_WRLCK, FREELIST_TOP, 0);
+
+       return ret;
+}
+
+static bool have_data_locks(const struct tdb_context *tdb)
+{
+       unsigned int i;
+
+       for (i = 0; i < tdb->num_lockrecs; i++) {
+               if (tdb->lockrecs[i].off >= lock_offset(-1))
+                       return true;
+       }
+       return false;
+}
+
 static int tdb_lock_list(struct tdb_context *tdb, int list, int ltype,
                         enum tdb_lock_flags waitflag)
 {
        int ret;
+       bool check = false;
 
        /* a allrecord lock allows us to avoid per chain locks */
        if (tdb->allrecord_lock.count &&
@@ -324,7 +358,18 @@ static int tdb_lock_list(struct tdb_context *tdb, int list, int ltype,
                tdb->ecode = TDB_ERR_LOCK;
                ret = -1;
        } else {
+               /* Only check when we grab first data lock. */
+               check = !have_data_locks(tdb);
                ret = tdb_nest_lock(tdb, lock_offset(list), ltype, waitflag);
+
+               if (ret == 0 && check && tdb_needs_recovery(tdb)) {
+                       tdb_nest_unlock(tdb, lock_offset(list), ltype, false);
+
+                       if (tdb_lock_and_recover(tdb) == -1) {
+                               return -1;
+                       }
+                       return tdb_lock_list(tdb, list, ltype, waitflag);
+               }
        }
        return ret;
 }
@@ -488,6 +533,21 @@ int tdb_allrecord_lock(struct tdb_context *tdb, int ltype,
        tdb->allrecord_lock.ltype = upgradable ? F_WRLCK : ltype;
        tdb->allrecord_lock.off = upgradable;
 
+       if (tdb_needs_recovery(tdb)) {
+               bool mark = flags & TDB_LOCK_MARK_ONLY;
+               tdb_allrecord_unlock(tdb, ltype, mark);
+               if (mark) {
+                       tdb->ecode = TDB_ERR_LOCK;
+                       TDB_LOG((tdb, TDB_DEBUG_ERROR,
+                                "tdb_lockall_mark cannot do recovery\n"));
+                       return -1;
+               }
+               if (tdb_lock_and_recover(tdb) == -1) {
+                       return -1;
+               }
+               return tdb_allrecord_lock(tdb, ltype, flags, upgradable);
+       }
+
        return 0;
 }
 

diff --git a/ccan/tdb/tdb_private.h b/ccan/tdb/tdb_private.h
index a31bf363887a7e82511efc3d114d550ef1aab1d1..1195c3b557ac65ede35e30caf417516f5c2a3c0e 100644 (file)
--- a/ccan/tdb/tdb_private.h
+++ b/ccan/tdb/tdb_private.h
@@ -282,6 +282,7 @@ int tdb_ofs_read(struct tdb_context *tdb, tdb_off_t offset, tdb_off_t *d);
 int tdb_ofs_write(struct tdb_context *tdb, tdb_off_t offset, tdb_off_t *d);
 int tdb_lock_record(struct tdb_context *tdb, tdb_off_t off);
 int tdb_unlock_record(struct tdb_context *tdb, tdb_off_t off);
+bool tdb_needs_recovery(struct tdb_context *tdb);
 int tdb_rec_read(struct tdb_context *tdb, tdb_off_t offset, struct tdb_record *rec);
 int tdb_rec_write(struct tdb_context *tdb, tdb_off_t offset, struct tdb_record *rec);
 int tdb_do_delete(struct tdb_context *tdb, tdb_off_t rec_ptr, struct tdb_record *rec);

diff --git a/ccan/tdb/test/external-transaction.c b/ccan/tdb/test/external-transaction.c
index 0a0b4b87685f03803c4b66e01bd172e818c5086d..23d8471386413f39a2d7a7d311ad13c3e22b0021 100644 (file)
--- a/ccan/tdb/test/external-transaction.c
+++ b/ccan/tdb/test/external-transaction.c
@@ -76,11 +76,7 @@ static int do_operation(enum operation op, const char *name)
        } else if (op == CHECK_KEEP_OPENED) {
                return tdb_check(tdb, NULL, 0) == 0;
        } else if (op == NEEDS_RECOVERY_KEEP_OPENED) {
-#if 0
-               return tdb_maybe_needs_recovery(tdb);
-#else
-               return 0;
-#endif
+               return tdb_needs_recovery(tdb);
        }
 
        alarmed = 0;

diff --git a/ccan/tdb/transaction.c b/ccan/tdb/transaction.c
index 9cbbcc9087fe1d441c4a829b7c121d560d57146a..11194773647e255f1a7c8743cd69d5c85bfeda59 100644 (file)
--- a/ccan/tdb/transaction.c
+++ b/ccan/tdb/transaction.c
@@ -1194,3 +1194,28 @@ int tdb_transaction_recover(struct tdb_context *tdb)
        /* all done */
        return 0;
 }
+
+/* Any I/O failures we say "needs recovery". */
+bool tdb_needs_recovery(struct tdb_context *tdb)
+{
+       tdb_off_t recovery_head;
+       struct tdb_record rec;
+
+       /* find the recovery area */
+       if (tdb_ofs_read(tdb, TDB_RECOVERY_HEAD, &recovery_head) == -1) {
+               return true;
+       }
+
+       if (recovery_head == 0) {
+               /* we have never allocated a recovery record */
+               return false;
+       }
+
+       /* read the recovery record */
+       if (tdb->methods->tdb_read(tdb, recovery_head, &rec,
+                                  sizeof(rec), DOCONV()) == -1) {
+               return true;
+       }
+
+       return (rec.magic == TDB_RECOVERY_MAGIC);
+}