RPKI: refactore thread's communication usign pipes

diff --git a/proto/rpki/rpki.c b/proto/rpki/rpki.c
index 854a86095753f178a4e598df5d4b159523e13992..3202e52ad41f3b75dfbc0afc33eb273b32264278 100644 (file)
--- a/proto/rpki/rpki.c
+++ b/proto/rpki/rpki.c
@@ -306,9 +306,9 @@ static void
 send_data_to_main_thread(struct rpki_proto *p, struct rpki_entry *e)
 {
   rpki_lock_notify(p);
-  add_tail(&p->notify_list, &e->n);
+  add_tail(&p->roa_update_list, &e->n);
   rpki_unlock_notify(p);
-  pipe_kick(p->notify_write_sk->fd);
+  pipe_kick(p->roa_update.write->fd);
 }
 
 static void
@@ -392,8 +392,8 @@ recv_data_in_main_thread(struct birdsock *sk, int size)
 
   rpki_lock_notify(p);
   init_list(&tmp_list);
-  add_tail_list(&tmp_list, &p->notify_list);
-  init_list(&p->notify_list);
+  add_tail_list(&tmp_list, &p->roa_update_list);
+  init_list(&p->roa_update_list);
   rpki_unlock_notify(p);
 
   WALK_LIST_FIRST(e, tmp_list)
@@ -420,7 +420,7 @@ recv_err_in_main_thread(struct birdsock *sk, int err)
 }
 
 static sock *
-create_socket(struct rpki_proto *p, int fd)
+create_pipe(struct rpki_proto *p, int fd)
 {
   sock *sk = sk_new(p->p.pool);
   sk->type = SK_MAGIC;
@@ -429,38 +429,37 @@ create_socket(struct rpki_proto *p, int fd)
   return sk;
 }
 
-static void
-create_read_socket(struct rpki_proto *p, int fd)
+static sock *
+create_read_pipe(struct rpki_proto *p, int fd)
 {
-  sock *sk = create_socket(p, fd);
+  sock *sk = create_pipe(p, fd);
   sk->rx_hook = recv_data_in_main_thread;
   sk->err_hook = recv_err_in_main_thread;
   if (sk_open(sk) < 0)
-    RPKI_DIE(p, "read socket sk_open() failed");
-  p->notify_read_sk = sk;
+    return NULL;
+  return sk;
 }
 
-static void
-create_write_socket(struct rpki_proto *p, int fd)
+static sock *
+create_write_pipe(struct rpki_proto *p, int fd)
 {
-  sock *sk = create_socket(p, fd);
+  sock *sk = create_pipe(p, fd);
   sk->flags = SKF_THREAD;
   if (sk_open(sk) < 0)
-    RPKI_DIE(p, "write socket sk_open() failed");
-  p->notify_write_sk = sk;
+    return NULL;
+  return sk;
 }
 
 static void
-create_rw_sockets(struct rpki_proto *p)
+create_pipe_pair(struct rpki_proto *p, struct rpki_rw_sk_pair *sk_pair)
 {
   int pipe_fildes[2];
 
-  int rv = pipe(pipe_fildes);
-  if (rv < 0)
+  if ((pipe(pipe_fildes) < 0)
+      || ((sk_pair->read = create_read_pipe(p, pipe_fildes[0])) == NULL)
+      || ((sk_pair->write = create_write_pipe(p, pipe_fildes[1])) == NULL)
+  )
     RPKI_DIE(p, "pipe: %m");
-
-  create_read_socket (p, pipe_fildes[0]);
-  create_write_socket(p, pipe_fildes[1]);
 }
 
 static uint
@@ -617,9 +616,9 @@ rpki_start(struct proto *P)
   struct rpki_proto *p = (struct rpki_proto *) P;
   struct rpki_config *cf = (struct rpki_config *) (P->cf);
 
-  create_rw_sockets(p);
-  init_list(&p->notify_list);
-  pthread_mutex_init(&p->notify_lock, NULL);
+  create_pipe_pair(p, &p->roa_update);
+  init_list(&p->roa_update_list);
+  pthread_mutex_init(&p->roa_update_lock, NULL);
 
   lock_rpki_proto_list();
   add_tail(&rpki_proto_list, &p->rpki_node);
@@ -668,7 +667,7 @@ rpki_shutdown(struct proto *P)
   rem2_node(&p->rpki_node);
   unlock_rpki_proto_list();
 
-  pthread_mutex_destroy(&p->notify_lock);
+  pthread_mutex_destroy(&p->roa_update_lock);
 
   return PS_DOWN;
 }

diff --git a/proto/rpki/rpki.h b/proto/rpki/rpki.h
index e88a561cf4f9bfc76482367a031aa7e32023770b..3f59f0e58a13f073de25a7c9d09f43699677ebff 100644 (file)
--- a/proto/rpki/rpki.h
+++ b/proto/rpki/rpki.h
@@ -64,6 +64,11 @@ struct rpki_config {
   struct roa_table_config *roa_table_cf;
 };
 
+struct rpki_rw_sk_pair {
+  sock *read;
+  sock *write;
+};
+
 struct rpki_proto {
   struct proto p;
   struct rpki_config *cf;
@@ -71,16 +76,15 @@ struct rpki_proto {
 
   struct rtr_mgr_config *rtr_conf;
 
-  sock *notify_read_sk;
-  sock *notify_write_sk;
-  list notify_list;
-  pthread_mutex_t notify_lock;
+  struct rpki_rw_sk_pair roa_update;
+  list roa_update_list;
+  pthread_mutex_t roa_update_lock;
 };
 
 struct rpki_cache *rpki_new_cache(void);
 
-static inline void rpki_lock_notify(struct rpki_proto *p) { pthread_mutex_lock(&p->notify_lock); }
-static inline void rpki_unlock_notify(struct rpki_proto *p) { pthread_mutex_unlock(&p->notify_lock); }
+static inline void rpki_lock_notify(struct rpki_proto *p) { pthread_mutex_lock(&p->roa_update_lock); }
+static inline void rpki_unlock_notify(struct rpki_proto *p) { pthread_mutex_unlock(&p->roa_update_lock); }
 
 void rpki_init_all(void);
 char *rpki_load_rtrlib(void);