ucw/mainloop.c

   1 /*
   2  *      UCW Library -- Main Loop
   3  *
   4  *      (c) 2004--2006 Martin Mares <mj@ucw.cz>
   5  *
   6  *      This software may be freely distributed and used according to the terms
   7  *      of the GNU Lesser General Public License.
   8  */
   9
  10 #undef LOCAL_DEBUG
  11
  12 #include "ucw/lib.h"
  13 #include "ucw/heap.h"
  14 #include "ucw/mainloop.h"
  15
  16 #include <stdio.h>
  17 #include <string.h>
  18 #include <unistd.h>
  19 #include <signal.h>
  20 #include <fcntl.h>
  21 #include <errno.h>
  22 #include <time.h>
  23 #include <sys/poll.h>
  24 #include <sys/wait.h>
  25 #include <sys/time.h>
  26
  27 timestamp_t main_now;
  28 ucw_time_t main_now_seconds;
  29 timestamp_t main_idle_time;
  30 uns main_shutdown;
  31
  32 #define GBUF_PREFIX(x) main_timer_table_##x
  33 #define GBUF_TYPE struct main_timer *
  34 #include "ucw/gbuf.h"
  35 static uns main_timer_cnt;
  36 static main_timer_table_t main_timer_table;
  37 #define MAIN_TIMER_LESS(x,y) ((x)->expires < (y)->expires)
  38 #define MAIN_TIMER_SWAP(heap,a,b,t) (t=heap[a], heap[a]=heap[b], heap[b]=t, heap[a]->index=(a), heap[b]->index=(b))
  39
  40 clist main_file_list, main_hook_list, main_hook_done_list, main_process_list;
  41 static uns main_file_cnt;
  42 static uns main_poll_table_obsolete, main_poll_table_size;
  43 static struct pollfd *main_poll_table;
  44 static uns main_sigchld_set_up;
  45 static volatile sig_atomic_t chld_received = 0;
  46
  47 #ifdef O_CLOEXEC
  48 // On recent Linux systems, O_CLOEXEC flag is available and we can get around
  49 // the race condition of poll().
  50 #define USE_SELF_PIPE
  51 static int sig_pipe_recv, sig_pipe_send;
  52 #endif
  53
  54 void
  55 main_get_time(void)
  56 {
  57   struct timeval tv;
  58   gettimeofday(&tv, NULL);
  59   main_now_seconds = tv.tv_sec;
  60   main_now = (timestamp_t)tv.tv_sec * 1000 + tv.tv_usec / 1000;
  61   // DBG("It's %lld o'clock", (long long) main_now);
  62 }
  63
  64 void
  65 main_init(void)
  66 {
  67   DBG("MAIN: Initializing");
  68   main_timer_cnt = 0;
  69   clist_init(&main_file_list);
  70   clist_init(&main_hook_list);
  71   clist_init(&main_hook_done_list);
  72   clist_init(&main_process_list);
  73   main_file_cnt = 0;
  74   main_poll_table_obsolete = 1;
  75   main_get_time();
  76 }
  77
  78 void
  79 timer_add(struct main_timer *tm, timestamp_t expires)
  80 {
  81   if (expires)
  82     DBG("MAIN: Setting timer %p (expire at now+%lld)", tm, (long long)(expires-main_now));
  83   else
  84     DBG("MAIN: Clearing timer %p", tm);
  85   if (tm->expires < expires)
  86     {
  87       if (!tm->expires)
  88         {
  89           tm->expires = expires;
  90           tm->index = ++main_timer_cnt;
  91           main_timer_table_grow(&main_timer_table, tm->index + 1);
  92           main_timer_table.ptr[tm->index] = tm;
  93           HEAP_INSERT(struct main_timer *, main_timer_table.ptr, main_timer_cnt, MAIN_TIMER_LESS, MAIN_TIMER_SWAP);
  94         }
  95       else
  96         {
  97           tm->expires = expires;
  98           HEAP_INCREASE(struct main_timer *, main_timer_table.ptr, main_timer_cnt, MAIN_TIMER_LESS, MAIN_TIMER_SWAP, tm->index);
  99         }
 100     }
 101   else if (tm->expires > expires)
 102     {
 103       if (!expires)
 104         {
 105           ASSERT(tm->index && tm->index <= main_timer_cnt);
 106           HEAP_DELETE(struct main_timer *, main_timer_table.ptr, main_timer_cnt, MAIN_TIMER_LESS, MAIN_TIMER_SWAP, tm->index);
 107           tm->index = 0;
 108           tm->expires = 0;
 109         }
 110       else
 111         {
 112           tm->expires = expires;
 113           HEAP_DECREASE(struct main_timer *, main_timer_table.ptr, main_timer_cnt, MAIN_TIMER_LESS, MAIN_TIMER_SWAP, tm->index);
 114         }
 115     }
 116 }
 117
 118 void
 119 timer_del(struct main_timer *tm)
 120 {
 121   timer_add(tm, 0);
 122 }
 123
 124 static void
 125 file_timer_expired(struct main_timer *tm)
 126 {
 127   struct main_file *fi = tm->data;
 128   timer_del(&fi->timer);
 129   if (fi->error_handler)
 130     fi->error_handler(fi, MFERR_TIMEOUT);
 131 }
 132
 133 void
 134 file_add(struct main_file *fi)
 135 {
 136   DBG("MAIN: Adding file %p (fd=%d)", fi, fi->fd);
 137   ASSERT(!fi->n.next);
 138   clist_add_tail(&main_file_list, &fi->n);
 139   fi->timer.handler = file_timer_expired;
 140   fi->timer.data = fi;
 141   main_file_cnt++;
 142   main_poll_table_obsolete = 1;
 143   if (fcntl(fi->fd, F_SETFL, O_NONBLOCK) < 0)
 144     msg(L_ERROR, "Error setting fd %d to non-blocking mode: %m. Keep fingers crossed.", fi->fd);
 145 }
 146
 147 void
 148 file_chg(struct main_file *fi)
 149 {
 150   struct pollfd *p = fi->pollfd;
 151   if (p)
 152     {
 153       p->events = 0;
 154       if (fi->read_handler)
 155         p->events |= POLLIN | POLLHUP | POLLERR;
 156       if (fi->write_handler)
 157         p->events |= POLLOUT | POLLERR;
 158     }
 159 }
 160
 161 void
 162 file_del(struct main_file *fi)
 163 {
 164   DBG("MAIN: Deleting file %p (fd=%d)", fi, fi->fd);
 165   ASSERT(fi->n.next);
 166   timer_del(&fi->timer);
 167   clist_remove(&fi->n);
 168   main_file_cnt--;
 169   main_poll_table_obsolete = 1;
 170   fi->n.next = fi->n.prev = NULL;
 171 }
 172
 173 static int
 174 file_read_handler(struct main_file *fi)
 175 {
 176   while (fi->rpos < fi->rlen)
 177     {
 178       int l = read(fi->fd, fi->rbuf + fi->rpos, fi->rlen - fi->rpos);
 179       DBG("MAIN: FD %d: read %d", fi->fd, l);
 180       if (l < 0)
 181         {
 182           if (errno != EINTR && errno != EAGAIN && fi->error_handler)
 183             fi->error_handler(fi, MFERR_READ);
 184           return 0;
 185         }
 186       else if (!l)
 187         break;
 188       fi->rpos += l;
 189     }
 190   DBG("MAIN: FD %d done read %d of %d", fi->fd, fi->rpos, fi->rlen);
 191   fi->read_handler = NULL;
 192   file_chg(fi);
 193   fi->read_done(fi);
 194   return 1;
 195 }
 196
 197 static int
 198 file_write_handler(struct main_file *fi)
 199 {
 200   while (fi->wpos < fi->wlen)
 201     {
 202       int l = write(fi->fd, fi->wbuf + fi->wpos, fi->wlen - fi->wpos);
 203       DBG("MAIN: FD %d: write %d", fi->fd, l);
 204       if (l < 0)
 205         {
 206           if (errno != EINTR && errno != EAGAIN && fi->error_handler)
 207             fi->error_handler(fi, MFERR_WRITE);
 208           return 0;
 209         }
 210       fi->wpos += l;
 211     }
 212   DBG("MAIN: FD %d done write %d", fi->fd, fi->wpos);
 213   fi->write_handler = NULL;
 214   file_chg(fi);
 215   fi->write_done(fi);
 216   return 1;
 217 }
 218
 219 void
 220 file_read(struct main_file *fi, void *buf, uns len)
 221 {
 222   ASSERT(fi->n.next);
 223   if (len)
 224     {
 225       fi->read_handler = file_read_handler;
 226       fi->rbuf = buf;
 227       fi->rpos = 0;
 228       fi->rlen = len;
 229     }
 230   else
 231     {
 232       fi->read_handler = NULL;
 233       fi->rbuf = NULL;
 234       fi->rpos = fi->rlen = 0;
 235     }
 236   file_chg(fi);
 237 }
 238
 239 void
 240 file_write(struct main_file *fi, void *buf, uns len)
 241 {
 242   ASSERT(fi->n.next);
 243   if (len)
 244     {
 245       fi->write_handler = file_write_handler;
 246       fi->wbuf = buf;
 247       fi->wpos = 0;
 248       fi->wlen = len;
 249     }
 250   else
 251     {
 252       fi->write_handler = NULL;
 253       fi->wbuf = NULL;
 254       fi->wpos = fi->wlen = 0;
 255     }
 256   file_chg(fi);
 257 }
 258
 259 void
 260 file_set_timeout(struct main_file *fi, timestamp_t expires)
 261 {
 262   ASSERT(fi->n.next);
 263   timer_add(&fi->timer, expires);
 264 }
 265
 266 void
 267 file_close_all(void)
 268 {
 269   CLIST_FOR_EACH(struct main_file *, f, main_file_list)
 270     close(f->fd);
 271 }
 272
 273 void
 274 hook_add(struct main_hook *ho)
 275 {
 276   DBG("MAIN: Adding hook %p", ho);
 277   ASSERT(!ho->n.next);
 278   clist_add_tail(&main_hook_list, &ho->n);
 279 }
 280
 281 void
 282 hook_del(struct main_hook *ho)
 283 {
 284   DBG("MAIN: Deleting hook %p", ho);
 285   ASSERT(ho->n.next);
 286   clist_remove(&ho->n);
 287   ho->n.next = ho->n.prev = NULL;
 288 }
 289
 290 #ifdef USE_SELF_PIPE
 291 static void
 292 main_sigchld_handler(int x UNUSED)
 293 {
 294   int old_errno = errno;
 295   DBG("SIGCHLD received");
 296   chld_received = 1;
 297   ssize_t result;
 298   while((result = write(sig_pipe_send, "c", 1)) == -1 && errno == EINTR);
 299   if(result == -1 && errno != EAGAIN)
 300     msg(L_SIGHANDLER|L_ERROR, "Could not write to self-pipe: %m");
 301   errno = old_errno;
 302 }
 303
 304 static int
 305 dummy_read_handler(struct main_file *mp)
 306 {
 307   char buffer[1024];
 308   ssize_t result = read(mp->fd, buffer, 1024);
 309   if(result == -1 && errno != EAGAIN)
 310     msg(L_ERROR, "Could not read from selfpipe: %m");
 311   file_chg(mp);
 312   return result == 1024;
 313 }
 314
 315 static void
 316 pipe_configure(int fd)
 317 {
 318   int flags;
 319   if((flags = fcntl(fd, F_GETFL)) == -1 || fcntl(fd, F_SETFL, flags|O_NONBLOCK))
 320     die("Could not set file descriptor %d to non-blocking: %m", fd);
 321   if((flags = fcntl(fd, F_GETFD)) == -1 || fcntl(fd, F_SETFD, flags|O_CLOEXEC))
 322     die("Could not set file descriptor %d to close-on-exec: %m", fd);
 323 }
 324 #else
 325 static void
 326 main_sigchld_handler(int x UNUSED)
 327 {
 328   DBG("SIGCHLD received");
 329 }
 330 #endif
 331
 332 void
 333 process_add(struct main_process *mp)
 334 {
 335   DBG("MAIN: Adding process %p (pid=%d)", mp, mp->pid);
 336   ASSERT(!mp->n.next);
 337   ASSERT(mp->handler);
 338   clist_add_tail(&main_process_list, &mp->n);
 339   if (!main_sigchld_set_up)
 340     {
 341 #ifdef USE_SELF_PIPE
 342       int pipe_result[2];
 343       if(pipe(pipe_result) == -1)
 344         die("Could not create selfpipe:%m");
 345       pipe_configure(pipe_result[0]);
 346       pipe_configure(pipe_result[1]);
 347       sig_pipe_recv = pipe_result[0];
 348       sig_pipe_send = pipe_result[1];
 349       static struct main_file self_pipe;
 350       self_pipe = (struct main_file) {
 351         .fd = sig_pipe_recv,
 352         .read_handler = dummy_read_handler
 353       };
 354       file_add(&self_pipe);
 355 #endif
 356       struct sigaction sa;
 357       bzero(&sa, sizeof(sa));
 358       sa.sa_handler = main_sigchld_handler;
 359       sa.sa_flags = SA_NOCLDSTOP | SA_RESTART;
 360       sigaction(SIGCHLD, &sa, NULL);
 361       main_sigchld_set_up = 1;
 362       chld_received = 1; // The signal may have come before the handler
 363     }
 364 }
 365
 366 void
 367 process_del(struct main_process *mp)
 368 {
 369   DBG("MAIN: Deleting process %p (pid=%d)", mp, mp->pid);
 370   ASSERT(mp->n.next);
 371   clist_remove(&mp->n);
 372   mp->n.next = NULL;
 373 }
 374
 375 int
 376 process_fork(struct main_process *mp)
 377 {
 378   pid_t pid = fork();
 379   if (pid < 0)
 380     {
 381       DBG("MAIN: Fork failed");
 382       mp->status = -1;
 383       format_exit_status(mp->status_msg, -1);
 384       mp->handler(mp);
 385       return 1;
 386     }
 387   else if (!pid)
 388     return 0;
 389   else
 390     {
 391       DBG("MAIN: Forked process %d", (int) pid);
 392       mp->pid = pid;
 393       process_add(mp);
 394       return 1;
 395     }
 396 }
 397
 398 void
 399 main_debug(void)
 400 {
 401 #ifdef CONFIG_DEBUG
 402   msg(L_DEBUG, "### Main loop status on %lld", (long long)main_now);
 403   msg(L_DEBUG, "\tActive timers:");
 404   for (uns i = 1; i <= main_timer_cnt; i++)
 405     {
 406       struct main_timer *tm = main_timer_table.ptr[i];
 407       msg(L_DEBUG, "\t\t%p (expires %lld, data %p)", tm, (long long)(tm->expires ? tm->expires-main_now : 999999), tm->data);
 408     }
 409   struct main_file *fi;
 410   msg(L_DEBUG, "\tActive files:");
 411   CLIST_WALK(fi, main_file_list)
 412     msg(L_DEBUG, "\t\t%p (fd %d, rh %p, wh %p, eh %p, expires %lld, data %p)",
 413         fi, fi->fd, fi->read_handler, fi->write_handler, fi->error_handler,
 414         (long long)(fi->timer.expires ? fi->timer.expires-main_now : 999999), fi->data);
 415   msg(L_DEBUG, "\tActive hooks:");
 416   struct main_hook *ho;
 417   CLIST_WALK(ho, main_hook_done_list)
 418     msg(L_DEBUG, "\t\t%p (func %p, data %p)", ho, ho->handler, ho->data);
 419   CLIST_WALK(ho, main_hook_list)
 420     msg(L_DEBUG, "\t\t%p (func %p, data %p)", ho, ho->handler, ho->data);
 421   msg(L_DEBUG, "\tActive processes:");
 422   struct main_process *pr;
 423   CLIST_WALK(pr, main_process_list)
 424     msg(L_DEBUG, "\t\t%p (pid %d, data %p)", pr, pr->pid, pr->data);
 425 #endif
 426 }
 427
 428 static void
 429 main_rebuild_poll_table(void)
 430 {
 431   struct main_file *fi;
 432   if (main_poll_table_size < main_file_cnt)
 433     {
 434       if (main_poll_table)
 435         xfree(main_poll_table);
 436       else
 437         main_poll_table_size = 1;
 438       while (main_poll_table_size < main_file_cnt)
 439         main_poll_table_size *= 2;
 440       main_poll_table = xmalloc(sizeof(struct pollfd) * main_poll_table_size);
 441     }
 442   struct pollfd *p = main_poll_table;
 443   DBG("MAIN: Rebuilding poll table: %d of %d entries set", main_file_cnt, main_poll_table_size);
 444   CLIST_WALK(fi, main_file_list)
 445     {
 446       p->fd = fi->fd;
 447       fi->pollfd = p++;
 448       file_chg(fi);
 449     }
 450   main_poll_table_obsolete = 0;
 451 }
 452
 453 void
 454 main_loop(void)
 455 {
 456   DBG("MAIN: Entering main_loop");
 457   ASSERT(main_hook_list.head.next);
 458
 459   struct main_file *fi;
 460   struct main_hook *ho;
 461   struct main_timer *tm;
 462   struct main_process *pr;
 463
 464   main_get_time();
 465   for (;;)
 466     {
 467       timestamp_t wake = main_now + 1000000000;
 468       while (main_timer_cnt && (tm = main_timer_table.ptr[1])->expires <= main_now)
 469         {
 470           DBG("MAIN: Timer %p expired at now-%lld", tm, (long long)(main_now - tm->expires));
 471           tm->handler(tm);
 472         }
 473       int hook_min = HOOK_RETRY;
 474       int hook_max = HOOK_SHUTDOWN;
 475       while (ho = clist_remove_head(&main_hook_list))
 476         {
 477           clist_add_tail(&main_hook_done_list, &ho->n);
 478           DBG("MAIN: Hook %p", ho);
 479           int ret = ho->handler(ho);
 480           hook_min = MIN(hook_min, ret);
 481           hook_max = MAX(hook_max, ret);
 482         }
 483       clist_move(&main_hook_list, &main_hook_done_list);
 484       if (hook_min == HOOK_SHUTDOWN ||
 485           hook_min == HOOK_DONE && hook_max == HOOK_DONE ||
 486           main_shutdown)
 487         {
 488           DBG("MAIN: Shut down by %s", main_shutdown ? "main_shutdown" : "a hook");
 489           return;
 490         }
 491       if (hook_max == HOOK_RETRY)
 492         wake = 0;
 493       if (main_poll_table_obsolete)
 494         main_rebuild_poll_table();
 495 #ifndef USE_SELF_PIPE
 496       // We don't have a reliable flag without the self-pipe.
 497       chld_received = 1;
 498 #endif
 499       if (chld_received && !clist_empty(&main_process_list))
 500         {
 501           int stat;
 502           pid_t pid;
 503           wake = MIN(wake, main_now + 10000);
 504           chld_received = 0;
 505           while ((pid = waitpid(-1, &stat, WNOHANG)) > 0)
 506             {
 507               DBG("MAIN: Child %d exited with status %x", pid, stat);
 508               CLIST_WALK(pr, main_process_list)
 509                 if (pr->pid == pid)
 510                   {
 511                     pr->status = stat;
 512                     process_del(pr);
 513                     format_exit_status(pr->status_msg, pr->status);
 514                     DBG("MAIN: Calling process exit handler");
 515                     pr->handler(pr);
 516                     break;
 517                   }
 518               wake = 0;
 519             }
 520         }
 521       if (main_timer_cnt && (tm = main_timer_table.ptr[1])->expires < wake)
 522         wake = tm->expires;
 523       main_get_time();
 524       int timeout = ((wake > main_now) ? wake - main_now : 0);
 525       DBG("MAIN: Poll for %d fds and timeout %d ms", main_file_cnt, timeout);
 526       int p = poll(main_poll_table, main_file_cnt, timeout);
 527       timestamp_t old_now = main_now;
 528       main_get_time();
 529       main_idle_time += main_now - old_now;
 530       if (p > 0)
 531         {
 532           struct pollfd *p = main_poll_table;
 533           CLIST_WALK(fi, main_file_list)
 534             {
 535               if (p->revents & (POLLIN | POLLHUP | POLLERR))
 536                 {
 537                   do
 538                     DBG("MAIN: Read event on fd %d", p->fd);
 539                   while (fi->read_handler && fi->read_handler(fi) && !main_poll_table_obsolete);
 540                   if (main_poll_table_obsolete) /* File entries have been inserted or deleted => better not risk continuing to nowhere */
 541                     break;
 542                 }
 543               if (p->revents & (POLLOUT | POLLERR))
 544                 {
 545                   do
 546                     DBG("MAIN: Write event on fd %d", p->fd);
 547                   while (fi->write_handler && fi->write_handler(fi) && !main_poll_table_obsolete);
 548                   if (main_poll_table_obsolete)
 549                     break;
 550                 }
 551               p++;
 552             }
 553         }
 554     }
 555 }
 556
 557 #ifdef TEST
 558
 559 static struct main_process mp;
 560 static struct main_file fin, fout;
 561 static struct main_hook hook;
 562 static struct main_timer tm;
 563
 564 static byte rb[16];
 565
 566 static void dread(struct main_file *fi)
 567 {
 568   if (fi->rpos < fi->rlen)
 569     {
 570       msg(L_INFO, "Read EOF");
 571       file_del(fi);
 572     }
 573   else
 574     {
 575       msg(L_INFO, "Read done");
 576       file_read(fi, rb, sizeof(rb));
 577     }
 578 }
 579
 580 static void derror(struct main_file *fi, int cause)
 581 {
 582   msg(L_INFO, "Error: %m !!! (cause %d)", cause);
 583   file_del(fi);
 584 }
 585
 586 static void dwrite(struct main_file *fi UNUSED)
 587 {
 588   msg(L_INFO, "Write done");
 589 }
 590
 591 static int dhook(struct main_hook *ho UNUSED)
 592 {
 593   msg(L_INFO, "Hook called");
 594   return 0;
 595 }
 596
 597 static void dtimer(struct main_timer *tm)
 598 {
 599   msg(L_INFO, "Timer tick");
 600   timer_add(tm, main_now + 10000);
 601 }
 602
 603 static void dentry(void)
 604 {
 605   msg(L_INFO, "*** SUBPROCESS START ***");
 606   sleep(2);
 607   msg(L_INFO, "*** SUBPROCESS FINISH ***");
 608   exit(0);
 609 }
 610
 611 static void dexit(struct main_process *pr)
 612 {
 613   msg(L_INFO, "Subprocess %d exited with status %x", pr->pid, pr->status);
 614 }
 615
 616 int
 617 main(void)
 618 {
 619   log_init(NULL);
 620   main_init();
 621
 622   fin.fd = 0;
 623   fin.read_done = dread;
 624   fin.error_handler = derror;
 625   file_add(&fin);
 626   file_read(&fin, rb, sizeof(rb));
 627
 628   fout.fd = 1;
 629   fout.write_done = dwrite;
 630   fout.error_handler = derror;
 631   file_add(&fout);
 632   file_write(&fout, "Hello, world!\n", 14);
 633
 634   hook.handler = dhook;
 635   hook_add(&hook);
 636
 637   tm.handler = dtimer;
 638   timer_add(&tm, main_now + 1000);
 639
 640   mp.handler = dexit;
 641   if (!process_fork(&mp))
 642     dentry();
 643
 644   main_debug();
 645
 646   main_loop();
 647   msg(L_INFO, "Finished.");
 648 }
 649
 650 #endif