diff --git a/ctdb/tests/simple/52_ctdb_fetch.sh b/ctdb/tests/simple/52_ctdb_fetch.sh index 23c33bfde67..60beef2c17a 100755 --- a/ctdb/tests/simple/52_ctdb_fetch.sh +++ b/ctdb/tests/simple/52_ctdb_fetch.sh @@ -38,7 +38,7 @@ try_command_on_node 0 "$CTDB listnodes" num_nodes=$(echo "$out" | wc -l) echo "Running ctdb_fetch on all $num_nodes nodes." -try_command_on_node -v -p all $CTDB_TEST_WRAPPER $VALGRIND ctdb_fetch -n $num_nodes +try_command_on_node -v -p all $CTDB_TEST_WRAPPER $VALGRIND fetch_ring -n $num_nodes pat='^(Fetch: [[:digit:]]+(\.[[:digit:]]+)? msgs/sec[[:space:]]?|msg_count=[[:digit:]]+ on node [[:digit:]]|Fetching final record|DATA:|Test data|Waiting for cluster[[:space:]]?|.*: Reqid wrap!|Sleeping for [[:digit:]]+ seconds|)+$' sanity_check_output 1 "$pat" "$out" diff --git a/ctdb/tests/src/ctdb_fetch.c b/ctdb/tests/src/ctdb_fetch.c deleted file mode 100644 index 600504b82b2..00000000000 --- a/ctdb/tests/src/ctdb_fetch.c +++ /dev/null @@ -1,312 +0,0 @@ -/* - simple ctdb benchmark - - Copyright (C) Andrew Tridgell 2006 - - This program is free software; you can redistribute it and/or modify - it under the terms of the GNU General Public License as published by - the Free Software Foundation; either version 3 of the License, or - (at your option) any later version. - - This program is distributed in the hope that it will be useful, - but WITHOUT ANY WARRANTY; without even the implied warranty of - MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - GNU General Public License for more details. - - You should have received a copy of the GNU General Public License - along with this program; if not, see . -*/ - -#include "replace.h" -#include "system/filesys.h" -#include "system/network.h" - -#include -#include -/* Allow use of deprecated function tevent_loop_allow_nesting() */ -#define TEVENT_DEPRECATED -#include -#include - -#include "lib/util/time.h" - -#include "ctdb_private.h" -#include "ctdb_client.h" - -#include "common/cmdline.h" -#include "common/common.h" - - -static struct timeval tp1,tp2; - -static void start_timer(void) -{ - gettimeofday(&tp1,NULL); -} - -static double end_timer(void) -{ - gettimeofday(&tp2,NULL); - return (tp2.tv_sec + (tp2.tv_usec*1.0e-6)) - - (tp1.tv_sec + (tp1.tv_usec*1.0e-6)); -} - - -static int timelimit = 10; -static int num_records = 10; -static int num_nodes; - -struct bench_data { - struct ctdb_context *ctdb; - struct tevent_context *ev; - int msg_count; -}; - -#define TESTKEY "testkey" - -/* - fetch a record - store a expanded record - send a message to next node to tell it to do the same -*/ -static void bench_fetch_1node(struct bench_data *bdata) -{ - struct ctdb_context *ctdb = bdata->ctdb; - TDB_DATA key, data, nulldata; - 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(TESTKEY); - - ctdb_db = ctdb_db_handle(ctdb, "test.tdb"); - - 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_pnn(ctdb)); - talloc_free(tmp_ctx); - return; - } - - if (data.dsize > 1000) { - data.dsize = 0; - } - - if (data.dsize == 0) { - data.dptr = (uint8_t *)talloc_asprintf(tmp_ctx, "Test data\n"); - } - data.dptr = (uint8_t *)talloc_asprintf_append((char *)data.dptr, - "msg_count=%d on node %d\n", - bdata->msg_count, - ctdb_get_pnn(ctdb)); - if (data.dptr == NULL) { - printf("Failed to create record\n"); - talloc_free(tmp_ctx); - return; - } - data.dsize = strlen((const char *)data.dptr)+1; - - ret = ctdb_record_store(h, data); - talloc_free(h); - if (ret != 0) { - printf("Failed to store record\n"); - } - - talloc_free(tmp_ctx); - - /* tell the next node to do the same */ - nulldata.dptr = NULL; - nulldata.dsize = 0; - - dest = (ctdb_get_pnn(ctdb) + 1) % num_nodes; - ctdb_client_send_message(ctdb, dest, 0, nulldata); -} - -/* - handler for messages in bench_ring() -*/ -static void message_handler(uint64_t srvid, TDB_DATA data, void *private_data) -{ - struct bench_data *bdata = talloc_get_type_abort( - private_data, struct bench_data); - - bdata->msg_count++; - bench_fetch_1node(bdata); -} - - -/* - * timeout handler - noop - */ -static void timeout_handler(struct tevent_context *ev, - struct tevent_timer *timer, - struct timeval curtime, void *private_data) -{ - return; -} - -/* - benchmark the following: - - fetch a record - store a expanded record - send a message to next node to tell it to do the same - -*/ -static void bench_fetch(struct bench_data *bdata) -{ - struct ctdb_context *ctdb = bdata->ctdb; - int pnn=ctdb_get_pnn(ctdb); - - if (pnn == num_nodes - 1) { - bench_fetch_1node(bdata); - } - - start_timer(); - tevent_add_timer(bdata->ev, bdata, timeval_current_ofs(timelimit,0), - timeout_handler, NULL); - - while (end_timer() < timelimit) { - if (pnn == 0 && bdata->msg_count % 100 == 0 && end_timer() > 0) { - printf("Fetch: %.2f msgs/sec\r", bdata->msg_count/end_timer()); - fflush(stdout); - } - if (tevent_loop_once(bdata->ev) != 0) { - printf("Event loop failed!\n"); - break; - } - } - - printf("Fetch: %.2f msgs/sec\n", bdata->msg_count/end_timer()); -} - -/* - handler for reconfigure message -*/ -static void reconfigure_handler(uint64_t srvid, TDB_DATA data, - void *private_data) -{ - int *ready = (int *)private_data; - *ready = 1; -} - -/* - main program -*/ -int main(int argc, const char *argv[]) -{ - struct ctdb_context *ctdb; - struct ctdb_db_context *ctdb_db; - - struct poptOption popt_options[] = { - POPT_AUTOHELP - POPT_CTDB_CMDLINE - { "timelimit", 't', POPT_ARG_INT, &timelimit, 0, "timelimit", "integer" }, - { "num-records", 'r', POPT_ARG_INT, &num_records, 0, "num_records", "integer" }, - { NULL, 'n', POPT_ARG_INT, &num_nodes, 0, "num_nodes", "integer" }, - POPT_TABLEEND - }; - int opt; - const char **extra_argv; - int extra_argc = 0; - poptContext pc; - struct tevent_context *ev; - TDB_DATA key, data; - struct ctdb_record_handle *h; - int cluster_ready=0; - struct bench_data *bdata; - - 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); - } - } - - /* talloc_enable_leak_report_full(); */ - - /* 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 (num_nodes == 0) { - printf("You must specify the number of nodes\n"); - exit(1); - } - - ev = tevent_context_init(NULL); - tevent_loop_allow_nesting(ev); - - ctdb = ctdb_cmdline_client(ev, timeval_current_ofs(3, 0)); - - if (ctdb == NULL) { - printf("failed to connect to ctdb daemon.\n"); - exit(1); - } - - ctdb_client_set_message_handler(ctdb, CTDB_SRVID_RECONFIGURE, - reconfigure_handler, &cluster_ready); - - /* attach to a specific database */ - ctdb_db = ctdb_attach(ctdb, timeval_current_ofs(2, 0), "test.tdb", - false, 0); - if (!ctdb_db) { - printf("ctdb_attach failed - %s\n", ctdb_errstr(ctdb)); - exit(1); - } - - bdata = talloc_zero(ctdb, struct bench_data); - if (bdata == NULL) { - printf("memory allocation error\n"); - exit(1); - } - - bdata->ctdb = ctdb; - bdata->ev = ev; - - ctdb_client_set_message_handler(ctdb, 0, message_handler, bdata); - - printf("Waiting for cluster\n"); - while (1) { - uint32_t recmode=1; - ctdb_ctrl_getrecmode(ctdb, ctdb, timeval_zero(), CTDB_CURRENT_NODE, &recmode); - if (recmode == 0) break; - tevent_loop_once(ev); - } - - /* This test has a race condition. If CTDB receives the message from previous - * node, before this node has registered for that message, this node will never - * receive that message and will block on receive. Sleeping for some time will - * hopefully ensure that the test program on all the nodes register for messages. - */ - printf("Sleeping for %d seconds\n", num_nodes); - sleep(num_nodes); - bench_fetch(bdata); - - key.dptr = discard_const(TESTKEY); - key.dsize = strlen(TESTKEY); - - printf("Fetching final record\n"); - - h = ctdb_fetch_lock(ctdb_db, ctdb, key, &data); - - if (h == NULL) { - printf("Failed to fetch record '%s' on node %d\n", - (const char *)key.dptr, ctdb_get_pnn(ctdb)); - exit(1); - } - - printf("DATA:\n%s\n", (char *)data.dptr); - - return 0; -} diff --git a/ctdb/tests/src/fetch_ring.c b/ctdb/tests/src/fetch_ring.c new file mode 100644 index 00000000000..d5b21027e7a --- /dev/null +++ b/ctdb/tests/src/fetch_ring.c @@ -0,0 +1,381 @@ +/* + simple ctdb benchmark + + Copyright (C) Amitay Isaacs 2015 + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program; if not, see . +*/ + +#include "replace.h" +#include "system/network.h" + +#include "lib/util/time.h" +#include "lib/util/tevent_unix.h" + +#include "client/client.h" +#include "tests/src/test_options.h" +#include "tests/src/cluster_wait.h" + +#define TESTDB "fetch_ring.tdb" +#define TESTKEY "testkey" + +#define MSG_ID_FETCH 0 + +static uint32_t next_node(struct ctdb_client_context *client, int num_nodes) +{ + return (ctdb_client_pnn(client) + 1) % num_nodes; +} + +struct fetch_ring_state { + struct tevent_context *ev; + struct ctdb_client_context *client; + struct ctdb_db_context *ctdb_db; + int num_nodes; + int timelimit; + TDB_DATA key; + int msg_count; + struct timeval start_time; +}; + +static void fetch_ring_msg_handler(uint64_t srvid, TDB_DATA data, + void *private_data); +static void fetch_ring_wait(struct tevent_req *subreq); +static void fetch_ring_start(struct tevent_req *subreq); +static void fetch_ring_update(struct tevent_req *subreq); +static void fetch_ring_msg_sent(struct tevent_req *subreq); +static void fetch_ring_finish(struct tevent_req *subreq); +static void fetch_ring_final_read(struct tevent_req *subreq); + +static struct tevent_req *fetch_ring_send(TALLOC_CTX *mem_ctx, + struct tevent_context *ev, + struct ctdb_client_context *client, + struct ctdb_db_context *ctdb_db, + int num_nodes, int timelimit) +{ + struct tevent_req *req, *subreq; + struct fetch_ring_state *state; + + req = tevent_req_create(mem_ctx, &state, struct fetch_ring_state); + if (req == NULL) { + return NULL; + } + + state->ev = ev; + state->client = client; + state->ctdb_db = ctdb_db; + state->num_nodes = num_nodes; + state->timelimit = timelimit; + state->key.dptr = discard_const(TESTKEY); + state->key.dsize = strlen(TESTKEY); + + subreq = ctdb_client_set_message_handler_send( + state, ev, client, MSG_ID_FETCH, + fetch_ring_msg_handler, req); + if (tevent_req_nomem(subreq, req)) { + return tevent_req_post(req, ev); + } + tevent_req_set_callback(subreq, fetch_ring_wait, req); + + return req; +} + +static void fetch_ring_msg_handler(uint64_t srvid, TDB_DATA data, + void *private_data) +{ + struct tevent_req *req = talloc_get_type_abort( + private_data, struct tevent_req); + struct fetch_ring_state *state = tevent_req_data( + req, struct fetch_ring_state); + struct tevent_req *subreq; + + state->msg_count += 1; + + subreq = ctdb_fetch_lock_send(state, state->ev, state->client, + state->ctdb_db, state->key, false); + if (tevent_req_nomem(subreq, req)) { + return; + } + tevent_req_set_callback(subreq, fetch_ring_update, req); +} + +static void fetch_ring_wait(struct tevent_req *subreq) +{ + struct tevent_req *req = tevent_req_callback_data( + subreq, struct tevent_req); + struct fetch_ring_state *state = tevent_req_data( + req, struct fetch_ring_state); + bool status; + int ret; + + status = ctdb_client_set_message_handler_recv(subreq, &ret); + TALLOC_FREE(subreq); + if (! status) { + tevent_req_error(req, ret); + return; + } + + subreq = cluster_wait_send(state, state->ev, state->client, + state->num_nodes); + if (tevent_req_nomem(subreq, req)) { + return; + } + tevent_req_set_callback(subreq, fetch_ring_start, req); +} + +static void fetch_ring_start(struct tevent_req *subreq) +{ + struct tevent_req *req = tevent_req_callback_data( + subreq, struct tevent_req); + struct fetch_ring_state *state = tevent_req_data( + req, struct fetch_ring_state); + bool status; + int ret; + + status = cluster_wait_recv(subreq, &ret); + TALLOC_FREE(subreq); + if (! status) { + tevent_req_error(req, ret); + return; + } + + state->start_time = tevent_timeval_current(); + + if (ctdb_client_pnn(state->client) == state->num_nodes-1) { + subreq = ctdb_fetch_lock_send(state, state->ev, state->client, + state->ctdb_db, state->key, + false); + if (tevent_req_nomem(subreq, req)) { + return; + } + tevent_req_set_callback(subreq, fetch_ring_update, req); + } + + subreq = tevent_wakeup_send(state, state->ev, + tevent_timeval_current_ofs( + state->timelimit, 0)); + if (tevent_req_nomem(subreq, req)) { + return; + } + tevent_req_set_callback(subreq, fetch_ring_finish, req); + +} + +static void fetch_ring_update(struct tevent_req *subreq) +{ + struct tevent_req *req = tevent_req_callback_data( + subreq, struct tevent_req); + struct fetch_ring_state *state = tevent_req_data( + req, struct fetch_ring_state); + struct ctdb_record_handle *h; + struct ctdb_req_message msg; + TDB_DATA data; + uint32_t pnn; + int ret; + + h = ctdb_fetch_lock_recv(subreq, NULL, state, &data, &ret); + TALLOC_FREE(subreq); + if (h == NULL) { + tevent_req_error(req, ret); + return; + } + + if (data.dsize > 1000) { + TALLOC_FREE(data.dptr); + data.dsize = 0; + } + + if (data.dsize == 0) { + data.dptr = (uint8_t *)talloc_asprintf(state, "Test data\n"); + if (tevent_req_nomem(data.dptr, req)) { + return; + } + } + + data.dptr = (uint8_t *)talloc_asprintf_append( + (char *)data.dptr, + "msg_count=%d on node %d\n", + state->msg_count, + ctdb_client_pnn(state->client)); + if (tevent_req_nomem(data.dptr, req)) { + return; + } + + data.dsize = strlen((const char *)data.dptr) + 1; + + ret = ctdb_store_record(h, data); + if (ret != 0) { + tevent_req_error(req, ret); + return; + } + + talloc_free(data.dptr); + talloc_free(h); + + msg.srvid = MSG_ID_FETCH; + msg.data.data = tdb_null; + + pnn = next_node(state->client, state->num_nodes); + + subreq = ctdb_client_message_send(state, state->ev, state->client, + pnn, &msg); + if (tevent_req_nomem(subreq, req)) { + return; + } + tevent_req_set_callback(subreq, fetch_ring_msg_sent, req); +} + +static void fetch_ring_msg_sent(struct tevent_req *subreq) +{ + struct tevent_req *req = tevent_req_callback_data( + subreq, struct tevent_req); + bool status; + int ret; + + status = ctdb_client_message_recv(subreq, &ret); + TALLOC_FREE(subreq); + if (! status) { + tevent_req_error(req, ret); + } +} + +static void fetch_ring_finish(struct tevent_req *subreq) +{ + struct tevent_req *req = tevent_req_callback_data( + subreq, struct tevent_req); + struct fetch_ring_state *state = tevent_req_data( + req, struct fetch_ring_state); + bool status; + double t; + + status = tevent_wakeup_recv(subreq); + TALLOC_FREE(subreq); + if (! status) { + tevent_req_error(req, EIO); + return; + } + + t = timeval_elapsed(&state->start_time); + + printf("Fetch: %.2f msgs/sec\n", state->msg_count / t); + + subreq = ctdb_fetch_lock_send(state, state->ev, state->client, + state->ctdb_db, state->key, false); + if (tevent_req_nomem(subreq, req)) { + return; + } + tevent_req_set_callback(subreq, fetch_ring_final_read, req); +} + +static void fetch_ring_final_read(struct tevent_req *subreq) +{ + struct tevent_req *req = tevent_req_callback_data( + subreq, struct tevent_req); + struct fetch_ring_state *state = tevent_req_data( + req, struct fetch_ring_state); + struct ctdb_record_handle *h; + TDB_DATA data; + int err; + + h = ctdb_fetch_lock_recv(subreq, NULL, state, &data, &err); + TALLOC_FREE(subreq); + if (h == NULL) { + tevent_req_error(req, err); + return; + } + + printf("DATA:\n%s\n", (char *)data.dptr); + talloc_free(data.dptr); + talloc_free(h); + + tevent_req_done(req); +} + +static bool fetch_ring_recv(struct tevent_req *req, int *perr) +{ + int err; + + if (tevent_req_is_unix_error(req, &err)) { + if (perr != NULL) { + *perr = err; + } + return false; + } + return true; +} + +int main(int argc, const char *argv[]) +{ + const struct test_options *opts; + TALLOC_CTX *mem_ctx; + struct tevent_context *ev; + struct ctdb_client_context *client; + struct ctdb_db_context *ctdb_db; + struct tevent_req *req; + int ret; + bool status; + + status = process_options_basic(argc, argv, &opts); + if (! status) { + exit(1); + } + + mem_ctx = talloc_new(NULL); + if (mem_ctx == NULL) { + fprintf(stderr, "Memory allocation error\n"); + exit(1); + } + + ev = tevent_context_init(mem_ctx); + if (ev == NULL) { + fprintf(stderr, "Memory allocation error\n"); + exit(1); + } + + ret = ctdb_client_init(mem_ctx, ev, opts->socket, &client); + if (ret != 0) { + fprintf(stderr, "Failed to initialize client, %s\n", + strerror(ret)); + exit(1); + } + + if (! ctdb_recovery_wait(ev, client)) { + fprintf(stderr, "Memory allocation error\n"); + exit(1); + } + + ret = ctdb_attach(ev, client, tevent_timeval_zero(), TESTDB, 0, + &ctdb_db); + if (ret != 0) { + fprintf(stderr, "Failed to attach to DB %s\n", TESTDB); + exit(1); + } + + req = fetch_ring_send(mem_ctx, ev, client, ctdb_db, + opts->num_nodes, opts->timelimit); + if (req == NULL) { + fprintf(stderr, "Memory allocation error\n"); + exit(1); + } + + tevent_req_poll(req, ev); + + status = fetch_ring_recv(req, NULL); + if (! status) { + fprintf(stderr, "fetch ring test failed\n"); + exit(1); + } + + talloc_free(mem_ctx); + return 0; +} diff --git a/ctdb/wscript b/ctdb/wscript index 3b8af4d913a..79e1ca87598 100755 --- a/ctdb/wscript +++ b/ctdb/wscript @@ -677,7 +677,6 @@ def build(bld): # Test binaries ctdb_tests = [ 'rb_test', - 'ctdb_fetch', 'ctdb_fetch_one', 'ctdb_fetch_readonly_once', 'ctdb_fetch_readonly_loop', @@ -710,7 +709,8 @@ def build(bld): ctdb_tests = [ 'g_lock_loop', - 'message_ring' + 'message_ring', + 'fetch_ring' ] for target in ctdb_tests: