From: Ronnie Sahlberg <ronniesahlberg@gmail.com>
Date: Mon, 6 Dec 2010 05:09:38 +0000 (+1100)
Subject: Add a new header flag for "migrated with data" and set this to 1
X-Git-Url: http://git.samba.org/?p=metze%2Fctdb%2Fwip.git;a=commitdiff_plain;h=a8cc35191df1cd4b866897df71d317ce5f198cb5

Add a new header flag for "migrated with data" and set this to 1
when we migrate a non-empty record onto the node
or a non-empty record off the node

When we migrate a record back to the lmaster and yield the dmaster role,
inspect this flag if if it is still not set, we can delete the record from
the local database as soon as we have migrated it back to the lmaster.
---

diff --git a/server/ctdb_call.c b/server/ctdb_call.c
index f3d93a8d..7bc4c351 100644
--- a/server/ctdb_call.c
+++ b/server/ctdb_call.c
@@ -202,9 +202,10 @@ static void ctdb_call_send_dmaster(struct ctdb_db_context *ctdb_db,
 	}
 
 	if (data->dsize != 0) {
-		header->flags &= CTDB_REC_FLAG_MIGRATED_WITH_DATA;
+		header->flags |= CTDB_REC_FLAG_MIGRATED_WITH_DATA;
 	}
 
+
 	if (lmaster == ctdb->pnn) {
 		ctdb_send_dmaster_reply(ctdb_db, header, *key, *data, 
 					c->hdr.srcnode, c->hdr.reqid);
@@ -226,10 +227,19 @@ static void ctdb_call_send_dmaster(struct ctdb_db_context *ctdb_db,
 	memcpy(&r->data[key->dsize], data->dptr, data->dsize);
 
 	header->dmaster = c->hdr.srcnode;
-	if (ctdb_ltdb_store(ctdb_db, *key, header, *data) != 0) {
-		ctdb_fatal(ctdb, "Failed to store record in ctdb_call_send_dmaster");
+
+	if (data->dsize == 0
+	&& lmaster != ctdb->pnn
+	&& (header->flags & CTDB_REC_FLAG_MIGRATED_WITH_DATA) == 0) {
+		if (ctdb_ltdb_delete(ctdb_db, *key) != 0) {
+			ctdb_fatal(ctdb, "Failed to delete empty record when migrating it off the node");
+		}
+	} else {
+		if (ctdb_ltdb_store(ctdb_db, *key, header, *data) != 0) {
+			ctdb_fatal(ctdb, "Failed to store record in ctdb_call_send_dmaster");
+		}
 	}
-	
+
 	ctdb_queue_packet(ctdb, &r->hdr);
 
 	talloc_free(r);
@@ -257,6 +267,10 @@ static void ctdb_become_dmaster(struct ctdb_db_context *ctdb_db,
 	header.rsn = rsn + 1;
 	header.dmaster = ctdb->pnn;
 
+	if (data.dsize != 0) {
+		header.flags |= CTDB_REC_FLAG_MIGRATED_WITH_DATA;
+	}
+
 	if (ctdb_ltdb_store(ctdb_db, key, &header, data) != 0) {
 		ctdb_fatal(ctdb, "ctdb_reply_dmaster store failed\n");
 
@@ -355,6 +369,10 @@ void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr
 		return;
 	}
 
+	if (data.dsize != 0) {
+		header.flags |= CTDB_REC_FLAG_MIGRATED_WITH_DATA;
+	}
+
 	if (ctdb_lmaster(ctdb, &key) != ctdb->pnn) {
 		DEBUG(DEBUG_ALERT,("pnn %u dmaster request to non-lmaster lmaster=%u gen=%u curgen=%u\n",
 			 ctdb->pnn, ctdb_lmaster(ctdb, &key),