Sockets: Unified API for main and other loops

Now sk_open() requires an explicit IO loop to open the socket in. Also
specific functions for socket RX pause / resume are added to allow for
BGP corking.

And last but not least, socket reloop is now synchronous to resolve
weird cases of the target loop stopping before actually picking up the
relooped socket. Now the caller must ensure that both loops are locked
while relooping, and this way all sockets always have their respective
loop.

11 files changed, 18 insertions, 37 deletions

diff --git a/proto/babel/packets.c b/proto/babel/packets.c
index d4acc170..f13bb5ba 100644
--- a/proto/babel/packets.c
+++ b/proto/babel/packets.c
@@ -1617,7 +1617,7 @@ babel_open_socket(struct babel_iface *ifa)
   sk->ttl = 1;
   sk->flags = SKF_LADDR_RX;
 
-  if (sk_open(sk) < 0)
+  if (sk_open(sk, p->p.loop) < 0)
     goto err;
 
   if (sk_setup_multicast(sk) < 0)
diff --git a/proto/bfd/bfd.c b/proto/bfd/bfd.c
index 1ef92b18..b6ccdb9a 100644
--- a/proto/bfd/bfd.c
+++ b/proto/bfd/bfd.c
@@ -603,16 +603,10 @@ bfd_free_iface(struct bfd_iface *ifa)
     return;
 
   if (ifa->sk)
-  {
-    sk_stop(ifa->sk);
     rfree(ifa->sk);
-  }
 
   if (ifa->rx)
-  {
-    sk_stop(ifa->rx);
     rfree(ifa->rx);
-  }
 
   rem_node(&ifa->n);
   mb_free(ifa);
@@ -1100,11 +1094,6 @@ bfd_shutdown(struct proto *P)
 
   bfd_drop_requests(p);
 
-  if (p->rx4_1) sk_stop(p->rx4_1);
-  if (p->rx4_m) sk_stop(p->rx4_m);
-  if (p->rx6_1) sk_stop(p->rx6_1);
-  if (p->rx6_m) sk_stop(p->rx6_m);
-
   return PS_DOWN;
 }
 
diff --git a/proto/bfd/packets.c b/proto/bfd/packets.c
index 2200ab09..a22f223b 100644
--- a/proto/bfd/packets.c
+++ b/proto/bfd/packets.c
@@ -430,12 +430,11 @@ bfd_open_rx_sk(struct bfd_proto *p, int multihop, int af)
   /* TODO: configurable ToS and priority */
   sk->tos = IP_PREC_INTERNET_CONTROL;
   sk->priority = sk_priority_control;
-  sk->flags = SKF_THREAD | SKF_LADDR_RX | (!multihop ? SKF_TTL_RX : 0);
+  sk->flags = SKF_LADDR_RX | (!multihop ? SKF_TTL_RX : 0);
 
-  if (sk_open(sk) < 0)
+  if (sk_open(sk, p->p.loop) < 0)
     goto err;
 
-  sk_start(sk);
   return sk;
 
  err:
@@ -462,12 +461,11 @@ bfd_open_rx_sk_bound(struct bfd_proto *p, ip_addr local, struct iface *ifa)
   /* TODO: configurable ToS and priority */
   sk->tos = IP_PREC_INTERNET_CONTROL;
   sk->priority = sk_priority_control;
-  sk->flags = SKF_THREAD | SKF_BIND | (ifa ? SKF_TTL_RX : 0);
+  sk->flags = SKF_BIND | (ifa ? SKF_TTL_RX : 0);
 
-  if (sk_open(sk) < 0)
+  if (sk_open(sk, p->p.loop) < 0)
     goto err;
 
-  sk_start(sk);
   return sk;
 
  err:
@@ -494,12 +492,11 @@ bfd_open_tx_sk(struct bfd_proto *p, ip_addr local, struct iface *ifa)
   sk->tos = IP_PREC_INTERNET_CONTROL;
   sk->priority = sk_priority_control;
   sk->ttl = ifa ? 255 : -1;
-  sk->flags = SKF_THREAD | SKF_BIND | SKF_HIGH_PORT;
+  sk->flags = SKF_BIND | SKF_HIGH_PORT;
 
-  if (sk_open(sk) < 0)
+  if (sk_open(sk, p->p.loop) < 0)
     goto err;
 
-  sk_start(sk);
   return sk;
 
  err:
diff --git a/proto/bgp/bgp.c b/proto/bgp/bgp.c
index 122b0c22..a9028353 100644
--- a/proto/bgp/bgp.c
+++ b/proto/bgp/bgp.c
@@ -263,7 +263,7 @@ bgp_listen_create(void *_ UNUSED)
       sk->rx_hook = bgp_incoming_connection;
       sk->err_hook = bgp_listen_sock_err;
 
-      if (sk_open(sk) < 0)
+      if (sk_open(sk, &main_birdloop) < 0)
       {
 	sk_log_error(sk, p->p.name);
 	log(L_ERR "%s: Cannot open listening socket", p->p.name);
@@ -1203,7 +1203,7 @@ bgp_connect(struct bgp_proto *p)	/* Enter Connect state and start establishing c
   bgp_setup_sk(conn, s);
   bgp_conn_set_state(conn, BS_CONNECT);
 
-  if (sk_open(s) < 0)
+  if (sk_open(s, p->p.loop) < 0)
     goto err;
 
   /* Set minimal receive TTL if needed */
diff --git a/proto/bgp/packets.c b/proto/bgp/packets.c
index 924d6828..c90798a0 100644
--- a/proto/bgp/packets.c
+++ b/proto/bgp/packets.c
@@ -3015,7 +3015,7 @@ bgp_kick_tx(void *vconn)
     ;
 
   if (!max && !ev_active(conn->tx_ev))
-    ev_schedule(conn->tx_ev);
+    proto_send_event(&conn->bgp->p, conn->tx_ev);
 }
 
 void
