X-Git-Url: https://git.sesse.net/?a=blobdiff_plain;f=libavformat%2Fudp.c;h=96d8b8ee11e4dfec2dcb7b59c7378c01b81a5e0a;hb=e0c8e517b6a0d82d8837f78425b62132443ecf15;hp=cf73d331e0472f20ecbab9c6c873a3a0ac98bef4;hpb=67d466d09b105b2b1d3d8da4c21d8975925741ae;p=ffmpeg

diff --git a/libavformat/udp.c b/libavformat/udp.c
index cf73d331e04..96d8b8ee11e 100644
--- a/libavformat/udp.c
+++ b/libavformat/udp.c
@@ -61,8 +61,13 @@
 #define IPPROTO_UDPLITE                                  136
 #endif
 
+#if HAVE_W32THREADS
+#undef HAVE_PTHREAD_CANCEL
+#define HAVE_PTHREAD_CANCEL 1
+#endif
+
 #if HAVE_PTHREAD_CANCEL
-#include <pthread.h>
+#include "libavutil/thread.h"
 #endif
 
 #ifndef IPV6_ADD_MEMBERSHIP
@@ -71,6 +76,7 @@
 #endif
 
 #define UDP_TX_BUF_SIZE 32768
+#define UDP_RX_BUF_SIZE 393216
 #define UDP_MAX_PKT_SIZE 65536
 #define UDP_HEADER_SIZE 8
 
@@ -274,7 +280,7 @@ static int udp_set_multicast_sources(URLContext *h,
         }
         return 0;
 #else
-        av_log(NULL, AV_LOG_ERROR,
+        av_log(h, AV_LOG_ERROR,
                "Setting multicast sources only supported for IPv4\n");
         return AVERROR(EINVAL);
 #endif
@@ -283,7 +289,7 @@ static int udp_set_multicast_sources(URLContext *h,
     for (i = 0; i < nb_sources; i++) {
         struct ip_mreq_source mreqs;
         if (sources[i].ss_family != AF_INET) {
-            av_log(NULL, AV_LOG_ERROR, "Source/block address %d is of incorrect protocol family\n", i + 1);
+            av_log(h, AV_LOG_ERROR, "Source/block address %d is of incorrect protocol family\n", i + 1);
             return AVERROR(EINVAL);
         }
 
@@ -298,9 +304,9 @@ static int udp_set_multicast_sources(URLContext *h,
                        include ? IP_ADD_SOURCE_MEMBERSHIP : IP_BLOCK_SOURCE,
                        (const void *)&mreqs, sizeof(mreqs)) < 0) {
             if (include)
-                ff_log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IP_ADD_SOURCE_MEMBERSHIP)");
+                ff_log_net_error(h, AV_LOG_ERROR, "setsockopt(IP_ADD_SOURCE_MEMBERSHIP)");
             else
-                ff_log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IP_BLOCK_SOURCE)");
+                ff_log_net_error(h, AV_LOG_ERROR, "setsockopt(IP_BLOCK_SOURCE)");
             return ff_neterrno();
         }
     }
@@ -519,14 +525,12 @@ static void *circular_buffer_task_tx( void *_URLContext)
 {
     URLContext *h = _URLContext;
     UDPContext *s = h->priv_data;
-    int old_cancelstate;
     int64_t target_timestamp = av_gettime_relative();
     int64_t start_timestamp = av_gettime_relative();
     int64_t sent_bits = 0;
     int64_t burst_interval = s->bitrate ? (s->burst_bits * 1000000 / s->bitrate) : 0;
     int64_t max_delay = s->bitrate ?  ((int64_t)h->max_packet_size * 8 * 1000000 / s->bitrate + 1) : 0;
 
-    pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancelstate);
     pthread_mutex_lock(&s->mutex);
 
     if (ff_socket_nonblock(s->udp_fd, 0) < 0) {
@@ -561,7 +565,6 @@ static void *circular_buffer_task_tx( void *_URLContext)
         av_fifo_generic_read(s->fifo, s->tmp, len, NULL);
 
         pthread_mutex_unlock(&s->mutex);
-        pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &old_cancelstate);
 
         if (s->bitrate) {
             timestamp = av_gettime_relative();
@@ -607,7 +610,6 @@ static void *circular_buffer_task_tx( void *_URLContext)
             }
         }
 
