1 files changed, 364 insertions, 0 deletions
diff --git a/source4/cluster/ctdb/direct/ctdbd_test.c b/source4/cluster/ctdb/direct/ctdbd_test.c
new file mode 100644
index 0000000000..019cdad30d
--- /dev/null
+++ b/source4/cluster/ctdb/direct/ctdbd_test.c
@@ -0,0 +1,364 @@
+/* 
+   test of messaging
+
+   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 "system/network.h"
+#include "../include/ctdb.h"
+#include "../include/ctdb_private.h"
+
+#define CTDB_SOCKET "/tmp/ctdb.socket.127.0.0.1"
+
+
+/*
+  connect to the unix domain socket
+*/
+static int ux_socket_connect(const char *name)
+{
+	struct sockaddr_un addr;
+	int fd;
+
+	memset(&addr, 0, sizeof(addr));
+	addr.sun_family = AF_UNIX;
+	strncpy(addr.sun_path, name, sizeof(addr.sun_path));
+
+	fd = socket(AF_UNIX, SOCK_STREAM, 0);
+	if (fd == -1) {
+		return -1;
+	}
+	
+	if (connect(fd, (struct sockaddr *)&addr, sizeof(addr)) == -1) {
+		close(fd);
+		return -1;
+	}
+
+	return fd;
+}
+
+void register_pid_with_daemon(int fd, int pid)
+{
+	struct ctdb_req_register r;
+
+	bzero(&r, sizeof(r));
+	r.hdr.length       = sizeof(r);
+	r.hdr.ctdb_magic   = CTDB_MAGIC;
+	r.hdr.ctdb_version = CTDB_VERSION;
+	r.hdr.operation    = CTDB_REQ_REGISTER;
+	r.srvid            = pid;
+
+	/* XXX must deal with partial writes here */
+	write(fd, &r, sizeof(r));
+}
+
+/* send a command to the cluster to wait until all nodes are connected
+   and the cluster is fully operational
+ */
+int wait_for_cluster(int fd)
+{
+	struct ctdb_req_connect_wait req;
+	struct ctdb_reply_connect_wait rep;
+	int cnt, tot;
+
+	/* send a connect wait command to the local node */
+	bzero(&req, sizeof(req));
+	req.hdr.length       = sizeof(req);
+	req.hdr.ctdb_magic   = CTDB_MAGIC;
+	req.hdr.ctdb_version = CTDB_VERSION;
+	req.hdr.operation    = CTDB_REQ_CONNECT_WAIT;
+
+	/* XXX must deal with partial writes here */
+	write(fd, &req, sizeof(req));
+
+
+	/* read the 4 bytes of length for the pdu */
+	cnt=0;
+	tot=4;
+	while(cnt!=tot){
+		int numread;
+		numread=read(fd, ((char *)&rep)+cnt, tot-cnt);
+		if(numread>0){
+			cnt+=numread;
+		}
+	}
+	/* read the rest of the pdu */
+	tot=rep.hdr.length;
+	while(cnt!=tot){
+		int numread;
+		numread=read(fd, ((char *)&rep)+cnt, tot-cnt);
+		if(numread>0){
+			cnt+=numread;
+		}
+	}
+
+	return rep.vnn;
+}
+
+
+int send_a_message(int fd, int ourvnn, int vnn, int pid, TDB_DATA data)
+{
+	struct ctdb_req_message r;
+	int len, cnt;
+
+	len = offsetof(struct ctdb_req_message, data) + data.dsize;
+	r.hdr.length     = len;
+	r.hdr.ctdb_magic = CTDB_MAGIC;
+	r.hdr.ctdb_version = CTDB_VERSION;
+	r.hdr.operation  = CTDB_REQ_MESSAGE;
+	r.hdr.destnode   = vnn;
+	r.hdr.srcnode    = ourvnn;
+	r.hdr.reqid      = 0;
+	r.srvid          = pid;
+	r.datalen        = data.dsize;
+	
+	/* write header */
+	cnt=write(fd, &r, offsetof(struct ctdb_req_message, data));
+	/* write data */
+	if(data.dsize){
+	    cnt=write(fd, data.dptr, data.dsize);
+	}
+	return 0;
+}
+
+int receive_a_message(int fd, struct ctdb_req_message **preply)
+{
+	int cnt,tot;
+	struct ctdb_req_message *rep;
+	uint32_t length;
+
+	/* read the 4 bytes of length for the pdu */
+	cnt=0;
+	tot=4;
+	while(cnt!=tot){
+		int numread;
+		numread=read(fd, ((char *)&length)+cnt, tot-cnt);
+		if(numread>0){
+			cnt+=numread;
+		}
+	}
+	
+	/* read the rest of the pdu */
+	rep = malloc(length);
+	rep->hdr.length = length;
+	cnt = 0;
+	tot = length-4;
+	while(cnt!=tot){
+		int numread;
+		numread=read(fd, ((char *)rep)+cnt, tot-cnt);
+		if(numread>0){
+			cnt+=numread;
+		}
+	}
+
+	*preply = rep;
+	return 0;
+}
+
+/*
+  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);  
+}
+
+void fetch_lock(int fd, uint32_t db_id, TDB_DATA key)
+{
+	struct ctdb_req_fetch_lock *req;
+	struct ctdb_reply_fetch_lock *rep;
+	uint32_t length;
+	int len, cnt, tot;
+
+	len = offsetof(struct ctdb_req_fetch_lock, key) + key.dsize;
+	req = malloc(len);
+
+	req->hdr.length      = len;
+	req->hdr.ctdb_magic  = CTDB_MAGIC;
+	req->hdr.ctdb_version = CTDB_VERSION;
+	req->hdr.operation   = CTDB_REQ_FETCH_LOCK;
+	req->hdr.reqid       = 1;
+	req->db_id           = db_id;
+	req->keylen          = key.dsize;
+	memcpy(&req->key[0], key.dptr, key.dsize);
+
+	cnt=write(fd, req, len);
+
+
+	/* wait fot the reply */
+	/* read the 4 bytes of length for the pdu */
+	cnt=0;
+	tot=4;
+	while(cnt!=tot){
+		int numread;
+		numread=read(fd, ((char *)&length)+cnt, tot-cnt);
+		if(numread>0){
+			cnt+=numread;
+		}
+	}
+	/* read the rest of the pdu */
+	rep = malloc(length);
+	tot=length;
+	while(cnt!=tot){
+		int numread;
+		numread=read(fd, ((char *)rep)+cnt, tot-cnt);
+		if(numread>0){
+			cnt+=numread;
+		}
+	}
+	printf("fetch lock reply: state:%d datalen:%d\n",rep->state,rep->datalen);
+	if(!rep->datalen){
+		printf("no data\n");
+	} else {
+		printf("data:[%s]\n",rep->data);
+	}
+
+}
+
+void store_unlock(int fd, uint32_t db_id, TDB_DATA key, TDB_DATA data)
+{
+	struct ctdb_req_store_unlock *req;
+	struct ctdb_reply_store_unlock *rep;
+	uint32_t length;
+	int len, cnt, tot;
+
+	len = offsetof(struct ctdb_req_store_unlock, data) + key.dsize + data.dsize;
+	req = malloc(len);
+
+	req->hdr.length      = len;
+	req->hdr.ctdb_magic  = CTDB_MAGIC;
+	req->hdr.ctdb_version = CTDB_VERSION;
+	req->hdr.operation   = CTDB_REQ_STORE_UNLOCK;
+	req->hdr.reqid       = 1;
+	req->db_id           = db_id;
+	req->keylen          = key.dsize;
+	req->datalen         = data.dsize;
+	memcpy(&req->data[0], key.dptr, key.dsize);
+	memcpy(&req->data[key.dsize], data.dptr, data.dsize);
+
+	cnt=write(fd, req, len);
+
+
+	/* wait fot the reply */
+	/* read the 4 bytes of length for the pdu */
+	cnt=0;
+	tot=4;
+	while(cnt!=tot){
+		int numread;
+		numread=read(fd, ((char *)&length)+cnt, tot-cnt);
+		if(numread>0){
+			cnt+=numread;
+		}
+	}
+	/* read the rest of the pdu */
+	rep = malloc(length);
+	tot=length;
+	while(cnt!=tot){
+		int numread;
+		numread=read(fd, ((char *)rep)+cnt, tot-cnt);
+		if(numread>0){
+			cnt+=numread;
+		}
+	}
+	printf("store unlock reply: state:%d\n",rep->state);
+}
+
+int main(int argc, const char *argv[])
+{
+	int fd, pid, vnn, dstvnn, dstpid;
+	TDB_DATA message;
+	struct ctdb_req_message *reply;
+	TDB_DATA dbname;
+	uint32_t db_id;
+	TDB_DATA key, data;
+	char str[256];
+
+	/* open the socket to talk to the local ctdb daemon */
+	fd=ux_socket_connect(CTDB_SOCKET);
+	if (fd==-1) {
+		printf("failed to open domain socket\n");
+		exit(10);
+	}
+
+
+	/* register our local server id with the daemon so that it knows
+	   where to send messages addressed to our local pid.
+	 */
+	pid=getpid();
+	register_pid_with_daemon(fd, pid);
+
+
+	/* do a connect wait to ensure that all nodes in the cluster are up 
+	   and operational.
+	   this also tells us the vnn of the local cluster.
+	   If someone wants to send us a emssage they should send it to
+	   this vnn and our pid
+	 */
+	vnn=wait_for_cluster(fd);
+	printf("our address is vnn:%d pid:%d  if someone wants to send us a message!\n",vnn,pid);
+
+
+	/* send a message to ourself */
+	dstvnn=vnn;
+	dstpid=pid;
+	message.dptr=discard_const("Test message");
+	message.dsize=strlen((const char *)message.dptr)+1;
+	printf("sending test message [%s] to ourself\n", message.dptr);
+	send_a_message(fd, vnn, dstvnn, dstpid, message);
+
+	/* wait for the message to come back */
+	receive_a_message(fd, &reply);
+	printf("received message: [%s]\n",&reply->data[0]);
+
+	/* create the db id for "test.tdb" */
+	dbname.dptr = discard_const("test.tdb");
+	dbname.dsize = strlen((const char *)(dbname.dptr));
+	db_id = ctdb_hash(&dbname);
+	printf("the has for the database id is 0x%08x\n",db_id);
+	printf("\n");
+
+	/* send a fetch lock */
+	key.dptr=discard_const("TestKey");
+	key.dsize=strlen((const char *)(key.dptr));
+	printf("fetch the test key:[%s]\n",key.dptr);
+	fetch_lock(fd, db_id, key);
+	printf("\n");
+
+
+	/* send a store unlock */
+	sprintf(str,"TestData_%d",getpid());
+	data.dptr=discard_const(str);
+	data.dsize=strlen((const char *)(data.dptr));
+	printf("store new data==[%s] for this record\n",data.dptr);
+	store_unlock(fd, db_id, key, data);
+	printf("\n");
+
+	/* send a fetch lock */
+	printf("fetch the test key:[%s]\n",key.dptr);
+	fetch_lock(fd, db_id, key);
+	printf("\n");
+
+
+	return 0;
+}