27 #define _DEFAULT_SOURCE
45 #include "TargetConditionals.h"
54 #define UDPLITE_SEND_CSCOV 10
55 #define UDPLITE_RECV_CSCOV 11
58 #ifndef IPPROTO_UDPLITE
59 #define IPPROTO_UDPLITE 136
63 #undef HAVE_PTHREAD_CANCEL
64 #define HAVE_PTHREAD_CANCEL 1
67 #if HAVE_PTHREAD_CANCEL
71 #ifndef IPV6_ADD_MEMBERSHIP
72 #define IPV6_ADD_MEMBERSHIP IPV6_JOIN_GROUP
73 #define IPV6_DROP_MEMBERSHIP IPV6_LEAVE_GROUP
76 #define UDP_TX_BUF_SIZE 32768
77 #define UDP_RX_BUF_SIZE 393216
78 #define UDP_MAX_PKT_SIZE 65536
79 #define UDP_HEADER_SIZE 8
104 #if HAVE_PTHREAD_CANCEL
120 #define OFFSET(x) offsetof(UDPContext, x)
121 #define D AV_OPT_FLAG_DECODING_PARAM
122 #define E AV_OPT_FLAG_ENCODING_PARAM
124 {
"buffer_size",
"System data size (in bytes)",
OFFSET(buffer_size),
AV_OPT_TYPE_INT, { .i64 = -1 }, -1, INT_MAX, .flags =
D|
E },
126 {
"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 },
130 {
"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 },
131 {
"pkt_size",
"Maximum UDP packet size",
OFFSET(pkt_size),
AV_OPT_TYPE_INT, { .i64 = 1472 }, -1, INT_MAX, .flags =
D|
E },
132 {
"reuse",
"explicitly allow reusing UDP sockets",
OFFSET(reuse_socket),
AV_OPT_TYPE_BOOL, { .i64 = -1 }, -1, 1,
D|
E },
133 {
"reuse_socket",
"explicitly allow reusing UDP sockets",
OFFSET(reuse_socket),
AV_OPT_TYPE_BOOL, { .i64 = -1 }, -1, 1, .flags =
D|
E },
134 {
"broadcast",
"explicitly allow or disallow broadcast destination",
OFFSET(is_broadcast),
AV_OPT_TYPE_BOOL, { .i64 = 0 }, 0, 1,
E },
136 {
"connect",
"set if connect() should be called on socket",
OFFSET(is_connected),
AV_OPT_TYPE_BOOL, { .i64 = 0 }, 0, 1, .flags =
D|
E },
137 {
"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 },
138 {
"overrun_nonfatal",
"survive in case of UDP receiving circular buffer overrun",
OFFSET(overrun_nonfatal),
AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1,
D },
139 {
"timeout",
"set raise error timeout, in microseconds (only in read mode)",
OFFSET(timeout),
AV_OPT_TYPE_INT, {.i64 = 0}, 0, INT_MAX,
D },
160 struct sockaddr *addr,
169 switch (addr->sa_family) {
170 #ifdef IP_MULTICAST_TTL
172 protocol = IPPROTO_IP;
173 cmd = IP_MULTICAST_TTL;
176 #ifdef IPV6_MULTICAST_HOPS
178 protocol = IPPROTO_IPV6;
179 cmd = IPV6_MULTICAST_HOPS;
186 if (setsockopt(sockfd, protocol, cmd, &mcastTTL,
sizeof(mcastTTL)) < 0) {
188 unsigned char ttl = (
unsigned char) mcastTTL;
191 if (setsockopt(sockfd, protocol, cmd, &ttl,
sizeof(ttl)) < 0) {
201 struct sockaddr *local_addr,
void *logctx)
203 #ifdef IP_ADD_MEMBERSHIP
204 if (addr->sa_family == AF_INET) {
207 mreq.imr_multiaddr.s_addr = ((
struct sockaddr_in *)addr)->sin_addr.s_addr;
209 mreq.imr_interface= ((
struct sockaddr_in *)local_addr)->sin_addr;
211 mreq.imr_interface.s_addr = INADDR_ANY;
212 if (setsockopt(sockfd, IPPROTO_IP, IP_ADD_MEMBERSHIP, (
const void *)&mreq,
sizeof(mreq)) < 0) {
218 #if HAVE_STRUCT_IPV6_MREQ && defined(IPPROTO_IPV6)
219 if (addr->sa_family == AF_INET6) {
220 struct ipv6_mreq mreq6;
222 memcpy(&mreq6.ipv6mr_multiaddr, &(((
struct sockaddr_in6 *)addr)->sin6_addr),
sizeof(
struct in6_addr));
224 mreq6.ipv6mr_interface = 0;
235 struct sockaddr *local_addr,
void *logctx)
237 #ifdef IP_DROP_MEMBERSHIP
238 if (addr->sa_family == AF_INET) {
241 mreq.imr_multiaddr.s_addr = ((
struct sockaddr_in *)addr)->sin_addr.s_addr;
243 mreq.imr_interface = ((
struct sockaddr_in *)local_addr)->sin_addr;
245 mreq.imr_interface.s_addr = INADDR_ANY;
246 if (setsockopt(sockfd, IPPROTO_IP, IP_DROP_MEMBERSHIP, (
const void *)&mreq,
sizeof(mreq)) < 0) {
252 #if HAVE_STRUCT_IPV6_MREQ && defined(IPPROTO_IPV6)
253 if (addr->sa_family == AF_INET6) {
254 struct ipv6_mreq mreq6;
256 memcpy(&mreq6.ipv6mr_multiaddr, &(((
struct sockaddr_in6 *)addr)->sin6_addr),
sizeof(
struct in6_addr));
258 mreq6.ipv6mr_interface = 0;
269 int sockfd,
struct sockaddr *addr,
272 int nb_sources,
int include)
275 if (addr->sa_family != AF_INET) {
276 #if HAVE_STRUCT_GROUP_SOURCE_REQ && defined(MCAST_BLOCK_SOURCE)
281 for (
i = 0;
i < nb_sources;
i++) {
282 struct group_source_req mreqs;
283 int level = addr->sa_family == AF_INET ? IPPROTO_IP : IPPROTO_IPV6;
286 mreqs.gsr_interface = 0;
287 memcpy(&mreqs.gsr_group, addr, addr_len);
290 if (setsockopt(sockfd,
level,
291 include ? MCAST_JOIN_SOURCE_GROUP : MCAST_BLOCK_SOURCE,
292 (
const void *)&mreqs,
sizeof(mreqs)) < 0) {
303 "Setting multicast sources only supported for IPv4\n");
307 #if HAVE_STRUCT_IP_MREQ_SOURCE && defined(IP_BLOCK_SOURCE)
308 for (
i = 0;
i < nb_sources;
i++) {
309 struct ip_mreq_source mreqs;
310 if (
sources[
i].ss_family != AF_INET) {
315 mreqs.imr_multiaddr.s_addr = ((
struct sockaddr_in *)addr)->sin_addr.s_addr;
317 mreqs.imr_interface = ((
struct sockaddr_in *)local_addr)->sin_addr;
319 mreqs.imr_interface.s_addr = INADDR_ANY;
320 mreqs.imr_sourceaddr.s_addr = ((
struct sockaddr_in *)&
sources[
i])->sin_addr.s_addr;
322 if (setsockopt(sockfd, IPPROTO_IP,
323 include ? IP_ADD_SOURCE_MEMBERSHIP : IP_BLOCK_SOURCE,
324 (
const void *)&mreqs,
sizeof(mreqs)) < 0) {
339 const char *hostname,
int port)
345 if (!res0)
return AVERROR(EIO);
354 socklen_t *addr_len,
const char *localaddr)
359 int family = AF_UNSPEC;
361 if (((
struct sockaddr *) &
s->dest_addr)->sa_family)
362 family = ((
struct sockaddr *) &
s->dest_addr)->sa_family;
368 for (res = res0; res; res=res->
ai_next) {
369 if (
s->udplite_coverage)
373 if (udp_fd != -1)
break;
397 char sbuf[
sizeof(
int)*3+1];
405 return strtol(sbuf,
NULL, 10);
428 char hostname[256], buf[10];
436 if (
s->dest_addr_len < 0) {
440 p = strchr(uri,
'?');
443 int was_connected =
s->is_connected;
444 s->is_connected = strtol(buf,
NULL, 10);
445 if (
s->is_connected && !was_connected) {
446 if (connect(
s->udp_fd, (
struct sockaddr *) &
s->dest_addr,
467 return s->local_port;
481 #if HAVE_PTHREAD_CANCEL
482 static void *circular_buffer_task_rx(
void *_URLContext)
494 s->circular_buffer_error =
AVERROR(EIO);
500 socklen_t addr_len =
sizeof(addr);
507 len = recvfrom(
s->udp_fd,
s->tmp+4,
sizeof(
s->tmp)-4, 0, (
struct sockaddr *)&addr, &addr_len);
523 if (
s->overrun_nonfatal) {
525 "Surviving due to overrun_nonfatal option\n");
529 "To avoid, increase fifo_size URL option. "
530 "To survive in such case, use overrun_nonfatal option\n");
531 s->circular_buffer_error =
AVERROR(EIO);
545 static void *circular_buffer_task_tx(
void *_URLContext)
551 int64_t sent_bits = 0;
552 int64_t burst_interval =
s->bitrate ? (
s->burst_bits * 1000000 /
s->bitrate) : 0;
553 int64_t max_delay =
s->bitrate ? ((int64_t)
h->max_packet_size * 8 * 1000000 /
s->bitrate + 1) : 0;
561 s->circular_buffer_error =
AVERROR(EIO);
592 if (timestamp < target_timestamp) {
593 int64_t delay = target_timestamp - timestamp;
594 if (delay > max_delay) {
596 start_timestamp = timestamp + delay;
601 if (timestamp - burst_interval > target_timestamp) {
602 start_timestamp = timestamp - burst_interval;
606 sent_bits +=
len * 8;
607 target_timestamp = start_timestamp + sent_bits * 1000000 /
s->bitrate;
614 if (!
s->is_connected) {
615 ret = sendto (
s->udp_fd, p,
len, 0,
616 (
struct sockaddr *) &
s->dest_addr,
619 ret = send(
s->udp_fd, p,
len, 0);
627 s->circular_buffer_error =
ret;
650 int port, udp_fd = -1,
tmp, bind_ret = -1, dscp = -1;
662 if (
s->buffer_size < 0)
675 p = strchr(uri,
'?');
679 s->reuse_socket = strtol(buf, &endptr, 10);
686 s->overrun_nonfatal = strtol(buf, &endptr, 10);
689 s->overrun_nonfatal = 1;
690 if (!HAVE_PTHREAD_CANCEL)
692 "'overrun_nonfatal' option was set but it is not supported "
693 "on this build (pthread support is required)\n");
696 s->ttl = strtol(buf,
NULL, 10);
697 if (
s->ttl < 0 ||
s->ttl > 255) {
704 s->udplite_coverage = strtol(buf,
NULL, 10);
707 s->local_port = strtol(buf,
NULL, 10);
710 s->pkt_size = strtol(buf,
NULL, 10);
713 s->buffer_size = strtol(buf,
NULL, 10);
716 s->is_connected = strtol(buf,
NULL, 10);
719 dscp = strtol(buf,
NULL, 10);
722 s->circular_buffer_size = strtol(buf,
NULL, 10);
723 if (!HAVE_PTHREAD_CANCEL)
725 "'circular_buffer_size' option was set but it is not supported "
726 "on this build (pthread support is required)\n");
729 s->bitrate = strtoll(buf,
NULL, 10);
730 if (!HAVE_PTHREAD_CANCEL)
732 "'bitrate' option was set but it is not supported "
733 "on this build (pthread support is required)\n");
736 s->burst_bits = strtoll(buf,
NULL, 10);
755 s->timeout = strtol(buf,
NULL, 10);
757 s->is_broadcast = strtol(buf,
NULL, 10);
760 s->circular_buffer_size *= 188;
762 h->max_packet_size =
s->pkt_size;
766 h->rw_timeout =
s->timeout;
772 if (hostname[0] ==
'\0' || hostname[0] ==
'?') {
784 s->local_port = port;
792 s->local_addr_storage=my_addr;
797 if (
s->reuse_socket > 0 || (
s->is_multicast &&
s->reuse_socket < 0)) {
799 if (setsockopt (udp_fd, SOL_SOCKET, SO_REUSEADDR, &(
s->reuse_socket),
sizeof(
s->reuse_socket)) != 0) {
805 if (
s->is_broadcast) {
807 if (setsockopt (udp_fd, SOL_SOCKET, SO_BROADCAST, &(
s->is_broadcast),
sizeof(
s->is_broadcast)) != 0) {
821 if (
s->udplite_coverage) {
831 if (setsockopt (udp_fd, IPPROTO_IP, IP_TOS, &dscp,
sizeof(dscp)) != 0) {
842 bind_ret = bind(udp_fd,(
struct sockaddr *)&
s->dest_addr,
len);
847 if (bind_ret < 0 && bind(udp_fd,(
struct sockaddr *)&my_addr,
len) < 0) {
853 len =
sizeof(my_addr);
854 getsockname(udp_fd, (
struct sockaddr *)&my_addr, &
len);
857 if (
s->is_multicast) {
865 if (
s->filters.nb_include_addrs) {
867 (
struct sockaddr *)&
s->dest_addr,
868 s->dest_addr_len, &
s->local_addr_storage,
869 s->filters.include_addrs,
870 s->filters.nb_include_addrs, 1)) < 0)
874 (
struct sockaddr *)&
s->local_addr_storage,
h)) < 0)
877 if (
s->filters.nb_exclude_addrs) {
879 (
struct sockaddr *)&
s->dest_addr,
880 s->dest_addr_len, &
s->local_addr_storage,
881 s->filters.exclude_addrs,
882 s->filters.nb_exclude_addrs, 0)) < 0)
890 tmp =
s->buffer_size;
891 if (setsockopt(udp_fd, SOL_SOCKET, SO_SNDBUF, &
tmp,
sizeof(
tmp)) < 0) {
898 tmp =
s->buffer_size;
899 if (setsockopt(udp_fd, SOL_SOCKET, SO_RCVBUF, &
tmp,
sizeof(
tmp)) < 0) {
903 if (getsockopt(udp_fd, SOL_SOCKET, SO_RCVBUF, &
tmp, &
len) < 0) {
907 if(tmp < s->buffer_size)
908 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);
914 if (
s->is_connected) {
915 if (connect(udp_fd, (
struct sockaddr *) &
s->dest_addr,
s->dest_addr_len)) {
924 #if HAVE_PTHREAD_CANCEL
931 if (is_output &&
s->bitrate && !
s->circular_buffer_size) {
933 av_log(
h,
AV_LOG_WARNING,
"'bitrate' option was set but 'circular_buffer_size' is not, but required\n");
936 if ((!is_output &&
s->circular_buffer_size) || (is_output &&
s->bitrate &&
s->circular_buffer_size)) {
955 ret =
pthread_create(&
s->circular_buffer_thread,
NULL, is_output?circular_buffer_task_tx:circular_buffer_task_rx,
h);
961 s->thread_started = 1;
966 #if HAVE_PTHREAD_CANCEL
995 socklen_t addr_len =
sizeof(addr);
996 #if HAVE_PTHREAD_CANCEL
1017 }
else if(
s->circular_buffer_error){
1018 int err =
s->circular_buffer_error;
1021 }
else if(nonblock) {
1028 struct timespec tv = { .tv_sec = t / 1000000,
1029 .tv_nsec = (t % 1000000) * 1000 };
1033 return AVERROR(err == ETIMEDOUT ? EAGAIN : err);
1046 ret = recvfrom(
s->udp_fd, buf,
size, 0, (
struct sockaddr *)&addr, &addr_len);
1059 #if HAVE_PTHREAD_CANCEL
1069 if (
s->circular_buffer_error<0) {
1070 int err =
s->circular_buffer_error;
1094 if (!
s->is_connected) {
1095 ret = sendto (
s->udp_fd, buf,
size, 0,
1096 (
struct sockaddr *) &
s->dest_addr,
1099 ret = send(
s->udp_fd, buf,
size, 0);
1108 #if HAVE_PTHREAD_CANCEL
1120 (
struct sockaddr *)&
s->local_addr_storage,
h);
1121 #if HAVE_PTHREAD_CANCEL
1122 if (
s->thread_started) {
1130 shutdown(
s->udp_fd, SD_RECEIVE);
1131 CancelIoEx((HANDLE)(SOCKET)
s->udp_fd,
NULL);
1133 pthread_cancel(
s->circular_buffer_thread);
1143 closesocket(
s->udp_fd);