Merge remote-tracking branch 'cigaes/master'

* cigaes/master:
  ffmpeg: notify when the thread message queue blocks.
  ffmpeg: allow to set the thread message queue size.
  lavd/alsa: set frame_size field.

Merged-by: Michael Niedermayer <michaelni@gmx.at>

5 files changed, 26 insertions, 2 deletions

diff --git a/doc/ffmpeg.texi b/doc/ffmpeg.texi
index 6e51c44f29..6772f2f685 100644
--- a/doc/ffmpeg.texi
+++ b/doc/ffmpeg.texi
@@ -1158,6 +1158,12 @@ This option enables or disables accurate seeking in input files with the
 transcoding. Use @option{-noaccurate_seek} to disable it, which may be useful
 e.g. when copying some streams and transcoding the others.
 
+@item -thread_message_queue @var{size} (@emph{input})
+This option sets the maximum number of queued packets when reading from the
+file or device. With low latency / high rate live streams, packets may be
+discarded if they are not read in a timely manner; raising this value can
+avoid it.
+
 @item -override_ffserver (@emph{global})
 Overrides the input specifications from @command{ffserver}. Using this
 option you can map any input stream to @command{ffserver} and control
diff --git a/ffmpeg.c b/ffmpeg.c
index becd5dfc33..c6f3dcdbde 100644
--- a/ffmpeg.c
+++ b/ffmpeg.c
@@ -3356,6 +3356,7 @@ static int check_keyboard_interaction(int64_t cur_time)
 static void *input_thread(void *arg)
 {
     InputFile *f = arg;
+    unsigned flags = f->non_blocking ? AV_THREAD_MESSAGE_NONBLOCK : 0;
     int ret = 0;
 
     while (1) {
@@ -3371,7 +3372,15 @@ static void *input_thread(void *arg)
             break;
         }
         av_dup_packet(&pkt);
-        ret = av_thread_message_queue_send(f->in_thread_queue, &pkt, 0);
+        ret = av_thread_message_queue_send(f->in_thread_queue, &pkt, flags);
+        if (flags && ret == AVERROR(EAGAIN)) {
+            flags = 0;
+            ret = av_thread_message_queue_send(f->in_thread_queue, &pkt, flags);
+            av_log(f->ctx, AV_LOG_WARNING,
+                   "Thread message queue blocking; consider raising the "
+                   "thread_queue_size option (current value: %d)\n",
+                   f->thread_queue_size);
+        }
         if (ret < 0) {
             if (ret != AVERROR_EOF)
                 av_log(f->ctx, AV_LOG_ERROR,
@@ -3420,7 +3429,7 @@ static int init_input_threads(void)
             strcmp(f->ctx->iformat->name, "lavfi"))
             f->non_blocking = 1;
         ret = av_thread_message_queue_alloc(&f->in_thread_queue,
-                                            8, sizeof(AVPacket));
+                                            f->thread_queue_size, sizeof(AVPacket));
         if (ret < 0)
             return ret;
 
diff --git a/ffmpeg.h b/ffmpeg.h
index d2e0c5c3e9..7c0c22cfad 100644
--- a/ffmpeg.h
+++ b/ffmpeg.h
@@ -111,6 +111,7 @@ typedef struct OptionsContext {
     int64_t input_ts_offset;
     int rate_emu;
     int accurate_seek;
+    int thread_queue_size;
 
     SpecifierOpt *ts_scale;
     int        nb_ts_scale;
@@ -350,6 +351,7 @@ typedef struct InputFile {
     pthread_t thread;           /* thread reading from this file */
     int non_blocking;           /* reading packets from the thread should not block */
     int joined;                 /* the thread has been joined */
+    int thread_queue_size;      /* maximum number of queued packets */
 #endif
 } InputFile;
 
diff --git a/ffmpeg_opt.c b/ffmpeg_opt.c
index 09e6e33b5c..39c5f49d67 100644
--- a/ffmpeg_opt.c
+++ b/ffmpeg_opt.c
@@ -951,6 +951,9 @@ static int open_input_file(OptionsContext *o, const char *filename)
     f->nb_streams = ic->nb_streams;
     f->rate_emu   = o->rate_emu;
     f->accurate_seek = o->accurate_seek;
+#if HAVE_PTHREADS
+    f->thread_queue_size = o->thread_queue_size > 0 ? o->thread_queue_size : 8;
+#endif
 
     /* check if all codec options have been used */
     unused_opts = strip_specifiers(o->g->codec_opts);
@@ -2957,6 +2960,9 @@ const OptionDef options[] = {
     { "disposition",    OPT_STRING | HAS_ARG | OPT_SPEC |
                         OPT_OUTPUT,                                  { .off = OFFSET(disposition) },
         "disposition", "" },
+    { "thread_queue_size", HAS_ARG | OPT_INT | OPT_OFFSET | OPT_EXPERT | OPT_INPUT,
+                                                                     { .off = OFFSET(thread_queue_size) },
+        "set the maximum number of queued packets from the demuxer" },
 
     /* video options */
     { "vframes",      OPT_VIDEO | HAS_ARG  | OPT_PERFILE | OPT_OUTPUT,           { .func_arg = opt_video_frames },
diff --git a/libavdevice/alsa-audio-dec.c b/libavdevice/alsa-audio-dec.c
index 7f8f8cdd59..efa6747678 100644
--- a/libavdevice/alsa-audio-dec.c
+++ b/libavdevice/alsa-audio-dec.c
@@ -80,6 +80,7 @@ static av_cold int audio_read_header(AVFormatContext *s1)
     st->codec->codec_id    = codec_id;
     st->codec->sample_rate = s->sample_rate;
     st->codec->channels    = s->channels;
+    st->codec->frame_size = s->frame_size;
     avpriv_set_pts_info(st, 64, 1, 1000000);  /* 64 bits pts in us */
     /* microseconds instead of seconds, MHz instead of Hz */
     s->timefilter = ff_timefilter_new(1000000.0 / s->sample_rate,