rtp.io: integrate notification channel

Allocate a single notification socketpair to be shared by
all opensips workers, pass one side to the hosted rtpproxy
process and provide API for the rtpproxy module to tap into
the other end.

Author: sobomax

modules/rtp.io/rtp_io.c

@@ -29,6 +29,8 @@
 #include "../../dprint.h"
 #include "../../timer.h"
 
+#include "../rtpproxy/rtpproxy.h"
+#include "../rtpproxy/notification_process.h"
 #include "rtp_io.h"
 #include "rtp_io_util.h"
 #include "rtp_io_params.h"
@@ -40,7 +42,7 @@ static void mod_destroy(void);
 
 static const dep_export_t deps = {
     { /* OpenSIPS module dependencies */
-        { MOD_TYPE_DEFAULT, "rtpproxy", DEP_SILENT|DEP_REVERSE },
+        { MOD_TYPE_DEFAULT, "rtpproxy", (DEP_SILENT|DEP_REVERSE) & (~DEP_REVERSE_CINIT) },
         { MOD_TYPE_NULL, NULL, 0 },
     },
 	{ /* modparam dependencies */
@@ -49,12 +51,14 @@ static const dep_export_t deps = {
 };
 
 static int rtp_io_getchildsock(int);
+static int rtp_io_getrnsock(struct rtpp_notify_cfg *);
 
 /*
  * Exported functions
  */
 static const cmd_export_t cmds[] = {
     {"rtp_io_getchildsock", (cmd_function)rtp_io_getchildsock, {0}, 0},
+    {"rtp_io_getrnsock", (cmd_function)rtp_io_getrnsock, {0}, 0},
     {0}
 };
 
@@ -162,17 +166,29 @@ static int mod_init(void)
         ENV_ADD(argv_stat[i], e1);
     }
 
+    struct rtpp_n_sock *n_sock = &rpi_descp->n_sock;
+    int *fdp = n_sock->_fds;
+    if (socketpair(AF_UNIX, SOCK_STREAM, 0, fdp) < 0)
+        goto e1;
+    snprintf(n_sock->_name, sizeof(n_sock->_name), "fd:%d", n_sock->fds.rtpp);
+    n_sock->name.s = n_sock->_name;
+    n_sock->name.len = strlen(n_sock->_name);
+    ENV_ADD("-n", e2);
+    ENV_ADD("%s", e2, n_sock->name.s);
     for (int i = 0; i < nsocks; i++) {
         int *fdp = &rpi_descp->socks->holder[i * 2];
         if (socketpair(AF_UNIX, SOCK_STREAM, 0, fdp) < 0)
-            goto e1;
-        ENV_ADD("-s", e1);
-        ENV_ADD("fd:%d", e1, fdp[0]);
+            goto e2;
+        ENV_ADD("-s", e2);
+        ENV_ADD("fd:%d", e2, fdp[0]);
     }
 
     rpi_descp->socks->n = nsocks;
 
     return 0;
+e2:
+    close(n_sock->_fds[0]);
+    close(n_sock->_fds[1]);
 e1:
     free(rpi_descp->socks);
 e0:
@@ -198,6 +214,7 @@ void mod_destroy(void)
     }
     rtp_io_close_serv_socks();
     rtp_io_close_cnlt_socks();
+    rtp_io_close_cnlt_nsock();
     free(rpi_descp->socks);
 }
 
