tdb: handle processes dying during transaction commit.
authorRusty Russell <rusty@rustcorp.com.au>
Wed, 24 Feb 2010 03:38:40 +0000 (14:08 +1030)
committerRusty Russell <rusty@rustcorp.com.au>
Wed, 24 Feb 2010 03:38:40 +0000 (14:08 +1030)
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>
ccan/tdb/lock.c
ccan/tdb/tdb_private.h
ccan/tdb/test/external-transaction.c
ccan/tdb/transaction.c

index fd451e58cd314b54167b476e8a74890aa8e23581..2a6f1230c9d7308b83592771ec03de5d7f5a95f1 100644 (file)
@@ -309,10 +309,44 @@ int tdb_nest_lock(struct tdb_context *tdb, uint32_t offset, int ltype,
        return 0;
 }
 
        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;
 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 &&
 
        /* 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 {
                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);
                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;
 }
        }
        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;
 
        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;
 }
 
        return 0;
 }
 
index a31bf363887a7e82511efc3d114d550ef1aab1d1..1195c3b557ac65ede35e30caf417516f5c2a3c0e 100644 (file)
@@ -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);
 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);
 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);
index 0a0b4b87685f03803c4b66e01bd172e818c5086d..23d8471386413f39a2d7a7d311ad13c3e22b0021 100644 (file)
@@ -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) {
        } 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;
        }
 
        alarmed = 0;
index 9cbbcc9087fe1d441c4a829b7c121d560d57146a..11194773647e255f1a7c8743cd69d5c85bfeda59 100644 (file)
@@ -1194,3 +1194,28 @@ int tdb_transaction_recover(struct tdb_context *tdb)
        /* all done */
        return 0;
 }
        /* 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);
+}