Adding a generic cork mechanism for events

6 files changed, 102 insertions, 8 deletions

diff --git a/lib/event.c b/lib/event.c
index 6c5c8b14..5031f314 100644
--- a/lib/event.c
+++ b/lib/event.c
@@ -114,17 +114,42 @@ ev_send(event_list *l, event *e)
 
   e->list = l;
 
-  LOCK_DOMAIN(event, l->lock);
-  if (enlisted(&e->n))
+  struct event_cork *ec = e->cork;
+
+  uint ping = 0;
+
+  if (ec)
   {
+    LOCK_DOMAIN(cork, ec->lock);
+    LOCK_DOMAIN(event, l->lock);
+
+    if (!enlisted(&e->n))
+      if (ec->count)
+	add_tail(&ec->events, &e->n);
+      else
+      {
+	add_tail(&l->events, &e->n);
+	ping = 1;
+      }
+
     UNLOCK_DOMAIN(event, l->lock);
-    return;
+    UNLOCK_DOMAIN(cork, ec->lock);
   }
+  else
+  {
+    LOCK_DOMAIN(event, l->lock);
 
-  add_tail(&l->events, &e->n);
-  UNLOCK_DOMAIN(event, l->lock);
+    if (!enlisted(&e->n))
+    {
+      add_tail(&l->events, &e->n);
+      ping = 1;
+    }
 
-  birdloop_ping(l->loop);
+    UNLOCK_DOMAIN(event, l->lock);
+  }
+
+  if (ping)
+    birdloop_ping(l->loop);
 }
 
 void io_log_event(void *hook, void *data);
@@ -224,3 +249,40 @@ ev_run_list_limited(event_list *l, uint limit)
 
   return repeat;
 }
+
+void ev_cork(struct event_cork *ec)
+{
+  LOCK_DOMAIN(cork, ec->lock);
+  ec->count++;
+  UNLOCK_DOMAIN(cork, ec->lock);
+}
+
+void ev_uncork(struct event_cork *ec)
+{
+  LOCK_DOMAIN(cork, ec->lock);
+
+  if (--ec->count)
+  {
+    UNLOCK_DOMAIN(cork, ec->lock);
+    return;
+  }
+
+  node *n;
+  WALK_LIST_FIRST(n, ec->events)
+    {
+      event *e = SKIP_BACK(event, n, n);
+      event_list *el = e->list;
+
+      rem_node(&e->n);
+
+      LOCK_DOMAIN(event, el->lock);
+      add_tail(&el->events, &e->n);
+      UNLOCK_DOMAIN(event, el->lock);
+
+      birdloop_ping(el->loop);
+    }
+
+  UNLOCK_DOMAIN(cork, ec->lock);
+
+  birdloop_ping(&main_birdloop);
+}
diff --git a/lib/event.h b/lib/event.h
index 6c358f84..cd85bf78 100644
--- a/lib/event.h
+++ b/lib/event.h
@@ -15,6 +15,7 @@
 #include <stdatomic.h>
 
 DEFINE_DOMAIN(event);
+DEFINE_DOMAIN(cork);
 
 typedef struct event {
   resource r;
@@ -22,6 +23,8 @@ typedef struct event {
   void *data;
   node n;				/* Internal link */
   struct event_list *list;		/* List where this event is put in */
+  struct event_cork *cork;		/* Event execution limiter */
+  node cork_node;
 } event;
 
 typedef struct event_list {
@@ -31,6 +34,12 @@ typedef struct event_list {
   DOMAIN(event) lock;
 } event_list;
 
+struct event_cork {
+  DOMAIN(cork) lock;
+  u32 count;
+  list events;
+};
+
 extern event_list global_event_list;
 extern event_list global_work_list;
 
@@ -44,6 +53,13 @@ static inline void ev_init_list(event_list *el, struct birdloop *loop, const cha
   el->lock = DOMAIN_NEW(event, name);
 }
 
+static inline void ev_init_cork(struct event_cork *ec, const char *name)
+{
+  init_list(&ec->events);
+  ec->lock = DOMAIN_NEW(cork, name);
+  ec->count = 0;
+};
+
 void ev_send(event_list *, event *);
 #define ev_send_loop(l, e) ev_send(birdloop_event_list((l)), (e))
 
@@ -56,6 +72,20 @@ int ev_run_list_limited(event_list *, uint);
 
 #define LEGACY_EVENT_LIST(l)  (((l) == &global_event_list) || ((l) == &global_work_list))
 
+void ev_cork(struct event_cork *);
+void ev_uncork(struct event_cork *);
+
+static inline u32 ev_corked(struct event_cork *ec)
+{
+  if (!ec)
+    return 0;
+
+  LOCK_DOMAIN(cork, ec->lock);
+  u32 out = ec->count;
+  UNLOCK_DOMAIN(cork, ec->lock);
+  return out;
+}
+
 _Bool birdloop_inside(struct birdloop *loop);
 
 static inline int
diff --git a/lib/locking.h b/lib/locking.h
index ab5c06af..0cbdead8 100644
--- a/lib/locking.h
+++ b/lib/locking.h
@@ -16,6 +16,7 @@ struct lock_order {
   struct domain_generic *the_bird;
   struct domain_generic *proto;
   struct domain_generic *rtable;
+  struct domain_generic *cork;
   struct domain_generic *event;
 };
 
diff --git a/lib/socket.h b/lib/socket.h
index 5bdab7f3..ff07660f 100644
--- a/lib/socket.h
+++ b/lib/socket.h
@@ -57,6 +57,7 @@ typedef struct birdsock {
   uint fast_rx;				/* RX has higher priority in event loop */
   uint rbsize;
   int (*rx_hook)(struct birdsock *, uint size); /* NULL=receiving turned off, returns 1 to clear rx buffer */
+  struct event_cork *cork;		/* Cork to temporarily stop receiving data */
 
   byte *tbuf, *tpos;			/* NULL=allocate automatically */
   byte *ttx;				/* Internal */
diff --git a/sysdep/unix/io-loop.c b/sysdep/unix/io-loop.c
index 275d38a0..c7cf4ad2 100644
--- a/sysdep/unix/io-loop.c
+++ b/sysdep/unix/io-loop.c
@@ -205,7 +205,7 @@ sk_stop(sock *s)
 }
 
 static inline uint sk_want_events(sock *s)
-{ return (s->rx_hook ? POLLIN : 0) | ((s->ttx != s->tpos) ? POLLOUT : 0); }
+{ return ((s->rx_hook && !ev_corked(s->cork)) ? POLLIN : 0) | ((s->ttx != s->tpos) ? POLLOUT : 0); }
 
 /*
 FIXME: this should be called from sock code
diff --git a/sysdep/unix/io.c b/sysdep/unix/io.c
index c91f2856..dd385c80 100644
--- a/sysdep/unix/io.c
+++ b/sysdep/unix/io.c
@@ -2234,7 +2234,7 @@ io_loop(void)
 	{
 	  pfd[nfds] = (struct pollfd) { .fd = -1 }; /* everything other set to 0 by this */
 	  s = SKIP_BACK(sock, n, n);
-	  if (s->rx_hook)
+	  if (s->rx_hook && !ev_corked(s->cork))
 	    {
 	      pfd[nfds].fd = s->fd;
 	      pfd[nfds].events |= POLLIN;