@@ -206,11 +223,25 @@ void mod_destroy(void)
 static int
 child_init(int rank)
 {
+    rtpproxy_is_nproc_t is_nproc_f;
+
+    is_nproc_f = (rtpproxy_is_nproc_t)find_export("rtpproxy_is_nproc", 0);
+    if (is_nproc_f == NULL) {
+        LM_ERR("rtpproxy_is_nproc() not found in the rtpproxy module\n");
+        return -1;
+    }
+    int is_nproc = is_nproc_f(NPROC_CHECK);
+    LM_DBG("rtp.io: child_init(%d), notifier: %d\n", rank, is_nproc);
     if (rank > rpi_descp->socks->n) {
         LM_ERR("BUG: rank is higher than the number of sockets!\n");
         return -1;
     }
 
+    if (!is_nproc && rtp_io_close_cnlt_nsock() != 0) {
+        LM_ERR("rtp_io_close_cnlt_nsock() failed\n");
+        return -1;
+    }
+
     if (rank <= 0) {
         if (rtp_io_close_cnlt_socks() != 0) {
             LM_ERR("rtp_io_close_cnlt_socks() failed\n");
@@ -240,3 +271,12 @@ static int rtp_io_getchildsock(int rank)
     int *fdp = &rpi_descp->socks->holder[(rank - 1) * 2];
     return (fdp[1]);
 }
+
+static int rtp_io_getrnsock(struct rtpp_notify_cfg *rn_cfg)
+{
+
+    rn_cfg->name = rpi_descp->n_sock.name;
+    rn_cfg->sock.rn_umode = CM_RTPIO;
+    rn_cfg->sock.fd = rpi_descp->n_sock.fds.osips;
+    return (0);
+};

modules/rtp.io/rtp_io.h

@@ -25,10 +25,23 @@ struct rtp_io_socks {
     int holder[];
 };
 
+struct rtpp_n_sock {
+    char _name[32];
+    str name;
+    union {
+        struct {
+            int rtpp;
+            int osips;
+        } fds;
+        int _fds[2];
+    };
+};
+
 struct rtp_io_desc {
     struct rtpp_cfg *rtpp_cfsp;
     struct rtpp_env_hd env;
     struct rtp_io_socks *socks;
+    struct rtpp_n_sock n_sock;
 };
 
 extern struct rtp_io_desc *rpi_descp;

modules/rtp.io/rtp_io_api.h

@@ -1,3 +1,6 @@
 #pragma once
 
+struct rtpp_notify_cfg;
+
 typedef int (*rtp_io_getchildsock_t)(int);
+typedef int (*rtp_io_getrnsock_t)(struct rtpp_notify_cfg *);

modules/rtp.io/rtp_io_host.c

@@ -72,6 +72,8 @@ void rtpproxy_host_process(int rank)
         goto e1;
     if (rtp_io_close_cnlt_socks() != 0)
         goto e1;
+    if (rtp_io_close_cnlt_nsock() != 0)
+        goto e1;
 
     OPT_RESTORE();
     rpi_descp->rtpp_cfsp = rtpp_main(argc, argv);
@@ -101,5 +103,8 @@ ipc_shutdown_rtpp_host(int sender, void *param)
         if (rpi_descp->socks->holder[i] != -1)
             close(rpi_descp->socks->holder[i]);
     }
+    if (rpi_descp->n_sock.fds.rtpp != -1) {
+        close(rpi_descp->n_sock.fds.rtpp);
+    }
     free(rpi_descp->socks);
 }

modules/rtp.io/rtp_io_util.c

@@ -85,19 +85,36 @@ int rtp_io_close_serv_socks(void)
 
     for (int i = 0; i < (rpi_descp->socks->n * 2); i+=2) {
         if (rpi_descp->socks->holder[i] != -1) {
-            close(rpi_descp->socks->holder[i]);
+            if (close(rpi_descp->socks->holder[i]) != 0)
+                return -1;
             rpi_descp->socks->holder[i] = -1;
         }
     }
+    if (rpi_descp->n_sock.fds.rtpp != -1) {
+        if (close(rpi_descp->n_sock.fds.rtpp) != 0)
+            return -1;
+        rpi_descp->n_sock.fds.rtpp = -1;
+    }
     return (0);
 }
 
+int rtp_io_close_cnlt_nsock(void)
+{
+    if (rpi_descp->n_sock.fds.osips != -1) {
+	if (close(rpi_descp->n_sock.fds.osips) != 0)
+            return -1;
+	rpi_descp->n_sock.fds.osips = -1;
+    }
+    return 0;
+}
+
 int rtp_io_close_cnlt_socks(void)
 {
 
     for (int i = 0; i < (rpi_descp->socks->n * 2); i+=2) {
         if (rpi_descp->socks->holder[i+1] != -1) {
-            close(rpi_descp->socks->holder[i+1]);
+            if (close(rpi_descp->socks->holder[i+1]) != 0)
+                return -1;
             rpi_descp->socks->holder[i+1] = -1;
         }
     }

modules/rtp.io/rtp_io_util.h

@@ -11,3 +11,4 @@ const char *const * rtp_io_env_gen_argv(struct rtpp_env_hd *, int *);
 
 int rtp_io_close_serv_socks(void);
 int rtp_io_close_cnlt_socks(void);
+int rtp_io_close_cnlt_nsock(void);