Subversion Repositories Kolibri OS

Rev

Go to most recent revision | Blame | Last modification | View Log | RSS feed

  1. /*
  2.  * UDP prototype streaming system
  3.  * Copyright (c) 2000, 2001, 2002 Fabrice Bellard
  4.  *
  5.  * This file is part of FFmpeg.
  6.  *
  7.  * FFmpeg is free software; you can redistribute it and/or
  8.  * modify it under the terms of the GNU Lesser General Public
  9.  * License as published by the Free Software Foundation; either
  10.  * version 2.1 of the License, or (at your option) any later version.
  11.  *
  12.  * FFmpeg is distributed in the hope that it will be useful,
  13.  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14.  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15.  * Lesser General Public License for more details.
  16.  *
  17.  * You should have received a copy of the GNU Lesser General Public
  18.  * License along with FFmpeg; if not, write to the Free Software
  19.  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  20.  */
  21.  
  22. /**
  23.  * @file
  24.  * UDP protocol
  25.  */
  26.  
  27. #define _BSD_SOURCE     /* Needed for using struct ip_mreq with recent glibc */
  28.  
  29. #include "avformat.h"
  30. #include "avio_internal.h"
  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"
  38. #include "internal.h"
  39. #include "network.h"
  40. #include "os_support.h"
  41. #include "url.h"
  42.  
  43. #if HAVE_PTHREAD_CANCEL
  44. #include <pthread.h>
  45. #endif
  46.  
  47. #ifndef HAVE_PTHREAD_CANCEL
  48. #define HAVE_PTHREAD_CANCEL 0
  49. #endif
  50.  
  51. #ifndef IPV6_ADD_MEMBERSHIP
  52. #define IPV6_ADD_MEMBERSHIP IPV6_JOIN_GROUP
  53. #define IPV6_DROP_MEMBERSHIP IPV6_LEAVE_GROUP
  54. #endif
  55.  
  56. #define UDP_TX_BUF_SIZE 32768
  57. #define UDP_MAX_PKT_SIZE 65536
  58.  
  59. typedef struct {
  60.     const AVClass *class;
  61.     int udp_fd;
  62.     int ttl;
  63.     int buffer_size;
  64.     int is_multicast;
  65.     int local_port;
  66.     int reuse_socket;
  67.     int overrun_nonfatal;
  68.     struct sockaddr_storage dest_addr;
  69.     int dest_addr_len;
  70.     int is_connected;
  71.  
  72.     /* Circular Buffer variables for use in UDP receive code */
  73.     int circular_buffer_size;
  74.     AVFifoBuffer *fifo;
  75.     int circular_buffer_error;
  76. #if HAVE_PTHREAD_CANCEL
  77.     pthread_t circular_buffer_thread;
  78.     pthread_mutex_t mutex;
  79.     pthread_cond_t cond;
  80.     int thread_started;
  81. #endif
  82.     uint8_t tmp[UDP_MAX_PKT_SIZE+4];
  83.     int remaining_in_dg;
  84.     char *local_addr;
  85.     int packet_size;
  86.     int timeout;
  87. } UDPContext;
  88.  
  89. #define OFFSET(x) offsetof(UDPContext, x)
  90. #define D AV_OPT_FLAG_DECODING_PARAM
  91. #define E AV_OPT_FLAG_ENCODING_PARAM
  92. static const AVOption options[] = {
  93. {"buffer_size", "Socket buffer size in bytes", OFFSET(buffer_size), AV_OPT_TYPE_INT, {.i64 = 0}, 0, INT_MAX, D|E },
  94. {"localport", "Set local port to bind to", OFFSET(local_port), AV_OPT_TYPE_INT, {.i64 = 0}, 0, INT_MAX, D|E },
  95. {"localaddr", "Choose local IP address", OFFSET(local_addr), AV_OPT_TYPE_STRING, {.str = ""}, 0, 0, D|E },
  96. {"pkt_size", "Set size of UDP packets", OFFSET(packet_size), AV_OPT_TYPE_INT, {.i64 = 1472}, 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 },
  100. /* TODO 'sources', 'block' option */
  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 },
  104. {NULL}
  105. };
  106.  
  107. static const AVClass udp_context_class = {
  108.     .class_name     = "udp",
  109.     .item_name      = av_default_item_name,
  110.     .option         = options,
  111.     .version        = LIBAVUTIL_VERSION_INT,
  112. };
  113.  
  114. static void log_net_error(void *ctx, int level, const char* prefix)
  115. {
  116.     char errbuf[100];
  117.     av_strerror(ff_neterrno(), errbuf, sizeof(errbuf));
  118.     av_log(ctx, level, "%s: %s\n", prefix, errbuf);
  119. }
  120.  
  121. static int udp_set_multicast_ttl(int sockfd, int mcastTTL,
  122.                                  struct sockaddr *addr)
  123. {
  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) {
  127.             log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IP_MULTICAST_TTL)");
  128.             return -1;
  129.         }
  130.     }
  131. #endif
  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) {
  135.             log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IPV6_MULTICAST_HOPS)");
  136.             return -1;
  137.         }
  138.     }
  139. #endif
  140.     return 0;
  141. }
  142.  
  143. static int udp_join_multicast_group(int sockfd, struct sockaddr *addr)
  144. {
  145. #ifdef IP_ADD_MEMBERSHIP
  146.     if (addr->sa_family == AF_INET) {
  147.         struct ip_mreq mreq;
  148.  
  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) {
  152.             log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IP_ADD_MEMBERSHIP)");
  153.             return -1;
  154.         }
  155.     }
  156. #endif
  157. #if HAVE_STRUCT_IPV6_MREQ && defined(IPPROTO_IPV6)
  158.     if (addr->sa_family == AF_INET6) {
  159.         struct ipv6_mreq mreq6;
  160.  
  161.         memcpy(&mreq6.ipv6mr_multiaddr, &(((struct sockaddr_in6 *)addr)->sin6_addr), sizeof(struct in6_addr));
  162.         mreq6.ipv6mr_interface= 0;
  163.         if (setsockopt(sockfd, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP, &mreq6, sizeof(mreq6)) < 0) {
  164.             log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IPV6_ADD_MEMBERSHIP)");
  165.             return -1;
  166.         }
  167.     }
  168. #endif
  169.     return 0;
  170. }
  171.  
  172. static int udp_leave_multicast_group(int sockfd, struct sockaddr *addr)
  173. {
  174. #ifdef IP_DROP_MEMBERSHIP
  175.     if (addr->sa_family == AF_INET) {
  176.         struct ip_mreq mreq;
  177.  
  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) {
  181.             log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IP_DROP_MEMBERSHIP)");
  182.             return -1;
  183.         }
  184.     }
  185. #endif
  186. #if HAVE_STRUCT_IPV6_MREQ && defined(IPPROTO_IPV6)
  187.     if (addr->sa_family == AF_INET6) {
  188.         struct ipv6_mreq mreq6;
  189.  
  190.         memcpy(&mreq6.ipv6mr_multiaddr, &(((struct sockaddr_in6 *)addr)->sin6_addr), sizeof(struct in6_addr));
  191.         mreq6.ipv6mr_interface= 0;
  192.         if (setsockopt(sockfd, IPPROTO_IPV6, IPV6_DROP_MEMBERSHIP, &mreq6, sizeof(mreq6)) < 0) {
  193.             log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IPV6_DROP_MEMBERSHIP)");
  194.             return -1;
  195.         }
  196.     }
  197. #endif
  198.     return 0;
  199. }
  200.  
  201. static struct addrinfo* udp_resolve_host(const char *hostname, int port,
  202.                                          int type, int family, int flags)
  203. {
  204.     struct addrinfo hints = { 0 }, *res = 0;
  205.     int error;
  206.     char sport[16];
  207.     const char *node = 0, *service = "0";
  208.  
  209.     if (port > 0) {
  210.         snprintf(sport, sizeof(sport), "%d", port);
  211.         service = sport;
  212.     }
  213.     if ((hostname) && (hostname[0] != '\0') && (hostname[0] != '?')) {
  214.         node = hostname;
  215.     }
  216.     hints.ai_socktype = type;
  217.     hints.ai_family   = family;
  218.     hints.ai_flags = flags;
  219.     if ((error = getaddrinfo(node, service, &hints, &res))) {
  220.         res = NULL;
  221.         av_log(NULL, AV_LOG_ERROR, "udp_resolve_host: %s\n", gai_strerror(error));
  222.     }
  223.  
  224.     return res;
  225. }
  226.  
  227. static int udp_set_multicast_sources(int sockfd, struct sockaddr *addr,
  228.                                      int addr_len, char **sources,
  229.                                      int nb_sources, int include)
  230. {
  231. #if HAVE_STRUCT_GROUP_SOURCE_REQ && defined(MCAST_BLOCK_SOURCE) && !defined(_WIN32)
  232.     /* These ones are available in the microsoft SDK, but don't seem to work
  233.      * as on linux, so just prefer the v4-only approach there for now. */
  234.     int i;
  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;
  238.         struct addrinfo *sourceaddr = udp_resolve_host(sources[i], 0,
  239.                                                        SOCK_DGRAM, AF_UNSPEC,
  240.                                                        0);
  241.         if (!sourceaddr)
  242.             return AVERROR(ENOENT);
  243.  
  244.         mreqs.gsr_interface = 0;
  245.         memcpy(&mreqs.gsr_group, addr, addr_len);
  246.         memcpy(&mreqs.gsr_source, sourceaddr->ai_addr, sourceaddr->ai_addrlen);
  247.         freeaddrinfo(sourceaddr);
  248.  
  249.         if (setsockopt(sockfd, level,
  250.                        include ? MCAST_JOIN_SOURCE_GROUP : MCAST_BLOCK_SOURCE,
  251.                        (const void *)&mreqs, sizeof(mreqs)) < 0) {
  252.             if (include)
  253.                 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(MCAST_JOIN_SOURCE_GROUP)");
  254.             else
  255.                 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(MCAST_BLOCK_SOURCE)");
  256.             return ff_neterrno();
  257.         }
  258.     }
  259. #elif HAVE_STRUCT_IP_MREQ_SOURCE && defined(IP_BLOCK_SOURCE)
  260.     int i;
  261.     if (addr->sa_family != AF_INET) {
  262.         av_log(NULL, AV_LOG_ERROR,
  263.                "Setting multicast sources only supported for IPv4\n");
  264.         return AVERROR(EINVAL);
  265.     }
  266.     for (i = 0; i < nb_sources; i++) {
  267.         struct ip_mreq_source mreqs;
  268.         struct addrinfo *sourceaddr = udp_resolve_host(sources[i], 0,
  269.                                                        SOCK_DGRAM, AF_UNSPEC,
  270.                                                        0);
  271.         if (!sourceaddr)
  272.             return AVERROR(ENOENT);
  273.         if (sourceaddr->ai_addr->sa_family != AF_INET) {
  274.             freeaddrinfo(sourceaddr);
  275.             av_log(NULL, AV_LOG_ERROR, "%s is of incorrect protocol family\n",
  276.                    sources[i]);
  277.             return AVERROR(EINVAL);
  278.         }
  279.  
  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;
  283.         freeaddrinfo(sourceaddr);
  284.  
  285.         if (setsockopt(sockfd, IPPROTO_IP,
  286.                        include ? IP_ADD_SOURCE_MEMBERSHIP : IP_BLOCK_SOURCE,
  287.                        (const void *)&mreqs, sizeof(mreqs)) < 0) {
  288.             if (include)
  289.                 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IP_ADD_SOURCE_MEMBERSHIP)");
  290.             else
  291.                 log_net_error(NULL, AV_LOG_ERROR, "setsockopt(IP_BLOCK_SOURCE)");
  292.             return ff_neterrno();
  293.         }
  294.     }
  295. #else
  296.     return AVERROR(ENOSYS);
  297. #endif
  298.     return 0;
  299. }
  300. static int udp_set_url(struct sockaddr_storage *addr,
  301.                        const char *hostname, int port)
  302. {
  303.     struct addrinfo *res0;
  304.     int addr_len;
  305.  
  306.     res0 = udp_resolve_host(hostname, port, SOCK_DGRAM, AF_UNSPEC, 0);
  307.     if (res0 == 0) return AVERROR(EIO);
  308.     memcpy(addr, res0->ai_addr, res0->ai_addrlen);
  309.     addr_len = res0->ai_addrlen;
  310.     freeaddrinfo(res0);
  311.  
  312.     return addr_len;
  313. }
  314.  
  315. static int udp_socket_create(UDPContext *s, struct sockaddr_storage *addr,
  316.                              socklen_t *addr_len, const char *localaddr)
  317. {
  318.     int udp_fd = -1;
  319.     struct addrinfo *res0 = NULL, *res = NULL;
  320.     int family = AF_UNSPEC;
  321.  
  322.     if (((struct sockaddr *) &s->dest_addr)->sa_family)
  323.         family = ((struct sockaddr *) &s->dest_addr)->sa_family;
  324.     res0 = udp_resolve_host(localaddr[0] ? localaddr : NULL, s->local_port,
  325.                             SOCK_DGRAM, family, AI_PASSIVE);
  326.     if (res0 == 0)
  327.         goto fail;
  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;
  331.         log_net_error(NULL, AV_LOG_ERROR, "socket");
  332.     }
  333.  
  334.     if (udp_fd < 0)
  335.         goto fail;
  336.  
  337.     memcpy(addr, res->ai_addr, res->ai_addrlen);
  338.     *addr_len = res->ai_addrlen;
  339.  
  340.     freeaddrinfo(res0);
  341.  
  342.     return udp_fd;
  343.  
  344.  fail:
  345.     if (udp_fd >= 0)
  346.         closesocket(udp_fd);
  347.     if(res0)
  348.         freeaddrinfo(res0);
  349.     return -1;
  350. }
  351.  
  352. static int udp_port(struct sockaddr_storage *addr, int addr_len)
  353. {
  354.     char sbuf[sizeof(int)*3+1];
  355.     int error;
  356.  
  357.     if ((error = getnameinfo((struct sockaddr *)addr, addr_len, NULL, 0,  sbuf, sizeof(sbuf), NI_NUMERICSERV)) != 0) {
  358.         av_log(NULL, AV_LOG_ERROR, "getnameinfo: %s\n", gai_strerror(error));
  359.         return -1;
  360.     }
  361.  
  362.     return strtol(sbuf, NULL, 10);
  363. }
  364.  
  365.  
  366. /**
  367.  * If no filename is given to av_open_input_file because you want to
  368.  * get the local port first, then you must call this function to set
  369.  * the remote server address.
  370.  *
  371.  * url syntax: udp://host:port[?option=val...]
  372.  * option: 'ttl=n'       : set the ttl value (for multicast only)
  373.  *         'localport=n' : set the local port
  374.  *         'pkt_size=n'  : set max packet size
  375.  *         'reuse=1'     : enable reusing the socket
  376.  *         'overrun_nonfatal=1': survive in case of circular buffer overrun
  377.  *
  378.  * @param h media file context
  379.  * @param uri of the remote server
  380.  * @return zero if no error.
  381.  */
  382. int ff_udp_set_remote_url(URLContext *h, const char *uri)
  383. {
  384.     UDPContext *s = h->priv_data;
  385.     char hostname[256], buf[10];
  386.     int port;
  387.     const char *p;
  388.  
  389.     av_url_split(NULL, 0, NULL, 0, hostname, sizeof(hostname), &port, NULL, 0, uri);
  390.  
  391.     /* set the destination address */
  392.     s->dest_addr_len = udp_set_url(&s->dest_addr, hostname, port);
  393.     if (s->dest_addr_len < 0) {
  394.         return AVERROR(EIO);
  395.     }
  396.     s->is_multicast = ff_is_multicast_address((struct sockaddr*) &s->dest_addr);
  397.     p = strchr(uri, '?');
  398.     if (p) {
  399.         if (av_find_info_tag(buf, sizeof(buf), "connect", p)) {
  400.             int was_connected = s->is_connected;
  401.             s->is_connected = strtol(buf, NULL, 10);
  402.             if (s->is_connected && !was_connected) {
  403.                 if (connect(s->udp_fd, (struct sockaddr *) &s->dest_addr,
  404.                             s->dest_addr_len)) {
  405.                     s->is_connected = 0;
  406.                     log_net_error(h, AV_LOG_ERROR, "connect");
  407.                     return AVERROR(EIO);
  408.                 }
  409.             }
  410.         }
  411.     }
  412.  
  413.     return 0;
  414. }
  415.  
  416. /**
  417.  * Return the local port used by the UDP connection
  418.  * @param h media file context
  419.  * @return the local port number
  420.  */
  421. int ff_udp_get_local_port(URLContext *h)
  422. {
  423.     UDPContext *s = h->priv_data;
  424.     return s->local_port;
  425. }
  426.  
  427. /**
  428.  * Return the udp file handle for select() usage to wait for several RTP
  429.  * streams at the same time.
  430.  * @param h media file context
  431.  */
  432. static int udp_get_file_handle(URLContext *h)
  433. {
  434.     UDPContext *s = h->priv_data;
  435.     return s->udp_fd;
  436. }
  437.  
  438. #if HAVE_PTHREAD_CANCEL
  439. static void *circular_buffer_task( void *_URLContext)
  440. {
  441.     URLContext *h = _URLContext;
  442.     UDPContext *s = h->priv_data;
  443.     int old_cancelstate;
  444.  
  445.     pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancelstate);
  446.     pthread_mutex_lock(&s->mutex);
  447.     if (ff_socket_nonblock(s->udp_fd, 0) < 0) {
  448.         av_log(h, AV_LOG_ERROR, "Failed to set blocking mode");
  449.         s->circular_buffer_error = AVERROR(EIO);
  450.         goto end;
  451.     }
  452.     while(1) {
  453.         int len;
  454.  
  455.         pthread_mutex_unlock(&s->mutex);
  456.         /* Blocking operations are always cancellation points;
  457.            see "General Information" / "Thread Cancelation Overview"
  458.            in Single Unix. */
  459.         pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &old_cancelstate);
  460.         len = recv(s->udp_fd, s->tmp+4, sizeof(s->tmp)-4, 0);
  461.         pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancelstate);
  462.         pthread_mutex_lock(&s->mutex);
  463.         if (len < 0) {
  464.             if (ff_neterrno() != AVERROR(EAGAIN) && ff_neterrno() != AVERROR(EINTR)) {
  465.                 s->circular_buffer_error = ff_neterrno();
  466.                 goto end;
  467.             }
  468.             continue;
  469.         }
  470.         AV_WL32(s->tmp, len);
  471.  
  472.         if(av_fifo_space(s->fifo) < len + 4) {
  473.             /* No Space left */
  474.             if (s->overrun_nonfatal) {
  475.                 av_log(h, AV_LOG_WARNING, "Circular buffer overrun. "
  476.                         "Surviving due to overrun_nonfatal option\n");
  477.                 continue;
  478.             } else {
  479.                 av_log(h, AV_LOG_ERROR, "Circular buffer overrun. "
  480.                         "To avoid, increase fifo_size URL option. "
  481.                         "To survive in such case, use overrun_nonfatal option\n");
  482.                 s->circular_buffer_error = AVERROR(EIO);
  483.                 goto end;
  484.             }
  485.         }
  486.         av_fifo_generic_write(s->fifo, s->tmp, len+4, NULL);
  487.         pthread_cond_signal(&s->cond);
  488.     }
  489.  
  490. end:
  491.     pthread_cond_signal(&s->cond);
  492.     pthread_mutex_unlock(&s->mutex);
  493.     return NULL;
  494. }
  495. #endif
  496.  
  497. static int parse_source_list(char *buf, char **sources, int *num_sources,
  498.                              int max_sources)
  499. {
  500.     char *source_start;
  501.  
  502.     source_start = buf;
  503.     while (1) {
  504.         char *next = strchr(source_start, ',');
  505.         if (next)
  506.             *next = '\0';
  507.         sources[*num_sources] = av_strdup(source_start);
  508.         if (!sources[*num_sources])
  509.             return AVERROR(ENOMEM);
  510.         source_start = next + 1;
  511.         (*num_sources)++;
  512.         if (*num_sources >= max_sources || !next)
  513.             break;
  514.     }
  515.     return 0;
  516. }
  517.  
  518. /* put it in UDP context */
  519. /* return non zero if error */
  520. static int udp_open(URLContext *h, const char *uri, int flags)
  521. {
  522.     char hostname[1024], localaddr[1024] = "";
  523.     int port, udp_fd = -1, tmp, bind_ret = -1;
  524.     UDPContext *s = h->priv_data;
  525.     int is_output;
  526.     const char *p;
  527.     char buf[256];
  528.     struct sockaddr_storage my_addr;
  529.     socklen_t len;
  530.     int reuse_specified = 0;
  531.     int i, num_include_sources = 0, num_exclude_sources = 0;
  532.     char *include_sources[32], *exclude_sources[32];
  533.  
  534.     h->is_streamed = 1;
  535.  
  536.     is_output = !(flags & AVIO_FLAG_READ);
  537.     if (!s->buffer_size) /* if not set explicitly */
  538.         s->buffer_size = is_output ? UDP_TX_BUF_SIZE : UDP_MAX_PKT_SIZE;
  539.  
  540.     p = strchr(uri, '?');
  541.     if (p) {
  542.         if (av_find_info_tag(buf, sizeof(buf), "reuse", p)) {
  543.             char *endptr = NULL;
  544.             s->reuse_socket = strtol(buf, &endptr, 10);
  545.             /* assume if no digits were found it is a request to enable it */
  546.             if (buf == endptr)
  547.                 s->reuse_socket = 1;
  548.             reuse_specified = 1;
  549.         }
  550.         if (av_find_info_tag(buf, sizeof(buf), "overrun_nonfatal", p)) {
  551.             char *endptr = NULL;
  552.             s->overrun_nonfatal = strtol(buf, &endptr, 10);
  553.             /* assume if no digits were found it is a request to enable it */
  554.             if (buf == endptr)
  555.                 s->overrun_nonfatal = 1;
  556.             if (!HAVE_PTHREAD_CANCEL)
  557.                 av_log(h, AV_LOG_WARNING,
  558.                        "'overrun_nonfatal' option was set but it is not supported "
  559.                        "on this build (pthread support is required)\n");
  560.         }
  561.         if (av_find_info_tag(buf, sizeof(buf), "ttl", p)) {
  562.             s->ttl = strtol(buf, NULL, 10);
  563.         }
  564.         if (av_find_info_tag(buf, sizeof(buf), "localport", p)) {
  565.             s->local_port = strtol(buf, NULL, 10);
  566.         }
  567.         if (av_find_info_tag(buf, sizeof(buf), "pkt_size", p)) {
  568.             s->packet_size = strtol(buf, NULL, 10);
  569.         }
  570.         if (av_find_info_tag(buf, sizeof(buf), "buffer_size", p)) {
  571.             s->buffer_size = strtol(buf, NULL, 10);
  572.         }
  573.         if (av_find_info_tag(buf, sizeof(buf), "connect", p)) {
  574.             s->is_connected = strtol(buf, NULL, 10);
  575.         }
  576.         if (av_find_info_tag(buf, sizeof(buf), "fifo_size", p)) {
  577.             s->circular_buffer_size = strtol(buf, NULL, 10);
  578.             if (!HAVE_PTHREAD_CANCEL)
  579.                 av_log(h, AV_LOG_WARNING,
  580.                        "'circular_buffer_size' option was set but it is not supported "
  581.                        "on this build (pthread support is required)\n");
  582.         }
  583.         if (av_find_info_tag(buf, sizeof(buf), "localaddr", p)) {
  584.             av_strlcpy(localaddr, buf, sizeof(localaddr));
  585.         }
  586.         if (av_find_info_tag(buf, sizeof(buf), "sources", p)) {
  587.             if (parse_source_list(buf, include_sources, &num_include_sources,
  588.                                   FF_ARRAY_ELEMS(include_sources)))
  589.                 goto fail;
  590.         }
  591.         if (av_find_info_tag(buf, sizeof(buf), "block", p)) {
  592.             if (parse_source_list(buf, exclude_sources, &num_exclude_sources,
  593.                                   FF_ARRAY_ELEMS(exclude_sources)))
  594.                 goto fail;
  595.         }
  596.         if (!is_output && av_find_info_tag(buf, sizeof(buf), "timeout", p))
  597.             s->timeout = strtol(buf, NULL, 10);
  598.     }
  599.     /* handling needed to support options picking from both AVOption and URL */
  600.     s->circular_buffer_size *= 188;
  601.     if (flags & AVIO_FLAG_WRITE) {
  602.         h->max_packet_size = s->packet_size;
  603.     } else {
  604.         h->max_packet_size = UDP_MAX_PKT_SIZE;
  605.     }
  606.     h->rw_timeout = s->timeout;
  607.  
  608.     /* fill the dest addr */
  609.     av_url_split(NULL, 0, NULL, 0, hostname, sizeof(hostname), &port, NULL, 0, uri);
  610.  
  611.     /* XXX: fix av_url_split */
  612.     if (hostname[0] == '\0' || hostname[0] == '?') {
  613.         /* only accepts null hostname if input */
  614.         if (!(flags & AVIO_FLAG_READ))
  615.             goto fail;
  616.     } else {
  617.         if (ff_udp_set_remote_url(h, uri) < 0)
  618.             goto fail;
  619.     }
  620.  
  621.     if ((s->is_multicast || !s->local_port) && (h->flags & AVIO_FLAG_READ))
  622.         s->local_port = port;
  623.     udp_fd = udp_socket_create(s, &my_addr, &len, localaddr[0] ? localaddr : s->local_addr);
  624.     if (udp_fd < 0)
  625.         goto fail;
  626.  
  627.     /* Follow the requested reuse option, unless it's multicast in which
  628.      * case enable reuse unless explicitly disabled.
  629.      */
  630.     if (s->reuse_socket || (s->is_multicast && !reuse_specified)) {
  631.         s->reuse_socket = 1;
  632.         if (setsockopt (udp_fd, SOL_SOCKET, SO_REUSEADDR, &(s->reuse_socket), sizeof(s->reuse_socket)) != 0)
  633.             goto fail;
  634.     }
  635.  
  636.     /* If multicast, try binding the multicast address first, to avoid
  637.      * receiving UDP packets from other sources aimed at the same UDP
  638.      * port. This fails on windows. This makes sending to the same address
  639.      * using sendto() fail, so only do it if we're opened in read-only mode. */
  640.     if (s->is_multicast && !(h->flags & AVIO_FLAG_WRITE)) {
  641.         bind_ret = bind(udp_fd,(struct sockaddr *)&s->dest_addr, len);
  642.     }
  643.     /* bind to the local address if not multicast or if the multicast
  644.      * bind failed */
  645.     /* the bind is needed to give a port to the socket now */
  646.     if (bind_ret < 0 && bind(udp_fd,(struct sockaddr *)&my_addr, len) < 0) {
  647.         log_net_error(h, AV_LOG_ERROR, "bind failed");
  648.         goto fail;
  649.     }
  650.  
  651.     len = sizeof(my_addr);
  652.     getsockname(udp_fd, (struct sockaddr *)&my_addr, &len);
  653.     s->local_port = udp_port(&my_addr, len);
  654.  
  655.     if (s->is_multicast) {
  656.         if (h->flags & AVIO_FLAG_WRITE) {
  657.             /* output */
  658.             if (udp_set_multicast_ttl(udp_fd, s->ttl, (struct sockaddr *)&s->dest_addr) < 0)
  659.                 goto fail;
  660.         }
  661.         if (h->flags & AVIO_FLAG_READ) {
  662.             /* input */
  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");
  665.                 goto fail;
  666.             }
  667.             if (num_include_sources) {
  668.                 if (udp_set_multicast_sources(udp_fd, (struct sockaddr *)&s->dest_addr, s->dest_addr_len, include_sources, num_include_sources, 1) < 0)
  669.                     goto fail;
  670.             } else {
  671.                 if (udp_join_multicast_group(udp_fd, (struct sockaddr *)&s->dest_addr) < 0)
  672.                     goto fail;
  673.             }
  674.             if (num_exclude_sources) {
  675.                 if (udp_set_multicast_sources(udp_fd, (struct sockaddr *)&s->dest_addr, s->dest_addr_len, exclude_sources, num_exclude_sources, 0) < 0)
  676.                     goto fail;
  677.             }
  678.         }
  679.     }
  680.  
  681.     if (is_output) {
  682.         /* limit the tx buf size to limit latency */
  683.         tmp = s->buffer_size;
  684.         if (setsockopt(udp_fd, SOL_SOCKET, SO_SNDBUF, &tmp, sizeof(tmp)) < 0) {
  685.             log_net_error(h, AV_LOG_ERROR, "setsockopt(SO_SNDBUF)");
  686.             goto fail;
  687.         }
  688.     } else {
  689.         /* set udp recv buffer size to the largest possible udp packet size to
  690.          * avoid losing data on OSes that set this too low by default. */
  691.         tmp = s->buffer_size;
  692.         if (setsockopt(udp_fd, SOL_SOCKET, SO_RCVBUF, &tmp, sizeof(tmp)) < 0) {
  693.             log_net_error(h, AV_LOG_WARNING, "setsockopt(SO_RECVBUF)");
  694.         }
  695.         /* make the socket non-blocking */
  696.         ff_socket_nonblock(udp_fd, 1);
  697.     }
  698.     if (s->is_connected) {
  699.         if (connect(udp_fd, (struct sockaddr *) &s->dest_addr, s->dest_addr_len)) {
  700.             log_net_error(h, AV_LOG_ERROR, "connect");
  701.             goto fail;
  702.         }
  703.     }
  704.  
  705.     for (i = 0; i < num_include_sources; i++)
  706.         av_freep(&include_sources[i]);
  707.     for (i = 0; i < num_exclude_sources; i++)
  708.         av_freep(&exclude_sources[i]);
  709.  
  710.     s->udp_fd = udp_fd;
  711.  
  712. #if HAVE_PTHREAD_CANCEL
  713.     if (!is_output && s->circular_buffer_size) {
  714.         int ret;
  715.  
  716.         /* start the task going */
  717.         s->fifo = av_fifo_alloc(s->circular_buffer_size);
  718.         ret = pthread_mutex_init(&s->mutex, NULL);
  719.         if (ret != 0) {
  720.             av_log(h, AV_LOG_ERROR, "pthread_mutex_init failed : %s\n", strerror(ret));
  721.             goto fail;
  722.         }
  723.         ret = pthread_cond_init(&s->cond, NULL);
  724.         if (ret != 0) {
  725.             av_log(h, AV_LOG_ERROR, "pthread_cond_init failed : %s\n", strerror(ret));
  726.             goto cond_fail;
  727.         }
  728.         ret = pthread_create(&s->circular_buffer_thread, NULL, circular_buffer_task, h);
  729.         if (ret != 0) {
  730.             av_log(h, AV_LOG_ERROR, "pthread_create failed : %s\n", strerror(ret));
  731.             goto thread_fail;
  732.         }
  733.         s->thread_started = 1;
  734.     }
  735. #endif
  736.  
  737.     return 0;
  738. #if HAVE_PTHREAD_CANCEL
  739.  thread_fail:
  740.     pthread_cond_destroy(&s->cond);
  741.  cond_fail:
  742.     pthread_mutex_destroy(&s->mutex);
  743. #endif
  744.  fail:
  745.     if (udp_fd >= 0)
  746.         closesocket(udp_fd);
  747.     av_fifo_free(s->fifo);
  748.     for (i = 0; i < num_include_sources; i++)
  749.         av_freep(&include_sources[i]);
  750.     for (i = 0; i < num_exclude_sources; i++)
  751.         av_freep(&exclude_sources[i]);
  752.     return AVERROR(EIO);
  753. }
  754.  
  755. static int udp_read(URLContext *h, uint8_t *buf, int size)
  756. {
  757.     UDPContext *s = h->priv_data;
  758.     int ret;
  759.     int avail, nonblock = h->flags & AVIO_FLAG_NONBLOCK;
  760.  
  761. #if HAVE_PTHREAD_CANCEL
  762.     if (s->fifo) {
  763.         pthread_mutex_lock(&s->mutex);
  764.         do {
  765.             avail = av_fifo_size(s->fifo);
  766.             if (avail) { // >=size) {
  767.                 uint8_t tmp[4];
  768.  
  769.                 av_fifo_generic_read(s->fifo, tmp, 4, NULL);
  770.                 avail= AV_RL32(tmp);
  771.                 if(avail > size){
  772.                     av_log(h, AV_LOG_WARNING, "Part of datagram lost due to insufficient buffer size\n");
  773.                     avail= size;
  774.                 }
  775.  
  776.                 av_fifo_generic_read(s->fifo, buf, avail, NULL);
  777.                 av_fifo_drain(s->fifo, AV_RL32(tmp) - avail);
  778.                 pthread_mutex_unlock(&s->mutex);
  779.                 return avail;
  780.             } else if(s->circular_buffer_error){
  781.                 int err = s->circular_buffer_error;
  782.                 pthread_mutex_unlock(&s->mutex);
  783.                 return err;
  784.             } else if(nonblock) {
  785.                 pthread_mutex_unlock(&s->mutex);
  786.                 return AVERROR(EAGAIN);
  787.             }
  788.             else {
  789.                 /* FIXME: using the monotonic clock would be better,
  790.                    but it does not exist on all supported platforms. */
  791.                 int64_t t = av_gettime() + 100000;
  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) {
  795.                     pthread_mutex_unlock(&s->mutex);
  796.                     return AVERROR(errno == ETIMEDOUT ? EAGAIN : errno);
  797.                 }
  798.                 nonblock = 1;
  799.             }
  800.         } while( 1);
  801.     }
  802. #endif
  803.  
  804.     if (!(h->flags & AVIO_FLAG_NONBLOCK)) {
  805.         ret = ff_network_wait_fd(s->udp_fd, 0);
  806.         if (ret < 0)
  807.             return ret;
  808.     }
  809.     ret = recv(s->udp_fd, buf, size, 0);
  810.  
  811.     return ret < 0 ? ff_neterrno() : ret;
  812. }
  813.  
  814. static int udp_write(URLContext *h, const uint8_t *buf, int size)
  815. {
  816.     UDPContext *s = h->priv_data;
  817.     int ret;
  818.  
  819.     if (!(h->flags & AVIO_FLAG_NONBLOCK)) {
  820.         ret = ff_network_wait_fd(s->udp_fd, 1);
  821.         if (ret < 0)
  822.             return ret;
  823.     }
  824.  
  825.     if (!s->is_connected) {
  826.         ret = sendto (s->udp_fd, buf, size, 0,
  827.                       (struct sockaddr *) &s->dest_addr,
  828.                       s->dest_addr_len);
  829.     } else
  830.         ret = send(s->udp_fd, buf, size, 0);
  831.  
  832.     return ret < 0 ? ff_neterrno() : ret;
  833. }
  834.  
  835. static int udp_close(URLContext *h)
  836. {
  837.     UDPContext *s = h->priv_data;
  838.     int ret;
  839.  
  840.     if (s->is_multicast && (h->flags & AVIO_FLAG_READ))
  841.         udp_leave_multicast_group(s->udp_fd, (struct sockaddr *)&s->dest_addr);
  842.     closesocket(s->udp_fd);
  843. #if HAVE_PTHREAD_CANCEL
  844.     if (s->thread_started) {
  845.         pthread_cancel(s->circular_buffer_thread);
  846.         ret = pthread_join(s->circular_buffer_thread, NULL);
  847.         if (ret != 0)
  848.             av_log(h, AV_LOG_ERROR, "pthread_join(): %s\n", strerror(ret));
  849.         pthread_mutex_destroy(&s->mutex);
  850.         pthread_cond_destroy(&s->cond);
  851.     }
  852. #endif
  853.     av_fifo_free(s->fifo);
  854.     return 0;
  855. }
  856.  
  857. URLProtocol ff_udp_protocol = {
  858.     .name                = "udp",
  859.     .url_open            = udp_open,
  860.     .url_read            = udp_read,
  861.     .url_write           = udp_write,
  862.     .url_close           = udp_close,
  863.     .url_get_file_handle = udp_get_file_handle,
  864.     .priv_data_size      = sizeof(UDPContext),
  865.     .priv_data_class     = &udp_context_class,
  866.     .flags               = URL_PROTOCOL_FLAG_NETWORK,
  867. };
  868.