Pipe kick-and-drain packed into a neat structure and functions.

3 files changed, 66 insertions, 41 deletions

diff --git a/sysdep/unix/io-loop.c b/sysdep/unix/io-loop.c
index 575e5403..2e2bb86e 100644
--- a/sysdep/unix/io-loop.c
+++ b/sysdep/unix/io-loop.c
@@ -88,74 +88,94 @@ birdloop_process_flags(struct birdloop *loop)
  *	Wakeup code for birdloop
  */
 
-static void
-pipe_new(int *pfds)
+void
+pipe_new(struct pipe *p)
 {
-  int rv = pipe(pfds);
+  int rv = pipe(p->fd);
   if (rv < 0)
     die("pipe: %m");
 
-  if (fcntl(pfds[0], F_SETFL, O_NONBLOCK) < 0)
+  if (fcntl(p->fd[0], F_SETFL, O_NONBLOCK) < 0)
     die("fcntl(O_NONBLOCK): %m");
 
-  if (fcntl(pfds[1], F_SETFL, O_NONBLOCK) < 0)
+  if (fcntl(p->fd[1], F_SETFL, O_NONBLOCK) < 0)
     die("fcntl(O_NONBLOCK): %m");
 }
 
 void
-pipe_drain(int fd)
+pipe_drain(struct pipe *p)
 {
-  char buf[64];
-  int rv;
-  
- try:
-  rv = read(fd, buf, 64);
-  if (rv < 0)
-  {
-    if (errno == EINTR)
-      goto try;
-    if (errno == EAGAIN)
+  while (1) {
+    char buf[64];
+    int rv = read(p->fd[0], buf, sizeof(buf));
+    if ((rv < 0) && (errno == EAGAIN))
       return;
-    die("wakeup read: %m");
+
+    if (rv == 0)
+      bug("wakeup read eof");
+    if ((rv < 0) && (errno != EINTR))
+      bug("wakeup read: %m");
+  }
+}
+
+int
+pipe_read_one(struct pipe *p)
+{
+  while (1) {
+    char v;
+    int rv = read(p->fd[0], &v, sizeof(v));
+    if (rv == 1)
+      return 1;
+    if ((rv < 0) && (errno == EAGAIN))
+      return 0;
+    if (rv > 1)
+      bug("wakeup read more bytes than expected: %d", rv);
+    if (rv == 0)
+      bug("wakeup read eof");
+    if (errno != EINTR)
+      bug("wakeup read: %m");
   }
-  if (rv == 64)
-    goto try;
 }
 
 void
-pipe_kick(int fd)
+pipe_kick(struct pipe *p)
 {
-  u64 v = 1;
+  char v = 1;
   int rv;
 
- try:
-  rv = write(fd, &v, sizeof(u64));
-  if (rv < 0)
-  {
-    if (errno == EINTR)
-      goto try;
-    if (errno == EAGAIN)
+  while (1) {
+    rv = write(p->fd[1], &v, sizeof(v));
+    if ((rv >= 0) || (errno == EAGAIN)) 
       return;
-    die("wakeup write: %m");
+    if (errno != EINTR)
+      bug("wakeup write: %m");
   }
 }
 
+void
+pipe_pollin(struct pipe *p, struct pollfd *pfd)
+{
+  pfd->fd = p->fd[0];
+  pfd->events = POLLIN;
+  pfd->revents = 0;
+}
+
 static inline void
 wakeup_init(struct birdloop *loop)
 {
-  pipe_new(loop->wakeup_fds);
+  pipe_new(&loop->wakeup);
 }
 
 static inline void
 wakeup_drain(struct birdloop *loop)
 {
-  pipe_drain(loop->wakeup_fds[0]);
+  pipe_drain(&loop->wakeup);
 }
 
 static inline void
 wakeup_do_kick(struct birdloop *loop)
 {
-  pipe_kick(loop->wakeup_fds[1]);
+  pipe_kick(&loop->wakeup);
 }
 
 static inline void
@@ -284,9 +304,7 @@ sockets_prepare(struct birdloop *loop)
 
   /* Add internal wakeup fd */
   *psk = NULL;
-  pfd->fd = loop->wakeup_fds[0];
-  pfd->events = POLLIN;
-  pfd->revents = 0;
+  pipe_pollin(&loop->wakeup, pfd);
 
   loop->poll_changed = 0;
 }
diff --git a/sysdep/unix/io-loop.h b/sysdep/unix/io-loop.h
index aec7a409..29ca96d6 100644
--- a/sysdep/unix/io-loop.h
+++ b/sysdep/unix/io-loop.h
@@ -9,6 +9,16 @@
 
 #include "lib/rcu.h"
 
+struct pipe
+{
+  int fd[2];
+};
+
+void pipe_new(struct pipe *);
+void pipe_pollin(struct pipe *, struct pollfd *);
+void pipe_drain(struct pipe *);
+void pipe_kick(struct pipe *);
+
 struct birdloop
 {
   pool *pool;
@@ -25,7 +35,7 @@ struct birdloop
 
   uint ping_pending;
   _Atomic u32 ping_sent;
-  int wakeup_fds[2];
+  struct pipe wakeup;
 
   pthread_t thread_id;
   pthread_attr_t thread_attr;
diff --git a/sysdep/unix/io.c b/sysdep/unix/io.c
index 23baffb2..6454f15f 100644
--- a/sysdep/unix/io.c
+++ b/sysdep/unix/io.c
@@ -2213,8 +2213,6 @@ static int short_loops = 0;
 #define SHORT_LOOP_MAX 10
 #define WORK_EVENTS_MAX 10
 
-void pipe_drain(int fd);
-
 void
 io_loop(void)
 {
@@ -2246,8 +2244,7 @@ io_loop(void)
       }
 
       /* A hack to reload main io_loop() when something has changed asynchronously. */
-      pfd[0].fd = main_birdloop.wakeup_fds[0];
-      pfd[0].events = POLLIN;
+      pipe_pollin(&main_birdloop.wakeup, &pfd[0]);
 
       nfds = 1;
 
@@ -2325,7 +2322,7 @@ io_loop(void)
 	  if (pfd[0].revents & POLLIN)
 	  {
 	    /* IO loop reload requested */
-	    pipe_drain(main_birdloop.wakeup_fds[0]);
+	    pipe_drain(&main_birdloop.wakeup);
 	    atomic_exchange_explicit(&main_birdloop.ping_sent, 0, memory_order_acq_rel);
 	    continue;
 	  }