![[BACK]](/icons/cvsweb/back.gif){kind=icon}
Return to io.c CVS log ![[TXT]](/icons/cvsweb/text.gif){kind=icon}
![[DIR]](/icons/cvsweb/dir.gif){kind=icon}
Up to [ELWIX - Embedded LightWeight unIX -] / embedaddon / bird / proto / bfd|
Return to io.c CVS log | Up to [ELWIX - Embedded LightWeight unIX -] / embedaddon / bird / proto / bfd |
1.1 misho 1: /*
2: * BIRD -- I/O and event loop
3: *
4: * Can be freely distributed and used under the terms of the GNU GPL.
5: */
6:
7: #include <stdio.h>
8: #include <stdlib.h>
9: #include <unistd.h>
10: #include <errno.h>
11: #include <fcntl.h>
12: #include <poll.h>
13: #include <pthread.h>
14: #include <time.h>
15: #include <sys/time.h>
16:
17: #include "nest/bird.h"
18: #include "proto/bfd/io.h"
19:
20: #include "lib/buffer.h"
21: #include "lib/heap.h"
22: #include "lib/lists.h"
23: #include "lib/resource.h"
24: #include "lib/event.h"
25: #include "lib/socket.h"
26:
27:
28: struct birdloop
29: {
30: pool *pool;
31: pthread_t thread;
32: pthread_mutex_t mutex;
33:
34: btime last_time;
35: btime real_time;
36: u8 use_monotonic_clock;
37:
38: u8 stop_called;
39: u8 poll_active;
40: u8 wakeup_masked;
41: int wakeup_fds[2];
42:
43: BUFFER(timer2 *) timers;
44: list event_list;
45: list sock_list;
46: uint sock_num;
47:
48: BUFFER(sock *) poll_sk;
49: BUFFER(struct pollfd) poll_fd;
50: u8 poll_changed;
51: u8 close_scheduled;
52: };
53:
54:
55: /*
56: * Current thread context
57: */
58:
59: static pthread_key_t current_loop_key;
60:
61: static inline struct birdloop *
62: birdloop_current(void)
63: {
64: return pthread_getspecific(current_loop_key);
65: }
66:
67: static inline void
68: birdloop_set_current(struct birdloop *loop)
69: {
70: pthread_setspecific(current_loop_key, loop);
71: }
72:
73: static inline void
74: birdloop_init_current(void)
75: {
76: pthread_key_create(¤t_loop_key, NULL);
77: }
78:
79:
80: /*
81: * Time clock
82: */
83:
84: static void times_update_alt(struct birdloop *loop);
85:
86: static void
87: times_init(struct birdloop *loop)
88: {
89: struct timespec ts;
90: int rv;
91:
92: rv = clock_gettime(CLOCK_MONOTONIC, &ts);
93: if (rv < 0)
94: {
95: log(L_WARN "Monotonic clock is missing");
96:
97: loop->use_monotonic_clock = 0;
98: loop->last_time = 0;
99: loop->real_time = 0;
100: times_update_alt(loop);
101: return;
102: }
103:
104: if ((ts.tv_sec < 0) || (((s64) ts.tv_sec) > ((s64) 1 << 40)))
105: log(L_WARN "Monotonic clock is crazy");
106:
107: loop->use_monotonic_clock = 1;
108: loop->last_time = ((s64) ts.tv_sec S) + (ts.tv_nsec / 1000);
109: loop->real_time = 0;
110: }
111:
112: static void
113: times_update_pri(struct birdloop *loop)
114: {
115: struct timespec ts;
116: int rv;
117:
118: rv = clock_gettime(CLOCK_MONOTONIC, &ts);
119: if (rv < 0)
120: die("clock_gettime: %m");
121:
122: btime new_time = ((s64) ts.tv_sec S) + (ts.tv_nsec / 1000);
123:
124: if (new_time < loop->last_time)
125: log(L_ERR "Monotonic clock is broken");
126:
127: loop->last_time = new_time;
128: loop->real_time = 0;
129: }
130:
131: static void
132: times_update_alt(struct birdloop *loop)
133: {
134: struct timeval tv;
135: int rv;
136:
137: rv = gettimeofday(&tv, NULL);
138: if (rv < 0)
139: die("gettimeofday: %m");
140:
141: btime new_time = ((s64) tv.tv_sec S) + tv.tv_usec;
142: btime delta = new_time - loop->real_time;
143:
144: if ((delta < 0) || (delta > (60 S)))
145: {
146: if (loop->real_time)
147: log(L_WARN "Time jump, delta %d us", (int) delta);
148:
149: delta = 100 MS;
150: }
151:
152: loop->last_time += delta;
153: loop->real_time = new_time;
154: }
155:
156: static void
157: times_update(struct birdloop *loop)
158: {
159: if (loop->use_monotonic_clock)
160: times_update_pri(loop);
161: else
162: times_update_alt(loop);
163: }
164:
165: btime
166: current_time(void)
167: {
168: return birdloop_current()->last_time;
169: }
170:
171:
172: /*
173: * Wakeup code for birdloop
174: */
175:
176: static void
177: pipe_new(int *pfds)
178: {
179: int rv = pipe(pfds);
180: if (rv < 0)
181: die("pipe: %m");
182:
183: if (fcntl(pfds[0], F_SETFL, O_NONBLOCK) < 0)
184: die("fcntl(O_NONBLOCK): %m");
185:
186: if (fcntl(pfds[1], F_SETFL, O_NONBLOCK) < 0)
187: die("fcntl(O_NONBLOCK): %m");
188: }
189:
190: void
191: pipe_drain(int fd)
192: {
193: char buf[64];
194: int rv;
195:
196: try:
197: rv = read(fd, buf, 64);
198: if (rv < 0)
199: {
200: if (errno == EINTR)
201: goto try;
202: if (errno == EAGAIN)
203: return;
204: die("wakeup read: %m");
205: }
206: if (rv == 64)
207: goto try;
208: }
209:
210: void
211: pipe_kick(int fd)
212: {
213: u64 v = 1;
214: int rv;
215:
216: try:
217: rv = write(fd, &v, sizeof(u64));
218: if (rv < 0)
219: {
220: if (errno == EINTR)
221: goto try;
222: if (errno == EAGAIN)
223: return;
224: die("wakeup write: %m");
225: }
226: }
227:
228: static inline void
229: wakeup_init(struct birdloop *loop)
230: {
231: pipe_new(loop->wakeup_fds);
232: }
233:
234: static inline void
235: wakeup_drain(struct birdloop *loop)
236: {
237: pipe_drain(loop->wakeup_fds[0]);
238: }
239:
240: static inline void
241: wakeup_do_kick(struct birdloop *loop)
242: {
243: pipe_kick(loop->wakeup_fds[1]);
244: }
245:
246: static inline void
247: wakeup_kick(struct birdloop *loop)
248: {
249: if (!loop->wakeup_masked)
250: wakeup_do_kick(loop);
251: else
252: loop->wakeup_masked = 2;
253: }
254:
255:
256: /*
257: * Events
258: */
259:
260: static inline uint
261: events_waiting(struct birdloop *loop)
262: {
263: return !EMPTY_LIST(loop->event_list);
264: }
265:
266: static inline void
267: events_init(struct birdloop *loop)
268: {
269: init_list(&loop->event_list);
270: }
271:
272: static void
273: events_fire(struct birdloop *loop)
274: {
275: times_update(loop);
276: ev_run_list(&loop->event_list);
277: }
278:
279: void
280: ev2_schedule(event *e)
281: {
282: struct birdloop *loop = birdloop_current();
283:
284: if (loop->poll_active && EMPTY_LIST(loop->event_list))
285: wakeup_kick(loop);
286:
287: if (e->n.next)
288: rem_node(&e->n);
289:
290: add_tail(&loop->event_list, &e->n);
291: }
292:
293:
294: /*
295: * Timers
296: */
297:
298: #define TIMER_LESS(a,b) ((a)->expires < (b)->expires)
299: #define TIMER_SWAP(heap,a,b,t) (t = heap[a], heap[a] = heap[b], heap[b] = t, \
300: heap[a]->index = (a), heap[b]->index = (b))
301:
302: static inline uint timers_count(struct birdloop *loop)
303: { return loop->timers.used - 1; }
304:
305: static inline timer2 *timers_first(struct birdloop *loop)
306: { return (loop->timers.used > 1) ? loop->timers.data[1] : NULL; }
307:
308:
309: static void
310: tm2_free(resource *r)
311: {
312: timer2 *t = (timer2 *) r;
313:
314: tm2_stop(t);
315: }
316:
317: static void
318: tm2_dump(resource *r)
319: {
320: timer2 *t = (timer2 *) r;
321:
322: debug("(code %p, data %p, ", t->hook, t->data);
323: if (t->randomize)
324: debug("rand %d, ", t->randomize);
325: if (t->recurrent)
326: debug("recur %d, ", t->recurrent);
327: if (t->expires)
328: debug("expires in %d ms)\n", (t->expires - current_time()) TO_MS);
329: else
330: debug("inactive)\n");
331: }
332:
333:
334: static struct resclass tm2_class = {
335: "Timer",
336: sizeof(timer2),
337: tm2_free,
338: tm2_dump,
339: NULL,
340: NULL
341: };
342:
343: timer2 *
344: tm2_new(pool *p)
345: {
346: timer2 *t = ralloc(p, &tm2_class);
347: t->index = -1;
348: return t;
349: }
350:
351: void
352: tm2_set(timer2 *t, btime when)
353: {
354: struct birdloop *loop = birdloop_current();
355: uint tc = timers_count(loop);
356:
357: if (!t->expires)
358: {
359: t->index = ++tc;
360: t->expires = when;
361: BUFFER_PUSH(loop->timers) = t;
362: HEAP_INSERT(loop->timers.data, tc, timer2 *, TIMER_LESS, TIMER_SWAP);
363: }
364: else if (t->expires < when)
365: {
366: t->expires = when;
367: HEAP_INCREASE(loop->timers.data, tc, timer2 *, TIMER_LESS, TIMER_SWAP, t->index);
368: }
369: else if (t->expires > when)
370: {
371: t->expires = when;
372: HEAP_DECREASE(loop->timers.data, tc, timer2 *, TIMER_LESS, TIMER_SWAP, t->index);
373: }
374:
375: if (loop->poll_active && (t->index == 1))
376: wakeup_kick(loop);
377: }
378:
379: void
380: tm2_start(timer2 *t, btime after)
381: {
382: tm2_set(t, current_time() + MAX(after, 0));
383: }
384:
385: void
386: tm2_stop(timer2 *t)
387: {
388: if (!t->expires)
389: return;
390:
391: struct birdloop *loop = birdloop_current();
392: uint tc = timers_count(loop);
393:
394: HEAP_DELETE(loop->timers.data, tc, timer2 *, TIMER_LESS, TIMER_SWAP, t->index);
395: BUFFER_POP(loop->timers);
396:
397: t->index = -1;
398: t->expires = 0;
399: }
400:
401: static void
402: timers_init(struct birdloop *loop)
403: {
404: BUFFER_INIT(loop->timers, loop->pool, 4);
405: BUFFER_PUSH(loop->timers) = NULL;
406: }
407:
408: static void
409: timers_fire(struct birdloop *loop)
410: {
411: btime base_time;
412: timer2 *t;
413:
414: times_update(loop);
415: base_time = loop->last_time;
416:
417: while (t = timers_first(loop))
418: {
419: if (t->expires > base_time)
420: return;
421:
422: if (t->recurrent)
423: {
424: btime when = t->expires + t->recurrent;
425:
426: if (when <= loop->last_time)
427: when = loop->last_time + t->recurrent;
428:
429: if (t->randomize)
430: when += random() % (t->randomize + 1);
431:
432: tm2_set(t, when);
433: }
434: else
435: tm2_stop(t);
436:
437: t->hook(t);
438: }
439: }
440:
441:
442: /*
443: * Sockets
444: */
445:
446: static void
447: sockets_init(struct birdloop *loop)
448: {
449: init_list(&loop->sock_list);
450: loop->sock_num = 0;
451:
452: BUFFER_INIT(loop->poll_sk, loop->pool, 4);
453: BUFFER_INIT(loop->poll_fd, loop->pool, 4);
454: loop->poll_changed = 1; /* add wakeup fd */
455: }
456:
457: static void
458: sockets_add(struct birdloop *loop, sock *s)
459: {
460: add_tail(&loop->sock_list, &s->n);
461: loop->sock_num++;
462:
463: s->index = -1;
464: loop->poll_changed = 1;
465:
466: if (loop->poll_active)
467: wakeup_kick(loop);
468: }
469:
470: void
471: sk_start(sock *s)
472: {
473: struct birdloop *loop = birdloop_current();
474:
475: sockets_add(loop, s);
476: }
477:
478: static void
479: sockets_remove(struct birdloop *loop, sock *s)
480: {
481: rem_node(&s->n);
482: loop->sock_num--;
483:
484: if (s->index >= 0)
485: loop->poll_sk.data[s->index] = NULL;
486:
487: s->index = -1;
488: loop->poll_changed = 1;
489:
490: /* Wakeup moved to sk_stop() */
491: }
492:
493: void
494: sk_stop(sock *s)
495: {
496: struct birdloop *loop = birdloop_current();
497:
498: sockets_remove(loop, s);
499:
500: if (loop->poll_active)
501: {
502: loop->close_scheduled = 1;
503: wakeup_kick(loop);
504: }
505: else
506: close(s->fd);
507:
508: s->fd = -1;
509: }
510:
511: static inline uint sk_want_events(sock *s)
512: { return (s->rx_hook ? POLLIN : 0) | ((s->ttx != s->tpos) ? POLLOUT : 0); }
513:
514: /*
515: FIXME: this should be called from sock code
516:
517: static void
518: sockets_update(struct birdloop *loop, sock *s)
519: {
520: if (s->index >= 0)
521: loop->poll_fd.data[s->index].events = sk_want_events(s);
522: }
523: */
524:
525: static void
526: sockets_prepare(struct birdloop *loop)
527: {
528: BUFFER_SET(loop->poll_sk, loop->sock_num + 1);
529: BUFFER_SET(loop->poll_fd, loop->sock_num + 1);
530:
531: struct pollfd *pfd = loop->poll_fd.data;
532: sock **psk = loop->poll_sk.data;
533: int i = 0;
534: node *n;
535:
536: WALK_LIST(n, loop->sock_list)
537: {
538: sock *s = SKIP_BACK(sock, n, n);
539:
540: ASSERT(i < loop->sock_num);
541:
542: s->index = i;
543: *psk = s;
544: pfd->fd = s->fd;
545: pfd->events = sk_want_events(s);
546: pfd->revents = 0;
547:
548: pfd++;
549: psk++;
550: i++;
551: }
552:
553: ASSERT(i == loop->sock_num);
554:
555: /* Add internal wakeup fd */
556: *psk = NULL;
557: pfd->fd = loop->wakeup_fds[0];
558: pfd->events = POLLIN;
559: pfd->revents = 0;
560:
561: loop->poll_changed = 0;
562: }
563:
564: static void
565: sockets_close_fds(struct birdloop *loop)
566: {
567: struct pollfd *pfd = loop->poll_fd.data;
568: sock **psk = loop->poll_sk.data;
569: int poll_num = loop->poll_fd.used - 1;
570:
571: int i;
572: for (i = 0; i < poll_num; i++)
573: if (psk[i] == NULL)
574: close(pfd[i].fd);
575:
576: loop->close_scheduled = 0;
577: }
578:
579: int sk_read(sock *s, int revents);
580: int sk_write(sock *s);
581:
582: static void
583: sockets_fire(struct birdloop *loop)
584: {
585: struct pollfd *pfd = loop->poll_fd.data;
586: sock **psk = loop->poll_sk.data;
587: int poll_num = loop->poll_fd.used - 1;
588:
589: times_update(loop);
590:
591: /* Last fd is internal wakeup fd */
592: if (pfd[poll_num].revents & POLLIN)
593: wakeup_drain(loop);
594:
595: int i;
596: for (i = 0; i < poll_num; pfd++, psk++, i++)
597: {
598: int e = 1;
599:
600: if (! pfd->revents)
601: continue;
602:
603: if (pfd->revents & POLLNVAL)
604: die("poll: invalid fd %d", pfd->fd);
605:
606: if (pfd->revents & POLLIN)
607: while (e && *psk && (*psk)->rx_hook)
608: e = sk_read(*psk, 0);
609:
610: e = 1;
611: if (pfd->revents & POLLOUT)
612: while (e && *psk)
613: e = sk_write(*psk);
614: }
615: }
616:
617:
618: /*
619: * Birdloop
620: */
621:
622: static void * birdloop_main(void *arg);
623:
624: struct birdloop *
625: birdloop_new(void)
626: {
627: /* FIXME: this init should be elsewhere and thread-safe */
628: static int init = 0;
629: if (!init)
630: { birdloop_init_current(); init = 1; }
631:
632: pool *p = rp_new(NULL, "Birdloop root");
633: struct birdloop *loop = mb_allocz(p, sizeof(struct birdloop));
634: loop->pool = p;
635: pthread_mutex_init(&loop->mutex, NULL);
636:
637: times_init(loop);
638: wakeup_init(loop);
639:
640: events_init(loop);
641: timers_init(loop);
642: sockets_init(loop);
643:
644: return loop;
645: }
646:
647: void
648: birdloop_start(struct birdloop *loop)
649: {
650: int rv = pthread_create(&loop->thread, NULL, birdloop_main, loop);
651: if (rv)
652: die("pthread_create(): %M", rv);
653: }
654:
655: void
656: birdloop_stop(struct birdloop *loop)
657: {
658: pthread_mutex_lock(&loop->mutex);
659: loop->stop_called = 1;
660: wakeup_do_kick(loop);
661: pthread_mutex_unlock(&loop->mutex);
662:
663: int rv = pthread_join(loop->thread, NULL);
664: if (rv)
665: die("pthread_join(): %M", rv);
666: }
667:
668: void
669: birdloop_free(struct birdloop *loop)
670: {
671: rfree(loop->pool);
672: }
673:
674:
675: void
676: birdloop_enter(struct birdloop *loop)
677: {
678: /* TODO: these functions could save and restore old context */
679: pthread_mutex_lock(&loop->mutex);
680: birdloop_set_current(loop);
681: }
682:
683: void
684: birdloop_leave(struct birdloop *loop)
685: {
686: /* TODO: these functions could save and restore old context */
687: birdloop_set_current(NULL);
688: pthread_mutex_unlock(&loop->mutex);
689: }
690:
691: void
692: birdloop_mask_wakeups(struct birdloop *loop)
693: {
694: pthread_mutex_lock(&loop->mutex);
695: loop->wakeup_masked = 1;
696: pthread_mutex_unlock(&loop->mutex);
697: }
698:
699: void
700: birdloop_unmask_wakeups(struct birdloop *loop)
701: {
702: pthread_mutex_lock(&loop->mutex);
703: if (loop->wakeup_masked == 2)
704: wakeup_do_kick(loop);
705: loop->wakeup_masked = 0;
706: pthread_mutex_unlock(&loop->mutex);
707: }
708:
709: static void *
710: birdloop_main(void *arg)
711: {
712: struct birdloop *loop = arg;
713: timer2 *t;
714: int rv, timeout;
715:
716: birdloop_set_current(loop);
717:
718: pthread_mutex_lock(&loop->mutex);
719: while (1)
720: {
721: events_fire(loop);
722: timers_fire(loop);
723:
724: times_update(loop);
725: if (events_waiting(loop))
726: timeout = 0;
727: else if (t = timers_first(loop))
728: timeout = (tm2_remains(t) TO_MS) + 1;
729: else
730: timeout = -1;
731:
732: if (loop->poll_changed)
733: sockets_prepare(loop);
734:
735: loop->poll_active = 1;
736: pthread_mutex_unlock(&loop->mutex);
737:
738: try:
739: rv = poll(loop->poll_fd.data, loop->poll_fd.used, timeout);
740: if (rv < 0)
741: {
742: if (errno == EINTR || errno == EAGAIN)
743: goto try;
744: die("poll: %m");
745: }
746:
747: pthread_mutex_lock(&loop->mutex);
748: loop->poll_active = 0;
749:
750: if (loop->close_scheduled)
751: sockets_close_fds(loop);
752:
753: if (loop->stop_called)
754: break;
755:
756: if (rv)
757: sockets_fire(loop);
758:
759: timers_fire(loop);
760: }
761:
762: loop->stop_called = 0;
763: pthread_mutex_unlock(&loop->mutex);
764:
765: return NULL;
766: }
767:
768: