Attachment 14173 Details for Bug 13411 – patch for 4.8

[patch] patch for 4.8

0001-ctdb-client-Remove-ununsed-functions-from-old-client.patch (text/plain), 81.56 KB, created by Andreas Schneider on 2018-05-03 06:49:41 UTC

(hide)

Description:

Filename:

MIME Type:

Creator: Andreas Schneider

Created: 2018-05-03 06:49:41 UTC

Size: 81.56 KB

patch

obsolete

>From 8b751255255bc93c80bd1849b07ccf36b8065925 Mon Sep 17 00:00:00 2001
>From: Amitay Isaacs <amitay@gmail.com>
>Date: Mon, 30 Apr 2018 19:32:13 +1000
>Subject: [PATCH] ctdb-client: Remove ununsed functions from old client code
>
>BUG: https://bugzilla.samba.org/show_bug.cgi?id=13411
>
>Signed-off-by: Amitay Isaacs <amitay@gmail.com>
>Reviewed-by: Martin Schwenke <martin@meltin.net>
>
>(cherry picked from commit 01c8dc7e15b8764a9b8c8e34b84d0cab790edf47)
>---
> ctdb/client/ctdb_client.c  | 2535 ++------------------------------------------
> ctdb/include/ctdb_client.h |  218 ----
> 2 files changed, 117 insertions(+), 2636 deletions(-)
>
>diff --git a/ctdb/client/ctdb_client.c b/ctdb/client/ctdb_client.c
>index d399cb551f6..635a4882715 100644
>--- a/ctdb/client/ctdb_client.c
>+++ b/ctdb/client/ctdb_client.c
>@@ -583,354 +583,6 @@ int ctdb_client_send_message(struct ctdb_context *ctdb, uint32_t pnn,
> }
> 
> 
>-/*
>-  cancel a ctdb_fetch_lock operation, releasing the lock
>- */
>-static int fetch_lock_destructor(struct ctdb_record_handle *h)
>-{
>-	ctdb_ltdb_unlock(h->ctdb_db, h->key);
>-	return 0;
>-}
>-
>-/*
>-  force the migration of a record to this node
>- */
>-static int ctdb_client_force_migration(struct ctdb_db_context *ctdb_db, TDB_DATA key)
>-{
>-	struct ctdb_call call;
>-	ZERO_STRUCT(call);
>-	call.call_id = CTDB_NULL_FUNC;
>-	call.key = key;
>-	call.flags = CTDB_IMMEDIATE_MIGRATION;
>-	return ctdb_call(ctdb_db, &call);
>-}
>-
>-/*
>-  try to fetch a readonly copy of a record
>- */
>-static int
>-ctdb_client_fetch_readonly(struct ctdb_db_context *ctdb_db, TDB_DATA key, TALLOC_CTX *mem_ctx, struct ctdb_ltdb_header **hdr, TDB_DATA *data)
>-{
>-	int ret;
>-
>-	struct ctdb_call call;
>-	ZERO_STRUCT(call);
>-
>-	call.call_id = CTDB_FETCH_WITH_HEADER_FUNC;
>-	call.call_data.dptr = NULL;
>-	call.call_data.dsize = 0;
>-	call.key = key;
>-	call.flags = CTDB_WANT_READONLY;
>-	ret = ctdb_call(ctdb_db, &call);
>-
>-	if (ret != 0) {
>-		return -1;
>-	}
>-	if (call.reply_data.dsize < sizeof(struct ctdb_ltdb_header)) {
>-		return -1;
>-	}
>-
>-	*hdr = talloc_memdup(mem_ctx, &call.reply_data.dptr[0], sizeof(struct ctdb_ltdb_header));
>-	if (*hdr == NULL) {
>-		talloc_free(call.reply_data.dptr);
>-		return -1;
>-	}
>-
>-	data->dsize = call.reply_data.dsize - sizeof(struct ctdb_ltdb_header);
>-	data->dptr  = talloc_memdup(mem_ctx, &call.reply_data.dptr[sizeof(struct ctdb_ltdb_header)], data->dsize);
>-	if (data->dptr == NULL) {
>-		talloc_free(call.reply_data.dptr);
>-		talloc_free(hdr);
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
>-/*
>-  get a lock on a record, and return the records data. Blocks until it gets the lock
>- */
>-struct ctdb_record_handle *ctdb_fetch_lock(struct ctdb_db_context *ctdb_db, TALLOC_CTX *mem_ctx, 
>-					   TDB_DATA key, TDB_DATA *data)
>-{
>-	int ret;
>-	struct ctdb_record_handle *h;
>-
>-	/*
>-	  procedure is as follows:
>-
>-	  1) get the chain lock. 
>-	  2) check if we are dmaster
>-	  3) if we are the dmaster then return handle 
>-	  4) if not dmaster then ask ctdb daemon to make us dmaster, and wait for
>-	     reply from ctdbd
>-	  5) when we get the reply, goto (1)
>-	 */
>-
>-	h = talloc_zero(mem_ctx, struct ctdb_record_handle);
>-	if (h == NULL) {
>-		return NULL;
>-	}
>-
>-	h->ctdb_db = ctdb_db;
>-	h->key     = key;
>-	h->key.dptr = talloc_memdup(h, key.dptr, key.dsize);
>-	if (h->key.dptr == NULL) {
>-		talloc_free(h);
>-		return NULL;
>-	}
>-	h->data    = data;
>-
>-	DEBUG(DEBUG_DEBUG,("ctdb_fetch_lock: key=%*.*s\n", (int)key.dsize, (int)key.dsize, 
>-		 (const char *)key.dptr));
>-
>-again:
>-	/* step 1 - get the chain lock */
>-	ret = ctdb_ltdb_lock(ctdb_db, key);
>-	if (ret != 0) {
>-		DEBUG(DEBUG_ERR, (__location__ " failed to lock ltdb record\n"));
>-		talloc_free(h);
>-		return NULL;
>-	}
>-
>-	DEBUG(DEBUG_DEBUG,("ctdb_fetch_lock: got chain lock\n"));
>-
>-	talloc_set_destructor(h, fetch_lock_destructor);
>-
>-	ret = ctdb_ltdb_fetch(ctdb_db, key, &h->header, h, data);
>-
>-	/* when torturing, ensure we test the remote path */
>-	if ((ctdb_db->ctdb->flags & CTDB_FLAG_TORTURE) &&
>-	    random() % 5 == 0) {
>-		h->header.dmaster = (uint32_t)-1;
>-	}
>-
>-
>-	DEBUG(DEBUG_DEBUG,("ctdb_fetch_lock: done local fetch\n"));
>-
>-	if (ret != 0 || h->header.dmaster != ctdb_db->ctdb->pnn) {
>-		ctdb_ltdb_unlock(ctdb_db, key);
>-		ret = ctdb_client_force_migration(ctdb_db, key);
>-		if (ret != 0) {
>-			DEBUG(DEBUG_DEBUG,("ctdb_fetch_lock: force_migration failed\n"));
>-			talloc_free(h);
>-			return NULL;
>-		}
>-		goto again;
>-	}
>-
>-	/* if this is a request for read/write and we have delegations
>-	   we have to revoke all delegations first
>-	*/
>-	if ((h->header.dmaster == ctdb_db->ctdb->pnn) &&
>-	    (h->header.flags & CTDB_REC_RO_HAVE_DELEGATIONS)) {
>-		ctdb_ltdb_unlock(ctdb_db, key);
>-		ret = ctdb_client_force_migration(ctdb_db, key);
>-		if (ret != 0) {
>-			DEBUG(DEBUG_DEBUG,("ctdb_fetch_readonly_lock: force_migration failed\n"));
>-			talloc_free(h);
>-			return NULL;
>-		}
>-		goto again;
>-	}
>-
>-	DEBUG(DEBUG_DEBUG,("ctdb_fetch_lock: we are dmaster - done\n"));
>-	return h;
>-}
>-
>-/*
>-  get a readonly lock on a record, and return the records data. Blocks until it gets the lock
>- */
>-struct ctdb_record_handle *
>-ctdb_fetch_readonly_lock(
>-	struct ctdb_db_context *ctdb_db, TALLOC_CTX *mem_ctx, 
>-	TDB_DATA key, TDB_DATA *data,
>-	int read_only)
>-{
>-	int ret;
>-	struct ctdb_record_handle *h;
>-	struct ctdb_ltdb_header *roheader = NULL;
>-
>-	h = talloc_zero(mem_ctx, struct ctdb_record_handle);
>-	if (h == NULL) {
>-		return NULL;
>-	}
>-
>-	h->ctdb_db = ctdb_db;
>-	h->key     = key;
>-	h->key.dptr = talloc_memdup(h, key.dptr, key.dsize);
>-	if (h->key.dptr == NULL) {
>-		talloc_free(h);
>-		return NULL;
>-	}
>-	h->data    = data;
>-
>-	data->dptr = NULL;
>-	data->dsize = 0;
>-
>-
>-again:
>-	talloc_free(roheader);
>-	roheader = NULL;
>-
>-	talloc_free(data->dptr);
>-	data->dptr = NULL;
>-	data->dsize = 0;
>-
>-	/* Lock the record/chain */
>-	ret = ctdb_ltdb_lock(ctdb_db, key);
>-	if (ret != 0) {
>-		DEBUG(DEBUG_ERR, (__location__ " failed to lock ltdb record\n"));
>-		talloc_free(h);
>-		return NULL;
>-	}
>-
>-	talloc_set_destructor(h, fetch_lock_destructor);
>-
>-	/* Check if record exists yet in the TDB */
>-	ret = ctdb_ltdb_fetch_with_header(ctdb_db, key, &h->header, h, data);
>-	if (ret != 0) {
>-		ctdb_ltdb_unlock(ctdb_db, key);
>-		ret = ctdb_client_force_migration(ctdb_db, key);
>-		if (ret != 0) {
>-			DEBUG(DEBUG_DEBUG,("ctdb_fetch_readonly_lock: force_migration failed\n"));
>-			talloc_free(h);
>-			return NULL;
>-		}
>-		goto again;
>-	}
>-
>-	/* if this is a request for read/write and we have delegations
>-	   we have to revoke all delegations first
>-	*/
>-	if ((read_only == 0) 
>-	&&  (h->header.dmaster == ctdb_db->ctdb->pnn)
>-	&&  (h->header.flags & CTDB_REC_RO_HAVE_DELEGATIONS)) {
>-		ctdb_ltdb_unlock(ctdb_db, key);
>-		ret = ctdb_client_force_migration(ctdb_db, key);
>-		if (ret != 0) {
>-			DEBUG(DEBUG_DEBUG,("ctdb_fetch_readonly_lock: force_migration failed\n"));
>-			talloc_free(h);
>-			return NULL;
>-		}
>-		goto again;
>-	}
>-
>-	/* if we are dmaster, just return the handle */
>-	if (h->header.dmaster == ctdb_db->ctdb->pnn) {
>-		return h;
>-	}
>-
>-	if (read_only != 0) {
>-		TDB_DATA rodata = {NULL, 0};
>-
>-		if ((h->header.flags & CTDB_REC_RO_HAVE_READONLY)
>-		||  (h->header.flags & CTDB_REC_RO_HAVE_DELEGATIONS)) {
>-			return h;
>-		}
>-
>-		ctdb_ltdb_unlock(ctdb_db, key);
>-		ret = ctdb_client_fetch_readonly(ctdb_db, key, h, &roheader, &rodata);
>-		if (ret != 0) {
>-			DEBUG(DEBUG_ERR,("ctdb_fetch_readonly_lock:  failed. force migration and try again\n"));
>-			ret = ctdb_client_force_migration(ctdb_db, key);
>-			if (ret != 0) {
>-				DEBUG(DEBUG_DEBUG,("ctdb_fetch_readonly_lock: force_migration failed\n"));
>-				talloc_free(h);
>-				return NULL;
>-			}
>-
>-			goto again;
>-		}
>-
>-		if (!(roheader->flags&CTDB_REC_RO_HAVE_READONLY)) {
>-			ret = ctdb_client_force_migration(ctdb_db, key);
>-			if (ret != 0) {
>-				DEBUG(DEBUG_DEBUG,("ctdb_fetch_readonly_lock: force_migration failed\n"));
>-				talloc_free(h);
>-				return NULL;
>-			}
>-
>-			goto again;
>-		}
>-
>-		ret = ctdb_ltdb_lock(ctdb_db, key);
>-		if (ret != 0) {
>-			DEBUG(DEBUG_ERR, (__location__ " failed to lock ltdb record\n"));
>-			talloc_free(h);
>-			return NULL;
>-		}
>-
>-		ret = ctdb_ltdb_fetch_with_header(ctdb_db, key, &h->header, h, data);
>-		if (ret != 0) {
>-			ctdb_ltdb_unlock(ctdb_db, key);
>-
>-			ret = ctdb_client_force_migration(ctdb_db, key);
>-			if (ret != 0) {
>-				DEBUG(DEBUG_DEBUG,("ctdb_fetch_readonly_lock: force_migration failed\n"));
>-				talloc_free(h);
>-				return NULL;
>-			}
>-
>-			goto again;
>-		}
>-
>-		return h;
>-	}
>-
>-	/* we are not dmaster and this was not a request for a readonly lock
>-	 * so unlock the record, migrate it and try again
>-	 */
>-	ctdb_ltdb_unlock(ctdb_db, key);
>-	ret = ctdb_client_force_migration(ctdb_db, key);
>-	if (ret != 0) {
>-		DEBUG(DEBUG_DEBUG,("ctdb_fetch_lock: force_migration failed\n"));
>-		talloc_free(h);
>-		return NULL;
>-	}
>-	goto again;
>-}
>-
>-/*
>-  store some data to the record that was locked with ctdb_fetch_lock()
>-*/
>-int ctdb_record_store(struct ctdb_record_handle *h, TDB_DATA data)
>-{
>-	if (! ctdb_db_volatile(h->ctdb_db)) {
>-		DEBUG(DEBUG_ERR,
>-		      ("ctdb_record_store prohibited for non-volatile dbs\n"));
>-		return -1;
>-	}
>-
>-	return ctdb_ltdb_store(h->ctdb_db, h->key, &h->header, data);
>-}
>-
>-/*
>-  non-locking fetch of a record
>- */
>-int ctdb_fetch(struct ctdb_db_context *ctdb_db, TALLOC_CTX *mem_ctx, 
>-	       TDB_DATA key, TDB_DATA *data)
>-{
>-	struct ctdb_call call;
>-	int ret;
>-
>-	call.call_id = CTDB_FETCH_FUNC;
>-	call.call_data.dptr = NULL;
>-	call.call_data.dsize = 0;
>-	call.key = key;
>-
>-	ret = ctdb_call(ctdb_db, &call);
>-
>-	if (ret == 0) {
>-		*data = call.reply_data;
>-		talloc_steal(mem_ctx, data->dptr);
>-	}
>-
>-	return ret;
>-}
>-
>-
>-
> /*
>    called when a control completes or timesout to invoke the callback
>    function the user provided
>@@ -1210,134 +862,6 @@ int ctdb_control(struct ctdb_context *ctdb, uint32_t destnode, uint64_t srvid,
> 			errormsg);
> }
> 
>-
>-
>-
>-/*
>-  a process exists call. Returns 0 if process exists, -1 otherwise
>- */
>-int ctdb_ctrl_process_exists(struct ctdb_context *ctdb, uint32_t destnode, pid_t pid)
>-{
>-	int ret;
>-	TDB_DATA data;
>-	int32_t status;
>-
>-	data.dptr = (uint8_t*)&pid;
>-	data.dsize = sizeof(pid);
>-
>-	ret = ctdb_control(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_PROCESS_EXISTS, 0, data, 
>-			   NULL, NULL, &status, NULL, NULL);
>-	if (ret != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for process_exists failed\n"));
>-		return -1;
>-	}
>-
>-	return status;
>-}
>-
>-/*
>-  get remote statistics
>- */
>-int ctdb_ctrl_statistics(struct ctdb_context *ctdb, uint32_t destnode, struct ctdb_statistics *status)
>-{
>-	int ret;
>-	TDB_DATA data;
>-	int32_t res;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_STATISTICS, 0, tdb_null, 
>-			   ctdb, &data, &res, NULL, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for statistics failed\n"));
>-		return -1;
>-	}
>-
>-	if (data.dsize != sizeof(struct ctdb_statistics)) {
>-		DEBUG(DEBUG_ERR,(__location__ " Wrong statistics size %u - expected %u\n",
>-			 (unsigned)data.dsize, (unsigned)sizeof(struct ctdb_statistics)));
>-		      return -1;
>-	}
>-
>-	*status = *(struct ctdb_statistics *)data.dptr;
>-	talloc_free(data.dptr);
>-			
>-	return 0;
>-}
>-
>-/*
>- * get db statistics
>- */
>-int ctdb_ctrl_dbstatistics(struct ctdb_context *ctdb, uint32_t destnode, uint32_t dbid,
>-			   TALLOC_CTX *mem_ctx, struct ctdb_db_statistics_old **dbstat)
>-{
>-	int ret;
>-	TDB_DATA indata, outdata;
>-	int32_t res;
>-	struct ctdb_db_statistics_old *wire, *s;
>-	char *ptr;
>-	int i;
>-
>-	indata.dptr = (uint8_t *)&dbid;
>-	indata.dsize = sizeof(dbid);
>-
>-	ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_GET_DB_STATISTICS,
>-			   0, indata, ctdb, &outdata, &res, NULL, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for dbstatistics failed\n"));
>-		return -1;
>-	}
>-
>-	if (outdata.dsize < offsetof(struct ctdb_db_statistics_old, hot_keys_wire)) {
>-		DEBUG(DEBUG_ERR,(__location__ " Wrong dbstatistics size %zi - expected >= %lu\n",
>-				 outdata.dsize,
>-				 (long unsigned int)sizeof(struct ctdb_statistics)));
>-		return -1;
>-	}
>-
>-	s = talloc_zero(mem_ctx, struct ctdb_db_statistics_old);
>-	if (s == NULL) {
>-		talloc_free(outdata.dptr);
>-		CTDB_NO_MEMORY(ctdb, s);
>-	}
>-
>-	wire = (struct ctdb_db_statistics_old *)outdata.dptr;
>-	memcpy(s, wire, offsetof(struct ctdb_db_statistics_old, hot_keys_wire));
>-	ptr = &wire->hot_keys_wire[0];
>-	for (i=0; i<wire->num_hot_keys; i++) {
>-		s->hot_keys[i].key.dptr = talloc_size(mem_ctx, s->hot_keys[i].key.dsize);
>-		if (s->hot_keys[i].key.dptr == NULL) {
>-			talloc_free(outdata.dptr);
>-			CTDB_NO_MEMORY(ctdb, s->hot_keys[i].key.dptr);
>-		}
>-
>-		memcpy(s->hot_keys[i].key.dptr, ptr, s->hot_keys[i].key.dsize);
>-		ptr += wire->hot_keys[i].key.dsize;
>-	}
>-
>-	talloc_free(outdata.dptr);
>-	*dbstat = s;
>-	return 0;
>-}
>-
>-/*
>-  shutdown a remote ctdb node
>- */
>-int ctdb_ctrl_shutdown(struct ctdb_context *ctdb, struct timeval timeout, uint32_t destnode)
>-{
>-	struct ctdb_client_control_state *state;
>-
>-	state = ctdb_control_send(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_SHUTDOWN, 0, tdb_null, 
>-			   NULL, &timeout, NULL);
>-	if (state == NULL) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for shutdown failed\n"));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
> /*
>   get vnn map from a remote node
>  */
>@@ -1554,259 +1078,68 @@ int ctdb_ctrl_getnodemap(struct ctdb_context *ctdb,
> 	return 0;
> }
> 
>-/*
>-  load nodes file on a remote node and return as a node map
>- */
>-int ctdb_ctrl_getnodesfile(struct ctdb_context *ctdb,
>-			   struct timeval timeout, uint32_t destnode,
>-			   TALLOC_CTX *mem_ctx, struct ctdb_node_map_old **nodemap)
>+int ctdb_ctrl_get_runstate(struct ctdb_context *ctdb,
>+			   struct timeval timeout,
>+			   uint32_t destnode,
>+			   uint32_t *runstate)
> {
>-	int ret;
> 	TDB_DATA outdata;
> 	int32_t res;
>+	int ret;
> 
>-	ret = ctdb_control(ctdb, destnode, 0,
>-			   CTDB_CONTROL_GET_NODES_FILE, 0, tdb_null,
>-			   mem_ctx, &outdata, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0 || outdata.dsize == 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for getnodes failed ret:%d res:%d\n", ret, res));
>+	ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_GET_RUNSTATE, 0,
>+			   tdb_null, ctdb, &outdata, &res, &timeout, NULL);
>+	if (ret != 0 || res != 0) {
>+		DEBUG(DEBUG_ERR,("ctdb_control for get_runstate failed\n"));
>+		return ret != 0 ? ret : res;
>+	}
>+
>+	if (outdata.dsize != sizeof(uint32_t)) {
>+		DEBUG(DEBUG_ERR,("Invalid return data in get_runstate\n"));
>+		talloc_free(outdata.dptr);
> 		return -1;
> 	}
> 
>-	*nodemap = (struct ctdb_node_map_old *)talloc_memdup(mem_ctx, outdata.dptr, outdata.dsize);
>+	if (runstate != NULL) {
>+		*runstate = *(uint32_t *)outdata.dptr;
>+	}
> 	talloc_free(outdata.dptr);
> 
> 	return 0;
> }
> 
> /*
>-  drop the transport, reload the nodes file and restart the transport
>+  find the real path to a ltdb
>  */
>-int ctdb_ctrl_reload_nodes_file(struct ctdb_context *ctdb, 
>-		    struct timeval timeout, uint32_t destnode)
>+int ctdb_ctrl_getdbpath(struct ctdb_context *ctdb, struct timeval timeout, uint32_t destnode, uint32_t dbid, TALLOC_CTX *mem_ctx,
>+		   const char **path)
> {
> 	int ret;
> 	int32_t res;
>+	TDB_DATA data;
>+
>+	data.dptr = (uint8_t *)&dbid;
>+	data.dsize = sizeof(dbid);
> 
> 	ret = ctdb_control(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_RELOAD_NODES_FILE, 0, tdb_null, 
>-			   NULL, NULL, &res, &timeout, NULL);
>+			   CTDB_CONTROL_GETDBPATH, 0, data,
>+			   mem_ctx, &data, &res, &timeout, NULL);
> 	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for reloadnodesfile failed\n"));
> 		return -1;
> 	}
> 
>+	(*path) = talloc_strndup(mem_ctx, (const char *)data.dptr, data.dsize);
>+	if ((*path) == NULL) {
>+		return -1;
>+	}
>+
>+	talloc_free(data.dptr);
>+
> 	return 0;
> }
> 
>-
> /*
>-  set vnn map on a node
>- */
>-int ctdb_ctrl_setvnnmap(struct ctdb_context *ctdb, struct timeval timeout, uint32_t destnode, 
>-			TALLOC_CTX *mem_ctx, struct ctdb_vnn_map *vnnmap)
>-{
>-	int ret;
>-	TDB_DATA data;
>-	int32_t res;
>-	struct ctdb_vnn_map_wire *map;
>-	size_t len;
>-
>-	len = offsetof(struct ctdb_vnn_map_wire, map) + sizeof(uint32_t)*vnnmap->size;
>-	map = talloc_size(mem_ctx, len);
>-	CTDB_NO_MEMORY(ctdb, map);
>-
>-	map->generation = vnnmap->generation;
>-	map->size = vnnmap->size;
>-	memcpy(map->map, vnnmap->map, sizeof(uint32_t)*map->size);
>-	
>-	data.dsize = len;
>-	data.dptr  = (uint8_t *)map;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_SETVNNMAP, 0, data, 
>-			   NULL, NULL, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for setvnnmap failed\n"));
>-		return -1;
>-	}
>-
>-	talloc_free(map);
>-
>-	return 0;
>-}
>-
>-
>-/*
>-  async send for pull database
>- */
>-struct ctdb_client_control_state *ctdb_ctrl_pulldb_send(
>-	struct ctdb_context *ctdb, uint32_t destnode, uint32_t dbid,
>-	uint32_t lmaster, TALLOC_CTX *mem_ctx, struct timeval timeout)
>-{
>-	TDB_DATA indata;
>-	struct ctdb_pulldb *pull;
>-	struct ctdb_client_control_state *state;
>-
>-	pull = talloc(mem_ctx, struct ctdb_pulldb);
>-	CTDB_NO_MEMORY_NULL(ctdb, pull);
>-
>-	pull->db_id   = dbid;
>-	pull->lmaster = lmaster;
>-
>-	indata.dsize = sizeof(struct ctdb_pulldb);
>-	indata.dptr  = (unsigned char *)pull;
>-
>-	state = ctdb_control_send(ctdb, destnode, 0, 
>-				  CTDB_CONTROL_PULL_DB, 0, indata, 
>-				  mem_ctx, &timeout, NULL);
>-	talloc_free(pull);
>-
>-	return state;
>-}
>-
>-/*
>-  async recv for pull database
>- */
>-int ctdb_ctrl_pulldb_recv(
>-	struct ctdb_context *ctdb, 
>-	TALLOC_CTX *mem_ctx, struct ctdb_client_control_state *state, 
>-	TDB_DATA *outdata)
>-{
>-	int ret;
>-	int32_t res;
>-
>-	ret = ctdb_control_recv(ctdb, state, mem_ctx, outdata, &res, NULL);
>-	if ( (ret != 0) || (res != 0) ){
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_ctrl_pulldb_recv failed\n"));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
>-/*
>-  pull all keys and records for a specific database on a node
>- */
>-int ctdb_ctrl_pulldb(struct ctdb_context *ctdb, uint32_t destnode, 
>-		uint32_t dbid, uint32_t lmaster, 
>-		TALLOC_CTX *mem_ctx, struct timeval timeout,
>-		TDB_DATA *outdata)
>-{
>-	struct ctdb_client_control_state *state;
>-
>-	state = ctdb_ctrl_pulldb_send(ctdb, destnode, dbid, lmaster, mem_ctx,
>-				      timeout);
>-	
>-	return ctdb_ctrl_pulldb_recv(ctdb, mem_ctx, state, outdata);
>-}
>-
>-
>-/*
>-  change dmaster for all keys in the database to the new value
>- */
>-int ctdb_ctrl_setdmaster(struct ctdb_context *ctdb, struct timeval timeout, uint32_t destnode, 
>-			 TALLOC_CTX *mem_ctx, uint32_t dbid, uint32_t dmaster)
>-{
>-	int ret;
>-	TDB_DATA indata;
>-	int32_t res;
>-
>-	indata.dsize = 2*sizeof(uint32_t);
>-	indata.dptr = (unsigned char *)talloc_array(mem_ctx, uint32_t, 2);
>-
>-	((uint32_t *)(&indata.dptr[0]))[0] = dbid;
>-	((uint32_t *)(&indata.dptr[0]))[1] = dmaster;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_SET_DMASTER, 0, indata, 
>-			   NULL, NULL, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for setdmaster failed\n"));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
>-/*
>-  ping a node, return number of clients connected
>- */
>-int ctdb_ctrl_ping(struct ctdb_context *ctdb, uint32_t destnode)
>-{
>-	int ret;
>-	int32_t res;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_PING, 0, 
>-			   tdb_null, NULL, NULL, &res, NULL, NULL);
>-	if (ret != 0) {
>-		return -1;
>-	}
>-	return res;
>-}
>-
>-int ctdb_ctrl_get_runstate(struct ctdb_context *ctdb, 
>-			   struct timeval timeout, 
>-			   uint32_t destnode,
>-			   uint32_t *runstate)
>-{
>-	TDB_DATA outdata;
>-	int32_t res;
>-	int ret;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_GET_RUNSTATE, 0,
>-			   tdb_null, ctdb, &outdata, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,("ctdb_control for get_runstate failed\n"));
>-		return ret != 0 ? ret : res;
>-	}
>-
>-	if (outdata.dsize != sizeof(uint32_t)) {
>-		DEBUG(DEBUG_ERR,("Invalid return data in get_runstate\n"));
>-		talloc_free(outdata.dptr);
>-		return -1;
>-	}
>-
>-	if (runstate != NULL) {
>-		*runstate = *(uint32_t *)outdata.dptr;
>-	}
>-	talloc_free(outdata.dptr);
>-
>-	return 0;
>-}
>-
>-/*
>-  find the real path to a ltdb 
>- */
>-int ctdb_ctrl_getdbpath(struct ctdb_context *ctdb, struct timeval timeout, uint32_t destnode, uint32_t dbid, TALLOC_CTX *mem_ctx, 
>-		   const char **path)
>-{
>-	int ret;
>-	int32_t res;
>-	TDB_DATA data;
>-
>-	data.dptr = (uint8_t *)&dbid;
>-	data.dsize = sizeof(dbid);
>-
>-	ret = ctdb_control(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_GETDBPATH, 0, data, 
>-			   mem_ctx, &data, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0) {
>-		return -1;
>-	}
>-
>-	(*path) = talloc_strndup(mem_ctx, (const char *)data.dptr, data.dsize);
>-	if ((*path) == NULL) {
>-		return -1;
>-	}
>-
>-	talloc_free(data.dptr);
>-
>-	return 0;
>-}
>-
>-/*
>-  find the name of a db 
>+  find the name of a db 
>  */
> int ctdb_ctrl_getdbname(struct ctdb_context *ctdb, struct timeval timeout, uint32_t destnode, uint32_t dbid, TALLOC_CTX *mem_ctx, 
> 		   const char **name)
>@@ -1835,80 +1168,6 @@ int ctdb_ctrl_getdbname(struct ctdb_context *ctdb, struct timeval timeout, uint3
> 	return 0;
> }
> 
>-/*
>-  get the health status of a db
>- */
>-int ctdb_ctrl_getdbhealth(struct ctdb_context *ctdb,
>-			  struct timeval timeout,
>-			  uint32_t destnode,
>-			  uint32_t dbid, TALLOC_CTX *mem_ctx,
>-			  const char **reason)
>-{
>-	int ret;
>-	int32_t res;
>-	TDB_DATA data;
>-
>-	data.dptr = (uint8_t *)&dbid;
>-	data.dsize = sizeof(dbid);
>-
>-	ret = ctdb_control(ctdb, destnode, 0,
>-			   CTDB_CONTROL_DB_GET_HEALTH, 0, data,
>-			   mem_ctx, &data, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0) {
>-		return -1;
>-	}
>-
>-	if (data.dsize == 0) {
>-		(*reason) = NULL;
>-		return 0;
>-	}
>-
>-	(*reason) = talloc_strndup(mem_ctx, (const char *)data.dptr, data.dsize);
>-	if ((*reason) == NULL) {
>-		return -1;
>-	}
>-
>-	talloc_free(data.dptr);
>-
>-	return 0;
>-}
>-
>-/*
>- * get db sequence number
>- */
>-int ctdb_ctrl_getdbseqnum(struct ctdb_context *ctdb, struct timeval timeout,
>-			  uint32_t destnode, uint32_t dbid, uint64_t *seqnum)
>-{
>-	int ret;
>-	int32_t res;
>-	TDB_DATA data, outdata;
>-	uint8_t buf[sizeof(uint64_t)] = { 0 };
>-
>-	*(uint32_t *)buf = dbid;
>-	data.dptr = buf;
>-	data.dsize = sizeof(uint64_t);
>-
>-	ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_GET_DB_SEQNUM,
>-			   0, data, ctdb, &outdata, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,("ctdb_control for getdbesqnum failed\n"));
>-		return -1;
>-	}
>-
>-	if (outdata.dsize != sizeof(uint64_t)) {
>-		DEBUG(DEBUG_ERR,("Invalid return data in get_dbseqnum\n"));
>-		talloc_free(outdata.dptr);
>-		return -1;
>-	}
>-
>-	if (seqnum != NULL) {
>-		*seqnum = *(uint64_t *)outdata.dptr;
>-	}
>-	talloc_free(outdata.dptr);
>-
>-	return 0;
>-}
>-
> /*
>   create a database
>  */
>@@ -1975,80 +1234,6 @@ int ctdb_ctrl_get_debuglevel(struct ctdb_context *ctdb, uint32_t destnode, int32
> 	return 0;
> }
> 
>-/*
>-  set debug level on a node
>- */
>-int ctdb_ctrl_set_debuglevel(struct ctdb_context *ctdb, uint32_t destnode, int32_t level)
>-{
>-	int ret;
>-	int32_t res;
>-	TDB_DATA data;
>-
>-	data.dptr = (uint8_t *)&level;
>-	data.dsize = sizeof(level);
>-
>-	ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_SET_DEBUG, 0, data, 
>-			   NULL, NULL, &res, NULL, NULL);
>-	if (ret != 0 || res != 0) {
>-		return -1;
>-	}
>-	return 0;
>-}
>-
>-
>-/*
>-  get a list of connected nodes
>- */
>-uint32_t *ctdb_get_connected_nodes(struct ctdb_context *ctdb, 
>-				struct timeval timeout,
>-				TALLOC_CTX *mem_ctx,
>-				uint32_t *num_nodes)
>-{
>-	struct ctdb_node_map_old *map=NULL;
>-	int ret, i;
>-	uint32_t *nodes;
>-
>-	*num_nodes = 0;
>-
>-	ret = ctdb_ctrl_getnodemap(ctdb, timeout, CTDB_CURRENT_NODE, mem_ctx, &map);
>-	if (ret != 0) {
>-		return NULL;
>-	}
>-
>-	nodes = talloc_array(mem_ctx, uint32_t, map->num);
>-	if (nodes == NULL) {
>-		return NULL;
>-	}
>-
>-	for (i=0;i<map->num;i++) {
>-		if (!(map->nodes[i].flags & NODE_FLAGS_DISCONNECTED)) {
>-			nodes[*num_nodes] = map->nodes[i].pnn;
>-			(*num_nodes)++;
>-		}
>-	}
>-
>-	return nodes;
>-}
>-
>-
>-/*
>-  reset remote status
>- */
>-int ctdb_statistics_reset(struct ctdb_context *ctdb, uint32_t destnode)
>-{
>-	int ret;
>-	int32_t res;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_STATISTICS_RESET, 0, tdb_null, 
>-			   NULL, NULL, &res, NULL, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for reset statistics failed\n"));
>-		return -1;
>-	}
>-	return 0;
>-}
>-
> /*
>  * Get db open flags
>  */
>@@ -2149,26 +1334,6 @@ struct ctdb_db_context *ctdb_attach(struct ctdb_context *ctdb,
> 	return ctdb_db;
> }
> 
>-/*
>- * detach from a specific database - client call
>- */
>-int ctdb_detach(struct ctdb_context *ctdb, uint32_t db_id)
>-{
>-	int ret;
>-	int32_t status;
>-	TDB_DATA data;
>-
>-	data.dsize = sizeof(db_id);
>-	data.dptr = (uint8_t *)&db_id;
>-
>-	ret = ctdb_control(ctdb, CTDB_CURRENT_NODE, 0, CTDB_CONTROL_DB_DETACH,
>-			   0, data, NULL, NULL, &status, NULL, NULL);
>-	if (ret != 0 || status != 0) {
>-		return -1;
>-	}
>-	return 0;
>-}
>-
> /*
>   setup a call for a database
>  */
>@@ -2185,222 +1350,6 @@ int ctdb_set_call(struct ctdb_db_context *ctdb_db, ctdb_fn_t fn, uint32_t id)
> 	return 0;
> }
> 
>-
>-struct traverse_state {
>-	bool done;
>-	uint32_t count;
>-	ctdb_traverse_func fn;
>-	void *private_data;
>-	bool listemptyrecords;
>-};
>-
>-/*
>-  called on each key during a ctdb_traverse
>- */
>-static void traverse_handler(uint64_t srvid, TDB_DATA data, void *p)
>-{
>-	struct traverse_state *state = (struct traverse_state *)p;
>-	struct ctdb_rec_data_old *d = (struct ctdb_rec_data_old *)data.dptr;
>-	TDB_DATA key;
>-
>-	if (data.dsize < sizeof(uint32_t) || d->length != data.dsize) {
>-		DEBUG(DEBUG_ERR, ("Bad data size %u in traverse_handler\n",
>-				  (unsigned)data.dsize));
>-		state->done = true;
>-		return;
>-	}
>-
>-	key.dsize = d->keylen;
>-	key.dptr  = &d->data[0];
>-	data.dsize = d->datalen;
>-	data.dptr = &d->data[d->keylen];
>-
>-	if (key.dsize == 0 && data.dsize == 0) {
>-		/* end of traverse */
>-		state->done = true;
>-		return;
>-	}
>-
>-	if (!state->listemptyrecords &&
>-	    data.dsize == sizeof(struct ctdb_ltdb_header))
>-	{
>-		/* empty records are deleted records in ctdb */
>-		return;
>-	}
>-
>-	if (state->fn(key, data, state->private_data) != 0) {
>-		state->done = true;
>-	}
>-
>-	state->count++;
>-}
>-
>-/**
>- * start a cluster wide traverse, calling the supplied fn on each record
>- * return the number of records traversed, or -1 on error
>- *
>- * Extendet variant with a flag to signal whether empty records should
>- * be listed.
>- */
>-static int ctdb_traverse_ext(struct ctdb_db_context *ctdb_db,
>-			     ctdb_traverse_func fn,
>-			     bool withemptyrecords,
>-			     void *private_data)
>-{
>-	TDB_DATA data;
>-	struct ctdb_traverse_start_ext t;
>-	int32_t status;
>-	int ret;
>-	uint64_t srvid = (getpid() | 0xFLL<<60);
>-	struct traverse_state state;
>-
>-	state.done = false;
>-	state.count = 0;
>-	state.private_data = private_data;
>-	state.fn = fn;
>-	state.listemptyrecords = withemptyrecords;
>-
>-	ret = ctdb_client_set_message_handler(ctdb_db->ctdb, srvid, traverse_handler, &state);
>-	if (ret != 0) {
>-		DEBUG(DEBUG_ERR,("Failed to setup traverse handler\n"));
>-		return -1;
>-	}
>-
>-	t.db_id = ctdb_db->db_id;
>-	t.srvid = srvid;
>-	t.reqid = 0;
>-	t.withemptyrecords = withemptyrecords;
>-
>-	data.dptr = (uint8_t *)&t;
>-	data.dsize = sizeof(t);
>-
>-	ret = ctdb_control(ctdb_db->ctdb, CTDB_CURRENT_NODE, 0, CTDB_CONTROL_TRAVERSE_START_EXT, 0,
>-			   data, NULL, NULL, &status, NULL, NULL);
>-	if (ret != 0 || status != 0) {
>-		DEBUG(DEBUG_ERR,("ctdb_traverse_all failed\n"));
>-		ctdb_client_remove_message_handler(ctdb_db->ctdb, srvid, &state);
>-		return -1;
>-	}
>-
>-	while (!state.done) {
>-		tevent_loop_once(ctdb_db->ctdb->ev);
>-	}
>-
>-	ret = ctdb_client_remove_message_handler(ctdb_db->ctdb, srvid, &state);
>-	if (ret != 0) {
>-		DEBUG(DEBUG_ERR,("Failed to remove ctdb_traverse handler\n"));
>-		return -1;
>-	}
>-
>-	return state.count;
>-}
>-
>-/**
>- * start a cluster wide traverse, calling the supplied fn on each record
>- * return the number of records traversed, or -1 on error
>- *
>- * Standard version which does not list the empty records:
>- * These are considered deleted.
>- */
>-int ctdb_traverse(struct ctdb_db_context *ctdb_db, ctdb_traverse_func fn, void *private_data)
>-{
>-	return ctdb_traverse_ext(ctdb_db, fn, false, private_data);
>-}
>-
>-#define ISASCII(x) (isprint(x) && !strchr("\"\\", (x)))
>-/*
>-  called on each key during a catdb
>- */
>-int ctdb_dumpdb_record(TDB_DATA key, TDB_DATA data, void *p)
>-{
>-	int i;
>-	struct ctdb_dump_db_context *c = (struct ctdb_dump_db_context *)p;
>-	FILE *f = c->f;
>-	struct ctdb_ltdb_header *h = (struct ctdb_ltdb_header *)data.dptr;
>-
>-	fprintf(f, "key(%u) = \"", (unsigned)key.dsize);
>-	for (i=0;i<key.dsize;i++) {
>-		if (ISASCII(key.dptr[i])) {
>-			fprintf(f, "%c", key.dptr[i]);
>-		} else {
>-			fprintf(f, "\\%02X", key.dptr[i]);
>-		}
>-	}
>-	fprintf(f, "\"\n");
>-
>-	fprintf(f, "dmaster: %u\n", h->dmaster);
>-	fprintf(f, "rsn: %llu\n", (unsigned long long)h->rsn);
>-
>-	if (c->printlmaster && c->ctdb->vnn_map != NULL) {
>-		fprintf(f, "lmaster: %u\n", ctdb_lmaster(c->ctdb, &key));
>-	}
>-
>-	if (c->printhash) {
>-		fprintf(f, "hash: 0x%08x\n", ctdb_hash(&key));
>-	}
>-
>-	if (c->printrecordflags) {
>-		fprintf(f, "flags: 0x%08x", h->flags);
>-		if (h->flags & CTDB_REC_FLAG_MIGRATED_WITH_DATA) printf(" MIGRATED_WITH_DATA");
>-		if (h->flags & CTDB_REC_FLAG_VACUUM_MIGRATED) printf(" VACUUM_MIGRATED");
>-		if (h->flags & CTDB_REC_FLAG_AUTOMATIC) printf(" AUTOMATIC");
>-		if (h->flags & CTDB_REC_RO_HAVE_DELEGATIONS) printf(" RO_HAVE_DELEGATIONS");
>-		if (h->flags & CTDB_REC_RO_HAVE_READONLY) printf(" RO_HAVE_READONLY");
>-		if (h->flags & CTDB_REC_RO_REVOKING_READONLY) printf(" RO_REVOKING_READONLY");
>-		if (h->flags & CTDB_REC_RO_REVOKE_COMPLETE) printf(" RO_REVOKE_COMPLETE");
>-		fprintf(f, "\n");
>-	}
>-
>-	if (c->printdatasize) {
>-		fprintf(f, "data size: %u\n", (unsigned)data.dsize);
>-	} else {
>-		fprintf(f, "data(%u) = \"", (unsigned)(data.dsize - sizeof(*h)));
>-		for (i=sizeof(*h);i<data.dsize;i++) {
>-			if (ISASCII(data.dptr[i])) {
>-				fprintf(f, "%c", data.dptr[i]);
>-			} else {
>-				fprintf(f, "\\%02X", data.dptr[i]);
>-			}
>-		}
>-		fprintf(f, "\"\n");
>-	}
>-
>-	fprintf(f, "\n");
>-
>-	return 0;
>-}
>-
>-/*
>-  convenience function to list all keys to stdout
>- */
>-int ctdb_dump_db(struct ctdb_db_context *ctdb_db,
>-		 struct ctdb_dump_db_context *ctx)
>-{
>-	return ctdb_traverse_ext(ctdb_db, ctdb_dumpdb_record,
>-				 ctx->printemptyrecords, ctx);
>-}
>-
>-/*
>-  get the pid of a ctdb daemon
>- */
>-int ctdb_ctrl_getpid(struct ctdb_context *ctdb, struct timeval timeout, uint32_t destnode, uint32_t *pid)
>-{
>-	int ret;
>-	int32_t res;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_GET_PID, 0, tdb_null, 
>-			   NULL, NULL, &res, &timeout, NULL);
>-	if (ret != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for getpid failed\n"));
>-		return -1;
>-	}
>-
>-	*pid = res;
>-
>-	return 0;
>-}
>-
> /* Freeze all databases */
> int ctdb_ctrl_freeze(struct ctdb_context *ctdb, struct timeval timeout,
> 		     uint32_t destnode)
>@@ -2438,181 +1387,6 @@ int ctdb_ctrl_getpnn(struct ctdb_context *ctdb, struct timeval timeout, uint32_t
> 	return res;
> }
> 
>-
>-/*
>-  sent to a node to make it take over an ip address
>-*/
>-int ctdb_ctrl_takeover_ip(struct ctdb_context *ctdb, struct timeval timeout,
>-			  uint32_t destnode, struct ctdb_public_ip *ip)
>-{
>-	TDB_DATA data;
>-	int ret;
>-	int32_t res;
>-
>-	data.dsize = sizeof(*ip);
>-	data.dptr  = (uint8_t *)ip;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_TAKEOVER_IP, 0,
>-			   data, NULL, NULL, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for takeover_ip failed\n"));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
>-
>-/*
>-  sent to a node to make it release an ip address
>-*/
>-int ctdb_ctrl_release_ip(struct ctdb_context *ctdb, struct timeval timeout,
>-			 uint32_t destnode, struct ctdb_public_ip *ip)
>-{
>-	TDB_DATA data;
>-	int ret;
>-	int32_t res;
>-
>-	data.dsize = sizeof(*ip);
>-	data.dptr  = (uint8_t *)ip;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_RELEASE_IP, 0,
>-			   data, NULL, NULL, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for release_ip failed\n"));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
>-
>-/*
>-  get a tunable
>- */
>-int ctdb_ctrl_get_tunable(struct ctdb_context *ctdb, 
>-			  struct timeval timeout, 
>-			  uint32_t destnode,
>-			  const char *name, uint32_t *value)
>-{
>-	struct ctdb_control_get_tunable *t;
>-	TDB_DATA data, outdata;
>-	int32_t res;
>-	int ret;
>-
>-	data.dsize = offsetof(struct ctdb_control_get_tunable, name) + strlen(name) + 1;
>-	data.dptr  = talloc_size(ctdb, data.dsize);
>-	CTDB_NO_MEMORY(ctdb, data.dptr);
>-
>-	t = (struct ctdb_control_get_tunable *)data.dptr;
>-	t->length = strlen(name)+1;
>-	memcpy(t->name, name, t->length);
>-
>-	ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_GET_TUNABLE, 0, data, ctdb,
>-			   &outdata, &res, &timeout, NULL);
>-	talloc_free(data.dptr);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for get_tunable failed\n"));
>-		return ret != 0 ? ret : res;
>-	}
>-
>-	if (outdata.dsize != sizeof(uint32_t)) {
>-		DEBUG(DEBUG_ERR,("Invalid return data in get_tunable\n"));
>-		talloc_free(outdata.dptr);
>-		return -1;
>-	}
>-	
>-	*value = *(uint32_t *)outdata.dptr;
>-	talloc_free(outdata.dptr);
>-
>-	return 0;
>-}
>-
>-/*
>-  set a tunable
>- */
>-int ctdb_ctrl_set_tunable(struct ctdb_context *ctdb, 
>-			  struct timeval timeout, 
>-			  uint32_t destnode,
>-			  const char *name, uint32_t value)
>-{
>-	struct ctdb_tunable_old *t;
>-	TDB_DATA data;
>-	int32_t res;
>-	int ret;
>-
>-	data.dsize = offsetof(struct ctdb_tunable_old, name) + strlen(name) + 1;
>-	data.dptr  = talloc_size(ctdb, data.dsize);
>-	CTDB_NO_MEMORY(ctdb, data.dptr);
>-
>-	t = (struct ctdb_tunable_old *)data.dptr;
>-	t->length = strlen(name)+1;
>-	memcpy(t->name, name, t->length);
>-	t->value = value;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_SET_TUNABLE, 0, data, NULL,
>-			   NULL, &res, &timeout, NULL);
>-	talloc_free(data.dptr);
>-	if ((ret != 0) || (res == -1)) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for set_tunable failed\n"));
>-		return -1;
>-	}
>-
>-	return res;
>-}
>-
>-/*
>-  list tunables
>- */
>-int ctdb_ctrl_list_tunables(struct ctdb_context *ctdb, 
>-			    struct timeval timeout, 
>-			    uint32_t destnode,
>-			    TALLOC_CTX *mem_ctx,
>-			    const char ***list, uint32_t *count)
>-{
>-	TDB_DATA outdata;
>-	int32_t res;
>-	int ret;
>-	struct ctdb_control_list_tunable *t;
>-	char *p, *s, *ptr;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_LIST_TUNABLES, 0, tdb_null, 
>-			   mem_ctx, &outdata, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for list_tunables failed\n"));
>-		return -1;
>-	}
>-
>-	t = (struct ctdb_control_list_tunable *)outdata.dptr;
>-	if (outdata.dsize < offsetof(struct ctdb_control_list_tunable, data) ||
>-	    t->length > outdata.dsize-offsetof(struct ctdb_control_list_tunable, data)) {
>-		DEBUG(DEBUG_ERR,("Invalid data in list_tunables reply\n"));
>-		talloc_free(outdata.dptr);
>-		return -1;		
>-	}
>-	
>-	p = talloc_strndup(mem_ctx, (char *)t->data, t->length);
>-	CTDB_NO_MEMORY(ctdb, p);
>-
>-	talloc_free(outdata.dptr);
>-	
>-	(*list) = NULL;
>-	(*count) = 0;
>-
>-	for (s=strtok_r(p, ":", &ptr); s; s=strtok_r(NULL, ":", &ptr)) {
>-		(*list) = talloc_realloc(mem_ctx, *list, const char *, 1+(*count));
>-		CTDB_NO_MEMORY(ctdb, *list);
>-		(*list)[*count] = talloc_strdup(*list, s);
>-		CTDB_NO_MEMORY(ctdb, (*list)[*count]);
>-		(*count)++;
>-	}
>-
>-	talloc_free(p);
>-
>-	return 0;
>-}
>-
>-
> int ctdb_ctrl_get_public_ips_flags(struct ctdb_context *ctdb,
> 				   struct timeval timeout, uint32_t destnode,
> 				   TALLOC_CTX *mem_ctx,
>@@ -2649,74 +1423,6 @@ int ctdb_ctrl_get_public_ips(struct ctdb_context *ctdb,
> 					      0, ips);
> }
> 
>-int ctdb_ctrl_get_public_ip_info(struct ctdb_context *ctdb,
>-				 struct timeval timeout, uint32_t destnode,
>-				 TALLOC_CTX *mem_ctx,
>-				 const ctdb_sock_addr *addr,
>-				 struct ctdb_public_ip_info_old **_info)
>-{
>-	int ret;
>-	TDB_DATA indata;
>-	TDB_DATA outdata;
>-	int32_t res;
>-	struct ctdb_public_ip_info_old *info;
>-	uint32_t len;
>-	uint32_t i;
>-
>-	indata.dptr = discard_const_p(uint8_t, addr);
>-	indata.dsize = sizeof(*addr);
>-
>-	ret = ctdb_control(ctdb, destnode, 0,
>-			   CTDB_CONTROL_GET_PUBLIC_IP_INFO, 0, indata,
>-			   mem_ctx, &outdata, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for get public ip info "
>-				"failed ret:%d res:%d\n",
>-				ret, res));
>-		return -1;
>-	}
>-
>-	len = offsetof(struct ctdb_public_ip_info_old, ifaces);
>-	if (len > outdata.dsize) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for get public ip info "
>-				"returned invalid data with size %u > %u\n",
>-				(unsigned int)outdata.dsize,
>-				(unsigned int)len));
>-		dump_data(DEBUG_DEBUG, outdata.dptr, outdata.dsize);
>-		return -1;
>-	}
>-
>-	info = (struct ctdb_public_ip_info_old *)outdata.dptr;
>-	len += info->num*sizeof(struct ctdb_iface);
>-
>-	if (len > outdata.dsize) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for get public ip info "
>-				"returned invalid data with size %u > %u\n",
>-				(unsigned int)outdata.dsize,
>-				(unsigned int)len));
>-		dump_data(DEBUG_DEBUG, outdata.dptr, outdata.dsize);
>-		return -1;
>-	}
>-
>-	/* make sure we null terminate the returned strings */
>-	for (i=0; i < info->num; i++) {
>-		info->ifaces[i].name[CTDB_IFACE_SIZE] = '\0';
>-	}
>-
>-	*_info = (struct ctdb_public_ip_info_old *)talloc_memdup(mem_ctx,
>-								outdata.dptr,
>-								outdata.dsize);
>-	talloc_free(outdata.dptr);
>-	if (*_info == NULL) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for get public ip info "
>-				"talloc_memdup size %u failed\n",
>-				(unsigned int)outdata.dsize));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
> int ctdb_ctrl_get_ifaces(struct ctdb_context *ctdb,
> 			 struct timeval timeout, uint32_t destnode,
> 			 TALLOC_CTX *mem_ctx,
>@@ -2780,31 +1486,6 @@ int ctdb_ctrl_get_ifaces(struct ctdb_context *ctdb,
> 	return 0;
> }
> 
>-int ctdb_ctrl_set_iface_link(struct ctdb_context *ctdb,
>-			     struct timeval timeout, uint32_t destnode,
>-			     TALLOC_CTX *mem_ctx,
>-			     const struct ctdb_iface *info)
>-{
>-	int ret;
>-	TDB_DATA indata;
>-	int32_t res;
>-
>-	indata.dptr = discard_const_p(uint8_t, info);
>-	indata.dsize = sizeof(*info);
>-
>-	ret = ctdb_control(ctdb, destnode, 0,
>-			   CTDB_CONTROL_SET_IFACE_LINK_STATE, 0, indata,
>-			   mem_ctx, NULL, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for set iface link "
>-				"failed ret:%d res:%d\n",
>-				ret, res));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
> /*
>   set/clear the permanent disabled bit on a remote node
>  */
>@@ -2901,123 +1582,7 @@ int ctdb_ctrl_get_all_tunables(struct ctdb_context *ctdb,
> }
> 
> /*
>-  add a public address to a node
>- */
>-int ctdb_ctrl_add_public_ip(struct ctdb_context *ctdb,
>-			    struct timeval timeout, uint32_t destnode,
>-			    struct ctdb_addr_info_old *pub)
>-{
>-	TDB_DATA data;
>-	int32_t res;
>-	int ret;
>-
>-	data.dsize = offsetof(struct ctdb_addr_info_old, iface) + pub->len;
>-	data.dptr  = (unsigned char *)pub;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_ADD_PUBLIC_IP, 0, data, NULL,
>-			   NULL, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for add_public_ip failed\n"));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
>-/*
>-  delete a public address from a node
>- */
>-int ctdb_ctrl_del_public_ip(struct ctdb_context *ctdb,
>-			    struct timeval timeout, uint32_t destnode,
>-			    struct ctdb_addr_info_old *pub)
>-{
>-	TDB_DATA data;
>-	int32_t res;
>-	int ret;
>-
>-	data.dsize = offsetof(struct ctdb_addr_info_old, iface) + pub->len;
>-	data.dptr  = (unsigned char *)pub;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_DEL_PUBLIC_IP, 0, data, NULL,
>-			   NULL, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for del_public_ip failed\n"));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
>-/*
>-  send a gratious arp
>- */
>-int ctdb_ctrl_gratious_arp(struct ctdb_context *ctdb,
>-			   struct timeval timeout, uint32_t destnode,
>-			   ctdb_sock_addr *addr, const char *ifname)
>-{
>-	TDB_DATA data;
>-	int32_t res;
>-	int ret, len;
>-	struct ctdb_addr_info_old *gratious_arp;
>-	TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
>-
>-
>-	len = strlen(ifname)+1;
>-	gratious_arp = talloc_size(tmp_ctx,
>-		offsetof(struct ctdb_addr_info_old, iface) + len);
>-	CTDB_NO_MEMORY(ctdb, gratious_arp);
>-
>-	gratious_arp->addr = *addr;
>-	gratious_arp->len = len;
>-	memcpy(&gratious_arp->iface[0], ifname, len);
>-
>-
>-	data.dsize = offsetof(struct ctdb_addr_info_old, iface) + len;
>-	data.dptr  = (unsigned char *)gratious_arp;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_SEND_GRATUITOUS_ARP, 0, data, NULL,
>-			   NULL, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for gratious_arp failed\n"));
>-		talloc_free(tmp_ctx);
>-		return -1;
>-	}
>-
>-	talloc_free(tmp_ctx);
>-	return 0;
>-}
>-
>-/*
>-  get a list of all tcp tickles that a node knows about for a particular vnn
>- */
>-int ctdb_ctrl_get_tcp_tickles(struct ctdb_context *ctdb, 
>-			      struct timeval timeout, uint32_t destnode, 
>-			      TALLOC_CTX *mem_ctx, 
>-			      ctdb_sock_addr *addr,
>-			      struct ctdb_tickle_list_old **list)
>-{
>-	int ret;
>-	TDB_DATA data, outdata;
>-	int32_t status;
>-
>-	data.dptr = (uint8_t*)addr;
>-	data.dsize = sizeof(ctdb_sock_addr);
>-
>-	ret = ctdb_control(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_GET_TCP_TICKLE_LIST, 0, data, 
>-			   mem_ctx, &outdata, &status, NULL, NULL);
>-	if (ret != 0 || status != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for get tcp tickles failed\n"));
>-		return -1;
>-	}
>-
>-	*list = (struct ctdb_tickle_list_old *)outdata.dptr;
>-
>-	return status;
>-}
>-
>-/*
>-  initialise the ctdb daemon for client applications
>+  initialise the ctdb daemon for client applications
> 
>   NOTE: In current code the daemon does not fork. This is for testing purposes only
>   and to simplify the code.
>@@ -3093,62 +1658,6 @@ uint32_t ctdb_get_pnn(struct ctdb_context *ctdb)
> 	return ctdb->pnn;
> }
> 
>-
>-/*
>-  get the uptime of a remote node
>- */
>-struct ctdb_client_control_state *
>-ctdb_ctrl_uptime_send(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx, struct timeval timeout, uint32_t destnode)
>-{
>-	return ctdb_control_send(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_UPTIME, 0, tdb_null, 
>-			   mem_ctx, &timeout, NULL);
>-}
>-
>-int ctdb_ctrl_uptime_recv(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx, struct ctdb_client_control_state *state, struct ctdb_uptime **uptime)
>-{
>-	int ret;
>-	int32_t res;
>-	TDB_DATA outdata;
>-
>-	ret = ctdb_control_recv(ctdb, state, mem_ctx, &outdata, &res, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_ctrl_uptime_recv failed\n"));
>-		return -1;
>-	}
>-
>-	*uptime = (struct ctdb_uptime *)talloc_steal(mem_ctx, outdata.dptr);
>-
>-	return 0;
>-}
>-
>-int ctdb_ctrl_uptime(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx, struct timeval timeout, uint32_t destnode, struct ctdb_uptime **uptime)
>-{
>-	struct ctdb_client_control_state *state;
>-
>-	state = ctdb_ctrl_uptime_send(ctdb, mem_ctx, timeout, destnode);
>-	return ctdb_ctrl_uptime_recv(ctdb, mem_ctx, state, uptime);
>-}
>-
>-/*
>-  send a control to execute the "recovered" event script on a node
>- */
>-int ctdb_ctrl_end_recovery(struct ctdb_context *ctdb, struct timeval timeout, uint32_t destnode)
>-{
>-	int ret;
>-	int32_t status;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_END_RECOVERY, 0, tdb_null, 
>-			   NULL, NULL, &status, &timeout, NULL);
>-	if (ret != 0 || status != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for end_recovery failed\n"));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
> /* 
>   callback for the async helpers used when sending the same control
>   to multiple nodes in parallell.
>@@ -3375,40 +1884,6 @@ uint32_t *list_of_connected_nodes(struct ctdb_context *ctdb,
> 			     include_self ? -1 : ctdb->pnn);
> }
> 
>-/* 
>-  this is used to test if a pnn lock exists and if it exists will return
>-  the number of connections that pnn has reported or -1 if that recovery
>-  daemon is not running.
>-*/
>-int
>-ctdb_read_pnn_lock(int fd, int32_t pnn)
>-{
>-	struct flock lock;
>-	char c;
>-
>-	lock.l_type = F_WRLCK;
>-	lock.l_whence = SEEK_SET;
>-	lock.l_start = pnn;
>-	lock.l_len = 1;
>-	lock.l_pid = 0;
>-
>-	if (fcntl(fd, F_GETLK, &lock) != 0) {
>-		DEBUG(DEBUG_ERR, (__location__ " F_GETLK failed with %s\n", strerror(errno)));
>-		return -1;
>-	}
>-
>-	if (lock.l_type == F_UNLCK) {
>-		return -1;
>-	}
>-
>-	if (pread(fd, &c, 1, pnn) == -1) {
>-		DEBUG(DEBUG_CRIT,(__location__ " failed read pnn count - %s\n", strerror(errno)));
>-		return -1;
>-	}
>-
>-	return c;
>-}
>-
> /*
>   get capabilities of a remote node
>  */
>@@ -3528,904 +2003,128 @@ bool ctdb_node_has_capabilities(struct ctdb_node_capabilities *caps,
> 		((*capp & capabilities_required) == capabilities_required);
> }
> 
>-
>-static struct ctdb_server_id server_id_fetch(struct ctdb_context *ctdb, uint32_t reqid)
>-{
>-	struct ctdb_server_id id;
>-
>-	id.pid = getpid();
>-	id.task_id = reqid;
>-	id.vnn = ctdb_get_pnn(ctdb);
>-	id.unique_id = id.vnn;
>-	id.unique_id = (id.unique_id << 32) | reqid;
>-
>-	return id;
>-}
>-
>-/* This is basically a copy from Samba's server_id.*.  However, a
>- * dependency chain stops us from using Samba's version, so use a
>- * renamed copy until a better solution is found. */
>-static bool ctdb_server_id_equal(struct ctdb_server_id *id1, struct ctdb_server_id *id2)
>+/*
>+  recovery daemon ping to main daemon
>+ */
>+int ctdb_ctrl_recd_ping(struct ctdb_context *ctdb)
> {
>-	if (id1->pid != id2->pid) {
>-		return false;
>-	}
>-
>-	if (id1->task_id != id2->task_id) {
>-		return false;
>-	}
>-
>-	if (id1->vnn != id2->vnn) {
>-		return false;
>-	}
>+	int ret;
>+	int32_t res;
> 
>-	if (id1->unique_id != id2->unique_id) {
>-		return false;
>+	ret = ctdb_control(ctdb, CTDB_CURRENT_NODE, 0, CTDB_CONTROL_RECD_PING, 0, tdb_null,
>+			   ctdb, NULL, &res, NULL, NULL);
>+	if (ret != 0 || res != 0) {
>+		DEBUG(DEBUG_ERR,("Failed to send recd ping\n"));
>+		return -1;
> 	}
> 
>-	return true;
>+	return 0;
> }
> 
>-static bool server_id_exists(struct ctdb_context *ctdb,
>-			     struct ctdb_server_id *id)
>+/*
>+  tell the main daemon how long it took to lock the reclock file
>+ */
>+int ctdb_ctrl_report_recd_lock_latency(struct ctdb_context *ctdb, struct timeval timeout, double latency)
> {
> 	int ret;
>+	int32_t res;
>+	TDB_DATA data;
>+
>+	data.dptr = (uint8_t *)&latency;
>+	data.dsize = sizeof(latency);
> 
>-	ret = ctdb_ctrl_process_exists(ctdb, id->vnn, id->pid);
>-	if (ret == 0) {
>-		return true;
>+	ret = ctdb_control(ctdb, CTDB_CURRENT_NODE, 0, CTDB_CONTROL_RECD_RECLOCK_LATENCY, 0, data,
>+			   ctdb, NULL, &res, NULL, NULL);
>+	if (ret != 0 || res != 0) {
>+		DEBUG(DEBUG_ERR,("Failed to send recd reclock latency\n"));
>+		return -1;
> 	}
> 
>-	return false;
>+	return 0;
> }
> 
>-static bool g_lock_parse(TALLOC_CTX *mem_ctx, TDB_DATA data,
>-			 struct ctdb_g_lock_list **locks)
>+int ctdb_ctrl_set_ban(struct ctdb_context *ctdb, struct timeval timeout,
>+		      uint32_t destnode, struct ctdb_ban_state *bantime)
> {
>-	struct ctdb_g_lock_list *recs;
>-
>-	recs = talloc_zero(mem_ctx, struct ctdb_g_lock_list);
>-	if (recs == NULL) {
>-		return false;
>-	}
>-
>-	if (data.dsize == 0) {
>-		goto done;
>-	}
>-
>-	if (data.dsize % sizeof(struct ctdb_g_lock) != 0) {
>-		DEBUG(DEBUG_ERR, (__location__ "invalid data size %lu in g_lock record\n",
>-				  (unsigned long)data.dsize));
>-		talloc_free(recs);
>-		return false;
>-	}
>+	int ret;
>+	TDB_DATA data;
>+	int32_t res;
> 
>-	recs->num = data.dsize / sizeof(struct ctdb_g_lock);
>-	recs->lock = talloc_memdup(mem_ctx, data.dptr, data.dsize);
>-	if (recs->lock == NULL) {
>-		talloc_free(recs);
>-		return false;
>-	}
>+	data.dsize = sizeof(*bantime);
>+	data.dptr  = (uint8_t *)bantime;
> 
>-done:
>-	if (locks != NULL) {
>-		*locks = recs;
>+	ret = ctdb_control(ctdb, destnode, 0,
>+			   CTDB_CONTROL_SET_BAN_STATE, 0, data,
>+			   NULL, NULL, &res, &timeout, NULL);
>+	if (ret != 0 || res != 0) {
>+		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for set ban state failed\n"));
>+		return -1;
> 	}
> 
>-	return true;
>+	return 0;
> }
> 
>-
>-static bool g_lock_lock(TALLOC_CTX *mem_ctx,
>-			struct ctdb_db_context *ctdb_db,
>-			const char *keyname, uint32_t reqid)
>+struct ctdb_client_control_state *
>+ctdb_ctrl_updaterecord_send(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx, struct timeval timeout, uint32_t destnode, struct ctdb_db_context *ctdb_db, TDB_DATA key, struct ctdb_ltdb_header *header, TDB_DATA data)
> {
>-	TDB_DATA key, data;
>-	struct ctdb_record_handle *h;
>-	struct ctdb_g_lock_list *locks;
>-	struct ctdb_server_id id;
>-	struct timeval t_start;
>-	int i;
>-
>-	key.dptr = (uint8_t *)discard_const(keyname);
>-	key.dsize = strlen(keyname) + 1;
>-
>-	t_start = timeval_current();
>-
>-again:
>-	/* Keep trying for an hour. */
>-	if (timeval_elapsed(&t_start) > 3600) {
>-		return false;
>-	}
>-
>-	h = ctdb_fetch_lock(ctdb_db, mem_ctx, key, &data);
>-	if (h == NULL) {
>-		return false;
>-	}
>+	struct ctdb_client_control_state *handle;
>+	struct ctdb_marshall_buffer *m;
>+	struct ctdb_rec_data_old *rec;
>+	TDB_DATA outdata;
> 
>-	if (!g_lock_parse(h, data, &locks)) {
>-		DEBUG(DEBUG_ERR, ("g_lock: error parsing locks\n"));
>-		talloc_free(data.dptr);
>-		talloc_free(h);
>-		return false;
>+	m = talloc_zero(mem_ctx, struct ctdb_marshall_buffer);
>+	if (m == NULL) {
>+		DEBUG(DEBUG_ERR, ("Failed to allocate marshall buffer for update record\n"));
>+		return NULL;
> 	}
> 
>-	talloc_free(data.dptr);
>-
>-	id = server_id_fetch(ctdb_db->ctdb, reqid);
>-
>-	i = 0;
>-	while (i < locks->num) {
>-		if (ctdb_server_id_equal(&locks->lock[i].sid, &id)) {
>-			/* Internal error */
>-			talloc_free(h);
>-			return false;
>-		}
>-
>-		if (!server_id_exists(ctdb_db->ctdb, &locks->lock[i].sid)) {
>-			if (i < locks->num-1) {
>-				locks->lock[i] = locks->lock[locks->num-1];
>-			}
>-			locks->num--;
>-			continue;
>-		}
>+	m->db_id = ctdb_db->db_id;
> 
>-		/* This entry is locked. */
>-		DEBUG(DEBUG_INFO, ("g_lock: lock already granted for "
>-				   "pid=0x%llx taskid=%x vnn=%d id=0x%llx\n",
>-				   (unsigned long long)id.pid,
>-				   id.task_id, id.vnn,
>-				   (unsigned long long)id.unique_id));
>-		talloc_free(h);
>-		goto again;
>+	rec = ctdb_marshall_record(m, 0, key, header, data);
>+	if (rec == NULL) {
>+		DEBUG(DEBUG_ERR,("Failed to marshall record for update record\n"));
>+		talloc_free(m);
>+		return NULL;
> 	}
>-
>-	locks->lock = talloc_realloc(locks, locks->lock, struct ctdb_g_lock,
>-				     locks->num+1);
>-	if (locks->lock == NULL) {
>-		talloc_free(h);
>-		return false;
>+	m = talloc_realloc_size(mem_ctx, m, rec->length + offsetof(struct ctdb_marshall_buffer, data));
>+	if (m == NULL) {
>+		DEBUG(DEBUG_CRIT,(__location__ " Failed to expand recdata\n"));
>+		talloc_free(m);
>+		return NULL;
> 	}
>+	m->count++;
>+	memcpy((uint8_t *)m + offsetof(struct ctdb_marshall_buffer, data), rec, rec->length);
> 
>-	locks->lock[locks->num].type = CTDB_G_LOCK_WRITE;
>-	locks->lock[locks->num].sid = id;
>-	locks->num++;
>-
>-	data.dptr = (uint8_t *)locks->lock;
>-	data.dsize = locks->num * sizeof(struct ctdb_g_lock);
>-
>-	if (ctdb_record_store(h, data) != 0) {
>-		DEBUG(DEBUG_ERR, ("g_lock: failed to write transaction lock for "
>-				  "pid=0x%llx taskid=%x vnn=%d id=0x%llx\n",
>-				  (unsigned long long)id.pid,
>-				  id.task_id, id.vnn,
>-				  (unsigned long long)id.unique_id));
>-		talloc_free(h);
>-		return false;
>-	}
> 
>-	DEBUG(DEBUG_INFO, ("g_lock: lock granted for "
>-			   "pid=0x%llx taskid=%x vnn=%d id=0x%llx\n",
>-			   (unsigned long long)id.pid,
>-			   id.task_id, id.vnn,
>-			   (unsigned long long)id.unique_id));
>+	outdata.dptr = (uint8_t *)m;
>+	outdata.dsize = talloc_get_size(m);
> 
>-	talloc_free(h);
>-	return true;
>+	handle = ctdb_control_send(ctdb, destnode, 0,
>+			   CTDB_CONTROL_UPDATE_RECORD, 0, outdata,
>+			   mem_ctx, &timeout, NULL);
>+	talloc_free(m);
>+	return handle;
> }
> 
>-static bool g_lock_unlock(TALLOC_CTX *mem_ctx,
>-			  struct ctdb_db_context *ctdb_db,
>-			  const char *keyname, uint32_t reqid)
>+int ctdb_ctrl_updaterecord_recv(struct ctdb_context *ctdb, struct ctdb_client_control_state *state)
> {
>-	TDB_DATA key, data;
>-	struct ctdb_record_handle *h;
>-	struct ctdb_g_lock_list *locks;
>-	struct ctdb_server_id id;
>-	int i;
>-	bool found = false;
>-
>-	key.dptr = (uint8_t *)discard_const(keyname);
>-	key.dsize = strlen(keyname) + 1;
>-	h = ctdb_fetch_lock(ctdb_db, mem_ctx, key, &data);
>-	if (h == NULL) {
>-		return false;
>-	}
>-
>-	if (!g_lock_parse(h, data, &locks)) {
>-		DEBUG(DEBUG_ERR, ("g_lock: error parsing locks\n"));
>-		talloc_free(data.dptr);
>-		talloc_free(h);
>-		return false;
>-	}
>-
>-	talloc_free(data.dptr);
>-
>-	id = server_id_fetch(ctdb_db->ctdb, reqid);
>-
>-	for (i=0; i<locks->num; i++) {
>-		if (ctdb_server_id_equal(&locks->lock[i].sid, &id)) {
>-			if (i < locks->num-1) {
>-				locks->lock[i] = locks->lock[locks->num-1];
>-			}
>-			locks->num--;
>-			found = true;
>-			break;
>-		}
>-	}
>-
>-	if (!found) {
>-		DEBUG(DEBUG_ERR, ("g_lock: lock not found\n"));
>-		talloc_free(h);
>-		return false;
>-	}
>-
>-	data.dptr = (uint8_t *)locks->lock;
>-	data.dsize = locks->num * sizeof(struct ctdb_g_lock);
>+	int ret;
>+	int32_t res;
> 
>-	if (ctdb_record_store(h, data) != 0) {
>-		talloc_free(h);
>-		return false;
>+	ret = ctdb_control_recv(ctdb, state, state, NULL, &res, NULL);
>+	if ( (ret != 0) || (res != 0) ){
>+		DEBUG(DEBUG_ERR,(__location__ " ctdb_ctrl_update_record_recv failed\n"));
>+		return -1;
> 	}
> 
>-	talloc_free(h);
>-	return true;
>-}
>-
>-
>-struct ctdb_transaction_handle {
>-	struct ctdb_db_context *ctdb_db;
>-	struct ctdb_db_context *g_lock_db;
>-	char *lock_name;
>-	uint32_t reqid;
>-	/*
>-	 * we store reads and writes done under a transaction:
>-	 * - one list stores both reads and writes (m_all)
>-	 * - the other just writes (m_write)
>-	 */
>-	struct ctdb_marshall_buffer *m_all;
>-	struct ctdb_marshall_buffer *m_write;
>-};
>-
>-static int ctdb_transaction_destructor(struct ctdb_transaction_handle *h)
>-{
>-	g_lock_unlock(h, h->g_lock_db, h->lock_name, h->reqid);
>-	reqid_remove(h->ctdb_db->ctdb->idr, h->reqid);
> 	return 0;
> }
> 
>-
>-/**
>- * start a transaction on a database
>- */
>-struct ctdb_transaction_handle *ctdb_transaction_start(struct ctdb_db_context *ctdb_db,
>-						       TALLOC_CTX *mem_ctx)
>+int
>+ctdb_ctrl_updaterecord(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx, struct timeval timeout, uint32_t destnode, struct ctdb_db_context *ctdb_db, TDB_DATA key, struct ctdb_ltdb_header *header, TDB_DATA data)
> {
>-	struct ctdb_transaction_handle *h;
>-
>-	h = talloc_zero(mem_ctx, struct ctdb_transaction_handle);
>-	if (h == NULL) {
>-		DEBUG(DEBUG_ERR, (__location__ " memory allocation error\n"));
>-		return NULL;
>-	}
>-
>-	h->ctdb_db = ctdb_db;
>-	h->lock_name = talloc_asprintf(h, "transaction_db_0x%08x",
>-				       (unsigned int)ctdb_db->db_id);
>-	if (h->lock_name == NULL) {
>-		DEBUG(DEBUG_ERR, (__location__ " talloc asprintf failed\n"));
>-		talloc_free(h);
>-		return NULL;
>-	}
>-
>-	h->g_lock_db = ctdb_attach(h->ctdb_db->ctdb, timeval_current_ofs(3,0),
>-				   "g_lock.tdb", 0);
>-	if (!h->g_lock_db) {
>-		DEBUG(DEBUG_ERR, (__location__ " unable to attach to g_lock.tdb\n"));
>-		talloc_free(h);
>-		return NULL;
>-	}
>-
>-	h->reqid = reqid_new(h->ctdb_db->ctdb->idr, h);
>-
>-	if (!g_lock_lock(h, h->g_lock_db, h->lock_name, h->reqid)) {
>-		DEBUG(DEBUG_ERR, (__location__ " Error locking g_lock.tdb\n"));
>-		talloc_free(h);
>-		return NULL;
>-	}
>-
>-	talloc_set_destructor(h, ctdb_transaction_destructor);
>-	return h;
>-}
>-
>-/**
>- * fetch a record inside a transaction
>- */
>-int ctdb_transaction_fetch(struct ctdb_transaction_handle *h,
>-			   TALLOC_CTX *mem_ctx,
>-			   TDB_DATA key, TDB_DATA *data)
>-{
>-	struct ctdb_ltdb_header header;
>-	int ret;
>-
>-	ZERO_STRUCT(header);
>-
>-	ret = ctdb_ltdb_fetch(h->ctdb_db, key, &header, mem_ctx, data);
>-	if (ret == -1 && header.dmaster == (uint32_t)-1) {
>-		/* record doesn't exist yet */
>-		*data = tdb_null;
>-		ret = 0;
>-	}
>-
>-	if (ret != 0) {
>-		return ret;
>-	}
>-
>-	h->m_all = ctdb_marshall_add(h, h->m_all, h->ctdb_db->db_id, 1, key, NULL, *data);
>-	if (h->m_all == NULL) {
>-		DEBUG(DEBUG_ERR,(__location__ " Failed to add to marshalling record\n"));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
>-/**
>- * stores a record inside a transaction
>- */
>-int ctdb_transaction_store(struct ctdb_transaction_handle *h,
>-			   TDB_DATA key, TDB_DATA data)
>-{
>-	TALLOC_CTX *tmp_ctx = talloc_new(h);
>-	struct ctdb_ltdb_header header;
>-	TDB_DATA olddata;
>-	int ret;
>-
>-	/* we need the header so we can update the RSN */
>-	ret = ctdb_ltdb_fetch(h->ctdb_db, key, &header, tmp_ctx, &olddata);
>-	if (ret == -1 && header.dmaster == (uint32_t)-1) {
>-		/* the record doesn't exist - create one with us as dmaster.
>-		   This is only safe because we are in a transaction and this
>-		   is a persistent database */
>-		ZERO_STRUCT(header);
>-	} else if (ret != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " Failed to fetch record\n"));
>-		talloc_free(tmp_ctx);
>-		return ret;
>-	}
>-
>-	if (data.dsize == olddata.dsize &&
>-	    memcmp(data.dptr, olddata.dptr, data.dsize) == 0 &&
>-	    header.rsn != 0) {
>-		/* save writing the same data */
>-		talloc_free(tmp_ctx);
>-		return 0;
>-	}
>-
>-	header.dmaster = h->ctdb_db->ctdb->pnn;
>-	header.rsn++;
>-
>-	h->m_all = ctdb_marshall_add(h, h->m_all, h->ctdb_db->db_id, 0, key, NULL, data);
>-	if (h->m_all == NULL) {
>-		DEBUG(DEBUG_ERR,(__location__ " Failed to add to marshalling record\n"));
>-		talloc_free(tmp_ctx);
>-		return -1;
>-	}
>-
>-	h->m_write = ctdb_marshall_add(h, h->m_write, h->ctdb_db->db_id, 0, key, &header, data);
>-	if (h->m_write == NULL) {
>-		DEBUG(DEBUG_ERR,(__location__ " Failed to add to marshalling record\n"));
>-		talloc_free(tmp_ctx);
>-		return -1;
>-	}
>-
>-	talloc_free(tmp_ctx);
>-	return 0;
>-}
>-
>-static int ctdb_fetch_db_seqnum(struct ctdb_db_context *ctdb_db, uint64_t *seqnum)
>-{
>-	const char *keyname = CTDB_DB_SEQNUM_KEY;
>-	TDB_DATA key, data;
>-	struct ctdb_ltdb_header header;
>-	int ret;
>-
>-	key.dptr = (uint8_t *)discard_const(keyname);
>-	key.dsize = strlen(keyname) + 1;
>-
>-	ret = ctdb_ltdb_fetch(ctdb_db, key, &header, ctdb_db, &data);
>-	if (ret != 0) {
>-		*seqnum = 0;
>-		return 0;
>-	}
>-
>-	if (data.dsize == 0) {
>-		*seqnum = 0;
>-		return 0;
>-	}
>-
>-	if (data.dsize != sizeof(*seqnum)) {
>-		DEBUG(DEBUG_ERR, (__location__ " Invalid data received len=%zi\n",
>-				  data.dsize));
>-		talloc_free(data.dptr);
>-		return -1;
>-	}
>-
>-	*seqnum = *(uint64_t *)data.dptr;
>-	talloc_free(data.dptr);
>-
>-	return 0;
>-}
>-
>-
>-static int ctdb_store_db_seqnum(struct ctdb_transaction_handle *h,
>-				uint64_t seqnum)
>-{
>-	const char *keyname = CTDB_DB_SEQNUM_KEY;
>-	TDB_DATA key, data;
>-
>-	key.dptr = (uint8_t *)discard_const(keyname);
>-	key.dsize = strlen(keyname) + 1;
>-
>-	data.dptr = (uint8_t *)&seqnum;
>-	data.dsize = sizeof(seqnum);
>-
>-	return ctdb_transaction_store(h, key, data);
>-}
>-
>-
>-/**
>- * commit a transaction
>- */
>-int ctdb_transaction_commit(struct ctdb_transaction_handle *h)
>-{
>-	int ret;
>-	uint64_t old_seqnum, new_seqnum;
>-	int32_t status;
>-	struct timeval timeout;
>-
>-	if (h->m_write == NULL) {
>-		/* no changes were made */
>-		talloc_free(h);
>-		return 0;
>-	}
>-
>-	ret = ctdb_fetch_db_seqnum(h->ctdb_db, &old_seqnum);
>-	if (ret != 0) {
>-		DEBUG(DEBUG_ERR, (__location__ " failed to fetch db sequence number\n"));
>-		ret = -1;
>-		goto done;
>-	}
>-
>-	new_seqnum = old_seqnum + 1;
>-	ret = ctdb_store_db_seqnum(h, new_seqnum);
>-	if (ret != 0) {
>-		DEBUG(DEBUG_ERR, (__location__ " failed to store db sequence number\n"));
>-		ret = -1;
>-		goto done;
>-	}
>-
>-again:
>-	timeout = timeval_current_ofs(30,0);
>-	ret = ctdb_control(h->ctdb_db->ctdb, CTDB_CURRENT_NODE,
>-			   h->ctdb_db->db_id,
>-			   CTDB_CONTROL_TRANS3_COMMIT, 0,
>-			   ctdb_marshall_finish(h->m_write), NULL, NULL,
>-			   &status, &timeout, NULL);
>-	if (ret != 0 || status != 0) {
>-		/*
>-		 * TRANS3_COMMIT control will only fail if recovery has been
>-		 * triggered.  Check if the database has been updated or not.
>-		 */
>-		ret = ctdb_fetch_db_seqnum(h->ctdb_db, &new_seqnum);
>-		if (ret != 0) {
>-			DEBUG(DEBUG_ERR, (__location__ " failed to fetch db sequence number\n"));
>-			goto done;
>-		}
>-
>-		if (new_seqnum == old_seqnum) {
>-			/* Database not yet updated, try again */
>-			goto again;
>-		}
>-
>-		if (new_seqnum != (old_seqnum + 1)) {
>-			DEBUG(DEBUG_ERR, (__location__ " new seqnum [%llu] != old seqnum [%llu] + 1\n",
>-					  (long long unsigned)new_seqnum,
>-					  (long long unsigned)old_seqnum));
>-			ret = -1;
>-			goto done;
>-		}
>-	}
>-
>-	ret = 0;
>-
>-done:
>-	talloc_free(h);
>-	return ret;
>-}
>-
>-/**
>- * cancel a transaction
>- */
>-int ctdb_transaction_cancel(struct ctdb_transaction_handle *h)
>-{
>-	talloc_free(h);
>-	return 0;
>-}
>-
>-
>-/*
>-  recovery daemon ping to main daemon
>- */
>-int ctdb_ctrl_recd_ping(struct ctdb_context *ctdb)
>-{
>-	int ret;
>-	int32_t res;
>-
>-	ret = ctdb_control(ctdb, CTDB_CURRENT_NODE, 0, CTDB_CONTROL_RECD_PING, 0, tdb_null, 
>-			   ctdb, NULL, &res, NULL, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,("Failed to send recd ping\n"));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
>-/*
>-  tell the main daemon how long it took to lock the reclock file
>- */
>-int ctdb_ctrl_report_recd_lock_latency(struct ctdb_context *ctdb, struct timeval timeout, double latency)
>-{
>-	int ret;
>-	int32_t res;
>-	TDB_DATA data;
>-
>-	data.dptr = (uint8_t *)&latency;
>-	data.dsize = sizeof(latency);
>-
>-	ret = ctdb_control(ctdb, CTDB_CURRENT_NODE, 0, CTDB_CONTROL_RECD_RECLOCK_LATENCY, 0, data, 
>-			   ctdb, NULL, &res, NULL, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,("Failed to send recd reclock latency\n"));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
>-/*
>-  get the name of the reclock file
>- */
>-int ctdb_ctrl_getreclock(struct ctdb_context *ctdb, struct timeval timeout,
>-			 uint32_t destnode, TALLOC_CTX *mem_ctx,
>-			 const char **name)
>-{
>-	int ret;
>-	int32_t res;
>-	TDB_DATA data;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_GET_RECLOCK_FILE, 0, tdb_null, 
>-			   mem_ctx, &data, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0) {
>-		return -1;
>-	}
>-
>-	if (data.dsize == 0) {
>-		*name = NULL;
>-	} else {
>-		*name = talloc_strdup(mem_ctx, discard_const(data.dptr));
>-	}
>-	talloc_free(data.dptr);
>-
>-	return 0;
>-}
>-
>-/*
>-  stop a node
>- */
>-int ctdb_ctrl_stop_node(struct ctdb_context *ctdb, struct timeval timeout, uint32_t destnode)
>-{
>-	int ret;
>-	int32_t res;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_STOP_NODE, 0, tdb_null, 
>-			   ctdb, NULL, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,("Failed to stop node\n"));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
>-/*
>-  continue a node
>- */
>-int ctdb_ctrl_continue_node(struct ctdb_context *ctdb, struct timeval timeout, uint32_t destnode)
>-{
>-	int ret;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_CONTINUE_NODE, 0, tdb_null, 
>-			   ctdb, NULL, NULL, &timeout, NULL);
>-	if (ret != 0) {
>-		DEBUG(DEBUG_ERR,("Failed to continue node\n"));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
>-/*
>-  set the lmaster role for a node
>- */
>-int ctdb_ctrl_setlmasterrole(struct ctdb_context *ctdb, struct timeval timeout, uint32_t destnode, uint32_t lmasterrole)
>-{
>-	int ret;
>-	TDB_DATA data;
>-	int32_t res;
>-
>-	data.dsize = sizeof(lmasterrole);
>-	data.dptr  = (uint8_t *)&lmasterrole;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_SET_LMASTERROLE, 0, data, 
>-			   NULL, NULL, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for setlmasterrole failed\n"));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
>-/*
>-  set the recmaster role for a node
>- */
>-int ctdb_ctrl_setrecmasterrole(struct ctdb_context *ctdb, struct timeval timeout, uint32_t destnode, uint32_t recmasterrole)
>-{
>-	int ret;
>-	TDB_DATA data;
>-	int32_t res;
>-
>-	data.dsize = sizeof(recmasterrole);
>-	data.dptr  = (uint8_t *)&recmasterrole;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_SET_RECMASTERROLE, 0, data, 
>-			   NULL, NULL, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for setrecmasterrole failed\n"));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
>-int ctdb_ctrl_set_ban(struct ctdb_context *ctdb, struct timeval timeout,
>-		      uint32_t destnode, struct ctdb_ban_state *bantime)
>-{
>-	int ret;
>-	TDB_DATA data;
>-	int32_t res;
>-
>-	data.dsize = sizeof(*bantime);
>-	data.dptr  = (uint8_t *)bantime;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_SET_BAN_STATE, 0, data, 
>-			   NULL, NULL, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for set ban state failed\n"));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
>-
>-int ctdb_ctrl_get_ban(struct ctdb_context *ctdb, struct timeval timeout,
>-		      uint32_t destnode, TALLOC_CTX *mem_ctx,
>-		      struct ctdb_ban_state **bantime)
>-{
>-	int ret;
>-	TDB_DATA outdata;
>-	int32_t res;
>-	TALLOC_CTX *tmp_ctx = talloc_new(NULL);
>-
>-	ret = ctdb_control(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_GET_BAN_STATE, 0, tdb_null,
>-			   tmp_ctx, &outdata, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for set ban state failed\n"));
>-		talloc_free(tmp_ctx);
>-		return -1;
>-	}
>-
>-	*bantime = (struct ctdb_ban_state *)talloc_steal(mem_ctx, outdata.dptr);
>-	talloc_free(tmp_ctx);
>-
>-	return 0;
>-}
>-
>-int ctdb_ctrl_getstathistory(struct ctdb_context *ctdb,
>-			     struct timeval timeout, uint32_t destnode,
>-			     TALLOC_CTX *mem_ctx,
>-			     struct ctdb_statistics_list_old **stats)
>-{
>-	int ret;
>-	TDB_DATA outdata;
>-	int32_t res;
>-
>-	ret = ctdb_control(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_GET_STAT_HISTORY, 0, tdb_null, 
>-			   mem_ctx, &outdata, &res, &timeout, NULL);
>-	if (ret != 0 || res != 0 || outdata.dsize == 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for getstathistory failed ret:%d res:%d\n", ret, res));
>-		return -1;
>-	}
>-
>-	*stats = (struct ctdb_statistics_list_old *)talloc_memdup(mem_ctx,
>-								  outdata.dptr,
>-								  outdata.dsize);
>-	talloc_free(outdata.dptr);
>-
>-	return 0;
>-}
>-
>-struct ctdb_ltdb_header *ctdb_header_from_record_handle(struct ctdb_record_handle *h)
>-{
>-	if (h == NULL) {
>-		return NULL;
>-	}
>-
>-	return &h->header;
>-}
>-
>-
>-struct ctdb_client_control_state *
>-ctdb_ctrl_updaterecord_send(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx, struct timeval timeout, uint32_t destnode, struct ctdb_db_context *ctdb_db, TDB_DATA key, struct ctdb_ltdb_header *header, TDB_DATA data)
>-{
>-	struct ctdb_client_control_state *handle;
>-	struct ctdb_marshall_buffer *m;
>-	struct ctdb_rec_data_old *rec;
>-	TDB_DATA outdata;
>-
>-	m = talloc_zero(mem_ctx, struct ctdb_marshall_buffer);
>-	if (m == NULL) {
>-		DEBUG(DEBUG_ERR, ("Failed to allocate marshall buffer for update record\n"));
>-		return NULL;
>-	}
>-
>-	m->db_id = ctdb_db->db_id;
>-
>-	rec = ctdb_marshall_record(m, 0, key, header, data);
>-	if (rec == NULL) {
>-		DEBUG(DEBUG_ERR,("Failed to marshall record for update record\n"));
>-		talloc_free(m);
>-		return NULL;
>-	}
>-	m = talloc_realloc_size(mem_ctx, m, rec->length + offsetof(struct ctdb_marshall_buffer, data));
>-	if (m == NULL) {
>-		DEBUG(DEBUG_CRIT,(__location__ " Failed to expand recdata\n"));
>-		talloc_free(m);
>-		return NULL;
>-	}
>-	m->count++;
>-	memcpy((uint8_t *)m + offsetof(struct ctdb_marshall_buffer, data), rec, rec->length);
>-
>-
>-	outdata.dptr = (uint8_t *)m;
>-	outdata.dsize = talloc_get_size(m);
>-
>-	handle = ctdb_control_send(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_UPDATE_RECORD, 0, outdata,
>-			   mem_ctx, &timeout, NULL);
>-	talloc_free(m);
>-	return handle;
>-}
>-
>-int ctdb_ctrl_updaterecord_recv(struct ctdb_context *ctdb, struct ctdb_client_control_state *state)
>-{
>-	int ret;
>-	int32_t res;
>-
>-	ret = ctdb_control_recv(ctdb, state, state, NULL, &res, NULL);
>-	if ( (ret != 0) || (res != 0) ){
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_ctrl_update_record_recv failed\n"));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
>-int
>-ctdb_ctrl_updaterecord(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx, struct timeval timeout, uint32_t destnode, struct ctdb_db_context *ctdb_db, TDB_DATA key, struct ctdb_ltdb_header *header, TDB_DATA data)
>-{
>-	struct ctdb_client_control_state *state;
>+	struct ctdb_client_control_state *state;
> 
> 	state = ctdb_ctrl_updaterecord_send(ctdb, mem_ctx, timeout, destnode, ctdb_db, key, header, data);
> 	return ctdb_ctrl_updaterecord_recv(ctdb, state);
> }
>-
>-
>-
>-
>-
>-
>-/*
>-  set a database to be readonly
>- */
>-struct ctdb_client_control_state *
>-ctdb_ctrl_set_db_readonly_send(struct ctdb_context *ctdb, uint32_t destnode, uint32_t dbid)
>-{
>-	TDB_DATA data;
>-
>-	data.dptr = (uint8_t *)&dbid;
>-	data.dsize = sizeof(dbid);
>-
>-	return ctdb_control_send(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_SET_DB_READONLY, 0, data, 
>-			   ctdb, NULL, NULL);
>-}
>-
>-int ctdb_ctrl_set_db_readonly_recv(struct ctdb_context *ctdb, struct ctdb_client_control_state *state)
>-{
>-	int ret;
>-	int32_t res;
>-
>-	ret = ctdb_control_recv(ctdb, state, ctdb, NULL, &res, NULL);
>-	if (ret != 0 || res != 0) {
>-	  DEBUG(DEBUG_ERR,(__location__ " ctdb_ctrl_set_db_readonly_recv failed  ret:%d res:%d\n", ret, res));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
>-int ctdb_ctrl_set_db_readonly(struct ctdb_context *ctdb, uint32_t destnode, uint32_t dbid)
>-{
>-	struct ctdb_client_control_state *state;
>-
>-	state = ctdb_ctrl_set_db_readonly_send(ctdb, destnode, dbid);
>-	return ctdb_ctrl_set_db_readonly_recv(ctdb, state);
>-}
>-
>-/*
>-  set a database to be sticky
>- */
>-struct ctdb_client_control_state *
>-ctdb_ctrl_set_db_sticky_send(struct ctdb_context *ctdb, uint32_t destnode, uint32_t dbid)
>-{
>-	TDB_DATA data;
>-
>-	data.dptr = (uint8_t *)&dbid;
>-	data.dsize = sizeof(dbid);
>-
>-	return ctdb_control_send(ctdb, destnode, 0, 
>-			   CTDB_CONTROL_SET_DB_STICKY, 0, data, 
>-			   ctdb, NULL, NULL);
>-}
>-
>-int ctdb_ctrl_set_db_sticky_recv(struct ctdb_context *ctdb, struct ctdb_client_control_state *state)
>-{
>-	int ret;
>-	int32_t res;
>-
>-	ret = ctdb_control_recv(ctdb, state, ctdb, NULL, &res, NULL);
>-	if (ret != 0 || res != 0) {
>-		DEBUG(DEBUG_ERR,(__location__ " ctdb_ctrl_set_db_sticky_recv failed  ret:%d res:%d\n", ret, res));
>-		return -1;
>-	}
>-
>-	return 0;
>-}
>-
>-int ctdb_ctrl_set_db_sticky(struct ctdb_context *ctdb, uint32_t destnode, uint32_t dbid)
>-{
>-	struct ctdb_client_control_state *state;
>-
>-	state = ctdb_ctrl_set_db_sticky_send(ctdb, destnode, dbid);
>-	return ctdb_ctrl_set_db_sticky_recv(ctdb, state);
>-}
>diff --git a/ctdb/include/ctdb_client.h b/ctdb/include/ctdb_client.h
>index d9040fb3550..e7f74192745 100644
>--- a/ctdb/include/ctdb_client.h
>+++ b/ctdb/include/ctdb_client.h
>@@ -100,24 +100,6 @@ int ctdb_client_remove_message_handler(struct ctdb_context *ctdb,
> int ctdb_client_send_message(struct ctdb_context *ctdb, uint32_t pnn,
> 			     uint64_t srvid, TDB_DATA data);
> 
>-/*
>-   Fetch a ctdb record from a remote node. Underneath this will force the
>-   dmaster for the record to be moved to the local node.
>-*/
>-struct ctdb_record_handle *ctdb_fetch_lock(struct ctdb_db_context *ctdb_db,
>-					   TALLOC_CTX *mem_ctx,
>-					   TDB_DATA key, TDB_DATA *data);
>-
>-struct ctdb_record_handle *ctdb_fetch_readonly_lock(
>-					struct ctdb_db_context *ctdb_db,
>-					TALLOC_CTX *mem_ctx, TDB_DATA key,
>-					TDB_DATA *data, int read_only);
>-
>-int ctdb_record_store(struct ctdb_record_handle *h, TDB_DATA data);
>-
>-int ctdb_fetch(struct ctdb_db_context *ctdb_db, TALLOC_CTX *mem_ctx,
>-	       TDB_DATA key, TDB_DATA *data);
>-
> struct ctdb_client_control_state *ctdb_control_send(struct ctdb_context *ctdb,
> 						    uint32_t destnode,
> 						    uint64_t srvid,
>@@ -136,24 +118,9 @@ int ctdb_control(struct ctdb_context *ctdb, uint32_t destnode, uint64_t srvid,
> 		 TALLOC_CTX *mem_ctx, TDB_DATA *outdata, int32_t *status,
> 		 struct timeval *timeout, char **errormsg);
> 
>-int ctdb_ctrl_process_exists(struct ctdb_context *ctdb, uint32_t destnode,
>-			     pid_t pid);
>-
>-int ctdb_ctrl_statistics(struct ctdb_context *ctdb, uint32_t destnode,
>-			 struct ctdb_statistics *status);
>-int ctdb_ctrl_dbstatistics(struct ctdb_context *ctdb, uint32_t destnode,
>-			   uint32_t dbid, TALLOC_CTX *mem_ctx,
>-			   struct ctdb_db_statistics_old **dbstat);
>-
>-int ctdb_ctrl_shutdown(struct ctdb_context *ctdb, struct timeval timeout,
>-		       uint32_t destnode);
>-
> int ctdb_ctrl_getvnnmap(struct ctdb_context *ctdb, struct timeval timeout,
> 			uint32_t destnode, TALLOC_CTX *mem_ctx,
> 			struct ctdb_vnn_map **vnnmap);
>-int ctdb_ctrl_setvnnmap(struct ctdb_context *ctdb, struct timeval timeout,
>-			uint32_t destnode, TALLOC_CTX *mem_ctx,
>-			struct ctdb_vnn_map *vnnmap);
> 
> /*
>   get the recovery mode of a remote node
>@@ -206,34 +173,6 @@ int ctdb_ctrl_getnodemap(struct ctdb_context *ctdb, struct timeval timeout,
> 			 uint32_t destnode, TALLOC_CTX *mem_ctx,
> 			 struct ctdb_node_map_old **nodemap);
> 
>-int ctdb_ctrl_getnodesfile(struct ctdb_context *ctdb, struct timeval timeout,
>-			   uint32_t destnode, TALLOC_CTX *mem_ctx,
>-			   struct ctdb_node_map_old **nodemap);
>-
>-int ctdb_ctrl_reload_nodes_file(struct ctdb_context *ctdb,
>-				struct timeval timeout, uint32_t destnode);
>-
>-struct ctdb_client_control_state *ctdb_ctrl_pulldb_send(
>-					struct ctdb_context *ctdb,
>-					uint32_t destnode, uint32_t dbid,
>-					uint32_t lmaster, TALLOC_CTX *mem_ctx,
>-					struct timeval timeout);
>-int ctdb_ctrl_pulldb_recv(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx,
>-			  struct ctdb_client_control_state *state,
>-			  TDB_DATA *outdata);
>-int ctdb_ctrl_pulldb(struct ctdb_context *ctdb, uint32_t destnode,
>-		     uint32_t dbid, uint32_t lmaster, TALLOC_CTX *mem_ctx,
>-		     struct timeval timeout, TDB_DATA *outdata);
>-
>-/*
>-  change dmaster for all keys in the database to the new value
>- */
>-int ctdb_ctrl_setdmaster(struct ctdb_context *ctdb, struct timeval timeout,
>-			 uint32_t destnode, TALLOC_CTX *mem_ctx,
>-			 uint32_t dbid, uint32_t dmaster);
>-
>-int ctdb_ctrl_ping(struct ctdb_context *ctdb, uint32_t destnode);
>-
> int ctdb_ctrl_get_runstate(struct ctdb_context *ctdb, struct timeval timeout,
> 			   uint32_t destnode, uint32_t *runstate);
> 
>@@ -243,11 +182,6 @@ int ctdb_ctrl_getdbpath(struct ctdb_context *ctdb, struct timeval timeout,
> int ctdb_ctrl_getdbname(struct ctdb_context *ctdb, struct timeval timeout,
> 			uint32_t destnode, uint32_t dbid,
> 			TALLOC_CTX *mem_ctx, const char **name);
>-int ctdb_ctrl_getdbhealth(struct ctdb_context *ctdb, struct timeval timeout,
>-			  uint32_t destnode, uint32_t dbid,
>-			  TALLOC_CTX *mem_ctx, const char **reason);
>-int ctdb_ctrl_getdbseqnum(struct ctdb_context *ctdb, struct timeval timeout,
>-			  uint32_t destnode, uint32_t dbid, uint64_t *seqnum);
> 
> int ctdb_ctrl_createdb(struct ctdb_context *ctdb, struct timeval timeout,
> 		       uint32_t destnode, TALLOC_CTX *mem_ctx,
>@@ -255,14 +189,6 @@ int ctdb_ctrl_createdb(struct ctdb_context *ctdb, struct timeval timeout,
> 
> int ctdb_ctrl_get_debuglevel(struct ctdb_context *ctdb, uint32_t destnode,
> 			     int32_t *level);
>-int ctdb_ctrl_set_debuglevel(struct ctdb_context *ctdb, uint32_t destnode,
>-			     int32_t level);
>-
>-uint32_t *ctdb_get_connected_nodes(struct ctdb_context *ctdb,
>-				   struct timeval timeout,
>-				   TALLOC_CTX *mem_ctx, uint32_t *num_nodes);
>-
>-int ctdb_statistics_reset(struct ctdb_context *ctdb, uint32_t destnode);
> 
> /*
>   attach to a ctdb database
>@@ -275,8 +201,6 @@ struct ctdb_db_context *ctdb_attach(struct ctdb_context *ctdb,
> 				    const char *name,
> 				    uint8_t db_flags);
> 
>-int ctdb_detach(struct ctdb_context *ctdb, uint32_t db_id);
>-
> /* a ctdb call function */
> typedef int (*ctdb_fn_t)(struct ctdb_call_info *);
> 
>@@ -285,54 +209,12 @@ typedef int (*ctdb_fn_t)(struct ctdb_call_info *);
> */
> int ctdb_set_call(struct ctdb_db_context *ctdb_db, ctdb_fn_t fn, uint32_t id);
> 
>-
>-typedef int (*ctdb_traverse_func)(TDB_DATA, TDB_DATA, void *);
>-
>-int ctdb_traverse(struct ctdb_db_context *ctdb_db, ctdb_traverse_func fn,
>-		  void *private_data);
>-
>-struct ctdb_dump_db_context {
>-	struct ctdb_context *ctdb;
>-	FILE *f;
>-	bool printemptyrecords;
>-	bool printdatasize;
>-	bool printlmaster;
>-	bool printhash;
>-	bool printrecordflags;
>-};
>-
>-int ctdb_dumpdb_record(TDB_DATA key, TDB_DATA data, void *p);
>-int ctdb_dump_db(struct ctdb_db_context *ctdb_db,
>-		 struct ctdb_dump_db_context *ctx);
>-
>-/*
>-  get the pid of a ctdb daemon
>- */
>-int ctdb_ctrl_getpid(struct ctdb_context *ctdb, struct timeval timeout,
>-		     uint32_t destnode, uint32_t *pid);
>-
> int ctdb_ctrl_freeze(struct ctdb_context *ctdb, struct timeval timeout,
> 		     uint32_t destnode);
> 
> int ctdb_ctrl_getpnn(struct ctdb_context *ctdb, struct timeval timeout,
> 		     uint32_t destnode);
> 
>-int ctdb_ctrl_takeover_ip(struct ctdb_context *ctdb, struct timeval timeout,
>-			  uint32_t destnode, struct ctdb_public_ip *ip);
>-int ctdb_ctrl_release_ip(struct ctdb_context *ctdb, struct timeval timeout,
>-			 uint32_t destnode, struct ctdb_public_ip *ip);
>-
>-int ctdb_ctrl_get_tunable(struct ctdb_context *ctdb,
>-			  struct timeval timeout, uint32_t destnode,
>-			  const char *name, uint32_t *value);
>-int ctdb_ctrl_set_tunable(struct ctdb_context *ctdb,
>-			  struct timeval timeout, uint32_t destnode,
>-			  const char *name, uint32_t value);
>-int ctdb_ctrl_list_tunables(struct ctdb_context *ctdb,
>-			    struct timeval timeout, uint32_t destnode,
>-			    TALLOC_CTX *mem_ctx,
>-			    const char ***list, uint32_t *count);
>-
> int ctdb_ctrl_get_public_ips_flags(struct ctdb_context *ctdb,
> 				   struct timeval timeout, uint32_t destnode,
> 				   TALLOC_CTX *mem_ctx, uint32_t flags,
>@@ -341,20 +223,11 @@ int ctdb_ctrl_get_public_ips(struct ctdb_context *ctdb,
> 			     struct timeval timeout, uint32_t destnode,
> 			     TALLOC_CTX *mem_ctx,
> 			     struct ctdb_public_ip_list_old **ips);
>-int ctdb_ctrl_get_public_ip_info(struct ctdb_context *ctdb,
>-				 struct timeval timeout, uint32_t destnode,
>-				 TALLOC_CTX *mem_ctx,
>-				 const ctdb_sock_addr *addr,
>-				 struct ctdb_public_ip_info_old **info);
> 
> int ctdb_ctrl_get_ifaces(struct ctdb_context *ctdb,
> 			 struct timeval timeout, uint32_t destnode,
> 			 TALLOC_CTX *mem_ctx,
> 			 struct ctdb_iface_list_old **ifaces);
>-int ctdb_ctrl_set_iface_link(struct ctdb_context *ctdb,
>-			     struct timeval timeout, uint32_t destnode,
>-			     TALLOC_CTX *mem_ctx,
>-			     const struct ctdb_iface *info);
> 
> int ctdb_ctrl_modflags(struct ctdb_context *ctdb,
> 		       struct timeval timeout,
>@@ -365,22 +238,6 @@ int ctdb_ctrl_get_all_tunables(struct ctdb_context *ctdb,
> 			       struct timeval timeout, uint32_t destnode,
> 			       struct ctdb_tunable_list *tunables);
> 
>-int ctdb_ctrl_add_public_ip(struct ctdb_context *ctdb,
>-			    struct timeval timeout, uint32_t destnode,
>-			    struct ctdb_addr_info_old *pub);
>-int ctdb_ctrl_del_public_ip(struct ctdb_context *ctdb,
>-			    struct timeval timeout, uint32_t destnode,
>-			    struct ctdb_addr_info_old *pub);
>-
>-int ctdb_ctrl_gratious_arp(struct ctdb_context *ctdb,
>-			   struct timeval timeout, uint32_t destnode,
>-			   ctdb_sock_addr *addr, const char *ifname);
>-
>-int ctdb_ctrl_get_tcp_tickles(struct ctdb_context *ctdb,
>-			      struct timeval timeout, uint32_t destnode,
>-			      TALLOC_CTX *mem_ctx, ctdb_sock_addr *addr,
>-			      struct ctdb_tickle_list_old **list);
>-
> /*
>   initialise ctdb subsystem
> */
>@@ -397,24 +254,6 @@ const char *ctdb_get_socketname(struct ctdb_context *ctdb);
> /* return pnn of this node */
> uint32_t ctdb_get_pnn(struct ctdb_context *ctdb);
> 
>-/*
>-  get the uptime of a remote node
>- */
>-struct ctdb_client_control_state *ctdb_ctrl_uptime_send(
>-					struct ctdb_context *ctdb,
>-					TALLOC_CTX *mem_ctx,
>-					struct timeval timeout,
>-					uint32_t destnode);
>-int ctdb_ctrl_uptime_recv(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx,
>-			  struct ctdb_client_control_state *state,
>-			  struct ctdb_uptime **uptime);
>-int ctdb_ctrl_uptime(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx,
>-		     struct timeval timeout, uint32_t destnode,
>-		     struct ctdb_uptime **uptime);
>-
>-int ctdb_ctrl_end_recovery(struct ctdb_context *ctdb, struct timeval timeout,
>-			   uint32_t destnode);
>-
> typedef void (*client_async_callback)(struct ctdb_context *ctdb,
> 				      uint32_t node_pnn, int32_t res,
> 				      TDB_DATA outdata, void *callback_data);
>@@ -455,8 +294,6 @@ uint32_t *list_of_connected_nodes(struct ctdb_context *ctdb,
> 				  struct ctdb_node_map_old *node_map,
> 				  TALLOC_CTX *mem_ctx, bool include_self);
> 
>-int ctdb_read_pnn_lock(int fd, int32_t pnn);
>-
> /*
>   get capabilities of a remote node
>  */
>@@ -495,52 +332,13 @@ uint32_t *ctdb_get_node_capabilities(struct ctdb_node_capabilities *caps,
> bool ctdb_node_has_capabilities(struct ctdb_node_capabilities *caps,
> 				uint32_t pnn, uint32_t capabilities_required);
> 
>-
>-struct ctdb_transaction_handle *ctdb_transaction_start(
>-					struct ctdb_db_context *ctdb_db,
>-					TALLOC_CTX *mem_ctx);
>-int ctdb_transaction_fetch(struct ctdb_transaction_handle *h,
>-			   TALLOC_CTX *mem_ctx,
>-			   TDB_DATA key, TDB_DATA *data);
>-int ctdb_transaction_store(struct ctdb_transaction_handle *h,
>-			   TDB_DATA key, TDB_DATA data);
>-int ctdb_transaction_commit(struct ctdb_transaction_handle *h);
>-int ctdb_transaction_cancel(struct ctdb_transaction_handle *h);
>-
> int ctdb_ctrl_recd_ping(struct ctdb_context *ctdb);
> 
> int ctdb_ctrl_report_recd_lock_latency(struct ctdb_context *ctdb,
> 				       struct timeval timeout, double latency);
> 
>-int ctdb_ctrl_getreclock(struct ctdb_context *ctdb,
>-			 struct timeval timeout, uint32_t destnode,
>-			 TALLOC_CTX *mem_ctx, const char **reclock);
>-
>-int ctdb_ctrl_stop_node(struct ctdb_context *ctdb, struct timeval timeout,
>-			uint32_t destnode);
>-int ctdb_ctrl_continue_node(struct ctdb_context *ctdb, struct timeval timeout,
>-			    uint32_t destnode);
>-
>-int ctdb_ctrl_setlmasterrole(struct ctdb_context *ctdb,
>-			     struct timeval timeout, uint32_t destnode,
>-			     uint32_t lmasterrole);
>-int ctdb_ctrl_setrecmasterrole(struct ctdb_context *ctdb,
>-			       struct timeval timeout, uint32_t destnode,
>-			       uint32_t recmasterrole);
>-
> int ctdb_ctrl_set_ban(struct ctdb_context *ctdb, struct timeval timeout,
> 		      uint32_t destnode, struct ctdb_ban_state *bantime);
>-int ctdb_ctrl_get_ban(struct ctdb_context *ctdb, struct timeval timeout,
>-		      uint32_t destnode, TALLOC_CTX *mem_ctx,
>-		      struct ctdb_ban_state **bantime);
>-
>-int ctdb_ctrl_getstathistory(struct ctdb_context *ctdb,
>-			     struct timeval timeout, uint32_t destnode,
>-			     TALLOC_CTX *mem_ctx,
>-			     struct ctdb_statistics_list_old **stats);
>-
>-struct ctdb_ltdb_header *ctdb_header_from_record_handle(
>-					struct ctdb_record_handle *h);
> 
> struct ctdb_client_control_state *ctdb_ctrl_updaterecord_send(
> 					struct ctdb_context *ctdb,
>@@ -558,20 +356,4 @@ int ctdb_ctrl_updaterecord(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx,
> 			   struct ctdb_db_context *ctdb_db, TDB_DATA key,
> 			   struct ctdb_ltdb_header *header, TDB_DATA data);
> 
>-struct ctdb_client_control_state *ctdb_ctrl_set_db_readonly_send(
>-					struct ctdb_context *ctdb,
>-					uint32_t destnode, uint32_t dbid);
>-int ctdb_ctrl_set_db_readonly_recv(struct ctdb_context *ctdb,
>-				   struct ctdb_client_control_state *state);
>-int ctdb_ctrl_set_db_readonly(struct ctdb_context *ctdb, uint32_t destnode,
>-			      uint32_t dbid);
>-
>-struct ctdb_client_control_state *ctdb_ctrl_set_db_sticky_send(
>-					struct ctdb_context *ctdb,
>-					uint32_t destnode, uint32_t dbid);
>-int ctdb_ctrl_set_db_sticky_recv(struct ctdb_context *ctdb,
>-				 struct ctdb_client_control_state *state);
>-int ctdb_ctrl_set_db_sticky(struct ctdb_context *ctdb, uint32_t destnode,
>-			    uint32_t dbid);
>-
> #endif /* _CTDB_CLIENT_H */
>-- 
>2.16.3
>

Flags:

amitay: review+

Actions: View

Attachments on bug 13411: 14173