Add a new header flag for "migrated with data" and set this to 1

[sahlberg/ctdb.git] / server / ctdb_call.c

diff --git a/server/ctdb_call.c b/server/ctdb_call.c
index ed741ed0a9d5f4cd7a6b9bf26836b4ee8760265d..a675e7f2eb058abaf35cbd26987bfc6b3fc2c50c 100644 (file)
--- 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),