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     write(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 (read(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     if (pipe(s->wakeup_pipe) < 0) {
 313         catta_free(s);
 314         return NULL;
 315     }
 316
 317     catta_set_nonblock(s->wakeup_pipe[0]);
 318     catta_set_nonblock(s->wakeup_pipe[1]);
 319
 320     s->api.userdata = s;
 321
 322     s->api.watch_new = watch_new;
 323     s->api.watch_free = watch_free;
 324     s->api.watch_update = watch_update;
 325     s->api.watch_get_events = watch_get_events;
 326
 327     s->api.timeout_new = timeout_new;
 328     s->api.timeout_free = timeout_free;
 329     s->api.timeout_update = timeout_update;
 330
 331     s->pollfds = NULL;
 332     s->max_pollfds = s->n_pollfds = 0;
 333     s->rebuild_pollfds = 1;
 334     s->quit = 0;
 335     s->n_watches = 0;
 336     s->events_valid = 0;
 337
 338     s->watch_req_cleanup = 0;
 339     s->timeout_req_cleanup = 0;
 340
 341     s->prepared_timeout = 0;
 342
 343     s->state = STATE_INIT;
 344
 345     s->wakeup_issued = 0;
 346
 347     catta_simple_poll_set_func(s, NULL, NULL);
 348
 349     CATTA_LLIST_HEAD_INIT(CattaWatch, s->watches);
 350     CATTA_LLIST_HEAD_INIT(CattaTimeout, s->timeouts);
 351
 352     return s;
 353 }
 354
 355 void catta_simple_poll_free(CattaSimplePoll *s) {
 356     assert(s);
 357
 358     cleanup_timeouts(s, 1);
 359     cleanup_watches(s, 1);
 360     assert(s->n_watches == 0);
 361
 362     catta_free(s->pollfds);
 363
 364     if (s->wakeup_pipe[0] >= 0)
 365         closesocket(s->wakeup_pipe[0]);
 366
 367     if (s->wakeup_pipe[1] >= 0)
 368         closesocket(s->wakeup_pipe[1]);
 369
 370     catta_free(s);
 371 }
 372
 373 static int rebuild(CattaSimplePoll *s) {
 374     CattaWatch *w;
 375     int idx;
 376
 377     assert(s);
 378
 379     if (s->n_watches+1 > s->max_pollfds) {
 380         struct pollfd *n;
 381
 382         s->max_pollfds = s->n_watches + 10;
 383
 384         if (!(n = catta_realloc(s->pollfds, sizeof(struct pollfd) * s->max_pollfds)))
 385             return -1;
 386
 387         s->pollfds = n;
 388     }
 389
 390
 391     s->pollfds[0].fd = s->wakeup_pipe[0];
 392     s->pollfds[0].events = POLLIN;
 393     s->pollfds[0].revents = 0;
 394
 395     idx = 1;
 396
 397     for (w = s->watches; w; w = w->watches_next) {
 398
 399         if(w->dead)
 400             continue;
 401
 402         assert(w->idx < s->max_pollfds);
 403         s->pollfds[w->idx = idx++] = w->pollfd;
 404     }
 405
 406     s->n_pollfds = idx;
 407     s->events_valid = 0;
 408     s->rebuild_pollfds = 0;
 409
 410     return 0;
 411 }
 412
 413 static CattaTimeout* find_next_timeout(CattaSimplePoll *s) {
 414     CattaTimeout *t, *n = NULL;
 415     assert(s);
 416
 417     for (t = s->timeouts; t; t = t->timeouts_next) {
 418
 419         if (t->dead || !t->enabled)
 420             continue;
 421
 422         if (!n || catta_timeval_compare(&t->expiry, &n->expiry) < 0)
 423             n = t;
 424     }
 425
 426     return n;
 427 }
 428
 429 static void timeout_callback(CattaTimeout *t) {
 430     assert(t);
 431     assert(!t->dead);
 432     assert(t->enabled);
 433
 434     t->enabled = 0;
 435     t->callback(t, t->userdata);
 436 }
 437
 438 int catta_simple_poll_prepare(CattaSimplePoll *s, int timeout) {
 439     CattaTimeout *next_timeout;
 440
 441     assert(s);
 442     assert(s->state == STATE_INIT || s->state == STATE_DISPATCHED || s->state == STATE_FAILURE);
 443     s->state = STATE_PREPARING;
 444
 445     /* Clear pending wakeup requests */
 446     clear_wakeup(s);
 447
 448     /* Cleanup things first */
 449     if (s->watch_req_cleanup)
 450         cleanup_watches(s, 0);
 451
 452     if (s->timeout_req_cleanup)
 453         cleanup_timeouts(s, 0);
 454
 455     /* Check whether a quit was requested */
 456     if (s->quit) {
 457         s->state = STATE_QUIT;
 458         return 1;
 459     }
 460
 461     /* Do we need to rebuild our array of pollfds? */
 462     if (s->rebuild_pollfds)
 463         if (rebuild(s) < 0) {
 464             s->state = STATE_FAILURE;
 465             return -1;
 466         }
 467
 468     /* Calculate the wakeup time */
 469     if ((next_timeout = find_next_timeout(s))) {
 470         struct timeval now;
 471         int t;
 472         CattaUsec usec;
 473
 474         if (next_timeout->expiry.tv_sec == 0 &&
 475             next_timeout->expiry.tv_usec == 0) {
 476
 477             /* Just a shortcut so that we don't need to call gettimeofday() */
 478             timeout = 0;
 479             goto finish;
 480         }
 481
 482         gettimeofday(&now, NULL);
 483         usec = catta_timeval_diff(&next_timeout->expiry, &now);
 484
 485         if (usec <= 0) {
 486             /* Timeout elapsed */
 487
 488             timeout = 0;
 489             goto finish;
 490         }
 491
 492         /* Calculate sleep time. We add 1ms because otherwise we'd
 493          * wake up too early most of the time */
 494         t = (int) (usec / 1000) + 1;
 495
 496         if (timeout < 0 || timeout > t)
 497             timeout = t;
 498     }
 499
 500 finish:
 501     s->prepared_timeout = timeout;
 502     s->state = STATE_PREPARED;
 503     return 0;
 504 }
 505
 506 int catta_simple_poll_run(CattaSimplePoll *s) {
 507     assert(s);
 508     assert(s->state == STATE_PREPARED || s->state == STATE_FAILURE);
 509
 510     s->state = STATE_RUNNING;
 511
 512     for (;;) {
 513         errno = 0;
 514
 515         if (s->poll_func(s->pollfds, s->n_pollfds, s->prepared_timeout, s->poll_func_userdata) < 0) {
 516
 517             if (errno == EINTR)
 518                 continue;
 519
 520             s->state = STATE_FAILURE;
 521             return -1;
 522         }
 523
 524         break;
 525     }
 526
 527     /* The poll events are now valid again */
 528     s->events_valid = 1;
 529
 530     /* Update state */
 531     s->state = STATE_RAN;
 532     return 0;
 533 }
 534
 535 int catta_simple_poll_dispatch(CattaSimplePoll *s) {
 536     CattaTimeout *next_timeout;
 537     CattaWatch *w;
 538
 539     assert(s);
 540     assert(s->state == STATE_RAN);
 541     s->state = STATE_DISPATCHING;
 542
 543     /* We execute only on callback in every iteration */
 544
 545     /* Check whether the wakeup time has been reached now */
 546     if ((next_timeout = find_next_timeout(s))) {
 547
 548         if (next_timeout->expiry.tv_sec == 0 && next_timeout->expiry.tv_usec == 0) {
 549
 550             /* Just a shortcut so that we don't need to call gettimeofday() */
 551             timeout_callback(next_timeout);
 552             goto finish;
 553         }
 554
 555         if (catta_age(&next_timeout->expiry) >= 0) {
 556
 557             /* Timeout elapsed */
 558             timeout_callback(next_timeout);
 559             goto finish;
 560         }
 561     }
 562
 563     /* Look for some kind of I/O event */
 564     for (w = s->watches; w; w = w->watches_next) {
 565
 566         if (w->dead)
 567             continue;
 568
 569         assert(w->idx >= 0);
 570         assert(w->idx < s->n_pollfds);
 571
 572         if (s->pollfds[w->idx].revents != 0) {
 573             w->callback(w, w->pollfd.fd, s->pollfds[w->idx].revents, w->userdata);
 574             goto finish;
 575         }
 576     }
 577
 578 finish:
 579
 580     s->state = STATE_DISPATCHED;
 581     return 0;
 582 }
 583
 584 int catta_simple_poll_iterate(CattaSimplePoll *s, int timeout) {
 585     int r;
 586
 587     if ((r = catta_simple_poll_prepare(s, timeout)) != 0)
 588         return r;
 589
 590     if ((r = catta_simple_poll_run(s)) != 0)
 591         return r;
 592
 593     if ((r = catta_simple_poll_dispatch(s)) != 0)
 594         return r;
 595
 596     return 0;
 597 }
 598
 599 void catta_simple_poll_quit(CattaSimplePoll *s) {
 600     assert(s);
 601
 602     s->quit = 1;
 603
 604     /* If there is a background thread running the poll() for us, tell it to exit the poll() */
 605     catta_simple_poll_wakeup(s);
 606 }
 607
 608 const CattaPoll* catta_simple_poll_get(CattaSimplePoll *s) {
 609     assert(s);
 610
 611     return &s->api;
 612 }
 613
 614 static int system_poll(struct pollfd *ufds, unsigned int nfds, int timeout, CATTA_GCC_UNUSED void *userdata) {
 615     return poll(ufds, nfds, timeout);
 616 }
 617
 618 void catta_simple_poll_set_func(CattaSimplePoll *s, CattaPollFunc func, void *userdata) {
 619     assert(s);
 620
 621     s->poll_func = func ? func : system_poll;
 622     s->poll_func_userdata = func ? userdata : NULL;
 623
 624     /* If there is a background thread running the poll() for us, tell it to exit the poll() */
 625     catta_simple_poll_wakeup(s);
 626 }
 627
 628 int catta_simple_poll_loop(CattaSimplePoll *s) {
 629     int r;
 630
 631     assert(s);
 632
 633     for (;;)
 634         if ((r = catta_simple_poll_iterate(s, -1)) != 0)
 635             if (r >= 0 || errno != EINTR)
 636                 return r;
 637 }