Leave sequence number alone when merely migrating records.

(Based on earlier version from Ronnie which modified tdb; this one
is standalone).

When storing records in a tdb that has "automatic seqnum updates"
also check if the actual data for the record has changed or not.

If it has not changed at all, except for possibly the header,
this is likely just a dmaster migration operation in which case
we want to write the record to the tdb but we do not want the tdb
sequence number to be increased.

This resolves the problem of notify.tdb being thrashed under load:
the heuristic in smbd to only reread this when the sequence number
increases (rarely) breaks down.

Before, running nbench --num-progs=512 across 4 nodes, we saw numbers like:
 512      1496  118.33 MB/sec  execute 60 sec  latency 0.00 msec
And turning on latency tracking, this was typical in the logs:
 ctdbd: High latency 9380914.000000s for operation lockwait on database notify.tdb

After this commit:
  512      2451  143.85 MB/sec  execute 60 sec  latency 0.00 msec
And no more latency messages...

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

diff --git a/common/ctdb_ltdb.c b/common/ctdb_ltdb.c
index 12fcf52d959f7d238261e90b9a0e7a1254d10c66..b2fd189943251fc3db6fe93b9c663ef67d63e9e5 100644 (file)
--- a/common/ctdb_ltdb.c
+++ b/common/ctdb_ltdb.c
@@ -128,6 +128,7 @@ int ctdb_ltdb_store(struct ctdb_db_context *ctdb_db, TDB_DATA key,
        struct ctdb_context *ctdb = ctdb_db->ctdb;
        TDB_DATA rec;
        int ret;
+       bool seqnum_suppressed = false;
 
        if (ctdb->flags & CTDB_FLAG_TORTURE) {
                struct ctdb_ltdb_header *h2;
@@ -147,10 +148,28 @@ int ctdb_ltdb_store(struct ctdb_db_context *ctdb_db, TDB_DATA key,
        memcpy(rec.dptr, header, sizeof(*header));
        memcpy(rec.dptr + sizeof(*header), data.dptr, data.dsize);
 
+       /* Databases with seqnum updates enabled only get their seqnum
+          changes when/if we modify the data */
+       if (ctdb_db->seqnum_update != NULL) {
+               TDB_DATA old;
+               old = tdb_fetch(ctdb_db->ltdb->tdb, key);
+
+               if ( (old.dsize == rec.dsize)
+               && !memcmp(old.dptr+sizeof(struct ctdb_ltdb_header),
+                         rec.dptr+sizeof(struct ctdb_ltdb_header),
+                         rec.dsize-sizeof(struct ctdb_ltdb_header)) ) {
+                       tdb_remove_flags(ctdb_db->ltdb->tdb, TDB_SEQNUM);
+                       seqnum_suppressed = true;
+               }
+               if (old.dptr) free(old.dptr);
+       }
        ret = tdb_store(ctdb_db->ltdb->tdb, key, rec, TDB_REPLACE);
        if (ret != 0) {
                DEBUG(DEBUG_ERR, (__location__ " Failed to store dynamic data\n"));
        }
+       if (seqnum_suppressed) {
+               tdb_add_flags(ctdb_db->ltdb->tdb, TDB_SEQNUM);
+       }
 
        talloc_free(rec.dptr);