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