@@ -3023,13 +3023,14 @@ bgp_tx(sock *sk)
 {
   struct bgp_conn *conn = sk->data;
 
+  ASSERT_DIE(birdloop_inside(conn->bgp->p.loop));
   DBG("BGP: TX hook\n");
   uint max = 1024;
   while (--max && (bgp_fire_tx(conn) > 0))
     ;
 
   if (!max && !ev_active(conn->tx_ev))
-    ev_schedule(conn->tx_ev);
+    proto_send_event(&conn->bgp->p, conn->tx_ev);
 }
 
 
diff --git a/proto/ospf/iface.c b/proto/ospf/iface.c
index 1919bccb..0aa7fa00 100644
--- a/proto/ospf/iface.c
+++ b/proto/ospf/iface.c
@@ -136,7 +136,7 @@ ospf_sk_open(struct ospf_iface *ifa)
   sk->flags = SKF_LADDR_RX | (ifa->check_ttl ? SKF_TTL_RX : 0);
   sk->ttl = ifa->cf->ttl_security ? 255 : 1;
 
-  if (sk_open(sk) < 0)
+  if (sk_open(sk, p->p.loop) < 0)
     goto err;
 
   /* 12 is an offset of the checksum in an OSPFv3 packet */
@@ -220,7 +220,7 @@ ospf_open_vlink_sk(struct ospf_proto *p)
   sk->data = (void *) p;
   sk->flags = 0;
 
-  if (sk_open(sk) < 0)
+  if (sk_open(sk, p->p.loop) < 0)
     goto err;
 
   /* 12 is an offset of the checksum in an OSPFv3 packet */
diff --git a/proto/radv/packets.c b/proto/radv/packets.c
index 5cd8b2de..c6b565d2 100644
--- a/proto/radv/packets.c
+++ b/proto/radv/packets.c
@@ -493,7 +493,7 @@ radv_sk_open(struct radv_iface *ifa)
   sk->data = ifa;
   sk->flags = SKF_LADDR_RX;
 
-  if (sk_open(sk) < 0)
+  if (sk_open(sk, ifa->ra->p.loop) < 0)
     goto err;
 
   /* We want listen just to ICMPv6 messages of type RS and RA */
diff --git a/proto/rip/packets.c b/proto/rip/packets.c
index 9c3bd7a3..fecdf896 100644
--- a/proto/rip/packets.c
+++ b/proto/rip/packets.c
@@ -1012,7 +1012,7 @@ rip_open_socket(struct rip_iface *ifa)
 
   /* sk->rbsize and sk->tbsize are handled in rip_iface_update_buffers() */
 
-  if (sk_open(sk) < 0)
+  if (sk_open(sk, p->p.loop) < 0)
     goto err;
 
   if (ifa->cf->mode == RIP_IM_MULTICAST)
diff --git a/proto/rpki/ssh_transport.c b/proto/rpki/ssh_transport.c
index 223afa80..425ad460 100644
--- a/proto/rpki/ssh_transport.c
+++ b/proto/rpki/ssh_transport.c
@@ -35,11 +35,9 @@ rpki_tr_ssh_open(struct rpki_tr_sock *tr)
   sk->ssh->subsystem = "rpki-rtr";
   sk->ssh->state = SK_SSH_CONNECT;
 
-  if (sk_open(sk) != 0)
+  if (sk_open(sk, cache->p->p.loop) != 0)
     return RPKI_TR_ERROR;
 
-  sk_start(sk);
-
   return RPKI_TR_SUCCESS;
 }
 
diff --git a/proto/rpki/tcp_transport.c b/proto/rpki/tcp_transport.c
index 4e850c44..ebb8030f 100644
--- a/proto/rpki/tcp_transport.c
+++ b/proto/rpki/tcp_transport.c
@@ -28,11 +28,9 @@ rpki_tr_tcp_open(struct rpki_tr_sock *tr)
 
   sk->type = SK_TCP_ACTIVE;
 
-  if (sk_open(sk) != 0)
+  if (sk_open(sk, tr->cache->p->p.loop) != 0)
     return RPKI_TR_ERROR;
 
-  sk_start(sk);
-
   return RPKI_TR_SUCCESS;
 }
 
diff --git a/proto/rpki/transport.c b/proto/rpki/transport.c
index 4026fca4..81bd6dd8 100644
--- a/proto/rpki/transport.c
+++ b/proto/rpki/transport.c
@@ -85,7 +85,6 @@ rpki_tr_open(struct rpki_tr_sock *tr)
   sk->rbsize = RPKI_RX_BUFFER_SIZE;
   sk->tbsize = RPKI_TX_BUFFER_SIZE;
   sk->tos = IP_PREC_INTERNET_CONTROL;
-  sk->flags |= SKF_THREAD;
   sk->vrf = cache->p->p.vrf;
 
   if (ipa_zero(sk->daddr) && sk->host)
@@ -121,7 +120,6 @@ rpki_tr_close(struct rpki_tr_sock *tr)
 
   if (tr->sk)
   {
-    sk_stop(tr->sk);
     rfree(tr->sk);
     tr->sk = NULL;
   }