Object locks use events

Instead of calling custom hooks from object locks, we use standard event
sending mechanism to inform protocols about object lock changes. This is
a backport from version 3 where these events are passed across threads.

This implementation of object locks doesn't use mutexes to lock the
whole data structure. In version 3, this data structure may get accessed
from multiple threads and must be protected by mutex.

2 files changed, 27 insertions, 36 deletions

diff --git a/nest/locks.c b/nest/locks.c
index 812a6534..e378fb1f 100644
--- a/nest/locks.c
+++ b/nest/locks.c
@@ -37,7 +37,6 @@
 #include "nest/iface.h"
 
 static list olock_list;
-static event *olock_event;
 
 static inline int
 olock_same(struct object_lock *x, struct object_lock *y)
@@ -54,7 +53,8 @@ olock_same(struct object_lock *x, struct object_lock *y)
 static void
 olock_free(resource *r)
 {
-  struct object_lock *q, *l = (struct object_lock *) r;
+  /* Called externally from rfree() */
+  struct object_lock *l = SKIP_BACK(struct object_lock, r, r);
   node *n;
 
   DBG("olock: Freeing %p\n", l);
@@ -63,21 +63,35 @@ olock_free(resource *r)
     case OLOCK_STATE_FREE:
       break;
     case OLOCK_STATE_LOCKED:
-    case OLOCK_STATE_EVENT:
+      /* Remove myself from the olock_list */
       rem_node(&l->n);
+
+      /* Maybe the notification is still pending. */
+      ev_postpone(&l->event);
+
+      /* Get new lock candidate */
       n = HEAD(l->waiters);
-      if (n->next)
+      if (NODE_VALID(n))
 	{
-	  DBG("olock: -> %p becomes locked\n", n);
-	  q = SKIP_BACK(struct object_lock, n, n);
+	  struct object_lock *q = SKIP_BACK(struct object_lock, n, n);
+
+	  /* Remove this candidate from waiters list */
 	  rem_node(n);
+
+	  /* Move waiter lists */
+	  DBG("olock: -> %p becomes locked\n", n);
 	  add_tail_list(&q->waiters, &l->waiters);
-	  q->state = OLOCK_STATE_EVENT;
+
+	  /* Add the new olock to olock_list */
 	  add_head(&olock_list, n);
-	  ev_schedule(olock_event);
+
+	  /* Inform */
+	  q->state = OLOCK_STATE_LOCKED;
+	  ev_schedule(&q->event);
 	}
       break;
     case OLOCK_STATE_WAITING:
+      /* Remove from the waiters list */
       rem_node(&l->n);
       break;
     default:
@@ -148,36 +162,16 @@ olock_acquire(struct object_lock *l)
 	  l->state = OLOCK_STATE_WAITING;
 	  add_tail(&q->waiters, &l->n);
 	  DBG("olock: %p waits\n", l);
+
 	  return;
 	}
     }
+
   DBG("olock: %p acquired immediately\n", l);
-  l->state = OLOCK_STATE_EVENT;
   add_head(&olock_list, &l->n);
-  ev_schedule(olock_event);
-}
 
-static void
-olock_run_event(void *unused UNUSED)
-{
-  node *n;
-  struct object_lock *q;
-
-  DBG("olock: Processing events\n");
-  for(;;)
-    {
-      n = HEAD(olock_list);
-      if (!n->next)
-	break;
-      q = SKIP_BACK(struct object_lock, n, n);
-      if (q->state != OLOCK_STATE_EVENT)
-	break;
-      DBG("olock: %p locked\n", q);
-      q->state = OLOCK_STATE_LOCKED;
-      rem_node(&q->n);
-      add_tail(&olock_list, &q->n);
-      q->hook(q);
-    }
+  l->state = OLOCK_STATE_LOCKED;
+  ev_schedule(&l->event);
 }
 
 /**
@@ -191,5 +185,4 @@ olock_init(void)
 {
   DBG("olock: init\n");
   init_list(&olock_list);
-  olock_event = ev_new_init(&root_pool, olock_run_event, NULL);
 }
diff --git a/nest/locks.h b/nest/locks.h
index 37026c68..0cb33db9 100644
--- a/nest/locks.h
+++ b/nest/locks.h
@@ -31,8 +31,7 @@ struct object_lock {
   uint inst;		/* ... instance ID */
   struct iface *iface;	/* ... interface */
   struct iface *vrf;	/* ... or VRF (if iface is unknown) */
-  void (*hook)(struct object_lock *);	/* Called when the lock succeeds */
-  void *data;		/* User data */
+  event event;		/* Enqueued when the lock succeeds */
   /* ... internal to lock manager, don't touch ... */
   node n;		/* Node in list of olocks */
   int state;		/* OLOCK_STATE_xxx */
@@ -50,6 +49,5 @@ void olock_init(void);
 #define OLOCK_STATE_FREE 0
 #define OLOCK_STATE_LOCKED 1
 #define OLOCK_STATE_WAITING 2
-#define OLOCK_STATE_EVENT 3		/* waiting for unlock processing */
 
 #endif