From: Jan Sebechlebsky <sebechlebsky...@gmail.com>
FIFO pseudo-muxer allows to separate decoder from the
actual output by using first-in-first-out queue and
running actual muxer asynchronously in separate thread.
It can be configured to attempt transparent recovery
of output on failure.
Signed-off-by: Jan Sebechlebsky <sebechlebsky...@gmail.com>
---
configure | 1 +
doc/muxers.texi | 77 ++++++
libavformat/Makefile | 1 +
libavformat/allformats.c | 1 +
libavformat/fifo.c | 657 +++++++++++++++++++++++++++++++++++++++++++++++
5 files changed, 737 insertions(+)
create mode 100644 libavformat/fifo.c
diff --git a/configure b/configure
index 007c953..eacda09 100755
--- a/configure
+++ b/configure
@@ -2826,6 +2826,7 @@ dv_muxer_select="dvprofile"
dxa_demuxer_select="riffdec"
eac3_demuxer_select="ac3_parser"
f4v_muxer_select="mov_muxer"
+fifo_muxer_deps="pthreads"
flac_demuxer_select="flac_parser"
hds_muxer_select="flv_muxer"
hls_muxer_select="mpegts_muxer"
diff --git a/doc/muxers.texi b/doc/muxers.texi
index c2ca0ba..e545bc7 100644
--- a/doc/muxers.texi
+++ b/doc/muxers.texi
@@ -1408,6 +1408,83 @@ Specify whether to remove all fragments when finished.
Default 0 (do not remove)
@end table
+@section fifo
+
+The fifo pseudo-muxer allows to separate encoding from any other muxer
+by using first-in-first-out queue and running the actual muxer in separate
+thread. This is especially useful in combination with tee muxer
+and output to several destinations with different reliability/writing
speed/latency.
+
+The fifo muxer can operate in regular or fully non-blocking mode. This
determines
+the behaviour in case the queue fills up. In regular mode the encoding is
blocked
+until muxer processes some of the packets from queue, in non-blocking mode the
packets
+are thrown away rather than blocking the encoder (this might be useful in
real-time
+streaming scenario).
+
+@table @option
+
+@item fifo_format
+Specify the format name. Useful if it cannot be guessed from the
+output name suffix.
+
+@item queue_size
+Specify size of the queue (number of packets)
+
+@item format_opts
+Specify format options for the underlying muxer. Muxer options can be specified
+as a list of @var{key}=@var{value} pairs separated by ':'.
+
+@item block_on_overflow 0|1
+If set to 0, fully non-blocking mode will be used and in case the queue
+fills up, packets will be dropped. By default this option is set to 1,
+so in case of queue overflow the encoder will be block until muxer
+processes some of the packets.
+
+@item attempt_recovery 0|1
+If failure happens, attempt to recover the output. This is especially useful
+when used with network output, allows to restart streaming transparently.
+By default this option is turned off.
+
+@item max_recovery_attempts
+Sets maximal number of successive unsucessfull recovery attempts after which
+the output fails permanently. Unlimited if set to zero.
+
+@item recovery_wait_time
+Waiting time before the next recovery attempt after previous unsuccessfull
+recovery attempt.
+
+@item recovery_wait_streamtime 0|1
+If set to 0 (default), the real time is used when waiting for the recovery
attempt
+(i.e. the recovery will be attempted after at least recovery_wait_time
seconds).
+If set to 1, the time of the processed stream is taken into account instead
+(i.e. the recovery will be attempted after at least recovery_wait_time seconds
+of the stream is ommited).
+
+@item recover_any_error 0|1
+If set to 1, recovery will be attempted regardless of type of the error causing
+the failure (by default, in case of certain errors the recovery is not
attempted
+even when attempt_recovery is on).
+
+@item restart_with_keyframe 0|1
+Specify whether to wait for the keyframe after recovering from
+queue overflow or failure.
+
+@end table
+
+@subsection Examples
+
+@itemize
+
+@item
+Stream something to rtmp server using non-blocking mode and recover
automatically
+in case failure happens (for example the network becomes unavailable for a
moment).
+@example
+ffmpeg -re -i ... -c:v libx264 -c:a mp2 -f fifo -fifo_format flv -map 0:v -map
0:a
+ -block_on_overflow 0 -attempt_recovery 1 rtmp://example.com/live/stream_name
+@end example
+
+@end itemize
+
@section tee
The tee muxer can be used to write the same data to several files or any
diff --git a/libavformat/Makefile b/libavformat/Makefile
index c49f9de..42fb9be 100644
--- a/libavformat/Makefile
+++ b/libavformat/Makefile
@@ -162,6 +162,7 @@ OBJS-$(CONFIG_FFM_DEMUXER) += ffmdec.o
OBJS-$(CONFIG_FFM_MUXER) += ffmenc.o
OBJS-$(CONFIG_FFMETADATA_DEMUXER) += ffmetadec.o
OBJS-$(CONFIG_FFMETADATA_MUXER) += ffmetaenc.o
+OBJS-$(CONFIG_FIFO_MUXER) += fifo.o
OBJS-$(CONFIG_FILMSTRIP_DEMUXER) += filmstripdec.o
OBJS-$(CONFIG_FILMSTRIP_MUXER) += filmstripenc.o
OBJS-$(CONFIG_FLAC_DEMUXER) += flacdec.o rawdec.o \
diff --git a/libavformat/allformats.c b/libavformat/allformats.c
index d490cc4..b21a3de 100644
--- a/libavformat/allformats.c
+++ b/libavformat/allformats.c
@@ -123,6 +123,7 @@ void av_register_all(void)
REGISTER_MUXER (F4V, f4v);
REGISTER_MUXDEMUX(FFM, ffm);
REGISTER_MUXDEMUX(FFMETADATA, ffmetadata);
+ REGISTER_MUXER (FIFO, fifo);
REGISTER_MUXDEMUX(FILMSTRIP, filmstrip);
REGISTER_MUXDEMUX(FLAC, flac);
REGISTER_DEMUXER (FLIC, flic);
diff --git a/libavformat/fifo.c b/libavformat/fifo.c
new file mode 100644
index 0000000..37e9f82
--- /dev/null
+++ b/libavformat/fifo.c
@@ -0,0 +1,657 @@
+/*
+ * FIFO pseudo-muxer
+ * Copyright (c) 2016 Jan Sebechlebsky
+ *
+ * This file is part of FFmpeg.
+ *
+ * FFmpeg is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public License
+ * as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * FFmpeg is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with FFmpeg; if not, write to the Free Software * Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+#include "libavutil/opt.h"
+#include "libavutil/time.h"
+#include "libavutil/threadmessage.h"
+#include "avformat.h"
+#include "internal.h"
+#include "pthread.h"
+
+#define FIFO_DEFAULT_QUEUE_SIZE 60
+#define FIFO_DEFAULT_MAX_RECOVERY_ATTEMPTS 16
+#define FIFO_DEFAULT_RECOVERY_WAIT_TIME 1000000 // 1 second
+
+typedef struct FifoContext {
+ const AVClass *class;
+ AVFormatContext *avf;
+
+ char *format;
+ AVOutputFormat *oformat;
+
+ char *format_options_str;
+ AVDictionary *format_options;
+
+ /* Whether to block in case the queue is full. */
+ uint8_t block_on_overflow;
+
+ /* Whether to wait for keyframe when recovering
+ * from failure or queue overflow */
+ uint8_t restart_with_keyframe;
+
+ /* Whether to attempt recovery from failure */
+ uint8_t attempt_recovery;
+
+ /* Maximal number of unsuccessful successive recovery attempts */
+ int max_recovery_attempts;
+
+ /* Number of current recovery process
+ * Value > 0 means we are in recovery process */
+ int recovery_nr;
+
+ /* Time to wait before next recovery attempt
+ * This can refer to the time in processed stream,
+ * or real time.*/
+ int64_t recovery_wait_time;
+
+ /* If >0 stream time will be used when waiting
+ * for the recovery attempt instead of real time */
+ uint8_t recovery_wait_streamtime;
+
+ /* If >0 recovery will be attempted regardless of error code
+ * (except AVERROR_EXIT, so exit request is never ignored)*/
+ uint8_t recover_any_error;
+
+ /* Timestamp of last failure.
+ * This is either pts in case stream time is used,
+ * or microseconds as returned by av_getttime_relative()*/
+ int64_t last_recovery_ts;
+
+ int queue_size;
+ AVThreadMessageQueue *queue;
+ pthread_t writer_thread;
+
+ /* Value > 0 signalizes the previous write_header call was successful
+ * so finalization by calling write_trailer and ff_io_close must be done
+ * before exiting / reinitialization of underlying muxer */
+ uint8_t header_written;
+
+ /* If > 0 all frames will be dropped until keyframe is received */
+ uint8_t drop_until_keyframe;
+
+ /* Return value of last write_trailer_call */
+ int write_trailer_ret;
+
+ pthread_mutex_t overflow_flag_lock;
+ /* Value > 0 signalizes queue overflow */
+ uint8_t overflow_flag;
+} FifoContext;
+
+typedef enum FifoMessageType {
+ FIFO_WRITE_HEADER,
+ FIFO_WRITE_PACKET,
+ FIFO_FLUSH_OUTPUT
+} FifoMessageType;
+
+typedef struct FifoMessage {
+ FifoMessageType type;
+ AVPacket pkt;
+} FifoMessage;
+
+static int fifo_thread_write_header(AVFormatContext *avf)
+{
+ int ret, i;
+ FifoContext *fifo = avf->priv_data;
+ AVFormatContext *avf2 = fifo->avf;
+ AVDictionary *format_options = NULL;
+
+ ret = av_dict_copy(&format_options, fifo->format_options, 0);
+ if (ret < 0)
+ return ret;
+
+ if (!(avf2->oformat->flags & AVFMT_NOFILE)) {
+ ret = avf2->io_open(avf2, &avf2->pb, avf->filename, AVIO_FLAG_WRITE,
&format_options);
+ if (ret < 0) {
+ av_log(avf, AV_LOG_ERROR, "Error opening %s: %s\n",
+ avf->filename, av_err2str(ret));
+ goto end;
+ }
+ }
+
+ for (i = 0;i < avf2->nb_streams; i++)
+ avf2->streams[i]->cur_dts = 0;
+
+ ret = avformat_write_header(avf2, &fifo->format_options);
+ if (!ret)
+ fifo->header_written = 1;
+
+ // Check for options unrecognized by underlying muxer
+ if (format_options) {
+ AVDictionaryEntry *entry = NULL;
+ while ((entry = av_dict_get(format_options, "", entry,
AV_DICT_IGNORE_SUFFIX)))
+ av_log(avf2, AV_LOG_ERROR, "Unknown option '%s'\n", entry->key);
+ ret = AVERROR(EINVAL);
+ }
+
+end:
+ av_dict_free(&format_options);
+ return ret;
+}
+
+static int fifo_thread_flush_output(AVFormatContext *avf)
+{
+ int ret;
+ FifoContext *fifo = avf->priv_data;
+ AVFormatContext *avf2 = fifo->avf;
+
+ ret = av_write_frame(avf2, NULL);
+ return ret;
+}
+
+static int fifo_thread_write_packet(AVFormatContext *avf, AVPacket *pkt)
+{
+ int ret, s_idx;
+ AVRational src_tb, dst_tb;
+ FifoContext *fifo = avf->priv_data;
+ AVFormatContext *avf2 = fifo->avf;
+
+ if (fifo->drop_until_keyframe) {
+ if (pkt->flags & AV_PKT_FLAG_KEY) {
+ fifo->drop_until_keyframe = 0;
+ av_log(avf, AV_LOG_INFO, "Keyframe received, recovering...\n");
+ } else {
+ av_log(avf, AV_LOG_INFO, "Dropping non-keyframe packet\n");
+ av_packet_unref(pkt);
+ return 0;
+ }
+ }
+
+ s_idx = pkt->stream_index;
+ src_tb = avf->streams[s_idx]->time_base;
+ dst_tb = avf2->streams[s_idx]->time_base;
+
+ pkt->pts = av_rescale_q(pkt->pts, src_tb, dst_tb);
+ pkt->dts = av_rescale_q(pkt->dts, src_tb, dst_tb);
+ pkt->duration = av_rescale_q(pkt->duration, src_tb, dst_tb);
+
+ ret = av_write_frame(avf2, pkt);
+ if (ret >= 0)
+ av_packet_unref(pkt);
+ return ret;
+}
+
+static int fifo_thread_write_trailer(AVFormatContext *avf)
+{
+ int ret;
+ FifoContext *fifo = avf->priv_data;
+ AVFormatContext *avf2 = fifo->avf;
+
+ if (!fifo->header_written)
+ return 0;
+
+ ret = av_write_trailer(avf2);
+ ff_format_io_close(avf2, &avf2->pb);
+
+ return ret;
+}
+
+static int fifo_dispatch_message(AVFormatContext *avf, FifoMessage *msg)
+{
+ int ret;
+
+ switch (msg->type) {
+ case FIFO_WRITE_HEADER:
+ ret = fifo_thread_write_header(avf);
+ break;
+ case FIFO_WRITE_PACKET:
+ ret = fifo_thread_write_packet(avf, &msg->pkt);
+ break;
+ case FIFO_FLUSH_OUTPUT:
+ ret = fifo_thread_flush_output(avf);
+ break;
+ }
+
+ return ret;
+}
+
+static int is_recoverable(FifoContext *fifo, int err_no) {
+ if (!fifo->attempt_recovery)
+ return 0;
+
+ if (fifo->recover_any_error)
+ return err_no != AVERROR_EXIT;
+
+ switch (err_no) {
+ case AVERROR(EINVAL):
+ case AVERROR(ENOSYS):
+ case AVERROR_EOF:
+ case AVERROR_EXIT:
+ case AVERROR_PATCHWELCOME:
+ return 0;
+ default:
+ return 1;
+ }
+}
+
+static int fifo_attempt_recovery(AVFormatContext *avf, FifoMessage *msg, int
err_no)
+{
+ int ret;
+ FifoContext *fifo = avf->priv_data;
+ AVPacket *pkt = &msg->pkt;
+ int64_t time_since_recovery;
+
+ if (!is_recoverable(fifo, err_no)) {
+ ret = err_no;
+ goto fail;
+ }
+
+ if (fifo->header_written) {
+ fifo->write_trailer_ret = fifo_thread_write_trailer(avf);
+ fifo->header_written = 0;
+ }
+
+ if (!fifo->recovery_nr)
+ fifo->last_recovery_ts = 0;
+ else {
+ if (fifo->recovery_wait_streamtime) {
+ AVRational tb = avf->streams[pkt->stream_index]->time_base;
+ time_since_recovery = av_rescale_q(pkt->pts -
fifo->last_recovery_ts,
+ tb, AV_TIME_BASE_Q);
+ } else {
+ time_since_recovery = av_gettime_relative() -
fifo->last_recovery_ts;
+ }
+
+ if (time_since_recovery < fifo->recovery_wait_time)
+ return AVERROR(EAGAIN);
+ }
+
+ fifo->recovery_nr++;
+
+ if (fifo->max_recovery_attempts)
+ av_log(avf, AV_LOG_INFO, "Recovery attempt #%d/%d\n",
+ fifo->recovery_nr, fifo->max_recovery_attempts);
+ else
+ av_log(avf, AV_LOG_INFO, "Recovery attempt #%d\n",
+ fifo->recovery_nr);
+
+
+ if (msg->type != FIFO_WRITE_HEADER) {
+ ret = fifo_thread_write_header(avf);
+ if (ret < 0)
+ goto recovery_fail;
+ }
+
+ if (fifo->restart_with_keyframe && !fifo->block_on_overflow)
+ fifo->drop_until_keyframe = 1;
+
+ ret = fifo_dispatch_message(avf, msg);
+ if (ret < 0) {
+ if (is_recoverable(fifo, ret))
+ goto recovery_fail;
+ else
+ goto fail;
+ } else {
+ av_log(avf, AV_LOG_INFO, "Recovery successful\n");
+ fifo->recovery_nr = 0;
+ }
+
+ return 0;
+
+recovery_fail:
+ av_log(avf, AV_LOG_INFO, "Recovery failed: %s\n",
+ av_err2str(ret));
+
+ fifo->last_recovery_ts = fifo->recovery_wait_streamtime ?
+ pkt->pts : av_gettime_relative();
+
+ if (fifo->max_recovery_attempts &&
+ fifo->recovery_nr >= fifo->max_recovery_attempts) {
+ av_log(avf, AV_LOG_ERROR,
+ "Maximal number of %d recovery attempts reached.\n",
+ fifo->max_recovery_attempts);
+ } else
+ return AVERROR(EAGAIN);
+fail:
+ if (msg->type == FIFO_WRITE_PACKET)
+ av_packet_unref(&msg->pkt);
+ return ret;
+}
+
+static int fifo_recover(AVFormatContext *avf, FifoMessage *msg, int err_no)
+{
+ int ret;
+ FifoContext *fifo = avf->priv_data;
+
+ do {
+ if (!fifo->recovery_wait_streamtime && fifo->recovery_nr > 0) {
+ int64_t time_since_recovery = av_gettime_relative() -
fifo->last_recovery_ts;
+ int64_t time_to_wait = FFMAX(0, fifo->recovery_wait_time -
time_since_recovery);
+ if (time_to_wait)
+ av_usleep(FFMIN(10000, time_to_wait));
+ }
+
+ ret = fifo_attempt_recovery(avf, msg, err_no);
+ } while(ret == AVERROR(EAGAIN) && fifo->block_on_overflow);
+
+ if (ret == AVERROR(EAGAIN) && !fifo->block_on_overflow) {
+ if (msg->type == FIFO_WRITE_PACKET)
+ av_packet_unref(&msg->pkt);
+ ret = 0;
+ }
+
+ return ret;
+}
+
+static void *fifo_writer_thread(void *data)
+{
+ int ret;
+ FifoMessage msg;
+ AVFormatContext *avf = data;
+ FifoContext *fifo = avf->priv_data;
+ AVThreadMessageQueue *queue = fifo->queue;
+
+ while (1) {
+ uint8_t just_flushed = 0;
+
+ /* Check and solve overflow */
+ pthread_mutex_lock(&fifo->overflow_flag_lock);
+ if (fifo->overflow_flag) {
+ av_thread_message_flush(queue);
+ if (fifo->restart_with_keyframe)
+ fifo->drop_until_keyframe = 1;
+ fifo->overflow_flag = 0;
+ just_flushed = 1;
+ }
+ pthread_mutex_unlock(&fifo->overflow_flag_lock);
+
+ if (just_flushed)
+ av_log(avf, AV_LOG_INFO, "FIFO queue flushed\n");
+
+ ret = av_thread_message_queue_recv(queue, &msg, 0);
+ if (ret < 0) {
+ av_thread_message_queue_set_err_send(queue, ret);
+ break;
+ }
+
+ if (!fifo->recovery_nr)
+ ret = fifo_dispatch_message(avf, &msg);
+
+ if (ret < 0 || fifo->recovery_nr > 0) {
+ int rec_ret = fifo_recover(avf, &msg, ret);
+ if (rec_ret < 0) {
+ av_thread_message_queue_set_err_send(queue, rec_ret);
+ break;
+ }
+ }
+ }
+
+ fifo->write_trailer_ret = fifo_thread_write_trailer(avf);
+
+ return NULL;
+}
+
+static int fifo_mux_init(AVFormatContext *avf)
+{
+ int ret = 0, i;
+ FifoContext *fifo = avf->priv_data;
+ AVFormatContext *avf2;
+
+ ret = avformat_alloc_output_context2(&avf2, fifo->oformat, NULL, NULL);
+ if (ret < 0) {
+ return ret;
+ }
+
+ fifo->avf = avf2;
+
+ avf2->interrupt_callback = avf->interrupt_callback;
+ avf2->max_delay = avf->max_delay;
+ ret = av_dict_copy(&avf2->metadata, avf->metadata, 0);
+ if (ret < 0)
+ return ret;
+ avf2->opaque = avf->opaque;
+ avf2->io_close = avf->io_close;
+ avf2->io_open = avf->io_open;
+ avf2->flags = avf->flags;
+
+ for (i = 0; i < avf->nb_streams; ++i) {
+ AVStream *st2, *st;
+ AVCodecParameters *ipar, *opar;
+
+ st = avf->streams[i];
+ st2 = avformat_new_stream(avf2, NULL);
+ if (!st)
+ return AVERROR(ENOMEM);
+
+ ipar = avf->streams[i]->codecpar;
+ opar = st2->codecpar;
+
+ ret = avcodec_parameters_copy(opar, ipar);
+ if (ret < 0)
+ return ret;
+ st2->id = st->id;
+ st2->r_frame_rate = st->r_frame_rate;
+ st2->time_base = st->time_base;
+ st2->start_time = st->start_time;
+ st2->duration = st->duration;
+ st2->nb_frames = st->nb_frames;
+ st2->disposition = st->disposition;
+ st2->sample_aspect_ratio = st->sample_aspect_ratio;
+ st2->avg_frame_rate = st->avg_frame_rate;
+
+ ret = av_dict_copy(&st2->metadata, st->metadata, 0);
+ if (ret < 0)
+ return ret;
+ }
+
+ return 0;
+}
+
+static void fifo_message_free(void *msg)
+{
+ FifoMessage *fifo_msg = msg;
+
+ if(fifo_msg->type == FIFO_WRITE_PACKET)
+ av_packet_unref(&fifo_msg->pkt);
+}
+
+static int fifo_init(AVFormatContext *avf)
+{
+ int ret = 0;
+ FifoContext *fifo = avf->priv_data;
+
+ if (fifo->recovery_wait_streamtime && fifo->block_on_overflow) {
+ av_log(avf, AV_LOG_ERROR, "recovery_wait_streamtime can be used only
when "
+ "--block_on_overflow is turned off\n");
+ return AVERROR(EINVAL);
+ }
+
+ if (fifo->format_options_str) {
+ ret = av_dict_parse_string(&fifo->format_options,
fifo->format_options_str,
+ "=", ":", 0);
+ if (ret < 0) {
+ av_log(avf, AV_LOG_ERROR, "Could not parse format options list
'%s'\n",
+ fifo->format_options_str);
+ return ret;
+ }
+ }
+
+ fifo->oformat = av_guess_format(fifo->format, avf->filename, NULL);
+ if (!fifo->oformat) {
+ ret = AVERROR_MUXER_NOT_FOUND;
+ return ret;
+ }
+
+ ret = fifo_mux_init(avf);
+ if (ret < 0)
+ return ret;
+
+ ret = av_thread_message_queue_alloc(&fifo->queue, (unsigned)
fifo->queue_size,
+ sizeof(FifoMessage));
+ if (!ret)
+ av_thread_message_queue_set_free_func(fifo->queue, fifo_message_free);
+
+ ret = pthread_mutex_init(&fifo->overflow_flag_lock, NULL);
+ if (ret < 0)
+ return AVERROR(ret);
+
+ return 0;
+}
+
+static int fifo_write_header(AVFormatContext *avf)
+{
+ int ret;
+ FifoContext * fifo = avf->priv_data;
+ FifoMessage message = {.type = FIFO_WRITE_HEADER};
+
+ ret = av_thread_message_queue_send(fifo->queue, &message, 0);
+ if (ret < 0)
+ return ret;
+
+ ret = pthread_create(&fifo->writer_thread, NULL, fifo_writer_thread, avf);
+ if (ret) {
+ av_log(avf, AV_LOG_ERROR, "Failed to start thread: %s\n",
+ av_err2str(AVERROR(ret)));
+ ret = AVERROR(ret);
+ }
+
+ return 0;
+}
+
+static int fifo_write_packet(AVFormatContext *avf, AVPacket *pkt)
+{
+ int ret = 0;
+ FifoContext *fifo = avf->priv_data;
+ FifoMessage msg = {.type = pkt ? FIFO_WRITE_PACKET : FIFO_FLUSH_OUTPUT};
+
+ if (pkt) {
+ memset(&msg.pkt, 0,sizeof(AVPacket));
+ ret = av_packet_ref(&msg.pkt,pkt);
+ if (ret < 0)
+ return ret;
+ }
+
+ ret = av_thread_message_queue_send(fifo->queue, &msg,
+ fifo->block_on_overflow ?
+ 0 : AV_THREAD_MESSAGE_NONBLOCK);
+ if (ret == AVERROR(EAGAIN)) {
+ uint8_t overflow_set = 0;
+
+ pthread_mutex_lock(&fifo->overflow_flag_lock);
+ if (!fifo->overflow_flag)
+ fifo->overflow_flag = overflow_set = 1;
+ pthread_mutex_unlock(&fifo->overflow_flag_lock);
+
+ if (overflow_set)
+ av_log(avf, AV_LOG_WARNING, "FIFO queue full\n");
+ ret = 0;
+ goto fail;
+ }else if (ret < 0) {
+ goto fail;
+ }
+
+ return ret;
+fail:
+ if (pkt)
+ av_packet_unref(&msg.pkt);
+ return ret;
+}
+
+static int fifo_write_trailer(AVFormatContext *avf)
+{
+ int ret = 0;
+ FifoContext *fifo= avf->priv_data;
+
+ av_thread_message_queue_set_err_recv(fifo->queue, AVERROR_EOF);
+
+ ret = pthread_join( fifo->writer_thread, NULL);
+ if (ret < 0) {
+ av_log(avf, AV_LOG_ERROR, "pthread join error: %s\n",
+ av_err2str(AVERROR(ret)));
+ return AVERROR(ret);
+ }
+
+ ret = fifo->write_trailer_ret;
+ return ret;
+}
+
+static void fifo_deinit(AVFormatContext *avf)
+{
+ FifoContext *fifo = avf->priv_data;
+
+ if (fifo->format_options)
+ av_dict_free(&fifo->format_options);
+
+ if (avf)
+ avformat_free_context(fifo->avf);
+
+ if (fifo->queue) {
+ av_thread_message_flush(fifo->queue);
+ av_thread_message_queue_free(&fifo->queue);
+ }
+
+ pthread_mutex_destroy(&fifo->overflow_flag_lock);
+}
+
+#define OFFSET(x) offsetof(FifoContext, x)
+static const AVOption options[] = {
+ {"fifo_format", "Target muxer", OFFSET(format),
+ AV_OPT_TYPE_STRING, {.str = NULL}, 0, 0, AV_OPT_FLAG_ENCODING_PARAM},
+
+ {"queue_size", "Size of fifo queue", OFFSET(queue_size),
+ AV_OPT_TYPE_INT, {.i64 = FIFO_DEFAULT_QUEUE_SIZE}, 1, 1024,
AV_OPT_FLAG_ENCODING_PARAM},
+
+ {"format_opts", "Options to be passed to underlying muxer",
OFFSET(format_options_str),
+ AV_OPT_TYPE_STRING, {.str = NULL}, 0, 0, AV_OPT_FLAG_ENCODING_PARAM},
+
+ {"block_on_overflow", "Block output on FIFO queue overflow until queue
frees up.", OFFSET(block_on_overflow),
+ AV_OPT_TYPE_BOOL, {.i64 = 1}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
+
+ {"restart_with_keyframe", "Wait for keyframe when restarting output",
OFFSET(restart_with_keyframe),
+ AV_OPT_TYPE_BOOL, {.i64 = 1}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
+
+ {"attempt_recovery", "Attempt recovery in case of failure",
OFFSET(attempt_recovery),
+ AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
+
+ {"max_recovery_attempts", "Maximal number of recovery attempts",
OFFSET(max_recovery_attempts),
+ AV_OPT_TYPE_INT, {.i64 = FIFO_DEFAULT_MAX_RECOVERY_ATTEMPTS}, 0,
INT_MAX, AV_OPT_FLAG_ENCODING_PARAM},
+
+ {"recovery_wait_time", "Waiting time between recovery attempts
(seconds)", OFFSET(recovery_wait_time),
+ AV_OPT_TYPE_DURATION, {.i64 = FIFO_DEFAULT_RECOVERY_WAIT_TIME}, 0,
INT64_MAX, AV_OPT_FLAG_ENCODING_PARAM},
+
+ {"recovery_wait_streamtime", "Use stream time instead of real time
while waiting for recovery",
+ OFFSET(recovery_wait_streamtime), AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1,
AV_OPT_FLAG_ENCODING_PARAM},
+
+ {"recover_any_error", "Attempt recovery regardless of type of the
error", OFFSET(recover_any_error),
+ AV_OPT_TYPE_BOOL, {.i64 = 0}, 0, 1, AV_OPT_FLAG_ENCODING_PARAM},
+
+ {NULL},
+};
+
+static const AVClass fifo_muxer_class = {
+ .class_name = "Fifo muxer",
+ .item_name = av_default_item_name,
+ .option = options,
+ .version = LIBAVUTIL_VERSION_INT,
+};
+
+AVOutputFormat ff_fifo_muxer = {
+ .name = "fifo",
+ .long_name = NULL_IF_CONFIG_SMALL("FIFO queue pseudo-muxer"),
+ .priv_data_size = sizeof(FifoContext),
+ .init = fifo_init,
+ .write_header = fifo_write_header,
+ .write_packet = fifo_write_packet,
+ .write_trailer = fifo_write_trailer,
+ .deinit = fifo_deinit,
+ .priv_class = &fifo_muxer_class,
+ .flags = AVFMT_NOFILE,
+};
+
--
1.9.1
_______________________________________________
ffmpeg-devel mailing list
ffmpeg-devel@ffmpeg.org
http://ffmpeg.org/mailman/listinfo/ffmpeg-devel
