diff options
Diffstat (limited to 'libavformat/udp.c')
-rw-r--r-- | libavformat/udp.c | 192 |
1 files changed, 185 insertions, 7 deletions
diff --git a/libavformat/udp.c b/libavformat/udp.c index 6571ab5..9387e21 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,12 +29,19 @@ #include "avformat.h" #include "avio_internal.h" #include "libavutil/parseutils.h" +#include "libavutil/fifo.h" +#include "libavutil/intreadwrite.h" #include "libavutil/avstring.h" #include <unistd.h> #include "internal.h" #include "network.h" #include "os_support.h" #include "url.h" + +#if HAVE_PTHREADS +#include <pthread.h> +#endif + #include <sys/time.h> #ifndef IPV6_ADD_MEMBERSHIP @@ -42,6 +49,9 @@ #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; @@ -49,13 +59,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_PTHREADS + 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 int udp_set_multicast_ttl(int sockfd, int mcastTTL, struct sockaddr *addr) @@ -238,6 +259,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 @@ -299,6 +321,61 @@ static int udp_get_file_handle(URLContext *h) return s->udp_fd; } +#if HAVE_PTHREADS +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) @@ -321,6 +398,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)) { @@ -331,6 +410,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); } @@ -346,6 +432,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)); } @@ -436,10 +525,43 @@ static int udp_open(URLContext *h, const char *uri, int flags) } s->udp_fd = udp_fd; + +#if HAVE_PTHREADS + 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_PTHREADS + 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); return AVERROR(EIO); } @@ -447,6 +569,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_PTHREADS + 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); @@ -454,6 +618,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; } @@ -481,10 +646,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); + av_fifo_free(s->fifo); +#if HAVE_PTHREADS + 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 return 0; } |