00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018
00019
00020
00021
00027 #define _BSD_SOURCE
00028
00029 #include "avformat.h"
00030 #include "avio_internal.h"
00031 #include "libavutil/parseutils.h"
00032 #include "libavutil/fifo.h"
00033 #include "libavutil/intreadwrite.h"
00034 #include "libavutil/avstring.h"
00035 #include "libavutil/opt.h"
00036 #include "libavutil/log.h"
00037 #include "internal.h"
00038 #include "network.h"
00039 #include "os_support.h"
00040 #include "url.h"
00041
00042 #if HAVE_PTHREAD_CANCEL
00043 #include <pthread.h>
00044 #endif
00045
00046 #ifndef HAVE_PTHREAD_CANCEL
00047 #define HAVE_PTHREAD_CANCEL 0
00048 #endif
00049
00050 #ifndef IPV6_ADD_MEMBERSHIP
00051 #define IPV6_ADD_MEMBERSHIP IPV6_JOIN_GROUP
00052 #define IPV6_DROP_MEMBERSHIP IPV6_LEAVE_GROUP
00053 #endif
00054
00055 #define UDP_TX_BUF_SIZE 32768
00056 #define UDP_MAX_PKT_SIZE 65536
00057
00058 typedef struct {
00059 const AVClass *class;
00060 int udp_fd;
00061 int ttl;
00062 int buffer_size;
00063 int is_multicast;
00064 int local_port;
00065 int reuse_socket;
00066 int overrun_nonfatal;
00067 struct sockaddr_storage dest_addr;
00068 int dest_addr_len;
00069 int is_connected;
00070
00071
00072 int circular_buffer_size;
00073 AVFifoBuffer *fifo;
00074 int circular_buffer_error;
00075 #if HAVE_PTHREAD_CANCEL
00076 pthread_t circular_buffer_thread;
00077 pthread_mutex_t mutex;
00078 pthread_cond_t cond;
00079 int thread_started;
00080 #endif
00081 uint8_t tmp[UDP_MAX_PKT_SIZE+4];
00082 int remaining_in_dg;
00083 char *local_addr;
00084 int packet_size;
00085 int timeout;
00086 } UDPContext;
00087
00088 #define OFFSET(x) offsetof(UDPContext, x)
00089 #define D AV_OPT_FLAG_DECODING_PARAM
00090 #define E AV_OPT_FLAG_ENCODING_PARAM
00091 static const AVOption options[] = {
00092 {"buffer_size", "Socket buffer size in bytes", OFFSET(buffer_size), AV_OPT_TYPE_INT, {.i64 = 0}, 0, INT_MAX, D|E },
00093 {"localport", "Set local port to bind to", OFFSET(local_port), AV_OPT_TYPE_INT, {.i64 = 0}, 0, INT_MAX, D|E },
00094 {"localaddr", "Choose local IP address", OFFSET(local_addr), AV_OPT_TYPE_STRING, {.str = ""}, 0, 0, D|E },
00095 {"pkt_size", "Set size of UDP packets", OFFSET(packet_size), AV_OPT_TYPE_INT, {.i64 = 1472}, 0, INT_MAX, D|E },
00096 {"reuse", "Explicitly allow or disallow reusing UDP sockets", OFFSET(reuse_socket), AV_OPT_TYPE_INT, {.i64 = 0}, 0, 1, D|E },
00097 {"ttl", "Set the time to live value (for multicast only)", OFFSET(ttl), AV_OPT_TYPE_INT, {.i64 = 16}, 0, INT_MAX, E },
00098 {"connect", "Should connect() be called on socket", OFFSET(is_connected), AV_OPT_TYPE_INT, {.i64 = 0}, 0, 1, D|E },
00099
00100 {"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 },
00101 {"overrun_nonfatal", "Survive in case of UDP receiving circular buffer overrun", OFFSET(overrun_nonfatal), AV_OPT_TYPE_INT, {.i64 = 0}, 0, 1, D },
00102 {"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 },
00103 {NULL}
00104 };
00105
00106 static const AVClass udp_context_class = {
00107 .class_name = "udp",
00108 .item_name = av_default_item_name,
00109 .option = options,
00110 .version = LIBAVUTIL_VERSION_INT,
00111 };
00112
00113 static void log_net_error(void *ctx, int level, const char* prefix)
00114 {
00115 char errbuf[100];
00116 av_strerror(ff_neterrno(), errbuf, sizeof(errbuf));
00117 av_log(ctx, level, "%s: %s\n", prefix, errbuf);
00118 }
00119
00120 static int udp_set_multicast_ttl(int sockfd, int mcastTTL,
00121 struct sockaddr *addr)
00122 {
00123 #ifdef IP_MULTICAST_TTL
00124 if (addr->sa_family == AF_INET) {
00125 if (setsockopt(sockfd, IPPROTO_IP, IP_MULTICAST_TTL, &mcastTTL, sizeof(mcastTTL)) < 0) {
00126 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IP_MULTICAST_TTL)");
00127 return -1;
00128 }
00129 }
00130 #endif
00131 #if defined(IPPROTO_IPV6) && defined(IPV6_MULTICAST_HOPS)
00132 if (addr->sa_family == AF_INET6) {
00133 if (setsockopt(sockfd, IPPROTO_IPV6, IPV6_MULTICAST_HOPS, &mcastTTL, sizeof(mcastTTL)) < 0) {
00134 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IPV6_MULTICAST_HOPS)");
00135 return -1;
00136 }
00137 }
00138 #endif
00139 return 0;
00140 }
00141
00142 static int udp_join_multicast_group(int sockfd, struct sockaddr *addr)
00143 {
00144 #ifdef IP_ADD_MEMBERSHIP
00145 if (addr->sa_family == AF_INET) {
00146 struct ip_mreq mreq;
00147
00148 mreq.imr_multiaddr.s_addr = ((struct sockaddr_in *)addr)->sin_addr.s_addr;
00149 mreq.imr_interface.s_addr= INADDR_ANY;
00150 if (setsockopt(sockfd, IPPROTO_IP, IP_ADD_MEMBERSHIP, (const void *)&mreq, sizeof(mreq)) < 0) {
00151 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IP_ADD_MEMBERSHIP)");
00152 return -1;
00153 }
00154 }
00155 #endif
00156 #if HAVE_STRUCT_IPV6_MREQ && defined(IPPROTO_IPV6)
00157 if (addr->sa_family == AF_INET6) {
00158 struct ipv6_mreq mreq6;
00159
00160 memcpy(&mreq6.ipv6mr_multiaddr, &(((struct sockaddr_in6 *)addr)->sin6_addr), sizeof(struct in6_addr));
00161 mreq6.ipv6mr_interface= 0;
00162 if (setsockopt(sockfd, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP, &mreq6, sizeof(mreq6)) < 0) {
00163 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IPV6_ADD_MEMBERSHIP)");
00164 return -1;
00165 }
00166 }
00167 #endif
00168 return 0;
00169 }
00170
00171 static int udp_leave_multicast_group(int sockfd, struct sockaddr *addr)
00172 {
00173 #ifdef IP_DROP_MEMBERSHIP
00174 if (addr->sa_family == AF_INET) {
00175 struct ip_mreq mreq;
00176
00177 mreq.imr_multiaddr.s_addr = ((struct sockaddr_in *)addr)->sin_addr.s_addr;
00178 mreq.imr_interface.s_addr= INADDR_ANY;
00179 if (setsockopt(sockfd, IPPROTO_IP, IP_DROP_MEMBERSHIP, (const void *)&mreq, sizeof(mreq)) < 0) {
00180 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IP_DROP_MEMBERSHIP)");
00181 return -1;
00182 }
00183 }
00184 #endif
00185 #if HAVE_STRUCT_IPV6_MREQ && defined(IPPROTO_IPV6)
00186 if (addr->sa_family == AF_INET6) {
00187 struct ipv6_mreq mreq6;
00188
00189 memcpy(&mreq6.ipv6mr_multiaddr, &(((struct sockaddr_in6 *)addr)->sin6_addr), sizeof(struct in6_addr));
00190 mreq6.ipv6mr_interface= 0;
00191 if (setsockopt(sockfd, IPPROTO_IPV6, IPV6_DROP_MEMBERSHIP, &mreq6, sizeof(mreq6)) < 0) {
00192 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IPV6_DROP_MEMBERSHIP)");
00193 return -1;
00194 }
00195 }
00196 #endif
00197 return 0;
00198 }
00199
00200 static struct addrinfo* udp_resolve_host(const char *hostname, int port,
00201 int type, int family, int flags)
00202 {
00203 struct addrinfo hints = { 0 }, *res = 0;
00204 int error;
00205 char sport[16];
00206 const char *node = 0, *service = "0";
00207
00208 if (port > 0) {
00209 snprintf(sport, sizeof(sport), "%d", port);
00210 service = sport;
00211 }
00212 if ((hostname) && (hostname[0] != '\0') && (hostname[0] != '?')) {
00213 node = hostname;
00214 }
00215 hints.ai_socktype = type;
00216 hints.ai_family = family;
00217 hints.ai_flags = flags;
00218 if ((error = getaddrinfo(node, service, &hints, &res))) {
00219 res = NULL;
00220 av_log(NULL, AV_LOG_ERROR, "udp_resolve_host: %s\n", gai_strerror(error));
00221 }
00222
00223 return res;
00224 }
00225
00226 static int udp_set_multicast_sources(int sockfd, struct sockaddr *addr,
00227 int addr_len, char **sources,
00228 int nb_sources, int include)
00229 {
00230 #if HAVE_STRUCT_GROUP_SOURCE_REQ && defined(MCAST_BLOCK_SOURCE) && !defined(_WIN32)
00231
00232
00233 int i;
00234 for (i = 0; i < nb_sources; i++) {
00235 struct group_source_req mreqs;
00236 int level = addr->sa_family == AF_INET ? IPPROTO_IP : IPPROTO_IPV6;
00237 struct addrinfo *sourceaddr = udp_resolve_host(sources[i], 0,
00238 SOCK_DGRAM, AF_UNSPEC,
00239 AI_NUMERICHOST);
00240 if (!sourceaddr)
00241 return AVERROR(ENOENT);
00242
00243 mreqs.gsr_interface = 0;
00244 memcpy(&mreqs.gsr_group, addr, addr_len);
00245 memcpy(&mreqs.gsr_source, sourceaddr->ai_addr, sourceaddr->ai_addrlen);
00246 freeaddrinfo(sourceaddr);
00247
00248 if (setsockopt(sockfd, level,
00249 include ? MCAST_JOIN_SOURCE_GROUP : MCAST_BLOCK_SOURCE,
00250 (const void *)&mreqs, sizeof(mreqs)) < 0) {
00251 if (include)
00252 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(MCAST_JOIN_SOURCE_GROUP)");
00253 else
00254 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(MCAST_BLOCK_SOURCE)");
00255 return ff_neterrno();
00256 }
00257 }
00258 #elif HAVE_STRUCT_IP_MREQ_SOURCE && defined(IP_BLOCK_SOURCE)
00259 int i;
00260 if (addr->sa_family != AF_INET) {
00261 av_log(NULL, AV_LOG_ERROR,
00262 "Setting multicast sources only supported for IPv4\n");
00263 return AVERROR(EINVAL);
00264 }
00265 for (i = 0; i < nb_sources; i++) {
00266 struct ip_mreq_source mreqs;
00267 struct addrinfo *sourceaddr = udp_resolve_host(sources[i], 0,
00268 SOCK_DGRAM, AF_UNSPEC,
00269 AI_NUMERICHOST);
00270 if (!sourceaddr)
00271 return AVERROR(ENOENT);
00272 if (sourceaddr->ai_addr->sa_family != AF_INET) {
00273 freeaddrinfo(sourceaddr);
00274 av_log(NULL, AV_LOG_ERROR, "%s is of incorrect protocol family\n",
00275 sources[i]);
00276 return AVERROR(EINVAL);
00277 }
00278
00279 mreqs.imr_multiaddr.s_addr = ((struct sockaddr_in *)addr)->sin_addr.s_addr;
00280 mreqs.imr_interface.s_addr = INADDR_ANY;
00281 mreqs.imr_sourceaddr.s_addr = ((struct sockaddr_in *)sourceaddr->ai_addr)->sin_addr.s_addr;
00282 freeaddrinfo(sourceaddr);
00283
00284 if (setsockopt(sockfd, IPPROTO_IP,
00285 include ? IP_ADD_SOURCE_MEMBERSHIP : IP_BLOCK_SOURCE,
00286 (const void *)&mreqs, sizeof(mreqs)) < 0) {
00287 if (include)
00288 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IP_ADD_SOURCE_MEMBERSHIP)");
00289 else
00290 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IP_BLOCK_SOURCE)");
00291 return ff_neterrno();
00292 }
00293 }
00294 #else
00295 return AVERROR(ENOSYS);
00296 #endif
00297 return 0;
00298 }
00299 static int udp_set_url(struct sockaddr_storage *addr,
00300 const char *hostname, int port)
00301 {
00302 struct addrinfo *res0;
00303 int addr_len;
00304
00305 res0 = udp_resolve_host(hostname, port, SOCK_DGRAM, AF_UNSPEC, 0);
00306 if (res0 == 0) return AVERROR(EIO);
00307 memcpy(addr, res0->ai_addr, res0->ai_addrlen);
00308 addr_len = res0->ai_addrlen;
00309 freeaddrinfo(res0);
00310
00311 return addr_len;
00312 }
00313
00314 static int udp_socket_create(UDPContext *s, struct sockaddr_storage *addr,
00315 socklen_t *addr_len, const char *localaddr)
00316 {
00317 int udp_fd = -1;
00318 struct addrinfo *res0 = NULL, *res = NULL;
00319 int family = AF_UNSPEC;
00320
00321 if (((struct sockaddr *) &s->dest_addr)->sa_family)
00322 family = ((struct sockaddr *) &s->dest_addr)->sa_family;
00323 res0 = udp_resolve_host(localaddr[0] ? localaddr : NULL, s->local_port,
00324 SOCK_DGRAM, family, AI_PASSIVE);
00325 if (res0 == 0)
00326 goto fail;
00327 for (res = res0; res; res=res->ai_next) {
00328 udp_fd = socket(res->ai_family, SOCK_DGRAM, 0);
00329 if (udp_fd != -1) break;
00330 log_net_error(NULL, AV_LOG_ERROR, "socket");
00331 }
00332
00333 if (udp_fd < 0)
00334 goto fail;
00335
00336 memcpy(addr, res->ai_addr, res->ai_addrlen);
00337 *addr_len = res->ai_addrlen;
00338
00339 freeaddrinfo(res0);
00340
00341 return udp_fd;
00342
00343 fail:
00344 if (udp_fd >= 0)
00345 closesocket(udp_fd);
00346 if(res0)
00347 freeaddrinfo(res0);
00348 return -1;
00349 }
00350
00351 static int udp_port(struct sockaddr_storage *addr, int addr_len)
00352 {
00353 char sbuf[sizeof(int)*3+1];
00354 int error;
00355
00356 if ((error = getnameinfo((struct sockaddr *)addr, addr_len, NULL, 0, sbuf, sizeof(sbuf), NI_NUMERICSERV)) != 0) {
00357 av_log(NULL, AV_LOG_ERROR, "getnameinfo: %s\n", gai_strerror(error));
00358 return -1;
00359 }
00360
00361 return strtol(sbuf, NULL, 10);
00362 }
00363
00364
00381 int ff_udp_set_remote_url(URLContext *h, const char *uri)
00382 {
00383 UDPContext *s = h->priv_data;
00384 char hostname[256], buf[10];
00385 int port;
00386 const char *p;
00387
00388 av_url_split(NULL, 0, NULL, 0, hostname, sizeof(hostname), &port, NULL, 0, uri);
00389
00390
00391 s->dest_addr_len = udp_set_url(&s->dest_addr, hostname, port);
00392 if (s->dest_addr_len < 0) {
00393 return AVERROR(EIO);
00394 }
00395 s->is_multicast = ff_is_multicast_address((struct sockaddr*) &s->dest_addr);
00396 p = strchr(uri, '?');
00397 if (p) {
00398 if (av_find_info_tag(buf, sizeof(buf), "connect", p)) {
00399 int was_connected = s->is_connected;
00400 s->is_connected = strtol(buf, NULL, 10);
00401 if (s->is_connected && !was_connected) {
00402 if (connect(s->udp_fd, (struct sockaddr *) &s->dest_addr,
00403 s->dest_addr_len)) {
00404 s->is_connected = 0;
00405 log_net_error(h, AV_LOG_ERROR, "connect");
00406 return AVERROR(EIO);
00407 }
00408 }
00409 }
00410 }
00411
00412 return 0;
00413 }
00414
00420 int ff_udp_get_local_port(URLContext *h)
00421 {
00422 UDPContext *s = h->priv_data;
00423 return s->local_port;
00424 }
00425
00431 static int udp_get_file_handle(URLContext *h)
00432 {
00433 UDPContext *s = h->priv_data;
00434 return s->udp_fd;
00435 }
00436
00437 #if HAVE_PTHREAD_CANCEL
00438 static void *circular_buffer_task( void *_URLContext)
00439 {
00440 URLContext *h = _URLContext;
00441 UDPContext *s = h->priv_data;
00442 int old_cancelstate;
00443
00444 pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancelstate);
00445 pthread_mutex_lock(&s->mutex);
00446 if (ff_socket_nonblock(s->udp_fd, 0) < 0) {
00447 av_log(h, AV_LOG_ERROR, "Failed to set blocking mode");
00448 s->circular_buffer_error = AVERROR(EIO);
00449 goto end;
00450 }
00451 while(1) {
00452 int len;
00453
00454 pthread_mutex_unlock(&s->mutex);
00455
00456
00457
00458 pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &old_cancelstate);
00459 len = recv(s->udp_fd, s->tmp+4, sizeof(s->tmp)-4, 0);
00460 pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancelstate);
00461 pthread_mutex_lock(&s->mutex);
00462 if (len < 0) {
00463 if (ff_neterrno() != AVERROR(EAGAIN) && ff_neterrno() != AVERROR(EINTR)) {
00464 s->circular_buffer_error = ff_neterrno();
00465 goto end;
00466 }
00467 continue;
00468 }
00469 AV_WL32(s->tmp, len);
00470
00471 if(av_fifo_space(s->fifo) < len + 4) {
00472
00473 if (s->overrun_nonfatal) {
00474 av_log(h, AV_LOG_WARNING, "Circular buffer overrun. "
00475 "Surviving due to overrun_nonfatal option\n");
00476 continue;
00477 } else {
00478 av_log(h, AV_LOG_ERROR, "Circular buffer overrun. "
00479 "To avoid, increase fifo_size URL option. "
00480 "To survive in such case, use overrun_nonfatal option\n");
00481 s->circular_buffer_error = AVERROR(EIO);
00482 goto end;
00483 }
00484 }
00485 av_fifo_generic_write(s->fifo, s->tmp, len+4, NULL);
00486 pthread_cond_signal(&s->cond);
00487 }
00488
00489 end:
00490 pthread_cond_signal(&s->cond);
00491 pthread_mutex_unlock(&s->mutex);
00492 return NULL;
00493 }
00494 #endif
00495
00496
00497
00498 static int udp_open(URLContext *h, const char *uri, int flags)
00499 {
00500 char hostname[1024], localaddr[1024] = "";
00501 int port, udp_fd = -1, tmp, bind_ret = -1;
00502 UDPContext *s = h->priv_data;
00503 int is_output;
00504 const char *p;
00505 char buf[256];
00506 struct sockaddr_storage my_addr;
00507 socklen_t len;
00508 int reuse_specified = 0;
00509 int i, include = 0, num_sources = 0;
00510 char *sources[32];
00511
00512 h->is_streamed = 1;
00513
00514 is_output = !(flags & AVIO_FLAG_READ);
00515 if (!s->buffer_size)
00516 s->buffer_size = is_output ? UDP_TX_BUF_SIZE : UDP_MAX_PKT_SIZE;
00517
00518 p = strchr(uri, '?');
00519 if (p) {
00520 if (av_find_info_tag(buf, sizeof(buf), "reuse", p)) {
00521 char *endptr = NULL;
00522 s->reuse_socket = strtol(buf, &endptr, 10);
00523
00524 if (buf == endptr)
00525 s->reuse_socket = 1;
00526 reuse_specified = 1;
00527 }
00528 if (av_find_info_tag(buf, sizeof(buf), "overrun_nonfatal", p)) {
00529 char *endptr = NULL;
00530 s->overrun_nonfatal = strtol(buf, &endptr, 10);
00531
00532 if (buf == endptr)
00533 s->overrun_nonfatal = 1;
00534 if (!HAVE_PTHREAD_CANCEL)
00535 av_log(h, AV_LOG_WARNING,
00536 "'overrun_nonfatal' option was set but it is not supported "
00537 "on this build (pthread support is required)\n");
00538 }
00539 if (av_find_info_tag(buf, sizeof(buf), "ttl", p)) {
00540 s->ttl = strtol(buf, NULL, 10);
00541 }
00542 if (av_find_info_tag(buf, sizeof(buf), "localport", p)) {
00543 s->local_port = strtol(buf, NULL, 10);
00544 }
00545 if (av_find_info_tag(buf, sizeof(buf), "pkt_size", p)) {
00546 s->packet_size = strtol(buf, NULL, 10);
00547 }
00548 if (av_find_info_tag(buf, sizeof(buf), "buffer_size", p)) {
00549 s->buffer_size = strtol(buf, NULL, 10);
00550 }
00551 if (av_find_info_tag(buf, sizeof(buf), "connect", p)) {
00552 s->is_connected = strtol(buf, NULL, 10);
00553 }
00554 if (av_find_info_tag(buf, sizeof(buf), "fifo_size", p)) {
00555 s->circular_buffer_size = strtol(buf, NULL, 10);
00556 if (!HAVE_PTHREAD_CANCEL)
00557 av_log(h, AV_LOG_WARNING,
00558 "'circular_buffer_size' option was set but it is not supported "
00559 "on this build (pthread support is required)\n");
00560 }
00561 if (av_find_info_tag(buf, sizeof(buf), "localaddr", p)) {
00562 av_strlcpy(localaddr, buf, sizeof(localaddr));
00563 }
00564 if (av_find_info_tag(buf, sizeof(buf), "sources", p))
00565 include = 1;
00566 if (include || av_find_info_tag(buf, sizeof(buf), "block", p)) {
00567 char *source_start;
00568
00569 source_start = buf;
00570 while (1) {
00571 char *next = strchr(source_start, ',');
00572 if (next)
00573 *next = '\0';
00574 sources[num_sources] = av_strdup(source_start);
00575 if (!sources[num_sources])
00576 goto fail;
00577 source_start = next + 1;
00578 num_sources++;
00579 if (num_sources >= FF_ARRAY_ELEMS(sources) || !next)
00580 break;
00581 }
00582 }
00583 if (!is_output && av_find_info_tag(buf, sizeof(buf), "timeout", p))
00584 s->timeout = strtol(buf, NULL, 10);
00585 }
00586
00587 s->circular_buffer_size *= 188;
00588 h->max_packet_size = s->packet_size;
00589 h->rw_timeout = s->timeout;
00590
00591
00592 av_url_split(NULL, 0, NULL, 0, hostname, sizeof(hostname), &port, NULL, 0, uri);
00593
00594
00595 if (hostname[0] == '\0' || hostname[0] == '?') {
00596
00597 if (!(flags & AVIO_FLAG_READ))
00598 goto fail;
00599 } else {
00600 if (ff_udp_set_remote_url(h, uri) < 0)
00601 goto fail;
00602 }
00603
00604 if ((s->is_multicast || !s->local_port) && (h->flags & AVIO_FLAG_READ))
00605 s->local_port = port;
00606 udp_fd = udp_socket_create(s, &my_addr, &len, localaddr[0] ? localaddr : s->local_addr);
00607 if (udp_fd < 0)
00608 goto fail;
00609
00610
00611
00612
00613 if (s->reuse_socket || (s->is_multicast && !reuse_specified)) {
00614 s->reuse_socket = 1;
00615 if (setsockopt (udp_fd, SOL_SOCKET, SO_REUSEADDR, &(s->reuse_socket), sizeof(s->reuse_socket)) != 0)
00616 goto fail;
00617 }
00618
00619
00620
00621
00622
00623 if (s->is_multicast && !(h->flags & AVIO_FLAG_WRITE)) {
00624 bind_ret = bind(udp_fd,(struct sockaddr *)&s->dest_addr, len);
00625 }
00626
00627
00628
00629 if (bind_ret < 0 && bind(udp_fd,(struct sockaddr *)&my_addr, len) < 0) {
00630 log_net_error(h, AV_LOG_ERROR, "bind failed");
00631 goto fail;
00632 }
00633
00634 len = sizeof(my_addr);
00635 getsockname(udp_fd, (struct sockaddr *)&my_addr, &len);
00636 s->local_port = udp_port(&my_addr, len);
00637
00638 if (s->is_multicast) {
00639 if (h->flags & AVIO_FLAG_WRITE) {
00640
00641 if (udp_set_multicast_ttl(udp_fd, s->ttl, (struct sockaddr *)&s->dest_addr) < 0)
00642 goto fail;
00643 }
00644 if (h->flags & AVIO_FLAG_READ) {
00645
00646 if (num_sources == 0 || !include) {
00647 if (udp_join_multicast_group(udp_fd, (struct sockaddr *)&s->dest_addr) < 0)
00648 goto fail;
00649
00650 if (num_sources) {
00651 if (udp_set_multicast_sources(udp_fd, (struct sockaddr *)&s->dest_addr, s->dest_addr_len, sources, num_sources, 0) < 0)
00652 goto fail;
00653 }
00654 } else if (include && num_sources) {
00655 if (udp_set_multicast_sources(udp_fd, (struct sockaddr *)&s->dest_addr, s->dest_addr_len, sources, num_sources, 1) < 0)
00656 goto fail;
00657 } else {
00658 av_log(NULL, AV_LOG_ERROR, "invalid udp settings: inclusive multicast but no sources given\n");
00659 goto fail;
00660 }
00661 }
00662 }
00663
00664 if (is_output) {
00665
00666 tmp = s->buffer_size;
00667 if (setsockopt(udp_fd, SOL_SOCKET, SO_SNDBUF, &tmp, sizeof(tmp)) < 0) {
00668 log_net_error(h, AV_LOG_ERROR, "setsockopt(SO_SNDBUF)");
00669 goto fail;
00670 }
00671 } else {
00672
00673
00674 tmp = s->buffer_size;
00675 if (setsockopt(udp_fd, SOL_SOCKET, SO_RCVBUF, &tmp, sizeof(tmp)) < 0) {
00676 log_net_error(h, AV_LOG_WARNING, "setsockopt(SO_RECVBUF)");
00677 }
00678
00679 ff_socket_nonblock(udp_fd, 1);
00680 }
00681 if (s->is_connected) {
00682 if (connect(udp_fd, (struct sockaddr *) &s->dest_addr, s->dest_addr_len)) {
00683 log_net_error(h, AV_LOG_ERROR, "connect");
00684 goto fail;
00685 }
00686 }
00687
00688 for (i = 0; i < num_sources; i++)
00689 av_freep(&sources[i]);
00690
00691 s->udp_fd = udp_fd;
00692
00693 #if HAVE_PTHREAD_CANCEL
00694 if (!is_output && s->circular_buffer_size) {
00695 int ret;
00696
00697
00698 s->fifo = av_fifo_alloc(s->circular_buffer_size);
00699 ret = pthread_mutex_init(&s->mutex, NULL);
00700 if (ret != 0) {
00701 av_log(h, AV_LOG_ERROR, "pthread_mutex_init failed : %s\n", strerror(ret));
00702 goto fail;
00703 }
00704 ret = pthread_cond_init(&s->cond, NULL);
00705 if (ret != 0) {
00706 av_log(h, AV_LOG_ERROR, "pthread_cond_init failed : %s\n", strerror(ret));
00707 goto cond_fail;
00708 }
00709 ret = pthread_create(&s->circular_buffer_thread, NULL, circular_buffer_task, h);
00710 if (ret != 0) {
00711 av_log(h, AV_LOG_ERROR, "pthread_create failed : %s\n", strerror(ret));
00712 goto thread_fail;
00713 }
00714 s->thread_started = 1;
00715 }
00716 #endif
00717
00718 return 0;
00719 #if HAVE_PTHREAD_CANCEL
00720 thread_fail:
00721 pthread_cond_destroy(&s->cond);
00722 cond_fail:
00723 pthread_mutex_destroy(&s->mutex);
00724 #endif
00725 fail:
00726 if (udp_fd >= 0)
00727 closesocket(udp_fd);
00728 av_fifo_free(s->fifo);
00729 for (i = 0; i < num_sources; i++)
00730 av_freep(&sources[i]);
00731 return AVERROR(EIO);
00732 }
00733
00734 static int udp_read(URLContext *h, uint8_t *buf, int size)
00735 {
00736 UDPContext *s = h->priv_data;
00737 int ret;
00738 int avail, nonblock = h->flags & AVIO_FLAG_NONBLOCK;
00739
00740 #if HAVE_PTHREAD_CANCEL
00741 if (s->fifo) {
00742 pthread_mutex_lock(&s->mutex);
00743 do {
00744 avail = av_fifo_size(s->fifo);
00745 if (avail) {
00746 uint8_t tmp[4];
00747
00748 av_fifo_generic_read(s->fifo, tmp, 4, NULL);
00749 avail= AV_RL32(tmp);
00750 if(avail > size){
00751 av_log(h, AV_LOG_WARNING, "Part of datagram lost due to insufficient buffer size\n");
00752 avail= size;
00753 }
00754
00755 av_fifo_generic_read(s->fifo, buf, avail, NULL);
00756 av_fifo_drain(s->fifo, AV_RL32(tmp) - avail);
00757 pthread_mutex_unlock(&s->mutex);
00758 return avail;
00759 } else if(s->circular_buffer_error){
00760 int err = s->circular_buffer_error;
00761 pthread_mutex_unlock(&s->mutex);
00762 return err;
00763 } else if(nonblock) {
00764 pthread_mutex_unlock(&s->mutex);
00765 return AVERROR(EAGAIN);
00766 }
00767 else {
00768
00769
00770 int64_t t = av_gettime() + 100000;
00771 struct timespec tv = { .tv_sec = t / 1000000,
00772 .tv_nsec = (t % 1000000) * 1000 };
00773 if (pthread_cond_timedwait(&s->cond, &s->mutex, &tv) < 0) {
00774 pthread_mutex_unlock(&s->mutex);
00775 return AVERROR(errno == ETIMEDOUT ? EAGAIN : errno);
00776 }
00777 nonblock = 1;
00778 }
00779 } while( 1);
00780 }
00781 #endif
00782
00783 if (!(h->flags & AVIO_FLAG_NONBLOCK)) {
00784 ret = ff_network_wait_fd(s->udp_fd, 0);
00785 if (ret < 0)
00786 return ret;
00787 }
00788 ret = recv(s->udp_fd, buf, size, 0);
00789
00790 return ret < 0 ? ff_neterrno() : ret;
00791 }
00792
00793 static int udp_write(URLContext *h, const uint8_t *buf, int size)
00794 {
00795 UDPContext *s = h->priv_data;
00796 int ret;
00797
00798 if (!(h->flags & AVIO_FLAG_NONBLOCK)) {
00799 ret = ff_network_wait_fd(s->udp_fd, 1);
00800 if (ret < 0)
00801 return ret;
00802 }
00803
00804 if (!s->is_connected) {
00805 ret = sendto (s->udp_fd, buf, size, 0,
00806 (struct sockaddr *) &s->dest_addr,
00807 s->dest_addr_len);
00808 } else
00809 ret = send(s->udp_fd, buf, size, 0);
00810
00811 return ret < 0 ? ff_neterrno() : ret;
00812 }
00813
00814 static int udp_close(URLContext *h)
00815 {
00816 UDPContext *s = h->priv_data;
00817 int ret;
00818
00819 if (s->is_multicast && (h->flags & AVIO_FLAG_READ))
00820 udp_leave_multicast_group(s->udp_fd, (struct sockaddr *)&s->dest_addr);
00821 closesocket(s->udp_fd);
00822 #if HAVE_PTHREAD_CANCEL
00823 if (s->thread_started) {
00824 pthread_cancel(s->circular_buffer_thread);
00825 ret = pthread_join(s->circular_buffer_thread, NULL);
00826 if (ret != 0)
00827 av_log(h, AV_LOG_ERROR, "pthread_join(): %s\n", strerror(ret));
00828 pthread_mutex_destroy(&s->mutex);
00829 pthread_cond_destroy(&s->cond);
00830 }
00831 #endif
00832 av_fifo_free(s->fifo);
00833 return 0;
00834 }
00835
00836 URLProtocol ff_udp_protocol = {
00837 .name = "udp",
00838 .url_open = udp_open,
00839 .url_read = udp_read,
00840 .url_write = udp_write,
00841 .url_close = udp_close,
00842 .url_get_file_handle = udp_get_file_handle,
00843 .priv_data_size = sizeof(UDPContext),
00844 .priv_data_class = &udp_context_class,
00845 .flags = URL_PROTOCOL_FLAG_NETWORK,
00846 };