Merge commit '7d99c92945e4b2d12ed2126365482e1900340da0'
authorMichael Niedermayer <michaelni@gmx.at>
Tue, 30 Jul 2013 09:40:05 +0000 (11:40 +0200)
committerMichael Niedermayer <michaelni@gmx.at>
Tue, 30 Jul 2013 09:40:05 +0000 (11:40 +0200)
* commit '7d99c92945e4b2d12ed2126365482e1900340da0':
  udp: Keep track of include and exclude sources separately

Conflicts:
libavformat/udp.c

Merged-by: Michael Niedermayer <michaelni@gmx.at>
1  2 
libavformat/udp.c

@@@ -435,65 -378,27 +435,86 @@@ static int udp_get_file_handle(URLConte
      return s->udp_fd;
  }
  
 +#if HAVE_PTHREAD_CANCEL
 +static void *circular_buffer_task( void *_URLContext)
 +{
 +    URLContext *h = _URLContext;
 +    UDPContext *s = h->priv_data;
 +    int old_cancelstate;
 +
 +    pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancelstate);
 +    pthread_mutex_lock(&s->mutex);
 +    if (ff_socket_nonblock(s->udp_fd, 0) < 0) {
 +        av_log(h, AV_LOG_ERROR, "Failed to set blocking mode");
 +        s->circular_buffer_error = AVERROR(EIO);
 +        goto end;
 +    }
 +    while(1) {
 +        int len;
 +
 +        pthread_mutex_unlock(&s->mutex);
 +        /* Blocking operations are always cancellation points;
 +           see "General Information" / "Thread Cancelation Overview"
 +           in Single Unix. */
 +        pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &old_cancelstate);
 +        len = recv(s->udp_fd, s->tmp+4, sizeof(s->tmp)-4, 0);
 +        pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old_cancelstate);
 +        pthread_mutex_lock(&s->mutex);
 +        if (len < 0) {
 +            if (ff_neterrno() != AVERROR(EAGAIN) && ff_neterrno() != AVERROR(EINTR)) {
 +                s->circular_buffer_error = ff_neterrno();
 +                goto end;
 +            }
 +            continue;
 +        }
 +        AV_WL32(s->tmp, len);
 +
 +        if(av_fifo_space(s->fifo) < len + 4) {
 +            /* No Space left */
 +            if (s->overrun_nonfatal) {
 +                av_log(h, AV_LOG_WARNING, "Circular buffer overrun. "
 +                        "Surviving due to overrun_nonfatal option\n");
 +                continue;
 +            } else {
 +                av_log(h, AV_LOG_ERROR, "Circular buffer overrun. "
 +                        "To avoid, increase fifo_size URL option. "
 +                        "To survive in such case, use overrun_nonfatal option\n");
 +                s->circular_buffer_error = AVERROR(EIO);
 +                goto end;
 +            }
 +        }
 +        av_fifo_generic_write(s->fifo, s->tmp, len+4, NULL);
 +        pthread_cond_signal(&s->cond);
 +    }
 +
 +end:
 +    pthread_cond_signal(&s->cond);
 +    pthread_mutex_unlock(&s->mutex);
 +    return NULL;
 +}
 +#endif
 +
+ static int parse_source_list(char *buf, char **sources, int *num_sources,
+                              int max_sources)
+ {
+     char *source_start;
+     source_start = buf;
+     while (1) {
+         char *next = strchr(source_start, ',');
+         if (next)
+             *next = '\0';
+         sources[*num_sources] = av_strdup(source_start);
+         if (!sources[*num_sources])
+             return AVERROR(ENOMEM);
+         source_start = next + 1;
+         (*num_sources)++;
+         if (*num_sources >= max_sources || !next)
+             break;
+     }
+     return 0;
+ }
  /* put it in UDP context */
  /* return non zero if error */
  static int udp_open(URLContext *h, const char *uri, int flags)
      struct sockaddr_storage my_addr;
      socklen_t len;
      int reuse_specified = 0;
