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

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

diff --git a/server/ctdb_traverse.c b/server/ctdb_traverse.c
index 60a8e90b1b5d0491e9bf05c03c1838acadd11f64..dcb16b227b2056301b012f571f09a8e1b5cec201 100644 (file)
--- a/server/ctdb_traverse.c
+++ b/server/ctdb_traverse.c
@@ -18,7 +18,7 @@
 */
 
 #include "includes.h"
-#include "lib/events/events.h"
+#include "lib/tevent/tevent.h"
 #include "system/filesys.h"
 #include "system/wait.h"
 #include "db_wrap.h"
@@ -92,7 +92,6 @@ static int ctdb_traverse_local_fn(struct tdb_context *tdb, TDB_DATA key, TDB_DAT
        struct ctdb_rec_data *d;
        struct ctdb_ltdb_header *hdr;
 
-       
        hdr = (struct ctdb_ltdb_header *)data.dptr;
 
        if (h->ctdb_db->persistent == 0) {
@@ -172,11 +171,15 @@ static struct ctdb_traverse_local_handle *ctdb_traverse_local(struct ctdb_db_con
        if (h->child == 0) {
                /* start the traverse in the child */
                close(h->fd[0]);
+               debug_extra = talloc_asprintf(NULL, "traverse_local-%s:",
+                                             ctdb_db->db_name);
                tdb_traverse_read(ctdb_db->ltdb->tdb, ctdb_traverse_local_fn, h);
                _exit(0);
        }
 
        close(h->fd[1]);
+       set_close_on_exec(h->fd[0]);
+
        talloc_set_destructor(h, traverse_local_destructor);
 
        DLIST_ADD(ctdb_db->traverse, h);
@@ -185,7 +188,10 @@ static struct ctdb_traverse_local_handle *ctdb_traverse_local(struct ctdb_db_con
          setup a packet queue between the child and the parent. This
          copes with all the async and packet boundary issues
         */
-       h->queue = ctdb_queue_setup(ctdb_db->ctdb, h, h->fd[0], 0, ctdb_traverse_local_handler, h);
+       DEBUG(DEBUG_DEBUG, (__location__ " Created PIPE FD:%d to child traverse\n", h->fd[0]));
+
+       h->queue = ctdb_queue_setup(ctdb_db->ctdb, h, h->fd[0], 0, ctdb_traverse_local_handler, h,
+                                   "to-ctdbd");
        if (h->queue == NULL) {
                talloc_free(h);
                return NULL;
@@ -229,10 +235,10 @@ static void ctdb_traverse_all_timeout(struct event_context *ev, struct timed_eve
 {
        struct ctdb_traverse_all_handle *state = talloc_get_type(private_data, struct ctdb_traverse_all_handle);
 
-       state->ctdb->statistics.timeouts.traverse++;
+       DEBUG(DEBUG_ERR,(__location__ " Traverse all timeout on database:%s\n", state->ctdb_db->db_name));
+       CTDB_INCREMENT_STAT(state->ctdb, timeouts.traverse);
 
        state->callback(state->private_data, tdb_null, tdb_null);
-       talloc_free(state);
 }
 
 
@@ -265,7 +271,7 @@ static struct ctdb_traverse_all_handle *ctdb_daemon_traverse_all(struct ctdb_db_
        struct ctdb_traverse_all r;
        uint32_t destination;
 
-       state = talloc(ctdb_db, struct ctdb_traverse_all_handle);
+       state = talloc(start_state, struct ctdb_traverse_all_handle);
        if (state == NULL) {
                return NULL;
        }
@@ -385,6 +391,16 @@ int32_t ctdb_control_traverse_all(struct ctdb_context *ctdb, TDB_DATA data, TDB_
                return -1;
        }
 
+       if (ctdb_db->unhealthy_reason) {
+               if (ctdb->tunable.allow_unhealthy_db_read == 0) {
+                       DEBUG(DEBUG_ERR,("db(%s) unhealty in ctdb_control_traverse_all: %s\n",
+                                       ctdb_db->db_name, ctdb_db->unhealthy_reason));
+                       return -1;
+               }
+               DEBUG(DEBUG_WARNING,("warn: db(%s) unhealty in ctdb_control_traverse_all: %s\n",
+                                    ctdb_db->db_name, ctdb_db->unhealthy_reason));
+       }
+
        state = talloc(ctdb_db, struct traverse_all_state);
        if (state == NULL) {
                return -1;
@@ -450,11 +466,6 @@ int32_t ctdb_control_traverse_data(struct ctdb_context *ctdb, TDB_DATA data, TDB
        private_data = state->private_data;
 
        callback(private_data, key, data);
-       if (key.dsize == 0 && data.dsize == 0) {
-               /* we've received all of the null replies, so all
-                  nodes are finished */
-               talloc_free(state);
-       }
        return 0;
 }      
 
@@ -495,6 +506,7 @@ static int ctdb_traverse_start_destructor(struct traverse_start_state *state)
        struct ctdb_traverse_start r;
        TDB_DATA data;
 
+       DEBUG(DEBUG_ERR,(__location__ " Traverse cancelled by client disconnect for database:0x%08x\n", state->db_id));
        r.db_id = state->db_id;
        r.reqid = state->reqid;
        r.srvid = state->srvid;
@@ -562,6 +574,16 @@ int32_t ctdb_control_traverse_start(struct ctdb_context *ctdb, TDB_DATA data,
                return -1;
        }
 
+       if (ctdb_db->unhealthy_reason) {
+               if (ctdb->tunable.allow_unhealthy_db_read == 0) {
+                       DEBUG(DEBUG_ERR,("db(%s) unhealty in ctdb_control_traverse_start: %s\n",
+                                       ctdb_db->db_name, ctdb_db->unhealthy_reason));
+                       return -1;
+               }
+               DEBUG(DEBUG_WARNING,("warn: db(%s) unhealty in ctdb_control_traverse_start: %s\n",
+                                    ctdb_db->db_name, ctdb_db->unhealthy_reason));
+       }
+
        state = talloc(client, struct traverse_start_state);
        if (state == NULL) {
                return -1;