r20889: import ctdb cluster backend from bzr

it will be interesting to see how the build farm handles this
(This used to be commit 53be449630bd67d649a9e70cc7e25a9799c0616b)

17 files changed, 3733 insertions, 1 deletions

diff --git a/source4/cluster/config.mk b/source4/cluster/config.mk
index 934bc55252..c5c2ea970a 100644
--- a/source4/cluster/config.mk
+++ b/source4/cluster/config.mk
@@ -1,4 +1,7 @@
+include ctdb/config.mk
 
 ####################
 [SUBSYSTEM::CLUSTER]
-OBJ_FILES = cluster.o
+OBJ_FILES = cluster.o \
+		local.o
+PRIVATE_DEPENDENCIES = ctdb
diff --git a/source4/cluster/ctdb/brlock_ctdb.c b/source4/cluster/ctdb/brlock_ctdb.c
new file mode 100644
index 0000000000..bcfa566b76
--- /dev/null
+++ b/source4/cluster/ctdb/brlock_ctdb.c
@@ -0,0 +1,971 @@
+/* 
+   Unix SMB/CIFS implementation.
+
+   generic byte range locking code - ctdb backend
+
+   Copyright (C) Andrew Tridgell 2006
+   
+   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 2 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, write to the Free Software
+   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+*/
+
+#include "includes.h"
+#include "system/filesys.h"
+#include "lib/tdb/include/tdb.h"
+#include "messaging/messaging.h"
+#include "db_wrap.h"
+#include "lib/messaging/irpc.h"
+#include "libcli/libcli.h"
+#include "cluster/cluster.h"
+#include "ntvfs/common/brlock.h"
+#include "cluster/ctdb/include/ctdb.h"
+
+enum my_functions {FUNC_BRL_LOCK=1, FUNC_BRL_UNLOCK=2, 
+		   FUNC_BRL_REMOVE_PENDING=3, FUNC_BRL_LOCKTEST=4,
+		   FUNC_BRL_CLOSE=5};
+
+/*
+  in this module a "DATA_BLOB *file_key" is a blob that uniquely identifies
+  a file. For a local posix filesystem this will usually be a combination
+  of the device and inode numbers of the file, but it can be anything 
+  that uniquely idetifies a file for locking purposes, as long
+  as it is applied consistently.
+*/
+
+/* this struct is typically attached to tcon */
+struct brl_context {
+	struct ctdb_context *ctdb;
+	struct server_id server;
+	struct messaging_context *messaging_ctx;
+};
+
+/*
+  the lock context contains the elements that define whether one
+  lock is the same as another lock
+*/
+struct lock_context {
+	struct server_id server;
+	uint16_t smbpid;
+	struct brl_context *ctx;
+};
+
+/* The data in brlock records is an unsorted linear array of these
+   records.  It is unnecessary to store the count as tdb provides the
+   size of the record */
+struct lock_struct {
+	struct lock_context context;
+	struct ntvfs_handle *ntvfs;
+	uint64_t start;
+	uint64_t size;
+	enum brl_type lock_type;
+	void *notify_ptr;
+};
+
+/* this struct is attached to on open file handle */
+struct brl_handle {
+	DATA_BLOB key;
+	struct ntvfs_handle *ntvfs;
+	struct lock_struct last_lock;
+};
+
+/*
+  Open up the brlock.tdb database. Close it down using
+  talloc_free(). We need the messaging_ctx to allow for
+  pending lock notifications.
+*/
+static struct brl_context *brl_ctdb_init(TALLOC_CTX *mem_ctx, struct server_id server, 
+				    struct messaging_context *messaging_ctx)
+{
+	struct ctdb_context *ctdb = talloc_get_type(cluster_private(), struct ctdb_context);
+	struct brl_context *brl;
+
+	brl = talloc(mem_ctx, struct brl_context);
+	if (brl == NULL) {
+		return NULL;
+	}
+
+	brl->ctdb = ctdb;
+	brl->server = server;
+	brl->messaging_ctx = messaging_ctx;
+
+	DEBUG(0,("brl_ctdb_init: brl=%p\n", brl));
+
+	return brl;
+}
+
+static struct brl_handle *brl_ctdb_create_handle(TALLOC_CTX *mem_ctx, struct ntvfs_handle *ntvfs, 
+						    DATA_BLOB *file_key)
+{
+	struct brl_handle *brlh;
+
+	brlh = talloc(mem_ctx, struct brl_handle);
+	if (brlh == NULL) {
+		return NULL;
+	}
+
+	brlh->key = *file_key;
+	brlh->ntvfs = ntvfs;
+	ZERO_STRUCT(brlh->last_lock);
+
+	return brlh;
+}
+
+/*
+  see if two locking contexts are equal
+*/
+static BOOL brl_ctdb_same_context(struct lock_context *ctx1, struct lock_context *ctx2)
+{
+	return (cluster_id_equal(&ctx1->server, &ctx2->server) &&
+		ctx1->smbpid == ctx2->smbpid &&
+		ctx1->ctx == ctx2->ctx);
+}
+
+/*
+  see if lck1 and lck2 overlap
+*/
+static BOOL brl_ctdb_overlap(struct lock_struct *lck1, 
+			struct lock_struct *lck2)
+{
+	/* this extra check is not redundent - it copes with locks
+	   that go beyond the end of 64 bit file space */
+	if (lck1->size != 0 &&
+	    lck1->start == lck2->start &&
+	    lck1->size == lck2->size) {
+		return True;
+	}
+	    
+	if (lck1->start >= (lck2->start+lck2->size) ||
+	    lck2->start >= (lck1->start+lck1->size)) {
+		return False;
+	}
+	return True;
+} 
+
+/*
+ See if lock2 can be added when lock1 is in place.
+*/
+static BOOL brl_ctdb_conflict(struct lock_struct *lck1, 
+			 struct lock_struct *lck2)
+{
+	/* pending locks don't conflict with anything */
+	if (lck1->lock_type >= PENDING_READ_LOCK ||
+	    lck2->lock_type >= PENDING_READ_LOCK) {
+		return False;
+	}
+
+	if (lck1->lock_type == READ_LOCK && lck2->lock_type == READ_LOCK) {
+		return False;
+	}
+
+	if (brl_ctdb_same_context(&lck1->context, &lck2->context) &&
+	    lck2->lock_type == READ_LOCK && lck1->ntvfs == lck2->ntvfs) {
+		return False;
+	}
+
+	return brl_ctdb_overlap(lck1, lck2);
+} 
+
+
+/*
+ Check to see if this lock conflicts, but ignore our own locks on the
+ same fnum only.
+*/
+static BOOL brl_ctdb_conflict_other(struct lock_struct *lck1, struct lock_struct *lck2)
+{
+	/* pending locks don't conflict with anything */
+	if (lck1->lock_type >= PENDING_READ_LOCK ||
+	    lck2->lock_type >= PENDING_READ_LOCK) {
+		return False;
+	}
+
+	if (lck1->lock_type == READ_LOCK && lck2->lock_type == READ_LOCK) 
+		return False;
+
+	/*
+	 * note that incoming write calls conflict with existing READ
+	 * locks even if the context is the same. JRA. See LOCKTEST7
+	 * in smbtorture.
+	 */
+	if (brl_ctdb_same_context(&lck1->context, &lck2->context) &&
+	    lck1->ntvfs == lck2->ntvfs &&
+	    (lck2->lock_type == READ_LOCK || lck1->lock_type == WRITE_LOCK)) {
+		return False;
+	}
+
+	return brl_ctdb_overlap(lck1, lck2);
+} 
+
+
+/*
+  amazingly enough, w2k3 "remembers" whether the last lock failure
+  is the same as this one and changes its error code. I wonder if any
+  app depends on this?
+*/
+static NTSTATUS brl_ctdb_lock_failed(struct brl_handle *brlh, struct lock_struct *lock)
+{
+	/*
+	 * this function is only called for non pending lock!
+	 */
+
+	/* 
+	 * if the notify_ptr is non NULL,
+	 * it means that we're at the end of a pending lock
+	 * and the real lock is requested after the timeout went by
+	 * In this case we need to remember the last_lock and always
+	 * give FILE_LOCK_CONFLICT
+	 */
+	if (lock->notify_ptr) {
+		brlh->last_lock = *lock;
+		return NT_STATUS_FILE_LOCK_CONFLICT;
+	}
+
+	/* 
+	 * amazing the little things you learn with a test
+	 * suite. Locks beyond this offset (as a 64 bit
+	 * number!) always generate the conflict error code,
+	 * unless the top bit is set
+	 */
+	if (lock->start >= 0xEF000000 && (lock->start >> 63) == 0) {
+		brlh->last_lock = *lock;
+		return NT_STATUS_FILE_LOCK_CONFLICT;
+	}
+
+	/*
+	 * if the current lock matches the last failed lock on the file handle
+	 * and starts at the same offset, then FILE_LOCK_CONFLICT should be returned
+	 */
+	if (cluster_id_equal(&lock->context.server, &brlh->last_lock.context.server) &&
+	    lock->context.ctx == brlh->last_lock.context.ctx &&
+	    lock->ntvfs == brlh->last_lock.ntvfs &&
+	    lock->start == brlh->last_lock.start) {
+		return NT_STATUS_FILE_LOCK_CONFLICT;
+	}
+
+	brlh->last_lock = *lock;
+	return NT_STATUS_LOCK_NOT_GRANTED;
+}
+
+
+static void show_locks(const char *op, struct lock_struct *locks, int count)
+{
+	int i;
+	DEBUG(0,("OP: %s\n", op));
+	for (i=0;i<count;i++) {
+		DEBUG(0,("%2d: %4d %4d %d.%d.%d %p %p\n",
+			 i, (int)locks[i].start, (int)locks[i].size, 
+			 locks[i].context.server.node,
+			 locks[i].context.server.id,
+			 locks[i].context.smbpid,
+			 locks[i].context.ctx,
+			 locks[i].ntvfs));
+	}
+}
+
+
+struct ctdb_lock_req {
+	uint16_t smbpid;
+	uint64_t start;
+	uint64_t size;
+	enum brl_type lock_type;
+	void *notify_ptr;
+	struct server_id server;
+	struct brl_context *brl;
+	struct ntvfs_handle *ntvfs;
+};
+
+/*
+  ctdb call handling brl_lock()
+*/
+static int brl_ctdb_lock_func(struct ctdb_call *call)
+{
+	struct ctdb_lock_req *req = (struct ctdb_lock_req *)call->call_data->dptr;
+	TDB_DATA dbuf;
+	int count=0, i;
+	struct lock_struct lock, *locks=NULL;
+	NTSTATUS status = NT_STATUS_OK;
+
+#if 0
+	/* if this is a pending lock, then with the chainlock held we
+	   try to get the real lock. If we succeed then we don't need
+	   to make it pending. This prevents a possible race condition
+	   where the pending lock gets created after the lock that is
+	   preventing the real lock gets removed */
+	if (lock_type >= PENDING_READ_LOCK) {
+		enum brl_type rw = (lock_type==PENDING_READ_LOCK? READ_LOCK : WRITE_LOCK);
+
+		/* here we need to force that the last_lock isn't overwritten */
+		lock = brlh->last_lock;
+		status = brl_ctdb_lock(brl, brlh, smbpid, start, size, rw, NULL);
+		brlh->last_lock = lock;
+
+		if (NT_STATUS_IS_OK(status)) {
+			tdb_chainunlock(brl->w->tdb, kbuf);
+			return NT_STATUS_OK;
+		}
+	}
+#endif
+
+	dbuf = call->record_data;
+
+	ZERO_STRUCT(lock);
+	lock.context.smbpid = req->smbpid;
+	lock.context.server = req->server;
+	lock.context.ctx = req->brl;
+	lock.ntvfs = req->ntvfs;
+	lock.start = req->start;
+	lock.size = req->size;
+	lock.lock_type = req->lock_type;
+	lock.notify_ptr = req->notify_ptr;
+
+	{
+		int xlen = sizeof(lock);
+		uint8_t *xx = &lock;
+		int ii, fd = open("/dev/null", O_WRONLY);
+		for (ii=0;ii<xlen;ii++) {
+			write(fd, &xx[ii], 1);
+		}
+		close(fd);
+	}
+
+	if (dbuf.dptr) {
+		/* there are existing locks - make sure they don't conflict */
+		locks = (struct lock_struct *)dbuf.dptr;
+		count = dbuf.dsize / sizeof(*locks);
+
+		show_locks("lock", locks, count);
+
+		for (i=0; i<count; i++) {
+			if (brl_ctdb_conflict(&locks[i], &lock)) {
+				status = NT_STATUS_LOCK_NOT_GRANTED;
+				goto reply;
+			}
+		}
+	}
+
+	call->new_data = talloc(call, TDB_DATA);
+	if (call->new_data == NULL) {
+		return CTDB_ERR_NOMEM;
+	}
+
+	call->new_data->dptr = talloc_size(call, dbuf.dsize + sizeof(lock));
+	if (call->new_data->dptr == NULL) {
+		return CTDB_ERR_NOMEM;
+	}
+	memcpy(call->new_data->dptr, locks, dbuf.dsize);
+	memcpy(call->new_data->dptr+dbuf.dsize, &lock, sizeof(lock));
+	call->new_data->dsize = dbuf.dsize + sizeof(lock);
+
+	if (req->lock_type >= PENDING_READ_LOCK) {
+		status = NT_STATUS_LOCK_NOT_GRANTED;
+	}
+
+	DEBUG(0,("lock: size now %d\n", call->new_data->dsize));
+
+reply:
+	call->reply_data = talloc(call, TDB_DATA);
+	if (call->reply_data == NULL) {
+		return CTDB_ERR_NOMEM;
+	}
+
+	call->reply_data->dptr = talloc_size(call, sizeof(NTSTATUS));
+	call->reply_data->dsize = sizeof(NTSTATUS);
+	if (call->reply_data->dptr == NULL) {
+		return CTDB_ERR_NOMEM;
+	}
+	*(NTSTATUS *)call->reply_data->dptr = status;
+
+	return 0;
+}
+
+
+/*
+  Lock a range of bytes.  The lock_type can be a PENDING_*_LOCK, in
+  which case a real lock is first tried, and if that fails then a
+  pending lock is created. When the pending lock is triggered (by
+  someone else closing an overlapping lock range) a messaging
+  notification is sent, identified by the notify_ptr
+*/
+static NTSTATUS brl_ctdb_lock(struct brl_context *brl,
+			      struct brl_handle *brlh,
+			      uint16_t smbpid,
+			      uint64_t start, uint64_t size, 
+			      enum brl_type lock_type,
+			      void *notify_ptr)
+{
+	TDB_DATA kbuf, rbuf, sbuf;
+	struct ctdb_lock_req req;
+	NTSTATUS status;
+	int ret;
+
+	kbuf.dptr = brlh->key.data;
+	kbuf.dsize = brlh->key.length;
+
+	rbuf.dptr = (uint8_t *)&req;
+	rbuf.dsize = sizeof(req);
+
+	ZERO_STRUCT(req);
+	req.smbpid = smbpid;
+	req.start  = start;
+	req.size   = size;
+	req.lock_type = lock_type;
+	req.notify_ptr = notify_ptr;
+	req.server = brl->server;
+	req.brl = brl;
+	req.ntvfs = brlh->ntvfs;
+		
+	ret = ctdb_call(brl->ctdb, kbuf, FUNC_BRL_LOCK, &rbuf, &sbuf);
+	if (ret == -1) {
+		DEBUG(0,("ctdb_call failed - %s\n", __location__));
+		return NT_STATUS_INTERNAL_DB_CORRUPTION;
+	}
+
+	status = *(NTSTATUS *)sbuf.dptr;
+	talloc_free(sbuf.dptr);
+
+	return status;
+}
+
+#if 0
+/*
+  we are removing a lock that might be holding up a pending lock. Scan for pending
+  locks that cover this range and if we find any then notify the server that it should
+  retry the lock
+*/
+static void brl_ctdb_notify_unlock(struct brl_context *brl,
+			      struct lock_struct *locks, int count, 
+			      struct lock_struct *removed_lock)
+{
+	int i, last_notice;
+
+	/* the last_notice logic is to prevent stampeding on a lock
+	   range. It prevents us sending hundreds of notifies on the
+	   same range of bytes. It doesn't prevent all possible
+	   stampedes, but it does prevent the most common problem */
+	last_notice = -1;
+
+	for (i=0;i<count;i++) {
+		if (locks[i].lock_type >= PENDING_READ_LOCK &&
+		    brl_ctdb_overlap(&locks[i], removed_lock)) {
+			if (last_notice != -1 && brl_ctdb_overlap(&locks[i], &locks[last_notice])) {
+				continue;
+			}
+			if (locks[i].lock_type == PENDING_WRITE_LOCK) {
+				last_notice = i;
+			}
+			messaging_send_ptr(brl->messaging_ctx, locks[i].context.server, 
+					   MSG_BRL_RETRY, locks[i].notify_ptr);
+		}
+	}
+}
+#endif
+
+/*
+  send notifications for all pending locks - the file is being closed by this
+  user
+*/
+static void brl_ctdb_notify_all(struct brl_context *brl,
+			   struct lock_struct *locks, int count)
+{
+	int i;
+	for (i=0;i<count;i++) {
+		if (locks->lock_type >= PENDING_READ_LOCK) {
+//			brl_ctdb_notify_unlock(brl, locks, count, &locks[i]);
+		}
+	}
+}
+
+struct ctdb_unlock_req {
+	uint16_t smbpid;
+	uint64_t start;
+	uint64_t size;
+	struct server_id server;
+	struct brl_context *brl;
+	struct ntvfs_handle *ntvfs;
+};
+
+/*
+ Unlock a range of bytes.
+*/
+static int brl_ctdb_unlock_func(struct ctdb_call *call)
+{
+	struct ctdb_unlock_req *req = (struct ctdb_unlock_req *)call->call_data->dptr;
+	TDB_DATA dbuf;
+	int count, i;
+	struct lock_struct *locks;
+	struct lock_context context;
+	NTSTATUS status = NT_STATUS_OK;
+
+	dbuf = call->record_data;
+
+	context.smbpid = req->smbpid;
+	context.server = req->server;
+	context.ctx = req->brl;
+
+	/* there are existing locks - find a match */
+	locks = (struct lock_struct *)dbuf.dptr;
+	count = dbuf.dsize / sizeof(*locks);
+
+	show_locks("unlock", locks, count);
+
+	for (i=0; i<count; i++) {
+		struct lock_struct *lock = &locks[i];
+		
+		if (brl_ctdb_same_context(&lock->context, &context) &&
+		    lock->ntvfs == req->ntvfs &&
+		    lock->start == req->start &&
+		    lock->size == req->size &&
+		    lock->lock_type < PENDING_READ_LOCK) {
+//			struct lock_struct removed_lock = *lock;
+
+			call->new_data = talloc(call, TDB_DATA);
+			if (call->new_data == NULL) {
+				return CTDB_ERR_NOMEM;
+			}
+
+			call->new_data->dptr = talloc_size(call, dbuf.dsize - sizeof(lock));
+			if (call->new_data->dptr == NULL) {
+				return CTDB_ERR_NOMEM;
+			}
+			call->new_data->dsize = dbuf.dsize - sizeof(lock);
+
+			memcpy(call->new_data->dptr, locks, i*sizeof(lock));
+			memcpy(call->new_data->dptr+i*sizeof(lock), locks+i+1,
+			       (count-(i+1))*sizeof(lock));
+			
+			if (count > 1) {
+				/* send notifications for any relevant pending locks */
+//				brl_ctdb_notify_unlock(req->brl, locks, count, &removed_lock);
+			}
+			break;
+		}
+	}
+
+	if (call->new_data) {
+		DEBUG(0,("unlock: size now %d\n", call->new_data->dsize));
+	}
+	
+	if (i == count) {
+		/* we didn't find it */
+		status = NT_STATUS_RANGE_NOT_LOCKED;
+	}
+
+	call->reply_data = talloc(call, TDB_DATA);
+	if (call->reply_data == NULL) {
+		return CTDB_ERR_NOMEM;
+	}
+
+	call->reply_data->dptr = talloc_size(call, sizeof(NTSTATUS));
+	call->reply_data->dsize = sizeof(NTSTATUS);
+	if (call->reply_data->dptr == NULL) {
+		return CTDB_ERR_NOMEM;
+	}
+	*(NTSTATUS *)call->reply_data->dptr = status;
+
+	return 0;
+}
+
+
+/*
+ Unlock a range of bytes.
+*/
+static NTSTATUS brl_ctdb_unlock(struct brl_context *brl,
+				struct brl_handle *brlh, 
+				uint16_t smbpid,
+				uint64_t start, uint64_t size)
+{
+	TDB_DATA kbuf, rbuf, sbuf;
+	struct ctdb_unlock_req req;
+	NTSTATUS status;
+	int ret;
+
+	kbuf.dptr = brlh->key.data;
+	kbuf.dsize = brlh->key.length;
+
+	rbuf.dptr = (uint8_t *)&req;
+	rbuf.dsize = sizeof(req);
+
+	ZERO_STRUCT(req);
+	req.smbpid = smbpid;
+	req.start  = start;
+	req.size   = size;
+	req.server = brl->server;
+	req.brl = brl;
+	req.ntvfs = brlh->ntvfs;
+		
+	ret = ctdb_call(brl->ctdb, kbuf, FUNC_BRL_UNLOCK, &rbuf, &sbuf);
+	if (ret == -1) {
+		DEBUG(0,("ctdb_call failed - %s\n", __location__));
+		return NT_STATUS_INTERNAL_DB_CORRUPTION;
+	}
+
+	status = *(NTSTATUS *)sbuf.dptr;
+	talloc_free(sbuf.dptr);
+
+	return status;
+}
+
+
+struct ctdb_remove_pending_req {
+	struct server_id server;
+	void *notify_ptr;
+};
+
+/*
+  remove a pending lock. This is called when the caller has either
+  given up trying to establish a lock or when they have succeeded in
+  getting it. In either case they no longer need to be notified.
+*/
+static int brl_ctdb_remove_pending_func(struct ctdb_call *call)
+{
+	struct ctdb_remove_pending_req *req = (struct ctdb_remove_pending_req *)call->call_data->dptr;
+	TDB_DATA dbuf;
+	int count, i;
+	struct lock_struct *locks;
+	NTSTATUS status = NT_STATUS_OK;
+
+	dbuf = call->record_data;
+
+	/* there are existing locks - find a match */
+	locks = (struct lock_struct *)dbuf.dptr;
+	count = dbuf.dsize / sizeof(*locks);
+
+	show_locks("remove_pending", locks, count);
+
+	for (i=0; i<count; i++) {
+		struct lock_struct *lock = &locks[i];
+		
+		if (lock->lock_type >= PENDING_READ_LOCK &&
+		    lock->notify_ptr == req->notify_ptr &&
+		    cluster_id_equal(&lock->context.server, &req->server)) {
+			call->new_data = talloc(call, TDB_DATA);
+			if (call->new_data == NULL) {
+				return CTDB_ERR_NOMEM;
+			}
+
+			call->new_data->dptr = talloc_size(call, dbuf.dsize - sizeof(lock));
+			if (call->new_data->dptr == NULL) {
+				return CTDB_ERR_NOMEM;
+			}
+			call->new_data->dsize = dbuf.dsize - sizeof(lock);
+
+			memcpy(call->new_data->dptr, locks, i*sizeof(lock));
+			memcpy(call->new_data->dptr+i*sizeof(lock), locks+i+1,
+			       (count-(i+1))*sizeof(lock));
+			break;
+		}
+	}
+	
+	if (call->new_data) {
+		DEBUG(0,("remove_pending: size now %d\n", call->new_data->dsize));
+	}
+
+	if (i == count) {
+		/* we didn't find it */
+		status = NT_STATUS_RANGE_NOT_LOCKED;
+	}
+
+	call->reply_data = talloc(call, TDB_DATA);
+	if (call->reply_data == NULL) {
+		return CTDB_ERR_NOMEM;
+	}
+
+	call->reply_data->dptr = talloc_size(call, sizeof(NTSTATUS));
+	call->reply_data->dsize = sizeof(NTSTATUS);
+	if (call->reply_data->dptr == NULL) {
+		return CTDB_ERR_NOMEM;
+	}
+	*(NTSTATUS *)call->reply_data->dptr = status;
+
+	return 0;
+}
+
+static NTSTATUS brl_ctdb_remove_pending(struct brl_context *brl,
+					struct brl_handle *brlh, 
+					void *notify_ptr)
+{
+	TDB_DATA kbuf, rbuf, sbuf;
+	struct ctdb_remove_pending_req req;
+	NTSTATUS status;
+	int ret;
+
+	kbuf.dptr = brlh->key.data;
+	kbuf.dsize = brlh->key.length;
+
+	rbuf.dptr = (uint8_t *)&req;
+	rbuf.dsize = sizeof(req);
+
+	ZERO_STRUCT(req);
+	req.notify_ptr = notify_ptr;
+	req.server = brl->server;
+		
+	ret = ctdb_call(brl->ctdb, kbuf, FUNC_BRL_REMOVE_PENDING, &rbuf, &sbuf);
+	if (ret == -1) {
+		DEBUG(0,("ctdb_call failed - %s\n", __location__));
+		return NT_STATUS_INTERNAL_DB_CORRUPTION;
+	}
+
+	status = *(NTSTATUS *)sbuf.dptr;
+	talloc_free(sbuf.dptr);
+
+	return status;
+}
+
+
+struct ctdb_locktest_req {
+	uint16_t smbpid;
+	uint64_t start;
+	uint64_t size;
+	enum brl_type lock_type;
+	struct brl_context *brl;
+	struct server_id server;
+	struct ntvfs_handle *ntvfs;
+};
+
+/*
+  remove a pending lock. This is called when the caller has either
+  given up trying to establish a lock or when they have succeeded in
+  getting it. In either case they no longer need to be notified.
+*/
+static int brl_ctdb_locktest_func(struct ctdb_call *call)
+{
+	struct ctdb_locktest_req *req = (struct ctdb_locktest_req *)call->call_data->dptr;
+	TDB_DATA dbuf;
+	int count, i;
+	struct lock_struct *locks, lock;
+	NTSTATUS status = NT_STATUS_OK;
+
+	lock.context.smbpid = req->smbpid;
+	lock.context.server = req->server;
+	lock.context.ctx = req->brl;
+	lock.ntvfs = req->ntvfs;
+	lock.start = req->start;
+	lock.size = req->size;
+	lock.lock_type = req->lock_type;
+
+	dbuf = call->record_data;
+
+	/* there are existing locks - find a match */
+	locks = (struct lock_struct *)dbuf.dptr;
+	count = dbuf.dsize / sizeof(*locks);
+
+	show_locks("locktest", locks, count);
+
+	for (i=0; i<count; i++) {
+		if (brl_ctdb_conflict_other(&locks[i], &lock)) {
+			status = NT_STATUS_FILE_LOCK_CONFLICT;
+			break;
+		}
+	}
+	
+	call->reply_data = talloc(call, TDB_DATA);
+	if (call->reply_data == NULL) {
+		return CTDB_ERR_NOMEM;
+	}
+
+	call->reply_data->dptr = talloc_size(call, sizeof(NTSTATUS));
+	call->reply_data->dsize = sizeof(NTSTATUS);
+	if (call->reply_data->dptr == NULL) {
+		return CTDB_ERR_NOMEM;
+	}
+	*(NTSTATUS *)call->reply_data->dptr = status;
+
+	return 0;
+}
+
+/*
+  Test if we are allowed to perform IO on a region of an open file
+*/
+static NTSTATUS brl_ctdb_locktest(struct brl_context *brl,
+				  struct brl_handle *brlh,
+				  uint16_t smbpid, 
+				  uint64_t start, uint64_t size, 
+				  enum brl_type lock_type)
+{
+	TDB_DATA kbuf, rbuf, sbuf;
+	struct ctdb_locktest_req req;
+	NTSTATUS status;
+	int ret;
+
+	kbuf.dptr = brlh->key.data;
+	kbuf.dsize = brlh->key.length;
+
+	rbuf.dptr = (uint8_t *)&req;
+	rbuf.dsize = sizeof(req);
+
+	ZERO_STRUCT(req);
+	req.smbpid = smbpid;
+	req.start  = start;
+	req.size   = size;
+	req.lock_type = lock_type;
+	req.server = brl->server;
+	req.brl = brl;
+	req.ntvfs = brlh->ntvfs;
+
+	ret = ctdb_call(brl->ctdb, kbuf, FUNC_BRL_LOCKTEST, &rbuf, &sbuf);
+	if (ret == -1) {
+		DEBUG(0,("ctdb_call failed - %s\n", __location__));
+		return NT_STATUS_INTERNAL_DB_CORRUPTION;
+	}
+
+	status = *(NTSTATUS *)sbuf.dptr;
+	talloc_free(sbuf.dptr);
+
+	return status;
+}
+
+
+struct ctdb_close_req {
+	struct brl_context *brl;
+	struct server_id server;
+	struct ntvfs_handle *ntvfs;
+};
+
+/*
+  remove a pending lock. This is called when the caller has either
+  given up trying to establish a lock or when they have succeeded in
+  getting it. In either case they no longer need to be notified.
+*/
+static int brl_ctdb_close_func(struct ctdb_call *call)
+{
+	struct ctdb_close_req *req = (struct ctdb_close_req *)call->call_data->dptr;
+	TDB_DATA dbuf;
+	int count, dcount=0, i;
+	struct lock_struct *locks;
+	NTSTATUS status = NT_STATUS_OK;
+
+	dbuf = call->record_data;
+
+	/* there are existing locks - find a match */
+	locks = (struct lock_struct *)dbuf.dptr;
+	count = dbuf.dsize / sizeof(*locks);
+
+	show_locks("close", locks, count);
+
+	DEBUG(0,("closing ctx=%p server=%d.%d ntvfs=%p\n",
+		 req->brl, req->server.node, req->server.id, req->ntvfs));
+
+	for (i=0; i<count; i++) {
+		struct lock_struct *lock = &locks[i];
+
+		if (lock->context.ctx == req->brl &&
+		    cluster_id_equal(&lock->context.server, &req->server) &&
+		    lock->ntvfs == req->ntvfs) {
+			/* found it - delete it */
+			if (count > 1 && i < count-1) {
+				memmove(&locks[i], &locks[i+1], 
+					sizeof(*locks)*((count-1) - i));
+			}
+			count--;
+			i--;
+			dcount++;
+		}
+	}
+
+	if (dcount > 0) {
+		call->new_data = talloc(call, TDB_DATA);
+		if (call->new_data == NULL) {
+			return CTDB_ERR_NOMEM;
+		}
+		
+		call->new_data->dptr = talloc_size(call, count*sizeof(struct lock_struct));
+		if (call->new_data->dptr == NULL) {
+			return CTDB_ERR_NOMEM;
+		}
+		call->new_data->dsize = count*sizeof(struct lock_struct);
+
+		memcpy(call->new_data->dptr, locks, count*sizeof(struct lock_struct));
+	}
+
+	if (call->new_data) {
+		DEBUG(0,("close: size now %d\n", call->new_data->dsize));
+	}
+
+	DEBUG(0,("brl_ctdb_close_func dcount=%d count=%d\n", dcount, count));
+	
+	call->reply_data = talloc(call, TDB_DATA);
+	if (call->reply_data == NULL) {
+		return CTDB_ERR_NOMEM;
+	}
+
+	call->reply_data->dptr = talloc_size(call, sizeof(NTSTATUS));
+	call->reply_data->dsize = sizeof(NTSTATUS);
+	if (call->reply_data->dptr == NULL) {
+		return CTDB_ERR_NOMEM;
+	}
+	*(NTSTATUS *)call->reply_data->dptr = status;
+
+	return 0;
+}
+
+/*
+  Test if we are allowed to perform IO on a region of an open file
+*/
+static NTSTATUS brl_ctdb_close(struct brl_context *brl,
+			       struct brl_handle *brlh)
+{
+	TDB_DATA kbuf, rbuf, sbuf;
+	struct ctdb_close_req req;
+	NTSTATUS status;
+	int ret;
+
+	kbuf.dptr = brlh->key.data;
+	kbuf.dsize = brlh->key.length;
+
+	rbuf.dptr = (uint8_t *)&req;
+	rbuf.dsize = sizeof(req);
+
+	ZERO_STRUCT(req);
+	req.brl = brl;
+	req.server = brl->server;
+	req.ntvfs = brlh->ntvfs;
+
+	DEBUG(0,("brl_ctdb_close %u.%u %p\n", req.server.node, req.server.id, brl));
+
+	ret = ctdb_call(brl->ctdb, kbuf, FUNC_BRL_CLOSE, &rbuf, &sbuf);
+	if (ret == -1) {
+		DEBUG(0,("ctdb_call failed - %s\n", __location__));
+		return NT_STATUS_INTERNAL_DB_CORRUPTION;
+	}
+
+	status = *(NTSTATUS *)sbuf.dptr;
+	talloc_free(sbuf.dptr);
+
+	return status;
+}
+
+
+static const struct brlock_ops brlock_tdb_ops = {
+	.brl_init           = brl_ctdb_init,
+	.brl_create_handle  = brl_ctdb_create_handle,
+	.brl_lock           = brl_ctdb_lock,
+	.brl_unlock         = brl_ctdb_unlock,
+	.brl_remove_pending = brl_ctdb_remove_pending,
+	.brl_locktest       = brl_ctdb_locktest,
+	.brl_close          = brl_ctdb_close
+};
+
+
+void brl_ctdb_init_ops(void)
+{
+	struct ctdb_context *ctdb = talloc_get_type(cluster_private(), struct ctdb_context);
+
+	brl_set_ops(&brlock_tdb_ops);
+
+	ctdb_set_call(ctdb, brl_ctdb_lock_func,  FUNC_BRL_LOCK);
+	ctdb_set_call(ctdb, brl_ctdb_unlock_func,  FUNC_BRL_UNLOCK);
+	ctdb_set_call(ctdb, brl_ctdb_remove_pending_func,  FUNC_BRL_REMOVE_PENDING);
+	ctdb_set_call(ctdb, brl_ctdb_locktest_func,  FUNC_BRL_LOCKTEST);
+	ctdb_set_call(ctdb, brl_ctdb_close_func,  FUNC_BRL_CLOSE);
+
+}
diff --git a/source4/cluster/ctdb/common/ctdb.c b/source4/cluster/ctdb/common/ctdb.c
new file mode 100644
index 0000000000..ad0345b3c7
--- /dev/null
+++ b/source4/cluster/ctdb/common/ctdb.c
@@ -0,0 +1,287 @@
+/* 
+   ctdb main protocol code
+
+   Copyright (C) Andrew Tridgell  2006
+
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library 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
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+#include "includes.h"
+#include "lib/tdb/include/tdb.h"
+#include "lib/events/events.h"
+#include "lib/util/dlinklist.h"
+#include "system/network.h"
+#include "system/filesys.h"
+#include "cluster/ctdb/include/ctdb_private.h"
+
+/*
+  choose the transport we will use
+*/
+int ctdb_set_transport(struct ctdb_context *ctdb, const char *transport)
+{
+	int ctdb_tcp_init(struct ctdb_context *ctdb);
+
+	if (strcmp(transport, "tcp") == 0) {
+		return ctdb_tcp_init(ctdb);
+	}
+	ctdb_set_error(ctdb, "Unknown transport '%s'\n", transport);
+	return -1;
+}
+
+/*
+  set some ctdb flags
+*/
+void ctdb_set_flags(struct ctdb_context *ctdb, unsigned flags)
+{
+	ctdb->flags |= flags;
+}
+
+
+/*
+  add a node to the list of active nodes
+*/
+static int ctdb_add_node(struct ctdb_context *ctdb, char *nstr)
+{
+	struct ctdb_node *node, **nodep;
+
+	nodep = talloc_realloc(ctdb, ctdb->nodes, struct ctdb_node *, ctdb->num_nodes+1);
+	CTDB_NO_MEMORY(ctdb, nodep);
+
+	ctdb->nodes = nodep;
+	nodep = &ctdb->nodes[ctdb->num_nodes];
+	(*nodep) = talloc_zero(ctdb->nodes, struct ctdb_node);
+	CTDB_NO_MEMORY(ctdb, *nodep);
+	node = *nodep;
+
+	if (ctdb_parse_address(ctdb, node, nstr, &node->address) != 0) {
+		return -1;
+	}
+	node->ctdb = ctdb;
+	node->name = talloc_asprintf(node, "%s:%u", 
+				     node->address.address, 
+				     node->address.port);
+	/* for now we just set the vnn to the line in the file - this
+	   will change! */
+	node->vnn = ctdb->num_nodes;
+
+	if (ctdb->methods->add_node(node) != 0) {
+		talloc_free(node);
+		return -1;
+	}
+
+	if (ctdb_same_address(&ctdb->address, &node->address)) {
+		ctdb->vnn = node->vnn;
+	}
+
+	ctdb->num_nodes++;
+
+	return 0;
+}
+
+/*
+  setup the node list from a file
+*/
+int ctdb_set_nlist(struct ctdb_context *ctdb, const char *nlist)
+{
+	char **lines;
+	int nlines;
+	int i;
+
+	lines = file_lines_load(nlist, &nlines, ctdb);
+	if (lines == NULL) {
+		ctdb_set_error(ctdb, "Failed to load nlist '%s'\n", nlist);
+		return -1;
+	}
+
+	for (i=0;i<nlines;i++) {
+		if (ctdb_add_node(ctdb, lines[i]) != 0) {
+			talloc_free(lines);
+			return -1;
+		}
+	}
+	
+	talloc_free(lines);
+	return 0;
+}
+
+/*
+  setup the local node address
+*/
+int ctdb_set_address(struct ctdb_context *ctdb, const char *address)
+{
+	if (ctdb_parse_address(ctdb, ctdb, address, &ctdb->address) != 0) {
+		return -1;
+	}
+	
+	ctdb->name = talloc_asprintf(ctdb, "%s:%u", 
+				     ctdb->address.address, 
+				     ctdb->address.port);
+	return 0;
+}
+
+/*
+  add a node to the list of active nodes
+*/
+int ctdb_set_call(struct ctdb_context *ctdb, ctdb_fn_t fn, int id)
+{
+	struct ctdb_registered_call *call;
+
+	call = talloc(ctdb, struct ctdb_registered_call);
+	call->fn = fn;
+	call->id = id;
+
+	DLIST_ADD(ctdb->calls, call);	
+	return 0;
+}
+
+/*
+  return the vnn of this node
+*/
+uint32_t ctdb_get_vnn(struct ctdb_context *ctdb)
+{
+	return ctdb->vnn;
+}
+
+/*
+  start the protocol going
+*/
+int ctdb_start(struct ctdb_context *ctdb)
+{
+	return ctdb->methods->start(ctdb);
+}
+
+/*
+  called by the transport layer when a packet comes in
+*/
+static void ctdb_recv_pkt(struct ctdb_context *ctdb, uint8_t *data, uint32_t length)
+{
+	struct ctdb_req_header *hdr;
+	if (length < sizeof(*hdr)) {
+		ctdb_set_error(ctdb, "Bad packet length %d\n", length);
+		return;
+	}
+	hdr = (struct ctdb_req_header *)data;
+	if (length != hdr->length) {
+		ctdb_set_error(ctdb, "Bad header length %d expected %d\n", 
+			       hdr->length, length);
+		return;
+	}
+
+	DEBUG(0,("got ctdb op %d reqid %d\n", hdr->operation, hdr->reqid));
+
+	switch (hdr->operation) {
+	case CTDB_REQ_CALL:
+		ctdb_request_call(ctdb, hdr);
+		break;
+
+	case CTDB_REPLY_CALL:
+		ctdb_reply_call(ctdb, hdr);
+		break;
+
+	case CTDB_REPLY_ERROR:
+		ctdb_reply_error(ctdb, hdr);
+		break;
+
+	case CTDB_REPLY_REDIRECT:
+		ctdb_reply_redirect(ctdb, hdr);
+		break;
+
+	case CTDB_REQ_DMASTER:
+		ctdb_request_dmaster(ctdb, hdr);
+		break;
+
+	case CTDB_REPLY_DMASTER:
+		ctdb_reply_dmaster(ctdb, hdr);
+		break;
+
+	default:
+		printf("Packet with unknown operation %d\n", hdr->operation);
+		talloc_free(hdr);
+		break;
+	}
+}
+
+/*
+  called by the transport layer when a node is dead
+*/
+static void ctdb_node_dead(struct ctdb_node *node)
+{
+	node->ctdb->num_connected--;
+	printf("%s: node %s is dead: %d connected\n", 
+	       node->ctdb->name, node->name, node->ctdb->num_connected);
+}
+
+/*
+  called by the transport layer when a node is dead
+*/
+static void ctdb_node_connected(struct ctdb_node *node)
+{
+	node->ctdb->num_connected++;
+	printf("%s: connected to %s - %d connected\n", 
+	       node->ctdb->name, node->name, node->ctdb->num_connected);
+}
+
+/*
+  wait for all nodes to be connected
+*/
+void ctdb_connect_wait(struct ctdb_context *ctdb)
+{
+	int expected = ctdb->num_nodes - 1;
+	if (ctdb->flags & CTDB_FLAG_SELF_CONNECT) {
+		expected++;
+	}
+	while (ctdb->num_connected != expected) {
+		event_loop_once(ctdb->ev);
+	}
+}
+
+/*
+  wait until we're the only node left
+*/
+void ctdb_wait_loop(struct ctdb_context *ctdb)
+{
+	int expected = 0;
+	if (ctdb->flags & CTDB_FLAG_SELF_CONNECT) {
+		expected++;
+	}
+	while (ctdb->num_connected > expected) {
+		event_loop_once(ctdb->ev);
+	}
+}
+
+static const struct ctdb_upcalls ctdb_upcalls = {
+	.recv_pkt       = ctdb_recv_pkt,
+	.node_dead      = ctdb_node_dead,
+	.node_connected = ctdb_node_connected
+};
+
+/*
+  initialise the ctdb daemon. 
+
+  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 event_context *ev)
+{
+	struct ctdb_context *ctdb;
+
+	ctdb = talloc_zero(ev, struct ctdb_context);
+	ctdb->ev = ev;
+	ctdb->upcalls = &ctdb_upcalls;
+	ctdb->idr = idr_init(ctdb);
+
+	return ctdb;
+}
+
diff --git a/source4/cluster/ctdb/common/ctdb_call.c b/source4/cluster/ctdb/common/ctdb_call.c
new file mode 100644
index 0000000000..2bedccc86a
--- /dev/null
+++ b/source4/cluster/ctdb/common/ctdb_call.c
@@ -0,0 +1,653 @@
+/* 
+   ctdb_call protocol code
+
+   Copyright (C) Andrew Tridgell  2006
+
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library 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
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+/*
+  see http://wiki.samba.org/index.php/Samba_%26_Clustering for
+  protocol design and packet details
+*/
+#include "includes.h"
+#include "lib/events/events.h"
+#include "lib/tdb/include/tdb.h"
+#include "system/network.h"
+#include "system/filesys.h"
+#include "cluster/ctdb/include/ctdb_private.h"
+
+
+/*
+  queue a packet or die
+*/
+static void ctdb_queue_packet(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
+{
+	struct ctdb_node *node;
+	DEBUG(0,("queueing destnode=%u srcnode=%u\n", hdr->destnode, hdr->srcnode));
+	node = ctdb->nodes[hdr->destnode];
+	if (ctdb->methods->queue_pkt(node, (uint8_t *)hdr, hdr->length) != 0) {
+		ctdb_fatal(ctdb, "Unable to queue packet\n");
+	}
+}
+
+
+/*
+  local version of ctdb_call
+*/
+static int ctdb_call_local(struct ctdb_context *ctdb, TDB_DATA key, 
+			   struct ctdb_ltdb_header *header, TDB_DATA *data,
+			   int call_id, TDB_DATA *call_data, TDB_DATA *reply_data,
+			   uint32_t caller)
+{
+	struct ctdb_call *c;
+	struct ctdb_registered_call *fn;
+
+	c = talloc(ctdb, struct ctdb_call);
+	CTDB_NO_MEMORY(ctdb, c);
+
+	c->key = key;
+	c->call_data = 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->new_data = NULL;
+	c->reply_data = NULL;
+
+	for (fn=ctdb->calls;fn;fn=fn->next) {
+		if (fn->id == call_id) break;
+	}
+	if (fn == NULL) {
+		ctdb_set_error(ctdb, "Unknown call id %u\n", call_id);
+		return -1;
+	}
+
+	if (fn->fn(c) != 0) {
+		ctdb_set_error(ctdb, "ctdb_call %u failed\n", call_id);
+		return -1;
+	}
+
+	if (header->laccessor != caller) {
+		header->lacount = 0;
+	}
+	header->laccessor = caller;
+	header->lacount++;
+
+	/* we need to force the record to be written out if this was a remote access,
+	   so that the lacount is updated */
+	if (c->new_data == NULL && header->laccessor != ctdb->vnn) {
+		c->new_data = &c->record_data;
+	}
+
+	if (c->new_data) {
+		if (ctdb_ltdb_store(ctdb, key, header, *c->new_data) != 0) {
+			ctdb_set_error(ctdb, "ctdb_call tdb_store failed\n");
+			return -1;
+		}
+	}
+
+	if (reply_data) {
+		if (c->reply_data) {
+			*reply_data = *c->reply_data;
+			talloc_steal(ctdb, reply_data->dptr);
+		} else {
+			reply_data->dptr = NULL;
+			reply_data->dsize = 0;
+		}
+	}
+
+	talloc_free(c);
+
+	return 0;
+}
+
+/*
+  send an error reply
+*/
+static void ctdb_send_error(struct ctdb_context *ctdb, 
+			    struct ctdb_req_header *hdr, uint32_t status,
+			    const char *fmt, ...)
+{
+	va_list ap;
+	struct ctdb_reply_error *r;
+	char *msg;
+	int len;
+
+	va_start(ap, fmt);
+	msg = talloc_vasprintf(ctdb, fmt, ap);
+	if (msg == NULL) {
+		ctdb_fatal(ctdb, "Unable to allocate error in ctdb_send_error\n");
+	}
+	va_end(ap);
+
+	len = strlen(msg)+1;
+	r = ctdb->methods->allocate_pkt(ctdb, sizeof(*r) + len);
+	CTDB_NO_MEMORY_FATAL(ctdb, r);
+	r->hdr.length = sizeof(*r) + len;
+	r->hdr.operation = CTDB_REPLY_ERROR;
+	r->hdr.destnode  = hdr->srcnode;
+	r->hdr.srcnode   = ctdb->vnn;
+	r->hdr.reqid     = hdr->reqid;
+	r->status        = status;
+	r->msglen        = len;
+	memcpy(&r->msg[0], msg, len);
+
+	talloc_free(msg);
+
+	ctdb_queue_packet(ctdb, &r->hdr);
+
+	talloc_free(r);
+}
+
+
+/*
+  send a redirect reply
+*/
+static void ctdb_call_send_redirect(struct ctdb_context *ctdb, 
+				    struct ctdb_req_call *c, 
+				    struct ctdb_ltdb_header *header)
+{
+	struct ctdb_reply_redirect *r;
+
+	r = ctdb->methods->allocate_pkt(ctdb, sizeof(*r));
+	CTDB_NO_MEMORY_FATAL(ctdb, r);
+	r->hdr.length = sizeof(*r);
+	r->hdr.operation = CTDB_REPLY_REDIRECT;
+	r->hdr.destnode  = c->hdr.srcnode;
+	r->hdr.srcnode   = ctdb->vnn;
+	r->hdr.reqid     = c->hdr.reqid;
+	r->dmaster       = header->dmaster;
+
+	ctdb_queue_packet(ctdb, &r->hdr);
+
+	talloc_free(r);
+}
+
+/*
+  send a dmaster request (give another node the dmaster for a record)
+
+  This is always sent to the lmaster, which ensures that the lmaster
+  always knows who the dmaster is. The lmaster will then send a
+  CTDB_REPLY_DMASTER to the new dmaster
+*/
+static void ctdb_call_send_dmaster(struct ctdb_context *ctdb, 
+				   struct ctdb_req_call *c, 
+				   struct ctdb_ltdb_header *header,
+				   TDB_DATA *key, TDB_DATA *data)
+{
+	struct ctdb_req_dmaster *r;
+	int len;
+	
+	len = sizeof(*r) + key->dsize + data->dsize;
+	r = ctdb->methods->allocate_pkt(ctdb, len);
+	CTDB_NO_MEMORY_FATAL(ctdb, r);
+	r->hdr.length    = len;
+	r->hdr.operation = CTDB_REQ_DMASTER;
+	r->hdr.destnode  = ctdb_lmaster(ctdb, key);
+	r->hdr.srcnode   = ctdb->vnn;
+	r->hdr.reqid     = c->hdr.reqid;
+	r->dmaster       = header->laccessor;
+	r->keylen        = key->dsize;
+	r->datalen       = data->dsize;
+	memcpy(&r->data[0], key->dptr, key->dsize);
+	memcpy(&r->data[key->dsize], data->dptr, data->dsize);
+
+	if (r->hdr.destnode == ctdb->vnn && !(ctdb->flags & CTDB_FLAG_SELF_CONNECT)) {
+		/* we are the lmaster - don't send to ourselves */
+		DEBUG(0,("XXXX local ctdb_req_dmaster\n"));
+		ctdb_request_dmaster(ctdb, &r->hdr);
+	} else {
+		ctdb_queue_packet(ctdb, &r->hdr);
+
+		/* update the ltdb to record the new dmaster */
+		header->dmaster = r->hdr.destnode;
+		ctdb_ltdb_store(ctdb, *key, header, *data);
+	}
+
+	talloc_free(r);
+}
+
+
+/*
+  called when a CTDB_REQ_DMASTER packet comes in
+
+  this comes into the lmaster for a record when the current dmaster
+  wants to give up the dmaster role and give it to someone else
+*/
+void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
+{
+	struct ctdb_req_dmaster *c = (struct ctdb_req_dmaster *)hdr;
+	struct ctdb_reply_dmaster *r;
+	TDB_DATA key, data;
+	struct ctdb_ltdb_header header;
+	int ret;
+
+	key.dptr = c->data;
+	key.dsize = c->keylen;
+	data.dptr = c->data + c->keylen;
+	data.dsize = c->datalen;
+
+	DEBUG(0,("request dmaster reqid=%d\n", hdr->reqid));
+
+	/* fetch the current record */
+	ret = ctdb_ltdb_fetch(ctdb, key, &header, &data);
+	if (ret != 0) {
+		ctdb_fatal(ctdb, "ctdb_req_dmaster failed to fetch record");
+		return;
+	}
+
+	{
+		int i, fd = open("/dev/null", O_WRONLY);
+		for (i=0;i<data.dsize;i++) {
+			write(fd, &data.dptr[i], 1);
+		}
+		close(fd);
+	}
+
+	/* its a protocol error if the sending node is not the current dmaster */
+	if (header.dmaster != hdr->srcnode) {
+		ctdb_fatal(ctdb, "dmaster request from non-master");
+		return;
+	}
+
+	DEBUG(0,("request dmaster reqid=%d %s\n", hdr->reqid, __location__));
+
+	header.dmaster = c->dmaster;
+	if (ctdb_ltdb_store(ctdb, key, &header, data) != 0) {
+		ctdb_fatal(ctdb, "ctdb_req_dmaster unable to update dmaster");
+		return;
+	}
+
+	{
+		int i, fd = open("/dev/null", O_WRONLY);
+		for (i=0;i<data.dsize;i++) {
+			write(fd, &data.dptr[i], 1);
+		}
+		close(fd);
+	}
+
+	/* send the CTDB_REPLY_DMASTER */
+	r = ctdb->methods->allocate_pkt(ctdb, sizeof(*r) + data.dsize);
+	CTDB_NO_MEMORY_FATAL(ctdb, r);
+	r->hdr.length = sizeof(*r) + data.dsize;
+	r->hdr.operation = CTDB_REPLY_DMASTER;
+	r->hdr.destnode  = c->dmaster;
+	r->hdr.srcnode   = ctdb->vnn;
+	r->hdr.reqid     = hdr->reqid;
+	r->datalen       = data.dsize;
+	memcpy(&r->data[0], data.dptr, data.dsize);
+
+	{
+		int i, fd = open("/dev/null", O_WRONLY);
+		for (i=0;i<data.dsize;i++) {
+			write(fd, &data.dptr[i], 1);
+		}
+		close(fd);
+	}
+
+	DEBUG(0,("request dmaster reqid=%d %s\n", hdr->reqid, __location__));
+
+	if (0 && r->hdr.destnode == r->hdr.srcnode) {
+		ctdb_reply_dmaster(ctdb, &r->hdr);
+	} else {
+		ctdb_queue_packet(ctdb, &r->hdr);
+		DEBUG(0,("request dmaster reqid=%d %s\n", hdr->reqid, __location__));
+
+		talloc_free(r);
+	}
+}
+
+
+/*
+  called when a CTDB_REQ_CALL packet comes in
+*/
+void ctdb_request_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
+{
+	struct ctdb_req_call *c = (struct ctdb_req_call *)hdr;
+	TDB_DATA key, data, call_data, reply_data;
+	struct ctdb_reply_call *r;
+	int ret;
+	struct ctdb_ltdb_header header;
+
+	key.dptr = c->data;
+	key.dsize = c->keylen;
+	call_data.dptr = c->data + c->keylen;
+	call_data.dsize = c->calldatalen;
+
+	/* determine if we are the dmaster for this key. This also
+	   fetches the record data (if any), thus avoiding a 2nd fetch of the data 
+	   if the call will be answered locally */
+	ret = ctdb_ltdb_fetch(ctdb, key, &header, &data);
+	if (ret != 0) {
+		ctdb_send_error(ctdb, hdr, ret, "ltdb fetch failed in ctdb_request_call");
+		return;
+	}
+
+	/* if we are not the dmaster, then send a redirect to the
+	   requesting node */
+	if (header.dmaster != ctdb->vnn) {
+		ctdb_call_send_redirect(ctdb, c, &header);
+		talloc_free(data.dptr);
+		return;
+	}
+
+	/* if this nodes has done enough consecutive calls on the same record
+	   then give them the record */
+	if (header.laccessor == c->hdr.srcnode &&
+	    header.lacount >= CTDB_MAX_LACOUNT) {
+		ctdb_call_send_dmaster(ctdb, c, &header, &key, &data);
+		talloc_free(data.dptr);
+		return;
+	}
+
+	ctdb_call_local(ctdb, key, &header, &data, c->callid, 
+			call_data.dsize?&call_data:NULL,
+			&reply_data, c->hdr.srcnode);
+
+	r = ctdb->methods->allocate_pkt(ctdb, sizeof(*r) + reply_data.dsize);
+	CTDB_NO_MEMORY_FATAL(ctdb, r);
+	r->hdr.length = sizeof(*r) + reply_data.dsize;
+	r->hdr.operation = CTDB_REPLY_CALL;
+	r->hdr.destnode  = hdr->srcnode;
+	r->hdr.srcnode   = hdr->destnode;
+	r->hdr.reqid     = hdr->reqid;
+	r->datalen       = reply_data.dsize;
+	memcpy(&r->data[0], reply_data.dptr, reply_data.dsize);
+
+	ctdb_queue_packet(ctdb, &r->hdr);
+
+	talloc_free(reply_data.dptr);
+	talloc_free(r);
+}
+
+enum call_state {CTDB_CALL_WAIT, CTDB_CALL_DONE, CTDB_CALL_ERROR};
+
+/*
+  state of a in-progress ctdb call
+*/
+struct ctdb_call_state {
+	enum call_state state;
+	struct ctdb_req_call *c;
+	struct ctdb_node *node;
+	const char *errmsg;
+	TDB_DATA call_data;
+	TDB_DATA reply_data;
+	TDB_DATA key;
+	int redirect_count;
+	struct ctdb_ltdb_header header;
+};
+
+
+/*
+  called when a CTDB_REPLY_CALL packet comes in
+
+  This packet comes in response to a CTDB_REQ_CALL request packet. It
+  contains any reply data freom the call
+*/
+void ctdb_reply_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
+{
+	struct ctdb_reply_call *c = (struct ctdb_reply_call *)hdr;
+	struct ctdb_call_state *state;
+	TDB_DATA reply_data;
+
+	state = idr_find(ctdb->idr, hdr->reqid);
+
+	reply_data.dptr = c->data;
+	reply_data.dsize = c->datalen;
+
+	state->reply_data = reply_data;
+
+	talloc_steal(state, c);
+
+	state->state = CTDB_CALL_DONE;
+}
+
+/*
+  called when a CTDB_REPLY_DMASTER packet comes in
+
+  This packet comes in from the lmaster response to a CTDB_REQ_CALL
+  request packet. It means that the current dmaster wants to give us
+  the dmaster role
+*/
+void ctdb_reply_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
+{
+	struct ctdb_reply_dmaster *c = (struct ctdb_reply_dmaster *)hdr;
+	struct ctdb_call_state *state;
+	TDB_DATA data;
+
+	state = idr_find(ctdb->idr, hdr->reqid);
+
+	data.dptr = c->data;
+	data.dsize = c->datalen;
+
+	talloc_steal(state, c);
+
+	/* we're now the dmaster - update our local ltdb with new header
+	   and data */
+	state->header.dmaster = ctdb->vnn;
+
+	if (ctdb_ltdb_store(ctdb, state->key, &state->header, data) != 0) {
+		ctdb_fatal(ctdb, "ctdb_reply_dmaster store failed\n");
+		return;
+	}
+
+	ctdb_call_local(ctdb, state->key, &state->header, &data, state->c->callid,
+			state->call_data.dsize?&state->call_data:NULL,
+			&state->reply_data, ctdb->vnn);
+
+	state->state = CTDB_CALL_DONE;
+}
+
+
+/*
+  called when a CTDB_REPLY_ERROR packet comes in
+*/
+void ctdb_reply_error(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
+{
+	struct ctdb_reply_error *c = (struct ctdb_reply_error *)hdr;
+	struct ctdb_call_state *state;
+
+	state = idr_find(ctdb->idr, hdr->reqid);
+
+	talloc_steal(state, c);
+
+	state->state  = CTDB_CALL_ERROR;
+	state->errmsg = (char *)c->msg;
+}
+
+
+/*
+  called when a CTDB_REPLY_REDIRECT packet comes in
+
+  This packet arrives when we have sent a CTDB_REQ_CALL request and
+  the node that received it is not the dmaster for the given key. We
+  are given a hint as to what node to try next.
+*/
+void ctdb_reply_redirect(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
+{
+	struct ctdb_reply_redirect *c = (struct ctdb_reply_redirect *)hdr;
+	struct ctdb_call_state *state;
+
+	state = idr_find(ctdb->idr, hdr->reqid);
+
+	talloc_steal(state, c);
+	
+	/* don't allow for too many redirects */
+	if (state->redirect_count++ == CTDB_MAX_REDIRECT) {
+		c->dmaster = ctdb_lmaster(ctdb, &state->key);
+	}
+
+	/* send it off again */
+	state->node = ctdb->nodes[c->dmaster];
+
+	ctdb_queue_packet(ctdb, &state->c->hdr);
+}
+
+/*
+  destroy a ctdb_call
+*/
+static int ctdb_call_destructor(struct ctdb_call_state *state)
+{
+//	idr_remove(state->node->ctdb->idr, state->c->hdr.reqid);
+	return 0;
+}
+
+
+/*
+  called when a ctdb_call times out
+*/
+void ctdb_call_timeout(struct event_context *ev, struct timed_event *te, 
+		       struct timeval t, void *private)
+{
+	struct ctdb_call_state *state = talloc_get_type(private, struct ctdb_call_state);
+	state->state = CTDB_CALL_ERROR;
+	ctdb_set_error(state->node->ctdb, "ctdb_call %u timed out",
+		       state->c->hdr.reqid);
+}
+
+/*
+  construct an event driven local ctdb_call
+
+  this is used so that locally processed ctdb_call requests are processed
+  in an event driven manner
+*/
+struct ctdb_call_state *ctdb_call_local_send(struct ctdb_context *ctdb, 
+					     TDB_DATA key, int call_id, 
+					     TDB_DATA *call_data, TDB_DATA *reply_data,
+					     struct ctdb_ltdb_header *header,
+					     TDB_DATA *data)
+{
+	struct ctdb_call_state *state;
+	int ret;
+
+	state = talloc_zero(ctdb, struct ctdb_call_state);
+	CTDB_NO_MEMORY_NULL(ctdb, state);
+
+	state->state = CTDB_CALL_DONE;
+	state->node = ctdb->nodes[ctdb->vnn];
+
+	ret = ctdb_call_local(ctdb, key, header, data, 
+			      call_id, call_data, &state->reply_data, 
+			      ctdb->vnn);
+	return state;
+}
+
+
+/*
+  make a remote ctdb call - async send
+
+  This constructs a ctdb_call request and queues it for processing. 
+  This call never blocks.
+*/
+struct ctdb_call_state *ctdb_call_send(struct ctdb_context *ctdb, 
+				       TDB_DATA key, int call_id, 
+				       TDB_DATA *call_data, TDB_DATA *reply_data)
+{
+	uint32_t len;
+	struct ctdb_call_state *state;
+	int ret;
+	struct ctdb_ltdb_header header;
+	TDB_DATA data;
+
+	/*
+	  if we are the dmaster for this key then we don't need to
+	  send it off at all, we can bypass the network and handle it
+	  locally. To find out if we are the dmaster we need to look
+	  in our ltdb
+	*/
+	ret = ctdb_ltdb_fetch(ctdb, key, &header, &data);
+	if (ret != 0) return NULL;
+
+	if (header.dmaster == ctdb->vnn && !(ctdb->flags & CTDB_FLAG_SELF_CONNECT)) {
+		return ctdb_call_local_send(ctdb, key, call_id, call_data, reply_data,
+					    &header, &data);
+	}
+
+	state = talloc_zero(ctdb, struct ctdb_call_state);
+	CTDB_NO_MEMORY_NULL(ctdb, state);
+
+	len = sizeof(*state->c) + key.dsize + (call_data?call_data->dsize:0);
+	state->c = ctdb->methods->allocate_pkt(ctdb, len);
+	CTDB_NO_MEMORY_NULL(ctdb, state->c);
+
+	state->c->hdr.length    = len;
+	state->c->hdr.operation = CTDB_REQ_CALL;
+	state->c->hdr.destnode  = header.dmaster;
+	state->c->hdr.srcnode   = ctdb->vnn;
+	/* this limits us to 16k outstanding messages - not unreasonable */
+	state->c->hdr.reqid     = idr_get_new(ctdb->idr, state, 0xFFFF);
+	DEBUG(0,("Allocate reqid %u\n", state->c->hdr.reqid));
+	state->c->callid        = call_id;
+	state->c->keylen        = key.dsize;
+	state->c->calldatalen   = call_data?call_data->dsize:0;
+	memcpy(&state->c->data[0], key.dptr, key.dsize);
+	if (call_data) {
+		memcpy(&state->c->data[key.dsize], call_data->dptr, call_data->dsize);
+		state->call_data.dptr = &state->c->data[key.dsize];
+		state->call_data.dsize = call_data->dsize;
+	}
+	state->key.dptr         = &state->c->data[0];
+	state->key.dsize        = key.dsize;
+
+	state->node   = ctdb->nodes[header.dmaster];
+	state->state  = CTDB_CALL_WAIT;
+	state->header = header;
+
+	talloc_set_destructor(state, ctdb_call_destructor);
+
+	ctdb_queue_packet(ctdb, &state->c->hdr);
+
+	event_add_timed(ctdb->ev, state, timeval_current_ofs(CTDB_REQ_TIMEOUT, 0), 
+			ctdb_call_timeout, state);
+	return state;
+}
+
+
+/*
+  make a remote ctdb call - async recv. 
+
+  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_call_state *state, TDB_DATA *reply_data)
+{
+	while (state->state < CTDB_CALL_DONE) {
+		event_loop_once(state->node->ctdb->ev);
+	}
+	if (state->state != CTDB_CALL_DONE) {
+		ctdb_set_error(state->node->ctdb, "%s", state->errmsg);
+		talloc_free(state);
+		return -1;
+	}
+	if (reply_data) {
+		reply_data->dptr = talloc_memdup(state->node->ctdb,
+						 state->reply_data.dptr,
+						 state->reply_data.dsize);
+		reply_data->dsize = state->reply_data.dsize;
+	}
+	talloc_free(state);
+	return 0;
+}
+
+/*
+  full ctdb_call. Equivalent to a ctdb_call_send() followed by a ctdb_call_recv()
+*/
+int ctdb_call(struct ctdb_context *ctdb, 
+	      TDB_DATA key, int call_id, 
+	      TDB_DATA *call_data, TDB_DATA *reply_data)
+{
+	struct ctdb_call_state *state;
+	state = ctdb_call_send(ctdb, key, call_id, call_data, reply_data);
+	return ctdb_call_recv(state, reply_data);
+}
diff --git a/source4/cluster/ctdb/common/ctdb_ltdb.c b/source4/cluster/ctdb/common/ctdb_ltdb.c
new file mode 100644
index 0000000000..bc15a3e898
--- /dev/null
+++ b/source4/cluster/ctdb/common/ctdb_ltdb.c
@@ -0,0 +1,139 @@
+/* 
+   ctdb ltdb code
+
+   Copyright (C) Andrew Tridgell  2006
+
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library 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
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+#include "includes.h"
+#include "lib/events/events.h"
+#include "lib/tdb/include/tdb.h"
+#include "system/network.h"
+#include "system/filesys.h"
+#include "cluster/ctdb/include/ctdb_private.h"
+
+/*
+  attach to a specific database
+*/
+int ctdb_attach(struct ctdb_context *ctdb, const char *name, int tdb_flags, 
+		int open_flags, mode_t mode)
+{
+	/* when we have a separate daemon this will need to be a real
+	   file, not a TDB_INTERNAL, so the parent can access it to
+	   for ltdb bypass */
+	ctdb->ltdb = tdb_open(name, 0, /* tdb_flags */ TDB_INTERNAL, open_flags, mode);
+	if (ctdb->ltdb == NULL) {
+		ctdb_set_error(ctdb, "Failed to open tdb %s\n", name);
+		return -1;
+	}
+	return 0;
+}
+
+/*
+  return the lmaster given a key
+*/
+uint32_t ctdb_lmaster(struct ctdb_context *ctdb, const TDB_DATA *key)
+{
+	return ctdb_hash(key) % ctdb->num_nodes;
+}
+
+
+/*
+  construct an initial header for a record with no ltdb header yet
+*/
+static void ltdb_initial_header(struct ctdb_context *ctdb, 
+				TDB_DATA key,
+				struct ctdb_ltdb_header *header)
+{
+	header->rsn = 0;
+	/* initial dmaster is the lmaster */
+	header->dmaster = ctdb_lmaster(ctdb, &key);
+	header->laccessor = header->dmaster;
+	header->lacount = 0;
+}
+
+
+/*
+  fetch a record from the ltdb, separating out the header information
+  and returning the body of the record. A valid (initial) header is
+  returned if the record is not present
+*/
+int ctdb_ltdb_fetch(struct ctdb_context *ctdb, 
+		    TDB_DATA key, struct ctdb_ltdb_header *header, TDB_DATA *data)
+{
+	TDB_DATA rec;
+
+	rec = tdb_fetch(ctdb->ltdb, key);
+	if (rec.dsize < sizeof(*header)) {
+		/* return an initial header */
+		free(rec.dptr);
+		ltdb_initial_header(ctdb, key, header);
+		data->dptr = NULL;
+		data->dsize = 0;
+		return 0;
+	}
+
+	*header = *(struct ctdb_ltdb_header *)rec.dptr;
+
+	data->dsize = rec.dsize - sizeof(struct ctdb_ltdb_header);
+	data->dptr = talloc_memdup(ctdb, sizeof(struct ctdb_ltdb_header)+rec.dptr,
+				   data->dsize);
+	free(rec.dptr);
+	CTDB_NO_MEMORY(ctdb, data->dptr);
+
+	{
+		int i, fd = open("/dev/null", O_WRONLY);
+		for (i=0;i<data->dsize;i++) {
+			write(fd, &data->dptr[i], 1);
+		}
+		close(fd);
+	}
+
+	return 0;
+}
+
+
+/*
+  fetch a record from the ltdb, separating out the header information
+  and returning the body of the record. A valid (initial) header is
+  returned if the record is not present
+*/
+int ctdb_ltdb_store(struct ctdb_context *ctdb, TDB_DATA key, 
+		    struct ctdb_ltdb_header *header, TDB_DATA data)
+{
+	TDB_DATA rec;
+	int ret;
+
+	rec.dsize = sizeof(*header) + data.dsize;
+	rec.dptr = talloc_size(ctdb, rec.dsize);
+	CTDB_NO_MEMORY(ctdb, rec.dptr);
+
+	memcpy(rec.dptr, header, sizeof(*header));
+	memcpy(rec.dptr + sizeof(*header), data.dptr, data.dsize);
+
+	{
+		int i, fd = open("/dev/null", O_WRONLY);
+		for (i=0;i<rec.dsize;i++) {
+			write(fd, &rec.dptr[i], 1);
+		}
+		close(fd);
+	}
+	
+	ret = tdb_store(ctdb->ltdb, key, rec, TDB_REPLACE);
+	talloc_free(rec.dptr);
+
+	return ret;
+}
diff --git a/source4/cluster/ctdb/common/ctdb_util.c b/source4/cluster/ctdb/common/ctdb_util.c
new file mode 100644
index 0000000000..8e25759609
--- /dev/null
+++ b/source4/cluster/ctdb/common/ctdb_util.c
@@ -0,0 +1,103 @@
+/* 
+   ctdb utility code
+
+   Copyright (C) Andrew Tridgell  2006
+
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library 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
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+#include "includes.h"
+#include "lib/events/events.h"
+#include "lib/tdb/include/tdb.h"
+#include "system/network.h"
+#include "system/filesys.h"
+#include "cluster/ctdb/include/ctdb_private.h"
+
+/*
+  return error string for last error
+*/
+const char *ctdb_errstr(struct ctdb_context *ctdb)
+{
+	return ctdb->err_msg;
+}
+
+
+/*
+  remember an error message
+*/
+void ctdb_set_error(struct ctdb_context *ctdb, const char *fmt, ...)
+{
+	va_list ap;
+	talloc_free(ctdb->err_msg);
+	va_start(ap, fmt);
+	ctdb->err_msg = talloc_vasprintf(ctdb, fmt, ap);
+	DEBUG(0,("ctdb error: %s\n", ctdb->err_msg));
+	va_end(ap);
+}
+
+
+/*
+  a fatal internal error occurred - no hope for recovery
+*/
+void ctdb_fatal(struct ctdb_context *ctdb, const char *msg)
+{
+	DEBUG(0,("ctdb fatal error: %s\n", msg));
+	fprintf(stderr, "ctdb fatal error: '%s'\n", msg);
+	abort();
+}
+
+/*
+  parse a IP:port pair
+*/
+int ctdb_parse_address(struct ctdb_context *ctdb,
+		       TALLOC_CTX *mem_ctx, const char *str,
+		       struct ctdb_address *address)
+{
+	char *p;
+	p = strchr(str, ':');
+	if (p == NULL) {
+		ctdb_set_error(ctdb, "Badly formed node '%s'\n", str);
+		return -1;
+	}
+	
+	address->address = talloc_strndup(mem_ctx, str, p-str);
+	address->port = strtoul(p+1, NULL, 0);
+	return 0;
+}
+
+
+/*
+  check if two addresses are the same
+*/
+bool ctdb_same_address(struct ctdb_address *a1, struct ctdb_address *a2)
+{
+	return strcmp(a1->address, a2->address) == 0 && a1->port == a2->port;
+}
+
+
+/*
+  hash function for mapping data to a VNN - taken from tdb
+*/
+uint32_t ctdb_hash(const TDB_DATA *key)
+{
+	uint32_t value;	/* Used to compute the hash value.  */
+	uint32_t i;	/* Used to cycle through random values. */
+
+	/* Set the initial value from the key size. */
+	for (value = 0x238F13AF * key->dsize, i=0; i < key->dsize; i++)
+		value = (value + (key->dptr[i] << (i*5 % 24)));
+
+	return (1103515243 * value + 12345);  
+}
diff --git a/source4/cluster/ctdb/config.mk b/source4/cluster/ctdb/config.mk
new file mode 100644
index 0000000000..90897eedd3
--- /dev/null
+++ b/source4/cluster/ctdb/config.mk
@@ -0,0 +1,24 @@
+##################
+[MODULE::brlock_ctdb]
+SUBSYSTEM = ntvfs_common
+OBJ_FILES = brlock_ctdb.o
+
+##################
+[MODULE::ctdb_tcp]
+SUBSYSTEM = CLUSTER
+OBJ_FILES = \
+		tcp/tcp_init.o \
+		tcp/tcp_io.o \
+		tcp/tcp_connect.o
+
+##################
+[MODULE::ctdb]
+SUBSYSTEM = CLUSTER
+OBJ_FILES = \
+		ctdb_cluster.o \
+		common/ctdb.o \
+		common/ctdb_call.o \
+		common/ctdb_ltdb.o \
+		common/ctdb_util.o
+PRIVATE_DEPENDENCIES = ctdb_tcp
+PUBLIC_DEPENDENCIES = LIBTDB LIBTALLOC
diff --git a/source4/cluster/ctdb/ctdb_cluster.c b/source4/cluster/ctdb/ctdb_cluster.c
new file mode 100644
index 0000000000..df16f2f8b5
--- /dev/null
+++ b/source4/cluster/ctdb/ctdb_cluster.c
@@ -0,0 +1,138 @@
+/* 
+   Unix SMB/CIFS implementation.
+
+   ctdb clustering hooks
+
+   Copyright (C) Andrew Tridgell 2006
+   
+   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 2 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, write to the Free Software
+   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+*/
+
+#include "includes.h"
+#include "lib/events/events.h"
+#include "cluster/cluster.h"
+#include "system/filesys.h"
+#include "cluster/cluster_private.h"
+#include "lib/tdb/include/tdb.h"
+#include "cluster/ctdb/include/ctdb.h"
+
+struct cluster_state {
+	struct ctdb_context *ctdb;
+};
+
+
+/*
+  return a server_id for a ctdb node
+*/
+static struct server_id ctdb_id(struct cluster_ops *ops, uint32_t id)
+{
+	struct ctdb_context *ctdb = ops->private;
+	struct server_id server_id;
+	server_id.node = ctdb_get_vnn(ctdb);
+	server_id.id = id;
+	return server_id;
+}
+
+
+/*
+  return a server_id as a string
+*/
+static const char *ctdb_id_string(struct cluster_ops *ops, 
+				  TALLOC_CTX *mem_ctx, struct server_id id)
+{
+	return talloc_asprintf(mem_ctx, "%u.%u", id.node, id.id);
+}
+
+static struct cluster_ops cluster_ctdb_ops = {
+	.cluster_id        = ctdb_id,
+	.cluster_id_string = ctdb_id_string,
+	.private           = NULL
+};
+
+/* initialise ctdb */
+void cluster_ctdb_init(struct event_context *ev)
+{
+	const char *nlist;
+	const char *address;
+	const char *transport;
+	struct cluster_state *state;
+	int ret;
+
+	nlist = lp_parm_string(-1, "ctdb", "nlist");
+	if (nlist == NULL) return;
+
+	address = lp_parm_string(-1, "ctdb", "address");
+	if (address == NULL) return;
+
+	transport = lp_parm_string(-1, "ctdb", "transport");
+	if (transport == NULL) {
+		transport = "tcp";
+	}
+
+	state = talloc(ev, struct cluster_state);
+	if (state == NULL) goto failed;
+
+	state->ctdb = ctdb_init(ev);
+	if (state->ctdb == NULL) goto failed;
+
+	cluster_ctdb_ops.private = state->ctdb;
+
+	ret = ctdb_set_transport(state->ctdb, transport);
+	if (ret == -1) {
+		DEBUG(0,("ctdb_set_transport failed - %s\n",
+			 ctdb_errstr(state->ctdb)));
+		goto failed;
+	}
+	
+//	ctdb_set_flags(state->ctdb, CTDB_FLAG_SELF_CONNECT);
+
+	/* tell ctdb what address to listen on */
+        ret = ctdb_set_address(state->ctdb, address);
+        if (ret == -1) {
+                DEBUG(0,("ctdb_set_address failed - %s\n", ctdb_errstr(state->ctdb)));
+		goto failed;
+        }
+
+        /* tell ctdb what nodes are available */
+        ret = ctdb_set_nlist(state->ctdb, nlist);
+        if (ret == -1) {
+                DEBUG(0,("ctdb_set_nlist failed - %s\n", ctdb_errstr(state->ctdb)));
+		goto failed;
+        }
+
+	ret = ctdb_attach(state->ctdb, "cluster.tdb", TDB_DEFAULT, O_RDWR|O_CREAT|O_TRUNC, 0666);
+	if (ret == -1) {
+		DEBUG(0,("ctdb_attach failed - %s\n", ctdb_errstr(state->ctdb)));
+		goto failed;
+	}
+
+	/* start the protocol running */
+	ret = ctdb_start(state->ctdb);
+        if (ret == -1) {
+                DEBUG(0,("ctdb_start failed - %s\n", ctdb_errstr(state->ctdb)));
+		goto failed;
+        }
+
+	/* wait until all nodes are connected (should not be needed
+	   outide of test code) */
+	ctdb_connect_wait(state->ctdb);
+
+	cluster_set_ops(&cluster_ctdb_ops);
+	return;
+	
+failed:
+	DEBUG(0,("cluster_ctdb_init failed\n"));
+	talloc_free(state);
+}
diff --git a/source4/cluster/ctdb/ctdb_cluster.h b/source4/cluster/ctdb/ctdb_cluster.h
new file mode 100644
index 0000000000..5f93df960c
--- /dev/null
+++ b/source4/cluster/ctdb/ctdb_cluster.h
@@ -0,0 +1,23 @@
+/* 
+   Unix SMB/CIFS implementation.
+
+   ctdb clustering hooks - header
+
+   Copyright (C) Andrew Tridgell 2006
+   
+   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 2 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, write to the Free Software
+   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+*/
+
+void cluster_ctdb_init(struct event_context *ev);
diff --git a/source4/cluster/ctdb/include/ctdb.h b/source4/cluster/ctdb/include/ctdb.h
new file mode 100644
index 0000000000..21b9b5d1ce
--- /dev/null
+++ b/source4/cluster/ctdb/include/ctdb.h
@@ -0,0 +1,117 @@
+/* 
+   ctdb database library
+
+   Copyright (C) Andrew Tridgell  2006
+
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library 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
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+#ifndef _CTDB_H
+#define _CTDB_H
+
+/*
+  structure passed to a ctdb call function
+*/
+struct ctdb_call {
+	TDB_DATA key;          /* record key */
+	TDB_DATA record_data;  /* current data in the record */
+	TDB_DATA *new_data;    /* optionally updated record data */
+	TDB_DATA *call_data;   /* optionally passed from caller */
+	TDB_DATA *reply_data;  /* optionally returned by function */
+};
+
+#define CTDB_ERR_INVALID 1
+#define CTDB_ERR_NOMEM 2
+
+/*
+  ctdb flags
+*/
+#define CTDB_FLAG_SELF_CONNECT (1<<0)
+
+
+struct event_context;
+
+/*
+  initialise ctdb subsystem
+*/
+struct ctdb_context *ctdb_init(struct event_context *ev);
+
+/*
+  choose the transport
+*/
+int ctdb_set_transport(struct ctdb_context *ctdb, const char *transport);
+
+/*
+  set some flags
+*/
+void ctdb_set_flags(struct ctdb_context *ctdb, unsigned flags);
+
+/*
+  tell ctdb what address to listen on, in transport specific format
+*/
+int ctdb_set_address(struct ctdb_context *ctdb, const char *address);
+
+/*
+  tell ctdb what nodes are available. This takes a filename, which will contain
+  1 node address per line, in a transport specific format
+*/
+int ctdb_set_nlist(struct ctdb_context *ctdb, const char *nlist);
+
+/*
+  start the ctdb protocol
+*/
+int ctdb_start(struct ctdb_context *ctdb);
+
+/*
+  error string for last ctdb error
+*/
+const char *ctdb_errstr(struct ctdb_context *);
+
+/* a ctdb call function */
+typedef int (*ctdb_fn_t)(struct ctdb_call *);
+
+/*
+  setup a ctdb call function
+*/
+int ctdb_set_call(struct ctdb_context *ctdb, ctdb_fn_t fn, int id);
+
+/*
+  attach to a ctdb database
+*/
+int ctdb_attach(struct ctdb_context *ctdb, const char *name, int tdb_flags, 
+		int open_flags, mode_t mode);
+
+
+/*
+  make a ctdb call. The associated ctdb call function will be called on the DMASTER
+  for the given record
+*/
+int ctdb_call(struct ctdb_context *ctdb, TDB_DATA key, int call_id, 
+	      TDB_DATA *call_data, TDB_DATA *reply_data);
+
+/*
+  wait for all nodes to be connected - useful for test code
+*/
+void ctdb_connect_wait(struct ctdb_context *ctdb);
+
+/*
+  wait until we're the only node left
+*/
+void ctdb_wait_loop(struct ctdb_context *ctdb);
+
+/* return vnn of this node */
+uint32_t ctdb_get_vnn(struct ctdb_context *ctdb);
+
+#endif
diff --git a/source4/cluster/ctdb/include/ctdb_private.h b/source4/cluster/ctdb/include/ctdb_private.h
new file mode 100644
index 0000000000..d373e3af32
--- /dev/null
+++ b/source4/cluster/ctdb/include/ctdb_private.h
@@ -0,0 +1,216 @@
+/* 
+   ctdb database library
+
+   Copyright (C) Andrew Tridgell  2006
+
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library 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
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+#ifndef _CTDB_PRIVATE_H
+#define _CTDB_PRIVATE_H
+
+#include "ctdb.h"
+
+/*
+  an installed ctdb remote call
+*/
+struct ctdb_registered_call {
+	struct ctdb_registered_call *next, *prev;
+	uint32_t id;
+	ctdb_fn_t fn;
+};
+
+/*
+  this address structure might need to be generalised later for some
+  transports
+*/
+struct ctdb_address {
+	const char *address;
+	int port;
+};
+
+/*
+  state associated with one node
+*/
+struct ctdb_node {
+	struct ctdb_context *ctdb;
+	struct ctdb_address address;
+	const char *name; /* for debug messages */
+	void *private; /* private to transport */
+	uint32_t vnn;
+};
+
+/*
+  transport specific methods
+*/
+struct ctdb_methods {
+	int (*start)(struct ctdb_context *); /* start protocol processing */	
+	int (*add_node)(struct ctdb_node *); /* setup a new node */	
+	int (*queue_pkt)(struct ctdb_node *, uint8_t *data, uint32_t length);
+	void *(*allocate_pkt)(struct ctdb_context *, size_t );
+};
+
+/*
+  transport calls up to the ctdb layer
+*/
+struct ctdb_upcalls {
+	/* recv_pkt is called when a packet comes in */
+	void (*recv_pkt)(struct ctdb_context *, uint8_t *data, uint32_t length);
+
+	/* node_dead is called when an attempt to send to a node fails */
+	void (*node_dead)(struct ctdb_node *);
+
+	/* node_connected is called when a connection to a node is established */
+	void (*node_connected)(struct ctdb_node *);
+};
+
+/* main state of the ctdb daemon */
+struct ctdb_context {
+	struct event_context *ev;
+	struct ctdb_address address;
+	const char *name;
+	uint32_t vnn; /* our own vnn */
+	uint32_t num_nodes;
+	uint32_t num_connected;
+	unsigned flags;
+	struct idr_context *idr;
+	struct ctdb_node **nodes; /* array of nodes in the cluster - indexed by vnn */
+	struct ctdb_registered_call *calls; /* list of registered calls */
+	char *err_msg;
+	struct tdb_context *ltdb;
+	const struct ctdb_methods *methods; /* transport methods */
+	const struct ctdb_upcalls *upcalls; /* transport upcalls */
+	void *private; /* private to transport */
+};
+
+#define CTDB_NO_MEMORY(ctdb, p) do { if (!(p)) { \
+          ctdb_set_error(ctdb, "Out of memory at %s:%d", __FILE__, __LINE__); \
+	  return -1; }} while (0)
+
+#define CTDB_NO_MEMORY_NULL(ctdb, p) do { if (!(p)) { \
+          ctdb_set_error(ctdb, "Out of memory at %s:%d", __FILE__, __LINE__); \
+	  return NULL; }} while (0)
+
+#define CTDB_NO_MEMORY_FATAL(ctdb, p) do { if (!(p)) { \
+          ctdb_fatal(ctdb, "Out of memory in " __location__ ); \
+	  }} while (0)
+
+/* arbitrary maximum timeout for ctdb operations */
+#define CTDB_REQ_TIMEOUT 10
+
+/* max number of redirects before we ask the lmaster */
+#define CTDB_MAX_REDIRECT 2
+
+/* number of consecutive calls from the same node before we give them
+   the record */
+#define CTDB_MAX_LACOUNT 7000
+
+/*
+  the extended header for records in the ltdb
+*/
+struct ctdb_ltdb_header {
+	uint64_t rsn;
+	uint32_t dmaster;
+	uint32_t laccessor;
+	uint32_t lacount;
+};
+
+
+/*
+  operation IDs
+*/
+enum ctdb_operation {
+	CTDB_REQ_CALL       = 0,
+	CTDB_REPLY_CALL     = 1,
+	CTDB_REPLY_REDIRECT = 2,
+	CTDB_REQ_DMASTER    = 3,
+	CTDB_REPLY_DMASTER  = 4,
+	CTDB_REPLY_ERROR    = 5
+};
+
+/*
+  packet structures
+*/
+struct ctdb_req_header {
+	uint32_t length;
+	uint32_t operation;
+	uint32_t destnode;
+	uint32_t srcnode;
+	uint32_t reqid;
+};
+
+struct ctdb_req_call {
+	struct ctdb_req_header hdr;
+	uint32_t callid;
+	uint32_t keylen;
+	uint32_t calldatalen;
+	uint8_t data[0]; /* key[] followed by calldata[] */
+};
+
+struct ctdb_reply_call {
+	struct ctdb_req_header hdr;
+	uint32_t datalen;
+	uint8_t  data[0];
+};
+
+struct ctdb_reply_error {
+	struct ctdb_req_header hdr;
+	uint32_t status;
+	uint32_t msglen;
+	uint8_t  msg[0];
+};
+
+struct ctdb_reply_redirect {
+	struct ctdb_req_header hdr;
+	uint32_t dmaster;
+};
+
+struct ctdb_req_dmaster {
+	struct ctdb_req_header hdr;
+	uint32_t dmaster;
+	uint32_t keylen;
+	uint32_t datalen;
+	uint8_t  data[0];
+};
+
+struct ctdb_reply_dmaster {
+	struct ctdb_req_header hdr;
+	uint32_t datalen;
+	uint8_t  data[0];
+};
+
+/* internal prototypes */
+void ctdb_set_error(struct ctdb_context *ctdb, const char *fmt, ...);
+void ctdb_fatal(struct ctdb_context *ctdb, const char *msg);
+bool ctdb_same_address(struct ctdb_address *a1, struct ctdb_address *a2);
+int ctdb_parse_address(struct ctdb_context *ctdb,
+		       TALLOC_CTX *mem_ctx, const char *str,
+		       struct ctdb_address *address);
+uint32_t ctdb_hash(const TDB_DATA *key);
+void ctdb_request_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
+void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
+void ctdb_reply_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
+void ctdb_reply_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
+void ctdb_reply_error(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
+void ctdb_reply_redirect(struct ctdb_context *ctdb, struct ctdb_req_header *hdr);
+
+uint32_t ctdb_lmaster(struct ctdb_context *ctdb, const TDB_DATA *key);
+int ctdb_ltdb_fetch(struct ctdb_context *ctdb, 
+		    TDB_DATA key, struct ctdb_ltdb_header *header, TDB_DATA *data);
+int ctdb_ltdb_store(struct ctdb_context *ctdb, TDB_DATA key, 
+		    struct ctdb_ltdb_header *header, TDB_DATA data);
+
+
+#endif
diff --git a/source4/cluster/ctdb/tcp/ctdb_tcp.h b/source4/cluster/ctdb/tcp/ctdb_tcp.h
new file mode 100644
index 0000000000..0f8ce300b4
--- /dev/null
+++ b/source4/cluster/ctdb/tcp/ctdb_tcp.h
@@ -0,0 +1,76 @@
+/* 
+   ctdb database library
+
+   Copyright (C) Andrew Tridgell  2006
+
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library 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
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+
+/* ctdb_tcp main state */
+struct ctdb_tcp {
+	int listen_fd;
+};
+
+/*
+  incoming packet structure - only used when we get a partial packet
+  on read
+*/
+struct ctdb_tcp_partial {
+	uint8_t *data;
+	uint32_t length;
+};
+
+
+/*
+  state associated with an incoming connection
+*/
+struct ctdb_incoming {
+	struct ctdb_context *ctdb;
+	int fd;
+	struct ctdb_tcp_partial partial;
+};
+
+/*
+  outgoing packet structure - only allocated when we can't write immediately
+  to the socket
+*/
+struct ctdb_tcp_packet {
+	struct ctdb_tcp_packet *next, *prev;
+	uint8_t *data;
+	uint32_t length;
+};
+
+/*
+  state associated with one tcp node
+*/
+struct ctdb_tcp_node {
+	int fd;
+	struct fd_event *fde;
+	struct ctdb_tcp_packet *queue;
+};
+
+
+/* prototypes internal to tcp transport */
+void ctdb_tcp_node_write(struct event_context *ev, struct fd_event *fde, 
+			 uint16_t flags, void *private);
+void ctdb_tcp_incoming_read(struct event_context *ev, struct fd_event *fde, 
+			    uint16_t flags, void *private);
+int ctdb_tcp_queue_pkt(struct ctdb_node *node, uint8_t *data, uint32_t length);
+int ctdb_tcp_listen(struct ctdb_context *ctdb);
+void ctdb_tcp_node_connect(struct event_context *ev, struct timed_event *te, 
+			   struct timeval t, void *private);
+
+#define CTDB_TCP_ALIGNMENT 8
diff --git a/source4/cluster/ctdb/tcp/tcp_connect.c b/source4/cluster/ctdb/tcp/tcp_connect.c
new file mode 100644
index 0000000000..2404144ac1
--- /dev/null
+++ b/source4/cluster/ctdb/tcp/tcp_connect.c
@@ -0,0 +1,191 @@
+/* 
+   ctdb over TCP
+
+   Copyright (C) Andrew Tridgell  2006
+
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library 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
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+#include "includes.h"
+#include "lib/events/events.h"
+#include "lib/tdb/include/tdb.h"
+#include "system/network.h"
+#include "system/filesys.h"
+#include "cluster/ctdb/include/ctdb_private.h"
+#include "ctdb_tcp.h"
+
+static void set_nonblocking(int fd)
+{
+	unsigned v;
+	v = fcntl(fd, F_GETFL, 0);
+        fcntl(fd, F_SETFL, v | O_NONBLOCK);
+}
+
+
+/*
+  called when socket becomes writeable on connect
+*/
+static void ctdb_node_connect_write(struct event_context *ev, struct fd_event *fde, 
+				    uint16_t flags, void *private)
+{
+	struct ctdb_node *node = talloc_get_type(private, struct ctdb_node);
+	struct ctdb_tcp_node *tnode = talloc_get_type(node->private, 
+						      struct ctdb_tcp_node);
+	struct ctdb_context *ctdb = node->ctdb;
+	int error = 0;
+	socklen_t len = sizeof(error);
+
+	if (getsockopt(tnode->fd, SOL_SOCKET, SO_ERROR, &error, &len) != 0 ||
+	    error != 0) {
+		talloc_free(fde);
+		close(tnode->fd);
+		tnode->fd = -1;
+		event_add_timed(ctdb->ev, node, timeval_current_ofs(1, 0), 
+				ctdb_tcp_node_connect, node);
+		return;
+	}
+
+	talloc_free(fde);
+	tnode->fde = event_add_fd(node->ctdb->ev, node, tnode->fd, EVENT_FD_READ, 
+				  ctdb_tcp_node_write, node);
+
+	/* tell the ctdb layer we are connected */
+	node->ctdb->upcalls->node_connected(node);
+
+	if (tnode->queue) {
+		EVENT_FD_WRITEABLE(tnode->fde);		
+	}
+}
+
+/*
+  called when we should try and establish a tcp connection to a node
+*/
+void ctdb_tcp_node_connect(struct event_context *ev, struct timed_event *te, 
+			   struct timeval t, void *private)
+{
+	struct ctdb_node *node = talloc_get_type(private, struct ctdb_node);
+	struct ctdb_tcp_node *tnode = talloc_get_type(node->private, 
+						      struct ctdb_tcp_node);
+	struct ctdb_context *ctdb = node->ctdb;
+        struct sockaddr_in sock_out;
+
+	tnode->fd = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
+
+	set_nonblocking(tnode->fd);
+
+	inet_pton(AF_INET, node->address.address, &sock_out.sin_addr);
+	sock_out.sin_port = htons(node->address.port);
+	sock_out.sin_family = PF_INET;
+	
+	if (connect(tnode->fd, (struct sockaddr *)&sock_out, sizeof(sock_out)) != 0 &&
+	    errno != EINPROGRESS) {
+		/* try again once a second */
+		close(tnode->fd);
+		event_add_timed(ctdb->ev, node, timeval_current_ofs(1, 0), 
+				ctdb_tcp_node_connect, node);
+		return;
+	}
+
+	/* non-blocking connect - wait for write event */
+	event_add_fd(node->ctdb->ev, node, tnode->fd, EVENT_FD_WRITE|EVENT_FD_READ, 
+		     ctdb_node_connect_write, node);
+}
+
+/*
+  destroy a ctdb_incoming structure 
+*/
+static int ctdb_incoming_destructor(struct ctdb_incoming *in)
+{
+	close(in->fd);
+	in->fd = -1;
+	return 0;
+}
+
+/*
+  called when we get contacted by another node
+  currently makes no attempt to check if the connection is really from a ctdb
+  node in our cluster
+*/
+static void ctdb_listen_event(struct event_context *ev, struct fd_event *fde, 
+			      uint16_t flags, void *private)
+{
+	struct ctdb_context *ctdb;
+	struct ctdb_tcp *ctcp;
+	struct sockaddr_in addr;
+	socklen_t len;
+	int fd;
+	struct ctdb_incoming *in;
+
+	ctdb = talloc_get_type(private, struct ctdb_context);
+	ctcp = talloc_get_type(ctdb->private, struct ctdb_tcp);
+	memset(&addr, 0, sizeof(addr));
+	len = sizeof(addr);
+	fd = accept(ctcp->listen_fd, (struct sockaddr *)&addr, &len);
+	if (fd == -1) return;
+
+	in = talloc_zero(ctdb, struct ctdb_incoming);
+	in->fd = fd;
+	in->ctdb = ctdb;
+
+	set_nonblocking(in->fd);
+
+	event_add_fd(ctdb->ev, in, in->fd, EVENT_FD_READ, 
+		     ctdb_tcp_incoming_read, in);	
+
+	talloc_set_destructor(in, ctdb_incoming_destructor);
+}
+
+
+/*
+  listen on our own address
+*/
+int ctdb_tcp_listen(struct ctdb_context *ctdb)
+{
+	struct ctdb_tcp *ctcp = talloc_get_type(ctdb->private, struct ctdb_tcp);
+        struct sockaddr_in sock;
+	int one = 1;
+
+        sock.sin_port = htons(ctdb->address.port);
+        sock.sin_family = PF_INET;
+	inet_pton(AF_INET, ctdb->address.address, &sock.sin_addr);
+
+        ctcp->listen_fd = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
+        if (ctcp->listen_fd == -1) {
+		ctdb_set_error(ctdb, "socket failed\n");
+		return -1;
+        }
+
+        setsockopt(ctcp->listen_fd,SOL_SOCKET,SO_REUSEADDR,(char *)&one,sizeof(one));
+
+        if (bind(ctcp->listen_fd, (struct sockaddr * )&sock, sizeof(sock)) != 0) {
+		ctdb_set_error(ctdb, "bind failed\n");
+		close(ctcp->listen_fd);
+		ctcp->listen_fd = -1;
+                return -1;
+        }
+
+	if (listen(ctcp->listen_fd, 10) == -1) {
+		ctdb_set_error(ctdb, "listen failed\n");
+		close(ctcp->listen_fd);
+		ctcp->listen_fd = -1;
+		return -1;
+	}
+
+	event_add_fd(ctdb->ev, ctdb, ctcp->listen_fd, EVENT_FD_READ, 
+		     ctdb_listen_event, ctdb);	
+
+	return 0;
+}
+
diff --git a/source4/cluster/ctdb/tcp/tcp_init.c b/source4/cluster/ctdb/tcp/tcp_init.c
new file mode 100644
index 0000000000..b8ee8cb30e
--- /dev/null
+++ b/source4/cluster/ctdb/tcp/tcp_init.c
@@ -0,0 +1,102 @@
+/* 
+   ctdb over TCP
+
+   Copyright (C) Andrew Tridgell  2006
+
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library 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
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+#include "includes.h"
+#include "lib/tdb/include/tdb.h"
+#include "lib/events/events.h"
+#include "system/network.h"
+#include "system/filesys.h"
+#include "cluster/ctdb/include/ctdb_private.h"
+#include "ctdb_tcp.h"
+
+/*
+  start the protocol going
+*/
+int ctdb_tcp_start(struct ctdb_context *ctdb)
+{
+	int i;
+
+	/* listen on our own address */
+	if (ctdb_tcp_listen(ctdb) != 0) return -1;
+
+	/* startup connections to the other servers - will happen on
+	   next event loop */
+	for (i=0;i<ctdb->num_nodes;i++) {
+		struct ctdb_node *node = *(ctdb->nodes + i);
+		if (!(ctdb->flags & CTDB_FLAG_SELF_CONNECT) &&
+		    ctdb_same_address(&ctdb->address, &node->address)) continue;
+		event_add_timed(ctdb->ev, node, timeval_zero(), 
+				ctdb_tcp_node_connect, node);
+	}
+
+	return 0;
+}
+
+
+/*
+  initialise tcp portion of a ctdb node 
+*/
+int ctdb_tcp_add_node(struct ctdb_node *node)
+{
+	struct ctdb_tcp_node *tnode;
+	tnode = talloc_zero(node, struct ctdb_tcp_node);
+	CTDB_NO_MEMORY(node->ctdb, tnode);
+
+	tnode->fd = -1;
+	node->private = tnode;
+	return 0;
+}
+
+
+/*
+  transport packet allocator - allows transport to control memory for packets
+*/
+void *ctdb_tcp_allocate_pkt(struct ctdb_context *ctdb, size_t size)
+{
+	/* tcp transport needs to round to 8 byte alignment to ensure
+	   that we can use a length header and 64 bit elements in
+	   structures */
+	size = (size+(CTDB_TCP_ALIGNMENT-1)) & ~(CTDB_TCP_ALIGNMENT-1);
+	return talloc_size(ctdb, size);
+}
+
+
+static const struct ctdb_methods ctdb_tcp_methods = {
+	.start     = ctdb_tcp_start,
+	.add_node  = ctdb_tcp_add_node,
+	.queue_pkt = ctdb_tcp_queue_pkt,
+	.allocate_pkt = ctdb_tcp_allocate_pkt
+};
+
+/*
+  initialise tcp portion of ctdb 
+*/
+int ctdb_tcp_init(struct ctdb_context *ctdb)
+{
+	struct ctdb_tcp *ctcp;
+	ctcp = talloc_zero(ctdb, struct ctdb_tcp);
+	CTDB_NO_MEMORY(ctdb, ctcp);
+
+	ctcp->listen_fd = -1;
+	ctdb->private = ctcp;
+	ctdb->methods = &ctdb_tcp_methods;
+	return 0;
+}
+
diff --git a/source4/cluster/ctdb/tcp/tcp_io.c b/source4/cluster/ctdb/tcp/tcp_io.c
new file mode 100644
index 0000000000..82e24f7260
--- /dev/null
+++ b/source4/cluster/ctdb/tcp/tcp_io.c
@@ -0,0 +1,254 @@
+/* 
+   ctdb over TCP
+
+   Copyright (C) Andrew Tridgell  2006
+
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library 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
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+#include "includes.h"
+#include "lib/events/events.h"
+#include "lib/util/dlinklist.h"
+#include "lib/tdb/include/tdb.h"
+#include "system/network.h"
+#include "system/filesys.h"
+#include "cluster/ctdb/include/ctdb_private.h"
+#include "ctdb_tcp.h"
+
+
+/*
+  called when we fail to send a message to a node
+*/
+static void ctdb_tcp_node_dead(struct event_context *ev, struct timed_event *te, 
+			       struct timeval t, void *private)
+{
+	struct ctdb_node *node = talloc_get_type(private, struct ctdb_node);
+	struct ctdb_tcp_node *tnode = talloc_get_type(node->private, 
+						      struct ctdb_tcp_node);
+
+	/* flush the queue */
+	while (tnode->queue) {
+		struct ctdb_tcp_packet *pkt = tnode->queue;
+		DLIST_REMOVE(tnode->queue, pkt);
+		talloc_free(pkt);
+	}
+
+	/* start a new connect cycle to try to re-establish the
+	   link */
+	talloc_free(tnode->fde);
+	close(tnode->fd);
+	tnode->fd = -1;
+	event_add_timed(node->ctdb->ev, node, timeval_zero(), 
+			ctdb_tcp_node_connect, node);
+}
+
+/*
+  called when socket becomes readable
+*/
+void ctdb_tcp_node_write(struct event_context *ev, struct fd_event *fde, 
+			 uint16_t flags, void *private)
+{
+	struct ctdb_node *node = talloc_get_type(private, struct ctdb_node);
+	struct ctdb_tcp_node *tnode = talloc_get_type(node->private, 
+						      struct ctdb_tcp_node);
+	if (flags & EVENT_FD_READ) {
+		/* getting a read event on this fd in the current tcp model is
+		   always an error, as we have separate read and write
+		   sockets. In future we may combine them, but for now it must
+		   mean that the socket is dead, so we try to reconnect */
+		talloc_free(tnode->fde);
+		close(tnode->fd);
+		tnode->fd = -1;
+		event_add_timed(node->ctdb->ev, node, timeval_zero(), 
+				ctdb_tcp_node_connect, node);
+		return;
+	}
+
+	while (tnode->queue) {
+		struct ctdb_tcp_packet *pkt = tnode->queue;
+		ssize_t n;
+
+		n = write(tnode->fd, pkt->data, pkt->length);
+
+		if (n == -1 && errno != EAGAIN && errno != EWOULDBLOCK) {
+			event_add_timed(node->ctdb->ev, node, timeval_zero(), 
+					ctdb_tcp_node_dead, node);
+			EVENT_FD_NOT_WRITEABLE(tnode->fde);
+			return;
+		}
+		if (n <= 0) return;
+		
+		if (n != pkt->length) {
+			pkt->length -= n;
+			pkt->data += n;
+			return;
+		}
+
+		DLIST_REMOVE(tnode->queue, pkt);
+		talloc_free(pkt);
+	}
+
+	EVENT_FD_NOT_WRITEABLE(tnode->fde);
+}
+
+
+/*
+  called when an incoming connection is readable
+*/
+void ctdb_tcp_incoming_read(struct event_context *ev, struct fd_event *fde, 
+			    uint16_t flags, void *private)
+{
+	struct ctdb_incoming *in = talloc_get_type(private, struct ctdb_incoming);
+	int num_ready = 0;
+	ssize_t nread;
+	uint8_t *data, *data_base;
+
+	if (ioctl(in->fd, FIONREAD, &num_ready) != 0 ||
+	    num_ready == 0) {
+		/* we've lost the link from another node. We don't
+		   notify the upper layers, as we only want to trigger
+		   a full node reorganisation when a send fails - that
+		   allows nodes to restart without penalty as long as
+		   the network is idle */
+		talloc_free(in);
+		return;
+	}
+
+	in->partial.data = talloc_realloc_size(in, in->partial.data, 
+					       num_ready + in->partial.length);
+	if (in->partial.data == NULL) {
+		/* not much we can do except drop the socket */
+		talloc_free(in);
+		return;
+	}
+
+	nread = read(in->fd, in->partial.data+in->partial.length, num_ready);
+	if (nread <= 0) {
+		/* the connection must be dead */
+		talloc_free(in);
+		return;
+	}
+
+	data = in->partial.data;
+	nread += in->partial.length;
+
+	in->partial.data = NULL;
+	in->partial.length = 0;
+
+	if (nread >= 4 && *(uint32_t *)data == nread) {
+		/* most common case - we got a whole packet in one go
+		   tell the ctdb layer above that we have a packet */
+		in->ctdb->upcalls->recv_pkt(in->ctdb, data, nread);
+		return;
+	}
+
+	data_base = data;
+
+	while (nread >= 4 && *(uint32_t *)data <= nread) {
+		/* we have at least one packet */
+		uint8_t *d2;
+		uint32_t len;
+		len = *(uint32_t *)data;
+		d2 = talloc_memdup(in, data, len);
+		if (d2 == NULL) {
+			/* sigh */
+			talloc_free(in);
+			return;
+		}
+		in->ctdb->upcalls->recv_pkt(in->ctdb, d2, len);
+		data += len;
+		nread -= len;		
+		return;
+	}
+
+	if (nread < 4 || *(uint32_t *)data > nread) {
+		/* we have only part of a packet */
+		if (data_base == data) {
+			in->partial.data = data;
+			in->partial.length = nread;
+		} else {
+			in->partial.data = talloc_memdup(in, data, nread);
+			if (in->partial.data == NULL) {
+				talloc_free(in);
+				return;
+			}
+			in->partial.length = nread;
+			talloc_free(data_base);
+		}
+		return;
+	}
+
+	talloc_free(data_base);
+}
+
+/*
+  queue a packet for sending
+*/
+int ctdb_tcp_queue_pkt(struct ctdb_node *node, uint8_t *data, uint32_t length)
+{
+	struct ctdb_tcp_node *tnode = talloc_get_type(node->private, 
+						      struct ctdb_tcp_node);
+	struct ctdb_tcp_packet *pkt;
+	uint32_t length2;
+
+	/* enforce the length and alignment rules from the tcp packet allocator */
+	length2 = (length+(CTDB_TCP_ALIGNMENT-1)) & ~(CTDB_TCP_ALIGNMENT-1);
+	*(uint32_t *)data = length2;
+
+	if (length2 != length) {
+		memset(data+length, 0, length2-length);
+	}
+	{
+		int i, fd = open("/dev/null", O_WRONLY);
+		for (i=0;i<length2;i++) {
+			write(fd, &data[i], 1);
+		}
+		close(fd);
+	}
+	
+	/* if the queue is empty then try an immediate write, avoiding
+	   queue overhead. This relies on non-blocking sockets */
+	if (tnode->queue == NULL && tnode->fd != -1) {
+		ssize_t n = write(tnode->fd, data, length2);
+		if (n == -1 && errno != EAGAIN && errno != EWOULDBLOCK) {
+			event_add_timed(node->ctdb->ev, node, timeval_zero(), 
+					ctdb_tcp_node_dead, node);
+			/* yes, we report success, as the dead node is 
+			   handled via a separate event */
+			return 0;
+		}
+		if (n > 0) {
+			data += n;
+			length2 -= n;
+		}
+		if (length2 == 0) return 0;
+	}
+
+	pkt = talloc(tnode, struct ctdb_tcp_packet);
+	CTDB_NO_MEMORY(node->ctdb, pkt);
+
+	pkt->data = talloc_memdup(pkt, data, length2);
+	CTDB_NO_MEMORY(node->ctdb, pkt->data);
+
+	pkt->length = length2;
+
+	if (tnode->queue == NULL && tnode->fd != -1) {
+		EVENT_FD_WRITEABLE(tnode->fde);
+	}
+
+	DLIST_ADD_END(tnode->queue, pkt, struct ctdb_tcp_packet *);
+
+	return 0;
+}
diff --git a/source4/cluster/ctdb/tests/ctdb_bench.c b/source4/cluster/ctdb/tests/ctdb_bench.c
new file mode 100644
index 0000000000..01a8cc0f15
--- /dev/null
+++ b/source4/cluster/ctdb/tests/ctdb_bench.c
@@ -0,0 +1,228 @@
+/* 
+   simple ctdb benchmark
+
+   Copyright (C) Andrew Tridgell  2006
+
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library 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
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+#include "includes.h"
+#include "lib/events/events.h"
+#include "system/filesys.h"
+#include "popt.h"
+
+#include <sys/time.h>
+#include <time.h>
+
+static struct timeval tp1,tp2;
+
+static void start_timer(void)
+{
+	gettimeofday(&tp1,NULL);
+}
+
+static double end_timer(void)
+{
+	gettimeofday(&tp2,NULL);
+	return (tp2.tv_sec + (tp2.tv_usec*1.0e-6)) - 
+		(tp1.tv_sec + (tp1.tv_usec*1.0e-6));
+}
+
+
+static int timelimit = 10;
+static int num_records = 10;
+static int num_repeats = 100;
+
+enum my_functions {FUNC_INCR=1, FUNC_FETCH=2};
+
+/*
+  ctdb call function to increment an integer
+*/
+static int incr_func(struct ctdb_call *call)
+{
+	if (call->record_data.dsize == 0) {
+		call->new_data = talloc(call, TDB_DATA);
+		if (call->new_data == NULL) {
+			return CTDB_ERR_NOMEM;
+		}
+		call->new_data->dptr = talloc_size(call, 4);
+		call->new_data->dsize = 4;
+		*(uint32_t *)call->new_data->dptr = 0;
+	} else {
+		call->new_data = &call->record_data;
+	}
+	(*(uint32_t *)call->new_data->dptr)++;
+	return 0;
+}
+
+/*
+  ctdb call function to fetch a record
+*/
+static int fetch_func(struct ctdb_call *call)
+{
+	call->reply_data = &call->record_data;
+	return 0;
+}
+
+/*
+  benchmark incrementing an integer
+*/
+static void bench_incr(struct ctdb_context *ctdb)
+{
+	TDB_DATA key, data;
+	int loops=0;
+	int ret, i;
+
+	start_timer();
+
+	while (1) {
+		uint32_t v = loops % num_records;
+		key.dptr = &v;
+		key.dsize = 4;
+		for (i=0;i<num_repeats;i++) {
+			ret = ctdb_call(ctdb, key, FUNC_INCR, NULL, NULL);
+			if (ret != 0) {
+				printf("incr call failed - %s\n", ctdb_errstr(ctdb));
+				return;
+			}
+		}
+		if (num_repeats * (++loops) % 10000 == 0) {
+			if (end_timer() > timelimit) break;
+			printf("Incr: %.2f ops/sec\r", num_repeats*loops/end_timer());
+			fflush(stdout);
+		}
+	}
+
+	ret = ctdb_call(ctdb, key, FUNC_FETCH, NULL, &data);
+	if (ret == -1) {
+		printf("ctdb_call FUNC_FETCH failed - %s\n", ctdb_errstr(ctdb));
+		return;
+	}
+
+	printf("Incr: %.2f ops/sec (loops=%d val=%d)\n", 
+	       num_repeats*loops/end_timer(), loops, *(uint32_t *)data.dptr);
+}
+
+/*
+  main program
+*/
+int main(int argc, const char *argv[])
+{
+	struct ctdb_context *ctdb;
+	const char *nlist = NULL;
+	const char *transport = "tcp";
+	const char *myaddress = NULL;
+	int self_connect=0;
+
+	struct poptOption popt_options[] = {
+		POPT_AUTOHELP
+		{ "nlist", 0, POPT_ARG_STRING, &nlist, 0, "node list file", "filename" },
+		{ "listen", 0, POPT_ARG_STRING, &myaddress, 0, "address to listen on", "address" },
+		{ "transport", 0, POPT_ARG_STRING, &transport, 0, "protocol transport", NULL },
+		{ "self-connect", 0, POPT_ARG_NONE, &self_connect, 0, "enable self connect", "boolean" },
+		{ "timelimit", 't', POPT_ARG_INT, &timelimit, 0, "timelimit", "integer" },
+		{ "num-records", 'r', POPT_ARG_INT, &num_records, 0, "num_records", "integer" },
+		POPT_TABLEEND
+	};
+	int opt;
+	const char **extra_argv;
+	int extra_argc = 0;
+	int ret;
+	poptContext pc;
+	struct event_context *ev;
+
+	pc = poptGetContext(argv[0], argc, argv, popt_options, POPT_CONTEXT_KEEP_FIRST);
+
+	while ((opt = poptGetNextOpt(pc)) != -1) {
+		switch (opt) {
+		default:
+			fprintf(stderr, "Invalid option %s: %s\n", 
+				poptBadOption(pc, 0), poptStrerror(opt));
+			exit(1);
+		}
+	}
+
+	/* setup the remaining options for the main program to use */
+	extra_argv = poptGetArgs(pc);
+	if (extra_argv) {
+		extra_argv++;
+		while (extra_argv[extra_argc]) extra_argc++;
+	}
+
+	if (nlist == NULL || myaddress == NULL) {
+		printf("You must provide a node list with --nlist and an address with --listen\n");
+		exit(1);
+	}
+
+	ev = event_context_init(NULL);
+
+	/* initialise ctdb */
+	ctdb = ctdb_init(ev);
+	if (ctdb == NULL) {
+		printf("Failed to init ctdb\n");
+		exit(1);
+	}
+
+	if (self_connect) {
+		ctdb_set_flags(ctdb, CTDB_FLAG_SELF_CONNECT);
+	}
+
+	ret = ctdb_set_transport(ctdb, transport);
+	if (ret == -1) {
+		printf("ctdb_set_transport failed - %s\n", ctdb_errstr(ctdb));
+		exit(1);
+	}
+
+	/* tell ctdb what address to listen on */
+	ret = ctdb_set_address(ctdb, myaddress);
+	if (ret == -1) {
+		printf("ctdb_set_address failed - %s\n", ctdb_errstr(ctdb));
+		exit(1);
+	}
+
+	/* tell ctdb what nodes are available */
+	ret = ctdb_set_nlist(ctdb, nlist);
+	if (ret == -1) {
+		printf("ctdb_set_nlist failed - %s\n", ctdb_errstr(ctdb));
+		exit(1);
+	}
+
+	/* setup a ctdb call function */
+	ret = ctdb_set_call(ctdb, incr_func,  FUNC_INCR);
+	ret = ctdb_set_call(ctdb, fetch_func, FUNC_FETCH);
+
+	/* attach to a specific database */
+	ret = ctdb_attach(ctdb, "test.tdb", TDB_DEFAULT, O_RDWR|O_CREAT|O_TRUNC, 0666);
+	if (ret == -1) {
+		printf("ctdb_attach failed - %s\n", ctdb_errstr(ctdb));
+		exit(1);
+	}
+
+	/* start the protocol running */
+	ret = ctdb_start(ctdb);
+
+	/* wait until all nodes are connected (should not be needed
+	   outside of test code) */
+	ctdb_connect_wait(ctdb);
+
+	bench_incr(ctdb);
+       
+	/* go into a wait loop to allow other nodes to complete */
+	ctdb_wait_loop(ctdb);
+
+	/* shut it down */
+	talloc_free(ctdb);
+	return 0;
+}
diff --git a/source4/cluster/ctdb/tests/ctdb_test.c b/source4/cluster/ctdb/tests/ctdb_test.c
new file mode 100644
index 0000000000..5bc35ad332
--- /dev/null
+++ b/source4/cluster/ctdb/tests/ctdb_test.c
@@ -0,0 +1,207 @@
+/* 
+   ctdb test harness
+
+   Copyright (C) Andrew Tridgell  2006
+
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 2 of the License, or (at your option) any later version.
+
+   This library 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
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General Public
+   License along with this library; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+*/
+
+#include "includes.h"
+#include "lib/events/events.h"
+#include "system/filesys.h"
+#include "popt.h"
+
+enum my_functions {FUNC_SORT=1, FUNC_FETCH=2};
+
+static int int_compare(int *i1, int *i2)
+{
+	return *i1 - *i2;
+}
+
+/*
+  add an integer into a record in sorted order
+*/
+static int sort_func(struct ctdb_call *call)
+{
+	if (call->call_data == NULL ||
+	    call->call_data->dsize != sizeof(int)) {
+		return CTDB_ERR_INVALID;
+	}
+	call->new_data = talloc(call, TDB_DATA);
+	if (call->new_data == NULL) {
+		return CTDB_ERR_NOMEM;
+	}
+	call->new_data->dptr = talloc_size(call, 
+					   call->record_data.dsize + 
+					   call->call_data->dsize);
+	if (call->new_data->dptr == NULL) {
+		return CTDB_ERR_NOMEM;
+	}
+	call->new_data->dsize = call->record_data.dsize + call->call_data->dsize;
+	memcpy(call->new_data->dptr,
+	       call->record_data.dptr, call->record_data.dsize);
+	memcpy(call->new_data->dptr+call->record_data.dsize,
+	       call->call_data->dptr, call->call_data->dsize);
+
+	qsort(call->new_data->dptr, call->new_data->dsize / sizeof(int),
+	      sizeof(int), (comparison_fn_t)int_compare);
+
+	return 0;
+}
+
+/*
+  ctdb call function to fetch a record
+*/
+static int fetch_func(struct ctdb_call *call)
+{
+	call->reply_data = &call->record_data;
+	return 0;
+}
+
+/*
+  main program
+*/
+int main(int argc, const char *argv[])
+{
+	struct ctdb_context *ctdb;
+	const char *nlist = NULL;
+	const char *transport = "tcp";
+	const char *myaddress = NULL;
+	int self_connect=0;
+
+	struct poptOption popt_options[] = {
+		POPT_AUTOHELP
+		{ "nlist", 0, POPT_ARG_STRING, &nlist, 0, "node list file", "filename" },
+		{ "listen", 0, POPT_ARG_STRING, &myaddress, 0, "address to listen on", "address" },
+		{ "transport", 0, POPT_ARG_STRING, &transport, 0, "protocol transport", NULL },
+		{ "self-connect", 0, POPT_ARG_NONE, &self_connect, 0, "enable self connect", "boolean" },
+		POPT_TABLEEND
+	};
+	int opt;
+	const char **extra_argv;
+	int extra_argc = 0;
+	int i, ret;
+	TDB_DATA key, data;
+	poptContext pc;
+	struct event_context *ev;
+
+	pc = poptGetContext(argv[0], argc, argv, popt_options, POPT_CONTEXT_KEEP_FIRST);
+
+	while ((opt = poptGetNextOpt(pc)) != -1) {
+		switch (opt) {
+		default:
+			fprintf(stderr, "Invalid option %s: %s\n", 
+				poptBadOption(pc, 0), poptStrerror(opt));
+			exit(1);
+		}
+	}
+
+	/* setup the remaining options for the main program to use */
+	extra_argv = poptGetArgs(pc);
+	if (extra_argv) {
+		extra_argv++;
+		while (extra_argv[extra_argc]) extra_argc++;
+	}
+
+	if (nlist == NULL || myaddress == NULL) {
+		printf("You must provide a node list with --nlist and an address with --listen\n");
+		exit(1);
+	}
+
+	ev = event_context_init(NULL);
+
+	/* initialise ctdb */
+	ctdb = ctdb_init(ev);
+	if (ctdb == NULL) {
+		printf("Failed to init ctdb\n");
+		exit(1);
+	}
+
+	if (self_connect) {
+		ctdb_set_flags(ctdb, CTDB_FLAG_SELF_CONNECT);
+	}
+
+	ret = ctdb_set_transport(ctdb, transport);
+	if (ret == -1) {
+		printf("ctdb_set_transport failed - %s\n", ctdb_errstr(ctdb));
+		exit(1);
+	}
+
+	/* tell ctdb what address to listen on */
+	ret = ctdb_set_address(ctdb, myaddress);
+	if (ret == -1) {
+		printf("ctdb_set_address failed - %s\n", ctdb_errstr(ctdb));
+		exit(1);
+	}
+
+	/* tell ctdb what nodes are available */
+	ret = ctdb_set_nlist(ctdb, nlist);
+	if (ret == -1) {
+		printf("ctdb_set_nlist failed - %s\n", ctdb_errstr(ctdb));
+		exit(1);
+	}
+
+	/* setup a ctdb call function */
+	ret = ctdb_set_call(ctdb, sort_func,  FUNC_SORT);
+	ret = ctdb_set_call(ctdb, fetch_func, FUNC_FETCH);
+
+	/* attach to a specific database */
+	ret = ctdb_attach(ctdb, "test.tdb", TDB_DEFAULT, O_RDWR|O_CREAT|O_TRUNC, 0666);
+	if (ret == -1) {
+		printf("ctdb_attach failed - %s\n", ctdb_errstr(ctdb));
+		exit(1);
+	}
+
+	/* start the protocol running */
+	ret = ctdb_start(ctdb);
+
+	/* wait until all nodes are connected (should not be needed
+	   outide of test code) */
+	ctdb_connect_wait(ctdb);
+       
+	key.dptr = "test";
+	key.dsize = strlen("test")+1;
+
+	/* add some random data */
+	for (i=0;i<10;i++) {
+		int v = random() % 1000;
+		data.dptr = (uint8_t *)&v;
+		data.dsize = sizeof(v);
+		ret = ctdb_call(ctdb, key, FUNC_SORT, &data, NULL);
+		if (ret == -1) {
+			printf("ctdb_call FUNC_SORT failed - %s\n", ctdb_errstr(ctdb));
+			exit(1);
+		}
+	}
+
+	/* fetch the record */
+	ret = ctdb_call(ctdb, key, FUNC_FETCH, NULL, &data);
+	if (ret == -1) {
+		printf("ctdb_call FUNC_FETCH failed - %s\n", ctdb_errstr(ctdb));
+		exit(1);
+	}
+
+	for (i=0;i<data.dsize/sizeof(int);i++) {
+		printf("%3d\n", ((int *)data.dptr)[i]);
+	}
+	talloc_free(data.dptr);
+
+	/* go into a wait loop to allow other nodes to complete */
+	ctdb_wait_loop(ctdb);
+
+	/* shut it down */
+	talloc_free(ctdb);
+	return 0;
+}