diff options
Diffstat (limited to 'source4/cluster/ctdb')
38 files changed, 2020 insertions, 1104 deletions
diff --git a/source4/cluster/ctdb/Makefile.in b/source4/cluster/ctdb/Makefile.in index 175418aa99..adc1f92d0a 100644 --- a/source4/cluster/ctdb/Makefile.in +++ b/source4/cluster/ctdb/Makefile.in @@ -21,8 +21,8 @@ LIB_FLAGS=@LDFLAGS@ -Llib @LIBS@ -lpopt @INFINIBAND_LIBS@ EVENTS_OBJ = lib/events/events.o lib/events/events_standard.o CTDB_COMMON_OBJ = common/ctdb.o common/ctdb_daemon.o common/ctdb_client.o common/ctdb_io.o common/util.o common/ctdb_util.o \ - common/ctdb_call.o common/ctdb_ltdb.o common/ctdb_message.o \ - lib/util/idtree.o lib/util/db_wrap.o + common/ctdb_call.o common/ctdb_ltdb.o common/ctdb_lockwait.o common/ctdb_message.o \ + common/cmdline.o lib/util/idtree.o lib/util/db_wrap.o lib/util/debug.o CTDB_TCP_OBJ = tcp/tcp_connect.o tcp/tcp_io.o tcp/tcp_init.o @@ -30,7 +30,7 @@ CTDB_OBJ = $(CTDB_COMMON_OBJ) $(CTDB_TCP_OBJ) OBJS = @TDBOBJ@ @TALLOCOBJ@ @LIBREPLACEOBJ@ @INFINIBAND_WRAPPER_OBJ@ $(EXTRA_OBJ) $(EVENTS_OBJ) $(CTDB_OBJ) -BINS = bin/ctdbd bin/ctdbd_test bin/ctdb_test bin/ctdb_bench bin/ctdb_messaging bin/ctdb_fetch bin/ctdb_fetch1 @INFINIBAND_BINS@ +BINS = bin/ctdbd bin/ctdbd_test bin/ctdb_test bin/ctdb_bench bin/ctdb_messaging bin/ctdb_fetch bin/ctdb_fetch1 bin/lockwait bin/ctdb_status @INFINIBAND_BINS@ DIRS = lib bin @@ -49,38 +49,46 @@ showflags: dirs: @mkdir -p $(DIRS) -bin/ctdb_test: $(OBJS) tests/ctdb_test.o tests/cmdline.o +bin/ctdb_test: $(OBJS) tests/ctdb_test.o @echo Linking $@ - @$(CC) $(CFLAGS) -o $@ tests/ctdb_test.o tests/cmdline.o $(OBJS) $(LIB_FLAGS) + @$(CC) $(CFLAGS) -o $@ tests/ctdb_test.o $(OBJS) $(LIB_FLAGS) -bin/ctdbd: $(OBJS) direct/ctdbd.o +bin/ctdbd: $(OBJS) direct/ctdbd.o @echo Linking $@ @$(CC) $(CFLAGS) -o $@ direct/ctdbd.o $(OBJS) $(LIB_FLAGS) +bin/ctdb_status: $(OBJS) tools/ctdb_status.o + @echo Linking $@ + @$(CC) $(CFLAGS) -o $@ tools/ctdb_status.o $(OBJS) $(LIB_FLAGS) + bin/ctdbd_test: $(OBJS) direct/ctdbd_test.o @echo Linking $@ @$(CC) $(CFLAGS) -o $@ direct/ctdbd_test.o -bin/ctdb_bench: $(OBJS) tests/ctdb_bench.o tests/cmdline.o +bin/ctdb_bench: $(OBJS) tests/ctdb_bench.o @echo Linking $@ - @$(CC) $(CFLAGS) -o $@ tests/ctdb_bench.o tests/cmdline.o $(OBJS) $(LIB_FLAGS) + @$(CC) $(CFLAGS) -o $@ tests/ctdb_bench.o $(OBJS) $(LIB_FLAGS) -bin/ctdb_fetch: $(OBJS) tests/ctdb_fetch.o tests/cmdline.o +bin/ctdb_fetch: $(OBJS) tests/ctdb_fetch.o @echo Linking $@ - @$(CC) $(CFLAGS) -o $@ tests/ctdb_fetch.o tests/cmdline.o $(OBJS) $(LIB_FLAGS) + @$(CC) $(CFLAGS) -o $@ tests/ctdb_fetch.o $(OBJS) $(LIB_FLAGS) -bin/ctdb_fetch1: $(OBJS) tests/ctdb_fetch1.o tests/cmdline.o +bin/ctdb_fetch1: $(OBJS) tests/ctdb_fetch1.o @echo Linking $@ - @$(CC) $(CFLAGS) -o $@ tests/ctdb_fetch1.o tests/cmdline.o $(OBJS) $(LIB_FLAGS) + @$(CC) $(CFLAGS) -o $@ tests/ctdb_fetch1.o $(OBJS) $(LIB_FLAGS) -bin/ctdb_messaging: $(OBJS) tests/ctdb_messaging.o tests/cmdline.o +bin/ctdb_messaging: $(OBJS) tests/ctdb_messaging.o @echo Linking $@ - @$(CC) $(CFLAGS) -o $@ tests/ctdb_messaging.o tests/cmdline.o $(OBJS) $(LIB_FLAGS) + @$(CC) $(CFLAGS) -o $@ tests/ctdb_messaging.o $(OBJS) $(LIB_FLAGS) bin/ibwrapper_test: $(OBJS) ib/ibwrapper_test.o @echo Linking $@ @$(CC) $(CFLAGS) -o $@ ib/ibwrapper_test.o $(OBJS) $(LIB_FLAGS) +bin/lockwait: $(OBJS) tests/lockwait.o + @echo Linking $@ + @$(CC) $(CFLAGS) -o $@ tests/lockwait.o $(OBJS) $(LIB_FLAGS) + clean: rm -f *.o */*.o */*/*.o rm -f $(BINS) diff --git a/source4/cluster/ctdb/brlock_ctdb.c b/source4/cluster/ctdb/brlock_ctdb.c index b1e8e64d40..2773827b59 100644 --- a/source4/cluster/ctdb/brlock_ctdb.c +++ b/source4/cluster/ctdb/brlock_ctdb.c @@ -389,6 +389,8 @@ static NTSTATUS brl_ctdb_lock(struct brl_context *brl, call.key.dsize = brlh->key.length; call.call_data.dptr = (uint8_t *)&req; call.call_data.dsize = sizeof(req); + call.flags = 0; + call.status = 0; ZERO_STRUCT(req); req.smbpid = smbpid; diff --git a/source4/cluster/ctdb/common/cmdline.c b/source4/cluster/ctdb/common/cmdline.c new file mode 100644 index 0000000000..699cb8fb22 --- /dev/null +++ b/source4/cluster/ctdb/common/cmdline.c @@ -0,0 +1,142 @@ +/* + common commandline code to ctdb test tools + + Copyright (C) Andrew Tridgell 2007 + + 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 "../include/ctdb.h" +#include "../include/ctdb_private.h" + +/* Handle common command line options for ctdb test progs + */ + +static struct { + const char *nlist; + const char *transport; + const char *myaddress; + int self_connect; + const char *db_dir; + int torture; +} ctdb_cmdline = { + .nlist = NULL, + .transport = "tcp", + .myaddress = NULL, + .self_connect = 0, + .db_dir = NULL, + .torture = 0 +}; + + +struct poptOption popt_ctdb_cmdline[] = { + { "nlist", 0, POPT_ARG_STRING, &ctdb_cmdline.nlist, 0, "node list file", "filename" }, + { "listen", 0, POPT_ARG_STRING, &ctdb_cmdline.myaddress, 0, "address to listen on", "address" }, + { "transport", 0, POPT_ARG_STRING, &ctdb_cmdline.transport, 0, "protocol transport", NULL }, + { "self-connect", 0, POPT_ARG_NONE, &ctdb_cmdline.self_connect, 0, "enable self connect", "boolean" }, + { "debug", 'd', POPT_ARG_INT, &LogLevel, 0, "debug level"}, + { "dbdir", 0, POPT_ARG_STRING, &ctdb_cmdline.db_dir, 0, "directory for the tdb files", NULL }, + { "torture", 0, POPT_ARG_NONE, &ctdb_cmdline.torture, 0, "enable nastiness in library", NULL }, + { NULL } +}; + + +/* + startup daemon side of ctdb according to command line options + */ +struct ctdb_context *ctdb_cmdline_init(struct event_context *ev) +{ + struct ctdb_context *ctdb; + int ret; + + if (ctdb_cmdline.nlist == NULL || ctdb_cmdline.myaddress == NULL) { + printf("You must provide a node list with --nlist and an address with --listen\n"); + exit(1); + } + + /* initialise ctdb */ + ctdb = ctdb_init(ev); + if (ctdb == NULL) { + printf("Failed to init ctdb\n"); + exit(1); + } + + if (ctdb_cmdline.self_connect) { + ctdb_set_flags(ctdb, CTDB_FLAG_SELF_CONNECT); + } + if (ctdb_cmdline.torture) { + ctdb_set_flags(ctdb, CTDB_FLAG_TORTURE); + } + + ret = ctdb_set_transport(ctdb, ctdb_cmdline.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, ctdb_cmdline.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, ctdb_cmdline.nlist); + if (ret == -1) { + printf("ctdb_set_nlist failed - %s\n", ctdb_errstr(ctdb)); + exit(1); + } + + ret = ctdb_set_tdb_dir(ctdb, ctdb_cmdline.db_dir); + if (ret == -1) { + printf("ctdb_set_tdb_dir failed - %s\n", ctdb_errstr(ctdb)); + exit(1); + } + + return ctdb; +} + + +/* + startup a client only ctdb context + */ +struct ctdb_context *ctdb_cmdline_client(struct event_context *ev, const char *ctdb_socket) +{ + struct ctdb_context *ctdb; + int ret; + + /* initialise ctdb */ + ctdb = ctdb_init(ev); + if (ctdb == NULL) { + printf("Failed to init ctdb\n"); + exit(1); + } + + ctdb->daemon.name = talloc_strdup(ctdb, ctdb_socket); + + ret = ctdb_socket_connect(ctdb); + if (ret != 0) { + DEBUG(0,(__location__ " Failed to connect to daemon\n")); + talloc_free(ctdb); + return NULL; + } + + return ctdb; +} diff --git a/source4/cluster/ctdb/common/ctdb.c b/source4/cluster/ctdb/common/ctdb.c index 8a8d52f3f1..6bd2fda529 100644 --- a/source4/cluster/ctdb/common/ctdb.c +++ b/source4/cluster/ctdb/common/ctdb.c @@ -74,6 +74,22 @@ void ctdb_set_max_lacount(struct ctdb_context *ctdb, unsigned count) } /* + set the directory for the local databases +*/ +int ctdb_set_tdb_dir(struct ctdb_context *ctdb, const char *dir) +{ + if (dir == NULL) { + ctdb->db_directory = talloc_asprintf(ctdb, "ctdb-%u", ctdb_get_vnn(ctdb)); + } else { + ctdb->db_directory = talloc_strdup(ctdb, dir); + } + if (ctdb->db_directory == NULL) { + return -1; + } + return 0; +} + +/* add a node to the list of active nodes */ static int ctdb_add_node(struct ctdb_context *ctdb, char *nstr) @@ -190,65 +206,102 @@ uint32_t ctdb_get_num_nodes(struct ctdb_context *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) +void ctdb_recv_pkt(struct ctdb_context *ctdb, uint8_t *data, uint32_t length) { - struct ctdb_req_header *hdr; + struct ctdb_req_header *hdr = (struct ctdb_req_header *)data; + TALLOC_CTX *tmp_ctx; + + ctdb->status.node_packets_recv++; + + /* place the packet as a child of the tmp_ctx. We then use + talloc_free() below to free it. If any of the calls want + to keep it, then they will steal it somewhere else, and the + talloc_free() will only free the tmp_ctx */ + tmp_ctx = talloc_new(ctdb); + talloc_steal(tmp_ctx, hdr); if (length < sizeof(*hdr)) { ctdb_set_error(ctdb, "Bad packet length %d\n", length); - return; + goto done; } - 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; + goto done; } if (hdr->ctdb_magic != CTDB_MAGIC) { ctdb_set_error(ctdb, "Non CTDB packet rejected\n"); - return; + goto done; } if (hdr->ctdb_version != CTDB_VERSION) { ctdb_set_error(ctdb, "Bad CTDB version 0x%x rejected\n", hdr->ctdb_version); - return; + goto done; } + DEBUG(3,(__location__ " ctdb request %d of type %d length %d from " + "node %d to %d\n", hdr->reqid, hdr->operation, hdr->length, + hdr->srcnode, hdr->destnode)); + switch (hdr->operation) { case CTDB_REQ_CALL: + ctdb->status.count.req_call++; ctdb_request_call(ctdb, hdr); break; case CTDB_REPLY_CALL: + ctdb->status.count.reply_call++; ctdb_reply_call(ctdb, hdr); break; case CTDB_REPLY_ERROR: + ctdb->status.count.reply_error++; ctdb_reply_error(ctdb, hdr); break; case CTDB_REPLY_REDIRECT: + ctdb->status.count.reply_redirect++; ctdb_reply_redirect(ctdb, hdr); break; case CTDB_REQ_DMASTER: + ctdb->status.count.req_dmaster++; ctdb_request_dmaster(ctdb, hdr); break; case CTDB_REPLY_DMASTER: + ctdb->status.count.reply_dmaster++; ctdb_reply_dmaster(ctdb, hdr); break; case CTDB_REQ_MESSAGE: + ctdb->status.count.req_message++; ctdb_request_message(ctdb, hdr); break; + case CTDB_REQ_FINISHED: + ctdb->status.count.req_finished++; + ctdb_request_finished(ctdb, hdr); + break; + default: - printf("Packet with unknown operation %d\n", hdr->operation); + DEBUG(0,("%s: Packet with unknown operation %d\n", + __location__, hdr->operation)); break; } - talloc_free(hdr); + +done: + talloc_free(tmp_ctx); +} + +/* + called by the transport layer when a packet comes in +*/ +void ctdb_recv_raw_pkt(void *p, uint8_t *data, uint32_t length) +{ + struct ctdb_context *ctdb = talloc_get_type(p, struct ctdb_context); + ctdb_recv_pkt(ctdb, data, length); } /* @@ -257,8 +310,8 @@ static void ctdb_recv_pkt(struct ctdb_context *ctdb, uint8_t *data, uint32_t len 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); + DEBUG(1,("%s: node %s is dead: %d connected\n", + node->ctdb->name, node->name, node->ctdb->num_connected)); } /* @@ -267,8 +320,8 @@ static void ctdb_node_dead(struct ctdb_node *node) 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); + DEBUG(1,("%s: connected to %s - %d connected\n", + node->ctdb->name, node->name, node->ctdb->num_connected)); } /* @@ -281,33 +334,62 @@ void ctdb_daemon_connect_wait(struct ctdb_context *ctdb) expected++; } while (ctdb->num_connected != expected) { + DEBUG(3,("ctdb_connect_wait: waiting for %d nodes (have %d)\n", + expected, ctdb->num_connected)); event_loop_once(ctdb->ev); } + DEBUG(3,("ctdb_connect_wait: got all %d nodes\n", expected)); } +struct queue_next { + struct ctdb_context *ctdb; + struct ctdb_req_header *hdr; +}; + + /* - wait until we're the only node left -*/ -void ctdb_wait_loop(struct ctdb_context *ctdb) + trigered when a deferred packet is due + */ +static void queue_next_trigger(struct event_context *ev, struct timed_event *te, + struct timeval t, void *private_data) { - int expected = 0; - if (ctdb->flags & CTDB_FLAG_SELF_CONNECT) { - expected++; + struct queue_next *q = talloc_get_type(private_data, struct queue_next); + ctdb_recv_pkt(q->ctdb, (uint8_t *)q->hdr, q->hdr->length); + talloc_free(q); +} + +/* + defer a packet, so it is processed on the next event loop + this is used for sending packets to ourselves + */ +static void ctdb_defer_packet(struct ctdb_context *ctdb, struct ctdb_req_header *hdr) +{ + struct queue_next *q; + q = talloc(ctdb, struct queue_next); + if (q == NULL) { + DEBUG(0,(__location__ " Failed to allocate deferred packet\n")); + return; } - while (ctdb->num_connected > expected) { - event_loop_once(ctdb->ev); + q->ctdb = ctdb; + q->hdr = talloc_memdup(ctdb, hdr, hdr->length); + if (q->hdr == NULL) { + DEBUG(0,("Error copying deferred packet to self\n")); + return; } + event_add_timed(ctdb->ev, q, timeval_zero(), queue_next_trigger, q); } - /* queue a packet or die */ void ctdb_queue_packet(struct ctdb_context *ctdb, struct ctdb_req_header *hdr) { struct ctdb_node *node; + ctdb->status.node_packets_sent++; node = ctdb->nodes[hdr->destnode]; - if (ctdb->methods->queue_pkt(node, (uint8_t *)hdr, hdr->length) != 0) { + if (hdr->destnode == ctdb->vnn && !(ctdb->flags & CTDB_FLAG_SELF_CONNECT)) { + ctdb_defer_packet(ctdb, hdr); + } else if (ctdb->methods->queue_pkt(node, (uint8_t *)hdr, hdr->length) != 0) { ctdb_fatal(ctdb, "Unable to queue packet\n"); } } @@ -338,11 +420,3 @@ struct ctdb_context *ctdb_init(struct event_context *ev) return ctdb; } -int ctdb_start(struct ctdb_context *ctdb) -{ - if (ctdb->flags&CTDB_FLAG_DAEMON_MODE) { - return ctdbd_start(ctdb); - } - - return ctdb->methods->start(ctdb); -} diff --git a/source4/cluster/ctdb/common/ctdb_call.c b/source4/cluster/ctdb/common/ctdb_call.c index ab5c2cce3b..76a7e97a87 100644 --- a/source4/cluster/ctdb/common/ctdb_call.c +++ b/source4/cluster/ctdb/common/ctdb_call.c @@ -47,9 +47,9 @@ /* local version of ctdb_call */ -static int ctdb_call_local(struct ctdb_db_context *ctdb_db, struct ctdb_call *call, - struct ctdb_ltdb_header *header, TDB_DATA *data, - uint32_t caller) +int ctdb_call_local(struct ctdb_db_context *ctdb_db, struct ctdb_call *call, + struct ctdb_ltdb_header *header, TDB_DATA *data, + uint32_t caller) { struct ctdb_call_info *c; struct ctdb_registered_call *fn; @@ -105,6 +105,7 @@ static int ctdb_call_local(struct ctdb_db_context *ctdb_db, struct ctdb_call *ca if (c->reply_data) { call->reply_data = *c->reply_data; talloc_steal(ctdb, call->reply_data.dptr); + talloc_set_name_const(call->reply_data.dptr, __location__); } else { call->reply_data.dptr = NULL; call->reply_data.dsize = 0; @@ -140,7 +141,7 @@ static void ctdb_send_error(struct ctdb_context *ctdb, msglen = strlen(msg)+1; len = offsetof(struct ctdb_reply_error, msg); - r = ctdb->methods->allocate_pkt(ctdb, len + msglen); + r = ctdb->methods->allocate_pkt(msg, len + msglen); CTDB_NO_MEMORY_FATAL(ctdb, r); talloc_set_name_const(r, "send_error packet"); @@ -155,11 +156,9 @@ static void ctdb_send_error(struct ctdb_context *ctdb, r->msglen = msglen; memcpy(&r->msg[0], msg, msglen); - talloc_free(msg); - ctdb_queue_packet(ctdb, &r->hdr); - talloc_free(r); + talloc_free(msg); } @@ -223,16 +222,12 @@ static void ctdb_call_send_dmaster(struct ctdb_db_context *ctdb_db, memcpy(&r->data[0], key->dptr, key->dsize); memcpy(&r->data[key->dsize], data->dptr, data->dsize); - if (r->hdr.destnode == ctdb->vnn) { - /* we are the lmaster - don't send to ourselves */ - 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_db, *key, header, *data); - } + /* XXX - probably not necessary when lmaster==dmaster + update the ltdb to record the new dmaster */ + header->dmaster = r->hdr.destnode; + ctdb_ltdb_store(ctdb_db, *key, header, *data); + + ctdb_queue_packet(ctdb, &r->hdr); talloc_free(r); } @@ -252,6 +247,7 @@ void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr struct ctdb_ltdb_header header; struct ctdb_db_context *ctdb_db; int ret, len; + TALLOC_CTX *tmp_ctx; key.dptr = c->data; key.dsize = c->keylen; @@ -267,28 +263,41 @@ void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr } /* fetch the current record */ - ret = ctdb_ltdb_fetch(ctdb_db, key, &header, hdr, &data2); - if (ret != 0) { + ret = ctdb_ltdb_lock_fetch_requeue(ctdb_db, key, &header, hdr, &data2, + ctdb_recv_raw_pkt, ctdb); + if (ret == -1) { ctdb_fatal(ctdb, "ctdb_req_dmaster failed to fetch record"); return; } - + if (ret == -2) { + DEBUG(2,(__location__ " deferring ctdb_request_dmaster\n")); + return; + } + /* its a protocol error if the sending node is not the current dmaster */ - if (header.dmaster != hdr->srcnode) { + if (header.dmaster != hdr->srcnode && + hdr->srcnode != ctdb_lmaster(ctdb_db->ctdb, &key)) { ctdb_fatal(ctdb, "dmaster request from non-master"); return; } - + header.dmaster = c->dmaster; - if (ctdb_ltdb_store(ctdb_db, key, &header, data) != 0) { + ret = ctdb_ltdb_store(ctdb_db, key, &header, data); + ctdb_ltdb_unlock(ctdb_db, key); + if (ret != 0) { ctdb_fatal(ctdb, "ctdb_req_dmaster unable to update dmaster"); return; } + /* put the packet on a temporary context, allowing us to safely free + it below even if ctdb_reply_dmaster() has freed it already */ + tmp_ctx = talloc_new(ctdb); + /* send the CTDB_REPLY_DMASTER */ len = offsetof(struct ctdb_reply_dmaster, data) + data.dsize; - r = ctdb->methods->allocate_pkt(ctdb, len); + r = ctdb->methods->allocate_pkt(tmp_ctx, len); CTDB_NO_MEMORY_FATAL(ctdb, r); + talloc_set_name_const(r, "reply_dmaster packet"); r->hdr.length = len; r->hdr.ctdb_magic = CTDB_MAGIC; @@ -300,13 +309,9 @@ void ctdb_request_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr r->datalen = data.dsize; memcpy(&r->data[0], data.dptr, data.dsize); - if (r->hdr.destnode == r->hdr.srcnode) { - ctdb_reply_dmaster(ctdb, &r->hdr); - } else { - ctdb_queue_packet(ctdb, &r->hdr); - } + ctdb_queue_packet(ctdb, &r->hdr); - talloc_free(r); + talloc_free(tmp_ctx); } @@ -341,17 +346,23 @@ void ctdb_request_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr) 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_db, call.key, &header, hdr, &data); - if (ret != 0) { + ret = ctdb_ltdb_lock_fetch_requeue(ctdb_db, call.key, &header, hdr, &data, + ctdb_recv_raw_pkt, ctdb); + if (ret == -1) { ctdb_send_error(ctdb, hdr, ret, "ltdb fetch failed in ctdb_request_call"); return; } + if (ret == -2) { + DEBUG(2,(__location__ " deferred ctdb_request_call\n")); + 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); + ctdb_ltdb_unlock(ctdb_db, call.key); return; } @@ -364,11 +375,14 @@ void ctdb_request_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr) || c->flags&CTDB_IMMEDIATE_MIGRATION ) { ctdb_call_send_dmaster(ctdb_db, c, &header, &call.key, &data); talloc_free(data.dptr); + ctdb_ltdb_unlock(ctdb_db, call.key); return; } ctdb_call_local(ctdb_db, &call, &header, &data, c->hdr.srcnode); + ctdb_ltdb_unlock(ctdb_db, call.key); + len = offsetof(struct ctdb_reply_call, data) + call.reply_data.dsize; r = ctdb->methods->allocate_pkt(ctdb, len); CTDB_NO_MEMORY_FATAL(ctdb, r); @@ -396,15 +410,18 @@ void ctdb_request_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr) 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 + contains any reply data from 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; - state = idr_find(ctdb->idr, hdr->reqid); - if (state == NULL) return; + state = idr_find_type(ctdb->idr, hdr->reqid, struct ctdb_call_state); + if (state == NULL) { + DEBUG(0, ("reqid %d not found\n", hdr->reqid)); + return; + } state->call.reply_data.dptr = c->data; state->call.reply_data.dsize = c->datalen; @@ -412,10 +429,6 @@ void ctdb_reply_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr) talloc_steal(state, c); - /* get an extra reference here - this prevents the free in ctdb_recv_pkt() - from freeing the data */ - (void)talloc_reference(state, c); - state->state = CTDB_CALL_DONE; if (state->async.fn) { state->async.fn(state); @@ -435,13 +448,25 @@ void ctdb_reply_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr) struct ctdb_call_state *state; struct ctdb_db_context *ctdb_db; TDB_DATA data; + int ret; - state = idr_find(ctdb->idr, hdr->reqid); + state = idr_find_type(ctdb->idr, hdr->reqid, struct ctdb_call_state); if (state == NULL) { return; } + ctdb_db = state->ctdb_db; + ret = ctdb_ltdb_lock_requeue(ctdb_db, state->call.key, hdr, + ctdb_recv_raw_pkt, ctdb); + if (ret == -2) { + return; + } + if (ret != 0) { + DEBUG(0,(__location__ " Failed to get lock in ctdb_reply_dmaster\n")); + return; + } + data.dptr = c->data; data.dsize = c->datalen; @@ -452,12 +477,17 @@ void ctdb_reply_dmaster(struct ctdb_context *ctdb, struct ctdb_req_header *hdr) state->header.dmaster = ctdb->vnn; if (ctdb_ltdb_store(ctdb_db, state->call.key, &state->header, data) != 0) { + ctdb_ltdb_unlock(ctdb_db, state->call.key); ctdb_fatal(ctdb, "ctdb_reply_dmaster store failed\n"); return; } ctdb_call_local(ctdb_db, &state->call, &state->header, &data, ctdb->vnn); + ctdb_ltdb_unlock(ctdb_db, state->call.key); + + talloc_steal(state, state->call.reply_data.dptr); + state->state = CTDB_CALL_DONE; if (state->async.fn) { state->async.fn(state); @@ -473,7 +503,7 @@ 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); + state = idr_find_type(ctdb->idr, hdr->reqid, struct ctdb_call_state); if (state == NULL) return; talloc_steal(state, c); @@ -498,7 +528,7 @@ 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); + state = idr_find_type(ctdb->idr, hdr->reqid, struct ctdb_call_state); if (state == NULL) return; talloc_steal(state, c); @@ -510,6 +540,7 @@ void ctdb_reply_redirect(struct ctdb_context *ctdb, struct ctdb_req_header *hdr) /* send it off again */ state->node = ctdb->nodes[c->dmaster]; + state->c->hdr.destnode = c->dmaster; ctdb_queue_packet(ctdb, &state->c->hdr); } @@ -578,6 +609,7 @@ struct ctdb_call_state *ctdb_call_local_send(struct ctdb_db_context *ctdb_db, state->ctdb_db = ctdb_db; ret = ctdb_call_local(ctdb_db, &state->call, header, data, ctdb->vnn); + talloc_steal(state, state->call.reply_data.dptr); event_add_timed(ctdb->ev, state, timeval_zero(), call_local_trigger, state); @@ -591,45 +623,27 @@ struct ctdb_call_state *ctdb_call_local_send(struct ctdb_db_context *ctdb_db, This constructs a ctdb_call request and queues it for processing. This call never blocks. */ -static struct ctdb_call_state *ctdb_daemon_call_send(struct ctdb_db_context *ctdb_db, - struct ctdb_call *call) +struct ctdb_call_state *ctdb_daemon_call_send_remote(struct ctdb_db_context *ctdb_db, + struct ctdb_call *call, + struct ctdb_ltdb_header *header) { uint32_t len; struct ctdb_call_state *state; - int ret; - struct ctdb_ltdb_header header; - TDB_DATA data; struct ctdb_context *ctdb = ctdb_db->ctdb; - /* - 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_db, call->key, &header, ctdb_db, &data); - if (ret != 0) return NULL; - - if (header.dmaster == ctdb->vnn && !(ctdb->flags & CTDB_FLAG_SELF_CONNECT)) { - return ctdb_call_local_send(ctdb_db, call, &header, &data); - } - state = talloc_zero(ctdb_db, struct ctdb_call_state); CTDB_NO_MEMORY_NULL(ctdb, state); - talloc_steal(state, data.dptr); - len = offsetof(struct ctdb_req_call, data) + call->key.dsize + call->call_data.dsize; - state->c = ctdb->methods->allocate_pkt(ctdb, len); + state->c = ctdb->methods->allocate_pkt(state, len); CTDB_NO_MEMORY_NULL(ctdb, state->c); talloc_set_name_const(state->c, "req_call packet"); - talloc_steal(state, state->c); state->c->hdr.length = len; state->c->hdr.ctdb_magic = CTDB_MAGIC; state->c->hdr.ctdb_version = CTDB_VERSION; state->c->hdr.operation = CTDB_REQ_CALL; - state->c->hdr.destnode = header.dmaster; + 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); @@ -645,9 +659,9 @@ static struct ctdb_call_state *ctdb_daemon_call_send(struct ctdb_db_context *ctd state->call.call_data.dptr = &state->c->data[call->key.dsize]; state->call.key.dptr = &state->c->data[0]; - state->node = ctdb->nodes[header.dmaster]; + state->node = ctdb->nodes[header->dmaster]; state->state = CTDB_CALL_WAIT; - state->header = header; + state->header = *header; state->ctdb_db = ctdb_db; talloc_set_destructor(state, ctdb_call_destructor); @@ -660,29 +674,13 @@ static struct ctdb_call_state *ctdb_daemon_call_send(struct ctdb_db_context *ctd } /* - 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_db_context *ctdb_db, struct ctdb_call *call) -{ - if (ctdb_db->ctdb->flags & CTDB_FLAG_DAEMON_MODE) { - return ctdb_client_call_send(ctdb_db, call); - } - return ctdb_daemon_call_send(ctdb_db, call); -} - -/* make a remote ctdb call - async recv - called in daemon context This is called when the program wants to wait for a ctdb_call to complete and get the results. This call will block unless the call has already completed. */ -static int ctdb_daemon_call_recv(struct ctdb_call_state *state, struct ctdb_call *call) +int ctdb_daemon_call_recv(struct ctdb_call_state *state, struct ctdb_call *call) { - struct ctdb_record_handle *rec; - while (state->state < CTDB_CALL_DONE) { event_loop_once(state->node->ctdb->ev); } @@ -692,16 +690,6 @@ static int ctdb_daemon_call_recv(struct ctdb_call_state *state, struct ctdb_call return -1; } - rec = state->fetch_private; - - /* ugly hack to manage forced migration */ - if (rec != NULL) { - rec->data->dptr = talloc_steal(rec, state->call.reply_data.dptr); - rec->data->dsize = state->call.reply_data.dsize; - talloc_free(state); - return 0; - } - if (state->call.reply_data.dsize) { call->reply_data.dptr = talloc_memdup(state->node->ctdb, state->call.reply_data.dptr, @@ -717,92 +705,3 @@ static int ctdb_daemon_call_recv(struct ctdb_call_state *state, struct ctdb_call } -/* - 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, struct ctdb_call *call) -{ - if (state->ctdb_db->ctdb->flags & CTDB_FLAG_DAEMON_MODE) { - return ctdb_client_call_recv(state, call); - } - return ctdb_daemon_call_recv(state, call); -} - -/* - full ctdb_call. Equivalent to a ctdb_call_send() followed by a ctdb_call_recv() -*/ -int ctdb_call(struct ctdb_db_context *ctdb_db, struct ctdb_call *call) -{ - struct ctdb_call_state *state; - - state = ctdb_call_send(ctdb_db, call); - return ctdb_call_recv(state, call); -} - - - -struct ctdb_record_handle *ctdb_fetch_lock(struct ctdb_db_context *ctdb_db, TALLOC_CTX *mem_ctx, - TDB_DATA key, TDB_DATA *data) -{ - struct ctdb_call call; - struct ctdb_record_handle *rec; - struct ctdb_call_state *state; - int ret; - - if (ctdb_db->ctdb->flags & CTDB_FLAG_DAEMON_MODE) { - return ctdb_client_fetch_lock(ctdb_db, mem_ctx, key, data); - } - - ZERO_STRUCT(call); - call.call_id = CTDB_FETCH_FUNC; - call.key = key; - call.flags = CTDB_IMMEDIATE_MIGRATION; - - rec = talloc(mem_ctx, struct ctdb_record_handle); - CTDB_NO_MEMORY_NULL(ctdb_db->ctdb, rec); - - rec->ctdb_db = ctdb_db; - rec->key = key; - rec->key.dptr = talloc_memdup(rec, key.dptr, key.dsize); - rec->data = data; - - state = ctdb_call_send(ctdb_db, &call); - state->fetch_private = rec; - - ret = ctdb_call_recv(state, &call); - if (ret != 0) { - talloc_free(rec); - return NULL; - } - - return rec; -} - - -int ctdb_store_unlock(struct ctdb_record_handle *rec, TDB_DATA data) -{ - int ret; - struct ctdb_ltdb_header header; - struct ctdb_db_context *ctdb_db = talloc_get_type(rec->ctdb_db, struct ctdb_db_context); - - if (ctdb_db->ctdb->flags & CTDB_FLAG_DAEMON_MODE) { - return ctdb_client_store_unlock(rec, data); - } - - /* should be avoided if possible hang header off rec ? */ - ret = ctdb_ltdb_fetch(rec->ctdb_db, rec->key, &header, NULL, NULL); - if (ret) { - ctdb_set_error(rec->ctdb_db->ctdb, "Fetch of locally held record failed"); - talloc_free(rec); - return ret; - } - - ret = ctdb_ltdb_store(rec->ctdb_db, rec->key, &header, data); - - talloc_free(rec); - - return ret; -} diff --git a/source4/cluster/ctdb/common/ctdb_client.c b/source4/cluster/ctdb/common/ctdb_client.c index 3cb27a1165..dbed8d3585 100644 --- a/source4/cluster/ctdb/common/ctdb_client.c +++ b/source4/cluster/ctdb/common/ctdb_client.c @@ -49,94 +49,87 @@ static void ctdb_reply_connect_wait(struct ctdb_context *ctdb, } /* - called in the client when we receive a CTDB_REPLY_FETCH_LOCK from the daemon + state of a in-progress ctdb call in client +*/ +struct ctdb_client_call_state { + enum call_state state; + uint32_t reqid; + struct ctdb_db_context *ctdb_db; + struct ctdb_call call; +}; + +/* + called when a CTDB_REPLY_CALL packet comes in in the client - This packet comes in response to a CTDB_REQ_FETCH_LOCK request packet. It + This packet comes in response to a CTDB_REQ_CALL request packet. It contains any reply data from the call */ -void ctdb_reply_fetch_lock(struct ctdb_context *ctdb, struct ctdb_req_header *hdr) +static void ctdb_client_reply_call(struct ctdb_context *ctdb, struct ctdb_req_header *hdr) { - struct ctdb_reply_fetch_lock *c = (struct ctdb_reply_fetch_lock *)hdr; - struct ctdb_call_state *state; + struct ctdb_reply_call *c = (struct ctdb_reply_call *)hdr; + struct ctdb_client_call_state *state; - state = idr_find(ctdb->idr, hdr->reqid); - if (state == NULL) return; + state = idr_find_type(ctdb->idr, hdr->reqid, struct ctdb_client_call_state); + if (state == NULL) { + DEBUG(0, ("reqid %d not found\n", hdr->reqid)); + return; + } state->call.reply_data.dptr = c->data; state->call.reply_data.dsize = c->datalen; - state->call.status = c->state; + state->call.status = c->status; talloc_steal(state, c); - /* get an extra reference here - this prevents the free in ctdb_recv_pkt() - from freeing the data */ - (void)talloc_reference(state, c); - state->state = CTDB_CALL_DONE; - if (state->async.fn) { - state->async.fn(state); - } } -/* - called in the client when we receive a CTDB_REPLY_STORE_UNLOCK from the daemon - - This packet comes in response to a CTDB_REQ_STORE_UNLOCK request packet. It - contains any reply data from the call -*/ -void ctdb_reply_store_unlock(struct ctdb_context *ctdb, struct ctdb_req_header *hdr) -{ - struct ctdb_reply_store_unlock *c = (struct ctdb_reply_store_unlock *)hdr; - struct ctdb_call_state *state; - - state = idr_find(ctdb->idr, hdr->reqid); - if (state == NULL) return; - - state->call.status = c->state; - - talloc_steal(state, c); - - /* get an extra reference here - this prevents the free in ctdb_recv_pkt() - from freeing the data */ - (void)talloc_reference(state, c); +static void ctdb_reply_status(struct ctdb_context *ctdb, struct ctdb_req_header *hdr); - state->state = CTDB_CALL_DONE; - if (state->async.fn) { - state->async.fn(state); - } -} /* this is called in the client, when data comes in from the daemon */ static void ctdb_client_read_cb(uint8_t *data, size_t cnt, void *args) { struct ctdb_context *ctdb = talloc_get_type(args, struct ctdb_context); - struct ctdb_req_header *hdr; + struct ctdb_req_header *hdr = (struct ctdb_req_header *)data; + TALLOC_CTX *tmp_ctx; + + /* place the packet as a child of a tmp_ctx. We then use + talloc_free() below to free it. If any of the calls want + to keep it, then they will steal it somewhere else, and the + talloc_free() will be a no-op */ + tmp_ctx = talloc_new(ctdb); + talloc_steal(tmp_ctx, hdr); + + if (cnt == 0) { + DEBUG(2,("Daemon has exited - shutting down client\n")); + exit(0); + } if (cnt < sizeof(*hdr)) { - ctdb_set_error(ctdb, "Bad packet length %d\n", cnt); - return; + DEBUG(0,("Bad packet length %d in client\n", cnt)); + goto done; } - hdr = (struct ctdb_req_header *)data; if (cnt != hdr->length) { - ctdb_set_error(ctdb, "Bad header length %d expected %d\n", + ctdb_set_error(ctdb, "Bad header length %d expected %d in client\n", hdr->length, cnt); - return; + goto done; } if (hdr->ctdb_magic != CTDB_MAGIC) { - ctdb_set_error(ctdb, "Non CTDB packet rejected\n"); - return; + ctdb_set_error(ctdb, "Non CTDB packet rejected in client\n"); + goto done; } if (hdr->ctdb_version != CTDB_VERSION) { - ctdb_set_error(ctdb, "Bad CTDB version 0x%x rejected\n", hdr->ctdb_version); - return; + ctdb_set_error(ctdb, "Bad CTDB version 0x%x rejected in client\n", hdr->ctdb_version); + goto done; } switch (hdr->operation) { case CTDB_REPLY_CALL: - ctdb_reply_call(ctdb, hdr); + ctdb_client_reply_call(ctdb, hdr); break; case CTDB_REQ_MESSAGE: @@ -147,23 +140,22 @@ static void ctdb_client_read_cb(uint8_t *data, size_t cnt, void *args) ctdb_reply_connect_wait(ctdb, hdr); break; - case CTDB_REPLY_FETCH_LOCK: - ctdb_reply_fetch_lock(ctdb, hdr); - break; - - case CTDB_REPLY_STORE_UNLOCK: - ctdb_reply_store_unlock(ctdb, hdr); + case CTDB_REPLY_STATUS: + ctdb_reply_status(ctdb, hdr); break; default: - printf("bogus operation code:%d\n",hdr->operation); + DEBUG(0,("bogus operation code:%d\n",hdr->operation)); } + +done: + talloc_free(tmp_ctx); } /* connect to a unix domain socket */ -static int ux_socket_connect(struct ctdb_context *ctdb) +int ctdb_socket_connect(struct ctdb_context *ctdb) { struct sockaddr_un addr; @@ -189,6 +181,13 @@ static int ux_socket_connect(struct ctdb_context *ctdb) } +struct ctdb_record_handle { + struct ctdb_db_context *ctdb_db; + TDB_DATA key; + TDB_DATA *data; + struct ctdb_ltdb_header header; +}; + /* make a recv call to the local ctdb daemon - called from client context @@ -196,31 +195,19 @@ static int ux_socket_connect(struct ctdb_context *ctdb) 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_client_call_recv(struct ctdb_call_state *state, struct ctdb_call *call) +int ctdb_call_recv(struct ctdb_client_call_state *state, struct ctdb_call *call) { - struct ctdb_record_handle *rec; - while (state->state < CTDB_CALL_DONE) { - event_loop_once(state->node->ctdb->ev); + event_loop_once(state->ctdb_db->ctdb->ev); } if (state->state != CTDB_CALL_DONE) { - ctdb_set_error(state->node->ctdb, "%s", state->errmsg); + DEBUG(0,(__location__ " ctdb_call_recv failed\n")); talloc_free(state); return -1; } - rec = state->fetch_private; - - /* ugly hack to manage forced migration */ - if (rec != NULL) { - rec->data->dptr = talloc_steal(rec, state->call.reply_data.dptr); - rec->data->dsize = state->call.reply_data.dsize; - talloc_free(state); - return 0; - } - if (state->call.reply_data.dsize) { - call->reply_data.dptr = talloc_memdup(state->node->ctdb, + call->reply_data.dptr = talloc_memdup(state->ctdb_db, state->call.reply_data.dptr, state->call.reply_data.dsize); call->reply_data.dsize = state->call.reply_data.dsize; @@ -240,13 +227,41 @@ int ctdb_client_call_recv(struct ctdb_call_state *state, struct ctdb_call *call) /* destroy a ctdb_call in client */ -static int ctdb_client_call_destructor(struct ctdb_call_state *state) +static int ctdb_client_call_destructor(struct ctdb_client_call_state *state) { - idr_remove(state->node->ctdb->idr, state->c->hdr.reqid); + idr_remove(state->ctdb_db->ctdb->idr, state->reqid); return 0; } +/* + construct an event driven local ctdb_call + + this is used so that locally processed ctdb_call requests are processed + in an event driven manner +*/ +static struct ctdb_client_call_state *ctdb_client_call_local_send(struct ctdb_db_context *ctdb_db, + struct ctdb_call *call, + struct ctdb_ltdb_header *header, + TDB_DATA *data) +{ + struct ctdb_client_call_state *state; + struct ctdb_context *ctdb = ctdb_db->ctdb; + int ret; + + state = talloc_zero(ctdb_db, struct ctdb_client_call_state); + CTDB_NO_MEMORY_NULL(ctdb, state); + talloc_steal(state, data->dptr); + + state->state = CTDB_CALL_DONE; + state->call = *call; + state->ctdb_db = ctdb_db; + + ret = ctdb_call_local(ctdb_db, &state->call, header, data, ctdb->vnn); + talloc_steal(state, state->call.reply_data.dptr); + + return state; +} /* make a ctdb call to the local daemon - async send. Called from client context. @@ -254,107 +269,109 @@ static int ctdb_client_call_destructor(struct ctdb_call_state *state) This constructs a ctdb_call request and queues it for processing. This call never blocks. */ -struct ctdb_call_state *ctdb_client_call_send(struct ctdb_db_context *ctdb_db, - struct ctdb_call *call) +struct ctdb_client_call_state *ctdb_call_send(struct ctdb_db_context *ctdb_db, + struct ctdb_call *call) { - struct ctdb_call_state *state; + struct ctdb_client_call_state *state; struct ctdb_context *ctdb = ctdb_db->ctdb; struct ctdb_ltdb_header header; TDB_DATA data; int ret; size_t len; + struct ctdb_req_call *c; /* if the domain socket is not yet open, open it */ if (ctdb->daemon.sd==-1) { - ux_socket_connect(ctdb); + ctdb_socket_connect(ctdb); } ret = ctdb_ltdb_lock(ctdb_db, call->key); if (ret != 0) { - printf("failed to lock ltdb record\n"); + DEBUG(0,(__location__ " Failed to get chainlock\n")); return NULL; } ret = ctdb_ltdb_fetch(ctdb_db, call->key, &header, ctdb_db, &data); if (ret != 0) { ctdb_ltdb_unlock(ctdb_db, call->key); + DEBUG(0,(__location__ " Failed to fetch record\n")); return NULL; } -#if 0 if (header.dmaster == ctdb->vnn && !(ctdb->flags & CTDB_FLAG_SELF_CONNECT)) { - state = ctdb_call_local_send(ctdb_db, call, &header, &data); + state = ctdb_client_call_local_send(ctdb_db, call, &header, &data); + talloc_free(data.dptr); ctdb_ltdb_unlock(ctdb_db, call->key); return state; } -#endif - state = talloc_zero(ctdb_db, struct ctdb_call_state); + ctdb_ltdb_unlock(ctdb_db, call->key); + talloc_free(data.dptr); + + state = talloc_zero(ctdb_db, struct ctdb_client_call_state); if (state == NULL) { - printf("failed to allocate state\n"); - ctdb_ltdb_unlock(ctdb_db, call->key); + DEBUG(0, (__location__ " failed to allocate state\n")); return NULL; } - talloc_steal(state, data.dptr); - len = offsetof(struct ctdb_req_call, data) + call->key.dsize + call->call_data.dsize; - state->c = ctdbd_allocate_pkt(ctdb, len); - if (state->c == NULL) { - printf("failed to allocate packet\n"); - ctdb_ltdb_unlock(ctdb_db, call->key); + c = ctdbd_allocate_pkt(state, len); + if (c == NULL) { + DEBUG(0, (__location__ " failed to allocate packet\n")); return NULL; } - talloc_set_name_const(state->c, "ctdbd req_call packet"); - talloc_steal(state, state->c); + talloc_set_name_const(c, "ctdb client req_call packet"); + memset(c, 0, offsetof(struct ctdb_req_call, data)); - state->c->hdr.length = len; - state->c->hdr.ctdb_magic = CTDB_MAGIC; - state->c->hdr.ctdb_version = CTDB_VERSION; - state->c->hdr.operation = CTDB_REQ_CALL; - state->c->hdr.destnode = header.dmaster; - state->c->hdr.srcnode = ctdb->vnn; + c->hdr.length = len; + c->hdr.ctdb_magic = CTDB_MAGIC; + c->hdr.ctdb_version = CTDB_VERSION; + c->hdr.operation = CTDB_REQ_CALL; /* this limits us to 16k outstanding messages - not unreasonable */ - state->c->hdr.reqid = idr_get_new(ctdb->idr, state, 0xFFFF); - state->c->flags = call->flags; - state->c->db_id = ctdb_db->db_id; - state->c->callid = call->call_id; - state->c->keylen = call->key.dsize; - state->c->calldatalen = call->call_data.dsize; - memcpy(&state->c->data[0], call->key.dptr, call->key.dsize); - memcpy(&state->c->data[call->key.dsize], + c->hdr.reqid = idr_get_new(ctdb->idr, state, 0xFFFF); + c->flags = call->flags; + c->db_id = ctdb_db->db_id; + c->callid = call->call_id; + c->keylen = call->key.dsize; + c->calldatalen = call->call_data.dsize; + memcpy(&c->data[0], call->key.dptr, call->key.dsize); + memcpy(&c->data[call->key.dsize], call->call_data.dptr, call->call_data.dsize); state->call = *call; - state->call.call_data.dptr = &state->c->data[call->key.dsize]; - state->call.key.dptr = &state->c->data[0]; + state->call.call_data.dptr = &c->data[call->key.dsize]; + state->call.key.dptr = &c->data[0]; - state->node = ctdb->nodes[header.dmaster]; state->state = CTDB_CALL_WAIT; - state->header = header; state->ctdb_db = ctdb_db; + state->reqid = c->hdr.reqid; talloc_set_destructor(state, ctdb_client_call_destructor); - ctdb_client_queue_pkt(ctdb, &state->c->hdr); - -/*XXX set up timeout to cleanup if server doesnt respond - event_add_timed(ctdb->ev, state, timeval_current_ofs(CTDB_REQ_TIMEOUT, 0), - ctdb_call_timeout, state); -*/ + ctdb_client_queue_pkt(ctdb, &c->hdr); - ctdb_ltdb_unlock(ctdb_db, call->key); return state; } +/* + full ctdb_call. Equivalent to a ctdb_call_send() followed by a ctdb_call_recv() +*/ +int ctdb_call(struct ctdb_db_context *ctdb_db, struct ctdb_call *call) +{ + struct ctdb_client_call_state *state; + + state = ctdb_call_send(ctdb_db, call); + return ctdb_call_recv(state, call); +} + /* tell the daemon what messaging srvid we will use, and register the message handler function in the client */ -int ctdb_client_set_message_handler(struct ctdb_context *ctdb, uint32_t srvid, - ctdb_message_fn_t handler, - void *private_data) +int ctdb_set_message_handler(struct ctdb_context *ctdb, uint32_t srvid, + ctdb_message_fn_t handler, + void *private_data) { struct ctdb_req_register c; @@ -362,7 +379,7 @@ int ctdb_client_set_message_handler(struct ctdb_context *ctdb, uint32_t srvid, /* if the domain socket is not yet open, open it */ if (ctdb->daemon.sd==-1) { - ux_socket_connect(ctdb); + ctdb_socket_connect(ctdb); } ZERO_STRUCT(c); @@ -383,26 +400,10 @@ int ctdb_client_set_message_handler(struct ctdb_context *ctdb, uint32_t srvid, } - -/* - setup handler for receipt of ctdb messages from ctdb_send_message() -*/ -int ctdb_set_message_handler(struct ctdb_context *ctdb, - uint32_t srvid, - ctdb_message_fn_t handler, - void *private_data) -{ - if (ctdb->flags & CTDB_FLAG_DAEMON_MODE) { - return ctdb_client_set_message_handler(ctdb, srvid, handler, private_data); - } - return ctdb_daemon_set_message_handler(ctdb, srvid, handler, private_data); -} - - /* send a message - from client context */ -int ctdb_client_send_message(struct ctdb_context *ctdb, uint32_t vnn, +int ctdb_send_message(struct ctdb_context *ctdb, uint32_t vnn, uint32_t srvid, TDB_DATA data) { struct ctdb_req_message *r; @@ -436,7 +437,7 @@ int ctdb_client_send_message(struct ctdb_context *ctdb, uint32_t vnn, /* wait for all nodes to be connected - from client */ -static void ctdb_client_connect_wait(struct ctdb_context *ctdb) +void ctdb_connect_wait(struct ctdb_context *ctdb) { struct ctdb_req_connect_wait r; int res; @@ -447,216 +448,235 @@ static void ctdb_client_connect_wait(struct ctdb_context *ctdb) r.hdr.ctdb_magic = CTDB_MAGIC; r.hdr.ctdb_version = CTDB_VERSION; r.hdr.operation = CTDB_REQ_CONNECT_WAIT; + + DEBUG(3,("ctdb_connect_wait: sending to ctdbd\n")); + + /* if the domain socket is not yet open, open it */ + if (ctdb->daemon.sd==-1) { + ctdb_socket_connect(ctdb); + } res = ctdb_queue_send(ctdb->daemon.queue, (uint8_t *)&r.hdr, r.hdr.length); if (res != 0) { - printf("Failed to queue a connect wait request\n"); + DEBUG(0,(__location__ " Failed to queue a connect wait request\n")); return; } + DEBUG(3,("ctdb_connect_wait: waiting\n")); + /* now we can go into the normal wait routine, as the reply packet will update the ctdb->num_connected variable */ ctdb_daemon_connect_wait(ctdb); } /* - wait for all nodes to be connected -*/ -void ctdb_connect_wait(struct ctdb_context *ctdb) + cancel a ctdb_fetch_lock operation, releasing the lock + */ +static int fetch_lock_destructor(struct ctdb_record_handle *h) { - if (!(ctdb->flags & CTDB_FLAG_DAEMON_MODE)) { - ctdb_daemon_connect_wait(ctdb); - return; - } - - ctdb_client_connect_wait(ctdb); + ctdb_ltdb_unlock(h->ctdb_db, h->key); + return 0; } +/* + force the migration of a record to this node + */ +static int ctdb_client_force_migration(struct ctdb_db_context *ctdb_db, TDB_DATA key) +{ + struct ctdb_call call; + ZERO_STRUCT(call); + call.call_id = CTDB_NULL_FUNC; + call.key = key; + call.flags = CTDB_IMMEDIATE_MIGRATION; + return ctdb_call(ctdb_db, &call); +} -struct ctdb_call_state *ctdb_client_fetch_lock_send(struct ctdb_db_context *ctdb_db, - TALLOC_CTX *mem_ctx, - TDB_DATA key) +/* + get a lock on a record, and return the records data. Blocks until it gets the lock + */ +struct ctdb_record_handle *ctdb_fetch_lock(struct ctdb_db_context *ctdb_db, TALLOC_CTX *mem_ctx, + TDB_DATA key, TDB_DATA *data) { - struct ctdb_call_state *state; - struct ctdb_context *ctdb = ctdb_db->ctdb; - struct ctdb_req_fetch_lock *req; - int len, res; + int ret; + struct ctdb_record_handle *h; - /* if the domain socket is not yet open, open it */ - if (ctdb->daemon.sd==-1) { - ux_socket_connect(ctdb); - } + /* + procedure is as follows: - state = talloc_zero(ctdb_db, struct ctdb_call_state); - if (state == NULL) { - printf("failed to allocate state\n"); - return NULL; - } - state->state = CTDB_CALL_WAIT; - state->ctdb_db = ctdb_db; - len = offsetof(struct ctdb_req_fetch_lock, key) + key.dsize; - state->c = ctdbd_allocate_pkt(ctdb, len); - if (state->c == NULL) { - printf("failed to allocate packet\n"); + 1) get the chain lock. + 2) check if we are dmaster + 3) if we are the dmaster then return handle + 4) if not dmaster then ask ctdb daemon to make us dmaster, and wait for + reply from ctdbd + 5) when we get the reply, goto (1) + */ + + h = talloc_zero(mem_ctx, struct ctdb_record_handle); + if (h == NULL) { return NULL; } - ZERO_STRUCT(*state->c); - talloc_set_name_const(state->c, "ctdbd req_fetch_lock packet"); - talloc_steal(state, state->c); - - req = (struct ctdb_req_fetch_lock *)state->c; - 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 = idr_get_new(ctdb->idr, state, 0xFFFF); - req->db_id = ctdb_db->db_id; - req->keylen = key.dsize; - memcpy(&req->key[0], key.dptr, key.dsize); - - res = ctdb_client_queue_pkt(ctdb, &req->hdr); - if (res != 0) { + + h->ctdb_db = ctdb_db; + h->key = key; + h->key.dptr = talloc_memdup(h, key.dptr, key.dsize); + if (h->key.dptr == NULL) { + talloc_free(h); return NULL; } + h->data = data; - talloc_free(req); - - return state; -} + DEBUG(3,("ctdb_fetch_lock: key=%*.*s\n", key.dsize, key.dsize, + (const char *)key.dptr)); +again: + /* step 1 - get the chain lock */ + ret = ctdb_ltdb_lock(ctdb_db, key); + if (ret != 0) { + DEBUG(0, (__location__ " failed to lock ltdb record\n")); + talloc_free(h); + return NULL; + } -struct ctdb_call_state *ctdb_client_store_unlock_send( - struct ctdb_record_handle *rh, - TALLOC_CTX *mem_ctx, - TDB_DATA data) -{ - struct ctdb_call_state *state; - struct ctdb_db_context *ctdb_db = talloc_get_type(rh->ctdb_db, struct ctdb_db_context); - struct ctdb_context *ctdb = ctdb_db->ctdb; - struct ctdb_req_store_unlock *req; - int len, res; + DEBUG(4,("ctdb_fetch_lock: got chain lock\n")); - /* if the domain socket is not yet open, open it */ - if (ctdb->daemon.sd==-1) { - ux_socket_connect(ctdb); - } + talloc_set_destructor(h, fetch_lock_destructor); - state = talloc_zero(ctdb_db, struct ctdb_call_state); - if (state == NULL) { - printf("failed to allocate state\n"); - return NULL; - } - state->state = CTDB_CALL_WAIT; - state->ctdb_db = ctdb_db; - len = offsetof(struct ctdb_req_store_unlock, data) + rh->key.dsize + data.dsize; - state->c = ctdbd_allocate_pkt(ctdb, len); - if (state->c == NULL) { - printf("failed to allocate packet\n"); + ret = ctdb_ltdb_fetch(ctdb_db, key, &h->header, h, data); + if (ret != 0) { + ctdb_ltdb_unlock(ctdb_db, key); + talloc_free(h); return NULL; } - ZERO_STRUCT(*state->c); - talloc_set_name_const(state->c, "ctdbd req_store_unlock packet"); - talloc_steal(state, state->c); - - req = (struct ctdb_req_store_unlock *)state->c; - 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 = idr_get_new(ctdb->idr, state, 0xFFFF); - req->db_id = ctdb_db->db_id; - req->keylen = rh->key.dsize; - req->datalen = data.dsize; - memcpy(&req->data[0], rh->key.dptr, rh->key.dsize); - memcpy(&req->data[req->keylen], data.dptr, data.dsize); - - res = ctdb_client_queue_pkt(ctdb, &req->hdr); - if (res != 0) { - return NULL; + + /* when torturing, ensure we test the remote path */ + if ((ctdb_db->ctdb->flags & CTDB_FLAG_TORTURE) && + random() % 5 == 0) { + h->header.dmaster = (uint32_t)-1; } - talloc_free(req); - return state; + DEBUG(4,("ctdb_fetch_lock: done local fetch\n")); + + if (h->header.dmaster != ctdb_db->ctdb->vnn) { + ctdb_ltdb_unlock(ctdb_db, key); + ret = ctdb_client_force_migration(ctdb_db, key); + if (ret != 0) { + DEBUG(4,("ctdb_fetch_lock: force_migration failed\n")); + talloc_free(h); + return NULL; + } + goto again; + } + + DEBUG(4,("ctdb_fetch_lock: we are dmaster - done\n")); + return h; } /* - make a recv call to the local ctdb daemon - called from client context + store some data to the record that was locked with ctdb_fetch_lock() +*/ +int ctdb_record_store(struct ctdb_record_handle *h, TDB_DATA data) +{ + return ctdb_ltdb_store(h->ctdb_db, h->key, &h->header, data); +} - This is called when the program wants to wait for a ctdb_fetch_lock to complete and get the - results. This call will block unless the call has already completed. +/* + wait until we're the only node left. + this function never returns */ -struct ctdb_record_handle *ctdb_client_fetch_lock_recv(struct ctdb_call_state *state, TALLOC_CTX *mem_ctx, TDB_DATA key, TDB_DATA *data) +void ctdb_shutdown(struct ctdb_context *ctdb) { - struct ctdb_record_handle *rec; + struct ctdb_req_shutdown r; + int len; - while (state->state < CTDB_CALL_DONE) { - event_loop_once(state->ctdb_db->ctdb->ev); - } - if (state->state != CTDB_CALL_DONE) { - ctdb_set_error(state->node->ctdb, "%s", state->errmsg); - talloc_free(state); - return NULL; + /* if the domain socket is not yet open, open it */ + if (ctdb->daemon.sd==-1) { + ctdb_socket_connect(ctdb); } - rec = talloc(mem_ctx, struct ctdb_record_handle); - CTDB_NO_MEMORY_NULL(state->ctdb_db->ctdb, rec); + len = sizeof(struct ctdb_req_shutdown); + ZERO_STRUCT(r); + r.hdr.length = len; + r.hdr.ctdb_magic = CTDB_MAGIC; + r.hdr.ctdb_version = CTDB_VERSION; + r.hdr.operation = CTDB_REQ_SHUTDOWN; + r.hdr.reqid = 0; - rec->ctdb_db = state->ctdb_db; - rec->key = key; - rec->key.dptr = talloc_memdup(rec, key.dptr, key.dsize); - rec->data = talloc(rec, TDB_DATA); - rec->data->dsize = state->call.reply_data.dsize; - rec->data->dptr = talloc_memdup(rec, state->call.reply_data.dptr, rec->data->dsize); + ctdb_client_queue_pkt(ctdb, &(r.hdr)); - if (data) { - *data = *rec->data; + /* this event loop will terminate once we receive the reply */ + while (1) { + event_loop_once(ctdb->ev); } - return rec; } -/* - make a recv call to the local ctdb daemon - called from client context +enum ctdb_status_states {CTDB_STATUS_WAIT, CTDB_STATUS_DONE}; - This is called when the program wants to wait for a ctdb_store_unlock to complete and get the - results. This call will block unless the call has already completed. -*/ -int ctdb_client_store_unlock_recv(struct ctdb_call_state *state, struct ctdb_record_handle *rec) +struct ctdb_status_state { + uint32_t reqid; + struct ctdb_status *status; + enum ctdb_status_states state; +}; + +/* + handle a ctdb_reply_status reply + */ +static void ctdb_reply_status(struct ctdb_context *ctdb, struct ctdb_req_header *hdr) { - while (state->state < CTDB_CALL_DONE) { - event_loop_once(state->ctdb_db->ctdb->ev); - } - if (state->state != CTDB_CALL_DONE) { - ctdb_set_error(state->node->ctdb, "%s", state->errmsg); + struct ctdb_reply_status *r = (struct ctdb_reply_status *)hdr; + struct ctdb_status_state *state; + + state = idr_find_type(ctdb->idr, hdr->reqid, struct ctdb_status_state); + if (state == NULL) { + DEBUG(0, ("reqid %d not found\n", hdr->reqid)); + return; } - talloc_free(state); - return state->state; + *state->status = r->status; + state->state = CTDB_STATUS_DONE; } -struct ctdb_record_handle *ctdb_client_fetch_lock(struct ctdb_db_context *ctdb_db, - TALLOC_CTX *mem_ctx, - TDB_DATA key, - TDB_DATA *data) +/* + wait until we're the only node left. + this function never returns +*/ +int ctdb_status(struct ctdb_context *ctdb, struct ctdb_status *status) { - struct ctdb_call_state *state; - struct ctdb_record_handle *rec; + struct ctdb_req_status r; + int ret; + struct ctdb_status_state *state; - state = ctdb_client_fetch_lock_send(ctdb_db, mem_ctx, key); - rec = ctdb_client_fetch_lock_recv(state, mem_ctx, key, data); + /* if the domain socket is not yet open, open it */ + if (ctdb->daemon.sd==-1) { + ctdb_socket_connect(ctdb); + } - return rec; -} + state = talloc(ctdb, struct ctdb_status_state); + CTDB_NO_MEMORY(ctdb, state); -int ctdb_client_store_unlock(struct ctdb_record_handle *rec, TDB_DATA data) -{ - struct ctdb_call_state *state; - int res; + state->reqid = idr_get_new(ctdb->idr, state, 0xFFFF); + state->status = status; + state->state = CTDB_STATUS_WAIT; + + ZERO_STRUCT(r); + r.hdr.length = sizeof(r); + r.hdr.ctdb_magic = CTDB_MAGIC; + r.hdr.ctdb_version = CTDB_VERSION; + r.hdr.operation = CTDB_REQ_STATUS; + r.hdr.reqid = state->reqid; - state = ctdb_client_store_unlock_send(rec, rec, data); - res = ctdb_client_store_unlock_recv(state, rec); + ret = ctdb_client_queue_pkt(ctdb, &(r.hdr)); + if (ret != 0) { + talloc_free(state); + return -1; + } + + while (state->state == CTDB_STATUS_WAIT) { + event_loop_once(ctdb->ev); + } - talloc_free(rec); + talloc_free(state); - return res; + return 0; } + diff --git a/source4/cluster/ctdb/common/ctdb_daemon.c b/source4/cluster/ctdb/common/ctdb_daemon.c index 945030d77e..ff3431a392 100644 --- a/source4/cluster/ctdb/common/ctdb_daemon.c +++ b/source4/cluster/ctdb/common/ctdb_daemon.c @@ -25,9 +25,23 @@ #include "lib/util/dlinklist.h" #include "system/network.h" #include "system/filesys.h" +#include "system/wait.h" #include "../include/ctdb.h" #include "../include/ctdb_private.h" +/* + structure describing a connected client in the daemon + */ +struct ctdb_client { + struct ctdb_context *ctdb; + int fd; + struct ctdb_queue *queue; +}; + + + +static void daemon_incoming_packet(void *, uint8_t *, uint32_t ); + static void ctdb_main_loop(struct ctdb_context *ctdb) { ctdb->methods->start(ctdb); @@ -35,7 +49,7 @@ static void ctdb_main_loop(struct ctdb_context *ctdb) /* go into a wait loop to allow other nodes to complete */ event_loop_wait(ctdb->ev); - printf("event_loop_wait() returned. this should not happen\n"); + DEBUG(0,("event_loop_wait() returned. this should not happen\n")); exit(1); } @@ -47,16 +61,27 @@ static void set_non_blocking(int fd) fcntl(fd, F_SETFL, v | O_NONBLOCK); } +static void block_signal(int signum) +{ + struct sigaction act; + + memset(&act, 0, sizeof(act)); + + act.sa_handler = SIG_IGN; + sigemptyset(&act.sa_mask); + sigaddset(&act.sa_mask, signum); + sigaction(signum, &act, NULL); +} + /* - structure describing a connected client in the daemon + send a packet to a client */ -struct ctdb_client { - struct ctdb_context *ctdb; - int fd; - struct ctdb_queue *queue; -}; - +static int daemon_queue_send(struct ctdb_client *client, struct ctdb_req_header *hdr) +{ + client->ctdb->status.client_packets_sent++; + return ctdb_queue_send(client->queue, (uint8_t *)hdr, hdr->length); +} /* message handler for when we are in daemon mode. This redirects the message @@ -73,10 +98,9 @@ static void daemon_message_handler(struct ctdb_context *ctdb, uint32_t srvid, len = offsetof(struct ctdb_req_message, data) + data.dsize; r = ctdbd_allocate_pkt(ctdb, len); -/*XXX cant use this since it returns an int CTDB_NO_MEMORY(ctdb, r);*/ talloc_set_name_const(r, "req_message packet"); - ZERO_STRUCT(*r); + memset(r, 0, offsetof(struct ctdb_req_message, data)); r->hdr.length = len; r->hdr.ctdb_magic = CTDB_MAGIC; @@ -85,11 +109,10 @@ static void daemon_message_handler(struct ctdb_context *ctdb, uint32_t srvid, r->srvid = srvid; r->datalen = data.dsize; memcpy(&r->data[0], data.dptr, data.dsize); - - ctdb_queue_send(client->queue, (uint8_t *)&r->hdr, len); + + daemon_queue_send(client, &r->hdr); talloc_free(r); - return; } @@ -105,187 +128,122 @@ static void daemon_request_register_message_handler(struct ctdb_client *client, c->srvid, daemon_message_handler, client); if (res != 0) { - printf("Failed to register handler %u in daemon\n", c->srvid); + DEBUG(0,(__location__ " Failed to register handler %u in daemon\n", + c->srvid)); + } else { + DEBUG(2,(__location__ " Registered message handler for srvid=%u\n", + c->srvid)); } } -static struct ctdb_call_state *ctdb_fetch_lock_send(struct ctdb_db_context *ctdb_db, - TALLOC_CTX *mem_ctx, - TDB_DATA key, TDB_DATA *data) +/* + called when the daemon gets a shutdown request from a client + */ +static void daemon_request_shutdown(struct ctdb_client *client, + struct ctdb_req_shutdown *f) { - struct ctdb_call *call; - struct ctdb_record_handle *rec; - struct ctdb_call_state *state; + struct ctdb_context *ctdb = talloc_get_type(client->ctdb, struct ctdb_context); + int len; + uint32_t node; - rec = talloc(mem_ctx, struct ctdb_record_handle); - CTDB_NO_MEMORY_NULL(ctdb_db->ctdb, rec); + /* we dont send to ourself so we can already count one daemon as + exiting */ + ctdb->num_finished++; - - call = talloc(rec, struct ctdb_call); - ZERO_STRUCT(*call); - call->call_id = CTDB_FETCH_FUNC; - call->key = key; - call->flags = CTDB_IMMEDIATE_MIGRATION; + /* loop over all nodes of the cluster */ + for (node=0; node<ctdb->num_nodes;node++) { + struct ctdb_req_finished *rf; - rec->ctdb_db = ctdb_db; - rec->key = key; - rec->key.dptr = talloc_memdup(rec, key.dptr, key.dsize); - rec->data = data; + /* dont send a message to ourself */ + if (ctdb->vnn == node) { + continue; + } - state = ctdb_call_send(ctdb_db, call); - state->fetch_private = rec; + len = sizeof(struct ctdb_req_finished); + rf = ctdb->methods->allocate_pkt(ctdb, len); + CTDB_NO_MEMORY_FATAL(ctdb, rf); + talloc_set_name_const(rf, "ctdb_req_finished packet"); - return state; -} + ZERO_STRUCT(*rf); + rf->hdr.length = len; + rf->hdr.ctdb_magic = CTDB_MAGIC; + rf->hdr.ctdb_version = CTDB_VERSION; + rf->hdr.operation = CTDB_REQ_FINISHED; + rf->hdr.destnode = node; + rf->hdr.srcnode = ctdb->vnn; + rf->hdr.reqid = 0; -struct client_fetch_lock_data { - struct ctdb_client *client; - uint32_t reqid; -}; -static void daemon_fetch_lock_complete(struct ctdb_call_state *state) -{ - struct ctdb_reply_fetch_lock *r; - struct client_fetch_lock_data *data = talloc_get_type(state->async.private_data, struct client_fetch_lock_data); - struct ctdb_client *client = talloc_get_type(data->client, struct ctdb_client); - int length, res; + ctdb_queue_packet(ctdb, &(rf->hdr)); - length = offsetof(struct ctdb_reply_fetch_lock, data) + state->call.reply_data.dsize; - r = ctdbd_allocate_pkt(client->ctdb, length); - if (r == NULL) { - printf("Failed to allocate reply_call in ctdb daemon\n"); - return; + talloc_free(rf); } - ZERO_STRUCT(*r); - r->hdr.length = length; - r->hdr.ctdb_magic = CTDB_MAGIC; - r->hdr.ctdb_version = CTDB_VERSION; - r->hdr.operation = CTDB_REPLY_FETCH_LOCK; - r->hdr.reqid = data->reqid; - r->state = state->state; - r->datalen = state->call.reply_data.dsize; - memcpy(&r->data[0], state->call.reply_data.dptr, r->datalen); - res = ctdb_queue_send(client->queue, (uint8_t *)&r->hdr, r->hdr.length); - if (res != 0) { - printf("Failed to queue packet from daemon to client\n"); + /* wait until all nodes have are prepared to shutdown */ + while (ctdb->num_finished != ctdb->num_nodes) { + event_loop_once(ctdb->ev); } - talloc_free(r); -} -/* - called when the daemon gets a fetch lock request from a client - */ -static void daemon_request_fetch_lock(struct ctdb_client *client, - struct ctdb_req_fetch_lock *f) -{ - struct ctdb_call_state *state; - TDB_DATA key, *data; - struct ctdb_db_context *ctdb_db; - struct client_fetch_lock_data *fl_data; - - ctdb_db = find_ctdb_db(client->ctdb, f->db_id); - - key.dsize = f->keylen; - key.dptr = &f->key[0]; - - data = talloc(client, TDB_DATA); - data->dptr = NULL; - data->dsize = 0; + /* all daemons have requested to finish - we now exit */ + DEBUG(1,("All daemons finished - exiting\n")); + _exit(0); +} - state = ctdb_fetch_lock_send(ctdb_db, client, key, data); - talloc_steal(state, data); - fl_data = talloc(state, struct client_fetch_lock_data); - fl_data->client = client; - fl_data->reqid = f->hdr.reqid; - state->async.fn = daemon_fetch_lock_complete; - state->async.private_data = fl_data; -} /* - called when the daemon gets a store unlock request from a client - - this would never block? + called when the daemon gets a connect wait request from a client */ -static void daemon_request_store_unlock(struct ctdb_client *client, - struct ctdb_req_store_unlock *f) +static void daemon_request_connect_wait(struct ctdb_client *client, + struct ctdb_req_connect_wait *c) { - struct ctdb_db_context *ctdb_db; - struct ctdb_reply_store_unlock r; - uint32_t caller = ctdb_get_vnn(client->ctdb); - struct ctdb_ltdb_header header; - TDB_DATA key, data; + struct ctdb_reply_connect_wait r; int res; - ctdb_db = find_ctdb_db(client->ctdb, f->db_id); - - /* write the data to ltdb */ - key.dsize = f->keylen; - key.dptr = &f->data[0]; - res = ctdb_ltdb_fetch(ctdb_db, key, &header, NULL, NULL); - if (res) { - ctdb_set_error(ctdb_db->ctdb, "Fetch of locally held record failed"); - res = -1; - goto done; - } - if (header.laccessor != caller) { - header.lacount = 0; - } - header.laccessor = caller; - header.lacount++; - data.dsize = f->datalen; - data.dptr = &f->data[f->keylen]; - res = ctdb_ltdb_store(ctdb_db, key, &header, data); - if ( res != 0) { - ctdb_set_error(ctdb_db->ctdb, "ctdb_call tdb_store failed\n"); - } - + /* first wait - in the daemon */ + ctdb_daemon_connect_wait(client->ctdb); -done: /* now send the reply */ ZERO_STRUCT(r); r.hdr.length = sizeof(r); r.hdr.ctdb_magic = CTDB_MAGIC; r.hdr.ctdb_version = CTDB_VERSION; - r.hdr.operation = CTDB_REPLY_STORE_UNLOCK; - r.hdr.reqid = f->hdr.reqid; - r.state = res; + r.hdr.operation = CTDB_REPLY_CONNECT_WAIT; + r.vnn = ctdb_get_vnn(client->ctdb); + r.num_connected = client->ctdb->num_connected; - res = ctdb_queue_send(client->queue, (uint8_t *)&r.hdr, r.hdr.length); + res = daemon_queue_send(client, &r.hdr); if (res != 0) { - printf("Failed to queue a store unlock response\n"); + DEBUG(0,(__location__ " Failed to queue a connect wait response\n")); return; } } + /* - called when the daemon gets a connect wait request from a client + called when the daemon gets a status request from a client */ -static void daemon_request_connect_wait(struct ctdb_client *client, - struct ctdb_req_connect_wait *c) +static void daemon_request_status(struct ctdb_client *client, + struct ctdb_req_status *c) { - struct ctdb_reply_connect_wait r; + struct ctdb_reply_status r; int res; - /* first wait - in the daemon */ - ctdb_daemon_connect_wait(client->ctdb); - /* now send the reply */ ZERO_STRUCT(r); r.hdr.length = sizeof(r); r.hdr.ctdb_magic = CTDB_MAGIC; r.hdr.ctdb_version = CTDB_VERSION; - r.hdr.operation = CTDB_REPLY_CONNECT_WAIT; - r.vnn = ctdb_get_vnn(client->ctdb); - r.num_connected = client->ctdb->num_connected; + r.hdr.operation = CTDB_REPLY_STATUS; + r.hdr.reqid = c->hdr.reqid; + r.status = client->ctdb->status; - res = ctdb_queue_send(client->queue, (uint8_t *)&r.hdr, r.hdr.length); + res = daemon_queue_send(client, &r.hdr); if (res != 0) { - printf("Failed to queue a connect wait response\n"); + DEBUG(0,(__location__ " Failed to queue a connect wait response\n")); return; } } @@ -323,11 +281,69 @@ static void daemon_request_message_from_client(struct ctdb_client *client, res = ctdb_daemon_send_message(client->ctdb, c->hdr.destnode, c->srvid, data); if (res != 0) { - printf("Failed to send message to remote node %u\n", - c->hdr.destnode); + DEBUG(0,(__location__ " Failed to send message to remote node %u\n", + c->hdr.destnode)); + } +} + + +struct daemon_call_state { + struct ctdb_client *client; + uint32_t reqid; + struct ctdb_call *call; + struct timeval start_time; +}; + +/* + complete a call from a client +*/ +static void daemon_call_from_client_callback(struct ctdb_call_state *state) +{ + struct daemon_call_state *dstate = talloc_get_type(state->async.private_data, + struct daemon_call_state); + struct ctdb_reply_call *r; + int res; + uint32_t length; + struct ctdb_client *client = dstate->client; + + talloc_steal(client, dstate); + talloc_steal(dstate, dstate->call); + + res = ctdb_daemon_call_recv(state, dstate->call); + if (res != 0) { + DEBUG(0, (__location__ " ctdbd_call_recv() returned error\n")); + client->ctdb->status.pending_calls--; + ctdb_latency(&client->ctdb->status.max_call_latency, dstate->start_time); + return; } + + length = offsetof(struct ctdb_reply_call, data) + dstate->call->reply_data.dsize; + r = ctdbd_allocate_pkt(dstate, length); + if (r == NULL) { + DEBUG(0, (__location__ " Failed to allocate reply_call in ctdb daemon\n")); + client->ctdb->status.pending_calls--; + ctdb_latency(&client->ctdb->status.max_call_latency, dstate->start_time); + return; + } + memset(r, 0, offsetof(struct ctdb_reply_call, data)); + r->hdr.length = length; + r->hdr.ctdb_magic = CTDB_MAGIC; + r->hdr.ctdb_version = CTDB_VERSION; + r->hdr.operation = CTDB_REPLY_CALL; + r->hdr.reqid = dstate->reqid; + r->datalen = dstate->call->reply_data.dsize; + memcpy(&r->data[0], dstate->call->reply_data.dptr, r->datalen); + + res = daemon_queue_send(client, &r->hdr); + if (res != 0) { + DEBUG(0, (__location__ "Failed to queue packet from daemon to client\n")); + } + ctdb_latency(&client->ctdb->status.max_call_latency, dstate->start_time); + talloc_free(dstate); + client->ctdb->status.pending_calls--; } + /* this is called when the ctdb daemon received a ctdb request call from a local client over the unix domain socket @@ -337,103 +353,159 @@ static void daemon_request_call_from_client(struct ctdb_client *client, { struct ctdb_call_state *state; struct ctdb_db_context *ctdb_db; - struct ctdb_call call; - struct ctdb_reply_call *r; - int res; - uint32_t length; + struct daemon_call_state *dstate; + struct ctdb_call *call; + struct ctdb_ltdb_header header; + TDB_DATA key, data; + int ret; + struct ctdb_context *ctdb = client->ctdb; + + ctdb->status.total_calls++; + ctdb->status.pending_calls++; ctdb_db = find_ctdb_db(client->ctdb, c->db_id); if (!ctdb_db) { - printf("Unknown database in request. db_id==0x%08x",c->db_id); + DEBUG(0, (__location__ " Unknown database in request. db_id==0x%08x", + c->db_id)); + ctdb->status.pending_calls--; return; } - ZERO_STRUCT(call); - call.call_id = c->callid; - call.key.dptr = c->data; - call.key.dsize = c->keylen; - call.call_data.dptr = c->data + c->keylen; - call.call_data.dsize = c->calldatalen; + key.dptr = c->data; + key.dsize = c->keylen; - state = ctdb_call_send(ctdb_db, &call); + ret = ctdb_ltdb_lock_fetch_requeue(ctdb_db, key, &header, + (struct ctdb_req_header *)c, &data, + daemon_incoming_packet, client); + if (ret == -2) { + /* will retry later */ + ctdb->status.pending_calls--; + return; + } -/* XXX this must be converted to fully async */ - res = ctdb_call_recv(state, &call); - if (res != 0) { - printf("ctdbd_call_recv() returned error\n"); - exit(1); + if (ret != 0) { + DEBUG(0,(__location__ " Unable to fetch record\n")); + ctdb->status.pending_calls--; + return; } - length = offsetof(struct ctdb_reply_call, data) + call.reply_data.dsize; - r = ctdbd_allocate_pkt(client->ctdb, length); - if (r == NULL) { - printf("Failed to allocate reply_call in ctdb daemon\n"); + dstate = talloc(client, struct daemon_call_state); + if (dstate == NULL) { + ctdb_ltdb_unlock(ctdb_db, key); + DEBUG(0,(__location__ " Unable to allocate dstate\n")); + ctdb->status.pending_calls--; + return; + } + dstate->start_time = timeval_current(); + dstate->client = client; + dstate->reqid = c->hdr.reqid; + talloc_steal(dstate, data.dptr); + + call = dstate->call = talloc_zero(dstate, struct ctdb_call); + if (call == NULL) { + ctdb_ltdb_unlock(ctdb_db, key); + DEBUG(0,(__location__ " Unable to allocate call\n")); + ctdb->status.pending_calls--; + ctdb_latency(&ctdb->status.max_call_latency, dstate->start_time); return; } - ZERO_STRUCT(*r); - r->hdr.length = length; - r->hdr.ctdb_magic = CTDB_MAGIC; - r->hdr.ctdb_version = CTDB_VERSION; - r->hdr.operation = CTDB_REPLY_CALL; - r->hdr.reqid = c->hdr.reqid; - r->datalen = call.reply_data.dsize; - memcpy(&r->data[0], call.reply_data.dptr, r->datalen); - res = ctdb_queue_send(client->queue, (uint8_t *)&r, r->hdr.length); - if (res != 0) { - printf("Failed to queue packet from daemon to client\n"); + call->call_id = c->callid; + call->key = key; + call->call_data.dptr = c->data + c->keylen; + call->call_data.dsize = c->calldatalen; + call->flags = c->flags; + + if (header.dmaster == ctdb->vnn && !(ctdb->flags & CTDB_FLAG_SELF_CONNECT)) { + state = ctdb_call_local_send(ctdb_db, call, &header, &data); + } else { + state = ctdb_daemon_call_send_remote(ctdb_db, call, &header); } - talloc_free(r); -} + ctdb_ltdb_unlock(ctdb_db, key); + + if (state == NULL) { + DEBUG(0,(__location__ " Unable to setup call send\n")); + ctdb->status.pending_calls--; + ctdb_latency(&ctdb->status.max_call_latency, dstate->start_time); + return; + } + talloc_steal(state, dstate); + talloc_steal(client, state); + + state->async.fn = daemon_call_from_client_callback; + state->async.private_data = dstate; +} /* data contains a packet from the client */ -static void client_incoming_packet(struct ctdb_client *client, void *data, size_t nread) +static void daemon_incoming_packet(void *p, uint8_t *data, uint32_t nread) { - struct ctdb_req_header *hdr = data; + struct ctdb_req_header *hdr = (struct ctdb_req_header *)data; + struct ctdb_client *client = talloc_get_type(p, struct ctdb_client); + TALLOC_CTX *tmp_ctx; + struct ctdb_context *ctdb = client->ctdb; + + /* place the packet as a child of a tmp_ctx. We then use + talloc_free() below to free it. If any of the calls want + to keep it, then they will steal it somewhere else, and the + talloc_free() will be a no-op */ + tmp_ctx = talloc_new(client); + talloc_steal(tmp_ctx, hdr); if (hdr->ctdb_magic != CTDB_MAGIC) { - ctdb_set_error(client->ctdb, "Non CTDB packet rejected\n"); + ctdb_set_error(client->ctdb, "Non CTDB packet rejected in daemon\n"); goto done; } if (hdr->ctdb_version != CTDB_VERSION) { - ctdb_set_error(client->ctdb, "Bad CTDB version 0x%x rejected\n", hdr->ctdb_version); + ctdb_set_error(client->ctdb, "Bad CTDB version 0x%x rejected in daemon\n", hdr->ctdb_version); goto done; } switch (hdr->operation) { case CTDB_REQ_CALL: + ctdb->status.client.req_call++; daemon_request_call_from_client(client, (struct ctdb_req_call *)hdr); break; case CTDB_REQ_REGISTER: + ctdb->status.client.req_register++; daemon_request_register_message_handler(client, (struct ctdb_req_register *)hdr); break; case CTDB_REQ_MESSAGE: + ctdb->status.client.req_message++; daemon_request_message_from_client(client, (struct ctdb_req_message *)hdr); break; case CTDB_REQ_CONNECT_WAIT: + ctdb->status.client.req_connect_wait++; daemon_request_connect_wait(client, (struct ctdb_req_connect_wait *)hdr); break; - case CTDB_REQ_FETCH_LOCK: - daemon_request_fetch_lock(client, (struct ctdb_req_fetch_lock *)hdr); + + case CTDB_REQ_SHUTDOWN: + ctdb->status.client.req_shutdown++; + daemon_request_shutdown(client, (struct ctdb_req_shutdown *)hdr); break; - case CTDB_REQ_STORE_UNLOCK: - daemon_request_store_unlock(client, (struct ctdb_req_store_unlock *)hdr); + + case CTDB_REQ_STATUS: + ctdb->status.client.req_status++; + daemon_request_status(client, (struct ctdb_req_status *)hdr); break; + default: - printf("daemon: unrecognized operation:%d\n",hdr->operation); + DEBUG(0,(__location__ " daemon: unrecognized operation %d\n", + hdr->operation)); } done: - talloc_free(data); + talloc_free(tmp_ctx); } - -static void ctdb_client_read_cb(uint8_t *data, size_t cnt, void *args) +/* + called when the daemon gets a incoming packet + */ +static void ctdb_daemon_read_cb(uint8_t *data, size_t cnt, void *args) { struct ctdb_client *client = talloc_get_type(args, struct ctdb_client); struct ctdb_req_header *hdr; @@ -443,13 +515,15 @@ static void ctdb_client_read_cb(uint8_t *data, size_t cnt, void *args) return; } + client->ctdb->status.client_packets_recv++; + if (cnt < sizeof(*hdr)) { - ctdb_set_error(client->ctdb, "Bad packet length %d\n", cnt); + ctdb_set_error(client->ctdb, "Bad packet length %d in daemon\n", cnt); return; } hdr = (struct ctdb_req_header *)data; if (cnt != hdr->length) { - ctdb_set_error(client->ctdb, "Bad header length %d expected %d\n", + ctdb_set_error(client->ctdb, "Bad header length %d expected %d\n in daemon", hdr->length, cnt); return; } @@ -460,12 +534,16 @@ static void ctdb_client_read_cb(uint8_t *data, size_t cnt, void *args) } if (hdr->ctdb_version != CTDB_VERSION) { - ctdb_set_error(client->ctdb, "Bad CTDB version 0x%x rejected\n", hdr->ctdb_version); + ctdb_set_error(client->ctdb, "Bad CTDB version 0x%x rejected in daemon\n", hdr->ctdb_version); return; } + DEBUG(3,(__location__ " client request %d of type %d length %d from " + "node %d to %d\n", hdr->reqid, hdr->operation, hdr->length, + hdr->srcnode, hdr->destnode)); + /* it is the responsibility of the incoming packet function to free 'data' */ - client_incoming_packet(client, data, cnt); + daemon_incoming_packet(client, data, cnt); } static void ctdb_accept_client(struct event_context *ev, struct fd_event *fde, @@ -490,7 +568,7 @@ static void ctdb_accept_client(struct event_context *ev, struct fd_event *fde, client->fd = fd; client->queue = ctdb_queue_setup(ctdb, client, fd, CTDB_DS_ALIGNMENT, - ctdb_client_read_cb, client); + ctdb_daemon_read_cb, client); talloc_set_destructor(client, ctdb_client_destructor); } @@ -507,10 +585,10 @@ static void ctdb_read_from_parent(struct event_context *ev, struct fd_event *fde /* XXX this is a good place to try doing some cleaning up before exiting */ cnt = read(*fd, &buf, 1); if (cnt==0) { - printf("parent process exited. filedescriptor dissappeared\n"); + DEBUG(2,(__location__ " parent process exited. filedescriptor dissappeared\n")); exit(1); } else { - printf("ctdb: did not expect data from parent process\n"); + DEBUG(0,(__location__ " ctdb: did not expect data from parent process\n")); exit(1); } } @@ -559,7 +637,7 @@ static int unlink_destructor(const char *name) /* start the protocol going */ -int ctdbd_start(struct ctdb_context *ctdb) +int ctdb_start(struct ctdb_context *ctdb) { pid_t pid; static int fd[2]; @@ -575,18 +653,18 @@ int ctdbd_start(struct ctdb_context *ctdb) /* create a unix domain stream socket to listen to */ res = ux_socket_bind(ctdb); if (res!=0) { - printf("Failed to open CTDB unix domain socket\n"); + DEBUG(0,(__location__ " Failed to open CTDB unix domain socket\n")); exit(10); } res = pipe(&fd[0]); if (res) { - printf("Failed to open pipe for CTDB\n"); + DEBUG(0,(__location__ " Failed to open pipe for CTDB\n")); exit(1); } pid = fork(); if (pid==-1) { - printf("Failed to fork CTDB daemon\n"); + DEBUG(0,(__location__ " Failed to fork CTDB daemon\n")); exit(1); } @@ -597,12 +675,14 @@ int ctdbd_start(struct ctdb_context *ctdb) return 0; } + block_signal(SIGPIPE); + /* ensure the socket is deleted on exit of the daemon */ domain_socket_name = talloc_strdup(talloc_autofree_context(), ctdb->daemon.name); talloc_set_destructor(domain_socket_name, unlink_destructor); close(fd[1]); - ctdb_clear_flags(ctdb, CTDB_FLAG_DAEMON_MODE); + ctdb->ev = event_context_init(NULL); fde = event_add_fd(ctdb->ev, ctdb, fd[0], EVENT_FD_READ, ctdb_read_from_parent, &fd[0]); fde = event_add_fd(ctdb->ev, ctdb, ctdb->daemon.sd, EVENT_FD_READ, ctdb_accept_client, ctdb); @@ -614,18 +694,18 @@ int ctdbd_start(struct ctdb_context *ctdb) /* allocate a packet for use in client<->daemon communication */ -void *ctdbd_allocate_pkt(struct ctdb_context *ctdb, size_t len) +void *ctdbd_allocate_pkt(TALLOC_CTX *mem_ctx, size_t len) { int size; size = (len+(CTDB_DS_ALIGNMENT-1)) & ~(CTDB_DS_ALIGNMENT-1); - return talloc_size(ctdb, size); + return talloc_size(mem_ctx, size); } -int ctdb_daemon_set_message_handler(struct ctdb_context *ctdb, uint32_t srvid, - ctdb_message_fn_t handler, - void *private_data) +/* + called when a CTDB_REQ_FINISHED packet comes in +*/ +void ctdb_request_finished(struct ctdb_context *ctdb, struct ctdb_req_header *hdr) { - return ctdb_register_message_handler(ctdb, ctdb, srvid, handler, private_data); + ctdb->num_finished++; } - diff --git a/source4/cluster/ctdb/common/ctdb_lockwait.c b/source4/cluster/ctdb/common/ctdb_lockwait.c new file mode 100644 index 0000000000..36b08796be --- /dev/null +++ b/source4/cluster/ctdb/common/ctdb_lockwait.c @@ -0,0 +1,139 @@ +/* + wait for a tdb chain lock + + 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 "system/wait.h" +#include "popt.h" +#include "db_wrap.h" +#include "lib/tdb/include/tdb.h" +#include "../include/ctdb_private.h" + + +struct lockwait_handle { + struct ctdb_context *ctdb; + struct fd_event *fde; + int fd[2]; + pid_t child; + void *private_data; + void (*callback)(void *); + struct timeval start_time; +}; + +static void lockwait_handler(struct event_context *ev, struct fd_event *fde, + uint16_t flags, void *private_data) +{ + struct lockwait_handle *h = talloc_get_type(private_data, + struct lockwait_handle); + void (*callback)(void *) = h->callback; + void *p = h->private_data; + pid_t child = h->child; + talloc_set_destructor(h, NULL); + close(h->fd[0]); + ctdb_latency(&h->ctdb->status.max_lockwait_latency, h->start_time); + h->ctdb->status.pending_lockwait_calls--; + talloc_free(h); + callback(p); + waitpid(child, NULL, 0); +} + +static int lockwait_destructor(struct lockwait_handle *h) +{ + h->ctdb->status.pending_lockwait_calls--; + close(h->fd[0]); + kill(h->child, SIGKILL); + waitpid(h->child, NULL, 0); + return 0; +} + +/* + setup a non-blocking chainlock on a tdb record. If this function + returns NULL then it could not get the chainlock. Otherwise it + returns a opaque handle, and will call callback() once it has + managed to get the chainlock. You can cancel it by using talloc_free + on the returned handle. + + It is the callers responsibility to unlock the chainlock once + acquired + */ +struct lockwait_handle *ctdb_lockwait(struct ctdb_db_context *ctdb_db, + TDB_DATA key, + void (*callback)(void *private_data), + void *private_data) +{ + struct lockwait_handle *result; + int ret; + + ctdb_db->ctdb->status.lockwait_calls++; + ctdb_db->ctdb->status.pending_lockwait_calls++; + + if (!(result = talloc_zero(ctdb_db, struct lockwait_handle))) { + ctdb_db->ctdb->status.pending_lockwait_calls--; + return NULL; + } + + ret = pipe(result->fd); + + if (ret != 0) { + talloc_free(result); + ctdb_db->ctdb->status.pending_lockwait_calls--; + return NULL; + } + + result->child = fork(); + + if (result->child == (pid_t)-1) { + close(result->fd[0]); + close(result->fd[1]); + talloc_free(result); + ctdb_db->ctdb->status.pending_lockwait_calls--; + return NULL; + } + + result->callback = callback; + result->private_data = private_data; + result->ctdb = ctdb_db->ctdb; + + if (result->child == 0) { + close(result->fd[0]); + /* + * Do we need a tdb_reopen here? + */ + tdb_chainlock(ctdb_db->ltdb->tdb, key); + _exit(0); + } + + close(result->fd[1]); + talloc_set_destructor(result, lockwait_destructor); + + result->fde = event_add_fd(ctdb_db->ctdb->ev, result, result->fd[0], + EVENT_FD_READ, lockwait_handler, + (void *)result); + if (result->fde == NULL) { + talloc_free(result); + ctdb_db->ctdb->status.pending_lockwait_calls--; + return NULL; + } + + result->start_time = timeval_current(); + + return result; +} diff --git a/source4/cluster/ctdb/common/ctdb_ltdb.c b/source4/cluster/ctdb/common/ctdb_ltdb.c index 785ccad9b3..cb07a72375 100644 --- a/source4/cluster/ctdb/common/ctdb_ltdb.c +++ b/source4/cluster/ctdb/common/ctdb_ltdb.c @@ -45,9 +45,8 @@ struct ctdb_db_context *ctdb_db_handle(struct ctdb_context *ctdb, const char *na /* this is the dummy null procedure that all databases support */ -static int ctdb_fetch_func(struct ctdb_call_info *call) +static int ctdb_null_func(struct ctdb_call_info *call) { - call->reply_data = &call->record_data; return 0; } @@ -82,10 +81,21 @@ struct ctdb_db_context *ctdb_attach(struct ctdb_context *ctdb, const char *name, } } + if (mkdir(ctdb->db_directory, 0700) == -1 && errno != EEXIST) { + DEBUG(0,(__location__ " Unable to create ctdb directory '%s'\n", + ctdb->db_directory)); + talloc_free(ctdb_db); + return NULL; + } + + /* add the node id to the database name, so when we run on loopback + we don't conflict in the local filesystem */ + name = talloc_asprintf(ctdb_db, "%s/%s", ctdb->db_directory, name); + /* 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_db->ltdb = tdb_wrap_open(ctdb, name, 0, TDB_INTERNAL, open_flags, mode); + ctdb_db->ltdb = tdb_wrap_open(ctdb, name, 0, TDB_CLEAR_IF_FIRST, open_flags, mode); if (ctdb_db->ltdb == NULL) { ctdb_set_error(ctdb, "Failed to open tdb %s\n", name); talloc_free(ctdb_db); @@ -94,9 +104,10 @@ struct ctdb_db_context *ctdb_attach(struct ctdb_context *ctdb, const char *name, /* - all databases support the "fetch" function. we need this in order to do forced migration of records + all databases support the "null" function. we need this in + order to do forced migration of records */ - ret = ctdb_set_call(ctdb_db, ctdb_fetch_func, CTDB_FETCH_FUNC); + ret = ctdb_set_call(ctdb_db, ctdb_null_func, CTDB_NULL_FUNC); if (ret != 0) { talloc_free(ctdb_db); return NULL; @@ -145,13 +156,15 @@ int ctdb_ltdb_fetch(struct ctdb_db_context *ctdb_db, rec = tdb_fetch(ctdb_db->ltdb->tdb, key); if (rec.dsize < sizeof(*header)) { + TDB_DATA d2; /* return an initial header */ - free(rec.dptr); + if (rec.dptr) free(rec.dptr); ltdb_initial_header(ctdb_db, key, header); + ZERO_STRUCT(d2); if (data) { - data->dptr = NULL; - data->dsize = 0; + *data = d2; } + ctdb_ltdb_store(ctdb_db, key, header, d2); return 0; } @@ -215,3 +228,118 @@ int ctdb_ltdb_unlock(struct ctdb_db_context *ctdb_db, TDB_DATA key) return tdb_chainunlock(ctdb_db->ltdb->tdb, key); } +struct lock_fetch_state { + struct ctdb_context *ctdb; + void (*recv_pkt)(void *, uint8_t *, uint32_t); + void *recv_context; + struct ctdb_req_header *hdr; +}; + +/* + called when we should retry the operation + */ +static void lock_fetch_callback(void *p) +{ + struct lock_fetch_state *state = talloc_get_type(p, struct lock_fetch_state); + state->recv_pkt(state->recv_context, (uint8_t *)state->hdr, state->hdr->length); + talloc_free(state); + DEBUG(2,(__location__ " PACKET REQUEUED\n")); +} + + +/* + do a non-blocking ltdb_lock, deferring this ctdb request until we + have the chainlock + + It does the following: + + 1) tries to get the chainlock. If it succeeds, then it returns 0 + + 2) if it fails to get a chainlock immediately then it sets up a + non-blocking chainlock via ctdb_lockwait, and when it gets the + chainlock it re-submits this ctdb request to the main packet + receive function + + This effectively queues all ctdb requests that cannot be + immediately satisfied until it can get the lock. This means that + the main ctdb daemon will not block waiting for a chainlock held by + a client + + There are 3 possible return values: + + 0: means that it got the lock immediately. + -1: means that it failed to get the lock, and won't retry + -2: means that it failed to get the lock immediately, but will retry + */ +int ctdb_ltdb_lock_requeue(struct ctdb_db_context *ctdb_db, + TDB_DATA key, struct ctdb_req_header *hdr, + void (*recv_pkt)(void *, uint8_t *, uint32_t ), + void *recv_context) +{ + int ret; + struct tdb_context *tdb = ctdb_db->ltdb->tdb; + struct lockwait_handle *h; + struct lock_fetch_state *state; + + ret = tdb_chainlock_nonblock(tdb, key); + + if (ret != 0 && + !(errno == EACCES || errno == EAGAIN || errno == EDEADLK)) { + /* a hard failure - don't try again */ + return -1; + } + + /* when torturing, ensure we test the contended path */ + if ((ctdb_db->ctdb->flags & CTDB_FLAG_TORTURE) && + random() % 5 == 0) { + ret = -1; + tdb_chainunlock(tdb, key); + } + + /* first the non-contended path */ + if (ret == 0) { + return 0; + } + + state = talloc(ctdb_db, struct lock_fetch_state); + state->ctdb = ctdb_db->ctdb; + state->hdr = hdr; + state->recv_pkt = recv_pkt; + state->recv_context = recv_context; + + /* now the contended path */ + h = ctdb_lockwait(ctdb_db, key, lock_fetch_callback, state); + if (h == NULL) { + tdb_chainunlock(tdb, key); + return -1; + } + + /* we need to move the packet off the temporary context in ctdb_recv_pkt(), + so it won't be freed yet */ + talloc_steal(state, hdr); + talloc_steal(state, h); + + /* now tell the caller than we will retry asynchronously */ + return -2; +} + +/* + a varient of ctdb_ltdb_lock_requeue that also fetches the record + */ +int ctdb_ltdb_lock_fetch_requeue(struct ctdb_db_context *ctdb_db, + TDB_DATA key, struct ctdb_ltdb_header *header, + struct ctdb_req_header *hdr, TDB_DATA *data, + void (*recv_pkt)(void *, uint8_t *, uint32_t ), + void *recv_context) +{ + int ret; + + ret = ctdb_ltdb_lock_requeue(ctdb_db, key, hdr, recv_pkt, recv_context); + if (ret == 0) { + ret = ctdb_ltdb_fetch(ctdb_db, key, header, hdr, data); + if (ret != 0) { + ctdb_ltdb_unlock(ctdb_db, key); + } + } + return ret; +} diff --git a/source4/cluster/ctdb/common/ctdb_message.c b/source4/cluster/ctdb/common/ctdb_message.c index ad88ec22d2..70fcf00c4d 100644 --- a/source4/cluster/ctdb/common/ctdb_message.c +++ b/source4/cluster/ctdb/common/ctdb_message.c @@ -42,7 +42,8 @@ static int ctdb_dispatch_message(struct ctdb_context *ctdb, uint32_t srvid, TDB_ if (ml->srvid == srvid || ml->srvid == CTDB_SRVID_ALL) break; } if (ml == NULL) { - printf("daemon vnn:%d no msg handler for srvid=%u\n", ctdb_get_vnn(ctdb), srvid); + DEBUG(1,(__location__ " daemon vnn:%d no msg handler for srvid=%u\n", + ctdb_get_vnn(ctdb), srvid)); /* no registered message handler */ return -1; } @@ -86,7 +87,7 @@ static void ctdb_local_message_trigger(struct event_context *ev, struct timed_ev res = ctdb_dispatch_message(m->ctdb, m->srvid, m->data); if (res != 0) { - printf("Failed to dispatch message for srvid=%u\n", m->srvid); + DEBUG(0, (__location__ " Failed to dispatch message for srvid=%u\n", m->srvid)); } talloc_free(m); } @@ -147,18 +148,6 @@ int ctdb_daemon_send_message(struct ctdb_context *ctdb, uint32_t vnn, return 0; } -/* - send a ctdb message -*/ -int ctdb_send_message(struct ctdb_context *ctdb, uint32_t vnn, - uint32_t srvid, TDB_DATA data) -{ - if (ctdb->flags & CTDB_FLAG_DAEMON_MODE) { - return ctdb_client_send_message(ctdb, vnn, srvid, data); - } - return ctdb_daemon_send_message(ctdb, vnn, srvid, data); -} - /* when a client goes away, we need to remove its srvid handler from the list diff --git a/source4/cluster/ctdb/common/ctdb_util.c b/source4/cluster/ctdb/common/ctdb_util.c index cf0c72a58b..9a5e51bfa0 100644 --- a/source4/cluster/ctdb/common/ctdb_util.c +++ b/source4/cluster/ctdb/common/ctdb_util.c @@ -25,6 +25,8 @@ #include "system/filesys.h" #include "../include/ctdb_private.h" +int LogLevel; + /* return error string for last error */ @@ -100,3 +102,29 @@ uint32_t ctdb_hash(const TDB_DATA *key) return (1103515243 * value + 12345); } + +/* + a type checking varient of idr_find + */ +void *_idr_find_type(struct idr_context *idp, int id, const char *type, const char *location) +{ + void *p = idr_find(idp, id); + if (p && talloc_check_name(p, type) == NULL) { + DEBUG(0,("%s idr_find_type expected type %s but got %s\n", + location, type, talloc_get_name(p))); + return NULL; + } + return p; +} + + +/* + update a max latency number + */ +void ctdb_latency(double *latency, struct timeval t) +{ + double l = timeval_elapsed(&t); + if (l > *latency) { + *latency = l; + } +} diff --git a/source4/cluster/ctdb/config.mk b/source4/cluster/ctdb/config.mk index 0e0629bfb1..076562297f 100644 --- a/source4/cluster/ctdb/config.mk +++ b/source4/cluster/ctdb/config.mk @@ -24,6 +24,7 @@ OBJ_FILES = \ common/ctdb_util.o \ common/ctdb_io.o \ common/ctdb_client.o \ - common/ctdb_daemon.o + common/ctdb_daemon.o \ + common/ctdb_lockwait.o PUBLIC_DEPENDENCIES = LIBTDB LIBTALLOC PRIVATE_DEPENDENCIES = ctdb_tcp diff --git a/source4/cluster/ctdb/ctdb_cluster.c b/source4/cluster/ctdb/ctdb_cluster.c index c9233696e8..e3e60685b0 100644 --- a/source4/cluster/ctdb/ctdb_cluster.c +++ b/source4/cluster/ctdb/ctdb_cluster.c @@ -104,12 +104,30 @@ static void *ctdb_backend_handle(struct cluster_ops *ops) return (void *)state->ctdb; } +struct ctdb_handler_state { + struct cluster_state *state; + cluster_message_fn_t handler; + struct messaging_context *msg; +}; + /* dispatch incoming ctdb messages */ static void ctdb_message_handler(struct ctdb_context *ctdb, uint32_t srvid, TDB_DATA data, void *private) { + struct ctdb_handler_state *s = talloc_get_type(private, + struct ctdb_handler_state); + DATA_BLOB blob; + blob.data = data.dptr; + blob.length = data.dsize; + s->handler(s->msg, blob); +} + +static int ctdb_handler_destructor(struct ctdb_handler_state *s) +{ + /* XXX - tell ctdb to de-register the message handler */ + return 0; } /* @@ -120,7 +138,28 @@ static NTSTATUS ctdb_message_init(struct cluster_ops *ops, struct server_id server, cluster_message_fn_t handler) { - /* nothing to do - we're now using the wildcard message handler */ + struct cluster_state *state = ops->private; + struct ctdb_handler_state *h; + int ret; + + h = talloc(msg, struct ctdb_handler_state); + NT_STATUS_HAVE_NO_MEMORY(h); + + h->state = state; + h->handler = handler; + h->msg = msg; + + talloc_set_destructor(h, ctdb_handler_destructor); + + /* setup a message handler */ + ret = ctdb_set_message_handler(state->ctdb, server.id, + ctdb_message_handler, h); + if (ret == -1) { + DEBUG(0,("ctdb_set_message_handler failed - %s\n", + ctdb_errstr(state->ctdb))); + exit(1); + } + return NT_STATUS_OK; } @@ -198,11 +237,6 @@ void cluster_ctdb_init(struct event_context *ev, const char *model) ctdb_set_flags(state->ctdb, CTDB_FLAG_SELF_CONNECT); } - if (strcmp(model, "single") != 0) { - DEBUG(0,("Enabling ctdb daemon mode\n")); - ctdb_set_flags(state->ctdb, CTDB_FLAG_DAEMON_MODE); - } - lacount = lp_parm_int(-1, "ctdb", "maxlacount", -1); if (lacount != -1) { ctdb_set_max_lacount(state->ctdb, lacount); @@ -215,6 +249,12 @@ void cluster_ctdb_init(struct event_context *ev, const char *model) goto failed; } + ret = ctdb_set_tdb_dir(state->ctdb, lp_lockdir()); + if (ret == -1) { + DEBUG(0,("ctdb_set_tdb_dir 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) { @@ -230,14 +270,13 @@ void cluster_ctdb_init(struct event_context *ev, const char *model) if (ctdb_db == NULL) goto failed; } - /* setup a global message handler */ - ret = ctdb_set_message_handler(state->ctdb, CTDB_SRVID_ALL, - ctdb_message_handler, state); - if (ret == -1) { - DEBUG(0,("ctdb_set_message_handler failed - %s\n", - ctdb_errstr(state->ctdb))); - exit(1); - } + cluster_set_ops(&cluster_ctdb_ops); + + /* nasty hack for now ... */ + { + void brl_ctdb_init_ops(void); + brl_ctdb_init_ops(); + } /* start the protocol running */ ret = ctdb_start(state->ctdb); @@ -250,8 +289,6 @@ void cluster_ctdb_init(struct event_context *ev, const char *model) outside of test code) */ ctdb_connect_wait(state->ctdb); - cluster_set_ops(&cluster_ctdb_ops); - return; failed: diff --git a/source4/cluster/ctdb/direct/ctdbd.c b/source4/cluster/ctdb/direct/ctdbd.c index 700416e5e9..674b54d47a 100644 --- a/source4/cluster/ctdb/direct/ctdbd.c +++ b/source4/cluster/ctdb/direct/ctdbd.c @@ -23,6 +23,7 @@ #include "system/filesys.h" #include "popt.h" #include "system/wait.h" +#include "cmdline.h" static void block_signal(int signum) { @@ -43,21 +44,12 @@ static void block_signal(int signum) 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; - int daemon_mode=0; const char *db_list = "test.tdb"; char *s, *tok; 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" }, - { "daemon", 0, POPT_ARG_NONE, &daemon_mode, 0, "spawn a ctdb daemon", "boolean" }, + POPT_CTDB_CMDLINE { "dblist", 0, POPT_ARG_STRING, &db_list, 0, "list of databases", NULL }, POPT_TABLEEND }; @@ -86,48 +78,11 @@ int main(int argc, const char *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); - } - block_signal(SIGPIPE); 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); - } - if (daemon_mode) { - ctdb_set_flags(ctdb, CTDB_FLAG_DAEMON_MODE); - } - - 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); - } + ctdb = ctdb_cmdline_init(ev); /* attach to the list of databases */ s = talloc_strdup(ctdb, db_list); diff --git a/source4/cluster/ctdb/direct/ctdbd.sh b/source4/cluster/ctdb/direct/ctdbd.sh index 366226260b..7224bdec6d 100755 --- a/source4/cluster/ctdb/direct/ctdbd.sh +++ b/source4/cluster/ctdb/direct/ctdbd.sh @@ -3,6 +3,6 @@ killall -q ctdbd echo "Starting 2 ctdb daemons" -bin/ctdbd --nlist direct/nodes.txt --listen 127.0.0.2:9001 --daemon & -bin/ctdbd --nlist direct/nodes.txt --listen 127.0.0.1:9001 --daemon & +bin/ctdbd --nlist direct/nodes.txt --listen 127.0.0.2:9001 & +bin/ctdbd --nlist direct/nodes.txt --listen 127.0.0.1:9001 & diff --git a/source4/cluster/ctdb/direct/ctdbd_test.c b/source4/cluster/ctdb/direct/ctdbd_test.c index 019cdad30d..00b9f967df 100644 --- a/source4/cluster/ctdb/direct/ctdbd_test.c +++ b/source4/cluster/ctdb/direct/ctdbd_test.c @@ -184,78 +184,32 @@ uint32_t ctdb_hash(const TDB_DATA *key) 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); +/* ask the daemon to migrate a record over so that the local node is the dmaster the client must not have the record locked when performing this call. - 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) + when the daemon has responded this node should be the dmaster (unless it has migrated off again) + */ +void fetch_record(int fd, uint32_t db_id, TDB_DATA key) { - struct ctdb_req_store_unlock *req; - struct ctdb_reply_store_unlock *rep; + struct ctdb_req_call *req; + struct ctdb_reply_call *rep; uint32_t length; int len, cnt, tot; - len = offsetof(struct ctdb_req_store_unlock, data) + key.dsize + data.dsize; + len = offsetof(struct ctdb_req_call, data) + 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_STORE_UNLOCK; + req->hdr.operation = CTDB_REQ_CALL; req->hdr.reqid = 1; + + req->flags = CTDB_IMMEDIATE_MIGRATION; req->db_id = db_id; + req->callid = CTDB_NULL_FUNC; req->keylen = key.dsize; - req->datalen = data.dsize; + req->calldatalen = 0; memcpy(&req->data[0], key.dptr, key.dsize); - memcpy(&req->data[key.dsize], data.dptr, data.dsize); cnt=write(fd, req, len); @@ -281,7 +235,7 @@ void store_unlock(int fd, uint32_t db_id, TDB_DATA key, TDB_DATA data) cnt+=numread; } } - printf("store unlock reply: state:%d\n",rep->state); + printf("fetch record reply: operation:%d state:%d\n",rep->hdr.operation,rep->status); } int main(int argc, const char *argv[]) @@ -291,8 +245,7 @@ int main(int argc, const char *argv[]) struct ctdb_req_message *reply; TDB_DATA dbname; uint32_t db_id; - TDB_DATA key, data; - char str[256]; + TDB_DATA key; /* open the socket to talk to the local ctdb daemon */ fd=ux_socket_connect(CTDB_SOCKET); @@ -338,25 +291,12 @@ int main(int argc, const char *argv[]) printf("the has for the database id is 0x%08x\n",db_id); printf("\n"); - /* send a fetch lock */ + /* send a request to migrate a record to the local node */ 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); + fetch_record(fd, db_id, key); printf("\n"); diff --git a/source4/cluster/ctdb/ib/ibw_ctdb_init.c b/source4/cluster/ctdb/ib/ibw_ctdb_init.c index 3b0c6ad28f..b4adfe6f12 100644 --- a/source4/cluster/ctdb/ib/ibw_ctdb_init.c +++ b/source4/cluster/ctdb/ib/ibw_ctdb_init.c @@ -158,10 +158,10 @@ static int ctdb_ibw_queue_pkt(struct ctdb_node *node, uint8_t *data, uint32_t le /* * transport packet allocator - allows transport to control memory for packets */ -static void *ctdb_ibw_allocate_pkt(struct ctdb_context *ctdb, size_t size) +static void *ctdb_ibw_allocate_pkt(TALLOC_CTX *mem_ctx, size_t size) { /* TODO: use ibw_alloc_send_buf instead... */ - return talloc_size(ctdb, size); + return talloc_size(mem_ctx, size); } #ifdef __NOTDEF__ diff --git a/source4/cluster/ctdb/include/cmdline.h b/source4/cluster/ctdb/include/cmdline.h new file mode 100644 index 0000000000..785595ee6c --- /dev/null +++ b/source4/cluster/ctdb/include/cmdline.h @@ -0,0 +1,7 @@ + +extern struct poptOption popt_ctdb_cmdline[]; + +#define POPT_CTDB_CMDLINE { NULL, 0, POPT_ARG_INCLUDE_TABLE, popt_ctdb_cmdline, 0, "Common ctdb test options:", NULL }, + +struct ctdb_context *ctdb_cmdline_init(struct event_context *ev); + diff --git a/source4/cluster/ctdb/include/ctdb.h b/source4/cluster/ctdb/include/ctdb.h index f24f1000a4..cb765884b6 100644 --- a/source4/cluster/ctdb/include/ctdb.h +++ b/source4/cluster/ctdb/include/ctdb.h @@ -50,10 +50,7 @@ struct ctdb_call_info { ctdb flags */ #define CTDB_FLAG_SELF_CONNECT (1<<0) -/* fork off a separate ctdb daemon */ -#define CTDB_FLAG_DAEMON_MODE (1<<1) -/* for test code only: make ctdb_start() block until all nodes are connected */ -#define CTDB_FLAG_CONNECT_WAIT (1<<2) +#define CTDB_FLAG_TORTURE (1<<1) /* @@ -74,6 +71,11 @@ struct ctdb_context *ctdb_init(struct event_context *ev); int ctdb_set_transport(struct ctdb_context *ctdb, const char *transport); /* + set the directory for the local databases +*/ +int ctdb_set_tdb_dir(struct ctdb_context *ctdb, const char *dir); + +/* set some flags */ void ctdb_set_flags(struct ctdb_context *ctdb, unsigned flags); @@ -142,9 +144,10 @@ int ctdb_call(struct ctdb_db_context *ctdb_db, struct ctdb_call *call); void ctdb_connect_wait(struct ctdb_context *ctdb); /* - wait until we're the only node left + initiate an ordered ctdb cluster shutdown + this function will never return */ -void ctdb_wait_loop(struct ctdb_context *ctdb); +void ctdb_shutdown(struct ctdb_context *ctdb); /* return vnn of this node */ uint32_t ctdb_get_vnn(struct ctdb_context *ctdb); @@ -163,8 +166,8 @@ int ctdb_set_message_handler(struct ctdb_context *ctdb, uint32_t srvid, int ctdb_call(struct ctdb_db_context *ctdb_db, struct ctdb_call *call); -struct ctdb_call_state *ctdb_call_send(struct ctdb_db_context *ctdb_db, struct ctdb_call *call); -int ctdb_call_recv(struct ctdb_call_state *state, struct ctdb_call *call); +struct ctdb_client_call_state *ctdb_call_send(struct ctdb_db_context *ctdb_db, struct ctdb_call *call); +int ctdb_call_recv(struct ctdb_client_call_state *state, struct ctdb_call *call); /* send a ctdb message */ int ctdb_send_message(struct ctdb_context *ctdb, uint32_t vnn, @@ -172,19 +175,29 @@ int ctdb_send_message(struct ctdb_context *ctdb, uint32_t vnn, /* - fetch and lock a ctdb record. Underneath this will force the + Fetch a ctdb record from a remote node + . Underneath this will force the dmaster for the record to be moved to the local node. - The lock is released when is talloc_free() is called on the - returned ctdb_record_handle. */ -struct ctdb_record_handle *ctdb_fetch_lock(struct ctdb_db_context *ctdb_db, TALLOC_CTX *mem_ctx, TDB_DATA key, TDB_DATA *data); +struct ctdb_record_handle *ctdb_fetch_lock(struct ctdb_db_context *ctdb_db, TALLOC_CTX *mem_ctx, + TDB_DATA key, TDB_DATA *data); + /* - change the data in a record held with a ctdb_record_handle - if the new data is zero length, this implies a delete of the record - */ -int ctdb_store_unlock(struct ctdb_record_handle *rec, TDB_DATA data); + do a fetch lock from a client to the local daemon +*/ +#define FETCH_LOCK_SUCCESS 0 +#define FETCH_LOCK_LOCKFAILED 1 +#define FETCH_LOCK_FETCHFAILED 2 +#define FETCH_LOCK_DMASTERFAILED 3 + +int ctdb_client_fetch_lock(struct ctdb_db_context *ctdb_db, + TALLOC_CTX *mem_ctx, + TDB_DATA key, TDB_DATA *data); + + +int ctdb_record_store(struct ctdb_record_handle *h, TDB_DATA data); int ctdb_register_message_handler(struct ctdb_context *ctdb, TALLOC_CTX *mem_ctx, @@ -194,4 +207,10 @@ int ctdb_register_message_handler(struct ctdb_context *ctdb, struct ctdb_db_context *find_ctdb_db(struct ctdb_context *ctdb, uint32_t id); + +struct ctdb_context *ctdb_cmdline_client(struct event_context *ev, const char *ctdb_socket); + +struct ctdb_status; +int ctdb_status(struct ctdb_context *ctdb, struct ctdb_status *status); + #endif diff --git a/source4/cluster/ctdb/include/ctdb_private.h b/source4/cluster/ctdb/include/ctdb_private.h index c50b481cf3..a8f7b48f49 100644 --- a/source4/cluster/ctdb/include/ctdb_private.h +++ b/source4/cluster/ctdb/include/ctdb_private.h @@ -30,8 +30,8 @@ #define CTDB_DS_ALIGNMENT 8 +#define CTDB_NULL_FUNC 0xf0000001 -#define CTDB_FETCH_FUNC 0xf0000001 /* an installed ctdb remote call */ @@ -50,6 +50,11 @@ struct ctdb_address { int port; }; +/* + check a vnn is valid + */ +#define ctdb_validate_vnn(ctdb, vnn) (((uint32_t)(vnn)) < (ctdb)->num_nodes) + /* called from the queue code when a packet comes in. Called with data==NULL on error */ @@ -68,12 +73,6 @@ struct ctdb_node { uint32_t vnn; }; -struct ctdb_record_handle { - struct ctdb_db_context *ctdb_db; - TDB_DATA key; - TDB_DATA *data; -}; - /* transport specific methods */ @@ -81,7 +80,7 @@ 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 ); + void *(*allocate_pkt)(TALLOC_CTX *mem_ctx, size_t ); }; /* @@ -115,14 +114,51 @@ struct ctdb_daemon_data { struct ctdb_queue *queue; }; +/* + ctdb status information + */ +struct ctdb_status { + uint32_t client_packets_sent; + uint32_t client_packets_recv; + uint32_t node_packets_sent; + uint32_t node_packets_recv; + struct { + uint32_t req_call; + uint32_t reply_call; + uint32_t reply_redirect; + uint32_t req_dmaster; + uint32_t reply_dmaster; + uint32_t reply_error; + uint32_t req_message; + uint32_t req_finished; + } count; + struct { + uint32_t req_call; + uint32_t req_message; + uint32_t req_finished; + uint32_t req_register; + uint32_t req_connect_wait; + uint32_t req_shutdown; + uint32_t req_status; + } client; + uint32_t total_calls; + uint32_t pending_calls; + uint32_t lockwait_calls; + uint32_t pending_lockwait_calls; + double max_call_latency; + double max_lockwait_latency; +}; + /* main state of the ctdb daemon */ struct ctdb_context { struct event_context *ev; struct ctdb_address address; const char *name; + const char *db_directory; uint32_t vnn; /* our own vnn */ uint32_t num_nodes; uint32_t num_connected; + uint32_t num_finished; unsigned flags; struct idr_context *idr; struct ctdb_node **nodes; /* array of nodes in the cluster - indexed by vnn */ @@ -134,6 +170,7 @@ struct ctdb_context { struct ctdb_db_context *db_list; struct ctdb_message_list *message_list; struct ctdb_daemon_data daemon; + struct ctdb_status status; }; struct ctdb_db_context { @@ -192,7 +229,6 @@ struct ctdb_call_state { struct ctdb_call call; int redirect_count; struct ctdb_ltdb_header header; - void *fetch_private; struct { void (*fn)(struct ctdb_call_state *); void *private_data; @@ -200,6 +236,14 @@ struct ctdb_call_state { }; +/* used for fetch_lock */ +struct ctdb_fetch_handle { + struct ctdb_db_context *ctdb_db; + TDB_DATA key; + TDB_DATA *data; + struct ctdb_ltdb_header header; +}; + /* operation IDs */ @@ -211,15 +255,15 @@ enum ctdb_operation { CTDB_REPLY_DMASTER = 4, CTDB_REPLY_ERROR = 5, CTDB_REQ_MESSAGE = 6, + CTDB_REQ_FINISHED = 7, /* only used on the domain socket */ CTDB_REQ_REGISTER = 1000, CTDB_REQ_CONNECT_WAIT = 1001, CTDB_REPLY_CONNECT_WAIT = 1002, - CTDB_REQ_FETCH_LOCK = 1003, - CTDB_REPLY_FETCH_LOCK = 1004, - CTDB_REQ_STORE_UNLOCK = 1005, - CTDB_REPLY_STORE_UNLOCK = 1006 + CTDB_REQ_SHUTDOWN = 1003, + CTDB_REQ_STATUS = 1004, + CTDB_REPLY_STATUS = 1005 }; #define CTDB_MAGIC 0x43544442 /* CTDB */ @@ -294,40 +338,31 @@ struct ctdb_req_message { uint8_t data[1]; }; -struct ctdb_req_connect_wait { +struct ctdb_req_finished { struct ctdb_req_header hdr; }; -struct ctdb_reply_connect_wait { +struct ctdb_req_shutdown { struct ctdb_req_header hdr; - uint32_t vnn; - uint32_t num_connected; }; -struct ctdb_req_fetch_lock { +struct ctdb_req_connect_wait { struct ctdb_req_header hdr; - uint32_t db_id; - uint32_t keylen; - uint8_t key[1]; /* key[] */ }; -struct ctdb_reply_fetch_lock { +struct ctdb_reply_connect_wait { struct ctdb_req_header hdr; - uint32_t state; - uint32_t datalen; - uint8_t data[1]; /* data[] */ + uint32_t vnn; + uint32_t num_connected; }; -struct ctdb_req_store_unlock { + +struct ctdb_req_status { struct ctdb_req_header hdr; - uint32_t db_id; - uint32_t keylen; - uint32_t datalen; - uint8_t data[1]; /* key[] and data[] */ }; -struct ctdb_reply_store_unlock { +struct ctdb_reply_status { struct ctdb_req_header hdr; - uint32_t state; + struct ctdb_status status; }; /* internal prototypes */ @@ -353,6 +388,16 @@ int ctdb_ltdb_fetch(struct ctdb_db_context *ctdb_db, int ctdb_ltdb_store(struct ctdb_db_context *ctdb_db, TDB_DATA key, struct ctdb_ltdb_header *header, TDB_DATA data); void ctdb_queue_packet(struct ctdb_context *ctdb, struct ctdb_req_header *hdr); +int ctdb_ltdb_lock_requeue(struct ctdb_db_context *ctdb_db, + TDB_DATA key, struct ctdb_req_header *hdr, + void (*recv_pkt)(void *, uint8_t *, uint32_t ), + void *recv_context); +int ctdb_ltdb_lock_fetch_requeue(struct ctdb_db_context *ctdb_db, + TDB_DATA key, struct ctdb_ltdb_header *header, + struct ctdb_req_header *hdr, TDB_DATA *data, + void (*recv_pkt)(void *, uint8_t *, uint32_t ), + void *recv_context); +void ctdb_recv_pkt(struct ctdb_context *ctdb, uint8_t *data, uint32_t length); struct ctdb_call_state *ctdb_call_local_send(struct ctdb_db_context *ctdb_db, struct ctdb_call *call, @@ -386,7 +431,7 @@ struct ctdb_queue *ctdb_queue_setup(struct ctdb_context *ctdb, /* allocate a packet for use in client<->daemon communication */ -void *ctdbd_allocate_pkt(struct ctdb_context *ctdb, size_t len); +void *ctdbd_allocate_pkt(TALLOC_CTX *mem_ctx, size_t len); /* @@ -437,16 +482,32 @@ int ctdb_daemon_send_message(struct ctdb_context *ctdb, uint32_t vnn, void ctdb_daemon_connect_wait(struct ctdb_context *ctdb); -/* - do a fetch lock from a client to the local daemon -*/ -struct ctdb_record_handle *ctdb_client_fetch_lock(struct ctdb_db_context *ctdb_db, - TALLOC_CTX *mem_ctx, - TDB_DATA key, TDB_DATA *data); +struct lockwait_handle *ctdb_lockwait(struct ctdb_db_context *ctdb_db, + TDB_DATA key, + void (*callback)(void *), void *private_data); -/* - do a store unlock from a client to the local daemon -*/ -int ctdb_client_store_unlock(struct ctdb_record_handle *rec, TDB_DATA data); +struct ctdb_call_state *ctdb_daemon_call_send(struct ctdb_db_context *ctdb_db, + struct ctdb_call *call); + +int ctdb_daemon_call_recv(struct ctdb_call_state *state, struct ctdb_call *call); + +struct ctdb_call_state *ctdb_daemon_call_send_remote(struct ctdb_db_context *ctdb_db, + struct ctdb_call *call, + struct ctdb_ltdb_header *header); + +void ctdb_request_finished(struct ctdb_context *ctdb, struct ctdb_req_header *hdr); + +int ctdb_call_local(struct ctdb_db_context *ctdb_db, struct ctdb_call *call, + struct ctdb_ltdb_header *header, TDB_DATA *data, + uint32_t caller); + +void *_idr_find_type(struct idr_context *idp, int id, const char *type, const char *location); +#define idr_find_type(idp, id, type) (type *)_idr_find_type(idp, id, #type, __location__) + +void ctdb_recv_raw_pkt(void *p, uint8_t *data, uint32_t length); + +int ctdb_socket_connect(struct ctdb_context *ctdb); + +void ctdb_latency(double *latency, struct timeval t); #endif diff --git a/source4/cluster/ctdb/include/includes.h b/source4/cluster/ctdb/include/includes.h index 994c25452c..bffc66b358 100644 --- a/source4/cluster/ctdb/include/includes.h +++ b/source4/cluster/ctdb/include/includes.h @@ -6,15 +6,16 @@ #include "idtree.h" #include "ctdb.h" #include "lib/util/dlinklist.h" +#include "lib/util/debug.h" typedef bool BOOL; #define True 1 #define False 0 -#define LogLevel 0 +extern int LogLevel; -#define DEBUG(lvl, x) if ((lvl) <= LogLevel) (printf x) +#define DEBUG(lvl, x) if ((lvl) <= LogLevel) (do_debug x) #define _PUBLIC_ @@ -32,5 +33,5 @@ int timeval_compare(const struct timeval *tv1, const struct timeval *tv2); struct timeval timeval_until(const struct timeval *tv1, const struct timeval *tv2); _PUBLIC_ struct timeval timeval_current_ofs(uint32_t secs, uint32_t usecs); +double timeval_elapsed(struct timeval *tv); char **file_lines_load(const char *fname, int *numlines, TALLOC_CTX *mem_ctx); - diff --git a/source4/cluster/ctdb/opendb_ctdb.c b/source4/cluster/ctdb/opendb_ctdb.c index 6e2748291e..df399515eb 100644 --- a/source4/cluster/ctdb/opendb_ctdb.c +++ b/source4/cluster/ctdb/opendb_ctdb.c @@ -223,7 +223,8 @@ static NTSTATUS odb_push_record(struct odb_lock *lck, struct opendb_file *file) if (!file->num_entries) { dbuf.dptr = NULL; dbuf.dsize = 0; - ctdb_store_unlock(lck->rec, dbuf); + ctdb_record_store(lck->rec, dbuf); + talloc_free(lck->rec); return NT_STATUS_OK; } @@ -233,7 +234,8 @@ static NTSTATUS odb_push_record(struct odb_lock *lck, struct opendb_file *file) dbuf.dptr = blob.data; dbuf.dsize = blob.length; - ret = ctdb_store_unlock(lck->rec, dbuf); + ret = ctdb_record_store(lck->rec, dbuf); + talloc_free(lck->rec); data_blob_free(&blob); if (ret != 0) { return NT_STATUS_INTERNAL_DB_CORRUPTION; diff --git a/source4/cluster/ctdb/tcp/tcp_connect.c b/source4/cluster/ctdb/tcp/tcp_connect.c index a1f2d331cf..4d9d8e8386 100644 --- a/source4/cluster/ctdb/tcp/tcp_connect.c +++ b/source4/cluster/ctdb/tcp/tcp_connect.c @@ -43,6 +43,10 @@ void ctdb_tcp_tnode_cb(uint8_t *data, size_t cnt, void *private_data) struct ctdb_tcp_node *tnode = talloc_get_type( node->private_data, struct ctdb_tcp_node); + if (data == NULL) { + node->ctdb->upcalls->node_dead(node); + } + /* start a new connect cycle to try to re-establish the link */ close(tnode->fd); diff --git a/source4/cluster/ctdb/tcp/tcp_init.c b/source4/cluster/ctdb/tcp/tcp_init.c index 20b9bc9e33..9b54bb75ba 100644 --- a/source4/cluster/ctdb/tcp/tcp_init.c +++ b/source4/cluster/ctdb/tcp/tcp_init.c @@ -46,12 +46,6 @@ static int ctdb_tcp_start(struct ctdb_context *ctdb) ctdb_tcp_node_connect, node); } - if (ctdb->flags&CTDB_FLAG_CONNECT_WAIT) { - /* wait until all nodes are connected (should not be needed - outide of test code) */ - ctdb_connect_wait(ctdb); - } - return 0; } @@ -78,13 +72,13 @@ static int ctdb_tcp_add_node(struct ctdb_node *node) /* transport packet allocator - allows transport to control memory for packets */ -static void *ctdb_tcp_allocate_pkt(struct ctdb_context *ctdb, size_t size) +static void *ctdb_tcp_allocate_pkt(TALLOC_CTX *mem_ctx, 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); + return talloc_size(mem_ctx, size); } diff --git a/source4/cluster/ctdb/tests/bench.sh b/source4/cluster/ctdb/tests/bench.sh index 50e9e08f99..3d0696f171 100755 --- a/source4/cluster/ctdb/tests/bench.sh +++ b/source4/cluster/ctdb/tests/bench.sh @@ -3,7 +3,7 @@ killall -q ctdb_bench echo "Trying 2 nodes" -bin/ctdb_bench --nlist tests/nodes.txt --listen 127.0.0.2:9001 $* & -bin/ctdb_bench --nlist tests/nodes.txt --listen 127.0.0.1:9001 $* +$VALGRIND bin/ctdb_bench --nlist tests/nodes.txt --listen 127.0.0.2:9001 $* & +$VALGRIND bin/ctdb_bench --nlist tests/nodes.txt --listen 127.0.0.1:9001 $* +wait -killall -q ctdb_bench diff --git a/source4/cluster/ctdb/tests/bench1.sh b/source4/cluster/ctdb/tests/bench1.sh index 3481d82be2..9adcf3198b 100755 --- a/source4/cluster/ctdb/tests/bench1.sh +++ b/source4/cluster/ctdb/tests/bench1.sh @@ -4,5 +4,4 @@ killall -q ctdb_bench echo "Trying 1 nodes" bin/ctdb_bench --nlist tests/1node.txt --listen 127.0.0.2:9001 $* - -killall -q ctdb_bench +wait diff --git a/source4/cluster/ctdb/tests/ctdb_bench.c b/source4/cluster/ctdb/tests/ctdb_bench.c index 37c095c0c0..02fcc1f2d4 100644 --- a/source4/cluster/ctdb/tests/ctdb_bench.c +++ b/source4/cluster/ctdb/tests/ctdb_bench.c @@ -22,7 +22,7 @@ #include "lib/events/events.h" #include "system/filesys.h" #include "popt.h" -#include "tests/cmdline.h" +#include "cmdline.h" #include <sys/time.h> #include <time.h> @@ -45,7 +45,6 @@ static double end_timer(void) static int timelimit = 10; static int num_records = 10; static int num_msgs = 1; -static int num_repeats = 100; enum my_functions {FUNC_INCR=1, FUNC_FETCH=2}; @@ -78,51 +77,6 @@ static int fetch_func(struct ctdb_call_info *call) return 0; } -/* - benchmark incrementing an integer -*/ -static void bench_incr(struct ctdb_context *ctdb, struct ctdb_db_context *ctdb_db) -{ - int loops=0; - int ret, i; - struct ctdb_call call; - - ZERO_STRUCT(call); - - start_timer(); - - while (1) { - uint32_t v = loops % num_records; - - call.call_id = FUNC_INCR; - call.key.dptr = (uint8_t *)&v; - call.key.dsize = 4; - - for (i=0;i<num_repeats;i++) { - ret = ctdb_call(ctdb_db, &call); - 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); - } - } - - call.call_id = FUNC_FETCH; - - ret = ctdb_call(ctdb_db, &call); - 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 *)call.reply_data.dptr); -} static int msg_count; static int msg_plus, msg_minus; @@ -259,6 +213,7 @@ int main(int argc, const char *argv[]) bench_ring(ctdb, ev); /* shut it down */ - talloc_free(ctdb); + ctdb_shutdown(ctdb); + return 0; } diff --git a/source4/cluster/ctdb/tests/ctdb_fetch.c b/source4/cluster/ctdb/tests/ctdb_fetch.c index febaf13fe4..39bd861fc1 100644 --- a/source4/cluster/ctdb/tests/ctdb_fetch.c +++ b/source4/cluster/ctdb/tests/ctdb_fetch.c @@ -22,7 +22,7 @@ #include "lib/events/events.h" #include "system/filesys.h" #include "popt.h" -#include "tests/cmdline.h" +#include "cmdline.h" #include <sys/time.h> #include <time.h> @@ -58,18 +58,18 @@ static int msg_count; static void bench_fetch_1node(struct ctdb_context *ctdb) { TDB_DATA key, data, nulldata; - struct ctdb_record_handle *rec; struct ctdb_db_context *ctdb_db; TALLOC_CTX *tmp_ctx = talloc_new(ctdb); int dest, ret; + struct ctdb_record_handle *h; - key.dptr = discard_const("testkey"); - key.dsize = strlen((const char *)key.dptr); + key.dptr = discard_const(TESTKEY); + key.dsize = strlen(TESTKEY); ctdb_db = ctdb_db_handle(ctdb, "test.tdb"); - rec = ctdb_fetch_lock(ctdb_db, tmp_ctx, key, &data); - if (rec == NULL) { + h = ctdb_fetch_lock(ctdb_db, tmp_ctx, key, &data); + if (h == NULL) { printf("Failed to fetch record '%s' on node %d\n", (const char *)key.dptr, ctdb_get_vnn(ctdb)); talloc_free(tmp_ctx); @@ -88,7 +88,8 @@ static void bench_fetch_1node(struct ctdb_context *ctdb) msg_count, ctdb_get_vnn(ctdb)); data.dsize = strlen((const char *)data.dptr)+1; - ret = ctdb_store_unlock(rec, data); + ret = ctdb_record_store(h, data); + talloc_free(h); if (ret != 0) { printf("Failed to store record\n"); } @@ -141,6 +142,10 @@ static void bench_fetch(struct ctdb_context *ctdb, struct event_context *ev) printf("Event loop failed!\n"); break; } + + if (LogLevel > 9) { + talloc_report_null_full(); + } } printf("Fetch: %.2f msgs/sec\n", msg_count/end_timer()); @@ -192,6 +197,8 @@ int main(int argc, const char *argv[]) } } + /* talloc_enable_leak_report_full(); */ + /* setup the remaining options for the main program to use */ extra_argv = poptGetArgs(pc); if (extra_argv) { @@ -240,7 +247,8 @@ int main(int argc, const char *argv[]) printf("DATA:\n%s\n", (char *)call.reply_data.dptr); - /* shut it down */ - talloc_free(ctdb); + /* go into a wait loop to allow other nodes to complete */ + ctdb_shutdown(ctdb); + return 0; } diff --git a/source4/cluster/ctdb/tests/ctdb_fetch1.c b/source4/cluster/ctdb/tests/ctdb_fetch1.c index ffe9c7c946..b92e7fcda3 100644 --- a/source4/cluster/ctdb/tests/ctdb_fetch1.c +++ b/source4/cluster/ctdb/tests/ctdb_fetch1.c @@ -24,7 +24,8 @@ #include "popt.h" #include "ctdb.h" #include "ctdb_private.h" -#include "tests/cmdline.h" +#include "cmdline.h" +#include <sys/time.h> #define PARENT_SRVID 0 #define CHILD1_SRVID 1 @@ -32,6 +33,20 @@ int num_msg=0; +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 void message_handler(struct ctdb_context *ctdb, uint32_t srvid, TDB_DATA data, void *private_data) { @@ -45,9 +60,9 @@ static void child_handler(struct ctdb_context *ctdb, uint32_t srvid, void test1(struct ctdb_db_context *ctdb_db) { - struct ctdb_record_handle *rh; TDB_DATA key, data, data2, store_data; int ret; + struct ctdb_record_handle *h; /* test 1 : write data and read it back. should all be the same @@ -55,13 +70,27 @@ void test1(struct ctdb_db_context *ctdb_db) printf("Test1: write and verify we can read it back: "); key.dptr = discard_const("Record"); key.dsize = strlen((const char *)key.dptr)+1; - rh = ctdb_fetch_lock(ctdb_db, ctdb_db, key, &data); + h = ctdb_fetch_lock(ctdb_db, ctdb_db, key, &data); + if (h == NULL) { + printf("test1: ctdb_fetch_lock() failed\n"); + exit(1); + } store_data.dptr = discard_const("data to store"); store_data.dsize = strlen((const char *)store_data.dptr)+1; - ret = ctdb_store_unlock(rh, store_data); + ret = ctdb_record_store(h, store_data); + talloc_free(h); + if (ret!=0) { + printf("test1: ctdb_record_store() failed\n"); + exit(1); + } + + h = ctdb_fetch_lock(ctdb_db, ctdb_db, key, &data2); + if (h == NULL) { + printf("test1: ctdb_fetch_lock() failed\n"); + exit(1); + } - rh = ctdb_fetch_lock(ctdb_db, ctdb_db, key, &data2); /* hopefully data2 will now contain the record written above */ if (!strcmp("data to store", (const char *)data2.dptr)) { printf("SUCCESS\n"); @@ -71,14 +100,19 @@ void test1(struct ctdb_db_context *ctdb_db) } /* just write it back to unlock it */ - ret = ctdb_store_unlock(rh, store_data); + ret = ctdb_record_store(h, store_data); + talloc_free(h); + if (ret!=0) { + printf("test1: ctdb_record_store() failed\n"); + exit(1); + } } void child(int srvid, struct event_context *ev, struct ctdb_context *ctdb, struct ctdb_db_context *ctdb_db) { TDB_DATA data; - struct ctdb_record_handle *rh; TDB_DATA key, data2; + struct ctdb_record_handle *h; data.dptr=discard_const("dummy message"); data.dsize=strlen((const char *)data.dptr)+1; @@ -94,13 +128,23 @@ void child(int srvid, struct event_context *ev, struct ctdb_context *ctdb, struc /* fetch and lock the record */ key.dptr = discard_const("Record"); key.dsize = strlen((const char *)key.dptr)+1; - rh = ctdb_fetch_lock(ctdb_db, ctdb_db, key, &data2); + printf("client:%d fetching the record\n",srvid); + h = ctdb_fetch_lock(ctdb_db, ctdb_db, key, &data2); + printf("client:%d the record is fetched and locked\n",srvid); + if (h == NULL) { + printf("client: ctdb_fetch_lock() failed\n"); + exit(1); + } ctdb_send_message(ctdb, ctdb_get_vnn(ctdb), PARENT_SRVID, data); - - while (1) { + /* wait until parent tells us to release the lock */ + while (num_msg==1) { event_loop_once(ev); } + + printf("child %d terminating\n",srvid); + exit(10); + } /* @@ -204,29 +248,48 @@ int main(int argc, const char *argv[]) */ data.dptr=discard_const("dummy message"); data.dsize=strlen((const char *)data.dptr)+1; - printf("Send message to child 1 to fetch_lock the record\n"); ctdb_send_message(ctdb, ctdb_get_vnn(ctdb), CHILD1_SRVID, data); /* wait for child 1 to complete fetching and locking the record */ while (num_msg!=3) { event_loop_once(ev); } - printf("Child 1 has fetched and locked the record\n"); /* now tell child 2 to fetch and lock the same record */ - printf("Send message to child 2 to fetch_lock the record\n"); ctdb_send_message(ctdb, ctdb_get_vnn(ctdb), CHILD2_SRVID, data); - /* wait for child 2 to complete fetching and locking the record */ - while (num_msg!=4) { + /* wait a while for child 2 to complete fetching and locking the + record, this should fail since the record is already locked + by the first child */ + start_timer(); + while ( (end_timer() < 1.0) && (num_msg!=4) ) { event_loop_once(ev); } - printf("Child 2 has fetched and locked the record\n"); + if (num_msg!=4) { + printf("Child 2 did not get the lock since it is held by client 1:SUCCESS\n"); + } else { + printf("Child 2 did get the lock:FAILURE\n"); + exit(10); + } + /* send message to child 1 to terminate, which should let child 2 + get the lock. + */ + ctdb_send_message(ctdb, ctdb_get_vnn(ctdb), CHILD1_SRVID, data); - while (1) { + + /* wait for a final message from child 2 it has received the lock + which indicates success */ + while (num_msg!=4) { event_loop_once(ev); } + printf("child 2 aquired the lock after child 1 terminated:SUCCESS\n"); + + /* send a message to child 2 to tell it to terminate too */ + ctdb_send_message(ctdb, ctdb_get_vnn(ctdb), CHILD2_SRVID, data); + + + printf("Test was SUCCESSFUL\n"); /* shut it down */ talloc_free(ctdb); diff --git a/source4/cluster/ctdb/tests/ctdb_messaging.c b/source4/cluster/ctdb/tests/ctdb_messaging.c index d02c2116e8..e7057b64fb 100644 --- a/source4/cluster/ctdb/tests/ctdb_messaging.c +++ b/source4/cluster/ctdb/tests/ctdb_messaging.c @@ -22,7 +22,7 @@ #include "lib/events/events.h" #include "system/filesys.h" #include "popt.h" -#include "tests/cmdline.h" +#include "cmdline.h" static int timelimit = 10; static int num_records = 10; @@ -130,7 +130,7 @@ int main(int argc, const char *argv[]) for (j=0;j<num_clients;j++) { printf("sending message to %d:%d\n", i, j); sprintf(buf,"Message from %d to vnn:%d srvid:%d",ctdb_get_vnn(ctdb),i,j); - data.dptr=buf; + data.dptr = (unsigned char *)buf; data.dsize=strlen(buf)+1; ctdb_send_message(ctdb, i, j, data); } @@ -141,6 +141,7 @@ int main(int argc, const char *argv[]) } /* shut it down */ - talloc_free(ctdb); + ctdb_shutdown(ctdb); + return 0; } diff --git a/source4/cluster/ctdb/tests/ctdb_test.c b/source4/cluster/ctdb/tests/ctdb_test.c index 8125cc4490..dec1ea5100 100644 --- a/source4/cluster/ctdb/tests/ctdb_test.c +++ b/source4/cluster/ctdb/tests/ctdb_test.c @@ -22,7 +22,7 @@ #include "lib/events/events.h" #include "system/filesys.h" #include "popt.h" -#include "tests/cmdline.h" +#include "cmdline.h" enum my_functions {FUNC_SORT=1, FUNC_FETCH=2}; @@ -133,6 +133,8 @@ int main(int argc, const char *argv[]) /* start the protocol running */ ret = ctdb_start(ctdb); + ctdb_connect_wait(ctdb); + ZERO_STRUCT(call); call.key.dptr = discard_const("test"); call.key.dsize = strlen("test")+1; @@ -169,14 +171,7 @@ int main(int argc, const char *argv[]) talloc_free(call.reply_data.dptr); /* go into a wait loop to allow other nodes to complete */ - ctdb_wait_loop(ctdb); - - /*talloc_report_full(ctdb, stdout);*/ - -/* sleep for a while so that our daemon will remaining alive for the other nodes in the cluster */ -sleep(10); + ctdb_shutdown(ctdb); - /* shut it down */ - talloc_free(ctdb); return 0; } diff --git a/source4/cluster/ctdb/tests/fetch.sh b/source4/cluster/ctdb/tests/fetch.sh index ba043e17c5..73192e70ae 100755 --- a/source4/cluster/ctdb/tests/fetch.sh +++ b/source4/cluster/ctdb/tests/fetch.sh @@ -3,7 +3,13 @@ killall -q ctdb_fetch echo "Trying 2 nodes" -bin/ctdb_fetch --nlist tests/nodes.txt --listen 127.0.0.2:9001 $* & -bin/ctdb_fetch --nlist tests/nodes.txt --listen 127.0.0.1:9001 $* +$VALGRIND bin/ctdb_fetch --nlist tests/nodes.txt --listen 127.0.0.2:9001 $* & +$VALGRIND bin/ctdb_fetch --nlist tests/nodes.txt --listen 127.0.0.1:9001 $* +wait -killall -q ctdb_fetch +echo "Trying 4 nodes" +$VALGRIND bin/ctdb_fetch --nlist tests/4nodes.txt --listen 127.0.0.4:9001 $* & +$VALGRIND bin/ctdb_fetch --nlist tests/4nodes.txt --listen 127.0.0.3:9001 $* & +$VALGRIND bin/ctdb_fetch --nlist tests/4nodes.txt --listen 127.0.0.2:9001 $* & +$VALGRIND bin/ctdb_fetch --nlist tests/4nodes.txt --listen 127.0.0.1:9001 $* +wait diff --git a/source4/cluster/ctdb/tests/fetch1.sh b/source4/cluster/ctdb/tests/fetch1.sh index 4331a57d3e..db584ec012 100755 --- a/source4/cluster/ctdb/tests/fetch1.sh +++ b/source4/cluster/ctdb/tests/fetch1.sh @@ -3,6 +3,5 @@ killall -q ctdb_fetch1 echo "Trying node" -bin/ctdb_fetch1 --nlist tests/1node.txt --listen 127.0.0.1:9001 --daemon $* - -killall -q ctdb_fetch1 +bin/ctdb_fetch1 --nlist tests/1node.txt --listen 127.0.0.1:9001 $* +wait diff --git a/source4/cluster/ctdb/tests/lockwait.c b/source4/cluster/ctdb/tests/lockwait.c new file mode 100644 index 0000000000..51f88c1168 --- /dev/null +++ b/source4/cluster/ctdb/tests/lockwait.c @@ -0,0 +1,245 @@ +/* + test a lock wait idea + + 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 "system/wait.h" +#include "popt.h" +#include "cmdline.h" + + +struct lockwait_handle { + struct fd_event *fde; + int fd[2]; + pid_t child; + void *private_data; + void (*callback)(void *); +}; + +static void lockwait_handler(struct event_context *ev, struct fd_event *fde, + uint16_t flags, void *private_data) +{ + struct lockwait_handle *h = talloc_get_type(private_data, + struct lockwait_handle); + void (*callback)(void *) = h->callback; + void *p = h->private_data; + talloc_set_destructor(h, NULL); + close(h->fd[0]); + talloc_free(h); + callback(p); + waitpid(h->child, NULL, 0); +} + +static int lockwait_destructor(struct lockwait_handle *h) +{ + close(h->fd[0]); + kill(h->child, SIGKILL); + waitpid(h->child, NULL, 0); + return 0; +} + + +static struct lockwait_handle *lockwait(struct event_context *ev, + TALLOC_CTX *mem_ctx, + int fd, off_t ofs, size_t len, + void (*callback)(void *), void *private_data) +{ + struct lockwait_handle *h; + int ret; + + h = talloc_zero(mem_ctx, struct lockwait_handle); + if (h == NULL) { + return NULL; + } + + ret = pipe(h->fd); + if (ret != 0) { + talloc_free(h); + return NULL; + } + + h->child = fork(); + if (h->child == (pid_t)-1) { + close(h->fd[0]); + close(h->fd[1]); + talloc_free(h); + return NULL; + } + + h->callback = callback; + h->private_data = private_data; + + if (h->child == 0) { + /* in child */ + struct flock lock; + close(h->fd[0]); + lock.l_type = F_WRLCK; + lock.l_whence = SEEK_SET; + lock.l_start = ofs; + lock.l_len = len; + lock.l_pid = 0; + fcntl(fd,F_SETLKW,&lock); + _exit(0); + } + + close(h->fd[1]); + talloc_set_destructor(h, lockwait_destructor); + + h->fde = event_add_fd(ev, h, h->fd[0], EVENT_FD_READ, lockwait_handler, h); + if (h->fde == NULL) { + talloc_free(h); + return NULL; + } + + return h; +} + + + + +static void fcntl_lock_callback(void *p) +{ + int *got_lock = (int *)p; + *got_lock = 1; +} + +/* + get an fcntl lock - waiting if necessary + */ +static int fcntl_lock(struct event_context *ev, + int fd, int op, off_t offset, off_t count, int type) +{ + struct flock lock; + int ret; + int use_lockwait = (op == F_SETLKW); + int got_lock = 0; + + lock.l_type = type; + lock.l_whence = SEEK_SET; + lock.l_start = offset; + lock.l_len = count; + lock.l_pid = 0; + + do { + ret = fcntl(fd,use_lockwait?F_SETLK:op,&lock); + if (ret == 0) { + return 0; + } + if (ret == -1 && + (errno == EACCES || errno == EAGAIN || errno == EDEADLK)) { + struct lockwait_handle *h; + h = lockwait(ev, ev, fd, offset, count, + fcntl_lock_callback, &got_lock); + if (h == NULL) { + errno = ENOLCK; + return -1; + } + /* in real code we would return to the event loop */ + while (!got_lock) { + event_loop_once(ev); + } + got_lock = 0; + } + } while (!got_lock); + + return ret; +} + +static void child(struct event_context *ev, int n) +{ + int fd; + int count=0; + struct timeval tv; + fd = open("test.dat", O_CREAT|O_RDWR, 0666); + if (fd == -1) { + perror("test.dat"); + exit(1); + } + + tv = timeval_current(); + + while (timeval_elapsed(&tv) < 10) { + int ret; + ret = fcntl_lock(ev, fd, F_SETLKW, 0, 1, F_WRLCK); + if (ret != 0) { + printf("Failed to get lock in child %d!\n", n); + break; + } + fcntl_lock(ev, fd, F_SETLK, 0, 1, F_UNLCK); + count++; + } + + printf("child %2d %.0f ops/sec\n", n, count/timeval_elapsed(&tv)); + _exit(0); +} + +static int timelimit = 10; + +/* + main program +*/ +int main(int argc, const char *argv[]) +{ + pid_t *pids; + int nprogs = 2; + int i; + struct event_context *ev; + struct poptOption popt_options[] = { + POPT_AUTOHELP + { "timelimit", 't', POPT_ARG_INT, &timelimit, 0, "timelimit", "integer" }, + { "num-progs", 'n', POPT_ARG_INT, &nprogs, 0, "num_progs", "integer" }, + POPT_TABLEEND + }; + poptContext pc; + int opt; + + 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); + } + } + + ev = event_context_init(NULL); + + pids = talloc_array(ev, pid_t, nprogs); + + /* create N processes fighting over the same lock */ + for (i=0;i<nprogs;i++) { + pids[i] = fork(); + if (pids[i] == 0) { + child(ev, i); + } + } + + printf("Waiting for %d children ...\n", nprogs); + + /* wait for our kids to finish playing */ + for (i=0;i<nprogs;i++) { + waitpid(pids[i], NULL, 0); + } + + return 0; +} diff --git a/source4/cluster/ctdb/tests/messaging.sh b/source4/cluster/ctdb/tests/messaging.sh index 179a2bef88..46f3e4dc77 100755 --- a/source4/cluster/ctdb/tests/messaging.sh +++ b/source4/cluster/ctdb/tests/messaging.sh @@ -5,5 +5,4 @@ killall -q ctdb_messaging echo "Trying 2 nodes" bin/ctdb_messaging --nlist tests/nodes.txt --listen 127.0.0.2:9001 $* & bin/ctdb_messaging --nlist tests/nodes.txt --listen 127.0.0.1:9001 $* - -killall -q ctdb_messaging +wait diff --git a/source4/cluster/ctdb/tests/test.sh b/source4/cluster/ctdb/tests/test.sh index 6197ce9616..23fdb8d6ce 100755 --- a/source4/cluster/ctdb/tests/test.sh +++ b/source4/cluster/ctdb/tests/test.sh @@ -3,33 +3,15 @@ killall -q ctdb_test -echo "Trying 2 nodes" -bin/ctdb_test --nlist tests/nodes.txt --listen 127.0.0.1:9001 & -bin/ctdb_test --nlist tests/nodes.txt --listen 127.0.0.2:9001 & +echo "Trying 2 nodes ..." +$VALGRIND bin/ctdb_test --nlist tests/nodes.txt --listen 127.0.0.1:9001 & +$VALGRIND bin/ctdb_test --nlist tests/nodes.txt --listen 127.0.0.2:9001 +wait + +echo "Trying 4 nodes ..." +$VALGRIND bin/ctdb_test --nlist tests/4nodes.txt --listen 127.0.0.1:9001 & +$VALGRIND bin/ctdb_test --nlist tests/4nodes.txt --listen 127.0.0.2:9001 & +$VALGRIND bin/ctdb_test --nlist tests/4nodes.txt --listen 127.0.0.3:9001 & +$VALGRIND bin/ctdb_test --nlist tests/4nodes.txt --listen 127.0.0.4:9001 +wait -sleep 3 -killall ctdb_test - -echo "Trying 4 nodes" -bin/ctdb_test --nlist tests/4nodes.txt --listen 127.0.0.1:9001 & -bin/ctdb_test --nlist tests/4nodes.txt --listen 127.0.0.2:9001 & -bin/ctdb_test --nlist tests/4nodes.txt --listen 127.0.0.3:9001 & -bin/ctdb_test --nlist tests/4nodes.txt --listen 127.0.0.4:9001 & -sleep 3 - -killall ctdb_test - -echo "Trying 2 nodes in daemon mode" -bin/ctdb_test --nlist tests/nodes.txt --listen 127.0.0.1:9001 --daemon & -bin/ctdb_test --nlist tests/nodes.txt --listen 127.0.0.2:9001 --daemon & - -sleep 3 -killall ctdb_test - -echo "Trying 4 nodes in daemon mode" -bin/ctdb_test --nlist tests/4nodes.txt --listen 127.0.0.1:9001 --daemon & -bin/ctdb_test --nlist tests/4nodes.txt --listen 127.0.0.2:9001 --daemon & -bin/ctdb_test --nlist tests/4nodes.txt --listen 127.0.0.3:9001 --daemon & -bin/ctdb_test --nlist tests/4nodes.txt --listen 127.0.0.4:9001 --daemon & -sleep 3 -killall ctdb_test diff --git a/source4/cluster/ctdb/tests/test1.sh b/source4/cluster/ctdb/tests/test1.sh index 64d5c9fc87..42256f22f9 100755 --- a/source4/cluster/ctdb/tests/test1.sh +++ b/source4/cluster/ctdb/tests/test1.sh @@ -1,8 +1,9 @@ #!/bin/sh -echo "Testing local send" -bin/ctdb_test --nlist tests/1node.txt --listen 127.0.0.1:9001 echo "Testing daemon mode" -bin/ctdb_test --nlist tests/1node.txt --listen 127.0.0.1:9001 --daemon +bin/ctdb_test --nlist tests/1node.txt --listen 127.0.0.1:9001 +wait + echo "Testing self connect" bin/ctdb_test --nlist tests/1node.txt --listen 127.0.0.1:9001 --self-connect +wait diff --git a/source4/cluster/ctdb/tools/ctdb_status.c b/source4/cluster/ctdb/tools/ctdb_status.c new file mode 100644 index 0000000000..af4c62488d --- /dev/null +++ b/source4/cluster/ctdb/tools/ctdb_status.c @@ -0,0 +1,133 @@ +/* + ctdb status tool + + Copyright (C) Andrew Tridgell 2007 + + 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 "cmdline.h" +#include "../include/ctdb_private.h" + + +/* + display status structure + */ +static void show_status(struct ctdb_status *s) +{ + printf(" client_packets_sent %u\n", s->client_packets_sent); + printf(" client_packets_recv %u\n", s->client_packets_recv); + printf(" req_call %u\n", s->client.req_call); + printf(" req_message %u\n", s->client.req_message); + printf(" req_finished %u\n", s->client.req_finished); + printf(" req_register %u\n", s->client.req_register); + printf(" req_connect_wait %u\n", s->client.req_connect_wait); + printf(" req_shutdown %u\n", s->client.req_shutdown); + printf(" req_status %u\n", s->client.req_status); + printf(" node_packets_sent %u\n", s->node_packets_sent); + printf(" node_packets_recv %u\n", s->node_packets_recv); + printf(" req_call %u\n", s->client.req_call); + printf(" reply_call %u\n", s->count.reply_call); + printf(" reply_redirect %u\n", s->count.reply_redirect); + printf(" req_dmaster %u\n", s->count.req_dmaster); + printf(" reply_dmaster %u\n", s->count.reply_dmaster); + printf(" reply_error %u\n", s->count.reply_error); + printf(" reply_redirect %u\n", s->count.reply_redirect); + printf(" req_message %u\n", s->count.req_message); + printf(" req_finished %u\n", s->count.req_finished); + printf(" total_calls %u\n", s->total_calls); + printf(" pending_calls %u\n", s->pending_calls); + printf(" lockwait_calls %u\n", s->lockwait_calls); + printf(" pending_lockwait_calls %u\n", s->pending_lockwait_calls); + printf(" max_call_latency %.6f seconds\n", s->max_call_latency); + printf(" max_lockwait_latency %.6f seconds\n", s->max_lockwait_latency); +} + +/* + show usage message + */ +static void usage(void) +{ + printf("Usage: ctdb_status <socketpath>\n"); + exit(1); +} + +/* + main program +*/ +int main(int argc, const char *argv[]) +{ + struct ctdb_context *ctdb; + struct poptOption popt_options[] = { + POPT_AUTOHELP + POPT_CTDB_CMDLINE + POPT_TABLEEND + }; + int opt; + const char **extra_argv; + int extra_argc = 0; + int ret; + poptContext pc; + struct event_context *ev; + const char *ctdb_socket; + struct ctdb_status status; + + 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 (extra_argc < 1) { + usage(); + } + + ctdb_socket = extra_argv[0]; + + ev = event_context_init(NULL); + + /* initialise ctdb */ + ctdb = ctdb_cmdline_client(ev, ctdb_socket); + if (ctdb == NULL) { + printf("Failed to init ctdb\n"); + exit(1); + } + + ret = ctdb_status(ctdb, &status); + if (ret != 0) { + printf("Failed to get ctdb status\n"); + exit(1); + } + + show_status(&status); + + return 0; +} |