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

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>

(This used to be ctdb commit 3cca0d4b48325d86de2cb0b44bb7811a30701352)

1 files changed, 31 insertions, 1 deletions

diff --git a/ctdb/server/ctdb_ltdb_server.c b/ctdb/server/ctdb_ltdb_server.c
index c00a048246e..b5eb712a450 100644
--- a/ctdb/server/ctdb_ltdb_server.c
+++ b/ctdb/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"));
 	}