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