27 #define _DEFAULT_SOURCE
47 #include "TargetConditionals.h"
56 #define UDPLITE_SEND_CSCOV 10
57 #define UDPLITE_RECV_CSCOV 11
60 #ifndef IPPROTO_UDPLITE
61 #define IPPROTO_UDPLITE 136
65 #undef HAVE_PTHREAD_CANCEL
66 #define HAVE_PTHREAD_CANCEL 1
69 #if HAVE_PTHREAD_CANCEL
73 #ifndef IPV6_ADD_MEMBERSHIP
74 #define IPV6_ADD_MEMBERSHIP IPV6_JOIN_GROUP
75 #define IPV6_DROP_MEMBERSHIP IPV6_LEAVE_GROUP
78 #define UDP_TX_BUF_SIZE 32768
79 #define UDP_RX_BUF_SIZE 393216
80 #define UDP_MAX_PKT_SIZE 65536
81 #define UDP_HEADER_SIZE 8
106 #if HAVE_PTHREAD_CANCEL
122 #define OFFSET(x) offsetof(UDPContext, x)
123 #define D AV_OPT_FLAG_DECODING_PARAM
124 #define E AV_OPT_FLAG_ENCODING_PARAM
126 {
"buffer_size",
"System data size (in bytes)",
OFFSET(buffer_size),
AV_OPT_TYPE_INT, { .i64 = -1 }, -1, INT_MAX, .flags =
D|
E },
128 {
"burst_bits",
"Max length of bursts in bits (when using bitrate)",
OFFSET(burst_bits),
AV_OPT_TYPE_INT64, { .i64 = 0 }, 0, INT64_MAX, .flags =
E },
132 {
"udplite_coverage",
"choose UDPLite head size which should be validated by checksum",
OFFSET(udplite_coverage),
AV_OPT_TYPE_INT, {.i64 = 0}, 0, INT_MAX,
D|
E },
133 {
"pkt_size",
"Maximum UDP packet size",
OFFSET(pkt_size),
AV_OPT_TYPE_INT, { .i64 = 1472 }, -1, INT_MAX, .flags =
D|
E },
134 {
"reuse",
"explicitly allow reusing UDP sockets",
OFFSET(reuse_socket),
AV_OPT_TYPE_BOOL, { .i64 = -1 }, -1, 1,
D|
E },
135 {
"reuse_socket",
"explicitly allow reusing UDP sockets",
OFFSET(reuse_socket),
AV_OPT_TYPE_BOOL, { .i64 = -1 }, -1, 1, .flags =
D|
E },
136 {
"broadcast",
"explicitly allow or disallow broadcast destination",
OFFSET(is_broadcast),
AV_OPT_TYPE_BOOL, { .i64 = 0 }, 0, 1,
E },
138 {
"connect",
"set if connect() should be called on socket",
OFFSET(is_connected),
AV_OPT_TYPE_BOOL, { .i64 = 0 }, 0, 1, .flags =
D|
E },
139 {
"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 },
140 {
"overrun_nonfatal",
"survive in case of UDP receiving circular buffer overrun",
OFFSET(overrun_nonfatal),
AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1,
D },
141 {
"timeout",
"set raise error timeout, in microseconds (only in read mode)",
OFFSET(timeout),
AV_OPT_TYPE_INT, {.i64 = 0}, 0, INT_MAX,
D },
162 struct sockaddr *addr)
164 #ifdef IP_MULTICAST_TTL
165 if (addr->sa_family == AF_INET) {
166 if (setsockopt(sockfd, IPPROTO_IP, IP_MULTICAST_TTL, &mcastTTL,
sizeof(mcastTTL)) < 0) {
172 #if defined(IPPROTO_IPV6) && defined(IPV6_MULTICAST_HOPS)
173 if (addr->sa_family == AF_INET6) {
174 if (setsockopt(sockfd, IPPROTO_IPV6, IPV6_MULTICAST_HOPS, &mcastTTL,
sizeof(mcastTTL)) < 0) {
185 #ifdef IP_ADD_MEMBERSHIP
186 if (addr->sa_family == AF_INET) {
189 mreq.imr_multiaddr.s_addr = ((
struct sockaddr_in *)addr)->sin_addr.s_addr;
191 mreq.imr_interface= ((
struct sockaddr_in *)local_addr)->sin_addr;
193 mreq.imr_interface.s_addr = INADDR_ANY;
194 if (setsockopt(sockfd, IPPROTO_IP, IP_ADD_MEMBERSHIP, (
const void *)&mreq,
sizeof(mreq)) < 0) {
200 #if HAVE_STRUCT_IPV6_MREQ && defined(IPPROTO_IPV6)
201 if (addr->sa_family == AF_INET6) {
202 struct ipv6_mreq mreq6;
204 memcpy(&mreq6.ipv6mr_multiaddr, &(((
struct sockaddr_in6 *)addr)->sin6_addr),
sizeof(
struct in6_addr));
206 mreq6.ipv6mr_interface = 0;
218 #ifdef IP_DROP_MEMBERSHIP
219 if (addr->sa_family == AF_INET) {
222 mreq.imr_multiaddr.s_addr = ((
struct sockaddr_in *)addr)->sin_addr.s_addr;
224 mreq.imr_interface = ((
struct sockaddr_in *)local_addr)->sin_addr;
226 mreq.imr_interface.s_addr = INADDR_ANY;
227 if (setsockopt(sockfd, IPPROTO_IP, IP_DROP_MEMBERSHIP, (
const void *)&mreq,
sizeof(mreq)) < 0) {
233 #if HAVE_STRUCT_IPV6_MREQ && defined(IPPROTO_IPV6)
234 if (addr->sa_family == AF_INET6) {
235 struct ipv6_mreq mreq6;
237 memcpy(&mreq6.ipv6mr_multiaddr, &(((
struct sockaddr_in6 *)addr)->sin6_addr),
sizeof(
struct in6_addr));
239 mreq6.ipv6mr_interface = 0;
250 int sockfd,
struct sockaddr *addr,
253 int nb_sources,
int include)
256 if (addr->sa_family != AF_INET) {
257 #if HAVE_STRUCT_GROUP_SOURCE_REQ && defined(MCAST_BLOCK_SOURCE)
262 for (
i = 0;
i < nb_sources;
i++) {
263 struct group_source_req mreqs;
264 int level = addr->sa_family == AF_INET ? IPPROTO_IP : IPPROTO_IPV6;
267 mreqs.gsr_interface = 0;
268 memcpy(&mreqs.gsr_group, addr, addr_len);
269 memcpy(&mreqs.gsr_source, &sources[
i],
sizeof(*sources));
271 if (setsockopt(sockfd,
level,
272 include ? MCAST_JOIN_SOURCE_GROUP : MCAST_BLOCK_SOURCE,
273 (
const void *)&mreqs,
sizeof(mreqs)) < 0) {
284 "Setting multicast sources only supported for IPv4\n");
288 #if HAVE_STRUCT_IP_MREQ_SOURCE && defined(IP_BLOCK_SOURCE)
289 for (
i = 0;
i < nb_sources;
i++) {
290 struct ip_mreq_source mreqs;
291 if (sources[
i].ss_family != AF_INET) {
296 mreqs.imr_multiaddr.s_addr = ((
struct sockaddr_in *)addr)->sin_addr.s_addr;
298 mreqs.imr_interface = ((
struct sockaddr_in *)local_addr)->sin_addr;
300 mreqs.imr_interface.s_addr = INADDR_ANY;
301 mreqs.imr_sourceaddr.s_addr = ((
struct sockaddr_in *)&sources[
i])->sin_addr.s_addr;
303 if (setsockopt(sockfd, IPPROTO_IP,
304 include ? IP_ADD_SOURCE_MEMBERSHIP : IP_BLOCK_SOURCE,
305 (
const void *)&mreqs,
sizeof(mreqs)) < 0) {
320 const char *hostname,
int port)
326 if (!res0)
return AVERROR(EIO);
335 socklen_t *addr_len,
const char *localaddr)
340 int family = AF_UNSPEC;
342 if (((
struct sockaddr *) &
s->dest_addr)->sa_family)
343 family = ((
struct sockaddr *) &
s->dest_addr)->sa_family;
349 for (res = res0; res; res=res->
ai_next) {
350 if (
s->udplite_coverage)
354 if (udp_fd != -1)
break;
378 char sbuf[
sizeof(
int)*3+1];
386 return strtol(sbuf,
NULL, 10);
409 char hostname[256], buf[10];
417 if (
s->dest_addr_len < 0) {
421 p = strchr(uri,
'?');
424 int was_connected =
s->is_connected;
425 s->is_connected = strtol(buf,
NULL, 10);
426 if (
s->is_connected && !was_connected) {
427 if (connect(
s->udp_fd, (
struct sockaddr *) &
s->dest_addr,
448 return s->local_port;
462 #if HAVE_PTHREAD_CANCEL
463 static void *circular_buffer_task_rx(
void *_URLContext)
473 s->circular_buffer_error =
AVERROR(EIO);
479 socklen_t addr_len =
sizeof(addr);
486 len = recvfrom(
s->udp_fd,
s->tmp+4,
sizeof(
s->tmp)-4, 0, (
struct sockaddr *)&addr, &addr_len);
502 if (
s->overrun_nonfatal) {
504 "Surviving due to overrun_nonfatal option\n");
508 "To avoid, increase fifo_size URL option. "
509 "To survive in such case, use overrun_nonfatal option\n");
510 s->circular_buffer_error =
AVERROR(EIO);
524 static void *circular_buffer_task_tx(
void *_URLContext)
530 int64_t sent_bits = 0;
531 int64_t burst_interval =
s->bitrate ? (
s->burst_bits * 1000000 /
s->bitrate) : 0;
532 int64_t max_delay =
s->bitrate ? ((int64_t)
h->max_packet_size * 8 * 1000000 /
s->bitrate + 1) : 0;
538 s->circular_buffer_error =
AVERROR(EIO);
571 if (timestamp < target_timestamp) {
572 int64_t delay = target_timestamp - timestamp;
573 if (delay > max_delay) {
575 start_timestamp = timestamp + delay;
580 if (timestamp - burst_interval > target_timestamp) {
581 start_timestamp = timestamp - burst_interval;
585 sent_bits +=
len * 8;
586 target_timestamp = start_timestamp + sent_bits * 1000000 /
s->bitrate;
593 if (!
s->is_connected) {
594 ret = sendto (
s->udp_fd, p,
len, 0,
595 (
struct sockaddr *) &
s->dest_addr,
598 ret = send(
s->udp_fd, p,
len, 0);
606 s->circular_buffer_error = ret;
628 char hostname[1024], localaddr[1024] =
"";
629 int port, udp_fd = -1,
tmp, bind_ret = -1, dscp = -1;
641 if (
s->buffer_size < 0)
654 p = strchr(uri,
'?');
658 s->reuse_socket = strtol(buf, &endptr, 10);
665 s->overrun_nonfatal = strtol(buf, &endptr, 10);
668 s->overrun_nonfatal = 1;
671 "'overrun_nonfatal' option was set but it is not supported "
672 "on this build (pthread support is required)\n");
675 s->ttl = strtol(buf,
NULL, 10);
678 s->udplite_coverage = strtol(buf,
NULL, 10);
681 s->local_port = strtol(buf,
NULL, 10);
684 s->pkt_size = strtol(buf,
NULL, 10);
687 s->buffer_size = strtol(buf,
NULL, 10);
690 s->is_connected = strtol(buf,
NULL, 10);
693 dscp = strtol(buf,
NULL, 10);
696 s->circular_buffer_size = strtol(buf,
NULL, 10);
699 "'circular_buffer_size' option was set but it is not supported "
700 "on this build (pthread support is required)\n");
703 s->bitrate = strtoll(buf,
NULL, 10);
706 "'bitrate' option was set but it is not supported "
707 "on this build (pthread support is required)\n");
710 s->burst_bits = strtoll(buf,
NULL, 10);
713 av_strlcpy(localaddr, buf,
sizeof(localaddr));
724 s->timeout = strtol(buf,
NULL, 10);
726 s->is_broadcast = strtol(buf,
NULL, 10);
729 s->circular_buffer_size *= 188;
731 h->max_packet_size =
s->pkt_size;
735 h->rw_timeout =
s->timeout;
741 if (hostname[0] ==
'\0' || hostname[0] ==
'?') {
753 s->local_port = port;
764 s->local_addr_storage=my_addr;
769 if (
s->reuse_socket > 0 || (
s->is_multicast &&
s->reuse_socket < 0)) {
771 if (setsockopt (udp_fd, SOL_SOCKET, SO_REUSEADDR, &(
s->reuse_socket),
sizeof(
s->reuse_socket)) != 0) {
777 if (
s->is_broadcast) {
779 if (setsockopt (udp_fd, SOL_SOCKET, SO_BROADCAST, &(
s->is_broadcast),
sizeof(
s->is_broadcast)) != 0) {
793 if (
s->udplite_coverage) {
803 if (setsockopt (udp_fd, IPPROTO_IP, IP_TOS, &dscp,
sizeof(dscp)) != 0) {
814 bind_ret = bind(udp_fd,(
struct sockaddr *)&
s->dest_addr,
len);
819 if (bind_ret < 0 && bind(udp_fd,(
struct sockaddr *)&my_addr,
len) < 0) {
825 len =
sizeof(my_addr);
826 getsockname(udp_fd, (
struct sockaddr *)&my_addr, &
len);
829 if (
s->is_multicast) {
837 if (
s->filters.nb_include_addrs) {
839 (
struct sockaddr *)&
s->dest_addr,
840 s->dest_addr_len, &
s->local_addr_storage,
841 s->filters.include_addrs,
842 s->filters.nb_include_addrs, 1)) < 0)
848 if (
s->filters.nb_exclude_addrs) {
850 (
struct sockaddr *)&
s->dest_addr,
851 s->dest_addr_len, &
s->local_addr_storage,
852 s->filters.exclude_addrs,
853 s->filters.nb_exclude_addrs, 0)) < 0)
861 tmp =
s->buffer_size;
862 if (setsockopt(udp_fd, SOL_SOCKET, SO_SNDBUF, &
tmp,
sizeof(
tmp)) < 0) {
869 tmp =
s->buffer_size;
870 if (setsockopt(udp_fd, SOL_SOCKET, SO_RCVBUF, &
tmp,
sizeof(
tmp)) < 0) {
874 if (getsockopt(udp_fd, SOL_SOCKET, SO_RCVBUF, &
tmp, &
len) < 0) {
878 if(tmp < s->buffer_size)
879 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);
885 if (
s->is_connected) {
886 if (connect(udp_fd, (
struct sockaddr *) &
s->dest_addr,
s->dest_addr_len)) {
895 #if HAVE_PTHREAD_CANCEL
902 if (is_output &&
s->bitrate && !
s->circular_buffer_size) {
904 av_log(
h,
AV_LOG_WARNING,
"'bitrate' option was set but 'circular_buffer_size' is not, but required\n");
907 if ((!is_output &&
s->circular_buffer_size) || (is_output &&
s->bitrate &&
s->circular_buffer_size)) {
926 ret =
pthread_create(&
s->circular_buffer_thread,
NULL, is_output?circular_buffer_task_tx:circular_buffer_task_rx,
h);
932 s->thread_started = 1;
937 #if HAVE_PTHREAD_CANCEL
966 socklen_t addr_len =
sizeof(addr);
967 #if HAVE_PTHREAD_CANCEL
988 }
else if(
s->circular_buffer_error){
989 int err =
s->circular_buffer_error;
992 }
else if(nonblock) {
999 struct timespec tv = { .tv_sec = t / 1000000,
1000 .tv_nsec = (t % 1000000) * 1000 };
1004 return AVERROR(err == ETIMEDOUT ? EAGAIN : err);
1017 ret = recvfrom(
s->udp_fd, buf,
size, 0, (
struct sockaddr *)&addr, &addr_len);
1030 #if HAVE_PTHREAD_CANCEL
1040 if (
s->circular_buffer_error<0) {
1041 int err =
s->circular_buffer_error;
1065 if (!
s->is_connected) {
1066 ret = sendto (
s->udp_fd, buf,
size, 0,
1067 (
struct sockaddr *) &
s->dest_addr,
1070 ret = send(
s->udp_fd, buf,
size, 0);
1079 #if HAVE_PTHREAD_CANCEL
1091 #if HAVE_PTHREAD_CANCEL
1092 if (
s->thread_started) {
1100 shutdown(
s->udp_fd, SD_RECEIVE);
1101 CancelIoEx((HANDLE)(SOCKET)
s->udp_fd,
NULL);
1103 pthread_cancel(
s->circular_buffer_thread);
1113 closesocket(
s->udp_fd);
simple assert() macros that are a bit more flexible than ISO C assert().
#define av_assert0(cond)
assert() equivalent, that is always enabled.
#define AVIO_FLAG_READ
read-only
#define AVIO_FLAG_WRITE
write-only
#define AVIO_FLAG_NONBLOCK
Use non-blocking mode.
#define flags(name, subs,...)
#define HAVE_PTHREAD_CANCEL
#define pthread_mutex_lock(a)
#define pthread_mutex_unlock(a)
a very simple circular buffer FIFO implementation
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.
#define AV_LOG_DEBUG
Stuff which is only useful for libav* developers.
#define AV_LOG_WARNING
Something somehow does not look correct.
#define AV_LOG_ERROR
Something went wrong and cannot losslessly be recovered.
const char * av_default_item_name(void *ptr)
Return the context name.
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.
#define LIBAVUTIL_VERSION_INT
int ff_ip_check_source_lists(struct sockaddr_storage *source_addr_ptr, IPSourceFilters *s)
Checks the source address against a given IP source filter.
void ff_ip_reset_filters(IPSourceFilters *filters)
Resets the IP filter list and frees the internal fields of an IPSourceFilters structure.
int ff_ip_parse_blocks(void *log_ctx, const char *buf, IPSourceFilters *filters)
Parses the address[,address] source block list in buf and adds it to the filters in the IPSourceFilte...
struct addrinfo * ff_ip_resolve_host(void *log_ctx, const char *hostname, int port, int type, int family, int flags)
Resolves hostname into an addrinfo structure.
int ff_ip_parse_sources(void *log_ctx, const char *buf, IPSourceFilters *filters)
Parses the address[,address] source list in buf and adds it to the filters in the IPSourceFilters str...
void av_fifo_drain(AVFifoBuffer *f, int size)
Discard data from the FIFO.
void av_fifo_freep(AVFifoBuffer **f)
Free an AVFifoBuffer and reset pointer to NULL.
int av_fifo_size(const AVFifoBuffer *f)
Return the amount of data in bytes in the AVFifoBuffer, that is the amount of data you can read from ...
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.
int av_fifo_space(const AVFifoBuffer *f)
Return the amount of space in bytes in the AVFifoBuffer, that is the amount of data you can write int...
AVFifoBuffer * av_fifo_alloc(unsigned int size)
Initialize an AVFifoBuffer.
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.
common internal API header
int ff_is_multicast_address(struct sockaddr *addr)
int ff_socket(int af, int type, int proto)
void ff_log_net_error(void *ctx, int level, const char *prefix)
int ff_network_wait_fd(int fd, int write)
int ff_socket_nonblock(int socket, int enable)
static av_always_inline int pthread_cond_signal(pthread_cond_t *cond)
static av_always_inline int pthread_cond_destroy(pthread_cond_t *cond)
static av_always_inline int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutexattr_t *attr)
static av_always_inline int pthread_join(pthread_t thread, void **value_ptr)
static av_always_inline int pthread_cond_init(pthread_cond_t *cond, const pthread_condattr_t *attr)
static av_always_inline int pthread_create(pthread_t *thread, const pthread_attr_t *attr, void *(*start_routine)(void *), void *arg)
static av_always_inline int pthread_cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex, const struct timespec *abstime)
static av_always_inline int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex)
static av_always_inline int pthread_mutex_destroy(pthread_mutex_t *mutex)
miscellaneous OS support macros and functions.
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.
int(* cond)(enum AVPixelFormat pix_fmt)
Describe the class of an AVClass context structure.
const char * class_name
The name of the class; usually it is the same name as the context structure type to which the AVClass...
Structure for storing IP (UDP) source filters or block lists.
struct sockaddr_storage dest_addr
struct sockaddr_storage local_addr_storage
int circular_buffer_error
uint8_t tmp[UDP_MAX_PKT_SIZE+4]
struct sockaddr * ai_addr
struct addrinfo * ai_next
static void error(const char *err)
int av_usleep(unsigned usec)
Sleep for a period of time.
int64_t av_gettime_relative(void)
Get the current time in microseconds since some unspecified starting point.
int64_t av_gettime(void)
Get the current time in microseconds.
static int udp_join_multicast_group(int sockfd, struct sockaddr *addr, struct sockaddr *local_addr)
static int udp_socket_create(URLContext *h, struct sockaddr_storage *addr, socklen_t *addr_len, const char *localaddr)
#define UDPLITE_SEND_CSCOV
static const AVOption options[]
static int udp_set_url(URLContext *h, struct sockaddr_storage *addr, const char *hostname, int port)
int ff_udp_get_local_port(URLContext *h)
Return the local port used by the UDP connection.
const URLProtocol ff_udp_protocol
#define IPV6_DROP_MEMBERSHIP
static int udp_close(URLContext *h)
static int udp_set_multicast_ttl(int sockfd, int mcastTTL, struct sockaddr *addr)
static int udplite_open(URLContext *h, const char *uri, int flags)
static int udp_open(URLContext *h, const char *uri, int flags)
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,...
static const AVClass udp_class
const URLProtocol ff_udplite_protocol
static const AVClass udplite_context_class
#define IPV6_ADD_MEMBERSHIP
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 int udp_leave_multicast_group(int sockfd, struct sockaddr *addr, struct sockaddr *local_addr)
static int udp_port(struct sockaddr_storage *addr, int addr_len)
static int udp_write(URLContext *h, const uint8_t *buf, int size)
#define UDPLITE_RECV_CSCOV
static int udp_set_multicast_sources(URLContext *h, int sockfd, struct sockaddr *addr, int addr_len, struct sockaddr_storage *local_addr, struct sockaddr_storage *sources, int nb_sources, int include)
static int udp_read(URLContext *h, uint8_t *buf, int size)
unbuffered private I/O API
#define URL_PROTOCOL_FLAG_NETWORK
static int pthread_setcancelstate(int state, int *oldstate)
#define PTHREAD_CANCEL_DISABLE
#define PTHREAD_CANCEL_ENABLE