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)

diff --git a/ccan/tdb2/transaction.c b/ccan/tdb2/transaction.c
index d9a1bc050ef6a4415bd69647d3dc9e6ce59573ac..805ea0948aad0e0ddfa11289aadc26334cfc8a44 100644 (file)
--- a/ccan/tdb2/transaction.c
+++ b/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;
        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) {
 
        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);
        }
 
                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);
        /* 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;
                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;
 
                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) {
                }
 
                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;
                }
                }
                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. */
                }
 
                /* 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;
                }
                        off += len + samelen;
                        offset += len + samelen;
                }

+               tdb_access_release(tdb, buffer);

        }
 
        *len = p - (unsigned char *)(rec + 1);
        }
 
        *len = p - (unsigned char *)(rec + 1);

+       tdb->methods = old_methods;

        return rec;
        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,
 }
 
 static tdb_off_t create_recovery_area(struct tdb_context *tdb,