ctdb/tests/src/transaction_loop.c

   1 /*
   2    simple ctdb benchmark for persistent databases
   3
   4    Copyright (C) Amitay Isaacs  2016
   5
   6    This program is free software; you can redistribute it and/or modify
   7    it under the terms of the GNU General Public License as published by
   8    the Free Software Foundation; either version 3 of the License, or
   9    (at your option) any later version.
  10
  11    This program is distributed in the hope that it will be useful,
  12    but WITHOUT ANY WARRANTY; without even the implied warranty of
  13    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14    GNU General Public License for more details.
  15
  16    You should have received a copy of the GNU General Public License
  17    along with this program; if not, see <http://www.gnu.org/licenses/>.
  18 */
  19
  20 #include "replace.h"
  21 #include "system/network.h"
  22
  23 #include "lib/util/debug.h"
  24 #include "lib/util/tevent_unix.h"
  25
  26 #include "client/client.h"
  27 #include "tests/src/test_options.h"
  28 #include "tests/src/cluster_wait.h"
  29
  30 struct transaction_loop_state {
  31         struct tevent_context *ev;
  32         struct ctdb_client_context *client;
  33         struct ctdb_db_context *ctdb_db;
  34         int num_nodes;
  35         int timelimit;
  36         int interactive;
  37         TDB_DATA key;
  38         uint32_t pnn;
  39         struct ctdb_transaction_handle *h;
  40         uint32_t *old_counter, *counter;
  41         struct tevent_req *subreq;
  42         bool done;
  43 };
  44
  45 static void transaction_loop_start(struct tevent_req *subreq);
  46 static void transaction_loop_started(struct tevent_req *subreq);
  47 static void transaction_loop_committed(struct tevent_req *subreq);
  48 static void transaction_loop_each_second(struct tevent_req *subreq);
  49 static bool transaction_loop_check_counters(struct tevent_req *req);
  50 static void transaction_loop_finish(struct tevent_req *subreq);
  51
  52 static struct tevent_req *transaction_loop_send(
  53                                 TALLOC_CTX *mem_ctx,
  54                                 struct tevent_context *ev,
  55                                 struct ctdb_client_context *client,
  56                                 struct ctdb_db_context *ctdb_db,
  57                                 int num_nodes, int timelimit, int interactive,
  58                                 const char *keystr)
  59 {
  60         struct tevent_req *req, *subreq;
  61         struct transaction_loop_state *state;
  62
  63         req = tevent_req_create(mem_ctx, &state,
  64                                 struct transaction_loop_state);
  65         if (req == NULL) {
  66                 return NULL;
  67         }
  68
  69         state->ev = ev;
  70         state->client = client;
  71         state->ctdb_db = ctdb_db;
  72         state->num_nodes = num_nodes;
  73         state->timelimit = timelimit;
  74         state->interactive = interactive;
  75         state->key.dptr = discard_const(keystr);
  76         state->key.dsize = strlen(keystr);
  77         state->pnn = ctdb_client_pnn(client);
  78         state->old_counter = talloc_zero_array(state, uint32_t, num_nodes);
  79         if (tevent_req_nomem(state->old_counter, req)) {
  80                 return tevent_req_post(req, ev);
  81         }
  82         state->counter = talloc_zero_array(state, uint32_t, num_nodes);
  83         if (tevent_req_nomem(state->counter, req)) {
  84                 return tevent_req_post(req, ev);
  85         }
  86
  87         subreq = cluster_wait_send(state, state->ev, state->client,
  88                                    state->num_nodes);
  89         if (tevent_req_nomem(subreq, req)) {
  90                 return tevent_req_post(req, ev);
  91         }
  92         tevent_req_set_callback(subreq, transaction_loop_start, req);
  93
  94         return req;
  95 }
  96
  97 static void transaction_loop_start(struct tevent_req *subreq)
  98 {
  99         struct tevent_req *req = tevent_req_callback_data(
 100                 subreq, struct tevent_req);
 101         struct transaction_loop_state *state = tevent_req_data(
 102                 req, struct transaction_loop_state);
 103         bool status;
 104         int ret;
 105
 106         status = cluster_wait_recv(subreq, &ret);
 107         TALLOC_FREE(subreq);
 108         if (! status) {
 109                 tevent_req_error(req, ret);
 110                 return;
 111         }
 112
 113         subreq = ctdb_transaction_start_send(state, state->ev, state->client,
 114                                              tevent_timeval_current_ofs(
 115                                                      state->timelimit, 0),
 116                                              state->ctdb_db, false);
 117         if (tevent_req_nomem(subreq, req)) {
 118                 return;
 119         }
 120         tevent_req_set_callback(subreq, transaction_loop_started, req);
 121         state->subreq = subreq;
 122
 123         if (ctdb_client_pnn(state->client) == 0) {
 124                 subreq = tevent_wakeup_send(state, state->ev,
 125                                             tevent_timeval_current_ofs(1, 0));
 126                 if (tevent_req_nomem(subreq, req)) {
 127                         return;
 128                 }
 129                 tevent_req_set_callback(subreq, transaction_loop_each_second,
 130                                         req);
 131         }
 132
 133         subreq = tevent_wakeup_send(state, state->ev,
 134                                     tevent_timeval_current_ofs(
 135                                             state->timelimit, 0));
 136         if (tevent_req_nomem(subreq, req)) {
 137                 return;
 138         }
 139         tevent_req_set_callback(subreq, transaction_loop_finish, req);
 140 }
 141
 142 static void transaction_loop_started(struct tevent_req *subreq)
 143 {
 144         struct tevent_req *req = tevent_req_callback_data(
 145                 subreq, struct tevent_req);
 146         struct transaction_loop_state *state = tevent_req_data(
 147                 req, struct transaction_loop_state);
 148         TDB_DATA data;
 149         int ret;
 150         uint32_t *counter;
 151
 152         state->h = ctdb_transaction_start_recv(subreq, &ret);
 153         TALLOC_FREE(subreq);
 154         state->subreq = NULL;
 155         if (state->h == NULL) {
 156                 fprintf(stderr, "transaction start failed\n");
 157                 tevent_req_error(req, ret);
 158                 return;
 159         }
 160
 161         ret = ctdb_transaction_fetch_record(state->h, state->key,
 162                                             state, &data);
 163         if (ret != 0) {
 164                 fprintf(stderr, "transaction fetch record failed\n");
 165                 tevent_req_error(req, ret);
 166                 return;
 167         }
 168
 169         if (data.dsize < state->num_nodes * sizeof(uint32_t)) {
 170                 TALLOC_FREE(data.dptr);
 171
 172                 data.dsize = state->num_nodes * sizeof(uint32_t);
 173                 data.dptr = (uint8_t *)talloc_zero_array(state, uint32_t,
 174                                                          state->num_nodes);
 175                 if (tevent_req_nomem(data.dptr, req)) {
 176                         return;
 177                 }
 178         }
 179
 180         counter = (uint32_t *)data.dptr;
 181         counter[state->pnn] += 1;
 182         memcpy(state->counter, counter, state->num_nodes * sizeof(uint32_t));
 183
 184         ret = ctdb_transaction_store_record(state->h, state->key, data);
 185         if (ret != 0) {
 186                 fprintf(stderr, "transaction store failed\n");
 187                 tevent_req_error(req, ret);
 188                 return;
 189         }
 190
 191         subreq = ctdb_transaction_commit_send(state, state->ev,
 192                                               tevent_timeval_current_ofs(
 193                                                       state->timelimit, 0),
 194                                               state->h);
 195         if (tevent_req_nomem(subreq, req)) {
 196                 return;
 197         }
 198         tevent_req_set_callback(subreq, transaction_loop_committed, req);
 199         state->subreq = subreq;
 200 }
 201
 202 static void transaction_loop_committed(struct tevent_req *subreq)
 203 {
 204         struct tevent_req *req = tevent_req_callback_data(
 205                 subreq, struct tevent_req);
 206         struct transaction_loop_state *state = tevent_req_data(
 207                 req, struct transaction_loop_state);
 208         int ret;
 209         bool status;
 210
 211         status = ctdb_transaction_commit_recv(subreq, &ret);
 212         TALLOC_FREE(subreq);
 213         state->subreq = NULL;
 214         if (! status) {
 215                 fprintf(stderr, "transaction commit failed - %s\n",
 216                         strerror(ret));
 217                 tevent_req_error(req, ret);
 218                 return;
 219         }
 220
 221         if (state->pnn == 0) {
 222                 if (! transaction_loop_check_counters(req)) {
 223                         return;
 224                 }
 225         }
 226
 227         if (state->done) {
 228                 int i;
 229
 230                 printf("Transaction[%u]: ", ctdb_client_pnn(state->client));
 231                 for (i=0; i<state->num_nodes; i++) {
 232                         printf("%6u ", state->counter[i]);
 233                 }
 234                 printf("\n");
 235
 236                 tevent_req_done(req);
 237
 238                 return;
 239         }
 240
 241         subreq = ctdb_transaction_start_send(state, state->ev, state->client,
 242                                              tevent_timeval_current_ofs(
 243                                                      state->timelimit, 0),
 244                                              state->ctdb_db, false);
 245         if (tevent_req_nomem(subreq, req)) {
 246                 return;
 247         }
 248         tevent_req_set_callback(subreq, transaction_loop_started, req);
 249 }
 250
 251 static void transaction_loop_each_second(struct tevent_req *subreq)
 252 {
 253         struct tevent_req *req = tevent_req_callback_data(
 254                 subreq, struct tevent_req);
 255         struct transaction_loop_state *state = tevent_req_data(
 256                 req, struct transaction_loop_state);
 257         bool status;
 258         int i;
 259
 260         status = tevent_wakeup_recv(subreq);
 261         TALLOC_FREE(subreq);
 262         if (! status) {
 263                 fprintf(stderr, "tevent wakeup failed\n");
 264                 tevent_req_error(req, EIO);
 265                 return;
 266         }
 267
 268         if (state->interactive == 1) {
 269                 printf("Transaction[%u]: ", ctdb_client_pnn(state->client));
 270                 for (i=0; i<state->num_nodes; i++) {
 271                         printf("%6u ", state->counter[i]);
 272                 }
 273                 printf("\n");
 274                 fflush(stdout);
 275         }
 276
 277         subreq = tevent_wakeup_send(state, state->ev,
 278                                     tevent_timeval_current_ofs(1, 0));
 279         if (tevent_req_nomem(subreq, req)) {
 280                 return;
 281         }
 282         tevent_req_set_callback(subreq, transaction_loop_each_second, req);
 283 }
 284
 285 static bool transaction_loop_check_counters(struct tevent_req *req)
 286 {
 287         struct transaction_loop_state *state = tevent_req_data(
 288                 req, struct transaction_loop_state);
 289         int i;
 290         bool monotonous = true;
 291
 292         for (i=0; i<state->num_nodes; i++) {
 293                 if (state->counter[i] < state->old_counter[i]) {
 294                         fprintf(stderr,
 295                                 "Counter reduced for node %d: %u -> %u\n",
 296                                 i, state->old_counter[i], state->counter[i]);
 297                         monotonous = false;
 298                         break;
 299                 }
 300         }
 301
 302         if (monotonous) {
 303                 memcpy(state->old_counter, state->counter,
 304                        state->num_nodes * sizeof(uint32_t));
 305         }
 306
 307         return monotonous;
 308 }
 309
 310 static void transaction_loop_finish(struct tevent_req *subreq)
 311 {
 312         struct tevent_req *req = tevent_req_callback_data(
 313                 subreq, struct tevent_req);
 314         struct transaction_loop_state *state = tevent_req_data(
 315                 req, struct transaction_loop_state);
 316         bool status;
 317
 318         status = tevent_wakeup_recv(subreq);
 319         TALLOC_FREE(subreq);
 320
 321         state->done = true;
 322
 323         if (! status) {
 324                 tevent_req_error(req, EIO);
 325                 return;
 326         }
 327 }
 328
 329 static bool transaction_loop_recv(struct tevent_req *req, int *perr)
 330 {
 331         int err;
 332
 333         if (tevent_req_is_unix_error(req, &err)) {
 334                 if (perr != NULL) {
 335                         *perr = err;
 336                 }
 337                 return false;
 338         }
 339         return true;
 340 }
 341
 342 int main(int argc, const char *argv[])
 343 {
 344         const struct test_options *opts;
 345         TALLOC_CTX *mem_ctx;
 346         struct tevent_context *ev;
 347         struct ctdb_client_context *client;
 348         struct ctdb_db_context *ctdb_db;
 349         struct tevent_req *req;
 350         uint8_t db_flags;
 351         int ret;
 352         bool status;
 353
 354         setup_logging("transaction_loop", DEBUG_STDERR);
 355
 356         status = process_options_database(argc, argv, &opts);
 357         if (! status) {
 358                 exit(1);
 359         }
 360
 361         mem_ctx = talloc_new(NULL);
 362         if (mem_ctx == NULL) {
 363                 fprintf(stderr, "Memory allocation error\n");
 364                 exit(1);
 365         }
 366
 367         ev = tevent_context_init(mem_ctx);
 368         if (ev == NULL) {
 369                 fprintf(stderr, "Memory allocation error\n");
 370                 exit(1);
 371         }
 372
 373         ret = ctdb_client_init(mem_ctx, ev, opts->socket, &client);
 374         if (ret != 0) {
 375                 fprintf(stderr, "Failed to initialize client, ret=%d\n", ret);
 376                 exit(1);
 377         }
 378
 379         if (! ctdb_recovery_wait(ev, client)) {
 380                 fprintf(stderr, "Memory allocation error\n");
 381                 exit(1);
 382         }
 383
 384         if (strcmp(opts->dbtype, "persistent") == 0) {
 385                 db_flags = CTDB_DB_FLAGS_PERSISTENT;
 386         } else if (strcmp(opts->dbtype, "replicated") == 0) {
 387                 db_flags = CTDB_DB_FLAGS_REPLICATED;
 388         } else {
 389                 fprintf(stderr, "Database must be persistent or replicated\n");
 390                 exit(1);
 391         }
 392
 393         ret = ctdb_attach(ev, client, tevent_timeval_zero(), opts->dbname,
 394                           db_flags, &ctdb_db);
 395         if (ret != 0) {
 396                 fprintf(stderr, "Failed to attach to persistent DB %s\n",
 397                         opts->dbname);
 398                 exit(1);
 399         }
 400
 401         req = transaction_loop_send(mem_ctx, ev, client, ctdb_db,
 402                                     opts->num_nodes, opts->timelimit,
 403                                     opts->interactive, opts->keystr);
 404         if (req == NULL) {
 405                 fprintf(stderr, "Memory allocation error\n");
 406                 exit(1);
 407         }
 408
 409         tevent_req_poll(req, ev);
 410
 411         status = transaction_loop_recv(req, &ret);
 412         if (! status) {
 413                 fprintf(stderr, "transaction loop test failed, ret=%d\n", ret);
 414                 exit(1);
 415         }
 416
 417         talloc_free(mem_ctx);
 418         return 0;
 419 }