X-Git-Url: https://git.sesse.net/?a=blobdiff_plain;f=libavformat%2Fudp.c;h=802cbd5975a11c88e04ca7805876a39b7fe04c45;hb=7b915726ebeab8ce4125dacb4afa79f04913a6fe;hp=d48d5417c75f1b7093462ea8a657ae202afbd7e4;hpb=10d2ae8cf78ed7e23c6342200fa4d8a5355e87f2;p=ffmpeg

diff --git a/libavformat/udp.c b/libavformat/udp.c
index d48d5417c75..802cbd5975a 100644
--- a/libavformat/udp.c
+++ b/libavformat/udp.c
@@ -25,15 +25,17 @@
  */
 
 #define _BSD_SOURCE     /* Needed for using struct ip_mreq with recent glibc */
-#define _DARWIN_C_SOURCE /* Needed for using IP_MULTICAST_TTL on OS X */
+
 #include "avformat.h"
 #include "avio_internal.h"
 #include "libavutil/parseutils.h"
+#include "libavutil/fifo.h"
 #include <unistd.h>
 #include "internal.h"
 #include "network.h"
 #include "os_support.h"
 #include "url.h"
+#include <pthread.h>
 #include <sys/time.h>
 
 #ifndef IPV6_ADD_MEMBERSHIP
@@ -51,6 +53,13 @@ typedef struct {
     struct sockaddr_storage dest_addr;
     int dest_addr_len;
     int is_connected;
+
+    /* Circular Buffer variables for use in UDP receive code */
+    int circular_buffer_size;
+    AVFifoBuffer *fifo;
+    int circular_buffer_available_max;
+    int circular_buffer_error;
+    pthread_t circular_buffer_thread;
 } UDPContext;
 
 #define UDP_TX_BUF_SIZE 32768
@@ -298,6 +307,66 @@ static int udp_get_file_handle(URLContext *h)
     return s->udp_fd;
 }
 
+static void *circular_buffer_task( void *_URLContext)
+{
+    URLContext *h = _URLContext;
+    UDPContext *s = h->priv_data;
+    fd_set rfds;
+    struct timeval tv;
+
+    for(;;) {
+        int left;
+        int ret;
+        int len;
+
+        if (url_interrupt_cb()) {
+            s->circular_buffer_error = EINTR;
+            return NULL;
+        }
+
+        FD_ZERO(&rfds);
+        FD_SET(s->udp_fd, &rfds);
+        tv.tv_sec = 1;
+        tv.tv_usec = 0;
+        ret = select(s->udp_fd + 1, &rfds, NULL, NULL, &tv);
+        if (ret < 0) {
+            if (ff_neterrno() == AVERROR(EINTR))
+                continue;
+            s->circular_buffer_error = EIO;
+            return NULL;
+        }
+
+        if (!(ret > 0 && FD_ISSET(s->udp_fd, &rfds)))
+            continue;
+
+        /* How much do we have left to the end of the buffer */
+        /* Whats the minimum we can read so that we dont comletely fill the buffer */
+        left = av_fifo_space(s->fifo);
+        left = FFMIN(left, s->fifo->end - s->fifo->wptr);
+
+        /* No Space left, error, what do we do now */
+        if( !left) {
+            av_log(h, AV_LOG_ERROR, "circular_buffer: OVERRUN\n");
+            s->circular_buffer_error = EIO;
+            return NULL;
+        }
+
+        len = recv(s->udp_fd, s->fifo->wptr, left, 0);
+        if (len < 0) {
+            if (ff_neterrno() != AVERROR(EAGAIN) && ff_neterrno() != AVERROR(EINTR)) {
+                s->circular_buffer_error = EIO;
+                return NULL;
+            }
+        }
+        s->fifo->wptr += len;
+        if (s->fifo->wptr >= s->fifo->end)
+            s->fifo->wptr = s->fifo->buffer;
+        s->fifo->wndx += len;
+    }
+
+    return NULL;
+}
+
 /* put it in UDP context */
 /* return non zero if error */
 static int udp_open(URLContext *h, const char *uri, int flags)
@@ -315,7 +384,7 @@ static int udp_open(URLContext *h, const char *uri, int flags)
     h->is_streamed = 1;
     h->max_packet_size = 1472;
 
-    is_output = (flags & AVIO_FLAG_WRITE);
+    is_output = !(flags & AVIO_FLAG_READ);
 
     s = av_mallocz(sizeof(UDPContext));
     if (!s)
@@ -325,10 +394,12 @@ static int udp_open(URLContext *h, const char *uri, int flags)
     s->ttl = 16;
     s->buffer_size = is_output ? UDP_TX_BUF_SIZE : UDP_MAX_PKT_SIZE;
 
