lib/tevent/testsuite.c

   1 /*
   2    Unix SMB/CIFS implementation.
   3
   4    testing of the events subsystem
   5
   6    Copyright (C) Stefan Metzmacher 2006-2009
   7    Copyright (C) Jeremy Allison    2013
   8
   9      ** NOTE! The following LGPL license applies to the tevent
  10      ** library. This does NOT imply that all of Samba is released
  11      ** under the LGPL
  12
  13    This library is free software; you can redistribute it and/or
  14    modify it under the terms of the GNU Lesser General Public
  15    License as published by the Free Software Foundation; either
  16    version 3 of the License, or (at your option) any later version.
  17
  18    This library is distributed in the hope that it will be useful,
  19    but WITHOUT ANY WARRANTY; without even the implied warranty of
  20    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  21    Lesser General Public License for more details.
  22
  23    You should have received a copy of the GNU Lesser General Public
  24    License along with this library; if not, see <http://www.gnu.org/licenses/>.
  25 */
  26
  27 #include "includes.h"
  28 #include "lib/tevent/tevent.h"
  29 #include "system/filesys.h"
  30 #include "system/select.h"
  31 #include "system/network.h"
  32 #include "torture/torture.h"
  33 #include "torture/local/proto.h"
  34 #ifdef HAVE_PTHREAD
  35 #include <pthread.h>
  36 #include <assert.h>
  37 #endif
  38
  39 static int fde_count;
  40
  41 static void do_read(int fd, void *buf, size_t count)
  42 {
  43         ssize_t ret;
  44
  45         do {
  46                 ret = read(fd, buf, count);
  47         } while (ret == -1 && errno == EINTR);
  48 }
  49
  50 static void fde_handler_read(struct tevent_context *ev_ctx, struct tevent_fd *f,
  51                         uint16_t flags, void *private_data)
  52 {
  53         int *fd = (int *)private_data;
  54         char c;
  55 #ifdef SA_SIGINFO
  56         kill(getpid(), SIGUSR1);
  57 #endif
  58         kill(getpid(), SIGALRM);
  59
  60         do_read(fd[0], &c, 1);
  61         fde_count++;
  62 }
  63
  64 static void do_write(int fd, void *buf, size_t count)
  65 {
  66         ssize_t ret;
  67
  68         do {
  69                 ret = write(fd, buf, count);
  70         } while (ret == -1 && errno == EINTR);
  71 }
  72
  73 static void fde_handler_write(struct tevent_context *ev_ctx, struct tevent_fd *f,
  74                         uint16_t flags, void *private_data)
  75 {
  76         int *fd = (int *)private_data;
  77         char c = 0;
  78
  79         do_write(fd[1], &c, 1);
  80 }
  81
  82
  83 /* This will only fire if the fd's returned from pipe() are bi-directional. */
  84 static void fde_handler_read_1(struct tevent_context *ev_ctx, struct tevent_fd *f,
  85                         uint16_t flags, void *private_data)
  86 {
  87         int *fd = (int *)private_data;
  88         char c;
  89 #ifdef SA_SIGINFO
  90         kill(getpid(), SIGUSR1);
  91 #endif
  92         kill(getpid(), SIGALRM);
  93
  94         do_read(fd[1], &c, 1);
  95         fde_count++;
  96 }
  97
  98 /* This will only fire if the fd's returned from pipe() are bi-directional. */
  99 static void fde_handler_write_1(struct tevent_context *ev_ctx, struct tevent_fd *f,
 100                         uint16_t flags, void *private_data)
 101 {
 102         int *fd = (int *)private_data;
 103         char c = 0;
 104         do_write(fd[0], &c, 1);
 105 }
 106
 107 static void finished_handler(struct tevent_context *ev_ctx, struct tevent_timer *te,
 108                              struct timeval tval, void *private_data)
 109 {
 110         int *finished = (int *)private_data;
 111         (*finished) = 1;
 112 }
 113
 114 static void count_handler(struct tevent_context *ev_ctx, struct tevent_signal *te,
 115                           int signum, int count, void *info, void *private_data)
 116 {
 117         int *countp = (int *)private_data;
 118         (*countp) += count;
 119 }
 120
 121 static bool test_event_context(struct torture_context *test,
 122                                const void *test_data)
 123 {
 124         struct tevent_context *ev_ctx;
 125         int fd[2] = { -1, -1 };
 126         const char *backend = (const char *)test_data;
 127         int alarm_count=0, info_count=0;
 128         struct tevent_fd *fde_read;
 129         struct tevent_fd *fde_read_1;
 130         struct tevent_fd *fde_write;
 131         struct tevent_fd *fde_write_1;
 132 #ifdef SA_RESTART
 133         struct tevent_signal *se1 = NULL;
 134 #endif
 135 #ifdef SA_RESETHAND
 136         struct tevent_signal *se2 = NULL;
 137 #endif
 138 #ifdef SA_SIGINFO
 139         struct tevent_signal *se3 = NULL;
 140 #endif
 141         int finished=0;
 142         struct timeval t;
 143         int ret;
 144
 145         ev_ctx = tevent_context_init_byname(test, backend);
 146         if (ev_ctx == NULL) {
 147                 torture_comment(test, "event backend '%s' not supported\n", backend);
 148                 return true;
 149         }
 150
 151         torture_comment(test, "backend '%s' - %s\n",
 152                         backend, __FUNCTION__);
 153
 154         /* reset globals */
 155         fde_count = 0;
 156
 157         /* create a pipe */
 158         ret = pipe(fd);
 159         torture_assert_int_equal(test, ret, 0, "pipe failed");
 160
 161         fde_read = tevent_add_fd(ev_ctx, ev_ctx, fd[0], TEVENT_FD_READ,
 162                             fde_handler_read, fd);
 163         fde_write_1 = tevent_add_fd(ev_ctx, ev_ctx, fd[0], TEVENT_FD_WRITE,
 164                             fde_handler_write_1, fd);
 165
 166         fde_write = tevent_add_fd(ev_ctx, ev_ctx, fd[1], TEVENT_FD_WRITE,
 167                             fde_handler_write, fd);
 168         fde_read_1 = tevent_add_fd(ev_ctx, ev_ctx, fd[1], TEVENT_FD_READ,
 169                             fde_handler_read_1, fd);
 170
 171         tevent_fd_set_auto_close(fde_read);
 172         tevent_fd_set_auto_close(fde_write);
 173
 174         tevent_add_timer(ev_ctx, ev_ctx, timeval_current_ofs(2,0),
 175                          finished_handler, &finished);
 176
 177 #ifdef SA_RESTART
 178         se1 = tevent_add_signal(ev_ctx, ev_ctx, SIGALRM, SA_RESTART, count_handler, &alarm_count);
 179         torture_assert(test, se1 != NULL, "failed to setup se1");
 180 #endif
 181 #ifdef SA_RESETHAND
 182         se2 = tevent_add_signal(ev_ctx, ev_ctx, SIGALRM, SA_RESETHAND, count_handler, &alarm_count);
 183         torture_assert(test, se2 != NULL, "failed to setup se2");
 184 #endif
 185 #ifdef SA_SIGINFO
 186         se3 = tevent_add_signal(ev_ctx, ev_ctx, SIGUSR1, SA_SIGINFO, count_handler, &info_count);
 187         torture_assert(test, se3 != NULL, "failed to setup se3");
 188 #endif
 189
 190         t = timeval_current();
 191         while (!finished) {
 192                 errno = 0;
 193                 if (tevent_loop_once(ev_ctx) == -1) {
 194                         talloc_free(ev_ctx);
 195                         torture_fail(test, talloc_asprintf(test, "Failed event loop %s\n", strerror(errno)));
 196                 }
 197         }
 198
 199         talloc_free(fde_read_1);
 200         talloc_free(fde_write_1);
 201         talloc_free(fde_read);
 202         talloc_free(fde_write);
 203
 204         while (alarm_count < fde_count+1) {
 205                 if (tevent_loop_once(ev_ctx) == -1) {
 206                         break;
 207                 }
 208         }
 209
 210         torture_comment(test, "Got %.2f pipe events/sec\n", fde_count/timeval_elapsed(&t));
 211
 212 #ifdef SA_RESTART
 213         talloc_free(se1);
 214 #endif
 215
 216         torture_assert_int_equal(test, alarm_count, 1+fde_count, "alarm count mismatch");
 217
 218 #ifdef SA_RESETHAND
 219         /*
 220          * we do not call talloc_free(se2)
 221          * because it is already gone,
 222          * after triggering the event handler.
 223          */
 224 #endif
 225
 226 #ifdef SA_SIGINFO
 227         talloc_free(se3);
 228         torture_assert_int_equal(test, info_count, fde_count, "info count mismatch");
 229 #endif
 230
 231         talloc_free(ev_ctx);
 232
 233         return true;
 234 }
 235
 236 struct test_event_fd1_state {
 237         struct torture_context *tctx;
 238         const char *backend;
 239         struct tevent_context *ev;
 240         int sock[2];
 241         struct tevent_timer *te;
 242         struct tevent_fd *fde0;
 243         struct tevent_fd *fde1;
 244         bool got_write;
 245         bool got_read;
 246         bool drain;
 247         bool drain_done;
 248         unsigned loop_count;
 249         bool finished;
 250         const char *error;
 251 };
 252
 253 static void test_event_fd1_fde_handler(struct tevent_context *ev_ctx,
 254                                        struct tevent_fd *fde,
 255                                        uint16_t flags,
 256                                        void *private_data)
 257 {
 258         struct test_event_fd1_state *state =
 259                 (struct test_event_fd1_state *)private_data;
 260
 261         if (state->drain_done) {
 262                 state->finished = true;
 263                 state->error = __location__;
 264                 return;
 265         }
 266
 267         if (state->drain) {
 268                 ssize_t ret;
 269                 uint8_t c = 0;
 270
 271                 if (!(flags & TEVENT_FD_READ)) {
 272                         state->finished = true;
 273                         state->error = __location__;
 274                         return;
 275                 }
 276
 277                 ret = read(state->sock[0], &c, 1);
 278                 if (ret == 1) {
 279                         return;
 280                 }
 281
 282                 /*
 283                  * end of test...
 284                  */
 285                 tevent_fd_set_flags(fde, 0);
 286                 state->drain_done = true;
 287                 return;
 288         }
 289
 290         if (!state->got_write) {
 291                 uint8_t c = 0;
 292
 293                 if (flags != TEVENT_FD_WRITE) {
 294                         state->finished = true;
 295                         state->error = __location__;
 296                         return;
 297                 }
 298                 state->got_write = true;
 299
 300                 /*
 301                  * we write to the other socket...
 302                  */
 303                 do_write(state->sock[1], &c, 1);
 304                 TEVENT_FD_NOT_WRITEABLE(fde);
 305                 TEVENT_FD_READABLE(fde);
 306                 return;
 307         }
 308
 309         if (!state->got_read) {
 310                 if (flags != TEVENT_FD_READ) {
 311                         state->finished = true;
 312                         state->error = __location__;
 313                         return;
 314                 }
 315                 state->got_read = true;
 316
 317                 TEVENT_FD_NOT_READABLE(fde);
 318                 return;
 319         }
 320
 321         state->finished = true;
 322         state->error = __location__;
 323         return;
 324 }
 325
 326 static void test_event_fd1_finished(struct tevent_context *ev_ctx,
 327                                     struct tevent_timer *te,
 328                                     struct timeval tval,
 329                                     void *private_data)
 330 {
 331         struct test_event_fd1_state *state =
 332                 (struct test_event_fd1_state *)private_data;
 333
 334         if (state->drain_done) {
 335                 state->finished = true;
 336                 return;
 337         }
 338
 339         if (!state->got_write) {
 340                 state->finished = true;
 341                 state->error = __location__;
 342                 return;
 343         }
 344
 345         if (!state->got_read) {
 346                 state->finished = true;
 347                 state->error = __location__;
 348                 return;
 349         }
 350
 351         state->loop_count++;
 352         if (state->loop_count > 3) {
 353                 state->finished = true;
 354                 state->error = __location__;
 355                 return;
 356         }
 357
 358         state->got_write = false;
 359         state->got_read = false;
 360
 361         tevent_fd_set_flags(state->fde0, TEVENT_FD_WRITE);
 362
 363         if (state->loop_count > 2) {
 364                 state->drain = true;
 365                 TALLOC_FREE(state->fde1);
 366                 TEVENT_FD_READABLE(state->fde0);
 367         }
 368
 369         state->te = tevent_add_timer(state->ev, state->ev,
 370                                     timeval_current_ofs(0,2000),
 371                                     test_event_fd1_finished, state);
 372 }
 373
 374 static bool test_event_fd1(struct torture_context *tctx,
 375                            const void *test_data)
 376 {
 377         struct test_event_fd1_state state;
 378
 379         ZERO_STRUCT(state);
 380         state.tctx = tctx;
 381         state.backend = (const char *)test_data;
 382
 383         state.ev = tevent_context_init_byname(tctx, state.backend);
 384         if (state.ev == NULL) {
 385                 torture_skip(tctx, talloc_asprintf(tctx,
 386                              "event backend '%s' not supported\n",
 387                              state.backend));
 388                 return true;
 389         }
 390
 391         tevent_set_debug_stderr(state.ev);
 392         torture_comment(tctx, "backend '%s' - %s\n",
 393                         state.backend, __FUNCTION__);
 394
 395         /*
 396          * This tests the following:
 397          *
 398          * It monitors the state of state.sock[0]
 399          * with tevent_fd, but we never read/write on state.sock[0]
 400          * while state.sock[1] * is only used to write a few bytes.
 401          *
 402          * We have a loop:
 403          *   - we wait only for TEVENT_FD_WRITE on state.sock[0]
 404          *   - we write 1 byte to state.sock[1]
 405          *   - we wait only for TEVENT_FD_READ on state.sock[0]
 406          *   - we disable events on state.sock[0]
 407          *   - the timer event restarts the loop
 408          * Then we close state.sock[1]
 409          * We have a loop:
 410          *   - we wait for TEVENT_FD_READ/WRITE on state.sock[0]
 411          *   - we try to read 1 byte
 412          *   - if the read gets an error of returns 0
 413          *     we disable the event handler
 414          *   - the timer finishes the test
 415          */
 416         state.sock[0] = -1;
 417         state.sock[1] = -1;
 418         socketpair(AF_UNIX, SOCK_STREAM, 0, state.sock);
 419
 420         state.te = tevent_add_timer(state.ev, state.ev,
 421                                     timeval_current_ofs(0,1000),
 422                                     test_event_fd1_finished, &state);
 423         state.fde0 = tevent_add_fd(state.ev, state.ev,
 424                                    state.sock[0], TEVENT_FD_WRITE,
 425                                    test_event_fd1_fde_handler, &state);
 426         /* state.fde1 is only used to auto close */
 427         state.fde1 = tevent_add_fd(state.ev, state.ev,
 428                                    state.sock[1], 0,
 429                                    test_event_fd1_fde_handler, &state);
 430
 431         tevent_fd_set_auto_close(state.fde0);
 432         tevent_fd_set_auto_close(state.fde1);
 433
 434         while (!state.finished) {
 435                 errno = 0;
 436                 if (tevent_loop_once(state.ev) == -1) {
 437                         talloc_free(state.ev);
 438                         torture_fail(tctx, talloc_asprintf(tctx,
 439                                      "Failed event loop %s\n",
 440                                      strerror(errno)));
 441                 }
 442         }
 443
 444         talloc_free(state.ev);
 445
 446         torture_assert(tctx, state.error == NULL, talloc_asprintf(tctx,
 447                        "%s", state.error));
 448
 449         return true;
 450 }
 451
 452 struct test_event_fd2_state {
 453         struct torture_context *tctx;
 454         const char *backend;
 455         struct tevent_context *ev;
 456         struct tevent_timer *te;
 457         struct test_event_fd2_sock {
 458                 struct test_event_fd2_state *state;
 459                 int fd;
 460                 struct tevent_fd *fde;
 461                 size_t num_written;
 462                 size_t num_read;
 463                 bool got_full;
 464         } sock0, sock1;
 465         bool finished;
 466         const char *error;
 467 };
 468
 469 static void test_event_fd2_sock_handler(struct tevent_context *ev_ctx,
 470                                         struct tevent_fd *fde,
 471                                         uint16_t flags,
 472                                         void *private_data)
 473 {
 474         struct test_event_fd2_sock *cur_sock =
 475                 (struct test_event_fd2_sock *)private_data;
 476         struct test_event_fd2_state *state = cur_sock->state;
 477         struct test_event_fd2_sock *oth_sock = NULL;
 478         uint8_t v = 0, c;
 479         ssize_t ret;
 480
 481         if (cur_sock == &state->sock0) {
 482                 oth_sock = &state->sock1;
 483         } else {
 484                 oth_sock = &state->sock0;
 485         }
 486
 487         if (oth_sock->num_written == 1) {
 488                 if (flags != (TEVENT_FD_READ | TEVENT_FD_WRITE)) {
 489                         state->finished = true;
 490                         state->error = __location__;
 491                         return;
 492                 }
 493         }
 494
 495         if (cur_sock->num_read == oth_sock->num_written) {
 496                 state->finished = true;
 497                 state->error = __location__;
 498                 return;
 499         }
 500
 501         if (!(flags & TEVENT_FD_READ)) {
 502                 state->finished = true;
 503                 state->error = __location__;
 504                 return;
 505         }
 506
 507         if (oth_sock->num_read >= PIPE_BUF) {
 508                 /*
 509                  * On Linux we become writable once we've read
 510                  * one byte. On Solaris we only become writable
 511                  * again once we've read 4096 bytes. PIPE_BUF
 512                  * is probably a safe bet to test against.
 513                  *
 514                  * There should be room to write a byte again
 515                  */
 516                 if (!(flags & TEVENT_FD_WRITE)) {
 517                         state->finished = true;
 518                         state->error = __location__;
 519                         return;
 520                 }
 521         }
 522
 523         if ((flags & TEVENT_FD_WRITE) && !cur_sock->got_full) {
 524                 v = (uint8_t)cur_sock->num_written;
 525                 ret = write(cur_sock->fd, &v, 1);
 526                 if (ret != 1) {
 527                         state->finished = true;
 528                         state->error = __location__;
 529                         return;
 530                 }
 531                 cur_sock->num_written++;
 532                 if (cur_sock->num_written > 0x80000000) {
 533                         state->finished = true;
 534                         state->error = __location__;
 535                         return;
 536                 }
 537                 return;
 538         }
 539
 540         if (!cur_sock->got_full) {
 541                 cur_sock->got_full = true;
 542
 543                 if (!oth_sock->got_full) {
 544                         /*
 545                          * cur_sock is full,
 546                          * lets wait for oth_sock
 547                          * to be filled
 548                          */
 549                         tevent_fd_set_flags(cur_sock->fde, 0);
 550                         return;
 551                 }
 552
 553                 /*
 554                  * oth_sock waited for cur_sock,
 555                  * lets restart it
 556                  */
 557                 tevent_fd_set_flags(oth_sock->fde,
 558                                     TEVENT_FD_READ|TEVENT_FD_WRITE);
 559         }
 560
 561         ret = read(cur_sock->fd, &v, 1);
 562         if (ret != 1) {
 563                 state->finished = true;
 564                 state->error = __location__;
 565                 return;
 566         }
 567         c = (uint8_t)cur_sock->num_read;
 568         if (c != v) {
 569                 state->finished = true;
 570                 state->error = __location__;
 571                 return;
 572         }
 573         cur_sock->num_read++;
 574
 575         if (cur_sock->num_read < oth_sock->num_written) {
 576                 /* there is more to read */
 577                 return;
 578         }
 579         /*
 580          * we read everything, we need to remove TEVENT_FD_WRITE
 581          * to avoid spinning
 582          */
 583         TEVENT_FD_NOT_WRITEABLE(cur_sock->fde);
 584
 585         if (oth_sock->num_read == cur_sock->num_written) {
 586                 /*
 587                  * both directions are finished
 588                  */
 589                 state->finished = true;
 590         }
 591
 592         return;
 593 }
 594
 595 static void test_event_fd2_finished(struct tevent_context *ev_ctx,
 596                                     struct tevent_timer *te,
 597                                     struct timeval tval,
 598                                     void *private_data)
 599 {
 600         struct test_event_fd2_state *state =
 601                 (struct test_event_fd2_state *)private_data;
 602
 603         /*
 604          * this should never be triggered
 605          */
 606         state->finished = true;
 607         state->error = __location__;
 608 }
 609
 610 static bool test_event_fd2(struct torture_context *tctx,
 611                            const void *test_data)
 612 {
 613         struct test_event_fd2_state state;
 614         int sock[2];
 615         uint8_t c = 0;
 616
 617         ZERO_STRUCT(state);
 618         state.tctx = tctx;
 619         state.backend = (const char *)test_data;
 620
 621         state.ev = tevent_context_init_byname(tctx, state.backend);
 622         if (state.ev == NULL) {
 623                 torture_skip(tctx, talloc_asprintf(tctx,
 624                              "event backend '%s' not supported\n",
 625                              state.backend));
 626                 return true;
 627         }
 628
 629         tevent_set_debug_stderr(state.ev);
 630         torture_comment(tctx, "backend '%s' - %s\n",
 631                         state.backend, __FUNCTION__);
 632
 633         /*
 634          * This tests the following
 635          *
 636          * - We write 1 byte to each socket
 637          * - We wait for TEVENT_FD_READ/WRITE on both sockets
 638          * - When we get TEVENT_FD_WRITE we write 1 byte
 639          *   until both socket buffers are full, which
 640          *   means both sockets only get TEVENT_FD_READ.
 641          * - Then we read 1 byte until we have consumed
 642          *   all bytes the other end has written.
 643          */
 644         sock[0] = -1;
 645         sock[1] = -1;
 646         socketpair(AF_UNIX, SOCK_STREAM, 0, sock);
 647
 648         /*
 649          * the timer should never expire
 650          */
 651         state.te = tevent_add_timer(state.ev, state.ev,
 652                                     timeval_current_ofs(600, 0),
 653                                     test_event_fd2_finished, &state);
 654         state.sock0.state = &state;
 655         state.sock0.fd = sock[0];
 656         state.sock0.fde = tevent_add_fd(state.ev, state.ev,
 657                                         state.sock0.fd,
 658                                         TEVENT_FD_READ | TEVENT_FD_WRITE,
 659                                         test_event_fd2_sock_handler,
 660                                         &state.sock0);
 661         state.sock1.state = &state;
 662         state.sock1.fd = sock[1];
 663         state.sock1.fde = tevent_add_fd(state.ev, state.ev,
 664                                         state.sock1.fd,
 665                                         TEVENT_FD_READ | TEVENT_FD_WRITE,
 666                                         test_event_fd2_sock_handler,
 667                                         &state.sock1);
 668
 669         tevent_fd_set_auto_close(state.sock0.fde);
 670         tevent_fd_set_auto_close(state.sock1.fde);
 671
 672         do_write(state.sock0.fd, &c, 1);
 673         state.sock0.num_written++;
 674         do_write(state.sock1.fd, &c, 1);
 675         state.sock1.num_written++;
 676
 677         while (!state.finished) {
 678                 errno = 0;
 679                 if (tevent_loop_once(state.ev) == -1) {
 680                         talloc_free(state.ev);
 681                         torture_fail(tctx, talloc_asprintf(tctx,
 682                                      "Failed event loop %s\n",
 683                                      strerror(errno)));
 684                 }
 685         }
 686
 687         talloc_free(state.ev);
 688
 689         torture_assert(tctx, state.error == NULL, talloc_asprintf(tctx,
 690                        "%s", state.error));
 691
 692         return true;
 693 }
 694
 695 #ifdef HAVE_PTHREAD
 696
 697 static pthread_mutex_t threaded_mutex = PTHREAD_MUTEX_INITIALIZER;
 698 static bool do_shutdown = false;
 699
 700 static void test_event_threaded_lock(void)
 701 {
 702         int ret;
 703         ret = pthread_mutex_lock(&threaded_mutex);
 704         assert(ret == 0);
 705 }
 706
 707 static void test_event_threaded_unlock(void)
 708 {
 709         int ret;
 710         ret = pthread_mutex_unlock(&threaded_mutex);
 711         assert(ret == 0);
 712 }
 713
 714 static void test_event_threaded_trace(enum tevent_trace_point point,
 715                                       void *private_data)
 716 {
 717         switch (point) {
 718         case TEVENT_TRACE_BEFORE_WAIT:
 719                 test_event_threaded_unlock();
 720                 break;
 721         case TEVENT_TRACE_AFTER_WAIT:
 722                 test_event_threaded_lock();
 723                 break;
 724         case TEVENT_TRACE_BEFORE_LOOP_ONCE:
 725         case TEVENT_TRACE_AFTER_LOOP_ONCE:
 726                 break;
 727         }
 728 }
 729
 730 static void test_event_threaded_timer(struct tevent_context *ev,
 731                                       struct tevent_timer *te,
 732                                       struct timeval current_time,
 733                                       void *private_data)
 734 {
 735         return;
 736 }
 737
 738 static void *test_event_poll_thread(void *private_data)
 739 {
 740         struct tevent_context *ev = (struct tevent_context *)private_data;
 741
 742         test_event_threaded_lock();
 743
 744         while (true) {
 745                 int ret;
 746                 ret = tevent_loop_once(ev);
 747                 assert(ret == 0);
 748                 if (do_shutdown) {
 749                         test_event_threaded_unlock();
 750                         return NULL;
 751                 }
 752         }
 753
 754 }
 755
 756 static void test_event_threaded_read_handler(struct tevent_context *ev,
 757                                              struct tevent_fd *fde,
 758                                              uint16_t flags,
 759                                              void *private_data)
 760 {
 761         int *pfd = (int *)private_data;
 762         char c;
 763         ssize_t nread;
 764
 765         if ((flags & TEVENT_FD_READ) == 0) {
 766                 return;
 767         }
 768
 769         do {
 770                 nread = read(*pfd, &c, 1);
 771         } while ((nread == -1) && (errno == EINTR));
 772
 773         assert(nread == 1);
 774 }
 775
 776 static bool test_event_context_threaded(struct torture_context *test,
 777                                         const void *test_data)
 778 {
 779         struct tevent_context *ev;
 780         struct tevent_timer *te;
 781         struct tevent_fd *fde;
 782         pthread_t poll_thread;
 783         int fds[2];
 784         int ret;
 785         char c = 0;
 786
 787         ev = tevent_context_init_byname(test, "poll_mt");
 788         torture_assert(test, ev != NULL, "poll_mt not supported");
 789
 790         tevent_set_trace_callback(ev, test_event_threaded_trace, NULL);
 791
 792         te = tevent_add_timer(ev, ev, timeval_current_ofs(5, 0),
 793                               test_event_threaded_timer, NULL);
 794         torture_assert(test, te != NULL, "Could not add timer");
 795
 796         ret = pthread_create(&poll_thread, NULL, test_event_poll_thread, ev);
 797         torture_assert(test, ret == 0, "Could not create poll thread");
 798
 799         ret = pipe(fds);
 800         torture_assert(test, ret == 0, "Could not create pipe");
 801
 802         poll(NULL, 0, 100);
 803
 804         test_event_threaded_lock();
 805
 806         fde = tevent_add_fd(ev, ev, fds[0], TEVENT_FD_READ,
 807                             test_event_threaded_read_handler, &fds[0]);
 808         torture_assert(test, fde != NULL, "Could not add fd event");
 809
 810         test_event_threaded_unlock();
 811
 812         poll(NULL, 0, 100);
 813
 814         do_write(fds[1], &c, 1);
 815
 816         poll(NULL, 0, 100);
 817
 818         test_event_threaded_lock();
 819         do_shutdown = true;
 820         test_event_threaded_unlock();
 821
 822         do_write(fds[1], &c, 1);
 823
 824         ret = pthread_join(poll_thread, NULL);
 825         torture_assert(test, ret == 0, "pthread_join failed");
 826
 827         return true;
 828 }
 829
 830 #define NUM_TEVENT_THREADS 100
 831
 832 /* Ugly, but needed for torture_comment... */
 833 static struct torture_context *thread_test_ctx;
 834 static pthread_t thread_map[NUM_TEVENT_THREADS];
 835 static unsigned thread_counter;
 836
 837 /* Called in master thread context */
 838 static void callback_nowait(struct tevent_context *ev,
 839                                 struct tevent_immediate *im,
 840                                 void *private_ptr)
 841 {
 842         pthread_t *thread_id_ptr =
 843                 talloc_get_type_abort(private_ptr, pthread_t);
 844         unsigned i;
 845
 846         for (i = 0; i < NUM_TEVENT_THREADS; i++) {
 847                 if (pthread_equal(*thread_id_ptr,
 848                                 thread_map[i])) {
 849                         break;
 850                 }
 851         }
 852         torture_comment(thread_test_ctx,
 853                         "Callback %u from thread %u\n",
 854                         thread_counter,
 855                         i);
 856         thread_counter++;
 857 }
 858
 859 /* Blast the master tevent_context with a callback, no waiting. */
 860 static void *thread_fn_nowait(void *private_ptr)
 861 {
 862         struct tevent_thread_proxy *master_tp =
 863                 talloc_get_type_abort(private_ptr, struct tevent_thread_proxy);
 864         struct tevent_immediate *im;
 865         pthread_t *thread_id_ptr;
 866
 867         im = tevent_create_immediate(NULL);
 868         if (im == NULL) {
 869                 return NULL;
 870         }
 871         thread_id_ptr = talloc(NULL, pthread_t);
 872         if (thread_id_ptr == NULL) {
 873                 return NULL;
 874         }
 875         *thread_id_ptr = pthread_self();
 876
 877         tevent_thread_proxy_schedule(master_tp,
 878                                 &im,
 879                                 callback_nowait,
 880                                 &thread_id_ptr);
 881         return NULL;
 882 }
 883
 884 static void timeout_fn(struct tevent_context *ev,
 885                         struct tevent_timer *te,
 886                         struct timeval tv, void *p)
 887 {
 888         thread_counter = NUM_TEVENT_THREADS * 10;
 889 }
 890
 891 static bool test_multi_tevent_threaded(struct torture_context *test,
 892                                         const void *test_data)
 893 {
 894         unsigned i;
 895         struct tevent_context *master_ev;
 896         struct tevent_thread_proxy *tp;
 897
 898         talloc_disable_null_tracking();
 899
 900         /* Ugly global stuff. */
 901         thread_test_ctx = test;
 902         thread_counter = 0;
 903
 904         master_ev = tevent_context_init(NULL);
 905         if (master_ev == NULL) {
 906                 return false;
 907         }
 908         tevent_set_debug_stderr(master_ev);
 909
 910         tp = tevent_thread_proxy_create(master_ev);
 911         if (tp == NULL) {
 912                 torture_fail(test,
 913                         talloc_asprintf(test,
 914                                 "tevent_thread_proxy_create failed\n"));
 915                 talloc_free(master_ev);
 916                 return false;
 917         }
 918
 919         for (i = 0; i < NUM_TEVENT_THREADS; i++) {
 920                 int ret = pthread_create(&thread_map[i],
 921                                 NULL,
 922                                 thread_fn_nowait,
 923                                 tp);
 924                 if (ret != 0) {
 925                         torture_fail(test,
 926                                 talloc_asprintf(test,
 927                                         "Failed to create thread %i, %d\n",
 928                                         i, ret));
 929                         return false;
 930                 }
 931         }
 932
 933         /* Ensure we don't wait more than 10 seconds. */
 934         tevent_add_timer(master_ev,
 935                         master_ev,
 936                         timeval_current_ofs(10,0),
 937                         timeout_fn,
 938                         NULL);
 939
 940         while (thread_counter < NUM_TEVENT_THREADS) {
 941                 int ret = tevent_loop_once(master_ev);
 942                 torture_assert(test, ret == 0, "tevent_loop_once failed");
 943         }
 944
 945         torture_assert(test, thread_counter == NUM_TEVENT_THREADS,
 946                 "thread_counter fail\n");
 947
 948         talloc_free(master_ev);
 949         return true;
 950 }
 951
 952 struct reply_state {
 953         struct tevent_thread_proxy *reply_tp;
 954         pthread_t thread_id;
 955         int *p_finished;
 956 };
 957
 958 static void thread_timeout_fn(struct tevent_context *ev,
 959                         struct tevent_timer *te,
 960                         struct timeval tv, void *p)
 961 {
 962         int *p_finished = (int *)p;
 963
 964         *p_finished = 2;
 965 }
 966
 967 /* Called in child-thread context */
 968 static void thread_callback(struct tevent_context *ev,
 969                                 struct tevent_immediate *im,
 970                                 void *private_ptr)
 971 {
 972         struct reply_state *rsp =
 973                 talloc_get_type_abort(private_ptr, struct reply_state);
 974
 975         talloc_steal(ev, rsp);
 976         *rsp->p_finished = 1;
 977 }
 978
 979 /* Called in master thread context */
 980 static void master_callback(struct tevent_context *ev,
 981                                 struct tevent_immediate *im,
 982                                 void *private_ptr)
 983 {
 984         struct reply_state *rsp =
 985                 talloc_get_type_abort(private_ptr, struct reply_state);
 986         unsigned i;
 987
 988         talloc_steal(ev, rsp);
 989
 990         for (i = 0; i < NUM_TEVENT_THREADS; i++) {
 991                 if (pthread_equal(rsp->thread_id,
 992                                 thread_map[i])) {
 993                         break;
 994                 }
 995         }
 996         torture_comment(thread_test_ctx,
 997                         "Callback %u from thread %u\n",
 998                         thread_counter,
 999                         i);
1000         /* Now reply to the thread ! */
1001         tevent_thread_proxy_schedule(rsp->reply_tp,
1002                                 &im,
1003                                 thread_callback,
1004                                 &rsp);
1005
1006         thread_counter++;
1007 }
1008
1009 static void *thread_fn_1(void *private_ptr)
1010 {
1011         struct tevent_thread_proxy *master_tp =
1012                 talloc_get_type_abort(private_ptr, struct tevent_thread_proxy);
1013         struct tevent_thread_proxy *tp;
1014         struct tevent_immediate *im;
1015         struct tevent_context *ev;
1016         struct reply_state *rsp;
1017         int finished = 0;
1018         int ret;
1019
1020         ev = tevent_context_init(NULL);
1021         if (ev == NULL) {
1022                 return NULL;
1023         }
1024
1025         tp = tevent_thread_proxy_create(ev);
1026         if (tp == NULL) {
1027                 talloc_free(ev);
1028                 return NULL;
1029         }
1030
1031         im = tevent_create_immediate(ev);
1032         if (im == NULL) {
1033                 talloc_free(ev);
1034                 return NULL;
1035         }
1036
1037         rsp = talloc(ev, struct reply_state);
1038         if (rsp == NULL) {
1039                 talloc_free(ev);
1040                 return NULL;
1041         }
1042
1043         rsp->thread_id = pthread_self();
1044         rsp->reply_tp = tp;
1045         rsp->p_finished = &finished;
1046
1047         /* Introduce a little randomness into the mix.. */
1048         usleep(random() % 7000);
1049
1050         tevent_thread_proxy_schedule(master_tp,
1051                                 &im,
1052                                 master_callback,
1053                                 &rsp);
1054
1055         /* Ensure we don't wait more than 10 seconds. */
1056         tevent_add_timer(ev,
1057                         ev,
1058                         timeval_current_ofs(10,0),
1059                         thread_timeout_fn,
1060                         &finished);
1061
1062         while (finished == 0) {
1063                 ret = tevent_loop_once(ev);
1064                 assert(ret == 0);
1065         }
1066
1067         if (finished > 1) {
1068                 /* Timeout ! */
1069                 abort();
1070         }
1071
1072         /*
1073          * NB. We should talloc_free(ev) here, but if we do
1074          * we currently get hit by helgrind Fix #323432
1075          * "When calling pthread_cond_destroy or pthread_mutex_destroy
1076          * with initializers as argument Helgrind (incorrectly) reports errors."
1077          *
1078          * http://valgrind.10908.n7.nabble.com/Helgrind-3-9-0-false-positive-
1079          * with-pthread-mutex-destroy-td47757.html
1080          *
1081          * Helgrind doesn't understand that the request/reply
1082          * messages provide synchronization between the lock/unlock
1083          * in tevent_thread_proxy_schedule(), and the pthread_destroy()
1084          * when the struct tevent_thread_proxy object is talloc_free'd.
1085          *
1086          * As a work-around for now return ev for the parent thread to free.
1087          */
1088         return ev;
1089 }
1090
1091 static bool test_multi_tevent_threaded_1(struct torture_context *test,
1092                                         const void *test_data)
1093 {
1094         unsigned i;
1095         struct tevent_context *master_ev;
1096         struct tevent_thread_proxy *master_tp;
1097         int ret;
1098
1099         talloc_disable_null_tracking();
1100
1101         /* Ugly global stuff. */
1102         thread_test_ctx = test;
1103         thread_counter = 0;
1104
1105         master_ev = tevent_context_init(NULL);
1106         if (master_ev == NULL) {
1107                 return false;
1108         }
1109         tevent_set_debug_stderr(master_ev);
1110
1111         master_tp = tevent_thread_proxy_create(master_ev);
1112         if (master_tp == NULL) {
1113                 torture_fail(test,
1114                         talloc_asprintf(test,
1115                                 "tevent_thread_proxy_create failed\n"));
1116                 talloc_free(master_ev);
1117                 return false;
1118         }
1119
1120         for (i = 0; i < NUM_TEVENT_THREADS; i++) {
1121                 ret = pthread_create(&thread_map[i],
1122                                 NULL,
1123                                 thread_fn_1,
1124                                 master_tp);
1125                 if (ret != 0) {
1126                         torture_fail(test,
1127                                 talloc_asprintf(test,
1128                                         "Failed to create thread %i, %d\n",
1129                                         i, ret));
1130                                 return false;
1131                 }
1132         }
1133
1134         while (thread_counter < NUM_TEVENT_THREADS) {
1135                 ret = tevent_loop_once(master_ev);
1136                 torture_assert(test, ret == 0, "tevent_loop_once failed");
1137         }
1138
1139         /* Wait for all the threads to finish - join 'em. */
1140         for (i = 0; i < NUM_TEVENT_THREADS; i++) {
1141                 void *retval;
1142                 ret = pthread_join(thread_map[i], &retval);
1143                 torture_assert(test, ret == 0, "pthread_join failed");
1144                 /* Free the child thread event context. */
1145                 talloc_free(retval);
1146         }
1147
1148         talloc_free(master_ev);
1149         return true;
1150 }
1151
1152 struct threaded_test_2 {
1153         struct tevent_threaded_context *tctx;
1154         struct tevent_immediate *im;
1155         pthread_t thread_id;
1156 };
1157
1158 static void master_callback_2(struct tevent_context *ev,
1159                               struct tevent_immediate *im,
1160                               void *private_data);
1161
1162 static void *thread_fn_2(void *private_data)
1163 {
1164         struct threaded_test_2 *state = private_data;
1165
1166         state->thread_id = pthread_self();
1167
1168         usleep(random() % 7000);
1169
1170         tevent_threaded_schedule_immediate(
1171                 state->tctx, state->im, master_callback_2, state);
1172
1173         return NULL;
1174 }
1175
1176 static void master_callback_2(struct tevent_context *ev,
1177                               struct tevent_immediate *im,
1178                               void *private_data)
1179 {
1180         struct threaded_test_2 *state = private_data;
1181         int i;
1182
1183         for (i = 0; i < NUM_TEVENT_THREADS; i++) {
1184                 if (pthread_equal(state->thread_id, thread_map[i])) {
1185                         break;
1186                 }
1187         }
1188         torture_comment(thread_test_ctx,
1189                         "Callback_2 %u from thread %u\n",
1190                         thread_counter,
1191                         i);
1192         thread_counter++;
1193 }
1194
1195 static bool test_multi_tevent_threaded_2(struct torture_context *test,
1196                                          const void *test_data)
1197 {
1198         unsigned i;
1199
1200         struct tevent_context *ev;
1201         struct tevent_threaded_context *tctx;
1202         int ret;
1203
1204         thread_test_ctx = test;
1205         thread_counter = 0;
1206
1207         ev = tevent_context_init(test);
1208         torture_assert(test, ev != NULL, "tevent_context_init failed");
1209
1210         /*
1211          * tevent_re_initialise used to have a bug where it did not
1212          * re-initialise the thread support after taking it
1213          * down. Excercise that code path.
1214          */
1215         ret = tevent_re_initialise(ev);
1216         torture_assert(test, ret == 0, "tevent_re_initialise failed");
1217
1218         tctx = tevent_threaded_context_create(ev, ev);
1219         torture_assert(test, tctx != NULL,
1220                        "tevent_threaded_context_create failed");
1221
1222         for (i=0; i<NUM_TEVENT_THREADS; i++) {
1223                 struct threaded_test_2 *state;
1224
1225                 state = talloc(ev, struct threaded_test_2);
1226                 torture_assert(test, state != NULL, "talloc failed");
1227
1228                 state->tctx = tctx;
1229                 state->im = tevent_create_immediate(state);
1230                 torture_assert(test, state->im != NULL,
1231                                "tevent_create_immediate failed");
1232
1233                 ret = pthread_create(&thread_map[i], NULL, thread_fn_2, state);
1234                 torture_assert(test, ret == 0, "pthread_create failed");
1235         }
1236
1237         while (thread_counter < NUM_TEVENT_THREADS) {
1238                 ret = tevent_loop_once(ev);
1239                 torture_assert(test, ret == 0, "tevent_loop_once failed");
1240         }
1241
1242         /* Wait for all the threads to finish - join 'em. */
1243         for (i = 0; i < NUM_TEVENT_THREADS; i++) {
1244                 void *retval;
1245                 ret = pthread_join(thread_map[i], &retval);
1246                 torture_assert(test, ret == 0, "pthread_join failed");
1247                 /* Free the child thread event context. */
1248         }
1249
1250         talloc_free(tctx);
1251         talloc_free(ev);
1252         return true;
1253 }
1254 #endif
1255
1256 struct torture_suite *torture_local_event(TALLOC_CTX *mem_ctx)
1257 {
1258         struct torture_suite *suite = torture_suite_create(mem_ctx, "event");
1259         const char **list = tevent_backend_list(suite);
1260         int i;
1261
1262         for (i=0;list && list[i];i++) {
1263                 struct torture_suite *backend_suite;
1264
1265                 backend_suite = torture_suite_create(mem_ctx, list[i]);
1266
1267                 torture_suite_add_simple_tcase_const(backend_suite,
1268                                                "context",
1269                                                test_event_context,
1270                                                (const void *)list[i]);
1271                 torture_suite_add_simple_tcase_const(backend_suite,
1272                                                "fd1",
1273                                                test_event_fd1,
1274                                                (const void *)list[i]);
1275                 torture_suite_add_simple_tcase_const(backend_suite,
1276                                                "fd2",
1277                                                test_event_fd2,
1278                                                (const void *)list[i]);
1279
1280                 torture_suite_add_suite(suite, backend_suite);
1281         }
1282
1283 #ifdef HAVE_PTHREAD
1284         torture_suite_add_simple_tcase_const(suite, "threaded_poll_mt",
1285                                              test_event_context_threaded,
1286                                              NULL);
1287
1288         torture_suite_add_simple_tcase_const(suite, "multi_tevent_threaded",
1289                                              test_multi_tevent_threaded,
1290                                              NULL);
1291
1292         torture_suite_add_simple_tcase_const(suite, "multi_tevent_threaded_1",
1293                                              test_multi_tevent_threaded_1,
1294                                              NULL);
1295
1296         torture_suite_add_simple_tcase_const(suite, "multi_tevent_threaded_2",
1297                                              test_multi_tevent_threaded_2,
1298                                              NULL);
1299
1300 #endif
1301
1302         return suite;
1303 }