merge from tridge

(This used to be ctdb commit a84e9b47a87fc7d4756b4a179aa2ea0bc7c54c78)

10 files changed, 277 insertions, 133 deletions

diff --git a/ctdb/Makefile.in b/ctdb/Makefile.in
index a80908c638..ff53535c90 100644
--- a/ctdb/Makefile.in
+++ b/ctdb/Makefile.in
@@ -108,8 +108,14 @@ distclean: clean
 install: all
 	mkdir -p $(bindir)
 	mkdir -p $(includedir)
-	cp $(BINS) $(bindir)
-	cp $(srcdir)/include/ctdb.h $(includedir)
+	rsync $(BINS) $(bindir)
+	rsync $(srcdir)/include/ctdb.h $(includedir)
+
+test:
+	tests/run_tests.sh
+
+valgrindtest:
+	VALGRIND="valgrind -q --trace-children=yes" tests/run_tests.sh
 
 test:
 	tests/run_tests.sh
diff --git a/ctdb/common/ctdb.c b/ctdb/common/ctdb.c
index d2cf41f647..6ae32f5a0f 100644
--- a/ctdb/common/ctdb.c
+++ b/ctdb/common/ctdb.c
@@ -105,6 +105,7 @@ static int ctdb_add_node(struct ctdb_context *ctdb, char *nstr)
 
 	if (ctdb_same_address(&ctdb->address, &node->address)) {
 		ctdb->vnn = node->vnn;
+		node->flags |= NODE_FLAGS_CONNECTED;
 	}
 
 	ctdb->num_nodes++;
@@ -278,11 +279,6 @@ void ctdb_recv_pkt(struct ctdb_context *ctdb, uint8_t *data, uint32_t length)
 		ctdb_reply_error(ctdb, hdr);
 		break;
 
-	case CTDB_REPLY_REDIRECT:
-		ctdb->status.count.reply_redirect++;
-		ctdb_reply_redirect(ctdb, hdr);
-		break;
-
 	case CTDB_REQ_DMASTER:
 		ctdb->status.count.req_dmaster++;
 		ctdb_request_dmaster(ctdb, hdr);