-     int i, include = 0, num_sources = 0;
-     char *sources[32];
+     int i, num_include_sources = 0, num_exclude_sources = 0;
+     char *include_sources[32], *exclude_sources[32];
  
      h->is_streamed = 1;
 -    h->max_packet_size = 1472;
  
      is_output = !(flags & AVIO_FLAG_READ);
 -
 -    s->ttl = 16;
 -    s->buffer_size = is_output ? UDP_TX_BUF_SIZE : UDP_MAX_PKT_SIZE;
 +    if (!s->buffer_size) /* if not set explicitly */
 +        s->buffer_size = is_output ? UDP_TX_BUF_SIZE : UDP_MAX_PKT_SIZE;
  
      p = strchr(uri, '?');
      if (p) {
          if (av_find_info_tag(buf, sizeof(buf), "localaddr", p)) {
              av_strlcpy(localaddr, buf, sizeof(localaddr));
          }
-         if (av_find_info_tag(buf, sizeof(buf), "sources", p))
-             include = 1;
-         if (include || av_find_info_tag(buf, sizeof(buf), "block", p)) {
-             char *source_start;
-             source_start = buf;
-             while (1) {
-                 char *next = strchr(source_start, ',');
-                 if (next)
-                     *next = '\0';
-                 sources[num_sources] = av_strdup(source_start);
-                 if (!sources[num_sources])
-                     goto fail;
-                 source_start = next + 1;
-                 num_sources++;
-                 if (num_sources >= FF_ARRAY_ELEMS(sources) || !next)
-                     break;
-             }
+         if (av_find_info_tag(buf, sizeof(buf), "sources", p)) {
+             if (parse_source_list(buf, include_sources, &num_include_sources,
+                                   FF_ARRAY_ELEMS(include_sources)))
+                 goto fail;
+         }
+         if (av_find_info_tag(buf, sizeof(buf), "block", p)) {
+             if (parse_source_list(buf, exclude_sources, &num_exclude_sources,
+                                   FF_ARRAY_ELEMS(exclude_sources)))
+                 goto fail;
          }
 +        if (!is_output && av_find_info_tag(buf, sizeof(buf), "timeout", p))
 +            s->timeout = strtol(buf, NULL, 10);
 +    }
 +    /* handling needed to support options picking from both AVOption and URL */
 +    s->circular_buffer_size *= 188;
 +    if (flags & AVIO_FLAG_WRITE) {
 +        h->max_packet_size = s->packet_size;
 +    } else {
 +        h->max_packet_size = UDP_MAX_PKT_SIZE;
      }
 +    h->rw_timeout = s->timeout;
  
      /* fill the dest addr */
      av_url_split(NULL, 0, NULL, 0, hostname, sizeof(hostname), &port, NULL, 0, uri);
          }
      }
  
-     for (i = 0; i < num_sources; i++)
-         av_freep(&sources[i]);
+     for (i = 0; i < num_include_sources; i++)
+         av_freep(&include_sources[i]);
+     for (i = 0; i < num_exclude_sources; i++)
+         av_freep(&exclude_sources[i]);
  
      s->udp_fd = udp_fd;
 +
 +#if HAVE_PTHREAD_CANCEL
 +    if (!is_output && s->circular_buffer_size) {
 +        int ret;
 +
 +        /* start the task going */
 +        s->fifo = av_fifo_alloc(s->circular_buffer_size);
 +        ret = pthread_mutex_init(&s->mutex, NULL);
 +        if (ret != 0) {
 +            av_log(h, AV_LOG_ERROR, "pthread_mutex_init failed : %s\n", strerror(ret));
 +            goto fail;
 +        }
 +        ret = pthread_cond_init(&s->cond, NULL);
 +        if (ret != 0) {
 +            av_log(h, AV_LOG_ERROR, "pthread_cond_init failed : %s\n", strerror(ret));
 +            goto cond_fail;
 +        }
 +        ret = pthread_create(&s->circular_buffer_thread, NULL, circular_buffer_task, h);
 +        if (ret != 0) {
 +            av_log(h, AV_LOG_ERROR, "pthread_create failed : %s\n", strerror(ret));
 +            goto thread_fail;
 +        }
 +        s->thread_started = 1;
 +    }
 +#endif
 +
      return 0;
 +#if HAVE_PTHREAD_CANCEL
 + thread_fail:
 +    pthread_cond_destroy(&s->cond);
 + cond_fail:
 +    pthread_mutex_destroy(&s->mutex);
 +#endif
   fail:
      if (udp_fd >= 0)
          closesocket(udp_fd);
-     for (i = 0; i < num_sources; i++)
-         av_freep(&sources[i]);
 +    av_fifo_free(s->fifo);
+     for (i = 0; i < num_include_sources; i++)
+         av_freep(&include_sources[i]);
+     for (i = 0; i < num_exclude_sources; i++)
+         av_freep(&exclude_sources[i]);
      return AVERROR(EIO);
  }