traverse: add a flag to enable transferring empty records in cluster wide traverse

This will be useful for also printing information about empty/deleted
records in "ctdb catdb", e.g. for debugging vacuuming issues.

diff --git a/client/ctdb_client.c b/client/ctdb_client.c
index 1d63fa0ad7d9522f336340394027d1cbfd156d1a..307c19ed55360b2ef9de0e31a80aa413d215f9ef 100644 (file)
--- a/client/ctdb_client.c
+++ b/client/ctdb_client.c
@@ -2071,6 +2071,7 @@ int ctdb_traverse(struct ctdb_db_context *ctdb_db, ctdb_traverse_func fn, void *
        t.db_id = ctdb_db->db_id;
        t.srvid = srvid;
        t.reqid = 0;
+       t.withemptyrecords = false;
 
        data.dptr = (uint8_t *)&t;
        data.dsize = sizeof(t);

diff --git a/include/ctdb_protocol.h b/include/ctdb_protocol.h
index f83399c9644de8f1c0c0ee966c0d30dfa632f4ab..f404ec79054fbdae4b3276ac56798e515c1a13cf 100644 (file)
--- a/include/ctdb_protocol.h
+++ b/include/ctdb_protocol.h
@@ -581,6 +581,7 @@ struct ctdb_traverse_start {
        uint32_t db_id;
        uint32_t reqid;
        uint64_t srvid;
+       bool withemptyrecords;
 };
 
 /*

diff --git a/libctdb/ctdb.c b/libctdb/ctdb.c
index 2f694a1c347370d92e1d8949a9254c240d6a0e02..a525d6fb6d4547f781a1f2e683da9eb8185a2676 100644 (file)
--- a/libctdb/ctdb.c
+++ b/libctdb/ctdb.c
@@ -1127,6 +1127,7 @@ static void traverse_msghnd_cb(struct ctdb_connection *ctdb,
        t.db_id = ctdb_db->id;
        t.srvid = state->srvid;
        t.reqid = 0;
+       t.withemptyrecords = false;
 
        state->handle = new_ctdb_control_request(ctdb,
                                CTDB_CONTROL_TRAVERSE_START,

diff --git a/server/ctdb_traverse.c b/server/ctdb_traverse.c
index 59ec9b2c751ca0d37a47377c3ff1e6f70dd787a5..d719f6f4106c93c0330a124880365373c98cd836 100644 (file)
--- a/server/ctdb_traverse.c
+++ b/server/ctdb_traverse.c
@@ -43,6 +43,7 @@ struct ctdb_traverse_local_handle {
        ctdb_traverse_fn_t callback;
        struct timeval start_time;
        struct ctdb_queue *queue;
+       bool withemptyrecords;
 };
 
 /*
@@ -96,7 +97,9 @@ static int ctdb_traverse_local_fn(struct tdb_context *tdb, TDB_DATA key, TDB_DAT
 
        if (h->ctdb_db->persistent == 0) {
                /* filter out zero-length records */
-               if (data.dsize <= sizeof(struct ctdb_ltdb_header)) {
+               if (!h->withemptyrecords &&
+                   data.dsize <= sizeof(struct ctdb_ltdb_header))
+               {
                        return 0;
                }
 
@@ -125,6 +128,7 @@ struct traverse_all_state {
        uint32_t srcnode;
        uint32_t client_reqid;
        uint64_t srvid;
+       bool withemptyrecords;
 };
 
 /*
@@ -167,6 +171,7 @@ static struct ctdb_traverse_local_handle *ctdb_traverse_local(struct ctdb_db_con
        h->ctdb_db = ctdb_db;
        h->client_reqid = all_state->client_reqid;
        h->srvid = all_state->srvid;
+       h->withemptyrecords = all_state->withemptyrecords;
 
        if (h->child == 0) {
                /* start the traverse in the child */
@@ -227,6 +232,7 @@ struct ctdb_traverse_all {
        uint32_t pnn;
        uint32_t client_reqid;
        uint64_t srvid;
+       bool withemptyrecords;
 };
 
 /* called when a traverse times out */
@@ -249,6 +255,7 @@ struct traverse_start_state {
        uint32_t reqid;
        uint32_t db_id;
        uint64_t srvid;
+       bool withemptyrecords;
 };
 
 
@@ -290,6 +297,7 @@ static struct ctdb_traverse_all_handle *ctdb_daemon_traverse_all(struct ctdb_db_
        r.pnn   = ctdb->pnn;
        r.client_reqid = start_state->reqid;
        r.srvid = start_state->srvid;
+       r.withemptyrecords = start_state->withemptyrecords;
 
        data.dptr = (uint8_t *)&r;
        data.dsize = sizeof(r);
@@ -411,6 +419,7 @@ int32_t ctdb_control_traverse_all(struct ctdb_context *ctdb, TDB_DATA data, TDB_
        state->ctdb = ctdb;
        state->client_reqid = c->client_reqid;
        state->srvid = c->srvid;
+       state->withemptyrecords = c->withemptyrecords;
 
        state->h = ctdb_traverse_local(ctdb_db, traverse_all_callback, state);
        if (state->h == NULL) {
@@ -594,6 +603,7 @@ int32_t ctdb_control_traverse_start(struct ctdb_context *ctdb, TDB_DATA data,
        state->srvid = d->srvid;
        state->db_id = d->db_id;
        state->ctdb = ctdb;
+       state->withemptyrecords = d->withemptyrecords;
 
        state->h = ctdb_daemon_traverse_all(ctdb_db, traverse_start_callback, state);
        if (state->h == NULL) {