Fixes core state machine.

The core state machine was broken - it didn't free resources
in START -> DOWN transition and might freed resources after
UP -> STOP transition before protocol turned down. It leads
to deadlock on olock acquisition when lock was not freed
during previous stop.

The current behavior is that resources, allocated during
DOWN -> * transition, are freed in * -> DOWN transition,
and flushing (scheduled in UP -> *) just counteract
feeding (scheduled in * -> UP). Protocol fell down
when both flushing is done (if needed) and protocol
reports DOWN.

BTW, is thera a reason why neighbour cache item acquired
by protocol is not tracked by resource mechanism?

nest/neighbor.c

@@ -254,7 +254,7 @@ neigh_if_down(struct iface *i)
 static inline void
 neigh_prune_one(neighbor *n)
 {
-  if (n->proto->core_state != FS_FLUSHING)
+  if (n->proto->proto_state != PS_DOWN)
     return;
   rem_node(&n->n);
   if (n->iface)

nest/proto.c

@@ -552,6 +552,30 @@ proto_feed(void *P)
   proto_feed_more(P);
 }
 
+static void
+proto_schedule_flush(struct proto *p)
+{
+  /* Need to abort feeding */
+  if (p->core_state == FS_FEEDING)
+    rt_feed_baby_abort(p);
+
+  DBG("%s: Scheduling flush\n", p->name);
+  p->core_state = FS_FLUSHING;
+  proto_relink(p);
+  proto_flush_hooks(p);
+  ev_schedule(proto_flush_event);
+}
+
+static void
+proto_schedule_feed(struct proto *p)
+{
+  DBG("%s: Scheduling meal\n", p->name);
+  p->core_state = FS_FEEDING;
+  proto_relink(p);
+  p->attn->hook = proto_feed;
+  ev_schedule(p->attn);
+}
+
 /**
  * proto_notify_state - notify core about protocol state change
  * @p: protocol the state of which has changed
@@ -562,7 +586,9 @@ proto_feed(void *P)
  * it should immediately notify the core about the change by calling
  * proto_notify_state() which will write the new state to the &proto
  * structure and take all the actions necessary to adapt to the new
- * state.
+ * state. State change to PS_DOWN immediately frees resources of protocol
+ * and might execute start callback of protocol; therefore,
+ * it should be used at tail positions of protocol callbacks.
  */
 void
 proto_notify_state(struct proto *p, unsigned ps)
@@ -574,23 +600,23 @@ proto_notify_state(struct proto *p, unsigned ps)
   if (ops == ps)
     return;
 
+  p->proto_state = ps;
+
   switch (ps)
     {
     case PS_DOWN:
+      neigh_prune(); // FIXME convert neighbors to resource?
+      rfree(p->pool);
+      p->pool = NULL;
+
       if (cs == FS_HUNGRY)		/* Shutdown finished */
 	{
-	  p->proto_state = ps;
 	  proto_fell_down(p);
 	  return;			/* The protocol might have ceased to exist */
 	}
-      else if (cs == FS_FLUSHING)	/* Still flushing... */
+      /* Otherwise, we have something to flush... */
-	;
+      else if (cs != FS_FLUSHING)
-      else
+	proto_schedule_flush(p);
-	{
-	  if (cs == FS_FEEDING)		/* Need to abort feeding */
-	    rt_feed_baby_abort(p);
-	  goto schedule_flush;		/* Need to start flushing */
-	}
       break;
     case PS_START:
       ASSERT(ops == PS_DOWN);
@@ -599,27 +625,15 @@ proto_notify_state(struct proto *p, unsigned ps)
     case PS_UP:
       ASSERT(ops == PS_DOWN || ops == PS_START);
       ASSERT(cs == FS_HUNGRY);
-      DBG("%s: Scheduling meal\n", p->name);
+      proto_schedule_feed(p);
-      cs = FS_FEEDING;
-      p->attn->hook = proto_feed;
-      ev_schedule(p->attn);
       break;
     case PS_STOP:
-      if (ops != PS_DOWN)
+      if ((cs = FS_FEEDING) || (cs == FS_HAPPY))
-	{
+	proto_schedule_flush(p);
-	schedule_flush:
-	  DBG("%s: Scheduling flush\n", p->name);
-	  proto_flush_hooks(p);
-	  cs = FS_FLUSHING;
-	  ev_schedule(proto_flush_event);
-	}
       break;
     default:
       bug("Invalid state transition for %s from %s/%s to */%s", p->name, c_states[cs], p_states[ops], p_states[ps]);
     }
-  p->proto_state = ps;
-  p->core_state = cs;
-  proto_relink(p);
 }
 
 static void
@@ -628,15 +642,13 @@ proto_flush_all(void *unused UNUSED)
   struct proto *p;
 
   rt_prune_all();
-  neigh_prune();
   while ((p = HEAD(flush_proto_list))->n.next)
     {
       DBG("Flushing protocol %s\n", p->name);
-      rfree(p->pool);
-      p->pool = NULL;
       p->core_state = FS_HUNGRY;
       proto_relink(p);
-      proto_fell_down(p);
+      if (p->proto_state == PS_DOWN)
+	proto_fell_down(p);
     }
 }