merge from tridge

(This used to be ctdb commit bf7bf6b2c03548b82dde262009d02630d27e9cae)

ctdb/lib/tdb/common/tdb.c

@@ -715,6 +715,11 @@ int tdb_wipe_all(struct tdb_context *tdb)
 		goto failed;
 	}
 
+	if (tdb_ofs_write(tdb, TDB_RECOVERY_HEAD, &offset) == -1) {
+		TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_wipe_all: failed to write recovery head\n"));
+		goto failed;		
+	}
+
 	/* add all the rest of the file to the freelist */
 	data_len = (tdb->map_size - TDB_DATA_START(tdb->header.hash_size)) - sizeof(struct list_struct);
 	if (data_len > 0) {
@@ -738,3 +743,45 @@ failed:
 	tdb_unlockall(tdb);
 	return -1;
 }
+
+
+/* 
+   validate the integrity of all tdb hash chains. Useful when debugging
+ */
+int tdb_validate(struct tdb_context *tdb)
+{
+	int h;
+	for (h=-1;h<(int)tdb->header.hash_size;h++) {
+		tdb_off_t rec_ptr;
+		uint32_t count = 0;
+		if (tdb_ofs_read(tdb, TDB_HASH_TOP(h), &rec_ptr) == -1) {
+			TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_validate: failed ofs_read at top of hash %d\n", h));
+			return -1;
+		}
+		while (rec_ptr) {
+			struct list_struct r;
+			tdb_off_t size;
+
+			if (tdb_rec_read(tdb, rec_ptr, &r) == -1) {
+				TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_validate: failed rec_read h=%d rec_ptr=%u count=%u\n",
+					 h, rec_ptr, count));
+				return -1;
+			}
+			if (tdb_ofs_read(tdb, rec_ptr + sizeof(r) + r.rec_len - sizeof(tdb_off_t), &size) == -1) {
+				TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_validate: failed ofs_read h=%d rec_ptr=%u count=%u\n",
+					 h, rec_ptr, count));
+				return -1;
+			}
+			if (size != r.rec_len + sizeof(r)) {
+				TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_validate: failed size check size=%u h=%d rec_ptr=%u count=%u\n",
+					 size, h, rec_ptr, count));
+				return -1;
+			}
+			rec_ptr = r.next;
+			count++;
+		}		
+	}
+	return 0;
+}
+
+

ctdb/lib/tdb/common/transaction.c

@@ -87,6 +87,7 @@
 
 */
 
+
 /*
   hold the context of any current transaction
 */
@@ -280,6 +281,53 @@ fail:
 	return -1;
 }
 
+
+/*
+  write while in a transaction - this varient never expands the transaction blocks, it only
+  updates existing blocks. This means it cannot change the recovery size
+*/
+static int transaction_write_existing(struct tdb_context *tdb, tdb_off_t off, 
+				      const void *buf, tdb_len_t len)
+{
+	uint32_t blk;
+
+	/* break it up into block sized chunks */
+	while (len + (off % tdb->transaction->block_size) > tdb->transaction->block_size) {
+		tdb_len_t len2 = tdb->transaction->block_size - (off % tdb->transaction->block_size);
+		if (transaction_write_existing(tdb, off, buf, len2) != 0) {
+			return -1;
+		}
+		len -= len2;
+		off += len2;
+		if (buf != NULL) {
+			buf = (const void *)(len2 + (const char *)buf);
+		}
+	}
+
+	if (len == 0) {
+		return 0;
+	}
+
+	blk = off / tdb->transaction->block_size;
+	off = off % tdb->transaction->block_size;
+
+	if (tdb->transaction->num_blocks <= blk ||
+	    tdb->transaction->blocks[blk] == NULL) {
+		return 0;
+	}
+
+	if (blk == tdb->transaction->num_blocks-1 &&
+	    off + len > tdb->transaction->last_block_size) {
+		len = tdb->transaction->last_block_size - off;
+	}
+
+	/* overwrite part of an existing block */
+	memcpy(tdb->transaction->blocks[blk] + off, buf, len);
+
+	return 0;
+}
+
+
 /*
   accelerated hash chain head search, using the cached hash heads
 */