-        pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancelstate);
         pthread_mutex_lock(&s->mutex);
     }
 
@@ -636,7 +638,7 @@ static int udp_open(URLContext *h, const char *uri, int flags)
 
     is_output = !(flags & AVIO_FLAG_READ);
     if (s->buffer_size < 0)
-        s->buffer_size = is_output ? UDP_TX_BUF_SIZE : UDP_MAX_PKT_SIZE;
+        s->buffer_size = is_output ? UDP_TX_BUF_SIZE : UDP_RX_BUF_SIZE;
 
     if (s->sources) {
         if (ff_ip_parse_sources(h, s->sources, &s->filters) < 0)
@@ -648,9 +650,6 @@ static int udp_open(URLContext *h, const char *uri, int flags)
             goto fail;
     }
 
-    if (s->pkt_size > 0)
-        h->max_packet_size = s->pkt_size;
-
     p = strchr(uri, '?');
     if (p) {
         if (av_find_info_tag(buf, sizeof(buf), "reuse", p)) {
@@ -797,7 +796,7 @@ static int udp_open(URLContext *h, const char *uri, int flags)
      * receiving UDP packets from other sources aimed at the same UDP
      * port. This fails on windows. This makes sending to the same address
      * using sendto() fail, so only do it if we're opened in read-only mode. */
-    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
@@ -850,7 +849,7 @@ static int udp_open(URLContext *h, const char *uri, int flags)
             goto fail;
         }
     } else {
-        /* set udp recv buffer size to the requested value (default 64K) */
+        /* set udp recv buffer size to the requested value (default UDP_RX_BUF_SIZE) */
         tmp = s->buffer_size;
         if (setsockopt(udp_fd, SOL_SOCKET, SO_RCVBUF, &tmp, sizeof(tmp)) < 0) {
             ff_log_net_error(h, AV_LOG_WARNING, "setsockopt(SO_RECVBUF)");
@@ -861,7 +860,7 @@ static int udp_open(URLContext *h, const char *uri, int flags)
         } else {
             av_log(h, AV_LOG_DEBUG, "end receive buffer size reported is %d\n", tmp);
             if(tmp < s->buffer_size)
-                av_log(h, AV_LOG_WARNING, "attempted to set receive buffer to size %d but it only ended up set as %d", s->buffer_size, tmp);
+                av_log(h, AV_LOG_WARNING, "attempted to set receive buffer to size %d but it only ended up set as %d\n", s->buffer_size, tmp);
         }
 
         /* make the socket non-blocking */
@@ -978,9 +977,10 @@ static int udp_read(URLContext *h, uint8_t *buf, int size)
                 int64_t t = av_gettime() + 100000;
                 struct timespec tv = { .tv_sec  =  t / 1000000,
                                        .tv_nsec = (t % 1000000) * 1000 };
-                if (pthread_cond_timedwait(&s->cond, &s->mutex, &tv) < 0) {
+                int err = pthread_cond_timedwait(&s->cond, &s->mutex, &tv);
+                if (err) {
                     pthread_mutex_unlock(&s->mutex);
-                    return AVERROR(errno == ETIMEDOUT ? EAGAIN : errno);
+                    return AVERROR(err == ETIMEDOUT ? EAGAIN : err);
                 }
                 nonblock = 1;
             }
@@ -1071,8 +1071,17 @@ static int udp_close(URLContext *h)
     if (s->thread_started) {
         int ret;
         // Cancel only read, as write has been signaled as success to the user
-        if (h->flags & AVIO_FLAG_READ)
+        if (h->flags & AVIO_FLAG_READ) {
+#ifdef _WIN32
+            /* recvfrom() is not a cancellation point for win32, so we shutdown
+             * the socket and abort pending IO, subsequent recvfrom() calls
+             * will fail with WSAESHUTDOWN causing the thread to exit. */
+            shutdown(s->udp_fd, SD_RECEIVE);
+            CancelIoEx((HANDLE)(SOCKET)s->udp_fd, NULL);
+#else
             pthread_cancel(s->circular_buffer_thread);
+#endif
+        }
         ret = pthread_join(s->circular_buffer_thread, NULL);
         if (ret != 0)
             av_log(h, AV_LOG_ERROR, "pthread_join(): %s\n", strerror(ret));