diff --git a/ctdb/common/ctdb_call.c b/ctdb/common/ctdb_call.c
index 18172996ca..012e6758a5 100644
--- a/ctdb/common/ctdb_call.c
+++ b/ctdb/common/ctdb_call.c
@@ -162,22 +162,18 @@ static void ctdb_send_error(struct ctdb_context *ctdb,
   send a redirect reply
 */
 static void ctdb_call_send_redirect(struct ctdb_context *ctdb, 
+				    TDB_DATA key,
 				    struct ctdb_req_call *c, 
 				    struct ctdb_ltdb_header *header)
 {
-	struct ctdb_reply_redirect *r;
-
-	r = ctdb_transport_allocate(ctdb, ctdb, CTDB_REPLY_REDIRECT, sizeof(*r), 
-				    struct ctdb_reply_redirect);
-	CTDB_NO_MEMORY_FATAL(ctdb, r);
-
-	r->hdr.destnode  = c->hdr.srcnode;
-	r->hdr.reqid     = c->hdr.reqid;
-	r->dmaster       = header->dmaster;
-
-	ctdb_queue_packet(ctdb, &r->hdr);
-
-	talloc_free(r);
+	
+	uint32_t lmaster = ctdb_lmaster(ctdb, &key);
+	if (ctdb->vnn == lmaster) {
+		c->hdr.destnode = header->dmaster;
+	} else {
+		c->hdr.destnode = lmaster;
+	}
+	ctdb_queue_packet(ctdb, &c->hdr);
 }
 
 
@@ -438,8 +434,8 @@ void ctdb_request_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
 	/* if we are not the dmaster, then send a redirect to the
 	   requesting node */
 	if (header.dmaster != ctdb->vnn) {
-		ctdb_call_send_redirect(ctdb, c, &header);
 		talloc_free(data.dptr);
+		ctdb_call_send_redirect(ctdb, call.key, c, &header);
 		ctdb_ltdb_unlock(ctdb_db, call.key);
 		return;
 	}
@@ -595,49 +591,11 @@ void ctdb_reply_error(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
 
 
 /*
-  called when a CTDB_REPLY_REDIRECT packet comes in
-
-  This packet arrives when we have sent a CTDB_REQ_CALL request and
-  the node that received it is not the dmaster for the given key. We
-  are given a hint as to what node to try next.
-*/
-void ctdb_reply_redirect(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
-{
-	struct ctdb_reply_redirect *c = (struct ctdb_reply_redirect *)hdr;
-	struct ctdb_call_state *state;
-
-	state = ctdb_reqid_find(ctdb, hdr->reqid, struct ctdb_call_state);
-	if (state == NULL) {
-		return;
-	}
-
-	if (hdr->reqid != state->reqid) {
-		/* we found a record  but it was the wrong one */
-		DEBUG(0, ("Dropped orphaned dmaster reply with reqid:%d\n",hdr->reqid));
-		return;
-	}
-
-	/* don't allow for too many redirects */
-	if ((++state->redirect_count) % CTDB_MAX_REDIRECT == 0) {
-		c->dmaster = ctdb_lmaster(ctdb, &state->call.key);
-		if (state->redirect_count > ctdb->status.max_redirect_count) {
-			ctdb->status.max_redirect_count = state->redirect_count;
-		}
-	}
-
-	/* send it off again */
-	state->node = ctdb->nodes[c->dmaster];
-	state->c->hdr.destnode = c->dmaster;
-
-	ctdb_queue_packet(ctdb, &state->c->hdr);
-}
-
-/*
   destroy a ctdb_call
 */
 static int ctdb_call_destructor(struct ctdb_call_state *state)
 {
-	ctdb_reqid_remove(state->node->ctdb, state->reqid);
+	ctdb_reqid_remove(state->ctdb_db->ctdb, state->reqid);
 	return 0;
 }
 
@@ -651,7 +609,7 @@ void ctdb_call_timeout(struct event_context *ev, struct timed_event *te,
 	struct ctdb_call_state *state = talloc_get_type(private_data, struct ctdb_call_state);
 	DEBUG(0,(__location__ " call timeout for reqid %d\n", state->c->hdr.reqid));
 	state->state = CTDB_CALL_ERROR;
-	ctdb_set_error(state->node->ctdb, "ctdb_call %u timed out",
+	ctdb_set_error(state->ctdb_db->ctdb, "ctdb_call %u timed out",
 		       state->c->hdr.reqid);
 	if (state->async.fn) {
 		state->async.fn(state);
@@ -692,7 +650,6 @@ struct ctdb_call_state *ctdb_call_local_send(struct ctdb_db_context *ctdb_db,
 	talloc_steal(state, data->dptr);
 
 	state->state = CTDB_CALL_DONE;
-	state->node = ctdb->nodes[ctdb->vnn];
 	state->call = *call;
 	state->ctdb_db = ctdb_db;
 
@@ -729,12 +686,14 @@ struct ctdb_call_state *ctdb_daemon_call_send_remote(struct ctdb_db_context *ctd
 					   struct ctdb_req_call);
 	CTDB_NO_MEMORY_NULL(ctdb, state->c);
 	state->c->hdr.destnode  = header->dmaster;
-	/*
-	   always sending the remote call straight to the lmaster
+
+#if 0
+	/*always sending the remote call straight to the lmaster
 	   improved performance slightly in some tests.
 	   worth investigating further in the future
-	state->c->hdr.destnode  = ctdb_lmaster(ctdb_db->ctdb, &(call->key));
 	*/
+	state->c->hdr.destnode  = ctdb_lmaster(ctdb_db->ctdb, &(call->key));
+#endif
 
 
 	/* this limits us to 16k outstanding messages - not unreasonable */
@@ -751,7 +710,6 @@ struct ctdb_call_state *ctdb_daemon_call_send_remote(struct ctdb_db_context *ctd
 	state->call.call_data.dptr = &state->c->data[call->key.dsize];
 	state->call.key.dptr       = &state->c->data[0];
 
-	state->node   = ctdb->nodes[header->dmaster];
 	state->state  = CTDB_CALL_WAIT;
 	state->header = *header;
 	state->ctdb_db = ctdb_db;
@@ -774,16 +732,16 @@ struct ctdb_call_state *ctdb_daemon_call_send_remote(struct ctdb_db_context *ctd
 int ctdb_daemon_call_recv(struct ctdb_call_state *state, struct ctdb_call *call)
 {
 	while (state->state < CTDB_CALL_DONE) {
-		event_loop_once(state->node->ctdb->ev);
+		event_loop_once(state->ctdb_db->ctdb->ev);
 	}
 	if (state->state != CTDB_CALL_DONE) {
-		ctdb_set_error(state->node->ctdb, "%s", state->errmsg);
+		ctdb_set_error(state->ctdb_db->ctdb, "%s", state->errmsg);
 		talloc_free(state);
 		return -1;
 	}
 
 	if (state->call.reply_data.dsize) {
-		call->reply_data.dptr = talloc_memdup(state->node->ctdb,
+		call->reply_data.dptr = talloc_memdup(state->ctdb_db->ctdb,
 						      state->call.reply_data.dptr,
 						      state->call.reply_data.dsize);
 		call->reply_data.dsize = state->call.reply_data.dsize;
diff --git a/ctdb/common/ctdb_client.c b/ctdb/common/ctdb_client.c
index d7dc7be54d..d11bab6fc4 100644
--- a/ctdb/common/ctdb_client.c
+++ b/ctdb/common/ctdb_client.c
@@ -879,13 +879,15 @@ int ctdb_getdbmap(struct ctdb_context *ctdb, uint32_t destnode, struct ctdb_dbid
 /*
   get a list of nodes (vnn and flags ) from a remote node
  */
-int ctdb_getnodemap(struct ctdb_context *ctdb, uint32_t destnode, struct ctdb_node_map *nodemap)
+int ctdb_getnodemap(struct ctdb_context *ctdb, uint32_t destnode, 
+		    TALLOC_CTX *mem_ctx, struct ctdb_node_map *nodemap)
 {
 	int ret;
 	TDB_DATA data, outdata;
 	int32_t i, res;
 
 	ZERO_STRUCT(data);
+	ZERO_STRUCT(*nodemap);
 	ret = ctdb_control(ctdb, destnode, 0, 
 			   CTDB_CONTROL_GET_NODEMAP, data, 
 			   ctdb, &outdata, &res);
@@ -895,15 +897,9 @@ int ctdb_getnodemap(struct ctdb_context *ctdb, uint32_t destnode, struct ctdb_no
 	}
 
 	nodemap->num = ((uint32_t *)outdata.dptr)[0];
-	if (nodemap->nodes) {
-		talloc_free(nodemap->nodes);
-		nodemap->nodes=NULL;
-	}
-	nodemap->nodes=talloc_array(nodemap, struct ctdb_node_and_flags, nodemap->num);
-	if (!nodemap->nodes) {
-		DEBUG(0,(__location__ " failed to talloc nodemap\n"));
-		return -1;
-	}
+	nodemap->nodes=talloc_array(mem_ctx, struct ctdb_node_and_flags, nodemap->num);
+	CTDB_NO_MEMORY(ctdb, nodemap->nodes);
+
 	for (i=0;i<nodemap->num;i++) {
 		nodemap->nodes[i].vnn = ((uint32_t *)outdata.dptr)[2*i+1];
 		nodemap->nodes[i].flags = ((uint32_t *)outdata.dptr)[2*i+2];
@@ -1083,7 +1079,7 @@ int ctdb_pulldb(struct ctdb_context *ctdb, uint32_t destnode, TALLOC_CTX *mem_ct
 }
 
 /*
-  ping a node
+  ping a node, return number of clients connected
  */
 int ctdb_ping(struct ctdb_context *ctdb, uint32_t destnode)
 {
@@ -1093,10 +1089,10 @@ int ctdb_ping(struct ctdb_context *ctdb, uint32_t destnode)
 
 	ZERO_STRUCT(data);
 	ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_PING, data, NULL, NULL, &res);
-	if (ret != 0 || res != 0) {
+	if (ret != 0) {
 		return -1;
 	}
-	return 0;
+	return res;
 }
 
 /*
@@ -1205,3 +1201,63 @@ int ctdb_set_debuglevel(struct ctdb_context *ctdb, uint32_t destnode, uint32_t l
 	}
 	return 0;
 }
+
+
+/*
+  get a list of connected nodes
+ */
+uint32_t *ctdb_get_connected_nodes(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx,
+				   uint32_t *num_nodes)
+{
+	struct ctdb_node_map *map;
+	int ret, i;
+	uint32_t *nodes;
+
+	*num_nodes = 0;
+
+	map = talloc(mem_ctx, struct ctdb_node_map);
+	CTDB_NO_MEMORY_VOID(ctdb, map);
+
+	ret = ctdb_getnodemap(ctdb, CTDB_CURRENT_NODE, map, map);
+	if (ret != 0) {
+		talloc_free(map);
+		return NULL;
+	}
+
+	nodes = talloc_array(mem_ctx, uint32_t, map->num);
+	if (nodes == NULL) {
+		talloc_free(map);
+		return NULL;
+	}
+
+	for (i=0;i<map->num;i++) {
+		if (map->nodes[i].flags & NODE_FLAGS_CONNECTED) {
+			nodes[*num_nodes] = map->nodes[i].vnn;
+			(*num_nodes)++;
+		}
+	}
+
+	talloc_free(map);
+	return nodes;
+}
+
+
+/*
+  reset remote status
+ */
+int ctdb_status_reset(struct ctdb_context *ctdb, uint32_t destnode)
+{
+	int ret;
+	TDB_DATA data;
+	int32_t res;
+
+	ZERO_STRUCT(data);
+	ret = ctdb_control(ctdb, destnode, 0, 
+			   CTDB_CONTROL_STATUS_RESET, data, 
+			   NULL, NULL, &res);
+	if (ret != 0 || res != 0) {
+		DEBUG(0,(__location__ " ctdb_control for reset status failed\n"));
+		return -1;
+	}
+	return 0;
+}
diff --git a/ctdb/common/ctdb_control.c b/ctdb/common/ctdb_control.c
index 102d0debb3..2fb299f44e 100644
--- a/ctdb/common/ctdb_control.c
+++ b/ctdb/common/ctdb_control.c
@@ -160,6 +160,12 @@ static int32_t ctdb_control_dispatch(struct ctdb_context *ctdb,
 		return 0;
 	}
 
+	case CTDB_CONTROL_STATUS_RESET: {
+		CHECK_CONTROL_DATA_SIZE(0);
+		ZERO_STRUCT(ctdb->status);
+		return 0;
+	}
+
 	case CTDB_CONTROL_GETVNNMAP: {
 		uint32_t i, len;
 		CHECK_CONTROL_DATA_SIZE(0);
@@ -377,7 +383,7 @@ static int32_t ctdb_control_dispatch(struct ctdb_context *ctdb,
 
 	case CTDB_CONTROL_PING:
 		CHECK_CONTROL_DATA_SIZE(0);
-		return 0;
+		return ctdb->num_clients;
 
 	case CTDB_CONTROL_GETDBPATH: {
 		uint32_t db_id;
diff --git a/ctdb/common/ctdb_daemon.c b/ctdb/common/ctdb_daemon.c
index d305718575..d0d2c1a46d 100644
--- a/ctdb/common/ctdb_daemon.c
+++ b/ctdb/common/ctdb_daemon.c
@@ -232,6 +232,7 @@ static void daemon_request_connect_wait(struct ctdb_client *client,
 */
 static int ctdb_client_destructor(struct ctdb_client *client)
 {
+	client->ctdb->num_clients--;
 	close(client->fd);
 	client->fd = -1;
 	return 0;
@@ -547,6 +548,7 @@ static void ctdb_accept_client(struct event_context *ev, struct fd_event *fde,
 	client = talloc_zero(ctdb, struct ctdb_client);
 	client->ctdb = ctdb;
 	client->fd = fd;
+	ctdb->num_clients++;
 
 	client->queue = ctdb_queue_setup(ctdb, client, fd, CTDB_DS_ALIGNMENT, 
 					 ctdb_daemon_read_cb, client);
diff --git a/ctdb/include/ctdb.h b/ctdb/include/ctdb.h
index aa2f31fa34..1e382b4b8c 100644
--- a/ctdb/include/ctdb.h
+++ b/ctdb/include/ctdb.h
@@ -239,7 +239,8 @@ struct ctdb_node_map {
 	uint32_t num;
 	struct ctdb_node_and_flags *nodes;
 };
-int ctdb_getnodemap(struct ctdb_context *ctdb, uint32_t destnode, struct ctdb_node_map *nodemap);
+int ctdb_getnodemap(struct ctdb_context *ctdb, uint32_t destnode, 
+		    TALLOC_CTX *mem_ctx, struct ctdb_node_map *nodemap);
 
 struct ctdb_key_list {
 	uint32_t num;
@@ -288,4 +289,9 @@ int ctdb_getrecmode(struct ctdb_context *ctdb, uint32_t destnode, uint32_t *recm
  */
 int ctdb_setrecmode(struct ctdb_context *ctdb, uint32_t destnode, uint32_t recmode);
 
+uint32_t *ctdb_get_connected_nodes(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx,
+				   uint32_t *num_nodes);
+
+int ctdb_status_reset(struct ctdb_context *ctdb, uint32_t destnode);
+
 #endif
diff --git a/ctdb/include/ctdb_private.h b/ctdb/include/ctdb_private.h
index 73de824f33..e3b33ff6bf 100644
--- a/ctdb/include/ctdb_private.h
+++ b/ctdb/include/ctdb_private.h
@@ -135,7 +135,6 @@ struct ctdb_status {
 	struct {
 		uint32_t req_call;
 		uint32_t reply_call;
-		uint32_t reply_redirect;
 		uint32_t req_dmaster;
 		uint32_t reply_dmaster;
 		uint32_t reply_error;
@@ -157,7 +156,7 @@ struct ctdb_status {
 	uint32_t pending_calls;
 	uint32_t lockwait_calls;
 	uint32_t pending_lockwait_calls;
-	uint32_t max_redirect_count;
+	uint32_t __last_counter; /* hack for control_status_all */
 	double max_call_latency;
 	double max_lockwait_latency;
 };
@@ -196,6 +195,7 @@ struct ctdb_context {
 	struct ctdb_daemon_data daemon;
 	struct ctdb_status status;
 	struct ctdb_vnn_map *vnn_map;
+	uint32_t num_clients;
 };
 
 struct ctdb_db_context {
@@ -228,9 +228,6 @@ struct ctdb_db_context {
 /* arbitrary maximum timeout for ctdb operations */
 #define CTDB_REQ_TIMEOUT 0
 
-/* max number of redirects before we ask the lmaster */
-#define CTDB_MAX_REDIRECT 2
-
 /* number of consecutive calls from the same node before we give them
    the record */
 #define CTDB_DEFAULT_MAX_LACOUNT 7
@@ -261,7 +258,8 @@ enum ctdb_controls {CTDB_CONTROL_PROCESS_EXISTS,
 		    CTDB_CONTROL_CLEAR_DB,
 		    CTDB_CONTROL_PULL_DB,
 		    CTDB_CONTROL_GET_RECMODE,
-		    CTDB_CONTROL_SET_RECMODE};
+		    CTDB_CONTROL_SET_RECMODE,
+		    CTDB_CONTROL_STATUS_RESET};
 
 enum call_state {CTDB_CALL_WAIT, CTDB_CALL_DONE, CTDB_CALL_ERROR};
 
@@ -273,10 +271,8 @@ struct ctdb_call_state {
 	uint32_t reqid;
 	struct ctdb_req_call *c;
 	struct ctdb_db_context *ctdb_db;
-	struct ctdb_node *node;
 	const char *errmsg;
 	struct ctdb_call call;
-	int redirect_count;
 	struct ctdb_ltdb_header header;
 	struct {
 		void (*fn)(struct ctdb_call_state *);
@@ -298,21 +294,20 @@ struct ctdb_fetch_handle {
 */
 enum ctdb_operation {
 	CTDB_REQ_CALL           = 0,
-	CTDB_REPLY_CALL         = 1,
-	CTDB_REPLY_REDIRECT     = 2,
-	CTDB_REQ_DMASTER        = 3,
-	CTDB_REPLY_DMASTER      = 4,
-	CTDB_REPLY_ERROR        = 5,
-	CTDB_REQ_MESSAGE        = 6,
-	CTDB_REQ_FINISHED       = 7,
-	CTDB_REQ_CONTROL        = 8,
-	CTDB_REPLY_CONTROL      = 9,
+	CTDB_REPLY_CALL,
+	CTDB_REQ_DMASTER,
+	CTDB_REPLY_DMASTER,
+	CTDB_REPLY_ERROR,
+	CTDB_REQ_MESSAGE,
+	CTDB_REQ_FINISHED,
+	CTDB_REQ_CONTROL,
+	CTDB_REPLY_CONTROL,
 	
 	/* only used on the domain socket */
 	CTDB_REQ_REGISTER       = 1000,     
-	CTDB_REQ_CONNECT_WAIT   = 1001,
-	CTDB_REPLY_CONNECT_WAIT = 1002,
-	CTDB_REQ_SHUTDOWN       = 1003
+	CTDB_REQ_CONNECT_WAIT,
+	CTDB_REPLY_CONNECT_WAIT,
+	CTDB_REQ_SHUTDOWN
 };
 
 #define CTDB_MAGIC 0x43544442 /* CTDB */
@@ -356,11 +351,6 @@ struct ctdb_reply_error {
 	uint8_t  msg[1];
 };
 
-struct ctdb_reply_redirect {
-	struct ctdb_req_header hdr;
-	uint32_t dmaster;
-};
-
 struct ctdb_req_dmaster {
 	struct ctdb_req_header hdr;
 	uint32_t db_id;
@@ -447,7 +437,6 @@ void ctdb_request_message(struct ctdb_context *ctdb, struct ctdb_req_header *hdr
 void ctdb_reply_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
 void ctdb_reply_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
 void ctdb_reply_error(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
-void ctdb_reply_redirect(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
 
 uint32_t ctdb_lmaster(struct ctdb_context *ctdb, const TDB_DATA *key);
 int ctdb_ltdb_fetch(struct ctdb_db_context *ctdb_db, 
diff --git a/ctdb/tests/run_tests.sh b/ctdb/tests/run_tests.sh
index 983afa0adb..e628bad800 100755
--- a/ctdb/tests/run_tests.sh
+++ b/ctdb/tests/run_tests.sh
@@ -1,7 +1,7 @@
 #!/bin/sh
 
-tests/fetch.sh || exit 1
-tests/bench.sh || exit 1
+tests/fetch.sh 4 || exit 1
+tests/bench.sh 4 || exit 1
 tests/test.sh || exit 1
 
 echo "All OK"
diff --git a/ctdb/tools/ctdb_control.c b/ctdb/tools/ctdb_control.c
index d2fa2a4308..469914da87 100644
--- a/ctdb/tools/ctdb_control.c
+++ b/ctdb/tools/ctdb_control.c
@@ -35,8 +35,9 @@ static void usage(void)
 	printf("\nControls:\n");
 	printf("  ping\n");
 	printf("  process-exists <vnn:pid>           see if a process exists\n");
-	printf("  status <vnn>                       show ctdb status on a node\n");
-	printf("  debug <vnn> <level>                set ctdb debug level on a node\n");
+	printf("  status <vnn|all>                   show ctdb status on a node\n");
+	printf("  statusreset <vnn|all>              reset status on a node\n");
+	printf("  debug <vnn|all> <level>            set ctdb debug level on a node\n");
 	printf("  debuglevel                         display ctdb debug levels\n");
 	printf("  getvnnmap <vnn>                    display ctdb vnnmap\n");
 	printf("  setvnnmap <vnn> <generation> <numslots> <lmaster>*\n");
@@ -95,23 +96,61 @@ static void show_status(struct ctdb_status *s)
 	printf(" node_packets_recv       %u\n", s->node_packets_recv);
 	printf("   req_call              %u\n", s->count.req_call);
 	printf("   reply_call            %u\n", s->count.reply_call);
-	printf("   reply_redirect        %u\n", s->count.reply_redirect);
 	printf("   req_dmaster           %u\n", s->count.req_dmaster);
 	printf("   reply_dmaster         %u\n", s->count.reply_dmaster);
 	printf("   reply_error           %u\n", s->count.reply_error);
-	printf("   reply_redirect        %u\n", s->count.reply_redirect);
 	printf("   req_message           %u\n", s->count.req_message);
 	printf("   req_finished          %u\n", s->count.req_finished);
 	printf(" total_calls             %u\n", s->total_calls);
 	printf(" pending_calls           %u\n", s->pending_calls);
 	printf(" lockwait_calls          %u\n", s->lockwait_calls);
 	printf(" pending_lockwait_calls  %u\n", s->pending_lockwait_calls);
-	printf(" max_redirect_count      %u\n", s->max_redirect_count);
 	printf(" max_call_latency        %.6f sec\n", s->max_call_latency);
 	printf(" max_lockwait_latency    %.6f sec\n", s->max_lockwait_latency);
 }
 
 /*
+  display remote ctdb status combined from all nodes
+ */
+static int control_status_all(struct ctdb_context *ctdb)
+{
+	int ret, i;
+	struct ctdb_status status;
+	uint32_t *nodes;
+	uint32_t num_nodes;
+
+	nodes = ctdb_get_connected_nodes(ctdb, ctdb, &num_nodes);
+	CTDB_NO_MEMORY(ctdb, nodes);
+	
+	ZERO_STRUCT(status);
+
+	for (i=0;i<num_nodes;i++) {
+		struct ctdb_status s1;
+		int j;
+		uint32_t *v1 = (uint32_t *)&s1;
+		uint32_t *v2 = (uint32_t *)&status;
+		uint32_t num_ints = 
+			offsetof(struct ctdb_status, __last_counter) / sizeof(uint32_t);
+		ret = ctdb_status(ctdb, nodes[i], &s1);
+		if (ret != 0) {
+			printf("Unable to get status from node %u\n", nodes[i]);
+			return ret;
+		}
+		for (j=0;j<num_ints;j++) {
+			v2[j] += v1[j];
+		}
+		status.max_call_latency = 
+			MAX(status.max_call_latency, s1.max_call_latency);
+		status.max_lockwait_latency = 
+			MAX(status.max_lockwait_latency, s1.max_lockwait_latency);
+	}
+	talloc_free(nodes);
+	printf("Gathered status for %u nodes\n", num_nodes);
+	show_status(&status);
+	return 0;
+}
+
+/*
   display remote ctdb status
  */
 static int control_status(struct ctdb_context *ctdb, int argc, const char **argv)
@@ -123,6 +162,10 @@ static int control_status(struct ctdb_context *ctdb, int argc, const char **argv
 		usage();
 	}
 
+	if (strcmp(argv[0], "all") == 0) {
+		return control_status_all(ctdb);
+	}
+
 	vnn = strtoul(argv[0], NULL, 0);
 
 	ret = ctdb_status(ctdb, vnn, &status);
@@ -134,6 +177,56 @@ static int control_status(struct ctdb_context *ctdb, int argc, const char **argv
 	return 0;
 }
 
+
+/*
+  reset status on all nodes
+ */
+static int control_status_reset_all(struct ctdb_context *ctdb)
+{
+	int ret, i;
+	uint32_t *nodes;
+	uint32_t num_nodes;
+
+	nodes = ctdb_get_connected_nodes(ctdb, ctdb, &num_nodes);
+	CTDB_NO_MEMORY(ctdb, nodes);
+	
+	for (i=0;i<num_nodes;i++) {
+		ret = ctdb_status_reset(ctdb, nodes[i]);
+		if (ret != 0) {
+			printf("Unable to reset status on node %u\n", nodes[i]);
+			return ret;
+		}
+	}
+	talloc_free(nodes);
+	return 0;
+}
+
+
+/*
+  reset remote ctdb status
+ */
+static int control_status_reset(struct ctdb_context *ctdb, int argc, const char **argv)
+{
+	uint32_t vnn;
+	int ret;
+	if (argc < 1) {
+		usage();
+	}
+
+	if (strcmp(argv[0], "all") == 0) {
+		return control_status_reset_all(ctdb);
+	}
+
+	vnn = strtoul(argv[0], NULL, 0);
+
+	ret = ctdb_status_reset(ctdb, vnn);
+	if (ret != 0) {
+		printf("Unable to reset status on node %u\n", vnn);
+		return ret;
+	}
+	return 0;
+}
+
 /*
   display remote ctdb vnn map
  */
@@ -303,7 +396,7 @@ static int control_getnodemap(struct ctdb_context *ctdb, int argc, const char **
 	vnn = strtoul(argv[0], NULL, 0);
 
 	nodemap = talloc_zero(ctdb, struct ctdb_node_map);
-	ret = ctdb_getnodemap(ctdb, vnn, nodemap);
+	ret = ctdb_getnodemap(ctdb, vnn, nodemap, nodemap);
 	if (ret != 0) {
 		printf("Unable to get nodemap from node %u\n", vnn);
 		talloc_free(nodemap);
@@ -428,17 +521,23 @@ static int control_pulldb(struct ctdb_context *ctdb, int argc, const char **argv
 static int control_ping(struct ctdb_context *ctdb, int argc, const char **argv)
 {
 	int ret, i;
+	uint32_t *nodes;
+	uint32_t num_nodes;
 
-	for (i=0;i<ctdb->num_nodes;i++) {
+	nodes = ctdb_get_connected_nodes(ctdb, ctdb, &num_nodes);
+	CTDB_NO_MEMORY(ctdb, nodes);
+
+	for (i=0;i<num_nodes;i++) {
 		struct timeval tv = timeval_current();
-		ret = ctdb_ping(ctdb, i);
-		if (ret != 0) {
-			printf("Unable to get ping response from node %u\n", i);
+		ret = ctdb_ping(ctdb, nodes[i]);
+		if (ret == -1) {
+			printf("Unable to get ping response from node %u\n", nodes[i]);
 		} else {
-			printf("response from %u time=%.6f sec\n", 
-			       i, timeval_elapsed(&tv));
+			printf("response from %u time=%.6f sec  (%d clients)\n", 
+			       nodes[i], timeval_elapsed(&tv), ret);
 		}
 	}
+	talloc_free(nodes);
 	return 0;
 }
 
@@ -449,16 +548,23 @@ static int control_ping(struct ctdb_context *ctdb, int argc, const char **argv)
 static int control_debuglevel(struct ctdb_context *ctdb, int argc, const char **argv)
 {
 	int ret, i;
+	uint32_t *nodes;
+	uint32_t num_nodes;
+
+	nodes = ctdb_get_connected_nodes(ctdb, ctdb, &num_nodes);
+	CTDB_NO_MEMORY(ctdb, nodes);
 
-	for (i=0;i<ctdb->num_nodes;i++) {
+	for (i=0;i<num_nodes;i++) {
 		uint32_t level;
-		ret = ctdb_get_debuglevel(ctdb, i, &level);
+		ret = ctdb_get_debuglevel(ctdb, nodes[i], &level);
 		if (ret != 0) {
-			printf("Unable to get debuglevel response from node %u\n", i);
+			printf("Unable to get debuglevel response from node %u\n", 
+				nodes[i]);
 		} else {
-			printf("Node %u is at debug level %u\n", i, level);
+			printf("Node %u is at debug level %u\n", nodes[i], level);
 		}
 	}
+	talloc_free(nodes);
 	return 0;
 }
 
@@ -468,19 +574,36 @@ static int control_debuglevel(struct ctdb_context *ctdb, int argc, const char **
 static int control_debug(struct ctdb_context *ctdb, int argc, const char **argv)
 {
 	int ret;
-	uint32_t vnn, level;
+	uint32_t vnn, level, i;
+	uint32_t *nodes;
+	uint32_t num_nodes;
 
 	if (argc < 2) {
 		usage();
 	}
 
-	vnn   = strtoul(argv[0], NULL, 0);
 	level = strtoul(argv[1], NULL, 0);
 
-	ret = ctdb_set_debuglevel(ctdb, vnn, level);
-	if (ret != 0) {
-		printf("Unable to set debug level on node %u\n", vnn);
+	if (strcmp(argv[0], "all") != 0) {
+		vnn = strtoul(argv[0], NULL, 0);
+		ret = ctdb_set_debuglevel(ctdb, vnn, level);
+		if (ret != 0) {
+			printf("Unable to set debug level on node %u\n", vnn);
+		}
+		
+		return 0;
+	}
+
+	nodes = ctdb_get_connected_nodes(ctdb, ctdb, &num_nodes);
+	CTDB_NO_MEMORY(ctdb, nodes);
+	for (i=0;i<num_nodes;i++) {
+		ret = ctdb_set_debuglevel(ctdb, nodes[i], level);
+		if (ret != 0) {
+			printf("Unable to set debug level on node %u\n", nodes[i]);
+			break;
+		}
 	}
+	talloc_free(nodes);
 	return 0;
 }
 
@@ -540,6 +663,8 @@ int main(int argc, const char *argv[])
 		ret = control_process_exists(ctdb, extra_argc-1, extra_argv+1);
 	} else if (strcmp(control, "status") == 0) {
 		ret = control_status(ctdb, extra_argc-1, extra_argv+1);
+	} else if (strcmp(control, "statusreset") == 0) {
+		ret = control_status_reset(ctdb, extra_argc-1, extra_argv+1);
 	} else if (strcmp(control, "getvnnmap") == 0) {
 		ret = control_getvnnmap(ctdb, extra_argc-1, extra_argv+1);
 	} else if (strcmp(control, "getdbmap") == 0) {