@@ -629,6 +677,10 @@ static int tdb_recovery_allocate(struct tdb_context *tdb,
 		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_recovery_allocate: failed to write recovery head\n"));
 		return -1;
 	}
+	if (transaction_write_existing(tdb, TDB_RECOVERY_HEAD, &recovery_head, sizeof(tdb_off_t)) == -1) {
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_recovery_allocate: failed to write recovery head\n"));
+		return -1;
+	}
 
 	return 0;
 }
@@ -726,6 +778,12 @@ static int transaction_setup_recovery(struct tdb_context *tdb,
 		tdb->ecode = TDB_ERR_IO;
 		return -1;
 	}
+	if (transaction_write_existing(tdb, recovery_offset, data, sizeof(*rec) + recovery_size) == -1) {
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_setup_recovery: failed to write secondary recovery data\n"));
+		free(data);
+		tdb->ecode = TDB_ERR_IO;
+		return -1;
+	}
 
 	/* as we don't have ordered writes, we have to sync the recovery
 	   data before we update the magic to indicate that the recovery
@@ -747,6 +805,11 @@ static int transaction_setup_recovery(struct tdb_context *tdb,
 		tdb->ecode = TDB_ERR_IO;
 		return -1;
 	}
+	if (transaction_write_existing(tdb, *magic_offset, &magic, sizeof(magic)) == -1) {
+		TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_setup_recovery: failed to write secondary recovery magic\n"));
+		tdb->ecode = TDB_ERR_IO;
+		return -1;
+	}
 
 	/* ensure the recovery magic marker is on disk */
 	if (transaction_sync(tdb, *magic_offset, sizeof(magic)) == -1) {
@@ -778,6 +841,7 @@ int tdb_transaction_commit(struct tdb_context *tdb)
 		return -1;
 	}
 
+
 	if (tdb->transaction->nesting != 0) {
 		tdb->transaction->nesting--;
 		return 0;
@@ -916,6 +980,7 @@ int tdb_transaction_commit(struct tdb_context *tdb)
 	/* use a transaction cancel to free memory and remove the
 	   transaction locks */
 	tdb_transaction_cancel(tdb);
+
 	return 0;
 }
 

ctdb/lib/tdb/include/tdb.h

@@ -157,6 +157,7 @@ int tdb_printfreelist(struct tdb_context *tdb);
 int tdb_validate_freelist(struct tdb_context *tdb, int *pnum_entries);
 int tdb_wipe_all(struct tdb_context *tdb);
 int tdb_freelist_size(struct tdb_context *tdb);
+int tdb_validate(struct tdb_context *tdb);
 
 extern TDB_DATA tdb_null;
 

ctdb/server/ctdb_recoverd.c

@@ -756,7 +756,13 @@ static void vacuum_fetch_next(struct vacuum_info *v)
 		}
 
 		data = tdb_fetch(v->ctdb_db->ltdb->tdb, call.key);
-		if (data.dptr == NULL || data.dsize < sizeof(struct ctdb_ltdb_header)) {
+		if (data.dptr == NULL) {
+			tdb_chainunlock(v->ctdb_db->ltdb->tdb, call.key);
+			continue;
+		}
+
+		if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
+			free(data.dptr);
 			tdb_chainunlock(v->ctdb_db->ltdb->tdb, call.key);
 			continue;
 		}
@@ -764,10 +770,13 @@ static void vacuum_fetch_next(struct vacuum_info *v)
 		hdr = (struct ctdb_ltdb_header *)data.dptr;
 		if (hdr->dmaster == v->rec->ctdb->pnn) {
 			/* its already local */
+			free(data.dptr);
 			tdb_chainunlock(v->ctdb_db->ltdb->tdb, call.key);
 			continue;
 		}
 
+		free(data.dptr);
+
 		state = ctdb_call_send(v->ctdb_db, &call);
 		tdb_chainunlock(v->ctdb_db->ltdb->tdb, call.key);
 		if (state == NULL) {