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.
}
if (data->dsize != 0) {
}
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);
if (lmaster == ctdb->pnn) {
ctdb_send_dmaster_reply(ctdb_db, header, *key, *data,
c->hdr.srcnode, c->hdr.reqid);
memcpy(&r->data[key->dsize], data->dptr, data->dsize);
header->dmaster = c->hdr.srcnode;
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);
ctdb_queue_packet(ctdb, &r->hdr);
talloc_free(r);
header.rsn = rsn + 1;
header.dmaster = ctdb->pnn;
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");
if (ctdb_ltdb_store(ctdb_db, key, &header, data) != 0) {
ctdb_fatal(ctdb, "ctdb_reply_dmaster store failed\n");
+ 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),
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),