git.meshlink.io Git - catta/blob - src/simple-watch.c

   1 /***
   2   This file is part of catta.
   3
   4   catta is free software; you can redistribute it and/or modify it
   5   under the terms of the GNU Lesser General Public License as
   6   published by the Free Software Foundation; either version 2.1 of the
   7   License, or (at your option) any later version.
   8
   9   catta is distributed in the hope that it will be useful, but WITHOUT
  10   ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
  11   or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General
  12   Public License for more details.
  13
  14   You should have received a copy of the GNU Lesser General Public
  15   License along with catta; if not, write to the Free Software
  16   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  17   USA.
  18 ***/
  19
  20 #ifdef HAVE_CONFIG_H
  21 #include <config.h>
  22 #endif
  23
  24 #include <sys/poll.h>
  25 #include <assert.h>
  26 #include <string.h>
  27 #include <errno.h>
  28 #include <unistd.h>
  29 #include <fcntl.h>
  30 #include <stdio.h>
  31
  32 #include <catta/llist.h>
  33 #include <catta/malloc.h>
  34 #include <catta/timeval.h>
  35 #include <catta/simple-watch.h>
  36 #include "fdutil.h"                 // catta_set_nonblock
  37 #include "internal.h"               // closesocket
  38
  39 struct CattaWatch {
  40     CattaSimplePoll *simple_poll;
  41     int dead;
  42
  43     int idx;
  44     struct pollfd pollfd;
  45
  46     CattaWatchCallback callback;
  47     void *userdata;
  48
  49     CATTA_LLIST_FIELDS(CattaWatch, watches);
  50 };
  51
  52 struct CattaTimeout {
  53     CattaSimplePoll *simple_poll;
  54     int dead;
  55
  56     int enabled;
  57     struct timeval expiry;
  58
  59     CattaTimeoutCallback callback;
  60     void  *userdata;
  61
  62     CATTA_LLIST_FIELDS(CattaTimeout, timeouts);
  63 };
  64
  65 struct CattaSimplePoll {
  66     CattaPoll api;
  67     CattaPollFunc poll_func;
  68     void *poll_func_userdata;
  69
  70     struct pollfd* pollfds;
  71     int n_pollfds, max_pollfds, rebuild_pollfds;
  72
  73     int watch_req_cleanup, timeout_req_cleanup;
  74     int quit;
  75     int events_valid;
  76
  77     int n_watches;
  78     CATTA_LLIST_HEAD(CattaWatch, watches);
  79     CATTA_LLIST_HEAD(CattaTimeout, timeouts);
  80
  81     int wakeup_pipe[2];
  82     int wakeup_issued;
  83
  84     int prepared_timeout;
  85
  86     enum {
  87         STATE_INIT,
  88         STATE_PREPARING,
  89         STATE_PREPARED,
  90         STATE_RUNNING,
  91         STATE_RAN,
  92         STATE_DISPATCHING,
  93         STATE_DISPATCHED,
  94         STATE_QUIT,
  95         STATE_FAILURE
  96     } state;
  97 };
  98
  99 void catta_simple_poll_wakeup(CattaSimplePoll *s) {
 100     char c = 'W';
 101     assert(s);
 102
 103     writepipe(s->wakeup_pipe[1], &c, sizeof(c));
 104     s->wakeup_issued = 1;
 105 }
 106
 107 static void clear_wakeup(CattaSimplePoll *s) {
 108     char c[10]; /* Read ten at a time */
 109
 110     if (!s->wakeup_issued)
 111         return;
 112
 113     s->wakeup_issued = 0;
 114
 115     for(;;)
 116         if (readpipe(s->wakeup_pipe[0], c, sizeof(c)) != sizeof(c))
 117             break;
 118 }
 119
 120 static CattaWatch* watch_new(const CattaPoll *api, int fd, CattaWatchEvent event, CattaWatchCallback callback, void *userdata) {
 121     CattaWatch *w;
 122     CattaSimplePoll *s;
 123
 124     assert(api);
 125     assert(fd >= 0);
 126     assert(callback);
 127
 128     s = api->userdata;
 129     assert(s);
 130
 131     if (!(w = catta_new(CattaWatch, 1)))
 132         return NULL;
 133
 134     /* If there is a background thread running the poll() for us, tell it to exit the poll() */
 135     catta_simple_poll_wakeup(s);
 136
 137     w->simple_poll = s;
 138     w->dead = 0;
 139
 140     w->pollfd.fd = fd;
 141     w->pollfd.events = event;
 142     w->pollfd.revents = 0;
 143
 144     w->callback = callback;
 145     w->userdata = userdata;
 146
 147     w->idx = -1;
 148     s->rebuild_pollfds = 1;
 149
 150     CATTA_LLIST_PREPEND(CattaWatch, watches, s->watches, w);
 151     s->n_watches++;
 152
 153     return w;
 154 }
 155
 156 static void watch_update(CattaWatch *w, CattaWatchEvent events) {
 157     assert(w);
 158     assert(!w->dead);
 159
 160     /* If there is a background thread running the poll() for us, tell it to exit the poll() */
 161     catta_simple_poll_wakeup(w->simple_poll);
 162
 163     w->pollfd.events = events;
 164
 165     if (w->idx != -1) {
 166         assert(w->simple_poll);
 167         w->simple_poll->pollfds[w->idx] = w->pollfd;
 168     } else
 169         w->simple_poll->rebuild_pollfds = 1;
 170 }
 171
 172 static CattaWatchEvent watch_get_events(CattaWatch *w) {
 173     assert(w);
 174     assert(!w->dead);
 175
 176     if (w->idx != -1 && w->simple_poll->events_valid)
 177         return w->simple_poll->pollfds[w->idx].revents;
 178
 179     return 0;
 180 }
 181
 182 static void remove_pollfd(CattaWatch *w) {
 183     assert(w);
 184
 185     if (w->idx == -1)
 186         return;
 187
 188     w->simple_poll->rebuild_pollfds = 1;
 189 }
 190
 191 static void watch_free(CattaWatch *w) {
 192     assert(w);
 193
 194     assert(!w->dead);
 195
 196     /* If there is a background thread running the poll() for us, tell it to exit the poll() */
 197     catta_simple_poll_wakeup(w->simple_poll);
 198
 199     remove_pollfd(w);
 200
 201     w->dead = 1;
 202     w->simple_poll->n_watches --;
 203     w->simple_poll->watch_req_cleanup = 1;
 204 }
 205
 206 static void destroy_watch(CattaWatch *w) {
 207     assert(w);
 208
 209     remove_pollfd(w);
 210     CATTA_LLIST_REMOVE(CattaWatch, watches, w->simple_poll->watches, w);
 211
 212     if (!w->dead)
 213         w->simple_poll->n_watches --;
 214
 215     catta_free(w);
 216 }
 217
 218 static void cleanup_watches(CattaSimplePoll *s, int all) {
 219     CattaWatch *w, *next;
 220     assert(s);
 221
 222     for (w = s->watches; w; w = next) {
 223         next = w->watches_next;
 224
 225         if (all || w->dead)
 226             destroy_watch(w);
 227     }
 228
 229     s->timeout_req_cleanup = 0;
 230 }
 231
 232 static CattaTimeout* timeout_new(const CattaPoll *api, const struct timeval *tv, CattaTimeoutCallback callback, void *userdata) {
 233     CattaTimeout *t;
 234     CattaSimplePoll *s;
 235
 236     assert(api);
 237     assert(callback);
 238
 239     s = api->userdata;
 240     assert(s);
 241
 242     if (!(t = catta_new(CattaTimeout, 1)))
 243         return NULL;
 244
 245     /* If there is a background thread running the poll() for us, tell it to exit the poll() */
 246     catta_simple_poll_wakeup(s);
 247
 248     t->simple_poll = s;
 249     t->dead = 0;
 250
 251     if ((t->enabled = !!tv))
 252         t->expiry = *tv;
 253
 254     t->callback = callback;
 255     t->userdata = userdata;
 256
 257     CATTA_LLIST_PREPEND(CattaTimeout, timeouts, s->timeouts, t);
 258     return t;
 259 }
 260
 261 static void timeout_update(CattaTimeout *t, const struct timeval *tv) {
 262     assert(t);
 263     assert(!t->dead);
 264
 265     /* If there is a background thread running the poll() for us, tell it to exit the poll() */
 266     catta_simple_poll_wakeup(t->simple_poll);
 267
 268     if ((t->enabled = !!tv))
 269         t->expiry = *tv;
 270 }
 271
 272 static void timeout_free(CattaTimeout *t) {
 273     assert(t);
 274     assert(!t->dead);
 275
 276     /* If there is a background thread running the poll() for us, tell it to exit the poll() */
 277     catta_simple_poll_wakeup(t->simple_poll);
 278
 279     t->dead = 1;
 280     t->simple_poll->timeout_req_cleanup = 1;
 281 }
 282
 283
 284 static void destroy_timeout(CattaTimeout *t) {
 285     assert(t);
 286
 287     CATTA_LLIST_REMOVE(CattaTimeout, timeouts, t->simple_poll->timeouts, t);
 288
 289     catta_free(t);
 290 }
 291
 292 static void cleanup_timeouts(CattaSimplePoll *s, int all) {
 293     CattaTimeout *t, *next;
 294     assert(s);
 295
 296     for (t = s->timeouts; t; t = next) {
 297         next = t->timeouts_next;
 298
 299         if (all || t->dead)
 300             destroy_timeout(t);
 301     }
 302
 303     s->timeout_req_cleanup = 0;
 304 }
 305
 306 CattaSimplePoll *catta_simple_poll_new(void) {
 307     CattaSimplePoll *s;
 308
 309     if (!(s = catta_new(CattaSimplePoll, 1)))
 310         return NULL;
 311
 312     winsock_init();  // on Windows, pipe uses sockets; no-op on other platforms
 313     if (pipe(s->wakeup_pipe) < 0) {
 314         catta_free(s);
 315         winsock_exit();
 316         return NULL;
 317     }
 318
 319     catta_set_nonblock(s->wakeup_pipe[0]);
 320     catta_set_nonblock(s->wakeup_pipe[1]);
 321
 322     s->api.userdata = s;
 323
 324     s->api.watch_new = watch_new;
 325     s->api.watch_free = watch_free;
 326     s->api.watch_update = watch_update;
 327     s->api.watch_get_events = watch_get_events;
 328
 329     s->api.timeout_new = timeout_new;
 330     s->api.timeout_free = timeout_free;
 331     s->api.timeout_update = timeout_update;
 332
 333     s->pollfds = NULL;
 334     s->max_pollfds = s->n_pollfds = 0;
 335     s->rebuild_pollfds = 1;
 336     s->quit = 0;
 337     s->n_watches = 0;
 338     s->events_valid = 0;
 339
 340     s->watch_req_cleanup = 0;
 341     s->timeout_req_cleanup = 0;
 342
 343     s->prepared_timeout = 0;
 344
 345     s->state = STATE_INIT;
 346
 347     s->wakeup_issued = 0;
 348
 349     catta_simple_poll_set_func(s, NULL, NULL);
 350
 351     CATTA_LLIST_HEAD_INIT(CattaWatch, s->watches);
 352     CATTA_LLIST_HEAD_INIT(CattaTimeout, s->timeouts);
 353
 354     return s;
 355 }
 356
 357 void catta_simple_poll_free(CattaSimplePoll *s) {
 358     assert(s);
 359
 360     cleanup_timeouts(s, 1);
 361     cleanup_watches(s, 1);
 362     assert(s->n_watches == 0);
 363
 364     catta_free(s->pollfds);
 365
 366     if (s->wakeup_pipe[0] >= 0)
 367         closepipe(s->wakeup_pipe[0]);
 368
 369     if (s->wakeup_pipe[1] >= 0)
 370         closepipe(s->wakeup_pipe[1]);
 371
 372     catta_free(s);
 373     winsock_exit();  // match the winsock_init in catta_simple_poll_new
 374 }
 375
 376 static int rebuild(CattaSimplePoll *s) {
 377     CattaWatch *w;
 378     int idx;
 379
 380     assert(s);
 381
 382     if (s->n_watches+1 > s->max_pollfds) {
 383         struct pollfd *n;
 384
 385         s->max_pollfds = s->n_watches + 10;
 386
 387         if (!(n = catta_realloc(s->pollfds, sizeof(struct pollfd) * s->max_pollfds)))
 388             return -1;
 389
 390         s->pollfds = n;
 391     }
 392
 393
 394     s->pollfds[0].fd = s->wakeup_pipe[0];
 395     s->pollfds[0].events = POLLIN;
 396     s->pollfds[0].revents = 0;
 397
 398     idx = 1;
 399
 400     for (w = s->watches; w; w = w->watches_next) {
 401
 402         if(w->dead)
 403             continue;
 404
 405         assert(w->idx < s->max_pollfds);
 406         s->pollfds[w->idx = idx++] = w->pollfd;
 407     }
 408
 409     s->n_pollfds = idx;
 410     s->events_valid = 0;
 411     s->rebuild_pollfds = 0;
 412
 413     return 0;
 414 }
 415
 416 static CattaTimeout* find_next_timeout(CattaSimplePoll *s) {
 417     CattaTimeout *t, *n = NULL;
 418     assert(s);
 419
 420     for (t = s->timeouts; t; t = t->timeouts_next) {
 421
 422         if (t->dead || !t->enabled)
 423             continue;
 424
 425         if (!n || catta_timeval_compare(&t->expiry, &n->expiry) < 0)
 426             n = t;
 427     }
 428
 429     return n;
 430 }
 431
 432 static void timeout_callback(CattaTimeout *t) {
 433     assert(t);
 434     assert(!t->dead);
 435     assert(t->enabled);
 436
 437     t->enabled = 0;
 438     t->callback(t, t->userdata);
 439 }
 440
 441 int catta_simple_poll_prepare(CattaSimplePoll *s, int timeout) {
 442     CattaTimeout *next_timeout;
 443
 444     assert(s);
 445     assert(s->state == STATE_INIT || s->state == STATE_DISPATCHED || s->state == STATE_FAILURE);
 446     s->state = STATE_PREPARING;
 447
 448     /* Clear pending wakeup requests */
 449     clear_wakeup(s);
 450
 451     /* Cleanup things first */
 452     if (s->watch_req_cleanup)
 453         cleanup_watches(s, 0);
 454
 455     if (s->timeout_req_cleanup)
 456         cleanup_timeouts(s, 0);
 457
 458     /* Check whether a quit was requested */
 459     if (s->quit) {
 460         s->state = STATE_QUIT;
 461         return 1;
 462     }
 463
 464     /* Do we need to rebuild our array of pollfds? */
 465     if (s->rebuild_pollfds)
 466         if (rebuild(s) < 0) {
 467             s->state = STATE_FAILURE;
 468             return -1;
 469         }
 470
 471     /* Calculate the wakeup time */
 472     if ((next_timeout = find_next_timeout(s))) {
 473         struct timeval now;
 474         int t;
 475         CattaUsec usec;
 476
 477         if (next_timeout->expiry.tv_sec == 0 &&
 478             next_timeout->expiry.tv_usec == 0) {
 479
 480             /* Just a shortcut so that we don't need to call gettimeofday() */
 481             timeout = 0;
 482             goto finish;
 483         }
 484
 485         gettimeofday(&now, NULL);
 486         usec = catta_timeval_diff(&next_timeout->expiry, &now);
 487
 488         if (usec <= 0) {
 489             /* Timeout elapsed */
 490
 491             timeout = 0;
 492             goto finish;
 493         }
 494
 495         /* Calculate sleep time. We add 1ms because otherwise we'd
 496          * wake up too early most of the time */
 497         t = (int) (usec / 1000) + 1;
 498
 499         if (timeout < 0 || timeout > t)
 500             timeout = t;
 501     }
 502
 503 finish:
 504     s->prepared_timeout = timeout;
 505     s->state = STATE_PREPARED;
 506     return 0;
 507 }
 508
 509 int catta_simple_poll_run(CattaSimplePoll *s) {
 510     assert(s);
 511     assert(s->state == STATE_PREPARED || s->state == STATE_FAILURE);
 512
 513     s->state = STATE_RUNNING;
 514
 515     for (;;) {
 516         errno = 0;
 517
 518         if (s->poll_func(s->pollfds, s->n_pollfds, s->prepared_timeout, s->poll_func_userdata) < 0) {
 519
 520             if (errno == EINTR)
 521                 continue;
 522
 523             s->state = STATE_FAILURE;
 524             return -1;
 525         }
 526
 527         break;
 528     }
 529
 530     /* The poll events are now valid again */
 531     s->events_valid = 1;
 532
 533     /* Update state */
 534     s->state = STATE_RAN;
 535     return 0;
 536 }
 537
 538 int catta_simple_poll_dispatch(CattaSimplePoll *s) {
 539     CattaTimeout *next_timeout;
 540     CattaWatch *w;
 541
 542     assert(s);
 543     assert(s->state == STATE_RAN);
 544     s->state = STATE_DISPATCHING;
 545
 546     /* We execute only on callback in every iteration */
 547
 548     /* Check whether the wakeup time has been reached now */
 549     if ((next_timeout = find_next_timeout(s))) {
 550
 551         if (next_timeout->expiry.tv_sec == 0 && next_timeout->expiry.tv_usec == 0) {
 552
 553             /* Just a shortcut so that we don't need to call gettimeofday() */
 554             timeout_callback(next_timeout);
 555             goto finish;
 556         }
 557
 558         if (catta_age(&next_timeout->expiry) >= 0) {
 559
 560             /* Timeout elapsed */
 561             timeout_callback(next_timeout);
 562             goto finish;
 563         }
 564     }
 565
 566     /* Look for some kind of I/O event */
 567     for (w = s->watches; w; w = w->watches_next) {
 568
 569         if (w->dead)
 570             continue;
 571
 572         assert(w->idx >= 0);
 573         assert(w->idx < s->n_pollfds);
 574
 575         if (s->pollfds[w->idx].revents != 0) {
 576             w->callback(w, w->pollfd.fd, s->pollfds[w->idx].revents, w->userdata);
 577             goto finish;
 578         }
 579     }
 580
 581 finish:
 582
 583     s->state = STATE_DISPATCHED;
 584     return 0;
 585 }
 586
 587 int catta_simple_poll_iterate(CattaSimplePoll *s, int timeout) {
 588     int r;
 589
 590     if ((r = catta_simple_poll_prepare(s, timeout)) != 0)
 591         return r;
 592
 593     if ((r = catta_simple_poll_run(s)) != 0)
 594         return r;
 595
 596     if ((r = catta_simple_poll_dispatch(s)) != 0)
 597         return r;
 598
 599     return 0;
 600 }
 601
 602 void catta_simple_poll_quit(CattaSimplePoll *s) {
 603     assert(s);
 604
 605     s->quit = 1;
 606
 607     /* If there is a background thread running the poll() for us, tell it to exit the poll() */
 608     catta_simple_poll_wakeup(s);
 609 }
 610
 611 const CattaPoll* catta_simple_poll_get(CattaSimplePoll *s) {
 612     assert(s);
 613
 614     return &s->api;
 615 }
 616
 617 static int system_poll(struct pollfd *ufds, unsigned int nfds, int timeout, CATTA_GCC_UNUSED void *userdata) {
 618     return poll(ufds, nfds, timeout);
 619 }
 620
 621 void catta_simple_poll_set_func(CattaSimplePoll *s, CattaPollFunc func, void *userdata) {
 622     assert(s);
 623
 624     s->poll_func = func ? func : system_poll;
 625     s->poll_func_userdata = func ? userdata : NULL;
 626
 627     /* If there is a background thread running the poll() for us, tell it to exit the poll() */
 628     catta_simple_poll_wakeup(s);
 629 }
 630
 631 int catta_simple_poll_loop(CattaSimplePoll *s) {
 632     int r;
 633
 634     assert(s);
 635
 636     for (;;)
 637         if ((r = catta_simple_poll_iterate(s, -1)) != 0)
 638             if (r >= 0 || errno != EINTR)
 639                 return r;
 640 }