RTP: dequeue and discard late packets

[vlc] / modules / access / rtp / input.c

diff --git a/modules/access/rtp/input.c b/modules/access/rtp/input.c
index 995c7ba3737881a9459fd0f5b92b3648dbf8eba0..622e91456be896b0e51c74986368a389d555c549 100644 (file)
--- a/modules/access/rtp/input.c
+++ b/modules/access/rtp/input.c
@@ -37,6 +37,12 @@
 #include "rtp.h"
 #include <srtp.h>
 
+static bool fd_dead (int fd)
+{
+    struct pollfd ufd = { .fd = fd, };
+    return (poll (&ufd, 1, 0) > 0) && (ufd.revents & POLLHUP);
+}
+
 /**
  * Gets a datagram from the network.
  * @param fd datagram file descriptor
@@ -54,8 +60,7 @@ static block_t *rtp_dgram_recv (vlc_object_t *obj, int fd)
                         block->p_buffer, block->i_buffer, false);
         vlc_cleanup_pop ();
 
-        if (((len <= 0) && poll (&(struct pollfd){ .fd = fd, }, 1, 0))
-         || !vlc_object_alive (obj))
+        if (((len <= 0) && fd_dead (fd)) || !vlc_object_alive (obj))
         {   /* POLLHUP -> permanent (DCCP) socket error */
             block_Release (block);
             return NULL;
@@ -160,43 +165,52 @@ void *rtp_thread (void *data)
     demux_t *demux = data;
     demux_sys_t *p_sys = demux->p_sys;
 
-    for (;;)
+    do
     {
         block_t *block = rtp_recv (demux);
-        if (block == NULL)
-            break; /* fatal error: abort */
 
         vlc_mutex_lock (&p_sys->lock);
-
-        /* Autodetect payload type, _before_ rtp_queue() */
-        if (p_sys->autodetect)
+        if (block == NULL)
+            p_sys->dead = true; /* Fatal error: abort */
+        else
         {
-            if (rtp_autodetect (demux, p_sys->session, block))
-            {
-                block_Release (block);
-                continue;
+            if (p_sys->autodetect)
+            {   /* Autodetect payload type, _before_ rtp_queue() */
+                if (rtp_autodetect (demux, p_sys->session, block))
+                {
+                    vlc_mutex_unlock (&p_sys->lock);
+                    block_Release (block);
+                    continue;
+                }
+                p_sys->autodetect = false;
             }
-            p_sys->autodetect = false;
+            rtp_queue (demux, p_sys->session, block);
         }
-
-        rtp_queue (demux, p_sys->session, block);
         vlc_cond_signal (&p_sys->wait);
         vlc_mutex_unlock (&p_sys->lock);
     }
-    /* TODO: return 0 from Demux */
+    while (!p_sys->dead);
+
     return NULL;
 }
 
 
-void rtp_process (demux_t *demux)
+/**
+ * Process one RTP packet from the de-jitter queue.
+ * @return 0 on success, -1 on EOF
+ */
+int rtp_process (demux_t *demux)
 {
     demux_sys_t *p_sys = demux->p_sys;
     mtime_t deadline = INT64_MAX;
 
     vlc_mutex_lock (&p_sys->lock);
-    if (!rtp_dequeue (demux, p_sys->session, &deadline))
-        deadline = mdate () + CLOCK_FREQ / 5;
-    vlc_cond_timedwait (&p_sys->wait, &p_sys->lock, deadline);
+    if (rtp_dequeue (demux, p_sys->session, &deadline))
+        /* Pace the demux thread */
+        vlc_cond_timedwait (&p_sys->wait, &p_sys->lock, deadline);
+    else
+        vlc_cond_wait (&p_sys->wait, &p_sys->lock);
     vlc_mutex_unlock (&p_sys->lock);
-}
 
+    return p_sys->dead ? -1 : 0;
+}