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