1 files changed, 131 insertions, 81 deletions

diff --git a/lib/event.c b/lib/event.c
index 33dc00b0..07d7dc53 100644
--- a/lib/event.c
+++ b/lib/event.c
@@ -19,20 +19,95 @@
  * events in them and explicitly ask to run them.
  */
 
+#undef LOCAL_DEBUG
+
 #include "nest/bird.h"
 #include "lib/event.h"
+#include "lib/io-loop.h"
 
 event_list global_event_list;
 event_list global_work_list;
 
-inline void
-ev_postpone(event *e)
+STATIC_ASSERT(OFFSETOF(event_list, _sentinel.next) >= OFFSETOF(event_list, _end[0]));
+
+void
+ev_init_list(event_list *el, struct birdloop *loop, const char *name)
+{
+  el->name = name;
+  el->loop = loop;
+
+  atomic_store_explicit(&el->receiver, &el->_sentinel, memory_order_relaxed);
+  atomic_store_explicit(&el->_executor, &el->_sentinel, memory_order_relaxed);
+  atomic_store_explicit(&el->_sentinel.next, NULL, memory_order_relaxed);
+}
+
+/*
+ * The event list should work as a message passing point. Sending a message
+ * must be a fairly fast process with no locks and low waiting times. OTOH,
+ * processing messages always involves running the assigned code and the
+ * receiver is always a single one thread with no concurrency at all. There is
+ * also a postponing requirement to synchronously remove an event from a queue,
+ * yet we allow this only when the caller has its receiver event loop locked.
+ * It still means that the event may get postponed from other event in the same
+ * list, therefore we have to be careful.
+ */
+
+static inline int
+ev_remove_from(event *e, event * _Atomic * head)
 {
-  if (ev_active(e))
+  /* The head pointer stores where cur is pointed to from */
+  event * _Atomic *prev = head;
+
+  /* The current event in queue to check */
+  event *cur = atomic_load_explicit(prev, memory_order_acquire);
+
+  /* Pre-loaded next pointer; if NULL, this is sentinel */
+  event *next = atomic_load_explicit(&cur->next, memory_order_acquire);
+
+  while (next)
+  {
+    if (e == cur)
     {
-      rem_node(&e->n);
-      e->n.next = NULL;
+      /* Check whether we have collided with somebody else
+       * adding an item to the queue. */
+      if (!atomic_compare_exchange_strong_explicit(
+	    prev, &cur, next,
+	    memory_order_acq_rel, memory_order_acquire))
+      {
+	/* This may happen only on list head */
+	ASSERT_DIE(prev == head);
+
+	/* Restart. The collision should never happen again. */
+	return ev_remove_from(e, head);
+      }
+
+      /* Successfully removed from the list; inactivate this event. */
+      atomic_store_explicit(&cur->next, NULL, memory_order_release);
+      return 1;
     }
+
+    /* Go to the next event. */
+    prev = &cur->next;
+    cur = next;
+    next = atomic_load_explicit(&cur->next, memory_order_acquire);
+  }
+
+  return 0;
+}
+
+inline void
+ev_postpone(event *e)
+{
+  /* Find the list to remove the event from */
+  event_list *sl = ev_get_list(e);
+  if (!sl)
+    return;
+
+  /* Postponing allowed only from the target loop */
+  ASSERT_DIE(birdloop_inside(sl->loop));
+
+  /* Remove from one of these lists. */
+  ASSERT(ev_remove_from(e, &sl->_executor) || ev_remove_from(e, &sl->receiver));
 }
 
 static void