+    s->circular_buffer_size = 7*188*4096;
+
     p = strchr(uri, '?');
     if (p) {
         if (av_find_info_tag(buf, sizeof(buf), "reuse", p)) {
-            const char *endptr=NULL;
+            char *endptr=NULL;
             s->reuse_socket = strtol(buf, &endptr, 10);
             /* assume if no digits were found it is a request to enable it */
             if (buf == endptr)
@@ -350,6 +421,9 @@ static int udp_open(URLContext *h, const char *uri, int flags)
         if (av_find_info_tag(buf, sizeof(buf), "connect", p)) {
             s->is_connected = strtol(buf, NULL, 10);
         }
+        if (av_find_info_tag(buf, sizeof(buf), "buf_size", p)) {
+            s->circular_buffer_size = strtol(buf, NULL, 10)*188;
+        }
     }
 
     /* fill the dest addr */
@@ -358,14 +432,14 @@ static int udp_open(URLContext *h, const char *uri, int flags)
     /* XXX: fix av_url_split */
     if (hostname[0] == '\0' || hostname[0] == '?') {
         /* only accepts null hostname if input */
-        if (flags & AVIO_FLAG_WRITE)
+        if (!(flags & AVIO_FLAG_READ))
             goto fail;
     } else {
         if (ff_udp_set_remote_url(h, uri) < 0)
             goto fail;
     }
 
-    if (s->is_multicast && !(h->flags & AVIO_FLAG_WRITE))
+    if (s->is_multicast && (h->flags & AVIO_FLAG_READ))
         s->local_port = port;
     udp_fd = udp_socket_create(s, &my_addr, &len);
     if (udp_fd < 0)
@@ -382,7 +456,7 @@ static int udp_open(URLContext *h, const char *uri, int flags)
 
     /* the bind is needed to give a port to the socket now */
     /* if multicast, try the multicast address bind first */
-    if (s->is_multicast && !(h->flags & AVIO_FLAG_WRITE)) {
+    if (s->is_multicast && (h->flags & AVIO_FLAG_READ)) {
         bind_ret = bind(udp_fd,(struct sockaddr *)&s->dest_addr, len);
     }
     /* bind to the local address if not multicast or if the multicast
@@ -395,7 +469,7 @@ static int udp_open(URLContext *h, const char *uri, int flags)
     s->local_port = udp_port(&my_addr, len);
 
     if (s->is_multicast) {
-        if (h->flags & AVIO_FLAG_WRITE) {
+        if (!(h->flags & AVIO_FLAG_READ)) {
             /* output */
             if (udp_set_multicast_ttl(udp_fd, s->ttl, (struct sockaddr *)&s->dest_addr) < 0)
                 goto fail;
@@ -431,10 +505,21 @@ static int udp_open(URLContext *h, const char *uri, int flags)
     }
 
     s->udp_fd = udp_fd;
+
+    if (!is_output && s->circular_buffer_size) {
+        /* start the task going */
+        s->fifo = av_fifo_alloc(s->circular_buffer_size);
+        if (pthread_create(&s->circular_buffer_thread, NULL, circular_buffer_task, h)) {
+            av_log(h, AV_LOG_ERROR, "pthread_create failed\n");
+            goto fail;
+        }
+    }
+
     return 0;
  fail:
     if (udp_fd >= 0)
         closesocket(udp_fd);
+        av_fifo_free(s->fifo);
     av_free(s);
     return AVERROR(EIO);
 }
@@ -443,6 +528,33 @@ static int udp_read(URLContext *h, uint8_t *buf, int size)
 {
     UDPContext *s = h->priv_data;
     int ret;
+    int avail;
+    int left;
+    fd_set rfds;
+    struct timeval tv;
+
+    if (s->fifo) {
+
+        do {
+            avail = av_fifo_size(s->fifo);
+            if (avail) { // >=size) {
+
+                // Maximum amount available
+                size = FFMIN( avail, size);
+                av_fifo_generic_read(s->fifo, buf, size, NULL);
+                return size;
+            }
+            else {
+                FD_ZERO(&rfds);
+                FD_SET(s->udp_fd, &rfds);
+                tv.tv_sec = 1;
+                tv.tv_usec = 0;
+                ret = select(s->udp_fd + 1, &rfds, NULL, NULL, &tv);
+                if (ret<0)
+                    return ret;
+            }
+        } while( 1);
+    }
 
     if (!(h->flags & AVIO_FLAG_NONBLOCK)) {
         ret = ff_network_wait_fd(s->udp_fd, 0);
@@ -450,6 +562,7 @@ static int udp_read(URLContext *h, uint8_t *buf, int size)
             return ret;
     }
     ret = recv(s->udp_fd, buf, size, 0);
+
     return ret < 0 ? ff_neterrno() : ret;
 }
 
@@ -478,9 +591,11 @@ static int udp_close(URLContext *h)
 {
     UDPContext *s = h->priv_data;
 
-    if (s->is_multicast && !(h->flags & AVIO_FLAG_WRITE))
+    if (s->is_multicast && (h->flags & AVIO_FLAG_READ))
         udp_leave_multicast_group(s->udp_fd, (struct sockaddr *)&s->dest_addr);
     closesocket(s->udp_fd);
+    av_log( h, AV_LOG_INFO, "circular_buffer_info max:%d%%\r\n", (s->circular_buffer_available_max*100)/s->circular_buffer_size);
+    av_fifo_free(s->fifo);
     av_free(s);
     return 0;
 }