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