diff options
Diffstat (limited to 'libavformat/udp.c')
-rw-r--r-- | libavformat/udp.c | 191 |
1 files changed, 184 insertions, 7 deletions
diff --git a/libavformat/udp.c b/libavformat/udp.c index e848ecd..e444a7d 100644 --- a/libavformat/udp.c +++ b/libavformat/udp.c @@ -2,20 +2,20 @@ * UDP prototype streaming system * Copyright (c) 2000, 2001, 2002 Fabrice Bellard * - * This file is part of Libav. + * This file is part of FFmpeg. * - * Libav is free software; you can redistribute it and/or + * FFmpeg is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public * License as published by the Free Software Foundation; either * version 2.1 of the License, or (at your option) any later version. * - * Libav is distributed in the hope that it will be useful, + * FFmpeg is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU * Lesser General Public License for more details. * * You should have received a copy of the GNU Lesser General Public - * License along with Libav; if not, write to the Free Software + * License along with FFmpeg; if not, write to the Free Software * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA */ @@ -29,17 +29,26 @@ #include "avformat.h" #include "avio_internal.h" #include "libavutil/parseutils.h" +#include "libavutil/fifo.h" +#include "libavutil/intreadwrite.h" #include "libavutil/avstring.h" #include "internal.h" #include "network.h" #include "os_support.h" #include "url.h" +#if HAVE_PTHREAD_CANCEL +#include <pthread.h> +#endif + #ifndef IPV6_ADD_MEMBERSHIP #define IPV6_ADD_MEMBERSHIP IPV6_JOIN_GROUP #define IPV6_DROP_MEMBERSHIP IPV6_LEAVE_GROUP #endif +#define UDP_TX_BUF_SIZE 32768 +#define UDP_MAX_PKT_SIZE 65536 + typedef struct { int udp_fd; int ttl; @@ -47,13 +56,24 @@ typedef struct { int is_multicast; int local_port; int reuse_socket; + int overrun_nonfatal; struct sockaddr_storage dest_addr; int dest_addr_len; int is_connected; -} UDPContext; -#define UDP_TX_BUF_SIZE 32768 -#define UDP_MAX_PKT_SIZE 65536 + /* Circular Buffer variables for use in UDP receive code */ + int circular_buffer_size; + AVFifoBuffer *fifo; + int circular_buffer_error; +#if HAVE_PTHREAD_CANCEL + pthread_t circular_buffer_thread; + pthread_mutex_t mutex; + pthread_cond_t cond; + int thread_started; +#endif + uint8_t tmp[UDP_MAX_PKT_SIZE+4]; + int remaining_in_dg; +} UDPContext; static void log_net_error(void *ctx, int level, const char* prefix) { @@ -317,6 +337,7 @@ static int udp_port(struct sockaddr_storage *addr, int addr_len) * 'localport=n' : set the local port * 'pkt_size=n' : set max packet size * 'reuse=1' : enable reusing the socket + * 'overrun_nonfatal=1': survive in case of circular buffer overrun * * @param h media file context * @param uri of the remote server @@ -378,6 +399,61 @@ static int udp_get_file_handle(URLContext *h) return s->udp_fd; } +#if HAVE_PTHREAD_CANCEL +static void *circular_buffer_task( void *_URLContext) +{ + URLContext *h = _URLContext; + UDPContext *s = h->priv_data; + int old_cancelstate; + + pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancelstate); + ff_socket_nonblock(s->udp_fd, 0); + pthread_mutex_lock(&s->mutex); + while(1) { + int len; + + pthread_mutex_unlock(&s->mutex); + /* Blocking operations are always cancellation points; + see "General Information" / "Thread Cancelation Overview" + in Single Unix. */ + pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &old_cancelstate); + len = recv(s->udp_fd, s->tmp+4, sizeof(s->tmp)-4, 0); + pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancelstate); + pthread_mutex_lock(&s->mutex); + if (len < 0) { + if (ff_neterrno() != AVERROR(EAGAIN) && ff_neterrno() != AVERROR(EINTR)) { + s->circular_buffer_error = ff_neterrno(); + goto end; + } + continue; + } + AV_WL32(s->tmp, len); + + if(av_fifo_space(s->fifo) < len + 4) { + /* No Space left */ + if (s->overrun_nonfatal) { + av_log(h, AV_LOG_WARNING, "Circular buffer overrun. " + "Surviving due to overrun_nonfatal option\n"); + continue; + } else { + av_log(h, AV_LOG_ERROR, "Circular buffer overrun. " + "To avoid, increase fifo_size URL option. " + "To survive in such case, use overrun_nonfatal option\n"); + s->circular_buffer_error = AVERROR(EIO); + goto end; + } + } + av_fifo_generic_write(s->fifo, s->tmp, len+4, NULL); + pthread_cond_signal(&s->cond); + } + +end: + pthread_cond_signal(&s->cond); + pthread_mutex_unlock(&s->mutex); + return NULL; +} +#endif + /* put it in UDP context */ /* return non zero if error */ static int udp_open(URLContext *h, const char *uri, int flags) @@ -402,6 +478,8 @@ 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)) { @@ -412,6 +490,13 @@ static int udp_open(URLContext *h, const char *uri, int flags) s->reuse_socket = 1; reuse_specified = 1; } + if (av_find_info_tag(buf, sizeof(buf), "overrun_nonfatal", p)) { + char *endptr = NULL; + s->overrun_nonfatal = strtol(buf, &endptr, 10); + /* assume if no digits were found it is a request to enable it */ + if (buf == endptr) + s->overrun_nonfatal = 1; + } if (av_find_info_tag(buf, sizeof(buf), "ttl", p)) { s->ttl = strtol(buf, NULL, 10); } @@ -427,6 +512,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), "fifo_size", p)) { + s->circular_buffer_size = strtol(buf, NULL, 10)*188; + } if (av_find_info_tag(buf, sizeof(buf), "localaddr", p)) { av_strlcpy(localaddr, buf, sizeof(localaddr)); } @@ -552,10 +640,43 @@ static int udp_open(URLContext *h, const char *uri, int flags) av_free(sources[i]); s->udp_fd = udp_fd; + +#if HAVE_PTHREAD_CANCEL + if (!is_output && s->circular_buffer_size) { + int ret; + + /* start the task going */ + s->fifo = av_fifo_alloc(s->circular_buffer_size); + ret = pthread_mutex_init(&s->mutex, NULL); + if (ret != 0) { + av_log(h, AV_LOG_ERROR, "pthread_mutex_init failed : %s\n", strerror(ret)); + goto fail; + } + ret = pthread_cond_init(&s->cond, NULL); + if (ret != 0) { + av_log(h, AV_LOG_ERROR, "pthread_cond_init failed : %s\n", strerror(ret)); + goto cond_fail; + } + ret = pthread_create(&s->circular_buffer_thread, NULL, circular_buffer_task, h); + if (ret != 0) { + av_log(h, AV_LOG_ERROR, "pthread_create failed : %s\n", strerror(ret)); + goto thread_fail; + } + s->thread_started = 1; + } +#endif + return 0; +#if HAVE_PTHREAD_CANCEL + thread_fail: + pthread_cond_destroy(&s->cond); + cond_fail: + pthread_mutex_destroy(&s->mutex); +#endif fail: if (udp_fd >= 0) closesocket(udp_fd); + av_fifo_free(s->fifo); for (i = 0; i < num_sources; i++) av_free(sources[i]); return AVERROR(EIO); @@ -565,6 +686,48 @@ static int udp_read(URLContext *h, uint8_t *buf, int size) { UDPContext *s = h->priv_data; int ret; + int avail, nonblock = h->flags & AVIO_FLAG_NONBLOCK; + +#if HAVE_PTHREAD_CANCEL + if (s->fifo) { + pthread_mutex_lock(&s->mutex); + do { + avail = av_fifo_size(s->fifo); + if (avail) { // >=size) { + uint8_t tmp[4]; + + av_fifo_generic_read(s->fifo, tmp, 4, NULL); + avail= AV_RL32(tmp); + if(avail > size){ + av_log(h, AV_LOG_WARNING, "Part of datagram lost due to insufficient buffer size\n"); + avail= size; + } + + av_fifo_generic_read(s->fifo, buf, avail, NULL); + av_fifo_drain(s->fifo, AV_RL32(tmp) - avail); + pthread_mutex_unlock(&s->mutex); + return avail; + } else if(s->circular_buffer_error){ + int err = s->circular_buffer_error; + pthread_mutex_unlock(&s->mutex); + return err; + } else if(nonblock) { + pthread_mutex_unlock(&s->mutex); + return AVERROR(EAGAIN); + } + else { + /* FIXME: using the monotonic clock would be better, + but it does not exist on all supported platforms. */ + 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) + return AVERROR(errno == ETIMEDOUT ? EAGAIN : errno); + nonblock = 1; + } + } while( 1); + } +#endif if (!(h->flags & AVIO_FLAG_NONBLOCK)) { ret = ff_network_wait_fd(s->udp_fd, 0); @@ -572,6 +735,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; } @@ -599,10 +763,23 @@ static int udp_write(URLContext *h, const uint8_t *buf, int size) static int udp_close(URLContext *h) { UDPContext *s = h->priv_data; + int ret; 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); +#if HAVE_PTHREAD_CANCEL + if (s->thread_started) { + pthread_cancel(s->circular_buffer_thread); + ret = pthread_join(s->circular_buffer_thread, NULL); + if (ret != 0) + av_log(h, AV_LOG_ERROR, "pthread_join(): %s\n", strerror(ret)); + } + + pthread_mutex_destroy(&s->mutex); + pthread_cond_destroy(&s->cond); +#endif + av_fifo_free(s->fifo); return 0; } |