@@ -43,7 +118,7 @@ ev_dump(resource *r)
   debug("(code %p, data %p, %s)\n",
 	e->hook,
 	e->data,
-	e->n.next ? "scheduled" : "inactive");
+	atomic_load_explicit(&e->next, memory_order_relaxed) ? "scheduled" : "inactive");
 }
 
 static struct resclass ev_class = {
@@ -95,40 +170,21 @@ ev_run(event *e)
  * list @l which can be run by calling ev_run_list().
  */
 inline void
-ev_enqueue(event_list *l, event *e)
-{
-  ev_postpone(e);
-  add_tail(l, &e->n);
-}
-
-/**
- * ev_schedule - schedule an event
- * @e: an event
- *
- * This function schedules an event by enqueueing it to a system-wide
- * event list which is run by the platform dependent code whenever
- * appropriate.
- */
-void
-ev_schedule(event *e)
-{
-  ev_enqueue(&global_event_list, e);
-}
-
-/**
- * ev_schedule_work - schedule a work-event.
- * @e: an event
- *
- * This function schedules an event by enqueueing it to a system-wide work-event
- * list which is run by the platform dependent code whenever appropriate. This
- * is designated for work-events instead of regular events. They are executed
- * less often in order to not clog I/O loop.
- */
-void
-ev_schedule_work(event *e)
+ev_send(event_list *l, event *e)
 {
-  if (!ev_active(e))
-    add_tail(&global_work_list, &e->n);
+  event_list *sl = ev_get_list(e);
+  if (sl == l)
+    return;
+  if (sl)
+    bug("Queuing an already queued event to another queue is not supported.");
+
+  event *next = atomic_load_explicit(&l->receiver, memory_order_acquire);
+  do atomic_store_explicit(&e->next, next, memory_order_relaxed);
+  while (!atomic_compare_exchange_strong_explicit(
+	&l->receiver, &next, e,
+	memory_order_acq_rel, memory_order_acquire));
+
+  birdloop_ping(l->loop);
 }
 
 void io_log_event(void *hook, void *data);
@@ -140,62 +196,56 @@ void io_log_event(void *hook, void *data);
  * This function calls ev_run() for all events enqueued in the list @l.
  */
 int
-ev_run_list(event_list *l)
+ev_run_list_limited(event_list *l, uint limit)
 {
-  node *n;
-  list tmp_list;
+  event * _Atomic *ep = &l->_executor;
 
-  init_list(&tmp_list);
-  add_tail_list(&tmp_list, l);
-  init_list(l);
-  WALK_LIST_FIRST(n, tmp_list)
-    {
-      event *e = SKIP_BACK(event, n, n);
-
-      /* This is ugly hack, we want to log just events executed from the main I/O loop */
-      if ((l == &global_event_list) || (l == &global_work_list))
-	io_log_event(e->hook, e->data);
+  /* No pending events, refill the queue. */
+  if (atomic_load_explicit(ep, memory_order_relaxed) == &l->_sentinel)
+  {
+    /* Move the current event list aside and create a new one. */
+    event *received = atomic_exchange_explicit(
+	&l->receiver, &l->_sentinel, memory_order_acq_rel);
 
-      ev_run(e);
-      tmp_flush();
-    }
+    /* No event to run. */
+    if (received == &l->_sentinel)
+      return 0;
 
-  return !EMPTY_LIST(*l);
-}
+    /* Setup the executor queue */
+    event *head = &l->_sentinel;
 
-int
-ev_run_list_limited(event_list *l, uint limit)
-{
-  node *n;
-  list tmp_list;
+    /* Flip the order of the events by relinking them one by one (push-pop) */
+    while (received != &l->_sentinel)
+    {
+      event *cur = received;
+      received = atomic_exchange_explicit(&cur->next, head, memory_order_relaxed);
+      head = cur;
+    }
 
-  init_list(&tmp_list);
-  add_tail_list(&tmp_list, l);
-  init_list(l);
+    /* Store the executor queue to its designated place */
+    atomic_store_explicit(ep, head, memory_order_relaxed);
+  }
 
-  WALK_LIST_FIRST(n, tmp_list)
+  /* Run the events in order. */
+  event *e;
+  while ((e = atomic_load_explicit(ep, memory_order_relaxed)) != &l->_sentinel)
     {
-      event *e = SKIP_BACK(event, n, n);
-
-      if (!limit)
-	break;
+      /* Check limit */
+      if (!--limit)
+	return 1;
 
       /* This is ugly hack, we want to log just events executed from the main I/O loop */
       if ((l == &global_event_list) || (l == &global_work_list))
 	io_log_event(e->hook, e->data);
 
-      ev_run(e);
+      /* Inactivate the event */
+      atomic_store_explicit(ep, atomic_load_explicit(&e->next, memory_order_relaxed), memory_order_relaxed);
+      atomic_store_explicit(&e->next, NULL, memory_order_relaxed);
+
+      /* Run the event */
+      e->hook(e->data);
       tmp_flush();
-      limit--;
     }
 
-  if (!EMPTY_LIST(tmp_list))
-  {
-    /* Attach new items after the unprocessed old items */
-    add_tail_list(&tmp_list, l);
-    init_list(l);
-    add_tail_list(l, &tmp_list);
-  }
-
-  return !EMPTY_LIST(*l);
+  return atomic_load_explicit(&l->receiver, memory_order_relaxed) != &l->_sentinel;
 }