31 #include "libavutil/parseutils.h"
32 #include "libavutil/fifo.h"
33 #include "libavutil/intreadwrite.h"
34 #include "libavutil/avstring.h"
35 #include "libavutil/opt.h"
36 #include "libavutil/log.h"
37 #include "libavutil/time.h"
43 #if HAVE_PTHREAD_CANCEL
47 #ifndef HAVE_PTHREAD_CANCEL
48 #define HAVE_PTHREAD_CANCEL 0
51 #ifndef IPV6_ADD_MEMBERSHIP
52 #define IPV6_ADD_MEMBERSHIP IPV6_JOIN_GROUP
53 #define IPV6_DROP_MEMBERSHIP IPV6_LEAVE_GROUP
56 #define UDP_TX_BUF_SIZE 32768
57 #define UDP_MAX_PKT_SIZE 65536
76 #if HAVE_PTHREAD_CANCEL
89 #define OFFSET(x) offsetof(UDPContext, x)
90 #define D AV_OPT_FLAG_DECODING_PARAM
91 #define E AV_OPT_FLAG_ENCODING_PARAM
93 {
"buffer_size",
"Socket buffer size in bytes",
OFFSET(buffer_size),
AV_OPT_TYPE_INT, {.i64 = 0}, 0, INT_MAX,
D|
E },
97 {
"reuse",
"Explicitly allow or disallow reusing UDP sockets",
OFFSET(reuse_socket),
AV_OPT_TYPE_INT, {.i64 = 0}, 0, 1,
D|
E },
98 {
"ttl",
"Set the time to live value (for multicast only)",
OFFSET(ttl),
AV_OPT_TYPE_INT, {.i64 = 16}, 0, INT_MAX,
E },
99 {
"connect",
"Should connect() be called on socket",
OFFSET(is_connected),
AV_OPT_TYPE_INT, {.i64 = 0}, 0, 1,
D|
E },
101 {
"fifo_size",
"Set the UDP receiving circular buffer size, expressed as a number of packets with size of 188 bytes",
OFFSET(circular_buffer_size),
AV_OPT_TYPE_INT, {.i64 = 7*4096}, 0, INT_MAX,
D },
102 {
"overrun_nonfatal",
"Survive in case of UDP receiving circular buffer overrun",
OFFSET(overrun_nonfatal),
AV_OPT_TYPE_INT, {.i64 = 0}, 0, 1,
D },
103 {
"timeout",
"In read mode: if no data arrived in more than this time interval, raise error",
OFFSET(timeout),
AV_OPT_TYPE_INT, {.i64 = 0}, 0, INT_MAX,
D },
118 av_log(ctx, level,
"%s: %s\n", prefix, errbuf);
122 struct sockaddr *addr)
124 #ifdef IP_MULTICAST_TTL
125 if (addr->sa_family == AF_INET) {
126 if (setsockopt(sockfd, IPPROTO_IP, IP_MULTICAST_TTL, &mcastTTL,
sizeof(mcastTTL)) < 0) {
132 #if defined(IPPROTO_IPV6) && defined(IPV6_MULTICAST_HOPS)
133 if (addr->sa_family == AF_INET6) {
134 if (setsockopt(sockfd, IPPROTO_IPV6, IPV6_MULTICAST_HOPS, &mcastTTL,
sizeof(mcastTTL)) < 0) {
145 #ifdef IP_ADD_MEMBERSHIP
146 if (addr->sa_family == AF_INET) {
149 mreq.imr_multiaddr.s_addr = ((
struct sockaddr_in *)addr)->sin_addr.s_addr;
150 mreq.imr_interface.s_addr= INADDR_ANY;
151 if (setsockopt(sockfd, IPPROTO_IP, IP_ADD_MEMBERSHIP, (
const void *)&mreq,
sizeof(mreq)) < 0) {
157 #if HAVE_STRUCT_IPV6_MREQ && defined(IPPROTO_IPV6)
158 if (addr->sa_family == AF_INET6) {
159 struct ipv6_mreq mreq6;
161 memcpy(&mreq6.ipv6mr_multiaddr, &(((
struct sockaddr_in6 *)addr)->sin6_addr),
sizeof(
struct in6_addr));
162 mreq6.ipv6mr_interface= 0;
174 #ifdef IP_DROP_MEMBERSHIP
175 if (addr->sa_family == AF_INET) {
178 mreq.imr_multiaddr.s_addr = ((
struct sockaddr_in *)addr)->sin_addr.s_addr;
179 mreq.imr_interface.s_addr= INADDR_ANY;
180 if (setsockopt(sockfd, IPPROTO_IP, IP_DROP_MEMBERSHIP, (
const void *)&mreq,
sizeof(mreq)) < 0) {
186 #if HAVE_STRUCT_IPV6_MREQ && defined(IPPROTO_IPV6)
187 if (addr->sa_family == AF_INET6) {
188 struct ipv6_mreq mreq6;
190 memcpy(&mreq6.ipv6mr_multiaddr, &(((
struct sockaddr_in6 *)addr)->sin6_addr),
sizeof(
struct in6_addr));
191 mreq6.ipv6mr_interface= 0;
207 const char *node = 0, *service =
"0";
210 snprintf(sport,
sizeof(sport),
"%d", port);
213 if ((hostname) && (hostname[0] !=
'\0') && (hostname[0] !=
'?')) {
228 int addr_len,
char **sources,
229 int nb_sources,
int include)
231 #if HAVE_STRUCT_GROUP_SOURCE_REQ && defined(MCAST_BLOCK_SOURCE) && !defined(_WIN32)
235 for (i = 0; i < nb_sources; i++) {
236 struct group_source_req mreqs;
237 int level = addr->sa_family == AF_INET ? IPPROTO_IP : IPPROTO_IPV6;
239 SOCK_DGRAM, AF_UNSPEC,
244 mreqs.gsr_interface = 0;
245 memcpy(&mreqs.gsr_group, addr, addr_len);
249 if (setsockopt(sockfd, level,
250 include ? MCAST_JOIN_SOURCE_GROUP : MCAST_BLOCK_SOURCE,
251 (
const void *)&mreqs,
sizeof(mreqs)) < 0) {
259 #elif HAVE_STRUCT_IP_MREQ_SOURCE && defined(IP_BLOCK_SOURCE)
261 if (addr->sa_family != AF_INET) {
263 "Setting multicast sources only supported for IPv4\n");
266 for (i = 0; i < nb_sources; i++) {
267 struct ip_mreq_source mreqs;
269 SOCK_DGRAM, AF_UNSPEC,
273 if (sourceaddr->
ai_addr->sa_family != AF_INET) {
280 mreqs.imr_multiaddr.s_addr = ((
struct sockaddr_in *)addr)->sin_addr.s_addr;
281 mreqs.imr_interface.s_addr = INADDR_ANY;
282 mreqs.imr_sourceaddr.s_addr = ((
struct sockaddr_in *)sourceaddr->
ai_addr)->sin_addr.s_addr;
285 if (setsockopt(sockfd, IPPROTO_IP,
286 include ? IP_ADD_SOURCE_MEMBERSHIP : IP_BLOCK_SOURCE,
287 (
const void *)&mreqs,
sizeof(mreqs)) < 0) {
301 const char *hostname,
int port)
307 if (res0 == 0)
return AVERROR(EIO);
316 socklen_t *addr_len,
const char *localaddr)
320 int family = AF_UNSPEC;
322 if (((
struct sockaddr *) &s->
dest_addr)->sa_family)
323 family = ((
struct sockaddr *) &s->
dest_addr)->sa_family;
328 for (res = res0;
res; res=res->ai_next) {
329 udp_fd =
ff_socket(res->ai_family, SOCK_DGRAM, 0);
330 if (udp_fd != -1)
break;
337 memcpy(addr, res->ai_addr, res->ai_addrlen);
338 *addr_len = res->ai_addrlen;
354 char sbuf[
sizeof(int)*3+1];
362 return strtol(sbuf, NULL, 10);
385 char hostname[256],
buf[10];
389 av_url_split(NULL, 0, NULL, 0, hostname,
sizeof(hostname), &port, NULL, 0, uri);
397 p = strchr(uri,
'?');
438 #if HAVE_PTHREAD_CANCEL
439 static void *circular_buffer_task(
void *_URLContext)
445 pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancelstate);
459 pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &old_cancelstate);
461 pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancelstate);
476 "Surviving due to overrun_nonfatal option\n");
480 "To avoid, increase fifo_size URL option. "
481 "To survive in such case, use overrun_nonfatal option\n");
504 char *next = strchr(source_start,
',');
507 sources[*num_sources] =
av_strdup(source_start);
508 if (!sources[*num_sources])
510 source_start = next + 1;
512 if (*num_sources >= max_sources || !next)
522 char hostname[1024], localaddr[1024] =
"";
523 int port, udp_fd = -1, tmp, bind_ret = -1;
530 int reuse_specified = 0;
531 int i, num_include_sources = 0, num_exclude_sources = 0;
532 char *include_sources[32], *exclude_sources[32];
540 p = strchr(uri,
'?');
558 "'overrun_nonfatal' option was set but it is not supported "
559 "on this build (pthread support is required)\n");
562 s->
ttl = strtol(buf, NULL, 10);
580 "'circular_buffer_size' option was set but it is not supported "
581 "on this build (pthread support is required)\n");
584 av_strlcpy(localaddr, buf,
sizeof(localaddr));
597 s->
timeout = strtol(buf, NULL, 10);
609 av_url_split(NULL, 0, NULL, 0, hostname,
sizeof(hostname), &port, NULL, 0, uri);
612 if (hostname[0] ==
'\0' || hostname[0] ==
'?') {
641 bind_ret = bind(udp_fd,(
struct sockaddr *)&s->
dest_addr, len);
646 if (bind_ret < 0 && bind(udp_fd,(
struct sockaddr *)&my_addr, len) < 0) {
651 len =
sizeof(my_addr);
652 getsockname(udp_fd, (
struct sockaddr *)&my_addr, &len);
656 if (h->
flags & AVIO_FLAG_WRITE) {
663 if (num_include_sources && num_exclude_sources) {
664 av_log(h,
AV_LOG_ERROR,
"Simultaneously including and excluding multicast sources is not supported\n");
667 if (num_include_sources) {
674 if (num_exclude_sources) {
684 if (setsockopt(udp_fd, SOL_SOCKET, SO_SNDBUF, &tmp,
sizeof(tmp)) < 0) {
692 if (setsockopt(udp_fd, SOL_SOCKET, SO_RCVBUF, &tmp,
sizeof(tmp)) < 0) {
705 for (i = 0; i < num_include_sources; i++)
707 for (i = 0; i < num_exclude_sources; i++)
712 #if HAVE_PTHREAD_CANCEL
728 ret =
pthread_create(&s->circular_buffer_thread, NULL, circular_buffer_task, h);
733 s->thread_started = 1;
738 #if HAVE_PTHREAD_CANCEL
748 for (i = 0; i < num_include_sources; i++)
750 for (i = 0; i < num_exclude_sources; i++)
761 #if HAVE_PTHREAD_CANCEL
784 }
else if(nonblock) {
792 struct timespec tv = { .tv_sec = t / 1000000,
793 .tv_nsec = (t % 1000000) * 1000 };
794 if (pthread_cond_timedwait(&s->cond, &s->mutex, &tv) < 0) {
796 return AVERROR(errno == ETIMEDOUT ? EAGAIN : errno);
809 ret = recv(s->
udp_fd, buf, size, 0);
826 ret = sendto (s->
udp_fd, buf, size, 0,
830 ret = send(s->
udp_fd, buf, size, 0);
843 #if HAVE_PTHREAD_CANCEL
844 if (s->thread_started) {
845 pthread_cancel(s->circular_buffer_thread);
865 .priv_data_class = &udp_context_class,
static int udp_open(URLContext *h, const char *uri, int flags)
void av_url_split(char *proto, int proto_size, char *authorization, int authorization_size, char *hostname, int hostname_size, int *port_ptr, char *path, int path_size, const char *url)
Split a URL string into components.
static void log_net_error(void *ctx, int level, const char *prefix)
static av_always_inline int pthread_mutex_destroy(pthread_mutex_t *mutex)
#define URL_PROTOCOL_FLAG_NETWORK
#define HAVE_PTHREAD_CANCEL
#define AV_LOG_WARNING
Something somehow does not look correct.
#define LIBAVUTIL_VERSION_INT
char * av_strdup(const char *s) av_malloc_attrib
Duplicate the string s.
static int udp_set_multicast_sources(int sockfd, struct sockaddr *addr, int addr_len, char **sources, int nb_sources, int include)
int is_streamed
true if streamed (no seek possible), default = false
static int parse_source_list(char *buf, char **sources, int *num_sources, int max_sources)
int64_t rw_timeout
maximum time to wait for (network) read/write operation completion, in mcs
void av_log(void *avcl, int level, const char *fmt,...) av_printf_format(3
Send the specified message to the log if the level is less than or equal to the current av_log_level...
static av_always_inline int pthread_cond_destroy(pthread_cond_t *cond)
int ff_socket(int af, int type, int proto)
void av_freep(void *ptr)
Free a memory block which has been allocated with av_malloc(z)() or av_realloc() and set the pointer ...
static struct addrinfo * udp_resolve_host(const char *hostname, int port, int type, int family, int flags)
int av_fifo_size(AVFifoBuffer *f)
Return the amount of data in bytes in the AVFifoBuffer, that is the amount of data you can read from ...
const char * av_default_item_name(void *ctx)
Return the context name.
const char * class_name
The name of the class; usually it is the same name as the context structure type to which the AVClass...
miscellaneous OS support macros and functions.
int ff_udp_get_local_port(URLContext *h)
Return the local port used by the UDP connection.
static av_cold int end(AVCodecContext *avctx)
int av_find_info_tag(char *arg, int arg_size, const char *tag1, const char *info)
Attempt to find a specific tag in a URL.
static int udp_port(struct sockaddr_storage *addr, int addr_len)
static av_always_inline int pthread_cond_signal(pthread_cond_t *cond)
void av_fifo_free(AVFifoBuffer *f)
Free an AVFifoBuffer.
static int udp_close(URLContext *h)
const OptionDef options[]
void av_fifo_drain(AVFifoBuffer *f, int size)
Read and discard the specified amount of data from an AVFifoBuffer.
#define AV_LOG_ERROR
Something went wrong and cannot losslessly be recovered.
int ff_udp_set_remote_url(URLContext *h, const char *uri)
If no filename is given to av_open_input_file because you want to get the local port first...
int ff_is_multicast_address(struct sockaddr *addr)
size_t av_strlcpy(char *dst, const char *src, size_t size)
Copy the string src to dst, but no more than size - 1 bytes, and null-terminate dst.
int av_fifo_generic_write(AVFifoBuffer *f, void *src, int size, int(*func)(void *, void *, int))
Feed data from a user-supplied callback to an AVFifoBuffer.
#define FF_ARRAY_ELEMS(a)
int64_t av_gettime(void)
Get the current time in microseconds.
static int udp_socket_create(UDPContext *s, struct sockaddr_storage *addr, socklen_t *addr_len, const char *localaddr)
uint8_t tmp[UDP_MAX_PKT_SIZE+4]
static av_always_inline int pthread_join(pthread_t thread, void **value_ptr)
static av_always_inline int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutexattr_t *attr)
static int udp_get_file_handle(URLContext *h)
Return the udp file handle for select() usage to wait for several RTP streams at the same time...
static pthread_mutex_t * mutex
int ff_socket_nonblock(int socket, int enable)
static int udp_leave_multicast_group(int sockfd, struct sockaddr *addr)
static av_always_inline int pthread_create(pthread_t *thread, const pthread_attr_t *attr, void *(*start_routine)(void *), void *arg)
#define IPV6_ADD_MEMBERSHIP
#define IPV6_DROP_MEMBERSHIP
static int udp_set_url(struct sockaddr_storage *addr, const char *hostname, int port)
Describe the class of an AVClass context structure.
AVFifoBuffer * av_fifo_alloc(unsigned int size)
Initialize an AVFifoBuffer.
int av_fifo_generic_read(AVFifoBuffer *f, void *dest, int buf_size, void(*func)(void *, void *, int))
Feed data from an AVFifoBuffer to a user-supplied callback.
static const AVClass udp_context_class
int av_strerror(int errnum, char *errbuf, size_t errbuf_size)
Put a description of the AVERROR code errnum in errbuf.
URLProtocol ff_udp_protocol
struct sockaddr_storage dest_addr
static int udp_read(URLContext *h, uint8_t *buf, int size)
static av_always_inline int pthread_cond_init(pthread_cond_t *cond, const pthread_condattr_t *attr)
static int udp_set_multicast_ttl(int sockfd, int mcastTTL, struct sockaddr *addr)
int circular_buffer_error
static av_always_inline int pthread_mutex_unlock(pthread_mutex_t *mutex)
int max_packet_size
if non zero, the stream is packetized with this max packet size
int ff_network_wait_fd(int fd, int write)
unbuffered private I/O API
static av_always_inline int pthread_mutex_lock(pthread_mutex_t *mutex)
static int udp_join_multicast_group(int sockfd, struct sockaddr *addr)
struct sockaddr * ai_addr
int av_fifo_space(AVFifoBuffer *f)
Return the amount of space in bytes in the AVFifoBuffer, that is the amount of data you can write int...
static int udp_write(URLContext *h, const uint8_t *buf, int size)