ctdb_ltdb_store_server: delete an empty record that is safe to delete instead of...

When storing a record that is being migrated off to another node
and has never been migrated with data, then we can safely delete it
from the local tdb instead of storing the record with empty data.

Note: This record is not deleted if we are its lmaster or dmaster.

Pair-Programmed-With: Stefan Metzmacher <metze@samba.org>

diff --git a/server/ctdb_ltdb_server.c b/server/ctdb_ltdb_server.c
index 43a1bf600f6129da999aac06322cd1120d21801b..5f9f4636de71a7965813285f80f086835f63b7a0 100644 (file)
--- a/server/ctdb_ltdb_server.c
+++ b/server/ctdb_ltdb_server.c
@@ -66,6 +66,8 @@ static int ctdb_ltdb_store_server(struct ctdb_db_context *ctdb_db,
        TDB_DATA rec;
        int ret;
        bool seqnum_suppressed = false;
+       bool keep = false;
+       uint32_t lmaster = ctdb_lmaster(ctdb_db->ctdb, &key);
 
        if (ctdb->flags & CTDB_FLAG_TORTURE) {
                struct ctdb_ltdb_header *h2;
@@ -78,6 +80,23 @@ static int ctdb_ltdb_store_server(struct ctdb_db_context *ctdb_db,
                if (rec.dptr) free(rec.dptr);
        }
 
+       /*
+        * If we migrate an empty record off to another node
+        * and the record has not been migrated with data,
+        * delete the record instead of storing the empty record.
+        */
+       if (data.dsize != 0) {
+               keep = true;
+       } else if (ctdb_db->persistent) {
+               keep = true;
+       } else if (ctdb_db->ctdb->pnn == lmaster) {
+               keep = true;
+       } else if (ctdb_db->ctdb->pnn == header->dmaster) {
+               keep = true;
+       } else if (header->flags & CTDB_REC_FLAG_MIGRATED_WITH_DATA) {
+               keep = true;
+       }
+
        rec.dsize = sizeof(*header) + data.dsize;
        rec.dptr = talloc_size(ctdb, rec.dsize);
        CTDB_NO_MEMORY(ctdb, rec.dptr);
@@ -100,7 +119,18 @@ static int ctdb_ltdb_store_server(struct ctdb_db_context *ctdb_db,
                }
                if (old.dptr) free(old.dptr);
        }
-       ret = tdb_store(ctdb_db->ltdb->tdb, key, rec, TDB_REPLACE);
+
+       DEBUG(DEBUG_DEBUG, (__location__ " db[%s]: %s record: hash[0x%08x]\n",
+                           ctdb_db->db_name,
+                           keep?"storing":"deleting",
+                           ctdb_hash(&key)));
+
+       if (keep) {
+               ret = tdb_store(ctdb_db->ltdb->tdb, key, rec, TDB_REPLACE);
+       } else {
+               ret = tdb_delete(ctdb_db->ltdb->tdb, key);
+       }
+
        if (ret != 0) {
                DEBUG(DEBUG_ERR, (__location__ " Failed to store dynamic data\n"));
        }