ctdb/tests/src/comm_test.c

   1 /*
   2    comm tests
   3
   4    Copyright (C) Amitay Isaacs  2015
   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/filesys.h"
  22
  23 #include <assert.h>
  24
  25 #include "common/pkt_read.c"
  26 #include "common/pkt_write.c"
  27 #include "common/comm.c"
  28
  29 /*
  30  * Test read_handler and dead_handler
  31  */
  32
  33 static void test1_read_handler(uint8_t *buf, size_t buflen,
  34                                void *private_data)
  35 {
  36         int *result = (int *)private_data;
  37
  38         *result = -1;
  39 }
  40
  41 static void test1_dead_handler(void *private_data)
  42 {
  43         int *result = (int *)private_data;
  44
  45         *result = 1;
  46 }
  47
  48 static void test1(void)
  49 {
  50         TALLOC_CTX *mem_ctx;
  51         struct tevent_context *ev;
  52         struct comm_context *comm;
  53         int fd[2];
  54         int result = 0;
  55         uint32_t data[2];
  56         int ret;
  57         ssize_t n;
  58
  59         mem_ctx = talloc_new(NULL);
  60         assert(mem_ctx != NULL);
  61
  62         ev = tevent_context_init(mem_ctx);
  63         assert(ev != NULL);
  64
  65         ret = pipe(fd);
  66         assert(ret == 0);
  67
  68         ret = comm_setup(ev, ev, fd[0], test1_read_handler, &result,
  69                          test1_dead_handler, &result, &comm);
  70         assert(ret == 0);
  71
  72         data[0] = 2 * sizeof(uint32_t);
  73         data[1] = 0;
  74
  75         n = write(fd[1], (void *)&data, data[0]);
  76         assert(n == data[0]);
  77
  78         while (result == 0) {
  79                 tevent_loop_once(ev);
  80         }
  81
  82         assert(result == -1);
  83
  84         result = 0;
  85         close(fd[1]);
  86
  87         while (result == 0) {
  88                 tevent_loop_once(ev);
  89         }
  90
  91         assert(result == 1);
  92
  93         talloc_free(mem_ctx);
  94 }
  95
  96 /*
  97  * Test that the tevent_req returned by comm_write_send() can be free'd.
  98  */
  99
 100 struct test2_state {
 101         TALLOC_CTX *mem_ctx;
 102         bool done;
 103 };
 104
 105 static void test2_read_handler(uint8_t *buf, size_t buflen,
 106                                void *private_data)
 107 {
 108         struct test2_state *state = (struct test2_state *)private_data;
 109
 110         TALLOC_FREE(state->mem_ctx);
 111 }
 112
 113 static void test2_dead_handler(void *private_data)
 114 {
 115         abort();
 116 }
 117
 118 struct test2_write_state {
 119         int count;
 120 };
 121
 122 static void test2_write_done(struct tevent_req *subreq);
 123
 124 static struct tevent_req *test2_write_send(TALLOC_CTX *mem_ctx,
 125                                            struct tevent_context *ev,
 126                                            struct comm_context *comm,
 127                                            uint8_t *buf, size_t buflen)
 128 {
 129         struct tevent_req *req, *subreq;
 130         struct test2_write_state *state;
 131         int i;
 132
 133         req = tevent_req_create(mem_ctx, &state, struct test2_write_state);
 134         if (req == NULL) {
 135                 return NULL;
 136         }
 137
 138         state->count = 0;
 139
 140         for (i=0; i<10; i++) {
 141                 subreq = comm_write_send(state, ev, comm, buf, buflen);
 142                 if (tevent_req_nomem(subreq, req)) {
 143                         return tevent_req_post(req, ev);
 144                 }
 145                 tevent_req_set_callback(subreq, test2_write_done, req);
 146         }
 147
 148         return req;
 149 }
 150
 151 static void test2_write_done(struct tevent_req *subreq)
 152 {
 153         struct tevent_req *req = tevent_req_callback_data(
 154                 subreq, struct tevent_req);
 155         struct test2_write_state *state = tevent_req_data(
 156                 req, struct test2_write_state);
 157         bool status;
 158         int ret;
 159
 160         status = comm_write_recv(subreq, &ret);
 161         TALLOC_FREE(subreq);
 162         if (! status) {
 163                 tevent_req_error(req, ret);
 164                 return;
 165         }
 166
 167         state->count += 1;
 168
 169         if (state->count == 10) {
 170                 tevent_req_done(req);
 171         }
 172 }
 173
 174 static void test2_timer_handler(struct tevent_context *ev,
 175                                 struct tevent_timer *te,
 176                                 struct timeval cur_time,
 177                                 void *private_data)
 178 {
 179         struct test2_state *state = (struct test2_state *)private_data;
 180
 181         state->done = true;
 182 }
 183
 184 static void test2(void)
 185 {
 186         TALLOC_CTX *mem_ctx;
 187         struct tevent_context *ev;
 188         struct comm_context *comm_reader, *comm_writer;
 189         struct test2_state test2_state;
 190         struct tevent_req *req;
 191         struct tevent_timer *te;
 192         int fd[2];
 193         uint32_t data[2];
 194         int ret;
 195
 196         mem_ctx = talloc_new(NULL);
 197         assert(mem_ctx != NULL);
 198
 199         test2_state.mem_ctx = talloc_new(mem_ctx);
 200         assert(test2_state.mem_ctx != NULL);
 201
 202         test2_state.done = false;
 203
 204         ev = tevent_context_init(mem_ctx);
 205         assert(ev != NULL);
 206
 207         ret = pipe(fd);
 208         assert(ret == 0);
 209
 210         ret = comm_setup(ev, ev, fd[0], test2_read_handler, &test2_state,
 211                          test2_dead_handler, NULL, &comm_reader);
 212         assert(ret == 0);
 213
 214         ret = comm_setup(ev, ev, fd[1], NULL, NULL, test2_dead_handler, NULL,
 215                          &comm_writer);
 216         assert(ret == 0);
 217
 218         data[0] = 2 * sizeof(uint32_t);
 219         data[1] = 0;
 220
 221         req = test2_write_send(test2_state.mem_ctx, ev, comm_writer,
 222                                (uint8_t *)data, data[0]);
 223         assert(req != NULL);
 224
 225         te = tevent_add_timer(ev, ev, tevent_timeval_current_ofs(5,0),
 226                               test2_timer_handler, &test2_state);
 227         assert(te != NULL);
 228
 229         while (! test2_state.done) {
 230                 tevent_loop_once(ev);
 231         }
 232
 233         talloc_free(mem_ctx);
 234 }
 235
 236 /*
 237  * Test that data is written and read correctly.
 238  */
 239
 240 static void test3_dead_handler(void *private_data)
 241 {
 242         int dead_data = *(int *)private_data;
 243
 244         assert(dead_data == 1 || dead_data == 2);
 245
 246         if (dead_data == 1) {
 247                 /* reader */
 248                 fprintf(stderr, "writer closed pipe\n");
 249         } else {
 250                 /* writer */
 251                 fprintf(stderr, "reader closed pipe\n");
 252         }
 253 }
 254
 255 struct test3_writer_state {
 256         struct tevent_context *ev;
 257         struct comm_context *comm;
 258         uint8_t *buf;
 259         size_t *pkt_size;
 260         int count, id;
 261 };
 262
 263 static void test3_writer_next(struct tevent_req *subreq);
 264
 265 static struct tevent_req *test3_writer_send(TALLOC_CTX *mem_ctx,
 266                                             struct tevent_context *ev,
 267                                             struct comm_context *comm,
 268                                             size_t *pkt_size, size_t count)
 269 {
 270         struct tevent_req *req, *subreq;
 271         struct test3_writer_state *state;
 272         size_t max_size = 0, buflen;
 273         size_t i;
 274
 275         for (i=0; i<count; i++) {
 276                 if (pkt_size[i] > max_size) {
 277                         max_size = pkt_size[i];
 278                 }
 279         }
 280
 281         req = tevent_req_create(mem_ctx, &state, struct test3_writer_state);
 282         if (req == NULL) {
 283                 return NULL;
 284         }
 285
 286         state->ev = ev;
 287         state->comm = comm;
 288         state->pkt_size = pkt_size;
 289         state->count = count;
 290         state->id = 0;
 291
 292         state->buf = talloc_array(state, uint8_t, max_size);
 293         if (state->buf == NULL) {
 294                 talloc_free(req);
 295                 return NULL;
 296         }
 297         for (i=0; i<max_size; i++) {
 298                 state->buf[i] = i%256;
 299         }
 300
 301         buflen = state->pkt_size[state->id];
 302         *(uint32_t *)state->buf = buflen;
 303         subreq = comm_write_send(state, state->ev, state->comm,
 304                                          state->buf, buflen);
 305         if (tevent_req_nomem(subreq, req)) {
 306                 return tevent_req_post(req, ev);
 307         }
 308
 309         tevent_req_set_callback(subreq, test3_writer_next, req);
 310         return req;
 311 }
 312
 313 static void test3_writer_next(struct tevent_req *subreq)
 314 {
 315         struct tevent_req *req = tevent_req_callback_data(
 316                 subreq, struct tevent_req);
 317         struct test3_writer_state *state = tevent_req_data(
 318                 req, struct test3_writer_state);
 319         bool ret;
 320         int err;
 321         size_t buflen;
 322
 323         ret = comm_write_recv(subreq, &err);
 324         TALLOC_FREE(subreq);
 325         if (!ret) {
 326                 tevent_req_error(req, err);
 327                 return;
 328         }
 329
 330         state->id++;
 331         if (state->id >= state->count) {
 332                 tevent_req_done(req);
 333                 return;
 334         }
 335
 336         buflen = state->pkt_size[state->id];
 337         *(uint32_t *)state->buf = buflen;
 338         subreq = comm_write_send(state, state->ev, state->comm,
 339                                          state->buf, buflen);
 340         if (tevent_req_nomem(subreq, req)) {
 341                 return;
 342         }
 343
 344         tevent_req_set_callback(subreq, test3_writer_next, req);
 345 }
 346
 347 static void test3_writer_recv(struct tevent_req *req, int *perr)
 348 {
 349         if (tevent_req_is_unix_error(req, perr)) {
 350                 return;
 351         }
 352         *perr = 0;
 353 }
 354
 355 static void test3_writer(int fd, size_t *pkt_size, size_t count)
 356 {
 357         TALLOC_CTX *mem_ctx;
 358         struct tevent_context *ev;
 359         struct comm_context *comm;
 360         struct tevent_req *req;
 361         int dead_data = 2;
 362         int err;
 363
 364         mem_ctx = talloc_new(NULL);
 365         assert(mem_ctx != NULL);
 366
 367         ev = tevent_context_init(mem_ctx);
 368         assert(ev != NULL);
 369
 370         err = comm_setup(mem_ctx, ev, fd, NULL, NULL,
 371                          test3_dead_handler, &dead_data, &comm);
 372         assert(err == 0);
 373         assert(comm != NULL);
 374
 375         req = test3_writer_send(mem_ctx, ev, comm, pkt_size, count);
 376         assert(req != NULL);
 377
 378         tevent_req_poll(req, ev);
 379
 380         test3_writer_recv(req, &err);
 381         assert(err == 0);
 382
 383         talloc_free(mem_ctx);
 384 }
 385
 386 struct test3_reader_state {
 387         size_t *pkt_size;
 388         int count, received;
 389         bool done;
 390 };
 391
 392 static void test3_reader_handler(uint8_t *buf, size_t buflen,
 393                                  void *private_data)
 394 {
 395         struct test3_reader_state *state = talloc_get_type_abort(
 396                 private_data, struct test3_reader_state);
 397
 398         assert(buflen == state->pkt_size[state->received]);
 399         printf("%zi ", buflen);
 400         state->received++;
 401
 402         if (state->received == state->count) {
 403                 printf("\n");
 404                 state->done = true;
 405         }
 406 }
 407
 408 static void test3_reader(int fd, size_t *pkt_size, int count)
 409 {
 410         TALLOC_CTX *mem_ctx;
 411         struct tevent_context *ev;
 412         struct comm_context *comm;
 413         struct test3_reader_state *state;
 414         int dead_data = 1;
 415         int err;
 416
 417         mem_ctx = talloc_new(NULL);
 418         assert(mem_ctx != NULL);
 419
 420         ev = tevent_context_init(mem_ctx);
 421         assert(ev != NULL);
 422
 423         state = talloc_zero(mem_ctx, struct test3_reader_state);
 424         assert(state != NULL);
 425
 426         state->pkt_size = pkt_size;
 427         state->count = count;
 428         state->received = 0;
 429         state->done = false;
 430
 431         err = comm_setup(mem_ctx, ev, fd, test3_reader_handler, state,
 432                          test3_dead_handler, &dead_data, &comm);
 433         assert(err == 0);
 434         assert(comm != NULL);
 435
 436         while (!state->done) {
 437                 tevent_loop_once(ev);
 438         }
 439
 440         talloc_free(mem_ctx);
 441 }
 442
 443 static void test3(void)
 444 {
 445         int fd[2];
 446         int ret;
 447         pid_t pid;
 448         size_t pkt_size[13] = { 100, 2048, 500, 4096, 1024, 8192,
 449                               200, 16384, 300, 32768, 400, 65536,
 450                               1024*1024 };
 451
 452         ret = pipe(fd);
 453         assert(ret == 0);
 454
 455         pid = fork();
 456         assert(pid != -1);
 457
 458         if (pid == 0) {
 459                 /* Child process */
 460                 close(fd[0]);
 461                 test3_writer(fd[1], pkt_size, 13);
 462                 close(fd[1]);
 463                 exit(0);
 464         }
 465
 466         close(fd[1]);
 467         test3_reader(fd[0], pkt_size, 13);
 468         close(fd[0]);
 469 }
 470
 471
 472 int main(int argc, const char **argv)
 473 {
 474         int num;
 475
 476         if (argc != 2) {
 477                 fprintf(stderr, "%s <testnum>\n", argv[0]);
 478                 exit(1);
 479         }
 480
 481         num = atoi(argv[1]);
 482
 483         switch (num) {
 484         case 1:
 485                 test1();
 486                 break;
 487
 488         case 2:
 489                 test2();
 490                 break;
 491
 492         case 3:
 493                 test3();
 494                 break;
 495
 496         default:
 497                 fprintf(stderr, "Unknown test number %s\n", argv[1]);
 498         }
 499
 500         return 0;
 501 }