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