tdb2: use direct access functions when creating recovery blob

We don't need to copy into a buffer to examine the old data: in the
common case, it's mmaped already.  It's made a bit trickier because
the tdb_access_read() function uses the current I/O methods, so we
need to restore that temporarily.

The difference was in the noise, however (the sync no-doubt
dominates).

Before:
$ time ./growtdb-bench 250000 10 > /dev/null && ls -l /tmp/growtdb.tdb && time ./tdbtorture -s 0 && ls -l torture.tdb && ./speed --transaction 2000000
real	0m45.021s
user	0m16.261s
sys	0m2.432s
-rw------- 1 rusty rusty 364469344 2011-04-27 22:55 /tmp/growtdb.tdb
testing with 3 processes, 5000 loops, seed=0
OK

real	1m10.144s
user	0m0.480s
sys	0m0.460s
-rw------- 1 rusty rusty 391992 2011-04-27 22:56 torture.tdb
Adding 2000000 records:  863 ns (110601144 bytes)
Finding 2000000 records:  565 ns (110601144 bytes)
Missing 2000000 records:  383 ns (110601144 bytes)
Traversing 2000000 records:  409 ns (110601144 bytes)
Deleting 2000000 records:  676 ns (225354680 bytes)
Re-adding 2000000 records:  784 ns (225354680 bytes)
Appending 2000000 records:  1191 ns (247890168 bytes)
Churning 2000000 records:  2166 ns (423133432 bytes)

After:
real	0m47.141s
user	0m16.073s
sys	0m2.460s
-rw------- 1 rusty rusty 364469344 2011-04-27 22:58 /tmp/growtdb.tdb
testing with 3 processes, 5000 loops, seed=0
OK

real	1m4.207s
user	0m0.416s
sys	0m0.504s
-rw------- 1 rusty rusty 313576 2011-04-27 22:59 torture.tdb
Adding 2000000 records:  874 ns (110601144 bytes)
Finding 2000000 records:  565 ns (110601144 bytes)
Missing 2000000 records:  393 ns (110601144 bytes)
Traversing 2000000 records:  404 ns (110601144 bytes)
Deleting 2000000 records:  684 ns (225354680 bytes)
Re-adding 2000000 records:  792 ns (225354680 bytes)
Appending 2000000 records:  1212 ns (247890168 bytes)
Churning 2000000 records:  2191 ns (423133432 bytes)

ccan/tdb2/transaction.c

@@ -711,7 +711,7 @@ static struct tdb_recovery_record *alloc_recovery(struct tdb_context *tdb,
 	size_t i;
 	enum TDB_ERROR ecode;
 	unsigned char *p;
-	const struct tdb_methods *methods = tdb->transaction->io_methods;
+	const struct tdb_methods *old_methods = tdb->methods;
 
 	rec = malloc(sizeof(*rec) + tdb_recovery_size(tdb));
 	if (!rec) {
@@ -721,6 +721,10 @@ static struct tdb_recovery_record *alloc_recovery(struct tdb_context *tdb,
 		return TDB_ERR_PTR(TDB_ERR_OOM);
 	}
 
+	/* We temporarily revert to the old I/O methods, so we can use
+	 * tdb_access_read */
+	tdb->methods = tdb->transaction->io_methods;
+
 	/* build the recovery data into a single blob to allow us to do a single
 	   large write, which should be more efficient */
 	p = (unsigned char *)(rec + 1);
@@ -728,7 +732,7 @@ static struct tdb_recovery_record *alloc_recovery(struct tdb_context *tdb,
 		tdb_off_t offset;
 		tdb_len_t length;
 		unsigned int off;
-		unsigned char buffer[PAGESIZE];
+		const unsigned char *buffer;
 
 		if (tdb->transaction->blocks[i] == NULL) {
 			continue;
@@ -745,21 +749,20 @@ static struct tdb_recovery_record *alloc_recovery(struct tdb_context *tdb,
 		}
 
 		if (offset + length > tdb->file->map_size) {
-			free(rec);
-			tdb_logerr(tdb, TDB_ERR_CORRUPT, TDB_LOG_ERROR,
-				   "tdb_transaction_setup_recovery:"
-				   " transaction data over new region"
-				   " boundary");
-			return TDB_ERR_PTR(TDB_ERR_CORRUPT);
+			ecode = tdb_logerr(tdb, TDB_ERR_CORRUPT, TDB_LOG_ERROR,
+					   "tdb_transaction_setup_recovery:"
+					   " transaction data over new region"
+					   " boundary");
+			goto fail;
 		}
 		if (offset + length > tdb->transaction->old_map_size) {
 			/* Short read at EOF. */
 			length = tdb->transaction->old_map_size - offset;
 		}
-		ecode = methods->tread(tdb, offset, buffer, length);
-		if (ecode != TDB_SUCCESS) {
-			free(rec);
-			return TDB_ERR_PTR(ecode);
+		buffer = tdb_access_read(tdb, offset, length, false);
+		if (TDB_PTR_IS_ERR(buffer)) {
+			ecode = TDB_PTR_ERR(buffer);
+			goto fail;
 		}
 
 		/* Skip over anything the same at the start. */
@@ -784,10 +787,17 @@ static struct tdb_recovery_record *alloc_recovery(struct tdb_context *tdb,
 			off += len + samelen;
 			offset += len + samelen;
 		}
+		tdb_access_release(tdb, buffer);
 	}
 
 	*len = p - (unsigned char *)(rec + 1);
+	tdb->methods = old_methods;
 	return rec;
+
+fail:
+	free(rec);
+	tdb->methods = old_methods;
+	return TDB_ERR_PTR(ecode);
 }
 
 static tdb_off_t create_recovery_area(struct tdb_context *tdb,