git.sesse.net Git - ffmpeg/blob - libavformat/udp.c

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