RPKI has its own loop

5 files changed, 22 insertions, 4 deletions

diff --git a/proto/rpki/config.Y b/proto/rpki/config.Y
index d6d326b8..743b5b42 100644
--- a/proto/rpki/config.Y
+++ b/proto/rpki/config.Y
@@ -42,6 +42,7 @@ proto: rpki_proto ;
 
 rpki_proto_start: proto_start RPKI {
   this_proto = proto_config_new(&proto_rpki, $1);
+  this_proto->loop_order = DOMAIN_ORDER(proto);
   RPKI_CFG->retry_interval = RPKI_RETRY_INTERVAL;
   RPKI_CFG->refresh_interval = RPKI_REFRESH_INTERVAL;
   RPKI_CFG->expire_interval = RPKI_EXPIRE_INTERVAL;
diff --git a/proto/rpki/rpki.c b/proto/rpki/rpki.c
index e3fccb48..cc86ab6a 100644
--- a/proto/rpki/rpki.c
+++ b/proto/rpki/rpki.c
@@ -109,6 +109,7 @@ static void rpki_schedule_next_expire_check(struct rpki_cache *cache);
 static void rpki_stop_refresh_timer_event(struct rpki_cache *cache);
 static void rpki_stop_retry_timer_event(struct rpki_cache *cache);
 static void rpki_stop_expire_timer_event(struct rpki_cache *cache);
+static void rpki_stop_all_timers(struct rpki_cache *cache);
 
 
 /*
@@ -219,6 +220,8 @@ rpki_force_restart_proto(struct rpki_proto *p)
 {
   if (p->cache)
   {
+    rpki_tr_close(p->cache->tr_sock);
+    rpki_stop_all_timers(p->cache);
     CACHE_DBG(p->cache, "Connection object destroying");
   }
 
@@ -342,7 +345,7 @@ rpki_schedule_next_refresh(struct rpki_cache *cache)
   btime t = cache->refresh_interval S;
 
   CACHE_DBG(cache, "after %t s", t);
-  tm_start(cache->refresh_timer, t);
+  tm_start_in(cache->refresh_timer, t, cache->p->p.loop);
 }
 
 static void
@@ -351,7 +354,7 @@ rpki_schedule_next_retry(struct rpki_cache *cache)
   btime t = cache->retry_interval S;
 
   CACHE_DBG(cache, "after %t s", t);
-  tm_start(cache->retry_timer, t);
+  tm_start_in(cache->retry_timer, t, cache->p->p.loop);
 }
 
 static void
@@ -362,7 +365,7 @@ rpki_schedule_next_expire_check(struct rpki_cache *cache)
   t = MAX(t, 1 S);
 
   CACHE_DBG(cache, "after %t s", t);
-  tm_start(cache->expire_timer, t);
+  tm_start_in(cache->expire_timer, t, cache->p->p.loop);
 }
 
 static void
@@ -379,13 +382,21 @@ rpki_stop_retry_timer_event(struct rpki_cache *cache)
   tm_stop(cache->retry_timer);
 }
 
-static void UNUSED
+static void
 rpki_stop_expire_timer_event(struct rpki_cache *cache)
 {
   CACHE_DBG(cache, "Stop");
   tm_stop(cache->expire_timer);
 }
 
+static void
+rpki_stop_all_timers(struct rpki_cache *cache)
+{
+  rpki_stop_refresh_timer_event(cache);
+  rpki_stop_retry_timer_event(cache);
+  rpki_stop_expire_timer_event(cache);
+}
+
 static int
 rpki_do_we_recv_prefix_pdu_in_last_seconds(struct rpki_cache *cache)
 {
diff --git a/proto/rpki/ssh_transport.c b/proto/rpki/ssh_transport.c
index 6333f367..223afa80 100644
--- a/proto/rpki/ssh_transport.c
+++ b/proto/rpki/ssh_transport.c
@@ -38,6 +38,8 @@ rpki_tr_ssh_open(struct rpki_tr_sock *tr)
   if (sk_open(sk) != 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 132f8e2d..4e850c44 100644
--- a/proto/rpki/tcp_transport.c
+++ b/proto/rpki/tcp_transport.c
@@ -31,6 +31,8 @@ rpki_tr_tcp_open(struct rpki_tr_sock *tr)
   if (sk_open(sk) != 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 a1ac7587..b52495dc 100644
--- a/proto/rpki/transport.c
+++ b/proto/rpki/transport.c
@@ -85,6 +85,7 @@ 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;
 
   if (ipa_zero(sk->daddr) && sk->host)
   {
@@ -119,6 +120,7 @@ rpki_tr_close(struct rpki_tr_sock *tr)
 
   if (tr->sk)
   {
+    sk_stop(tr->sk);
     rfree(tr->sk);
     tr->sk = NULL;
   }