BGP and HCU uncorking is processed in the right loop

closes #86

The uncork events are running from mainloop so these should just
dispatch the right event to the right loop. Doing anything long there
is bad for performance and latency as the uncork list may be huge.

nest/rt-table.c

@@ -129,6 +129,7 @@ struct rt_cork rt_cork;
 /* Data structures for export journal */
 
 static void rt_free_hostcache(struct rtable_private *tab);
+static void rt_hcu_uncork(void *_tab);
 static void rt_update_hostcache(void *tab);
 static void rt_next_hop_update(void *_tab);
 static void rt_nhu_uncork(void *_tab);
@@ -4725,7 +4726,7 @@ rt_init_hostcache(struct rtable_private *tab)
   hc->tab = RT_PUB(tab);
 
   tab->hcu_event = ev_new_init(tab->rp, rt_update_hostcache, tab);
-  tab->hcu_uncork_event = ev_new_init(tab->rp, rt_update_hostcache, tab);
+  tab->hcu_uncork_event = ev_new_init(tab->rp, rt_hcu_uncork, tab);
   tab->hostcache = hc;
 
   ev_send_loop(tab->loop, tab->hcu_event);
@@ -4877,6 +4878,13 @@ done:
   return old_src != new_src;
 }
 
+static void
+rt_hcu_uncork(void *_tab)
+{
+  RT_LOCKED((rtable *) _tab, tab)
+    ev_send_loop(tab->loop, tab->hcu_event);
+}
+
 static void
 rt_update_hostcache(void *data)
 {

proto/bgp/bgp.c

@@ -573,6 +573,8 @@ bgp_down(struct bgp_proto *p)
     bgp_close(p);
   }
 
+  rfree(p->uncork_do_ev);
+
   p->neigh = NULL;
 
   BGP_TRACE(D_EVENTS, "Down");
@@ -624,7 +626,7 @@ void
 bgp_stop(struct bgp_proto *p, int subcode, byte *data, uint len)
 {
   proto_notify_state(&p->p, PS_STOP);
-  p->uncork_ev->data = NULL;
+
   bgp_graceful_close_conn(&p->outgoing_conn, subcode, data, len);
   bgp_graceful_close_conn(&p->incoming_conn, subcode, data, len);
 
@@ -1730,7 +1732,8 @@ bgp_start(struct proto *P)
   p->last_rx_update = 0;
 
   p->event = ev_new_init(p->p.pool, bgp_decision, p);
-  p->uncork_ev = ev_new_init(p->p.pool, bgp_uncork, p);
+  p->uncork_main_ev = ev_new_init(p->p.pool, bgp_uncork_main, p);
+  p->uncork_do_ev = ev_new_init(p->p.pool, bgp_do_uncork, p);
 
   p->startup_timer = tm_new_init(p->p.pool, bgp_startup_timeout, p, 0, 0);
   p->gr_timer = tm_new_init(p->p.pool, bgp_graceful_restart_timeout, p, 0, 0);

proto/bgp/bgp.h

@@ -364,7 +364,8 @@ struct bgp_proto {
   struct bgp_listen_request listen;	/* Shared listening socket */
   struct bfd_request *bfd_req;		/* BFD request, if BFD is used */
   struct birdsock *postponed_sk;	/* Postponed incoming socket for dynamic BGP */
-  event *uncork_ev;			/* Uncork event in case of congestion */
+  event *uncork_main_ev;		/* Uncork event for mainloop */
+  event *uncork_do_ev;			/* Uncork event to actually uncork */
   struct bgp_stats stats;		/* BGP statistics */
   btime last_established;		/* Last time of enter/leave of established state */
   btime last_rx_update;			/* Last time of RX update */
@@ -710,7 +711,8 @@ void bgp_schedule_packet(struct bgp_conn *conn, struct bgp_channel *c, int type)
 void bgp_kick_tx(void *vconn);
 void bgp_tx(struct birdsock *sk);
 int bgp_rx(struct birdsock *sk, uint size);
-void bgp_uncork(void *vp);
+void bgp_uncork_main(void *vp);
+void bgp_do_uncork(void *vp);
 const char * bgp_error_dsc(unsigned code, unsigned subcode);
 void bgp_log_error(struct bgp_proto *p, u8 class, char *msg, unsigned code, unsigned subcode, byte *data, unsigned len);
 

proto/bgp/packets.c

@@ -3438,18 +3438,15 @@ bgp_rx_packet(struct bgp_conn *conn, byte *pkt, uint len)
 }
 
 void
-bgp_uncork(void *vp)
+bgp_do_uncork(void *vp)
 {
-  /* The uncork event is run from &main_birdloop and there is no useful way how
-   * to assign the target loop to it, thus we have to lock it ourselves. */
-
   struct bgp_proto *p = vp;
-  if (!p)
+  ASSERT_DIE(birdloop_inside(p->p.loop));
-    return;
+  ASSERT_DIE(p->p.active_loops--);
 
-  birdloop_enter(p->p.loop);
+  if (p->p.proto_state == PS_DOWN)
-
+    ev_send_loop(p->p.loop, p->p.event);
-  if (p && p->conn && (p->conn->state == BS_ESTABLISHED) && !p->conn->sk->rx_hook)
+  else if (p->conn && (p->conn->state == BS_ESTABLISHED) && !p->conn->sk->rx_hook)
   {
     struct birdsock *sk = p->conn->sk;
     ASSERT_DIE(sk->rpos > sk->rbuf);
@@ -3457,8 +3454,16 @@ bgp_uncork(void *vp)
     bgp_rx(sk, sk->rpos - sk->rbuf);
     BGP_TRACE(D_PACKETS, "Uncorked");
   }
+}
 
-  birdloop_leave(p->p.loop);
+void
+bgp_uncork_main(void *vp)
+{
+  /* The uncork event is run from &main_birdloop and there is no useful way how
+   * to assign the target loop to it, thus we have to lock it ourselves. */
+
+  struct bgp_proto *p = vp;
+  ev_send_loop(p->p.loop, p->uncork_do_ev);
 }
 
 /**
@@ -3485,9 +3490,10 @@ bgp_rx(sock *sk, uint size)
     {
       if ((conn->state == BS_CLOSE) || (conn->sk != sk))
 	return 0;
-      if ((conn->state == BS_ESTABLISHED) && rt_cork_check(conn->bgp->uncork_ev))
+      if ((conn->state == BS_ESTABLISHED) && rt_cork_check(conn->bgp->uncork_main_ev))
       {
 	sk_pause_rx(p->p.loop, sk);
+	p->p.active_loops++;
 	BGP_TRACE(D_PACKETS, "Corked");
 	break;
       }