ctdb-daemon: Move ctdb_client.c to server/ subdir

It is used only by the code in the server directory.  It's mainly used
in recovery daemon and vacuuming child process.

Signed-off-by: Amitay Isaacs <amitay@gmail.com>
Reviewed-by: Martin Schwenke <martin@meltin.net>

1 files changed, 0 insertions, 2127 deletions

diff --git a/ctdb/client/ctdb_client.c b/ctdb/client/ctdb_client.c
deleted file mode 100644
index 7d5471dc4b5..00000000000
--- a/ctdb/client/ctdb_client.c
+++ /dev/null
@@ -1,2127 +0,0 @@
-/* 
-   ctdb daemon code
-
-   Copyright (C) Andrew Tridgell  2007
-   Copyright (C) Ronnie Sahlberg  2007
-
-   This program is free software; you can redistribute it and/or modify
-   it under the terms of the GNU General Public License as published by
-   the Free Software Foundation; either version 3 of the License, or
-   (at your option) any later version.
-   
-   This program is distributed in the hope that it will be useful,
-   but WITHOUT ANY WARRANTY; without even the implied warranty of
-   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-   GNU General Public License for more details.
-   
-   You should have received a copy of the GNU General Public License
-   along with this program; if not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "replace.h"
-#include "system/network.h"
-#include "system/filesys.h"
-#include "system/locale.h"
-
-#include <talloc.h>
-#include <tevent.h>
-#include <tdb.h>
-
-#include "lib/tdb_wrap/tdb_wrap.h"
-#include "lib/util/dlinklist.h"
-#include "lib/util/time.h"
-#include "lib/util/debug.h"
-#include "lib/util/samba_util.h"
-
-#include "ctdb_private.h"
-#include "ctdb_client.h"
-
-#include "common/reqid.h"
-#include "common/system.h"
-#include "common/common.h"
-#include "common/logging.h"
-
-/*
-  allocate a packet for use in client<->daemon communication
- */
-struct ctdb_req_header *_ctdbd_allocate_pkt(struct ctdb_context *ctdb,
-					    TALLOC_CTX *mem_ctx, 
-					    enum ctdb_operation operation, 
-					    size_t length, size_t slength,
-					    const char *type)
-{
-	int size;
-	struct ctdb_req_header *hdr;
-
-	length = MAX(length, slength);
-	size = (length+(CTDB_DS_ALIGNMENT-1)) & ~(CTDB_DS_ALIGNMENT-1);
-
-	hdr = (struct ctdb_req_header *)talloc_zero_size(mem_ctx, size);
-	if (hdr == NULL) {
-		DEBUG(DEBUG_ERR,("Unable to allocate packet for operation %u of length %u\n",
-			 operation, (unsigned)length));
-		return NULL;
-	}
-	talloc_set_name_const(hdr, type);
-	hdr->length       = length;
-	hdr->operation    = operation;
-	hdr->ctdb_magic   = CTDB_MAGIC;
-	hdr->ctdb_version = CTDB_PROTOCOL;
-	hdr->srcnode      = ctdb->pnn;
-	if (ctdb->vnn_map) {
-		hdr->generation = ctdb->vnn_map->generation;
-	}
-
-	return hdr;
-}
-
-/*
-  local version of ctdb_call
-*/
-int ctdb_call_local(struct ctdb_db_context *ctdb_db, struct ctdb_call *call,
-		    struct ctdb_ltdb_header *header, TALLOC_CTX *mem_ctx,
-		    TDB_DATA *data, bool updatetdb)
-{
-	struct ctdb_call_info *c;
-	struct ctdb_registered_call *fn;
-	struct ctdb_context *ctdb = ctdb_db->ctdb;
-	
-	c = talloc_zero(mem_ctx, struct ctdb_call_info);
-	CTDB_NO_MEMORY(ctdb, c);
-
-	c->key = call->key;
-	c->call_data = &call->call_data;
-	c->record_data.dptr = talloc_memdup(c, data->dptr, data->dsize);
-	c->record_data.dsize = data->dsize;
-	CTDB_NO_MEMORY(ctdb, c->record_data.dptr);
-	c->header = header;
-
-	for (fn=ctdb_db->calls;fn;fn=fn->next) {
-		if (fn->id == call->call_id) break;
-	}
-	if (fn == NULL) {
-		ctdb_set_error(ctdb, "Unknown call id %u\n", call->call_id);
-		talloc_free(c);
-		return -1;
-	}
-
-	if (fn->fn(c) != 0) {
-		ctdb_set_error(ctdb, "ctdb_call %u failed\n", call->call_id);
-		talloc_free(c);
-		return -1;
-	}
-
-	/* we need to force the record to be written out if this was a remote access */
-	if (c->new_data == NULL) {
-		c->new_data = &c->record_data;
-	}
-
-	if (c->new_data && updatetdb) {
-		/* XXX check that we always have the lock here? */
-		if (ctdb_ltdb_store(ctdb_db, call->key, header, *c->new_data) != 0) {
-			ctdb_set_error(ctdb, "ctdb_call tdb_store failed\n");
-			talloc_free(c);
-			return -1;
-		}
-	}
-
-	if (c->reply_data) {
-		call->reply_data = *c->reply_data;
-
-		talloc_steal(call, call->reply_data.dptr);
-		talloc_set_name_const(call->reply_data.dptr, __location__);
-	} else {
-		call->reply_data.dptr = NULL;
-		call->reply_data.dsize = 0;
-	}
-	call->status = c->status;
-
-	talloc_free(c);
-
-	return 0;
-}
-
-
-/*
-  queue a packet for sending from client to daemon
-*/
-static int ctdb_client_queue_pkt(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
-{
-	return ctdb_queue_send(ctdb->daemon.queue, (uint8_t *)hdr, hdr->length);
-}
-
-
-/*
-  called when a CTDB_REPLY_CALL packet comes in in the client
-
-  This packet comes in response to a CTDB_REQ_CALL request packet. It
-  contains any reply data from the call
-*/
-static void ctdb_client_reply_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
-{
-	struct ctdb_reply_call_old *c = (struct ctdb_reply_call_old *)hdr;
-	struct ctdb_client_call_state *state;
-
-	state = reqid_find(ctdb->idr, hdr->reqid, struct ctdb_client_call_state);
-	if (state == NULL) {
-		DEBUG(DEBUG_ERR,(__location__ " reqid %u not found\n", hdr->reqid));
-		return;
-	}
-
-	if (hdr->reqid != state->reqid) {
-		/* we found a record  but it was the wrong one */
-		DEBUG(DEBUG_ERR, ("Dropped client call reply with reqid:%u\n",hdr->reqid));
-		return;
-	}
-
-	state->call->reply_data.dptr = c->data;
-	state->call->reply_data.dsize = c->datalen;
-	state->call->status = c->status;
-
-	talloc_steal(state, c);
-
-	state->state = CTDB_CALL_DONE;
-
-	if (state->async.fn) {
-		state->async.fn(state);
-	}
-}
-
-void ctdb_request_message(struct ctdb_context *ctdb,
-			  struct ctdb_req_header *hdr)
-{
-	struct ctdb_req_message_old *c = (struct ctdb_req_message_old *)hdr;
-	TDB_DATA data;
-
-	data.dsize = c->datalen;
-	data.dptr = talloc_memdup(c, &c->data[0], c->datalen);
-	if (data.dptr == NULL) {
-		DEBUG(DEBUG_ERR, (__location__ " Memory allocation failure\n"));
-		return;
-	}
-
-	srvid_dispatch(ctdb->srv, c->srvid, CTDB_SRVID_ALL, data);
-}
-
-static void ctdb_client_reply_control(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
-
-/*
-  this is called in the client, when data comes in from the daemon
- */
-void ctdb_client_read_cb(uint8_t *data, size_t cnt, void *args)
-{
-	struct ctdb_context *ctdb = talloc_get_type(args, struct ctdb_context);
-	struct ctdb_req_header *hdr = (struct ctdb_req_header *)data;
-	TALLOC_CTX *tmp_ctx;
-
-	/* place the packet as a child of a tmp_ctx. We then use
-	   talloc_free() below to free it. If any of the calls want
-	   to keep it, then they will steal it somewhere else, and the
-	   talloc_free() will be a no-op */
-	tmp_ctx = talloc_new(ctdb);
-	talloc_steal(tmp_ctx, hdr);
-
-	if (cnt == 0) {
-		DEBUG(DEBUG_CRIT,("Daemon has exited - shutting down client\n"));
-		exit(1);
-	}
-
-	if (cnt < sizeof(*hdr)) {
-		DEBUG(DEBUG_CRIT,("Bad packet length %u in client\n", (unsigned)cnt));
-		goto done;
-	}
-	if (cnt != hdr->length) {
-		ctdb_set_error(ctdb, "Bad header length %u expected %u in client\n", 
-			       (unsigned)hdr->length, (unsigned)cnt);
-		goto done;
-	}
-
-	if (hdr->ctdb_magic != CTDB_MAGIC) {
-		ctdb_set_error(ctdb, "Non CTDB packet rejected in client\n");
-		goto done;
-	}
-
-	if (hdr->ctdb_version != CTDB_PROTOCOL) {
-		ctdb_set_error(ctdb, "Bad CTDB version 0x%x rejected in client\n", hdr->ctdb_version);
-		goto done;
-	}
-
-	switch (hdr->operation) {
-	case CTDB_REPLY_CALL:
-		ctdb_client_reply_call(ctdb, hdr);
-		break;
-
-	case CTDB_REQ_MESSAGE:
-		ctdb_request_message(ctdb, hdr);
-		break;
-
-	case CTDB_REPLY_CONTROL:
-		ctdb_client_reply_control(ctdb, hdr);
-		break;
-
-	default:
-		DEBUG(DEBUG_CRIT,("bogus operation code:%u\n",hdr->operation));
-	}
-
-done:
-	talloc_free(tmp_ctx);
-}
-
-/*
-  connect to a unix domain socket
-*/
-int ctdb_socket_connect(struct ctdb_context *ctdb)
-{
-	struct sockaddr_un addr;
-	int ret;
-
-	memset(&addr, 0, sizeof(addr));
-	addr.sun_family = AF_UNIX;
-	strncpy(addr.sun_path, ctdb->daemon.name, sizeof(addr.sun_path)-1);
-
-	ctdb->daemon.sd = socket(AF_UNIX, SOCK_STREAM, 0);
-	if (ctdb->daemon.sd == -1) {
-		DEBUG(DEBUG_ERR,(__location__ " Failed to open client socket. Errno:%s(%d)\n", strerror(errno), errno));
-		return -1;
-	}
-
-	if (connect(ctdb->daemon.sd, (struct sockaddr *)&addr, sizeof(addr)) == -1) {
-		DEBUG(DEBUG_ERR,
-		      (__location__
-		       "Failed to connect client socket to daemon (%s)\n",
-		       strerror(errno)));
-		close(ctdb->daemon.sd);
-		ctdb->daemon.sd = -1;
-		return -1;
-	}
-
-	ret = set_blocking(ctdb->daemon.sd, false);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR,
-		      (__location__
-		       " failed to set socket non-blocking (%s)\n",
-		       strerror(errno)));
-		close(ctdb->daemon.sd);
-		ctdb->daemon.sd = -1;
-		return -1;
-	}
-
-	set_close_on_exec(ctdb->daemon.sd);
-
-	ctdb->daemon.queue = ctdb_queue_setup(ctdb, ctdb, ctdb->daemon.sd, 
-					      CTDB_DS_ALIGNMENT, 
-					      ctdb_client_read_cb, ctdb, "to-ctdbd");
-	return 0;
-}
-
-
-struct ctdb_record_handle {
-	struct ctdb_db_context *ctdb_db;
-	TDB_DATA key;
-	TDB_DATA *data;
-	struct ctdb_ltdb_header header;
-};
-
-
-/*
-  make a recv call to the local ctdb daemon - called from client context
-
-  This is called when the program wants to wait for a ctdb_call to complete and get the 
-  results. This call will block unless the call has already completed.
-*/
-int ctdb_call_recv(struct ctdb_client_call_state *state, struct ctdb_call *call)
-{
-	if (state == NULL) {
-		return -1;
-	}
-
-	while (state->state < CTDB_CALL_DONE) {
-		tevent_loop_once(state->ctdb_db->ctdb->ev);
-	}
-	if (state->state != CTDB_CALL_DONE) {
-		DEBUG(DEBUG_ERR,(__location__ " ctdb_call_recv failed\n"));
-		talloc_free(state);
-		return -1;
-	}
-
-	if (state->call->reply_data.dsize) {
-		call->reply_data.dptr = talloc_memdup(state->ctdb_db,
-						      state->call->reply_data.dptr,
-						      state->call->reply_data.dsize);
-		call->reply_data.dsize = state->call->reply_data.dsize;
-	} else {
-		call->reply_data.dptr = NULL;
-		call->reply_data.dsize = 0;
-	}
-	call->status = state->call->status;
-	talloc_free(state);
-
-	return call->status;
-}
-
-
-
-
-/*
-  destroy a ctdb_call in client
-*/
-static int ctdb_client_call_destructor(struct ctdb_client_call_state *state)	
-{
-	reqid_remove(state->ctdb_db->ctdb->idr, state->reqid);
-	return 0;
-}
-
-/*
-  construct an event driven local ctdb_call
-
-  this is used so that locally processed ctdb_call requests are processed
-  in an event driven manner
-*/
-static struct ctdb_client_call_state *ctdb_client_call_local_send(struct ctdb_db_context *ctdb_db, 
-								  struct ctdb_call *call,
-								  struct ctdb_ltdb_header *header,
-								  TDB_DATA *data)
-{
-	struct ctdb_client_call_state *state;
-	struct ctdb_context *ctdb = ctdb_db->ctdb;
-	int ret;
-
-	state = talloc_zero(ctdb_db, struct ctdb_client_call_state);
-	CTDB_NO_MEMORY_NULL(ctdb, state);
-	state->call = talloc_zero(state, struct ctdb_call);
-	CTDB_NO_MEMORY_NULL(ctdb, state->call);
-
-	talloc_steal(state, data->dptr);
-
-	state->state   = CTDB_CALL_DONE;
-	*(state->call) = *call;
-	state->ctdb_db = ctdb_db;
-
-	ret = ctdb_call_local(ctdb_db, state->call, header, state, data, true);
-	if (ret != 0) {
-		DEBUG(DEBUG_DEBUG,("ctdb_call_local() failed, ignoring return code %d\n", ret));
-	}
-
-	return state;
-}
-
-/*
-  make a ctdb call to the local daemon - async send. Called from client context.
-
-  This constructs a ctdb_call request and queues it for processing. 
-  This call never blocks.
-*/
-struct ctdb_client_call_state *ctdb_call_send(struct ctdb_db_context *ctdb_db, 
-					      struct ctdb_call *call)
-{
-	struct ctdb_client_call_state *state;
-	struct ctdb_context *ctdb = ctdb_db->ctdb;
-	struct ctdb_ltdb_header header;
-	TDB_DATA data;
-	int ret;
-	size_t len;
-	struct ctdb_req_call_old *c;
-
-	/* if the domain socket is not yet open, open it */
-	if (ctdb->daemon.sd==-1) {
-		ctdb_socket_connect(ctdb);
-	}
-
-	ret = ctdb_ltdb_lock(ctdb_db, call->key);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR,(__location__ " Failed to get chainlock\n"));
-		return NULL;
-	}
-
-	ret = ctdb_ltdb_fetch(ctdb_db, call->key, &header, ctdb_db, &data);
-
-	if ((call->flags & CTDB_IMMEDIATE_MIGRATION) && (header.flags & CTDB_REC_RO_HAVE_DELEGATIONS)) {
-		ret = -1;
-	}
-
-	if (ret == 0 && header.dmaster == ctdb->pnn) {
-		state = ctdb_client_call_local_send(ctdb_db, call, &header, &data);
-		talloc_free(data.dptr);
-		ctdb_ltdb_unlock(ctdb_db, call->key);
-		return state;
-	}
-
-	ctdb_ltdb_unlock(ctdb_db, call->key);
-	talloc_free(data.dptr);
-
-	state = talloc_zero(ctdb_db, struct ctdb_client_call_state);
-	if (state == NULL) {
-		DEBUG(DEBUG_ERR, (__location__ " failed to allocate state\n"));
-		return NULL;
-	}
-	state->call = talloc_zero(state, struct ctdb_call);
-	if (state->call == NULL) {
-		DEBUG(DEBUG_ERR, (__location__ " failed to allocate state->call\n"));
-		return NULL;
-	}
-
-	len = offsetof(struct ctdb_req_call_old, data) + call->key.dsize + call->call_data.dsize;
-	c = ctdbd_allocate_pkt(ctdb, state, CTDB_REQ_CALL, len, struct ctdb_req_call_old);
-	if (c == NULL) {
-		DEBUG(DEBUG_ERR, (__location__ " failed to allocate packet\n"));
-		return NULL;
-	}
-
-	state->reqid     = reqid_new(ctdb->idr, state);
-	state->ctdb_db = ctdb_db;
-	talloc_set_destructor(state, ctdb_client_call_destructor);
-
-	c->hdr.reqid     = state->reqid;
-	c->flags         = call->flags;
-	c->db_id         = ctdb_db->db_id;
-	c->callid        = call->call_id;
-	c->hopcount      = 0;
-	c->keylen        = call->key.dsize;
-	c->calldatalen   = call->call_data.dsize;
-	memcpy(&c->data[0], call->key.dptr, call->key.dsize);
-	memcpy(&c->data[call->key.dsize], 
-	       call->call_data.dptr, call->call_data.dsize);
-	*(state->call)              = *call;
-	state->call->call_data.dptr = &c->data[call->key.dsize];
-	state->call->key.dptr       = &c->data[0];
-
-	state->state  = CTDB_CALL_WAIT;
-
-
-	ctdb_client_queue_pkt(ctdb, &c->hdr);
-
-	return state;
-}
-
-
-/*
-  full ctdb_call. Equivalent to a ctdb_call_send() followed by a ctdb_call_recv()
-*/
-int ctdb_call(struct ctdb_db_context *ctdb_db, struct ctdb_call *call)
-{
-	struct ctdb_client_call_state *state;
-
-	state = ctdb_call_send(ctdb_db, call);
-	return ctdb_call_recv(state, call);
-}
-
-
-/*
-  tell the daemon what messaging srvid we will use, and register the message
-  handler function in the client
-*/
-int ctdb_client_set_message_handler(struct ctdb_context *ctdb, uint64_t srvid,
-				    srvid_handler_fn handler,
-				    void *private_data)
-{
-	int res;
-	int32_t status;
-
-	res = ctdb_control(ctdb, CTDB_CURRENT_NODE, srvid,
-			   CTDB_CONTROL_REGISTER_SRVID, 0,
-			   tdb_null, NULL, NULL, &status, NULL, NULL);
-	if (res != 0 || status != 0) {
-		DEBUG(DEBUG_ERR,
-		      ("Failed to register srvid %llu\n",
-		       (unsigned long long)srvid));
-		return -1;
-	}
-
-	/* also need to register the handler with our own ctdb structure */
-	return srvid_register(ctdb->srv, ctdb, srvid, handler, private_data);
-}
-
-/*
-  tell the daemon we no longer want a srvid
-*/
-int ctdb_client_remove_message_handler(struct ctdb_context *ctdb,
-				       uint64_t srvid, void *private_data)
-{
-	int res;
-	int32_t status;
-
-	res = ctdb_control(ctdb, CTDB_CURRENT_NODE, srvid,
-			   CTDB_CONTROL_DEREGISTER_SRVID, 0,
-			   tdb_null, NULL, NULL, &status, NULL, NULL);
-	if (res != 0 || status != 0) {
-		DEBUG(DEBUG_ERR,
-		      ("Failed to deregister srvid %llu\n",
-		       (unsigned long long)srvid));
-		return -1;
-	}
-
-	/* also need to register the handler with our own ctdb structure */
-	srvid_deregister(ctdb->srv, srvid, private_data);
-	return 0;
-}
-
-/*
-  send a message - from client context
- */
-int ctdb_client_send_message(struct ctdb_context *ctdb, uint32_t pnn,
-		      uint64_t srvid, TDB_DATA data)
-{
-	struct ctdb_req_message_old *r;
-	int len, res;
-
-	len = offsetof(struct ctdb_req_message_old, data) + data.dsize;
-	r = ctdbd_allocate_pkt(ctdb, ctdb, CTDB_REQ_MESSAGE, 
-			       len, struct ctdb_req_message_old);
-	CTDB_NO_MEMORY(ctdb, r);
-
-	r->hdr.destnode  = pnn;
-	r->srvid         = srvid;
-	r->datalen       = data.dsize;
-	memcpy(&r->data[0], data.dptr, data.dsize);
-	
-	res = ctdb_client_queue_pkt(ctdb, &r->hdr);
-	talloc_free(r);
-	return res;
-}
-
-
-/*
-   called when a control completes or timesout to invoke the callback
-   function the user provided
-*/
-static void invoke_control_callback(struct tevent_context *ev,
-				    struct tevent_timer *te,
-				    struct timeval t, void *private_data)
-{
-	struct ctdb_client_control_state *state;
-	TALLOC_CTX *tmp_ctx = talloc_new(NULL);
-	int ret;
-
-	state = talloc_get_type(private_data, struct ctdb_client_control_state);
-	talloc_steal(tmp_ctx, state);
-
-	ret = ctdb_control_recv(state->ctdb, state, state,
-			NULL, 
-			NULL, 
-			NULL);
-	if (ret != 0) {
-		DEBUG(DEBUG_DEBUG,("ctdb_control_recv() failed, ignoring return code %d\n", ret));
-	}
-
-	talloc_free(tmp_ctx);
-}
-
-/*
-  called when a CTDB_REPLY_CONTROL packet comes in in the client
-
-  This packet comes in response to a CTDB_REQ_CONTROL request packet. It
-  contains any reply data from the control
-*/
-static void ctdb_client_reply_control(struct ctdb_context *ctdb, 
-				      struct ctdb_req_header *hdr)
-{
-	struct ctdb_reply_control_old *c = (struct ctdb_reply_control_old *)hdr;
-	struct ctdb_client_control_state *state;
-
-	state = reqid_find(ctdb->idr, hdr->reqid, struct ctdb_client_control_state);
-	if (state == NULL) {
-		DEBUG(DEBUG_ERR,(__location__ " reqid %u not found\n", hdr->reqid));
-		return;
-	}
-
-	if (hdr->reqid != state->reqid) {
-		/* we found a record  but it was the wrong one */
-		DEBUG(DEBUG_ERR, ("Dropped orphaned reply control with reqid:%u\n",hdr->reqid));
-		return;
-	}
-
-	state->outdata.dptr = c->data;
-	state->outdata.dsize = c->datalen;
-	state->status = c->status;
-	if (c->errorlen) {
-		state->errormsg = talloc_strndup(state, 
-						 (char *)&c->data[c->datalen], 
-						 c->errorlen);
-	}
-
-	/* state->outdata now uses resources from c so we don't want c
-	   to just dissappear from under us while state is still alive
-	*/
-	talloc_steal(state, c);
-
-	state->state = CTDB_CONTROL_DONE;
-
-	/* if we had a callback registered for this control, pull the response
-	   and call the callback.
-	*/
-	if (state->async.fn) {
-		tevent_add_timer(ctdb->ev, state, timeval_zero(),
-				 invoke_control_callback, state);
-	}
-}
-
-
-/*
-  destroy a ctdb_control in client
-*/
-static int ctdb_client_control_destructor(struct ctdb_client_control_state *state)
-{
-	reqid_remove(state->ctdb->idr, state->reqid);
-	return 0;
-}
-
-
-/* time out handler for ctdb_control */
-static void control_timeout_func(struct tevent_context *ev,
-				 struct tevent_timer *te,
-				 struct timeval t, void *private_data)
-{
-	struct ctdb_client_control_state *state = talloc_get_type(private_data, struct ctdb_client_control_state);
-
-	DEBUG(DEBUG_ERR,(__location__ " control timed out. reqid:%u opcode:%u "
-			 "dstnode:%u\n", state->reqid, state->c->opcode,
-			 state->c->hdr.destnode));
-
-	state->state = CTDB_CONTROL_TIMEOUT;
-
-	/* if we had a callback registered for this control, pull the response
-	   and call the callback.
-	*/
-	if (state->async.fn) {
-		tevent_add_timer(state->ctdb->ev, state, timeval_zero(),
-				 invoke_control_callback, state);
-	}
-}
-
-/* async version of send control request */
-struct ctdb_client_control_state *ctdb_control_send(struct ctdb_context *ctdb, 
-		uint32_t destnode, uint64_t srvid, 
-		uint32_t opcode, uint32_t flags, TDB_DATA data, 
-		TALLOC_CTX *mem_ctx,
-		struct timeval *timeout,
-		char **errormsg)
-{
-	struct ctdb_client_control_state *state;
-	size_t len;
-	struct ctdb_req_control_old *c;
-	int ret;
-
-	if (errormsg) {
-		*errormsg = NULL;
-	}
-
-	/* if the domain socket is not yet open, open it */
-	if (ctdb->daemon.sd==-1) {
-		ctdb_socket_connect(ctdb);
-	}
-
-	state = talloc_zero(mem_ctx, struct ctdb_client_control_state);
-	CTDB_NO_MEMORY_NULL(ctdb, state);
-
-	state->ctdb       = ctdb;
-	state->reqid      = reqid_new(ctdb->idr, state);
-	state->state      = CTDB_CONTROL_WAIT;
-	state->errormsg   = NULL;
-
-	talloc_set_destructor(state, ctdb_client_control_destructor);
-
-	len = offsetof(struct ctdb_req_control_old, data) + data.dsize;
-	c = ctdbd_allocate_pkt(ctdb, state, CTDB_REQ_CONTROL, 
-			       len, struct ctdb_req_control_old);
-	state->c            = c;	
-	CTDB_NO_MEMORY_NULL(ctdb, c);
-	c->hdr.reqid        = state->reqid;
-	c->hdr.destnode     = destnode;
-	c->opcode           = opcode;
-	c->client_id        = 0;
-	c->flags            = flags;
-	c->srvid            = srvid;
-	c->datalen          = data.dsize;
-	if (data.dsize) {
-		memcpy(&c->data[0], data.dptr, data.dsize);
-	}
-
-	/* timeout */
-	if (timeout && !timeval_is_zero(timeout)) {
-		tevent_add_timer(ctdb->ev, state, *timeout,
-				 control_timeout_func, state);
-	}
-
-	ret = ctdb_client_queue_pkt(ctdb, &(c->hdr));
-	if (ret != 0) {
-		talloc_free(state);
-		return NULL;
-	}
-
-	if (flags & CTDB_CTRL_FLAG_NOREPLY) {
-		talloc_free(state);
-		return NULL;
-	}
-
-	return state;
-}
-
-
-/* async version of receive control reply */
-int ctdb_control_recv(struct ctdb_context *ctdb, 
-		struct ctdb_client_control_state *state, 
-		TALLOC_CTX *mem_ctx,
-		TDB_DATA *outdata, int32_t *status, char **errormsg)
-{
-	TALLOC_CTX *tmp_ctx;
-
-	if (status != NULL) {
-		*status = -1;
-	}
-	if (errormsg != NULL) {
-		*errormsg = NULL;
-	}
-
-	if (state == NULL) {
-		return -1;
-	}
-
-	/* prevent double free of state */
-	tmp_ctx = talloc_new(ctdb);
-	talloc_steal(tmp_ctx, state);
-
-	/* loop one event at a time until we either timeout or the control
-	   completes.
-	*/
-	while (state->state == CTDB_CONTROL_WAIT) {
-		tevent_loop_once(ctdb->ev);
-	}
-
-	if (state->state != CTDB_CONTROL_DONE) {
-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control_recv failed\n"));
-		if (state->async.fn) {
-			state->async.fn(state);
-		}
-		talloc_free(tmp_ctx);
-		return -1;
-	}
-
-	if (state->errormsg) {
-		int s = (state->status == 0 ? -1 : state->status);
-		DEBUG(DEBUG_ERR,("ctdb_control error: '%s'\n", state->errormsg));
-		if (errormsg) {
-			(*errormsg) = talloc_move(mem_ctx, &state->errormsg);
-		}
-		if (state->async.fn) {
-			state->async.fn(state);
-		}
-		talloc_free(tmp_ctx);
-		return s;
-	}
-
-	if (outdata) {
-		*outdata = state->outdata;
-		outdata->dptr = talloc_memdup(mem_ctx, outdata->dptr, outdata->dsize);
-	}
-
-	if (status) {
-		*status = state->status;
-	}
-
-	if (state->async.fn) {
-		state->async.fn(state);
-	}
-
-	talloc_free(tmp_ctx);
-	return 0;
-}
-
-
-
-/*
-  send a ctdb control message
-  timeout specifies how long we should wait for a reply.
-  if timeout is NULL we wait indefinitely
- */
-int ctdb_control(struct ctdb_context *ctdb, uint32_t destnode, uint64_t srvid, 
-		 uint32_t opcode, uint32_t flags, TDB_DATA data, 
-		 TALLOC_CTX *mem_ctx, TDB_DATA *outdata, int32_t *status,
-		 struct timeval *timeout,
-		 char **errormsg)
-{
-	struct ctdb_client_control_state *state;
-
-	state = ctdb_control_send(ctdb, destnode, srvid, opcode, 
-			flags, data, mem_ctx,
-			timeout, errormsg);
-
-	/* FIXME: Error conditions in ctdb_control_send return NULL without
-	 * setting errormsg.  So, there is no way to distinguish between sucess
-	 * and failure when CTDB_CTRL_FLAG_NOREPLY is set */
-	if (flags & CTDB_CTRL_FLAG_NOREPLY) {
-		if (status != NULL) {
-			*status = 0;
-		}
-		return 0;
-	}
-
-	return ctdb_control_recv(ctdb, state, mem_ctx, outdata, status, 
-			errormsg);
-}
-
-/*
-  get vnn map from a remote node
- */
-int ctdb_ctrl_getvnnmap(struct ctdb_context *ctdb, struct timeval timeout, uint32_t destnode, TALLOC_CTX *mem_ctx, struct ctdb_vnn_map **vnnmap)
-{
-	int ret;
-	TDB_DATA outdata;
-	int32_t res;
-	struct ctdb_vnn_map_wire *map;
-
-	ret = ctdb_control(ctdb, destnode, 0, 
-			   CTDB_CONTROL_GETVNNMAP, 0, tdb_null, 
-			   mem_ctx, &outdata, &res, &timeout, NULL);
-	if (ret != 0 || res != 0) {
-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for getvnnmap failed\n"));
-		return -1;
-	}
-	
-	map = (struct ctdb_vnn_map_wire *)outdata.dptr;
-	if (outdata.dsize < offsetof(struct ctdb_vnn_map_wire, map) ||
-	    outdata.dsize != map->size*sizeof(uint32_t) + offsetof(struct ctdb_vnn_map_wire, map)) {
-		DEBUG(DEBUG_ERR,("Bad vnn map size received in ctdb_ctrl_getvnnmap\n"));
-		return -1;
-	}
-
-	(*vnnmap) = talloc(mem_ctx, struct ctdb_vnn_map);
-	CTDB_NO_MEMORY(ctdb, *vnnmap);
-	(*vnnmap)->generation = map->generation;
-	(*vnnmap)->size       = map->size;
-	(*vnnmap)->map        = talloc_array(*vnnmap, uint32_t, map->size);
-
-	CTDB_NO_MEMORY(ctdb, (*vnnmap)->map);
-	memcpy((*vnnmap)->map, map->map, sizeof(uint32_t)*map->size);
-	talloc_free(outdata.dptr);
-		    
-	return 0;
-}
-
-
-/*
-  get the recovery mode of a remote node
- */
-struct ctdb_client_control_state *
-ctdb_ctrl_getrecmode_send(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx, struct timeval timeout, uint32_t destnode)
-{
-	return ctdb_control_send(ctdb, destnode, 0, 
-			   CTDB_CONTROL_GET_RECMODE, 0, tdb_null, 
-			   mem_ctx, &timeout, NULL);
-}
-
-int ctdb_ctrl_getrecmode_recv(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx, struct ctdb_client_control_state *state, uint32_t *recmode)
-{
-	int ret;
-	int32_t res;
-
-	ret = ctdb_control_recv(ctdb, state, mem_ctx, NULL, &res, NULL);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR,(__location__ " ctdb_ctrl_getrecmode_recv failed\n"));
-		return -1;
-	}
-
-	if (recmode) {
-		*recmode = (uint32_t)res;
-	}
-
-	return 0;
-}
-
-int ctdb_ctrl_getrecmode(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx, struct timeval timeout, uint32_t destnode, uint32_t *recmode)
-{
-	struct ctdb_client_control_state *state;
-
-	state = ctdb_ctrl_getrecmode_send(ctdb, mem_ctx, timeout, destnode);
-	return ctdb_ctrl_getrecmode_recv(ctdb, mem_ctx, state, recmode);
-}
-
-
-
-
-/*
-  set the recovery mode of a remote node
- */
-int ctdb_ctrl_setrecmode(struct ctdb_context *ctdb, struct timeval timeout, uint32_t destnode, uint32_t recmode)
-{
-	int ret;
-	TDB_DATA data;
-	int32_t res;
-
-	data.dsize = sizeof(uint32_t);
-	data.dptr = (unsigned char *)&recmode;
-
-	ret = ctdb_control(ctdb, destnode, 0, 
-			   CTDB_CONTROL_SET_RECMODE, 0, data, 
-			   NULL, NULL, &res, &timeout, NULL);
-	if (ret != 0 || res != 0) {
-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for setrecmode failed\n"));
-		return -1;
-	}
-
-	return 0;
-}
-
-
-
-/*
-  get the recovery master of a remote node
- */
-struct ctdb_client_control_state *
-ctdb_ctrl_getrecmaster_send(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx, 
-			struct timeval timeout, uint32_t destnode)
-{
-	return ctdb_control_send(ctdb, destnode, 0, 
-			   CTDB_CONTROL_GET_RECMASTER, 0, tdb_null, 
-			   mem_ctx, &timeout, NULL);
-}
-
-int ctdb_ctrl_getrecmaster_recv(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx, struct ctdb_client_control_state *state, uint32_t *recmaster)
-{
-	int ret;
-	int32_t res;
-
-	ret = ctdb_control_recv(ctdb, state, mem_ctx, NULL, &res, NULL);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR,(__location__ " ctdb_ctrl_getrecmaster_recv failed\n"));
-		return -1;
-	}
-
-	if (recmaster) {
-		*recmaster = (uint32_t)res;
-	}
-
-	return 0;
-}
-
-int ctdb_ctrl_getrecmaster(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx, struct timeval timeout, uint32_t destnode, uint32_t *recmaster)
-{
-	struct ctdb_client_control_state *state;
-
-	state = ctdb_ctrl_getrecmaster_send(ctdb, mem_ctx, timeout, destnode);
-	return ctdb_ctrl_getrecmaster_recv(ctdb, mem_ctx, state, recmaster);
-}
-
-
-/*
-  set the recovery master of a remote node
- */
-int ctdb_ctrl_setrecmaster(struct ctdb_context *ctdb, struct timeval timeout, uint32_t destnode, uint32_t recmaster)
-{
-	int ret;
-	TDB_DATA data;
-	int32_t res;
-
-	ZERO_STRUCT(data);
-	data.dsize = sizeof(uint32_t);
-	data.dptr = (unsigned char *)&recmaster;
-
-	ret = ctdb_control(ctdb, destnode, 0, 
-			   CTDB_CONTROL_SET_RECMASTER, 0, data, 
-			   NULL, NULL, &res, &timeout, NULL);
-	if (ret != 0 || res != 0) {
-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for setrecmaster failed\n"));
-		return -1;
-	}
-
-	return 0;
-}
-
-
-/*
-  get a list of databases off a remote node
- */
-int ctdb_ctrl_getdbmap(struct ctdb_context *ctdb, struct timeval timeout, uint32_t destnode, 
-		       TALLOC_CTX *mem_ctx, struct ctdb_dbid_map_old **dbmap)
-{
-	int ret;
-	TDB_DATA outdata;
-	int32_t res;
-
-	ret = ctdb_control(ctdb, destnode, 0, 
-			   CTDB_CONTROL_GET_DBMAP, 0, tdb_null, 
-			   mem_ctx, &outdata, &res, &timeout, NULL);
-	if (ret != 0 || res != 0) {
-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for getdbmap failed ret:%d res:%d\n", ret, res));
-		return -1;
-	}
-
-	*dbmap = (struct ctdb_dbid_map_old *)talloc_memdup(mem_ctx, outdata.dptr, outdata.dsize);
-	talloc_free(outdata.dptr);
-		    
-	return 0;
-}
-
-/*
-  get a list of nodes (vnn and flags ) from a remote node
- */
-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 ret;
-	TDB_DATA outdata;
-	int32_t res;
-
-	ret = ctdb_control(ctdb, destnode, 0,
-			   CTDB_CONTROL_GET_NODEMAP, 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));
-		return -1;
-	}
-
-	*nodemap = (struct ctdb_node_map_old *)talloc_memdup(mem_ctx, outdata.dptr, outdata.dsize);
-	talloc_free(outdata.dptr);
-	return 0;
-}
-
-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 
- */
-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 ret;
-	int32_t res;
-	TDB_DATA data;
-
-	data.dptr = (uint8_t *)&dbid;
-	data.dsize = sizeof(dbid);
-
-	ret = ctdb_control(ctdb, destnode, 0, 
-			   CTDB_CONTROL_GET_DBNAME, 0, data, 
-			   mem_ctx, &data, &res, &timeout, NULL);
-	if (ret != 0 || res != 0) {
-		return -1;
-	}
-
-	(*name) = talloc_strndup(mem_ctx, (const char *)data.dptr, data.dsize);
-	if ((*name) == NULL) {
-		return -1;
-	}
-
-	talloc_free(data.dptr);
-
-	return 0;
-}
-
-/*
-  create a database
- */
-int ctdb_ctrl_createdb(struct ctdb_context *ctdb, struct timeval timeout,
-		       uint32_t destnode, TALLOC_CTX *mem_ctx,
-		       const char *name, uint8_t db_flags, uint32_t *db_id)
-{
-	int ret;
-	int32_t res;
-	TDB_DATA data;
-	uint32_t opcode;
-
-	data.dptr = discard_const(name);
-	data.dsize = strlen(name)+1;
-
-	if (db_flags & CTDB_DB_FLAGS_PERSISTENT) {
-		opcode = CTDB_CONTROL_DB_ATTACH_PERSISTENT;
-	} else if (db_flags & CTDB_DB_FLAGS_REPLICATED) {
-		opcode = CTDB_CONTROL_DB_ATTACH_REPLICATED;
-	} else {
-		opcode = CTDB_CONTROL_DB_ATTACH;
-	}
-
-	ret = ctdb_control(ctdb, destnode, 0, opcode, 0, data,
-			   mem_ctx, &data, &res, &timeout, NULL);
-
-	if (ret != 0 || res != 0) {
-		return -1;
-	}
-
-	if (data.dsize != sizeof(uint32_t)) {
-		TALLOC_FREE(data.dptr);
-		return -1;
-	}
-	if (db_id != NULL) {
-		*db_id = *(uint32_t *)data.dptr;
-	}
-	talloc_free(data.dptr);
-
-	return 0;
-}
-
-/*
-  get debug level on a node
- */
-int ctdb_ctrl_get_debuglevel(struct ctdb_context *ctdb, uint32_t destnode, int32_t *level)
-{
-	int ret;
-	int32_t res;
-	TDB_DATA data;
-
-	ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_GET_DEBUG, 0, tdb_null, 
-			   ctdb, &data, &res, NULL, NULL);
-	if (ret != 0 || res != 0) {
-		return -1;
-	}
-	if (data.dsize != sizeof(int32_t)) {
-		DEBUG(DEBUG_ERR,("Bad control reply size in ctdb_get_debuglevel (got %u)\n",
-			 (unsigned)data.dsize));
-		return -1;
-	}
-	*level = *(int32_t *)data.dptr;
-	talloc_free(data.dptr);
-	return 0;
-}
-
-/*
- * Get db open flags
- */
-int ctdb_ctrl_db_open_flags(struct ctdb_context *ctdb, uint32_t db_id,
-			    int *tdb_flags)
-{
-	TDB_DATA indata, outdata;
-	int ret;
-	int32_t res;
-
-	indata.dptr = (uint8_t *)&db_id;
-	indata.dsize = sizeof(db_id);
-
-	ret = ctdb_control(ctdb, CTDB_CURRENT_NODE, 0,
-			   CTDB_CONTROL_DB_OPEN_FLAGS, 0, indata,
-			   ctdb, &outdata, &res, NULL, NULL);
-	if (ret != 0 || res != 0) {
-		D_ERR("ctdb control for db open flags failed\n");
-		return  -1;
-	}
-
-	if (outdata.dsize != sizeof(int32_t)) {
-		D_ERR(__location__ " expected %zi bytes, received %zi bytes\n",
-		      sizeof(int32_t), outdata.dsize);
-		talloc_free(outdata.dptr);
-		return -1;
-	}
-
-	*tdb_flags = *(int32_t *)outdata.dptr;
-	talloc_free(outdata.dptr);
-	return 0;
-}
-
-/*
-  attach to a specific database - client call
-*/
-struct ctdb_db_context *ctdb_attach(struct ctdb_context *ctdb,
-				    struct timeval timeout,
-				    const char *name,
-				    uint8_t db_flags)
-{
-	struct ctdb_db_context *ctdb_db;
-	int ret;
-	int tdb_flags;
-
-	ctdb_db = ctdb_db_handle(ctdb, name);
-	if (ctdb_db) {
-		return ctdb_db;
-	}
-
-	ctdb_db = talloc_zero(ctdb, struct ctdb_db_context);
-	CTDB_NO_MEMORY_NULL(ctdb, ctdb_db);
-
-	ctdb_db->ctdb = ctdb;
-	ctdb_db->db_name = talloc_strdup(ctdb_db, name);
-	CTDB_NO_MEMORY_NULL(ctdb, ctdb_db->db_name);
-
-	/* tell ctdb daemon to attach */
-	ret = ctdb_ctrl_createdb(ctdb, timeout, CTDB_CURRENT_NODE,
-				 ctdb_db, name, db_flags, &ctdb_db->db_id);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR,("Failed to attach to database '%s'\n", name));
-		talloc_free(ctdb_db);
-		return NULL;
-	}
-
-	ret = ctdb_ctrl_getdbpath(ctdb, timeout, CTDB_CURRENT_NODE, ctdb_db->db_id, ctdb_db, &ctdb_db->db_path);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR,("Failed to get dbpath for database '%s'\n", name));
-		talloc_free(ctdb_db);
-		return NULL;
-	}
-
-	ret = ctdb_ctrl_db_open_flags(ctdb, ctdb_db->db_id, &tdb_flags);
-	if (ret != 0) {
-		D_ERR("Failed to get tdb_flags for database '%s'\n", name);
-		talloc_free(ctdb_db);
-		return NULL;
-	}
-
-	ctdb_db->ltdb = tdb_wrap_open(ctdb_db, ctdb_db->db_path, 0, tdb_flags,
-				      O_RDWR, 0);
-	if (ctdb_db->ltdb == NULL) {
-		ctdb_set_error(ctdb, "Failed to open tdb '%s'\n", ctdb_db->db_path);
-		talloc_free(ctdb_db);
-		return NULL;
-	}
-
-	ctdb_db->db_flags = db_flags;
-
-	DLIST_ADD(ctdb->db_list, ctdb_db);
-
-	/* add well known functions */
-	ctdb_set_call(ctdb_db, ctdb_null_func, CTDB_NULL_FUNC);
-	ctdb_set_call(ctdb_db, ctdb_fetch_func, CTDB_FETCH_FUNC);
-	ctdb_set_call(ctdb_db, ctdb_fetch_with_header_func, CTDB_FETCH_WITH_HEADER_FUNC);
-
-	return ctdb_db;
-}
-
-/*
-  setup a call for a database
- */
-int ctdb_set_call(struct ctdb_db_context *ctdb_db, ctdb_fn_t fn, uint32_t id)
-{
-	struct ctdb_registered_call *call;
-
-	/* register locally */
-	call = talloc(ctdb_db, struct ctdb_registered_call);
-	call->fn = fn;
-	call->id = id;
-
-	DLIST_ADD(ctdb_db->calls, call);
-	return 0;
-}
-
-/* Freeze all databases */
-int ctdb_ctrl_freeze(struct ctdb_context *ctdb, struct timeval timeout,
-		     uint32_t destnode)
-{
-	int ret;
-	int32_t res;
-
-	ret = ctdb_control(ctdb, destnode, 0,
-			   CTDB_CONTROL_FREEZE, 0, tdb_null,
-			   NULL, NULL, &res, &timeout, NULL);
-	if (ret != 0 || res != 0) {
-		DEBUG(DEBUG_ERR, ("ctdb_ctrl_freeze_priority failed\n"));
-		return -1;
-	}
-
-	return 0;
-}
-
-/*
-  get pnn of a node, or -1
- */
-int ctdb_ctrl_getpnn(struct ctdb_context *ctdb, struct timeval timeout, uint32_t destnode)
-{
-	int ret;
-	int32_t res;
-
-	ret = ctdb_control(ctdb, destnode, 0, 
-			   CTDB_CONTROL_GET_PNN, 0, tdb_null, 
-			   NULL, NULL, &res, &timeout, NULL);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for getpnn failed\n"));
-		return -1;
-	}
-
-	return res;
-}
-
-int ctdb_ctrl_get_public_ips_flags(struct ctdb_context *ctdb,
-				   struct timeval timeout, uint32_t destnode,
-				   TALLOC_CTX *mem_ctx,
-				   uint32_t flags,
-				   struct ctdb_public_ip_list_old **ips)
-{
-	int ret;
-	TDB_DATA outdata;
-	int32_t res;
-
-	ret = ctdb_control(ctdb, destnode, 0,
-			   CTDB_CONTROL_GET_PUBLIC_IPS, flags, tdb_null,
-			   mem_ctx, &outdata, &res, &timeout, NULL);
-	if (ret != 0 || res != 0) {
-		DEBUG(DEBUG_ERR,(__location__
-				 " ctdb_control for getpublicips failed ret:%d res:%d\n",
-				 ret, res));
-		return -1;
-	}
-
-	*ips = (struct ctdb_public_ip_list_old *)talloc_memdup(mem_ctx, outdata.dptr, outdata.dsize);
-	talloc_free(outdata.dptr);
-
-	return 0;
-}
-
-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)
-{
-	return ctdb_ctrl_get_public_ips_flags(ctdb, timeout,
-					      destnode, mem_ctx,
-					      0, ips);
-}
-
-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 ret;
-	TDB_DATA outdata;
-	int32_t res;
-	struct ctdb_iface_list_old *ifaces;
-	uint32_t len;
-	uint32_t i;
-
-	ret = ctdb_control(ctdb, destnode, 0,
-			   CTDB_CONTROL_GET_IFACES, 0, tdb_null,
-			   mem_ctx, &outdata, &res, &timeout, NULL);
-	if (ret != 0 || res != 0) {
-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for get ifaces "
-				"failed ret:%d res:%d\n",
-				ret, res));
-		return -1;
-	}
-
-	len = offsetof(struct ctdb_iface_list_old, ifaces);
-	if (len > outdata.dsize) {
-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for get ifaces "
-				"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;
-	}
-
-	ifaces = (struct ctdb_iface_list_old *)outdata.dptr;
-	len += ifaces->num*sizeof(struct ctdb_iface);
-
-	if (len > outdata.dsize) {
-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for get ifaces "
-				"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 < ifaces->num; i++) {
-		ifaces->ifaces[i].name[CTDB_IFACE_SIZE] = '\0';
-	}
-
-	*_ifaces = (struct ctdb_iface_list_old *)talloc_memdup(mem_ctx,
-								  outdata.dptr,
-								  outdata.dsize);
-	talloc_free(outdata.dptr);
-	if (*_ifaces == NULL) {
-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for get ifaces "
-				"talloc_memdup size %u failed\n",
-				(unsigned int)outdata.dsize));
-		return -1;
-	}
-
-	return 0;
-}
-
-/*
-  set/clear the permanent disabled bit on a remote node
- */
-int ctdb_ctrl_modflags(struct ctdb_context *ctdb, struct timeval timeout, uint32_t destnode, 
-		       uint32_t set, uint32_t clear)
-{
-	int ret;
-	TDB_DATA data;
-	struct ctdb_node_map_old *nodemap=NULL;
-	struct ctdb_node_flag_change c;
-	TALLOC_CTX *tmp_ctx = talloc_new(ctdb);
-	uint32_t recmaster;
-	uint32_t *nodes;
-
-
-	/* find the recovery master */
-	ret = ctdb_ctrl_getrecmaster(ctdb, tmp_ctx, timeout, CTDB_CURRENT_NODE, &recmaster);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR, (__location__ " Unable to get recmaster from local node\n"));
-		talloc_free(tmp_ctx);
-		return ret;
-	}
-
-
-	/* read the node flags from the recmaster */
-	ret = ctdb_ctrl_getnodemap(ctdb, timeout, recmaster, tmp_ctx, &nodemap);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR, (__location__ " Unable to get nodemap from node %u\n", destnode));
-		talloc_free(tmp_ctx);
-		return -1;
-	}
-	if (destnode >= nodemap->num) {
-		DEBUG(DEBUG_ERR,(__location__ " Nodemap from recmaster does not contain node %d\n", destnode));
-		talloc_free(tmp_ctx);
-		return -1;
-	}
-
-	c.pnn       = destnode;
-	c.old_flags = nodemap->nodes[destnode].flags;
-	c.new_flags = c.old_flags;
-	c.new_flags |= set;
-	c.new_flags &= ~clear;
-
-	data.dsize = sizeof(c);
-	data.dptr = (unsigned char *)&c;
-
-	/* send the flags update to all connected nodes */
-	nodes = list_of_connected_nodes(ctdb, nodemap, tmp_ctx, true);
-
-	if (ctdb_client_async_control(ctdb, CTDB_CONTROL_MODIFY_FLAGS,
-					nodes, 0,
-					timeout, false, data,
-					NULL, NULL,
-					NULL) != 0) {
-		DEBUG(DEBUG_ERR, (__location__ " Unable to update nodeflags on remote nodes\n"));
-
-		talloc_free(tmp_ctx);
-		return -1;
-	}
-
-	talloc_free(tmp_ctx);
-	return 0;
-}
-
-
-/*
-  get all tunables
- */
-int ctdb_ctrl_get_all_tunables(struct ctdb_context *ctdb,
-			       struct timeval timeout,
-			       uint32_t destnode,
-			       struct ctdb_tunable_list *tunables)
-{
-	TDB_DATA outdata;
-	int ret;
-	int32_t res;
-
-	ret = ctdb_control(ctdb, destnode, 0, CTDB_CONTROL_GET_ALL_TUNABLES, 0, tdb_null, ctdb,
-			   &outdata, &res, &timeout, NULL);
-	if (ret != 0 || res != 0) {
-		DEBUG(DEBUG_ERR,(__location__ " ctdb_control for get all tunables failed\n"));
-		return -1;
-	}
-
-	if (outdata.dsize != sizeof(*tunables)) {
-		DEBUG(DEBUG_ERR,(__location__ " bad data size %u in ctdb_ctrl_get_all_tunables should be %u\n",
-			 (unsigned)outdata.dsize, (unsigned)sizeof(*tunables)));
-		return -1;
-	}
-
-	*tunables = *(struct ctdb_tunable_list *)outdata.dptr;
-	talloc_free(outdata.dptr);
-	return 0;
-}
-
-/*
-  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.
-*/
-struct ctdb_context *ctdb_init(struct tevent_context *ev)
-{
-	int ret;
-	struct ctdb_context *ctdb;
-
-	ctdb = talloc_zero(ev, struct ctdb_context);
-	if (ctdb == NULL) {
-		DEBUG(DEBUG_ERR,(__location__ " talloc_zero failed.\n"));
-		return NULL;
-	}
-	ctdb->ev  = ev;
-	/* Wrap early to exercise code. */
-	ret = reqid_init(ctdb, INT_MAX-200, &ctdb->idr);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR, ("reqid_init failed (%s)\n", strerror(ret)));
-		talloc_free(ctdb);
-		return NULL;
-	}
-
-	ret = srvid_init(ctdb, &ctdb->srv);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR, ("srvid_init failed (%s)\n", strerror(ret)));
-		talloc_free(ctdb);
-		return NULL;
-	}
-
-	ret = ctdb_set_socketname(ctdb, CTDB_SOCKET);
-	if (ret != 0) {
-		DEBUG(DEBUG_ERR,(__location__ " ctdb_set_socketname failed.\n"));
-		talloc_free(ctdb);
-		return NULL;
-	}
-
-	ctdb->statistics.statistics_start_time = timeval_current();
-
-	return ctdb;
-}
-
-
-/*
-  set some ctdb flags
-*/
-void ctdb_set_flags(struct ctdb_context *ctdb, unsigned flags)
-{
-	ctdb->flags |= flags;
-}
-
-/*
-  setup the local socket name
-*/
-int ctdb_set_socketname(struct ctdb_context *ctdb, const char *socketname)
-{
-	ctdb->daemon.name = talloc_strdup(ctdb, socketname);
-	CTDB_NO_MEMORY(ctdb, ctdb->daemon.name);
-
-	return 0;
-}
-
-const char *ctdb_get_socketname(struct ctdb_context *ctdb)
-{
-	return ctdb->daemon.name;
-}
-
-/*
-  return the pnn of this node
-*/
-uint32_t ctdb_get_pnn(struct ctdb_context *ctdb)
-{
-	return ctdb->pnn;
-}
-
-/* 
-  callback for the async helpers used when sending the same control
-  to multiple nodes in parallell.
-*/
-static void async_callback(struct ctdb_client_control_state *state)
-{
-	struct client_async_data *data = talloc_get_type(state->async.private_data, struct client_async_data);
-	struct ctdb_context *ctdb = talloc_get_type(state->ctdb, struct ctdb_context);
-	int ret;
-	TDB_DATA outdata;
-	int32_t res = -1;
-	uint32_t destnode = state->c->hdr.destnode;
-
-	outdata.dsize = 0;
-	outdata.dptr = NULL;
-
-	/* one more node has responded with recmode data */
-	data->count--;
-
-	/* if we failed to push the db, then return an error and let
-	   the main loop try again.
-	*/
-	if (state->state != CTDB_CONTROL_DONE) {
-		if ( !data->dont_log_errors) {
-			DEBUG(DEBUG_ERR,("Async operation failed with state %d, opcode:%u\n", state->state, data->opcode));
-		}
-		data->fail_count++;
-		if (state->state == CTDB_CONTROL_TIMEOUT) {
-			res = -ETIME;
-		} else {
-			res = -1;
-		}
-		if (data->fail_callback) {
-			data->fail_callback(ctdb, destnode, res, outdata,
-					data->callback_data);
-		}
-		return;
-	}
-	
-	state->async.fn = NULL;
-
-	ret = ctdb_control_recv(ctdb, state, data, &outdata, &res, NULL);
-	if ((ret != 0) || (res != 0)) {
-		if ( !data->dont_log_errors) {
-			DEBUG(DEBUG_ERR,("Async operation failed with ret=%d res=%d opcode=%u\n", ret, (int)res, data->opcode));
-		}
-		data->fail_count++;
-		if (data->fail_callback) {
-			data->fail_callback(ctdb, destnode, res, outdata,
-					data->callback_data);
-		}
-	}
-	if ((ret == 0) && (data->callback != NULL)) {
-		data->callback(ctdb, destnode, res, outdata,
-					data->callback_data);
-	}
-}
-
-
-void ctdb_client_async_add(struct client_async_data *data, struct ctdb_client_control_state *state)
-{
-	/* set up the callback functions */
-	state->async.fn = async_callback;
-	state->async.private_data = data;
-	
-	/* one more control to wait for to complete */
-	data->count++;
-}
-
-
-/* wait for up to the maximum number of seconds allowed
-   or until all nodes we expect a response from has replied
-*/
-int ctdb_client_async_wait(struct ctdb_context *ctdb, struct client_async_data *data)
-{
-	while (data->count > 0) {
-		tevent_loop_once(ctdb->ev);
-	}
-	if (data->fail_count != 0) {
-		if (!data->dont_log_errors) {
-			DEBUG(DEBUG_ERR,("Async wait failed - fail_count=%u\n", 
-				 data->fail_count));
-		}
-		return -1;
-	}
-	return 0;
-}
-
-
-/* 
-   perform a simple control on the listed nodes
-   The control cannot return data
- */
-int ctdb_client_async_control(struct ctdb_context *ctdb,
-				enum ctdb_controls opcode,
-				uint32_t *nodes,
-				uint64_t srvid,
-				struct timeval timeout,
-				bool dont_log_errors,
-				TDB_DATA data,
-				client_async_callback client_callback,
-			        client_async_callback fail_callback,
-				void *callback_data)
-{
-	struct client_async_data *async_data;
-	struct ctdb_client_control_state *state;
-	int j, num_nodes;
-
-	async_data = talloc_zero(ctdb, struct client_async_data);
-	CTDB_NO_MEMORY_FATAL(ctdb, async_data);
-	async_data->dont_log_errors = dont_log_errors;
-	async_data->callback = client_callback;
-	async_data->fail_callback = fail_callback;
-	async_data->callback_data = callback_data;
-	async_data->opcode        = opcode;
-
-	num_nodes = talloc_get_size(nodes) / sizeof(uint32_t);
-
-	/* loop over all nodes and send an async control to each of them */
-	for (j=0; j<num_nodes; j++) {
-		uint32_t pnn = nodes[j];
-
-		state = ctdb_control_send(ctdb, pnn, srvid, opcode, 
-					  0, data, async_data, &timeout, NULL);
-		if (state == NULL) {
-			DEBUG(DEBUG_ERR,(__location__ " Failed to call async control %u\n", (unsigned)opcode));
-			talloc_free(async_data);
-			return -1;
-		}
-		
-		ctdb_client_async_add(async_data, state);
-	}
-
-	if (ctdb_client_async_wait(ctdb, async_data) != 0) {
-		talloc_free(async_data);
-		return -1;
-	}
-
-	talloc_free(async_data);
-	return 0;
-}
-
-uint32_t *list_of_vnnmap_nodes(struct ctdb_context *ctdb,
-				struct ctdb_vnn_map *vnn_map,
-				TALLOC_CTX *mem_ctx,
-				bool include_self)
-{
-	int i, j, num_nodes;
-	uint32_t *nodes;
-
-	for (i=num_nodes=0;i<vnn_map->size;i++) {
-		if (vnn_map->map[i] == ctdb->pnn && !include_self) {
-			continue;
-		}
-		num_nodes++;
-	} 
-
-	nodes = talloc_array(mem_ctx, uint32_t, num_nodes);
-	CTDB_NO_MEMORY_FATAL(ctdb, nodes);
-
-	for (i=j=0;i<vnn_map->size;i++) {
-		if (vnn_map->map[i] == ctdb->pnn && !include_self) {
-			continue;
-		}
-		nodes[j++] = vnn_map->map[i];
-	} 
-
-	return nodes;
-}
-
-/* Get list of nodes not including those with flags specified by mask.
- * If exclude_pnn is not -1 then exclude that pnn from the list.
- */
-uint32_t *list_of_nodes(struct ctdb_context *ctdb,
-			struct ctdb_node_map_old *node_map,
-			TALLOC_CTX *mem_ctx,
-			uint32_t mask,
-			int exclude_pnn)
-{
-	int i, j, num_nodes;
-	uint32_t *nodes;
-
-	for (i=num_nodes=0;i<node_map->num;i++) {
-		if (node_map->nodes[i].flags & mask) {
-			continue;
-		}
-		if (node_map->nodes[i].pnn == exclude_pnn) {
-			continue;
-		}
-		num_nodes++;
-	} 
-
-	nodes = talloc_array(mem_ctx, uint32_t, num_nodes);
-	CTDB_NO_MEMORY_FATAL(ctdb, nodes);
-
-	for (i=j=0;i<node_map->num;i++) {
-		if (node_map->nodes[i].flags & mask) {
-			continue;
-		}
-		if (node_map->nodes[i].pnn == exclude_pnn) {
-			continue;
-		}
-		nodes[j++] = node_map->nodes[i].pnn;
-	} 
-
-	return nodes;
-}
-
-uint32_t *list_of_active_nodes(struct ctdb_context *ctdb,
-				struct ctdb_node_map_old *node_map,
-				TALLOC_CTX *mem_ctx,
-				bool include_self)
-{
-	return list_of_nodes(ctdb, node_map, mem_ctx, NODE_FLAGS_INACTIVE,
-			     include_self ? -1 : ctdb->pnn);
-}
-
-uint32_t *list_of_connected_nodes(struct ctdb_context *ctdb,
-				struct ctdb_node_map_old *node_map,
-				TALLOC_CTX *mem_ctx,
-				bool include_self)
-{
-	return list_of_nodes(ctdb, node_map, mem_ctx, NODE_FLAGS_DISCONNECTED,
-			     include_self ? -1 : ctdb->pnn);
-}
-
-/*
-  get capabilities of a remote node
- */
-struct ctdb_client_control_state *
-ctdb_ctrl_getcapabilities_send(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx, struct timeval timeout, uint32_t destnode)
-{
-	return ctdb_control_send(ctdb, destnode, 0, 
-			   CTDB_CONTROL_GET_CAPABILITIES, 0, tdb_null, 
-			   mem_ctx, &timeout, NULL);
-}
-
-int ctdb_ctrl_getcapabilities_recv(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx, struct ctdb_client_control_state *state, uint32_t *capabilities)
-{
-	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_getcapabilities_recv failed\n"));
-		return -1;
-	}
-
-	if (capabilities) {
-		*capabilities = *((uint32_t *)outdata.dptr);
-	}
-
-	return 0;
-}
-
-int ctdb_ctrl_getcapabilities(struct ctdb_context *ctdb, struct timeval timeout, uint32_t destnode, uint32_t *capabilities)
-{
-	struct ctdb_client_control_state *state;
-	TALLOC_CTX *tmp_ctx = talloc_new(NULL);
-	int ret;
-
-	state = ctdb_ctrl_getcapabilities_send(ctdb, tmp_ctx, timeout, destnode);
-	ret = ctdb_ctrl_getcapabilities_recv(ctdb, tmp_ctx, state, capabilities);
-	talloc_free(tmp_ctx);
-	return ret;
-}
-
-static void get_capabilities_callback(struct ctdb_context *ctdb,
-				      uint32_t node_pnn, int32_t res,
-				      TDB_DATA outdata, void *callback_data)
-{
-	struct ctdb_node_capabilities *caps =
-		talloc_get_type(callback_data,
-				struct ctdb_node_capabilities);
-
-	if ( (outdata.dsize != sizeof(uint32_t)) || (outdata.dptr == NULL) ) {
-		DEBUG(DEBUG_ERR, (__location__ " Invalid length/pointer for getcap callback : %u %p\n",  (unsigned)outdata.dsize, outdata.dptr));
-		return;
-	}
-
-	if (node_pnn >= talloc_array_length(caps)) {
-		DEBUG(DEBUG_ERR,
-		      (__location__ " unexpected PNN %u\n", node_pnn));
-		return;
-	}
-
-	caps[node_pnn].retrieved = true;
-	caps[node_pnn].capabilities = *((uint32_t *)outdata.dptr);
-}
-
-struct ctdb_node_capabilities *
-ctdb_get_capabilities(struct ctdb_context *ctdb,
-		      TALLOC_CTX *mem_ctx,
-		      struct timeval timeout,
-		      struct ctdb_node_map_old *nodemap)
-{
-	uint32_t *nodes;
-	uint32_t i, res;
-	struct ctdb_node_capabilities *ret;
-
-	nodes = list_of_active_nodes(ctdb, nodemap, mem_ctx, true);
-
-	ret = talloc_array(mem_ctx, struct ctdb_node_capabilities,
-			   nodemap->num);
-	CTDB_NO_MEMORY_NULL(ctdb, ret);
-	/* Prepopulate the expected PNNs */
-	for (i = 0; i < talloc_array_length(ret); i++) {
-		ret[i].retrieved = false;
-	}
-
-	res = ctdb_client_async_control(ctdb, CTDB_CONTROL_GET_CAPABILITIES,
-					nodes, 0, timeout,
-					false, tdb_null,
-					get_capabilities_callback, NULL,
-					ret);
-	if (res != 0) {
-		DEBUG(DEBUG_ERR,
-		      (__location__ " Failed to read node capabilities.\n"));
-		TALLOC_FREE(ret);
-	}
-
-	return ret;
-}
-
-uint32_t *
-ctdb_get_node_capabilities(struct ctdb_node_capabilities *caps,
-			   uint32_t pnn)
-{
-	if (pnn < talloc_array_length(caps) && caps[pnn].retrieved) {
-		return &caps[pnn].capabilities;
-	}
-
-	return NULL;
-}
-
-bool ctdb_node_has_capabilities(struct ctdb_node_capabilities *caps,
-				uint32_t pnn,
-				uint32_t capabilities_required)
-{
-	uint32_t *capp = ctdb_get_node_capabilities(caps, pnn);
-	return (capp != NULL) &&
-		((*capp & capabilities_required) == capabilities_required);
-}
-
-/*
-  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;
-}
-
-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;
-}
-
-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;
-
-	state = ctdb_ctrl_updaterecord_send(ctdb, mem_ctx, timeout, destnode, ctdb_db, key, header, data);
-	return ctdb_ctrl_updaterecord_recv(ctdb, state);
-}