Add rtpp_packetport_flush() to cause packetport to flush outbound

queue.

Author: sobomax

src/rtpp_packetport.c

@@ -64,6 +64,7 @@
 #include "rtp_packet.h"
 #include "rtp_packet_priv.h"
 #include "rtpp_packetport.h"
+#include "librtpproxy/packetport.h"
 #include "rtpp_packet_ext.h"
 #include "rtpp_pipe.h"
 #include "rtpp_session.h"
@@ -80,6 +81,7 @@
 #define RTPP_PACKETPORT_IN_PORT0 2
 #define RTPP_PPSIG_TERM 0
 #define RTPP_PPSIG_NUDGE 1
+#define RTPP_PPSIG_FLUSH 2
 
 static const struct timespec rtpp_packetport_tick = {
     .tv_sec = 0,
@@ -93,6 +95,13 @@ struct rtpp_packetport_int_priv {
     unsigned int capacity;
     _Atomic(unsigned int) out_port;
     _Atomic(unsigned int) in_port;
+    struct {
+        _Atomic(uint64_t) out_run;
+        uint64_t out_run_lc;
+    } __attribute__((aligned(CACHELINE_SIZE))) out_run_state;
+    struct {
+        uint64_t last_flush_run;
+    } __attribute__((aligned(CACHELINE_SIZE))) flush_state;
     pthread_t worker_id;
     struct rtpp_queue *pqueue;
     struct rtpp_hash_table *streams_ht;
@@ -136,6 +145,7 @@ static int rtpp_packetport_queue_wi(struct rtpp_packetport_int_priv *,
   struct rtpp_wi *);
 static int rtpp_packetport_proc_wi_batch(struct rtpp_packetport_int_priv *,
   struct rtpp_wi **, int);
+static uint64_t rtpp_packetport_out_run(struct rtpp_packetport_int_priv *);
 static void rtpp_packetport_drain_queue(struct SPMCQueue *);
 static void rtpp_packetport_drain_out(struct rtpp_packetport_int_priv *);
 static int rtpp_packetport_ismapped(const void *);
@@ -318,6 +328,8 @@ rtpp_packetport_ctor(unsigned int capacity)
     pvt->id = RTPP_PACKETPORT_PRIV_ID;
     atomic_init(&pvt->out_port, (unsigned int)RTPP_PACKETPORT_OUT_PORT0);
     atomic_init(&pvt->in_port, (unsigned int)RTPP_PACKETPORT_IN_PORT0);
+    atomic_init(&pvt->out_run_state.out_run, 0);
+    pvt->flush_state.last_flush_run = UINT64_MAX;
     PUBINST_FININIT(&pvt->pub, pvt, rtpp_packetport_dtor_obj);
 #if HAVE_PTHREAD_SETNAME_NP
     (void)pthread_setname_np(pvt->worker_id, "rtpp_packetport");
@@ -348,6 +360,31 @@ rtpp_packetport_try_push(struct rtpp_packetport *ext, struct rtp_packet_ext *pkt
     return (try_push(pvt->ext.out, pktxp) ? 0 : -1);
 }
 
+int
+rtpp_packetport_flush(struct rtpp_packetport *ext)
+{
+    struct rtpp_packetport_int_priv *pvt;
+    struct rtpp_wi *wi;
+    uint64_t seen_run;
+
+    EXT2PVT(ext, pvt);
+    seen_run = atomic_load_explicit(&pvt->out_run_state.out_run,
+      memory_order_relaxed);
+    if (seen_run == pvt->flush_state.last_flush_run) {
+        return (0);
+    }
+    wi = rtpp_wi_malloc_sgnl(RTPP_PPSIG_FLUSH, &seen_run, sizeof(seen_run));
+    if (wi == NULL) {
+        return (-1);
+    }
+    if (rtpp_queue_put_item(wi, pvt->pqueue) != 0) {
+        RTPP_OBJ_DECREF(wi);
+        return (-1);
+    }
+    pvt->flush_state.last_flush_run = seen_run;
+    return (0);
+}
+
 struct rtpp_packetport_int *
 rtpp_packetport_get_int(struct rtpp_packetport *ext)
 {
@@ -424,14 +461,16 @@ rtpp_packetport_run(void *argp)
         if (qlen == 0)
             qlen = CALL_SMETHOD(pvt->streams_ht, get_length) > 0 ? 16 : 0;
         if (qlen > 0) {
-            rtpp_packetport_drain_out(pvt);
             if (ddp == NULL) {
                 ddp = &deadline;
                 dtime2mtimespec(getdtime(), ddp);
             }
-            rtpp_packetport_deadline_step(ddp);
             nwis = rtpp_queue_get_items_by(pvt->pqueue, pvt->wi_batch,
               (int)pvt->capacity, ddp, NULL);
+            if (nwis == 0) {
+                rtpp_packetport_drain_out(pvt);
+                rtpp_packetport_deadline_step(ddp);
+            }
             qlen -= 1;
         } else {
             ddp = NULL;
@@ -579,16 +618,31 @@ rtpp_packetport_proc_wi_batch(struct rtpp_packetport_int_priv *pvt,
     struct rtpp_wi_pvt *wipp;
     struct rtp_packet_ext *pktxp;
     struct rtp_packet *pktp;
-    int i, signum, term_seen;
+    uint64_t sig_run;
+    size_t dlen;
+    int flush_seen, i, signum, term_seen;
 
     term_seen = 0;
+    flush_seen = 0;
     for (i = 0; i < nwis; i++) {
         if (wis[i]->wi_type == RTPP_WI_TYPE_SGNL) {
             signum = rtpp_wi_sgnl_get_signum(wis[i]);
             if (signum == RTPP_PPSIG_TERM) {
                 term_seen = 1;
                 break;
             }
+            if (signum == RTPP_PPSIG_FLUSH) {
+                const uint64_t *srp;
+
+                srp = rtpp_wi_sgnl_get_data(wis[i], &dlen);
+                if (dlen == sizeof(*srp)) {
+                    sig_run = *srp;
+                    if (sig_run >= pvt->out_run_state.out_run_lc) {
+                        flush_seen = 1;
+                    }
+                }
+                continue;
+            }
             RTPP_DBG_ASSERT(signum == RTPP_PPSIG_NUDGE);
             continue;
         }
@@ -606,9 +660,23 @@ rtpp_packetport_proc_wi_batch(struct rtpp_packetport_int_priv *pvt,
         RTPP_OBJ_DECREF(wis[i]);
     if (term_seen != 0)
         return (-1);
+    if (flush_seen != 0) {
+        rtpp_packetport_drain_out(pvt);
+    }
     return (0);
 }
 
+static uint64_t
+rtpp_packetport_out_run(struct rtpp_packetport_int_priv *pvt)
+{
+    uint64_t run;
+
+    run = atomic_fetch_add_explicit(&pvt->out_run_state.out_run, 1,
+      memory_order_relaxed) + 1;
+    pvt->out_run_state.out_run_lc = run;
+    return (run);
+}
+
 static void
 rtpp_packetport_drain_queue(struct SPMCQueue *queue)
 {
@@ -631,6 +699,7 @@ rtpp_packetport_drain_out(struct rtpp_packetport_int_priv *pvt)
     struct pkt_proc_ctx pktx;
     size_t i, nitems;
 
+    (void)rtpp_packetport_out_run(pvt);
     while ((nitems = try_pop_many(pvt->ext.out, pvt->out_batch,
       (size_t)pvt->capacity)) > 0) {
         for (i = 0; i < nitems; i++) {