Merge branch 'mq-coro' into mq-config-new

aclocal.m4

@@ -17,8 +17,6 @@ AC_DEFUN([BIRD_CHECK_PTHREADS],
 	    [
 	      pthread_t pt;
 	      pthread_create(&pt, NULL, NULL, NULL);
-	      pthread_spinlock_t lock;
-	      pthread_spin_lock(&lock);
 	    ]
 	  )
 	],
@@ -31,6 +29,30 @@ AC_DEFUN([BIRD_CHECK_PTHREADS],
   CFLAGS="$bird_tmp_cflags"
 ])
 
+dnl ** Android API before version 24 doesn't implement spinlocks.
+AC_DEFUN([BIRD_CHECK_PTHREAD_SPINLOCK],
+[
+  AC_CACHE_CHECK(
+    [whether POSIX threads provide spinlocks],
+    [bird_cv_lib_pthread_spinlock],
+    [
+      AC_LINK_IFELSE(
+	[
+	  AC_LANG_PROGRAM(
+	    [ #include <pthread.h> ],
+	    [
+	      pthread_spinlock_t lock;
+	      pthread_spin_lock(&lock);
+	    ]
+	  )
+	],
+	[bird_cv_lib_pthread_spinlock=yes],
+	[bird_cv_lib_pthread_spinlock=no]
+      )
+    ]
+  )
+])
+
 AC_DEFUN([BIRD_CHECK_MPLS_KERNEL],
 [
   AC_CACHE_CHECK(

configure.ac

@@ -120,7 +120,13 @@ if test "$enable_pthreads" != no ; then
     AC_DEFINE([USE_PTHREADS], [1], [Define to 1 if pthreads are enabled])
     CFLAGS="$CFLAGS -pthread"
     LDFLAGS="$LDFLAGS -pthread"
+
+    BIRD_CHECK_PTHREAD_SPINLOCK
+    if test "$bird_cv_lib_pthread_spinlock" = yes ; then
+      AC_DEFINE([USE_PTHREAD_SPINLOCK], [1], [Define to 1 if pthreads provide spinlocks])
       proto_bfd=bfd
+    fi
+
   elif test "$enable_pthreads" = yes ; then
     AC_MSG_ERROR([POSIX threads not available.])
   fi

doc/bird.sgml

@@ -653,7 +653,7 @@ agreement").
 
 	<tag><label id="proto-description">description "<m/text/"</tag>
 	This is an optional description of the protocol. It is displayed as a
-	part of the output of 'show route all' command.
+	part of the output of 'show protocols all' command.
 
 	<tag><label id="proto-vrf">vrf "<m/text/"</tag>
 	Associate the protocol with specific VRF. The protocol will be

lib/coroutine.h

@@ -0,0 +1,26 @@
+/*
+ *	BIRD Coroutines
+ *
+ *	(c) 2017 Martin Mares <mj@ucw.cz>
+ *
+ *	Can be freely distributed and used under the terms of the GNU GPL.
+ */
+
+#ifndef _BIRD_COROUTINE_H_
+#define _BIRD_COROUTINE_H_
+
+#include "lib/event.h"
+
+// The structure is completely opaque, implemented by sysdep
+typedef struct coroutine coroutine;
+
+coroutine *coro_new(struct pool *pool, void (*entry_point)(void *arg), void *arg);
+void coro_suspend(void);
+void coro_resume(coroutine *c);
+void coro_done(event *e) NORET;
+
+struct birdsock;
+int coro_sk_read(struct birdsock *s);
+void coro_sk_write(struct birdsock *s, unsigned len);
+
+#endif

lib/socket.h

@@ -36,6 +36,8 @@ struct ssh_sock {
 };
 #endif
 
+struct coroutine;
+
 typedef struct birdsock {
   resource r;
   pool *pool;				/* Pool where incoming connections should be allocated (for SK_xxx_PASSIVE) */
@@ -79,6 +81,9 @@ typedef struct birdsock {
   char *password;			/* Password for MD5 authentication */
   const char *err;			/* Error message */
   struct ssh_sock *ssh;			/* Used in SK_SSH */
+
+  struct coroutine *rx_coroutine;
+  struct coroutine *tx_coroutine;
 } sock;
 
 sock *sock_new(pool *);			/* Allocate new socket */

nest/cli.c

@@ -1,7 +1,7 @@
 /*
  *	BIRD Internet Routing Daemon -- Command-Line Interface
  *
- *	(c) 1999--2000 Martin Mares <mj@ucw.cz>
+ *	(c) 1999--2017 Martin Mares <mj@ucw.cz>
  *
  *	Can be freely distributed and used under the terms of the GNU GPL.
  */
@@ -32,44 +32,55 @@
  * Each CLI session is internally represented by a &cli structure and a
  * resource pool containing all resources associated with the connection,
  * so that it can be easily freed whenever the connection gets closed, not depending
- * on the current state of command processing.
+ * on the current state of command processing. A socket is associated with
+ * the session, over which requests and replies are sent.
  *
  * The CLI commands are declared as a part of the configuration grammar
  * by using the |CF_CLI| macro. When a command is received, it is processed
  * by the same lexical analyzer and parser as used for the configuration, but
  * it's switched to a special mode by prepending a fake token to the text,
  * so that it uses only the CLI command rules. Then the parser invokes
- * an execution routine corresponding to the command, which either constructs
+ * an execution routine corresponding to the command, which constructs the
- * the whole reply and returns it back or (in case it expects the reply will be long)
+ * reply.
- * it prints a partial reply and asks the CLI module (using the @cont hook)
+ *
- * to call it again when the output is transferred to the user.
+ * Replies are buffered in memory and then sent asynchronously. Commands
+ * which produce long outputs must split them to pieces and yield to other
+ * operations between pieces. To simplify this (and possibly also complex
+ * parsing of input), the CLI session runs in a coroutine with its own
+ * execution context. At any time, cli_yield() can be called to interrupt
+ * the current coroutine and have the buffered output sent.
+ *
+ * Alternatively, a long sequence of replies can be split to parts
+ * using the @cont hook, which translates to yielding internally.
  *
  * The @this_cli variable points to a &cli structure of the session being
- * currently parsed, but it's of course available only in command handlers
+ * currently parsed, but it's available only before the first yield.
- * not entered using the @cont hook.
- *
- * TX buffer management works as follows: At cli.tx_buf there is a
- * list of TX buffers (struct cli_out), cli.tx_write is the buffer
- * currently used by the producer (cli_printf(), cli_alloc_out()) and
- * cli.tx_pos is the buffer currently used by the consumer
- * (cli_write(), in system dependent code). The producer uses
- * cli_out.wpos ptr as the current write position and the consumer
- * uses cli_out.outpos ptr as the current read position. When the
- * producer produces something, it calls cli_write_trigger(). If there
- * is not enough space in the current buffer, the producer allocates
- * the new one. When the consumer processes everything in the buffer
- * queue, it calls cli_written(), tha frees all buffers (except the
- * first one) and schedules cli.event .
  *
+ * A note on transmit buffer management: cli.tx_buf is a head of a list
+ * of TX buffers (struct cli_out). A buffer pointed to by cli.tx_write
+ * is the one currently written to using cli_printf() and cli_alloc_out(),
+ * its wpos field points to the position of the write head in that buffer.
+ * On the other side, cli.tx_pos is the buffer being set to the socket
+ * and its outpos field is the position of the read head.
  */
 
+#define LOCAL_DEBUG 1
+
 #include "nest/bird.h"
 #include "nest/cli.h"
 #include "conf/conf.h"
+#include "lib/coroutine.h"
 #include "lib/string.h"
 
 pool *cli_pool;
 
+/* Hack for scheduled undo notification */
+extern cli *cmd_reconfig_stored_cli;
+
+/*
+ *	Output buffering
+ */
+
 static byte *
 cli_alloc_out(cli *c, int size)
 {
@@ -229,114 +240,54 @@ cli_free_out(cli *c)
   c->async_msg_size = 0;
 }
 
-void
+static void
-cli_written(cli *c)
+cli_write(cli *c)
 {
+  sock *s = c->socket;
+
+  while (c->tx_pos)
+    {
+      struct cli_out *o = c->tx_pos;
+
+      int len = o->wpos - o->outpos;
+      s->tbuf = o->outpos;
+      o->outpos = o->wpos;
+
+      coro_sk_write(s, len);
+
+      c->tx_pos = o->next;
+    }
+
+  /* Everything is written */
+  s->tbuf = NULL;
   cli_free_out(c);
   ev_schedule(c->event);
 }
 
-
-struct cli *this_cli;
-
-struct cli_conf_order {
-  struct conf_order co;
-  struct cli *cli;
-};
-
-static void
-cli_cmd_error(struct conf_order *co, const char *msg, va_list args)
-{
-  struct cli_conf_order *cco = (struct cli_conf_order *) co;
-  cli_vprintf(cco->cli, 9001, msg, args);
-}
-
-static void
-cli_command(struct cli *c)
-{
-  struct conf_state state = {
-    .name = "",
-    .lino = 1
-  };
-
-  struct cli_conf_order o = {
-    .co = {
-      .ctx = NULL,
-      .state = &state,
-      .buf = c->rx_buf,
-      .len = strlen(c->rx_buf),
-      .cf_include = NULL,
-      .cf_outclude = NULL,
-      .cf_error = cli_cmd_error,
-      .lp = c->parser_pool,
-      .pool = c->pool,
-    },
-    .cli = c,
-  };
-
-  if (config->cli_debug > 1)
-    log(L_TRACE "CLI: %s", c->rx_buf);
-  
-  lp_flush(c->parser_pool);
-  this_cli = c;
-  cli_parse(&(o.co));
-}
-
-static void
-cli_event(void *data)
-{
-  cli *c = data;
-  int err;
-
-  while (c->ring_read != c->ring_write &&
-      c->async_msg_size < CLI_MAX_ASYNC_QUEUE)
-    cli_copy_message(c);
-
-  if (c->tx_pos)
-    ;
-  else if (c->cont)
-    c->cont(c);
-  else
-    {
-      err = cli_get_command(c);
-      if (!err)
-	return;
-      if (err < 0)
-	cli_printf(c, 9000, "Command too long");
-      else
-	cli_command(c);
-    }
-
-  cli_write_trigger(c);
-}
-
-cli *
-cli_new(void *priv)
-{
-  pool *p = rp_new(cli_pool, "CLI");
-  cli *c = mb_alloc(p, sizeof(cli));
-
-  bzero(c, sizeof(cli));
-  c->pool = p;
-  c->priv = priv;
-  c->event = ev_new(p);
-  c->event->hook = cli_event;
-  c->event->data = c;
-  c->cont = cli_hello;
-  c->show_pool = lp_new_default(c->pool);
-  c->parser_pool = lp_new_default(c->pool);
-  c->rx_buf = mb_alloc(c->pool, CLI_RX_BUF_SIZE);
-  ev_schedule(c->event);
-  return c;
-}
-
 void
-cli_kick(cli *c)
+cli_write_trigger(cli *c)
 {
-  if (!c->cont && !c->tx_pos)
+  if (c->tx_pos && c->socket->tbuf == NULL)
-    ev_schedule(c->event);
+    cli_write(c);
 }
 
+static void
+cli_err_hook(sock *s, int err)
+{
+  if (config->cli_debug)
+    {
+      if (err)
+	log(L_INFO "CLI connection dropped: %s", strerror(err));
+      else
+	log(L_INFO "CLI connection closed");
+    }
+  cli_free(s->data);
+}
+
+/*
+ *	Echoing of asynchronous messages
+ */
+
 static list cli_log_hooks;
 static int cli_log_inited;
 
@@ -407,12 +358,209 @@ cli_echo(uint class, byte *msg)
     }
 }
 
-/* Hack for scheduled undo notification */
+/*
-extern cli *cmd_reconfig_stored_cli;
+ *	Reading of input
+ */
+
+static int
+cli_getchar(cli *c)
+{
+  sock *s = c->socket;
+
+  if (c->rx_aux == s->rpos)
+    {
+      DBG("CLI: Waiting on read\n");
+      c->rx_aux = s->rpos = s->rbuf;
+      c->state = CLI_STATE_WAIT_RX;
+      int n = coro_sk_read(s);
+      c->state = CLI_STATE_RUN;
+      DBG("CLI: Read returned %d bytes\n", n);
+      ASSERT(n);
+    }
+  return *c->rx_aux++;
+}
+
+static int
+cli_read_line(cli *c)
+{
+  byte *d = c->rx_buf;
+  byte *dend = c->rx_buf + CLI_RX_BUF_SIZE - 2;
+  for (;;)
+    {
+      int ch = cli_getchar(c);
+      if (ch == '\r')
+	;
+      else if (ch == '\n')
+	break;
+      else if (d < dend)
+	*d++ = ch;
+    }
+
+  if (d >= dend)
+    return 0;
+
+  *d = 0;
+  return 1;
+}
+
+/*
+ *	Execution of commands
+ */
+
+struct cli *this_cli;
+
+struct cli_conf_order {
+  struct conf_order co;
+  struct cli *cli;
+};
+
+static void
+cli_cmd_error(struct conf_order *co, const char *msg, va_list args)
+{
+  struct cli_conf_order *cco = (struct cli_conf_order *) co;
+  cli_vprintf(cco->cli, 9001, msg, args);
+}
+
+static void
+cli_command(struct cli *c)
+{
+  struct conf_state state = {
+    .name = "",
+    .lino = 1
+  };
+
+  struct cli_conf_order o = {
+    .co = {
+      .ctx = NULL,
+      .state = &state,
+      .buf = c->rx_buf,
+      .len = strlen(c->rx_buf),
+      .cf_include = NULL,
+      .cf_outclude = NULL,
+      .cf_error = cli_cmd_error,
+      .lp = c->parser_pool,
+      .pool = c->pool,
+    },
+    .cli = c,
+  };
+
+  if (config->cli_debug > 1)
+    log(L_TRACE "CLI: %s", c->rx_buf);
+  
+  lp_flush(c->parser_pool);
+  this_cli = c;
+  cli_parse(&(o.co));
+}
+
+/*
+ *	Session control
+ */
+
+static void
+cli_event(void *data)
+{
+  cli *c = data;
+  DBG("CLI: Event in state %u\n", (int) c->state);
+
+  while (c->ring_read != c->ring_write &&
+      c->async_msg_size < CLI_MAX_ASYNC_QUEUE)
+    cli_copy_message(c);
+
+  cli_write_trigger(c);
+
+  if (c->state == CLI_STATE_YIELD ||
+      c->state == CLI_STATE_WAIT_TX && !c->tx_pos)
+    coro_resume(c->coro);
+}
+
+void
+cli_yield(cli *c)
+{
+  c->state = CLI_STATE_YIELD;
+  DBG("CLI: Yielding\n");
+  ev_schedule(c->event);
+  coro_suspend();
+  c->state = CLI_STATE_RUN;
+  DBG("CLI: Yield resumed\n");
+}
+
+static void
+cli_coroutine(void *_c)
+{
+  cli *c = _c;
+  sock *s = c->socket;
+
+  DBG("CLI: Coroutine started\n");
+  c->rx_aux = s->rbuf;
+
+  for (;;)
+    {
+      while (c->tx_pos)
+	{
+	  DBG("CLI: Sleeping on write\n");
+	  c->state = CLI_STATE_WAIT_TX;
+	  coro_suspend();
+	  c->state = CLI_STATE_RUN;
+	  DBG("CLI: Woke up on write\n");
+	}
+
+      if (c->cont)
+	{
+	  c->cont(c);
+	  cli_write_trigger(c);
+	  cli_yield(c);
+	  continue;
+	}
+
+      if (!cli_read_line(c))
+	cli_printf(c, 9000, "Command too long");
+      else
+	cli_command(c);
+      cli_write_trigger(c);
+    }
+}
+
+cli *
+cli_new(sock *s)
+{
+  pool *p = rp_new(cli_pool, "CLI session");
+  cli *c = mb_alloc(p, sizeof(cli));
+  DBG("CLI: Created new session\n");
+
+  bzero(c, sizeof(cli));
+  c->pool = p;
+  c->socket = s;
+  c->event = ev_new(p);
+  c->event->hook = cli_event;
+  c->event->data = c;
+  c->cont = cli_hello;
+  c->parser_pool = lp_new_default(c->pool);
+  c->show_pool = lp_new_default(c->pool);
+  c->rx_buf = mb_alloc(c->pool, CLI_RX_BUF_SIZE);
+
+  s->pool = c->pool;		/* We need to have all the socket buffers allocated in the cli pool */
+  rmove(s, c->pool);
+  s->err_hook = cli_err_hook;
+  s->data = c;
+
+  return c;
+}
+
+void
+cli_run(cli *c)
+{
+  DBG("CLI: Running\n");
+  c->state = CLI_STATE_RUN;
+  c->rx_pos = c->rx_buf;
+  c->rx_aux = NULL;
+  c->coro = coro_new(c->pool, cli_coroutine, c);
+  coro_resume(c->coro);
+}
 
 void
 cli_free(cli *c)
 {
+  DBG("CLI: Destroying session\n");
   cli_set_log_echo(c, 0, 0);
   if (c->cleanup)
     c->cleanup(c);

nest/cli.h

@@ -1,7 +1,7 @@
 /*
  *	BIRD Internet Routing Daemon -- Command-Line Interface
  *
- *	(c) 1999--2000 Martin Mares <mj@ucw.cz>
+ *	(c) 1999--2017 Martin Mares <mj@ucw.cz>
  *
  *	Can be freely distributed and used under the terms of the GNU GPL.
  */
@@ -10,7 +10,9 @@
 #define _BIRD_CLI_H_
 
 #include "lib/resource.h"
+#include "lib/coroutine.h"
 #include "lib/event.h"
+#include "lib/socket.h"
 
 #define CLI_RX_BUF_SIZE 4096
 #define CLI_TX_BUF_SIZE 4096
@@ -25,20 +27,38 @@ struct cli_out {
   byte buf[0];
 };
 
+enum cli_state {
+  CLI_STATE_INIT,
+  CLI_STATE_RUN,
+  CLI_STATE_WAIT_RX,
+  CLI_STATE_WAIT_TX,
+  CLI_STATE_YIELD,
+};
+
 typedef struct cli {
   node n;				/* Node in list of all log hooks */
   pool *pool;
-  void *priv;				/* Private to sysdep layer */
+  coroutine *coro;
-  byte *rx_buf, *rx_pos, *rx_aux;	/* sysdep */
+  enum cli_state state;
+  int restricted;			/* CLI is restricted to read-only commands */
+
+  /* I/O */
+  sock *socket;
+  byte *rx_buf, *rx_pos, *rx_aux;
   struct cli_out *tx_buf, *tx_pos, *tx_write;
   event *event;
+
+  /* Continuation mechanism */
   void (*cont)(struct cli *c);
   void (*cleanup)(struct cli *c);
   void *rover;				/* Private to continuation routine */
   int last_reply;
-  int restricted;			/* CLI is restricted to read-only commands */
+
+  /* Pools */
   struct linpool *parser_pool;		/* Pool used during parsing */
   struct linpool *show_pool;		/* Pool used during route show */
+
+  /* Asynchronous messages */
   byte *ring_buf;			/* Ring buffer for asynchronous messages */
   byte *ring_end, *ring_read, *ring_write;	/* Pointers to the ring buffer */
   uint ring_overflow;			/* Counter of ring overflows */
@@ -56,15 +76,16 @@ extern struct cli *this_cli;		/* Used during parsing */
 
 void cli_printf(cli *, int, char *, ...);
 #define cli_msg(x...) cli_printf(this_cli, x)
+void cli_write_trigger(cli *c);
 void cli_set_log_echo(cli *, uint mask, uint size);
+void cli_yield(cli *c);
 
 /* Functions provided to sysdep layer */
 
-cli *cli_new(void *);
 void cli_init(void);
+cli *cli_new(sock *s);
+void cli_run(cli *);
 void cli_free(cli *);
-void cli_kick(cli *);
-void cli_written(cli *);
 void cli_echo(uint class, byte *msg);
 
 static inline int cli_access_restricted(void)
@@ -75,9 +96,4 @@ static inline int cli_access_restricted(void)
     return 0;
 }
 
-/* Functions provided by sysdep layer */
-
-void cli_write_trigger(cli *);
-int cli_get_command(cli *);
-
 #endif

sysdep/unix/Makefile

@@ -1,4 +1,4 @@
-src := io.c krt.c log.c main.c random.c conf.c
+src := conf.c coroutine.c io.c krt.c log.c main.c random.c
 obj := $(src-o-files)
 $(all-daemon)
 $(cf-local)

sysdep/unix/coroutine.c

@@ -0,0 +1,332 @@
+/*
+ *	BIRD Coroutines
+ *
+ *	(c) 2017 Martin Mares <mj@ucw.cz>
+ *
+ *	Can be freely distributed and used under the terms of the GNU GPL.
+ */
+
+#ifndef _GNU_SOURCE
+#define _GNU_SOURCE
+#endif
+
+#include <stdlib.h>
+
+#include "nest/bird.h"
+#include "lib/coroutine.h"
+#include "lib/resource.h"
+#include "lib/socket.h"
+#include "lib/timer.h"
+#include "sysdep/unix/unix.h"
+
+#define CORO_STACK_SIZE 65536
+
+#if ! USE_PTHREADS
+
+/*
+ *	Implementation of coroutines based on <ucontext.h>
+ */
+
+#include <ucontext.h>
+
+struct coroutine {
+  resource r;
+  ucontext_t ctx;
+  void *stack;
+  void (*entry_point)(void *arg);
+  void *arg;
+  event *ev;
+};
+
+static ucontext_t *main_context;
+static coroutine *coro_current;		// NULL for main context
+
+static void
+coro_free(resource *r)
+{
+  coroutine *c = (coroutine *) r;
+  xfree(c->stack);
+}
+
+static void
+coro_dump(resource *r UNUSED)
+{
+  debug("\n");
+}
+
+static size_t
+coro_memsize(resource *r)
+{
+  coroutine *c = (coroutine *) r;
+  return sizeof(*c) + CORO_STACK_SIZE + 2*ALLOC_OVERHEAD;
+}
+
+static struct resclass coro_class = {
+  .name = "Coroutine",
+  .size = sizeof(struct coroutine),
+  .free = coro_free,
+  .dump = coro_dump,
+  .memsize = coro_memsize,
+};
+
+static void
+coro_do_start(void)
+{
+  ASSERT(coro_current);
+  coro_current->entry_point(coro_current->arg);
+  bug("Coroutine returned unexpectedly");
+}
+
+struct coroutine *
+coro_new(pool *p, void (*entry_point)(void *), void *arg)
+{
+  if (!main_context)
+    {
+      main_context = xmalloc(sizeof(*main_context));
+      if (getcontext(main_context) < 0)
+	bug("getcontext() failed");
+    }
+
+  coroutine *c = ralloc(p, &coro_class);
+  c->entry_point = entry_point;
+  c->arg = arg;
+  if (getcontext(&c->ctx) < 0)
+    bug("getcontext() failed");
+  c->stack = xmalloc(CORO_STACK_SIZE);
+  c->ctx.uc_stack.ss_sp = c->stack;
+  c->ctx.uc_stack.ss_size = CORO_STACK_SIZE;
+
+  makecontext(&c->ctx, coro_do_start, 0);
+
+  return c;
+}
+
+void
+coro_done(event *e)
+{
+  ASSERT(coro_inited);
+  ASSERT(coro_current);
+  if (e)
+    ev_schedule(e);
+  c->ev = e;
+  coroutine *c = coro_current;
+  coro_suspend();
+  bug("Coroutine suspend after coro_done() should never return");
+}
+
+void
+coro_suspend(void)
+{
+  ASSERT(coro_current);
+  ASSERT(main_context);
+  coroutine *c = coro_current;
+  coro_current = NULL;
+  swapcontext(&c->ctx, main_context);
+  ASSERT(coro_current == c);
+}
+
+void
+coro_resume(coroutine *c)
+{
+  ASSERT(!coro_current);
+  coro_current = c;
+  swapcontext(main_context, &c->ctx);
+  ASSERT(!coro_current);
+}
+
+#else
+
+/*
+ *	Implementation of coroutines based on POSIX threads
+ */
+
+#include <pthread.h>
+#include <semaphore.h>
+
+#define CORO_STOP 1	/* The coroutine should stop at first coro_suspend(). */
+#define CORO_DONE 2	/* The coroutine has already stopped. */
+
+struct coroutine {
+  resource r;
+  pthread_t thread;
+  void (*entry_point)(void *arg);
+  void *arg;
+  event *ev;
+  sem_t sem;
+  uint flags;
+};
+
+static coroutine *coro_current;		// NULL for main context
+static int coro_inited;
+static sem_t coro_main_sem;
+static pthread_attr_t coro_thread_attrs;
+
+static void
+coro_free(resource *r)
+{
+  coroutine *c = (coroutine *) r;
+  ASSERT(coro_current != c);
+
+  if (!(c->flags & CORO_DONE))
+    {
+      c->flags |= CORO_STOP;
+      coro_resume(c);
+    }
+
+  ASSERT(c->flags & CORO_DONE);
+  pthread_join(c->thread, NULL);
+}
+
+static void
+coro_dump(resource *r UNUSED)
+{
+  debug("\n");
+}
+
+static size_t
+coro_memsize(resource *r)
+{
+  coroutine *c = (coroutine *) r;
+  return sizeof(*c) + CORO_STACK_SIZE + 2*ALLOC_OVERHEAD;
+}
+
+static struct resclass coro_class = {
+  .name = "Coroutine",
+  .size = sizeof(struct coroutine),
+  .free = coro_free,
+  .dump = coro_dump,
+  .memsize = coro_memsize,
+};
+
+extern pthread_key_t current_time_key;
+
+static void *
+coro_do_start(void *c_)
+{
+  coroutine *c = c_;
+  pthread_setspecific(current_time_key, &main_timeloop);
+  while (sem_wait(&c->sem) < 0)
+    ;
+  coro_current = c;
+  c->entry_point(c->arg);
+  bug("Coroutine returned unexpectedly");
+}
+
+struct coroutine *
+coro_new(pool *p, void (*entry_point)(void *), void *arg)
+{
+  if (!coro_inited)
+    {
+      if (sem_init(&coro_main_sem, 0, 0) < 0)
+	bug("sem_init() failed");
+      if (pthread_attr_init(&coro_thread_attrs))
+	bug("pthread_attr_init() failed");
+      if (pthread_attr_setstacksize(&coro_thread_attrs, CORO_STACK_SIZE))
+	bug("pthread_attr_setstacksize() failed");
+      coro_inited = 1;
+    }
+
+  coroutine *c = ralloc(p, &coro_class);
+  c->entry_point = entry_point;
+  c->arg = arg;
+  if (sem_init(&c->sem, 0, 0) < 0)
+    bug("sem_init() failed");
+  if (pthread_create(&c->thread, &coro_thread_attrs, coro_do_start, c))
+    bug("pthread_create() failed");
+
+  return c;
+}
+
+static inline void
+coro_check_stop(void)
+{
+  ASSERT(coro_inited);
+  ASSERT(coro_current);
+  coroutine *c = coro_current;
+  if (c->flags & CORO_STOP)
+    coro_done(NULL);
+}
+
+void
+coro_done(event *e)
+{
+  ASSERT(coro_inited);
+  ASSERT(coro_current);
+  coroutine *c = coro_current;
+  c->flags |= CORO_DONE;
+  c->ev = e;
+  if (e)
+    ev_schedule(e);
+  sem_post(&coro_main_sem);
+  pthread_exit(NULL);
+  bug("pthread_exit should never return");
+}
+
+void
+coro_suspend(void)
+{
+  ASSERT(coro_inited);
+  ASSERT(coro_current);
+  coroutine *c = coro_current;
+  coro_check_stop();
+  sem_post(&coro_main_sem);
+  while (sem_wait(&c->sem) < 0)
+    ;
+  coro_current = c;
+  coro_check_stop();
+}
+
+void
+coro_resume(coroutine *c)
+{
+  ASSERT(coro_inited);
+  ASSERT(!coro_current);
+  sem_post(&c->sem);
+  while (sem_wait(&coro_main_sem) < 0)
+    ;
+  coro_current = NULL;
+}
+
+#endif
+
+/* Coroutine-based I/O */
+
+static int
+coro_sk_rx_hook(sock *sk, uint size UNUSED)
+{
+  ASSERT(sk->rx_coroutine);
+  ASSERT(!coro_current);
+  coro_resume(sk->rx_coroutine);
+  return 0;
+}
+
+static void
+coro_sk_tx_hook(sock *sk)
+{
+  ASSERT(sk->tx_coroutine);
+  ASSERT(!coro_current);
+  coro_resume(sk->tx_coroutine);
+}
+
+int
+coro_sk_read(sock *s)
+{
+  ASSERT(coro_current);
+  s->rx_coroutine = coro_current;
+  s->rx_hook = coro_sk_rx_hook;
+  coro_suspend();
+  s->rx_hook = NULL;
+  return s->rpos - s->rbuf;
+}
+
+void
+coro_sk_write(sock *s, unsigned len)
+{
+  ASSERT(coro_current);
+  s->tx_coroutine = coro_current;
+  s->tx_hook = coro_sk_tx_hook;
+  s->ttx = s->tbuf;
+  s->tpos = s->tbuf + len;
+  coro_suspend();
+  s->tx_hook = NULL;
+}

sysdep/unix/main.c

@@ -23,6 +23,7 @@
 #include <libgen.h>
 
 #include "nest/bird.h"
+#include "lib/coroutine.h"
 #include "lib/lists.h"
 #include "lib/resource.h"
 #include "lib/socket.h"
@@ -98,95 +99,6 @@ drop_gid(gid_t gid)
 static sock *cli_sk;
 static char *path_control_socket = PATH_CONTROL_SOCKET;
 
-
-static void
-cli_write(cli *c)
-{
-  sock *s = c->priv;
-
-  while (c->tx_pos)
-    {
-      struct cli_out *o = c->tx_pos;
-
-      int len = o->wpos - o->outpos;
-      s->tbuf = o->outpos;
-      o->outpos = o->wpos;
-
-      if (sk_send(s, len) <= 0)
-	return;
-
-      c->tx_pos = o->next;
-    }
-
-  /* Everything is written */
-  s->tbuf = NULL;
-  cli_written(c);
-}
-
-void
-cli_write_trigger(cli *c)
-{
-  sock *s = c->priv;
-
-  if (s->tbuf == NULL)
-    cli_write(c);
-}
-
-static void
-cli_tx(sock *s)
-{
-  cli_write(s->data);
-}
-
-int
-cli_get_command(cli *c)
-{
-  sock *s = c->priv;
-  byte *t = c->rx_aux ? : s->rbuf;
-  byte *tend = s->rpos;
-  byte *d = c->rx_pos;
-  byte *dend = c->rx_buf + CLI_RX_BUF_SIZE - 2;
-
-  while (t < tend)
-    {
-      if (*t == '\r')
-	t++;
-      else if (*t == '\n')
-	{
-	  t++;
-	  c->rx_pos = c->rx_buf;
-	  c->rx_aux = t;
-	  *d = 0;
-	  return (d < dend) ? 1 : -1;
-	}
-      else if (d < dend)
-	*d++ = *t++;
-    }
-  c->rx_aux = s->rpos = s->rbuf;
-  c->rx_pos = d;
-  return 0;
-}
-
-static int
-cli_rx(sock *s, uint size UNUSED)
-{
-  cli_kick(s->data);
-  return 0;
-}
-
-static void
-cli_err(sock *s, int err)
-{
-  if (config->cli_debug)
-    {
-      if (err)
-	log(L_INFO "CLI connection dropped: %s", strerror(err));
-      else
-	log(L_INFO "CLI connection closed");
-    }
-  cli_free(s->data);
-}
-
 static int
 cli_connect(sock *s, uint size UNUSED)
 {
@@ -194,15 +106,9 @@ cli_connect(sock *s, uint size UNUSED)
 
   if (config->cli_debug)
     log(L_INFO "CLI connect");
-  s->rx_hook = cli_rx;
+  c = cli_new(s);
-  s->tx_hook = cli_tx;
-  s->err_hook = cli_err;
-  s->data = c = cli_new(s);
-  s->pool = c->pool;		/* We need to have all the socket buffers allocated in the cli pool */
   s->fast_rx = 1;
-  c->rx_pos = c->rx_buf;
+  cli_run(c);
-  c->rx_aux = NULL;
-  rmove(s, c->pool);
   return 1;
 }
 
@@ -370,7 +276,7 @@ signal_init(void)
  *	Parsing of command-line arguments
  */
 
-static char *opt_list = "c:dD:ps:P:u:g:flRh";
+static char *opt_list = "bc:dD:ps:P:u:g:flRh";
 static int parse_and_exit;
 char *bird_name;
 static char *use_user;
@@ -391,6 +297,7 @@ display_help(void)
   fprintf(stderr,
     "\n"
     "Options: \n"
+    "  -b                   Run bird in background\n"
     "  -c <config-file>     Use given configuration file instead\n"
     "                       of prefix/etc/bird.conf\n"
     "  -d                   Enable debug messages and run bird in foreground\n"
@@ -495,16 +402,21 @@ parse_args(int argc, char **argv)
   while ((c = getopt(argc, argv, opt_list)) >= 0)
     switch (c)
       {
+      case 'b':
+	run_in_foreground = 0;
+	break;
       case 'c':
 	config_name = optarg;
 	config_changed = 1;
 	break;
       case 'd':
 	debug_flag |= 1;
+	run_in_foreground = 1;
 	break;
       case 'D':
 	log_init_debug(optarg);
 	debug_flag |= 2;
+	run_in_foreground = 1;
 	break;
       case 'p':
 	parse_and_exit = 1;
@@ -603,7 +515,7 @@ main(int argc, char **argv)
   if (parse_and_exit)
     exit(0);
 
-  if (!(debug_flag||run_in_foreground))
+  if (!run_in_foreground)
     {
       pid_t pid = fork();
       if (pid < 0)