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>
(cherry picked from commit 3cca0d4b48325d86de2cb0b44bb7811a30701352)

Signed-off-by: Stefan Metzmacher <metze@samba.org>

diff --git a/server/ctdb_ltdb_server.c b/server/ctdb_ltdb_server.c
index 72e3122533421c8d2ea854623ae643944b13f6c9..9ce97056be4d92c76289237ca4ad8d582fa36b1a 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"));
        }