videolan/ffmpeg
1
Fork 0
mirror of https://git.videolan.org/git/ffmpeg.git synced 2024-09-29 16:23:10 +02:00
Code Releases Activity
b9fcc135c5
ffmpeg/fftools/ffmpeg_sched.c
Mark Thompson 7f4b8d2f5e ffmpeg: set extra_hw_frames to account for frames held in queues 
Since e0da916b8f the ffmpeg utility has
held multiple frames output by the decoder in internal queues without
telling the decoder that it is going to do so.  When the decoder has a
fixed-size pool of frames (common in some hardware APIs where the output
frames must be stored as an array texture) this could lead to the pool
being exhausted and the decoder getting stuck.  Fix this by telling the
decoder to allocate additional frames according to the queue size.
2024-03-19 22:56:56 +00:00

2521 lines
65 KiB
C
Raw Blame History

/*
* Inter-thread scheduling/synchronization.
* Copyright (c) 2023 Anton Khirnov
*
* 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 <stdatomic.h>
#include <stddef.h>
#include <stdint.h>
#include "cmdutils.h"
#include "ffmpeg_sched.h"
#include "ffmpeg_utils.h"
#include "sync_queue.h"
#include "thread_queue.h"
#include "libavcodec/packet.h"
#include "libavutil/avassert.h"
#include "libavutil/error.h"
#include "libavutil/fifo.h"
#include "libavutil/frame.h"
#include "libavutil/mem.h"
#include "libavutil/thread.h"
#include "libavutil/threadmessage.h"
#include "libavutil/time.h"
// 100 ms
// FIXME: some other value? make this dynamic?
#define SCHEDULE_TOLERANCE (100 * 1000)
enum QueueType {
QUEUE_PACKETS,
QUEUE_FRAMES,
};
typedef struct SchWaiter {
pthread_mutex_t lock;
pthread_cond_t cond;
atomic_int choked;
// the following are internal state of schedule_update_locked() and must not
// be accessed outside of it
int choked_prev;
int choked_next;
} SchWaiter;
typedef struct SchTask {
Scheduler *parent;
SchedulerNode node;
SchThreadFunc func;
void *func_arg;
pthread_t thread;
int thread_running;
} SchTask;
typedef struct SchDec {
const AVClass *class;
SchedulerNode src;
SchedulerNode *dst;
uint8_t *dst_finished;
unsigned nb_dst;
SchTask task;
// Queue for receiving input packets, one stream.
ThreadQueue *queue;
// Queue for sending post-flush end timestamps back to the source
AVThreadMessageQueue *queue_end_ts;
int expect_end_ts;
// temporary storage used by sch_dec_send()
AVFrame *send_frame;
} SchDec;
typedef struct SchSyncQueue {
SyncQueue *sq;
AVFrame *frame;
pthread_mutex_t lock;
unsigned *enc_idx;
unsigned nb_enc_idx;
} SchSyncQueue;
typedef struct SchEnc {
const AVClass *class;
SchedulerNode src;
SchedulerNode *dst;
uint8_t *dst_finished;
unsigned nb_dst;
// [0] - index of the sync queue in Scheduler.sq_enc,
// [1] - index of this encoder in the sq
int sq_idx[2];
/* Opening encoders is somewhat nontrivial due to their interaction with
* sync queues, which are (among other things) responsible for maintaining
* constant audio frame size, when it is required by the encoder.
*
* Opening the encoder requires stream parameters, obtained from the first
* frame. However, that frame cannot be properly chunked by the sync queue
* without knowing the required frame size, which is only available after
* opening the encoder.
*
* This apparent circular dependency is resolved in the following way:
* - the caller creating the encoder gives us a callback which opens the
* encoder and returns the required frame size (if any)
* - when the first frame is sent to the encoder, the sending thread
* - calls this callback, opening the encoder
* - passes the returned frame size to the sync queue
*/
int (*open_cb)(void *opaque, const AVFrame *frame);
int opened;
SchTask task;
// Queue for receiving input frames, one stream.
ThreadQueue *queue;
// tq_send() to queue returned EOF
int in_finished;
// temporary storage used by sch_enc_send()
AVPacket *send_pkt;
} SchEnc;
typedef struct SchDemuxStream {
SchedulerNode *dst;
uint8_t *dst_finished;
unsigned nb_dst;
} SchDemuxStream;
typedef struct SchDemux {
const AVClass *class;
SchDemuxStream *streams;
unsigned nb_streams;
SchTask task;
SchWaiter waiter;
// temporary storage used by sch_demux_send()
AVPacket *send_pkt;
// protected by schedule_lock
int task_exited;
} SchDemux;
typedef struct PreMuxQueue {
/**
* Queue for buffering the packets before the muxer task can be started.
*/
AVFifo *fifo;
/**
* Maximum number of packets in fifo.
*/
int max_packets;
/*
* The size of the AVPackets' buffers in queue.
* Updated when a packet is either pushed or pulled from the queue.
*/
size_t data_size;
/* Threshold after which max_packets will be in effect */
size_t data_threshold;
} PreMuxQueue;
typedef struct SchMuxStream {
SchedulerNode src;
SchedulerNode src_sched;
unsigned *sub_heartbeat_dst;
unsigned nb_sub_heartbeat_dst;
PreMuxQueue pre_mux_queue;
// an EOF was generated while flushing the pre-mux queue
int init_eof;
////////////////////////////////////////////////////////////
// The following are protected by Scheduler.schedule_lock //
/* dts+duration of the last packet sent to this stream
in AV_TIME_BASE_Q */
int64_t last_dts;
// this stream no longer accepts input
int source_finished;
////////////////////////////////////////////////////////////
} SchMuxStream;
typedef struct SchMux {
const AVClass *class;
SchMuxStream *streams;
unsigned nb_streams;
unsigned nb_streams_ready;
int (*init)(void *arg);
SchTask task;
/**
* Set to 1 after starting the muxer task and flushing the
* pre-muxing queues.
* Set either before any tasks have started, or with
* Scheduler.mux_ready_lock held.
*/
atomic_int mux_started;
ThreadQueue *queue;
unsigned queue_size;
AVPacket *sub_heartbeat_pkt;
} SchMux;
typedef struct SchFilterIn {
SchedulerNode src;
SchedulerNode src_sched;
int send_finished;
int receive_finished;
} SchFilterIn;
typedef struct SchFilterOut {
SchedulerNode dst;
} SchFilterOut;
typedef struct SchFilterGraph {
const AVClass *class;
SchFilterIn *inputs;
unsigned nb_inputs;
atomic_uint nb_inputs_finished_send;
unsigned nb_inputs_finished_receive;
SchFilterOut *outputs;
unsigned nb_outputs;
SchTask task;
// input queue, nb_inputs+1 streams
// last stream is control
ThreadQueue *queue;
SchWaiter waiter;
// protected by schedule_lock
unsigned best_input;
int task_exited;
} SchFilterGraph;
struct Scheduler {
const AVClass *class;
SchDemux *demux;
unsigned nb_demux;
SchMux *mux;
unsigned nb_mux;
unsigned nb_mux_ready;
pthread_mutex_t mux_ready_lock;
unsigned nb_mux_done;
pthread_mutex_t mux_done_lock;
pthread_cond_t mux_done_cond;
SchDec *dec;
unsigned nb_dec;
SchEnc *enc;
unsigned nb_enc;
SchSyncQueue *sq_enc;
unsigned nb_sq_enc;
SchFilterGraph *filters;
unsigned nb_filters;
char *sdp_filename;
int sdp_auto;
int transcode_started;
atomic_int terminate;
atomic_int task_failed;
pthread_mutex_t schedule_lock;
atomic_int_least64_t last_dts;
};
/**
* Wait until this task is allowed to proceed.
*
* @retval 0 the caller should proceed
* @retval 1 the caller should terminate
*/
static int waiter_wait(Scheduler *sch, SchWaiter *w)
{
int terminate;
if (!atomic_load(&w->choked))
return 0;
pthread_mutex_lock(&w->lock);
while (atomic_load(&w->choked) && !atomic_load(&sch->terminate))
pthread_cond_wait(&w->cond, &w->lock);
terminate = atomic_load(&sch->terminate);
pthread_mutex_unlock(&w->lock);
return terminate;
}
static void waiter_set(SchWaiter *w, int choked)
{
pthread_mutex_lock(&w->lock);
atomic_store(&w->choked, choked);
pthread_cond_signal(&w->cond);
pthread_mutex_unlock(&w->lock);
}
static int waiter_init(SchWaiter *w)
{
int ret;
atomic_init(&w->choked, 0);
ret = pthread_mutex_init(&w->lock, NULL);
if (ret)
return AVERROR(ret);
ret = pthread_cond_init(&w->cond, NULL);
if (ret)
return AVERROR(ret);
return 0;
}
static void waiter_uninit(SchWaiter *w)
{
pthread_mutex_destroy(&w->lock);
pthread_cond_destroy(&w->cond);
}
static int queue_alloc(ThreadQueue **ptq, unsigned nb_streams, unsigned queue_size,
enum QueueType type)
{
ThreadQueue *tq;
ObjPool *op;
if (queue_size <= 0) {
if (type == QUEUE_FRAMES)
queue_size = DEFAULT_FRAME_THREAD_QUEUE_SIZE;
else
queue_size = DEFAULT_PACKET_THREAD_QUEUE_SIZE;
}
if (type == QUEUE_FRAMES) {
// This queue length is used in the decoder code to ensure that
// there are enough entries in fixed-size frame pools to account
// for frames held in queues inside the ffmpeg utility. If this
// can ever dynamically change then the corresponding decode
// code needs to be updated as well.
av_assert0(queue_size == DEFAULT_FRAME_THREAD_QUEUE_SIZE);
}
op = (type == QUEUE_PACKETS) ? objpool_alloc_packets() :
objpool_alloc_frames();
if (!op)
return AVERROR(ENOMEM);
tq = tq_alloc(nb_streams, queue_size, op,
(type == QUEUE_PACKETS) ? pkt_move : frame_move);
if (!tq) {
objpool_free(&op);
return AVERROR(ENOMEM);
}
*ptq = tq;
return 0;
}
static void *task_wrapper(void *arg);
static int task_stop(SchTask *task)
{
int ret;
void *thread_ret;
if (!task->thread_running)
return 0;
ret = pthread_join(task->thread, &thread_ret);
av_assert0(ret == 0);
task->thread_running = 0;
return (intptr_t)thread_ret;
}
static int task_start(SchTask *task)
{
int ret;
av_log(task->func_arg, AV_LOG_VERBOSE, "Starting thread...\n");
av_assert0(!task->thread_running);
ret = pthread_create(&task->thread, NULL, task_wrapper, task);
if (ret) {
av_log(task->func_arg, AV_LOG_ERROR, "pthread_create() failed: %s\n",
strerror(ret));
return AVERROR(ret);
}
task->thread_running = 1;
return 0;
}
static void task_init(Scheduler *sch, SchTask *task, enum SchedulerNodeType type, unsigned idx,
SchThreadFunc func, void *func_arg)
{
task->parent = sch;
task->node.type = type;
task->node.idx = idx;
task->func = func;
task->func_arg = func_arg;
}
static int64_t trailing_dts(const Scheduler *sch, int count_finished)
{
int64_t min_dts = INT64_MAX;
for (unsigned i = 0; i < sch->nb_mux; i++) {
const SchMux *mux = &sch->mux[i];
for (unsigned j = 0; j < mux->nb_streams; j++) {
const SchMuxStream *ms = &mux->streams[j];
if (ms->source_finished && !count_finished)
continue;
if (ms->last_dts == AV_NOPTS_VALUE)
return AV_NOPTS_VALUE;
min_dts = FFMIN(min_dts, ms->last_dts);
}
}
return min_dts == INT64_MAX ? AV_NOPTS_VALUE : min_dts;
}
int sch_stop(Scheduler *sch, int64_t *finish_ts)
{
int ret = 0, err;
atomic_store(&sch->terminate, 1);
for (unsigned type = 0; type < 2; type++)
for (unsigned i = 0; i < (type ? sch->nb_demux : sch->nb_filters); i++) {
SchWaiter *w = type ? &sch->demux[i].waiter : &sch->filters[i].waiter;
waiter_set(w, 1);
}
for (unsigned i = 0; i < sch->nb_demux; i++) {
SchDemux *d = &sch->demux[i];
err = task_stop(&d->task);
ret = err_merge(ret, err);
}
for (unsigned i = 0; i < sch->nb_dec; i++) {
SchDec *dec = &sch->dec[i];
err = task_stop(&dec->task);
ret = err_merge(ret, err);
}
for (unsigned i = 0; i < sch->nb_filters; i++) {
SchFilterGraph *fg = &sch->filters[i];
err = task_stop(&fg->task);
ret = err_merge(ret, err);
}
for (unsigned i = 0; i < sch->nb_enc; i++) {
SchEnc *enc = &sch->enc[i];
err = task_stop(&enc->task);
ret = err_merge(ret, err);
}
for (unsigned i = 0; i < sch->nb_mux; i++) {
SchMux *mux = &sch->mux[i];
err = task_stop(&mux->task);
ret = err_merge(ret, err);
}
if (finish_ts)
*finish_ts = trailing_dts(sch, 1);
return ret;
}
void sch_free(Scheduler **psch)
{
Scheduler *sch = *psch;
if (!sch)
return;
sch_stop(sch, NULL);
for (unsigned i = 0; i < sch->nb_demux; i++) {
SchDemux *d = &sch->demux[i];
for (unsigned j = 0; j < d->nb_streams; j++) {
SchDemuxStream *ds = &d->streams[j];
av_freep(&ds->dst);
av_freep(&ds->dst_finished);
}
av_freep(&d->streams);
av_packet_free(&d->send_pkt);
waiter_uninit(&d->waiter);
}
av_freep(&sch->demux);
for (unsigned i = 0; i < sch->nb_mux; i++) {
SchMux *mux = &sch->mux[i];
for (unsigned j = 0; j < mux->nb_streams; j++) {
SchMuxStream *ms = &mux->streams[j];
if (ms->pre_mux_queue.fifo) {
AVPacket *pkt;
while (av_fifo_read(ms->pre_mux_queue.fifo, &pkt, 1) >= 0)
av_packet_free(&pkt);
av_fifo_freep2(&ms->pre_mux_queue.fifo);
}
av_freep(&ms->sub_heartbeat_dst);
}
av_freep(&mux->streams);
av_packet_free(&mux->sub_heartbeat_pkt);
tq_free(&mux->queue);
}
av_freep(&sch->mux);
for (unsigned i = 0; i < sch->nb_dec; i++) {
SchDec *dec = &sch->dec[i];
tq_free(&dec->queue);
av_thread_message_queue_free(&dec->queue_end_ts);
av_freep(&dec->dst);
av_freep(&dec->dst_finished);
av_frame_free(&dec->send_frame);
}
av_freep(&sch->dec);
for (unsigned i = 0; i < sch->nb_enc; i++) {
SchEnc *enc = &sch->enc[i];
tq_free(&enc->queue);
av_packet_free(&enc->send_pkt);
av_freep(&enc->dst);
av_freep(&enc->dst_finished);
}
av_freep(&sch->enc);
for (unsigned i = 0; i < sch->nb_sq_enc; i++) {
SchSyncQueue *sq = &sch->sq_enc[i];
sq_free(&sq->sq);
av_frame_free(&sq->frame);
pthread_mutex_destroy(&sq->lock);
av_freep(&sq->enc_idx);
}
av_freep(&sch->sq_enc);
for (unsigned i = 0; i < sch->nb_filters; i++) {
SchFilterGraph *fg = &sch->filters[i];
tq_free(&fg->queue);
av_freep(&fg->inputs);
av_freep(&fg->outputs);
waiter_uninit(&fg->waiter);
}
av_freep(&sch->filters);
av_freep(&sch->sdp_filename);
pthread_mutex_destroy(&sch->schedule_lock);
pthread_mutex_destroy(&sch->mux_ready_lock);
pthread_mutex_destroy(&sch->mux_done_lock);
pthread_cond_destroy(&sch->mux_done_cond);
av_freep(psch);
}
static const AVClass scheduler_class = {
.class_name = "Scheduler",
.version = LIBAVUTIL_VERSION_INT,
};
Scheduler *sch_alloc(void)
{
Scheduler *sch;
int ret;
sch = av_mallocz(sizeof(*sch));
if (!sch)
return NULL;
sch->class = &scheduler_class;
sch->sdp_auto = 1;
ret = pthread_mutex_init(&sch->schedule_lock, NULL);
if (ret)
goto fail;
ret = pthread_mutex_init(&sch->mux_ready_lock, NULL);
if (ret)
goto fail;
ret = pthread_mutex_init(&sch->mux_done_lock, NULL);
if (ret)
goto fail;
ret = pthread_cond_init(&sch->mux_done_cond, NULL);
if (ret)
goto fail;
return sch;
fail:
sch_free(&sch);
return NULL;
}
int sch_sdp_filename(Scheduler *sch, const char *sdp_filename)
{
av_freep(&sch->sdp_filename);
sch->sdp_filename = av_strdup(sdp_filename);
return sch->sdp_filename ? 0 : AVERROR(ENOMEM);
}
static const AVClass sch_mux_class = {
.class_name = "SchMux",
.version = LIBAVUTIL_VERSION_INT,
.parent_log_context_offset = offsetof(SchMux, task.func_arg),
};
int sch_add_mux(Scheduler *sch, SchThreadFunc func, int (*init)(void *),
void *arg, int sdp_auto, unsigned thread_queue_size)
{
const unsigned idx = sch->nb_mux;
SchMux *mux;
int ret;
ret = GROW_ARRAY(sch->mux, sch->nb_mux);
if (ret < 0)
return ret;
mux = &sch->mux[idx];
mux->class = &sch_mux_class;
mux->init = init;
mux->queue_size = thread_queue_size;
task_init(sch, &mux->task, SCH_NODE_TYPE_MUX, idx, func, arg);
sch->sdp_auto &= sdp_auto;
return idx;
}
int sch_add_mux_stream(Scheduler *sch, unsigned mux_idx)
{
SchMux *mux;
SchMuxStream *ms;
unsigned stream_idx;
int ret;
av_assert0(mux_idx < sch->nb_mux);
mux = &sch->mux[mux_idx];
ret = GROW_ARRAY(mux->streams, mux->nb_streams);
if (ret < 0)
return ret;
stream_idx = mux->nb_streams - 1;
ms = &mux->streams[stream_idx];
ms->pre_mux_queue.fifo = av_fifo_alloc2(8, sizeof(AVPacket*), 0);
if (!ms->pre_mux_queue.fifo)
return AVERROR(ENOMEM);
ms->last_dts = AV_NOPTS_VALUE;
return stream_idx;
}
static const AVClass sch_demux_class = {
.class_name = "SchDemux",
.version = LIBAVUTIL_VERSION_INT,
.parent_log_context_offset = offsetof(SchDemux, task.func_arg),
};
int sch_add_demux(Scheduler *sch, SchThreadFunc func, void *ctx)
{
const unsigned idx = sch->nb_demux;
SchDemux *d;
int ret;
ret = GROW_ARRAY(sch->demux, sch->nb_demux);
if (ret < 0)
return ret;
d = &sch->demux[idx];
task_init(sch, &d->task, SCH_NODE_TYPE_DEMUX, idx, func, ctx);
d->class = &sch_demux_class;
d->send_pkt = av_packet_alloc();
if (!d->send_pkt)
return AVERROR(ENOMEM);
ret = waiter_init(&d->waiter);
if (ret < 0)
return ret;
return idx;
}
int sch_add_demux_stream(Scheduler *sch, unsigned demux_idx)
{
SchDemux *d;
int ret;
av_assert0(demux_idx < sch->nb_demux);
d = &sch->demux[demux_idx];
ret = GROW_ARRAY(d->streams, d->nb_streams);
return ret < 0 ? ret : d->nb_streams - 1;
}
static const AVClass sch_dec_class = {
.class_name = "SchDec",
.version = LIBAVUTIL_VERSION_INT,
.parent_log_context_offset = offsetof(SchDec, task.func_arg),
};
int sch_add_dec(Scheduler *sch, SchThreadFunc func, void *ctx,
int send_end_ts)
{
const unsigned idx = sch->nb_dec;
SchDec *dec;
int ret;
ret = GROW_ARRAY(sch->dec, sch->nb_dec);
if (ret < 0)
return ret;
dec = &sch->dec[idx];
task_init(sch, &dec->task, SCH_NODE_TYPE_DEC, idx, func, ctx);
dec->class = &sch_dec_class;
dec->send_frame = av_frame_alloc();
if (!dec->send_frame)
return AVERROR(ENOMEM);
ret = queue_alloc(&dec->queue, 1, 0, QUEUE_PACKETS);
if (ret < 0)
return ret;
if (send_end_ts) {
ret = av_thread_message_queue_alloc(&dec->queue_end_ts, 1, sizeof(Timestamp));
if (ret < 0)
return ret;
}
return idx;
}
static const AVClass sch_enc_class = {
.class_name = "SchEnc",
.version = LIBAVUTIL_VERSION_INT,
.parent_log_context_offset = offsetof(SchEnc, task.func_arg),
};
int sch_add_enc(Scheduler *sch, SchThreadFunc func, void *ctx,
int (*open_cb)(void *opaque, const AVFrame *frame))
{
const unsigned idx = sch->nb_enc;
SchEnc *enc;
int ret;
ret = GROW_ARRAY(sch->enc, sch->nb_enc);
if (ret < 0)
return ret;
enc = &sch->enc[idx];
enc->class = &sch_enc_class;
enc->open_cb = open_cb;
enc->sq_idx[0] = -1;
enc->sq_idx[1] = -1;
task_init(sch, &enc->task, SCH_NODE_TYPE_ENC, idx, func, ctx);
enc->send_pkt = av_packet_alloc();
if (!enc->send_pkt)
return AVERROR(ENOMEM);
ret = queue_alloc(&enc->queue, 1, 0, QUEUE_FRAMES);
if (ret < 0)
return ret;
return idx;
}
static const AVClass sch_fg_class = {
.class_name = "SchFilterGraph",
.version = LIBAVUTIL_VERSION_INT,
.parent_log_context_offset = offsetof(SchFilterGraph, task.func_arg),
};
int sch_add_filtergraph(Scheduler *sch, unsigned nb_inputs, unsigned nb_outputs,
SchThreadFunc func, void *ctx)
{
const unsigned idx = sch->nb_filters;
SchFilterGraph *fg;
int ret;
ret = GROW_ARRAY(sch->filters, sch->nb_filters);
if (ret < 0)
return ret;
fg = &sch->filters[idx];
fg->class = &sch_fg_class;
task_init(sch, &fg->task, SCH_NODE_TYPE_FILTER_IN, idx, func, ctx);
if (nb_inputs) {
fg->inputs = av_calloc(nb_inputs, sizeof(*fg->inputs));
if (!fg->inputs)
return AVERROR(ENOMEM);
fg->nb_inputs = nb_inputs;
}
if (nb_outputs) {
fg->outputs = av_calloc(nb_outputs, sizeof(*fg->outputs));
if (!fg->outputs)
return AVERROR(ENOMEM);
fg->nb_outputs = nb_outputs;
}
ret = waiter_init(&fg->waiter);
if (ret < 0)
return ret;
ret = queue_alloc(&fg->queue, fg->nb_inputs + 1, 0, QUEUE_FRAMES);
if (ret < 0)
return ret;
return idx;
}
int sch_add_sq_enc(Scheduler *sch, uint64_t buf_size_us, void *logctx)
{
SchSyncQueue *sq;
int ret;
ret = GROW_ARRAY(sch->sq_enc, sch->nb_sq_enc);
if (ret < 0)
return ret;
sq = &sch->sq_enc[sch->nb_sq_enc - 1];
sq->sq = sq_alloc(SYNC_QUEUE_FRAMES, buf_size_us, logctx);
if (!sq->sq)
return AVERROR(ENOMEM);
sq->frame = av_frame_alloc();
if (!sq->frame)
return AVERROR(ENOMEM);
ret = pthread_mutex_init(&sq->lock, NULL);
if (ret)
return AVERROR(ret);
return sq - sch->sq_enc;
}
int sch_sq_add_enc(Scheduler *sch, unsigned sq_idx, unsigned enc_idx,
int limiting, uint64_t max_frames)
{
SchSyncQueue *sq;
SchEnc *enc;
int ret;
av_assert0(sq_idx < sch->nb_sq_enc);
sq = &sch->sq_enc[sq_idx];
av_assert0(enc_idx < sch->nb_enc);
enc = &sch->enc[enc_idx];
ret = GROW_ARRAY(sq->enc_idx, sq->nb_enc_idx);
if (ret < 0)
return ret;
sq->enc_idx[sq->nb_enc_idx - 1] = enc_idx;
ret = sq_add_stream(sq->sq, limiting);
if (ret < 0)
return ret;
enc->sq_idx[0] = sq_idx;
enc->sq_idx[1] = ret;
if (max_frames != INT64_MAX)
sq_limit_frames(sq->sq, enc->sq_idx[1], max_frames);
return 0;
}
int sch_connect(Scheduler *sch, SchedulerNode src, SchedulerNode dst)
{
int ret;
switch (src.type) {
case SCH_NODE_TYPE_DEMUX: {
SchDemuxStream *ds;
av_assert0(src.idx < sch->nb_demux &&
src.idx_stream < sch->demux[src.idx].nb_streams);
ds = &sch->demux[src.idx].streams[src.idx_stream];
ret = GROW_ARRAY(ds->dst, ds->nb_dst);
if (ret < 0)
return ret;
ds->dst[ds->nb_dst - 1] = dst;
// demuxed packets go to decoding or streamcopy
switch (dst.type) {
case SCH_NODE_TYPE_DEC: {
SchDec *dec;
av_assert0(dst.idx < sch->nb_dec);
dec = &sch->dec[dst.idx];
av_assert0(!dec->src.type);
dec->src = src;
break;
}
case SCH_NODE_TYPE_MUX: {
SchMuxStream *ms;
av_assert0(dst.idx < sch->nb_mux &&
dst.idx_stream < sch->mux[dst.idx].nb_streams);
ms = &sch->mux[dst.idx].streams[dst.idx_stream];
av_assert0(!ms->src.type);
ms->src = src;
break;
}
default: av_assert0(0);
}
break;
}
case SCH_NODE_TYPE_DEC: {
SchDec *dec;
av_assert0(src.idx < sch->nb_dec);
dec = &sch->dec[src.idx];
ret = GROW_ARRAY(dec->dst, dec->nb_dst);
if (ret < 0)
return ret;
dec->dst[dec->nb_dst - 1] = dst;
// decoded frames go to filters or encoding
switch (dst.type) {
case SCH_NODE_TYPE_FILTER_IN: {
SchFilterIn *fi;
av_assert0(dst.idx < sch->nb_filters &&
dst.idx_stream < sch->filters[dst.idx].nb_inputs);
fi = &sch->filters[dst.idx].inputs[dst.idx_stream];
av_assert0(!fi->src.type);
fi->src = src;
break;
}
case SCH_NODE_TYPE_ENC: {
SchEnc *enc;
av_assert0(dst.idx < sch->nb_enc);
enc = &sch->enc[dst.idx];
av_assert0(!enc->src.type);
enc->src = src;
break;
}
default: av_assert0(0);
}
break;
}
case SCH_NODE_TYPE_FILTER_OUT: {
SchFilterOut *fo;
SchEnc *enc;
av_assert0(src.idx < sch->nb_filters &&
src.idx_stream < sch->filters[src.idx].nb_outputs);
// filtered frames go to encoding
av_assert0(dst.type == SCH_NODE_TYPE_ENC &&
dst.idx < sch->nb_enc);
fo = &sch->filters[src.idx].outputs[src.idx_stream];
enc = &sch->enc[dst.idx];
av_assert0(!fo->dst.type && !enc->src.type);
fo->dst = dst;
enc->src = src;
break;
}
case SCH_NODE_TYPE_ENC: {
SchEnc *enc;
av_assert0(src.idx < sch->nb_enc);
enc = &sch->enc[src.idx];
ret = GROW_ARRAY(enc->dst, enc->nb_dst);
if (ret < 0)
return ret;
enc->dst[enc->nb_dst - 1] = dst;
// encoding packets go to muxing or decoding
switch (dst.type) {
case SCH_NODE_TYPE_MUX: {
SchMuxStream *ms;
av_assert0(dst.idx < sch->nb_mux &&
dst.idx_stream < sch->mux[dst.idx].nb_streams);
ms = &sch->mux[dst.idx].streams[dst.idx_stream];
av_assert0(!ms->src.type);
ms->src = src;
break;
}
case SCH_NODE_TYPE_DEC: {
SchDec *dec;
av_assert0(dst.idx < sch->nb_dec);
dec = &sch->dec[dst.idx];
av_assert0(!dec->src.type);
dec->src = src;
break;
}
default: av_assert0(0);
}
break;
}
default: av_assert0(0);
}
return 0;
}
static int mux_task_start(SchMux *mux)
{
int ret = 0;
ret = task_start(&mux->task);
if (ret < 0)
return ret;
/* flush the pre-muxing queues */
for (unsigned i = 0; i < mux->nb_streams; i++) {
SchMuxStream *ms = &mux->streams[i];
AVPacket *pkt;
while (av_fifo_read(ms->pre_mux_queue.fifo, &pkt, 1) >= 0) {
if (pkt) {
if (!ms->init_eof)
ret = tq_send(mux->queue, i, pkt);
av_packet_free(&pkt);
if (ret == AVERROR_EOF)
ms->init_eof = 1;
else if (ret < 0)
return ret;
} else
tq_send_finish(mux->queue, i);
}
}
atomic_store(&mux->mux_started, 1);
return 0;
}
int print_sdp(const char *filename);
static int mux_init(Scheduler *sch, SchMux *mux)
{
int ret;
ret = mux->init(mux->task.func_arg);
if (ret < 0)
return ret;
sch->nb_mux_ready++;
if (sch->sdp_filename || sch->sdp_auto) {
if (sch->nb_mux_ready < sch->nb_mux)
return 0;
ret = print_sdp(sch->sdp_filename);
if (ret < 0) {
av_log(sch, AV_LOG_ERROR, "Error writing the SDP.\n");
return ret;
}
/* SDP is written only after all the muxers are ready, so now we
* start ALL the threads */
for (unsigned i = 0; i < sch->nb_mux; i++) {
ret = mux_task_start(&sch->mux[i]);
if (ret < 0)
return ret;
}
} else {
ret = mux_task_start(mux);
if (ret < 0)
return ret;
}
return 0;
}
void sch_mux_stream_buffering(Scheduler *sch, unsigned mux_idx, unsigned stream_idx,
size_t data_threshold, int max_packets)
{
SchMux *mux;
SchMuxStream *ms;
av_assert0(mux_idx < sch->nb_mux);
mux = &sch->mux[mux_idx];
av_assert0(stream_idx < mux->nb_streams);
ms = &mux->streams[stream_idx];
ms->pre_mux_queue.max_packets = max_packets;
ms->pre_mux_queue.data_threshold = data_threshold;
}
int sch_mux_stream_ready(Scheduler *sch, unsigned mux_idx, unsigned stream_idx)
{
SchMux *mux;
int ret = 0;
av_assert0(mux_idx < sch->nb_mux);
mux = &sch->mux[mux_idx];
av_assert0(stream_idx < mux->nb_streams);
pthread_mutex_lock(&sch->mux_ready_lock);
av_assert0(mux->nb_streams_ready < mux->nb_streams);
// this may be called during initialization - do not start
// threads before sch_start() is called
if (++mux->nb_streams_ready == mux->nb_streams && sch->transcode_started)
ret = mux_init(sch, mux);
pthread_mutex_unlock(&sch->mux_ready_lock);
return ret;
}
int sch_mux_sub_heartbeat_add(Scheduler *sch, unsigned mux_idx, unsigned stream_idx,
unsigned dec_idx)
{
SchMux *mux;
SchMuxStream *ms;
int ret = 0;
av_assert0(mux_idx < sch->nb_mux);
mux = &sch->mux[mux_idx];
av_assert0(stream_idx < mux->nb_streams);
ms = &mux->streams[stream_idx];
ret = GROW_ARRAY(ms->sub_heartbeat_dst, ms->nb_sub_heartbeat_dst);
if (ret < 0)
return ret;
av_assert0(dec_idx < sch->nb_dec);
ms->sub_heartbeat_dst[ms->nb_sub_heartbeat_dst - 1] = dec_idx;
if (!mux->sub_heartbeat_pkt) {
mux->sub_heartbeat_pkt = av_packet_alloc();
if (!mux->sub_heartbeat_pkt)
return AVERROR(ENOMEM);
}
return 0;
}
static void unchoke_for_stream(Scheduler *sch, SchedulerNode src)
{
while (1) {
SchFilterGraph *fg;
// fed directly by a demuxer (i.e. not through a filtergraph)
if (src.type == SCH_NODE_TYPE_DEMUX) {
sch->demux[src.idx].waiter.choked_next = 0;
return;
}
av_assert0(src.type == SCH_NODE_TYPE_FILTER_OUT);
fg = &sch->filters[src.idx];
// the filtergraph contains internal sources and
// requested to be scheduled directly
if (fg->best_input == fg->nb_inputs) {
fg->waiter.choked_next = 0;
return;
}
src = fg->inputs[fg->best_input].src_sched;
}
}
static void schedule_update_locked(Scheduler *sch)
{
int64_t dts;
int have_unchoked = 0;
// on termination request all waiters are choked,
// we are not to unchoke them
if (atomic_load(&sch->terminate))
return;
dts = trailing_dts(sch, 0);
atomic_store(&sch->last_dts, dts);
// initialize our internal state
for (unsigned type = 0; type < 2; type++)
for (unsigned i = 0; i < (type ? sch->nb_filters : sch->nb_demux); i++) {
SchWaiter *w = type ? &sch->filters[i].waiter : &sch->demux[i].waiter;
w->choked_prev = atomic_load(&w->choked);
w->choked_next = 1;
}
// figure out the sources that are allowed to proceed
for (unsigned i = 0; i < sch->nb_mux; i++) {
SchMux *mux = &sch->mux[i];
for (unsigned j = 0; j < mux->nb_streams; j++) {
SchMuxStream *ms = &mux->streams[j];
// unblock sources for output streams that are not finished
// and not too far ahead of the trailing stream
if (ms->source_finished)
continue;
if (dts == AV_NOPTS_VALUE && ms->last_dts != AV_NOPTS_VALUE)
continue;
if (dts != AV_NOPTS_VALUE && ms->last_dts - dts >= SCHEDULE_TOLERANCE)
continue;
// resolve the source to unchoke
unchoke_for_stream(sch, ms->src_sched);
have_unchoked = 1;
}
}
// make sure to unchoke at least one source, if still available
for (unsigned type = 0; !have_unchoked && type < 2; type++)
for (unsigned i = 0; i < (type ? sch->nb_filters : sch->nb_demux); i++) {
int exited = type ? sch->filters[i].task_exited : sch->demux[i].task_exited;
SchWaiter *w = type ? &sch->filters[i].waiter : &sch->demux[i].waiter;
if (!exited) {
w->choked_next = 0;
have_unchoked = 1;
break;
}
}
for (unsigned type = 0; type < 2; type++)
for (unsigned i = 0; i < (type ? sch->nb_filters : sch->nb_demux); i++) {
SchWaiter *w = type ? &sch->filters[i].waiter : &sch->demux[i].waiter;
if (w->choked_prev != w->choked_next)
waiter_set(w, w->choked_next);
}
}
enum {
CYCLE_NODE_NEW = 0,
CYCLE_NODE_STARTED,
CYCLE_NODE_DONE,
};
static int
check_acyclic_for_output(const Scheduler *sch, SchedulerNode src,
uint8_t *filters_visited, SchedulerNode *filters_stack)
{
unsigned nb_filters_stack = 0;
memset(filters_visited, 0, sch->nb_filters * sizeof(*filters_visited));
while (1) {
const SchFilterGraph *fg = &sch->filters[src.idx];
filters_visited[src.idx] = CYCLE_NODE_STARTED;
// descend into every input, depth first
if (src.idx_stream < fg->nb_inputs) {
const SchFilterIn *fi = &fg->inputs[src.idx_stream++];
// connected to demuxer, no cycles possible
if (fi->src_sched.type == SCH_NODE_TYPE_DEMUX)
continue;
// otherwise connected to another filtergraph
av_assert0(fi->src_sched.type == SCH_NODE_TYPE_FILTER_OUT);
// found a cycle
if (filters_visited[fi->src_sched.idx] == CYCLE_NODE_STARTED)
return AVERROR(EINVAL);
// place current position on stack and descend
av_assert0(nb_filters_stack < sch->nb_filters);
filters_stack[nb_filters_stack++] = src;
src = (SchedulerNode){ .idx = fi->src_sched.idx, .idx_stream = 0 };
continue;
}
filters_visited[src.idx] = CYCLE_NODE_DONE;
// previous search finished,
if (nb_filters_stack) {
src = filters_stack[--nb_filters_stack];
continue;
}
return 0;
}
}
static int check_acyclic(Scheduler *sch)
{
uint8_t *filters_visited = NULL;
SchedulerNode *filters_stack = NULL;
int ret = 0;
if (!sch->nb_filters)
return 0;
filters_visited = av_malloc_array(sch->nb_filters, sizeof(*filters_visited));
if (!filters_visited)
return AVERROR(ENOMEM);
filters_stack = av_malloc_array(sch->nb_filters, sizeof(*filters_stack));
if (!filters_stack) {
ret = AVERROR(ENOMEM);
goto fail;
}
// trace the transcoding graph upstream from every output stream
// fed by a filtergraph
for (unsigned i = 0; i < sch->nb_mux; i++) {
SchMux *mux = &sch->mux[i];
for (unsigned j = 0; j < mux->nb_streams; j++) {
SchMuxStream *ms = &mux->streams[j];
SchedulerNode src = ms->src_sched;
if (src.type != SCH_NODE_TYPE_FILTER_OUT)
continue;
src.idx_stream = 0;
ret = check_acyclic_for_output(sch, src, filters_visited, filters_stack);
if (ret < 0) {
av_log(mux, AV_LOG_ERROR, "Transcoding graph has a cycle\n");
goto fail;
}
}
}
fail:
av_freep(&filters_visited);
av_freep(&filters_stack);
return ret;
}
static int start_prepare(Scheduler *sch)
{
int ret;
for (unsigned i = 0; i < sch->nb_demux; i++) {
SchDemux *d = &sch->demux[i];
for (unsigned j = 0; j < d->nb_streams; j++) {
SchDemuxStream *ds = &d->streams[j];
if (!ds->nb_dst) {
av_log(d, AV_LOG_ERROR,
"Demuxer stream %u not connected to any sink\n", j);
return AVERROR(EINVAL);
}
ds->dst_finished = av_calloc(ds->nb_dst, sizeof(*ds->dst_finished));
if (!ds->dst_finished)
return AVERROR(ENOMEM);
}
}
for (unsigned i = 0; i < sch->nb_dec; i++) {
SchDec *dec = &sch->dec[i];
if (!dec->src.type) {
av_log(dec, AV_LOG_ERROR,
"Decoder not connected to a source\n");
return AVERROR(EINVAL);
}
if (!dec->nb_dst) {
av_log(dec, AV_LOG_ERROR,
"Decoder not connected to any sink\n");
return AVERROR(EINVAL);
}
dec->dst_finished = av_calloc(dec->nb_dst, sizeof(*dec->dst_finished));
if (!dec->dst_finished)
return AVERROR(ENOMEM);
}
for (unsigned i = 0; i < sch->nb_enc; i++) {
SchEnc *enc = &sch->enc[i];
if (!enc->src.type) {
av_log(enc, AV_LOG_ERROR,
"Encoder not connected to a source\n");
return AVERROR(EINVAL);
}
if (!enc->nb_dst) {
av_log(enc, AV_LOG_ERROR,
"Encoder not connected to any sink\n");
return AVERROR(EINVAL);
}
enc->dst_finished = av_calloc(enc->nb_dst, sizeof(*enc->dst_finished));
if (!enc->dst_finished)
return AVERROR(ENOMEM);
}
for (unsigned i = 0; i < sch->nb_mux; i++) {
SchMux *mux = &sch->mux[i];
for (unsigned j = 0; j < mux->nb_streams; j++) {
SchMuxStream *ms = &mux->streams[j];
switch (ms->src.type) {
case SCH_NODE_TYPE_ENC: {
SchEnc *enc = &sch->enc[ms->src.idx];
if (enc->src.type == SCH_NODE_TYPE_DEC) {
ms->src_sched = sch->dec[enc->src.idx].src;
av_assert0(ms->src_sched.type == SCH_NODE_TYPE_DEMUX);
} else {
ms->src_sched = enc->src;
av_assert0(ms->src_sched.type == SCH_NODE_TYPE_FILTER_OUT);
}
break;
}
case SCH_NODE_TYPE_DEMUX:
ms->src_sched = ms->src;
break;
default:
av_log(mux, AV_LOG_ERROR,
"Muxer stream #%u not connected to a source\n", j);
return AVERROR(EINVAL);
}
}
ret = queue_alloc(&mux->queue, mux->nb_streams, mux->queue_size,
QUEUE_PACKETS);
if (ret < 0)
return ret;
}
for (unsigned i = 0; i < sch->nb_filters; i++) {
SchFilterGraph *fg = &sch->filters[i];
for (unsigned j = 0; j < fg->nb_inputs; j++) {
SchFilterIn *fi = &fg->inputs[j];
SchDec *dec;
if (!fi->src.type) {
av_log(fg, AV_LOG_ERROR,
"Filtergraph input %u not connected to a source\n", j);
return AVERROR(EINVAL);
}
av_assert0(fi->src.type == SCH_NODE_TYPE_DEC);
dec = &sch->dec[fi->src.idx];
switch (dec->src.type) {
case SCH_NODE_TYPE_DEMUX: fi->src_sched = dec->src; break;
case SCH_NODE_TYPE_ENC: fi->src_sched = sch->enc[dec->src.idx].src; break;
default: av_assert0(0);
}
}
for (unsigned j = 0; j < fg->nb_outputs; j++) {
SchFilterOut *fo = &fg->outputs[j];
if (!fo->dst.type) {
av_log(fg, AV_LOG_ERROR,
"Filtergraph %u output %u not connected to a sink\n", i, j);
return AVERROR(EINVAL);
}
}
}
// Check that the transcoding graph has no cycles.
ret = check_acyclic(sch);
if (ret < 0)
return ret;
return 0;
}
int sch_start(Scheduler *sch)
{
int ret;
ret = start_prepare(sch);
if (ret < 0)
return ret;
sch->transcode_started = 1;
for (unsigned i = 0; i < sch->nb_mux; i++) {
SchMux *mux = &sch->mux[i];
if (mux->nb_streams_ready == mux->nb_streams) {
ret = mux_init(sch, mux);
if (ret < 0)
return ret;
}
}
for (unsigned i = 0; i < sch->nb_enc; i++) {
SchEnc *enc = &sch->enc[i];
ret = task_start(&enc->task);
if (ret < 0)
return ret;
}
for (unsigned i = 0; i < sch->nb_filters; i++) {
SchFilterGraph *fg = &sch->filters[i];
ret = task_start(&fg->task);
if (ret < 0)
return ret;
}
for (unsigned i = 0; i < sch->nb_dec; i++) {
SchDec *dec = &sch->dec[i];
ret = task_start(&dec->task);
if (ret < 0)
return ret;
}
for (unsigned i = 0; i < sch->nb_demux; i++) {
SchDemux *d = &sch->demux[i];
if (!d->nb_streams)
continue;
ret = task_start(&d->task);
if (ret < 0)
return ret;
}
pthread_mutex_lock(&sch->schedule_lock);
schedule_update_locked(sch);
pthread_mutex_unlock(&sch->schedule_lock);
return 0;
}
int sch_wait(Scheduler *sch, uint64_t timeout_us, int64_t *transcode_ts)
{
int ret, err;
// convert delay to absolute timestamp
timeout_us += av_gettime();
pthread_mutex_lock(&sch->mux_done_lock);
if (sch->nb_mux_done < sch->nb_mux) {
struct timespec tv = { .tv_sec = timeout_us / 1000000,
.tv_nsec = (timeout_us % 1000000) * 1000 };
pthread_cond_timedwait(&sch->mux_done_cond, &sch->mux_done_lock, &tv);
}
ret = sch->nb_mux_done == sch->nb_mux;
pthread_mutex_unlock(&sch->mux_done_lock);
*transcode_ts = atomic_load(&sch->last_dts);
// abort transcoding if any task failed
err = atomic_load(&sch->task_failed);
return ret || err;
}
static int enc_open(Scheduler *sch, SchEnc *enc, const AVFrame *frame)
{
int ret;
ret = enc->open_cb(enc->task.func_arg, frame);
if (ret < 0)
return ret;
// ret>0 signals audio frame size, which means sync queue must
// have been enabled during encoder creation
if (ret > 0) {
SchSyncQueue *sq;
av_assert0(enc->sq_idx[0] >= 0);
sq = &sch->sq_enc[enc->sq_idx[0]];
pthread_mutex_lock(&sq->lock);
sq_frame_samples(sq->sq, enc->sq_idx[1], ret);
pthread_mutex_unlock(&sq->lock);
}
return 0;
}
static int send_to_enc_thread(Scheduler *sch, SchEnc *enc, AVFrame *frame)
{
int ret;
if (!frame) {
tq_send_finish(enc->queue, 0);
return 0;
}
if (enc->in_finished)
return AVERROR_EOF;
ret = tq_send(enc->queue, 0, frame);
if (ret < 0)
enc->in_finished = 1;
return ret;
}
static int send_to_enc_sq(Scheduler *sch, SchEnc *enc, AVFrame *frame)
{
SchSyncQueue *sq = &sch->sq_enc[enc->sq_idx[0]];
int ret = 0;
// inform the scheduling code that no more input will arrive along this path;
// this is necessary because the sync queue may not send an EOF downstream
// until other streams finish
// TODO: consider a cleaner way of passing this information through
// the pipeline
if (!frame) {
for (unsigned i = 0; i < enc->nb_dst; i++) {
SchMux *mux;
SchMuxStream *ms;
if (enc->dst[i].type != SCH_NODE_TYPE_MUX)
continue;
mux = &sch->mux[enc->dst[i].idx];
ms = &mux->streams[enc->dst[i].idx_stream];
pthread_mutex_lock(&sch->schedule_lock);
ms->source_finished = 1;
schedule_update_locked(sch);
pthread_mutex_unlock(&sch->schedule_lock);
}
}
pthread_mutex_lock(&sq->lock);
ret = sq_send(sq->sq, enc->sq_idx[1], SQFRAME(frame));
if (ret < 0)
goto finish;
while (1) {
SchEnc *enc;
// TODO: the SQ API should be extended to allow returning EOF
// for individual streams
ret = sq_receive(sq->sq, -1, SQFRAME(sq->frame));
if (ret < 0) {
ret = (ret == AVERROR(EAGAIN)) ? 0 : ret;
break;
}
enc = &sch->enc[sq->enc_idx[ret]];
ret = send_to_enc_thread(sch, enc, sq->frame);
if (ret < 0) {
av_frame_unref(sq->frame);
if (ret != AVERROR_EOF)
break;
sq_send(sq->sq, enc->sq_idx[1], SQFRAME(NULL));
continue;
}
}
if (ret < 0) {
// close all encoders fed from this sync queue
for (unsigned i = 0; i < sq->nb_enc_idx; i++) {
int err = send_to_enc_thread(sch, &sch->enc[sq->enc_idx[i]], NULL);
// if the sync queue error is EOF and closing the encoder
// produces a more serious error, make sure to pick the latter
ret = err_merge((ret == AVERROR_EOF && err < 0) ? 0 : ret, err);
}
}
finish:
pthread_mutex_unlock(&sq->lock);
return ret;
}
static int send_to_enc(Scheduler *sch, SchEnc *enc, AVFrame *frame)
{
if (enc->open_cb && frame && !enc->opened) {
int ret = enc_open(sch, enc, frame);
if (ret < 0)
return ret;
enc->opened = 1;
// discard empty frames that only carry encoder init parameters
if (!frame->buf[0]) {
av_frame_unref(frame);
return 0;
}
}
return (enc->sq_idx[0] >= 0) ?
send_to_enc_sq (sch, enc, frame) :
send_to_enc_thread(sch, enc, frame);
}
static int mux_queue_packet(SchMux *mux, SchMuxStream *ms, AVPacket *pkt)
{
PreMuxQueue *q = &ms->pre_mux_queue;
AVPacket *tmp_pkt = NULL;
int ret;
if (!av_fifo_can_write(q->fifo)) {
size_t packets = av_fifo_can_read(q->fifo);
size_t pkt_size = pkt ? pkt->size : 0;
int thresh_reached = (q->data_size + pkt_size) > q->data_threshold;
size_t max_packets = thresh_reached ? q->max_packets : SIZE_MAX;
size_t new_size = FFMIN(2 * packets, max_packets);
if (new_size <= packets) {
av_log(mux, AV_LOG_ERROR,
"Too many packets buffered for output stream.\n");
return AVERROR(ENOSPC);
}
ret = av_fifo_grow2(q->fifo, new_size - packets);
if (ret < 0)
return ret;
}
if (pkt) {
tmp_pkt = av_packet_alloc();
if (!tmp_pkt)
return AVERROR(ENOMEM);
av_packet_move_ref(tmp_pkt, pkt);
q->data_size += tmp_pkt->size;
}
av_fifo_write(q->fifo, &tmp_pkt, 1);
return 0;
}
static int send_to_mux(Scheduler *sch, SchMux *mux, unsigned stream_idx,
AVPacket *pkt)
{
SchMuxStream *ms = &mux->streams[stream_idx];
int64_t dts = (pkt && pkt->dts != AV_NOPTS_VALUE) ?
av_rescale_q(pkt->dts + pkt->duration, pkt->time_base, AV_TIME_BASE_Q) :
AV_NOPTS_VALUE;
// queue the packet if the muxer cannot be started yet
if (!atomic_load(&mux->mux_started)) {
int queued = 0;
// the muxer could have started between the above atomic check and
// locking the mutex, then this block falls through to normal send path
pthread_mutex_lock(&sch->mux_ready_lock);
if (!atomic_load(&mux->mux_started)) {
int ret = mux_queue_packet(mux, ms, pkt);
queued = ret < 0 ? ret : 1;
}
pthread_mutex_unlock(&sch->mux_ready_lock);
if (queued < 0)
return queued;
else if (queued)
goto update_schedule;
}
if (pkt) {
int ret;
if (ms->init_eof)
return AVERROR_EOF;
ret = tq_send(mux->queue, stream_idx, pkt);
if (ret < 0)
return ret;
} else
tq_send_finish(mux->queue, stream_idx);
update_schedule:
// TODO: use atomics to check whether this changes trailing dts
// to avoid locking unnecesarily
if (dts != AV_NOPTS_VALUE || !pkt) {
pthread_mutex_lock(&sch->schedule_lock);
if (pkt) ms->last_dts = dts;
else ms->source_finished = 1;
schedule_update_locked(sch);
pthread_mutex_unlock(&sch->schedule_lock);
}
return 0;
}
static int
demux_stream_send_to_dst(Scheduler *sch, const SchedulerNode dst,
uint8_t *dst_finished, AVPacket *pkt, unsigned flags)
{
int ret;
if (*dst_finished)
return AVERROR_EOF;
if (pkt && dst.type == SCH_NODE_TYPE_MUX &&
(flags & DEMUX_SEND_STREAMCOPY_EOF)) {
av_packet_unref(pkt);
pkt = NULL;
}
if (!pkt)
goto finish;
ret = (dst.type == SCH_NODE_TYPE_MUX) ?
send_to_mux(sch, &sch->mux[dst.idx], dst.idx_stream, pkt) :
tq_send(sch->dec[dst.idx].queue, 0, pkt);
if (ret == AVERROR_EOF)
goto finish;
return ret;
finish:
if (dst.type == SCH_NODE_TYPE_MUX)
send_to_mux(sch, &sch->mux[dst.idx], dst.idx_stream, NULL);
else
tq_send_finish(sch->dec[dst.idx].queue, 0);
*dst_finished = 1;
return AVERROR_EOF;
}
static int demux_send_for_stream(Scheduler *sch, SchDemux *d, SchDemuxStream *ds,
AVPacket *pkt, unsigned flags)
{
unsigned nb_done = 0;
for (unsigned i = 0; i < ds->nb_dst; i++) {
AVPacket *to_send = pkt;
uint8_t *finished = &ds->dst_finished[i];
int ret;
// sending a packet consumes it, so make a temporary reference if needed
if (pkt && i < ds->nb_dst - 1) {
to_send = d->send_pkt;
ret = av_packet_ref(to_send, pkt);
if (ret < 0)
return ret;
}
ret = demux_stream_send_to_dst(sch, ds->dst[i], finished, to_send, flags);
if (to_send)
av_packet_unref(to_send);
if (ret == AVERROR_EOF)
nb_done++;
else if (ret < 0)
return ret;
}
return (nb_done == ds->nb_dst) ? AVERROR_EOF : 0;
}
static int demux_flush(Scheduler *sch, SchDemux *d, AVPacket *pkt)
{
Timestamp max_end_ts = (Timestamp){ .ts = AV_NOPTS_VALUE };
av_assert0(!pkt->buf && !pkt->data && !pkt->side_data_elems);
for (unsigned i = 0; i < d->nb_streams; i++) {
SchDemuxStream *ds = &d->streams[i];
for (unsigned j = 0; j < ds->nb_dst; j++) {
const SchedulerNode *dst = &ds->dst[j];
SchDec *dec;
int ret;
if (ds->dst_finished[j] || dst->type != SCH_NODE_TYPE_DEC)
continue;
dec = &sch->dec[dst->idx];
ret = tq_send(dec->queue, 0, pkt);
if (ret < 0)
return ret;
if (dec->queue_end_ts) {
Timestamp ts;
ret = av_thread_message_queue_recv(dec->queue_end_ts, &ts, 0);
if (ret < 0)
return ret;
if (max_end_ts.ts == AV_NOPTS_VALUE ||
(ts.ts != AV_NOPTS_VALUE &&
av_compare_ts(max_end_ts.ts, max_end_ts.tb, ts.ts, ts.tb) < 0))
max_end_ts = ts;
}
}
}
pkt->pts = max_end_ts.ts;
pkt->time_base = max_end_ts.tb;
return 0;
}
int sch_demux_send(Scheduler *sch, unsigned demux_idx, AVPacket *pkt,
unsigned flags)
{
SchDemux *d;
int terminate;
av_assert0(demux_idx < sch->nb_demux);
d = &sch->demux[demux_idx];
terminate = waiter_wait(sch, &d->waiter);
if (terminate)
return AVERROR_EXIT;
// flush the downstreams after seek
if (pkt->stream_index == -1)
return demux_flush(sch, d, pkt);
av_assert0(pkt->stream_index < d->nb_streams);
return demux_send_for_stream(sch, d, &d->streams[pkt->stream_index], pkt, flags);
}
static int demux_done(Scheduler *sch, unsigned demux_idx)
{
SchDemux *d = &sch->demux[demux_idx];
int ret = 0;
for (unsigned i = 0; i < d->nb_streams; i++) {
int err = demux_send_for_stream(sch, d, &d->streams[i], NULL, 0);
if (err != AVERROR_EOF)
ret = err_merge(ret, err);
}
pthread_mutex_lock(&sch->schedule_lock);
d->task_exited = 1;
schedule_update_locked(sch);
pthread_mutex_unlock(&sch->schedule_lock);
return ret;
}
int sch_mux_receive(Scheduler *sch, unsigned mux_idx, AVPacket *pkt)
{
SchMux *mux;
int ret, stream_idx;
av_assert0(mux_idx < sch->nb_mux);
mux = &sch->mux[mux_idx];
ret = tq_receive(mux->queue, &stream_idx, pkt);
pkt->stream_index = stream_idx;
return ret;
}
void sch_mux_receive_finish(Scheduler *sch, unsigned mux_idx, unsigned stream_idx)
{
SchMux *mux;
av_assert0(mux_idx < sch->nb_mux);
mux = &sch->mux[mux_idx];
av_assert0(stream_idx < mux->nb_streams);
tq_receive_finish(mux->queue, stream_idx);
pthread_mutex_lock(&sch->schedule_lock);
mux->streams[stream_idx].source_finished = 1;
schedule_update_locked(sch);
pthread_mutex_unlock(&sch->schedule_lock);
}
int sch_mux_sub_heartbeat(Scheduler *sch, unsigned mux_idx, unsigned stream_idx,
const AVPacket *pkt)
{
SchMux *mux;
SchMuxStream *ms;
av_assert0(mux_idx < sch->nb_mux);
mux = &sch->mux[mux_idx];
av_assert0(stream_idx < mux->nb_streams);
ms = &mux->streams[stream_idx];
for (unsigned i = 0; i < ms->nb_sub_heartbeat_dst; i++) {
SchDec *dst = &sch->dec[ms->sub_heartbeat_dst[i]];
int ret;
ret = av_packet_copy_props(mux->sub_heartbeat_pkt, pkt);
if (ret < 0)
return ret;
tq_send(dst->queue, 0, mux->sub_heartbeat_pkt);
}
return 0;
}
static int mux_done(Scheduler *sch, unsigned mux_idx)
{
SchMux *mux = &sch->mux[mux_idx];
pthread_mutex_lock(&sch->schedule_lock);
for (unsigned i = 0; i < mux->nb_streams; i++) {
tq_receive_finish(mux->queue, i);
mux->streams[i].source_finished = 1;
}
schedule_update_locked(sch);
pthread_mutex_unlock(&sch->schedule_lock);
pthread_mutex_lock(&sch->mux_done_lock);
av_assert0(sch->nb_mux_done < sch->nb_mux);
sch->nb_mux_done++;
pthread_cond_signal(&sch->mux_done_cond);
pthread_mutex_unlock(&sch->mux_done_lock);
return 0;
}
int sch_dec_receive(Scheduler *sch, unsigned dec_idx, AVPacket *pkt)
{
SchDec *dec;
int ret, dummy;
av_assert0(dec_idx < sch->nb_dec);
dec = &sch->dec[dec_idx];
// the decoder should have given us post-flush end timestamp in pkt
if (dec->expect_end_ts) {
Timestamp ts = (Timestamp){ .ts = pkt->pts, .tb = pkt->time_base };
ret = av_thread_message_queue_send(dec->queue_end_ts, &ts, 0);
if (ret < 0)
return ret;
dec->expect_end_ts = 0;
}
ret = tq_receive(dec->queue, &dummy, pkt);
av_assert0(dummy <= 0);
// got a flush packet, on the next call to this function the decoder
// will give us post-flush end timestamp
if (ret >= 0 && !pkt->data && !pkt->side_data_elems && dec->queue_end_ts)
dec->expect_end_ts = 1;
return ret;
}
static int send_to_filter(Scheduler *sch, SchFilterGraph *fg,
unsigned in_idx, AVFrame *frame)
{
if (frame)
return tq_send(fg->queue, in_idx, frame);
if (!fg->inputs[in_idx].send_finished) {
fg->inputs[in_idx].send_finished = 1;
tq_send_finish(fg->queue, in_idx);
// close the control stream when all actual inputs are done
if (atomic_fetch_add(&fg->nb_inputs_finished_send, 1) == fg->nb_inputs - 1)
tq_send_finish(fg->queue, fg->nb_inputs);
}
return 0;
}
static int dec_send_to_dst(Scheduler *sch, const SchedulerNode dst,
uint8_t *dst_finished, AVFrame *frame)
{
int ret;
if (*dst_finished)
return AVERROR_EOF;
if (!frame)
goto finish;
ret = (dst.type == SCH_NODE_TYPE_FILTER_IN) ?
send_to_filter(sch, &sch->filters[dst.idx], dst.idx_stream, frame) :
send_to_enc(sch, &sch->enc[dst.idx], frame);
if (ret == AVERROR_EOF)
goto finish;
return ret;
finish:
if (dst.type == SCH_NODE_TYPE_FILTER_IN)
send_to_filter(sch, &sch->filters[dst.idx], dst.idx_stream, NULL);
else
send_to_enc(sch, &sch->enc[dst.idx], NULL);
*dst_finished = 1;
return AVERROR_EOF;
}
int sch_dec_send(Scheduler *sch, unsigned dec_idx, AVFrame *frame)
{
SchDec *dec;
int ret = 0;
unsigned nb_done = 0;
av_assert0(dec_idx < sch->nb_dec);
dec = &sch->dec[dec_idx];
for (unsigned i = 0; i < dec->nb_dst; i++) {
uint8_t *finished = &dec->dst_finished[i];
AVFrame *to_send = frame;
// sending a frame consumes it, so make a temporary reference if needed
if (i < dec->nb_dst - 1) {
to_send = dec->send_frame;
// frame may sometimes contain props only,
// e.g. to signal EOF timestamp
ret = frame->buf[0] ? av_frame_ref(to_send, frame) :
av_frame_copy_props(to_send, frame);
if (ret < 0)
return ret;
}
ret = dec_send_to_dst(sch, dec->dst[i], finished, to_send);
if (ret < 0) {
av_frame_unref(to_send);
if (ret == AVERROR_EOF) {
nb_done++;
ret = 0;
continue;
}
return ret;
}
}
return (nb_done == dec->nb_dst) ? AVERROR_EOF : 0;
}
static int dec_done(Scheduler *sch, unsigned dec_idx)
{
SchDec *dec = &sch->dec[dec_idx];
int ret = 0;
tq_receive_finish(dec->queue, 0);
// make sure our source does not get stuck waiting for end timestamps
// that will never arrive
if (dec->queue_end_ts)
av_thread_message_queue_set_err_recv(dec->queue_end_ts, AVERROR_EOF);
for (unsigned i = 0; i < dec->nb_dst; i++) {
int err = dec_send_to_dst(sch, dec->dst[i], &dec->dst_finished[i], NULL);
if (err < 0 && err != AVERROR_EOF)
ret = err_merge(ret, err);
}
return ret;
}
int sch_enc_receive(Scheduler *sch, unsigned enc_idx, AVFrame *frame)
{
SchEnc *enc;
int ret, dummy;
av_assert0(enc_idx < sch->nb_enc);
enc = &sch->enc[enc_idx];
ret = tq_receive(enc->queue, &dummy, frame);
av_assert0(dummy <= 0);
return ret;
}
static int enc_send_to_dst(Scheduler *sch, const SchedulerNode dst,
uint8_t *dst_finished, AVPacket *pkt)
{
int ret;
if (*dst_finished)
return AVERROR_EOF;
if (!pkt)
goto finish;
ret = (dst.type == SCH_NODE_TYPE_MUX) ?
send_to_mux(sch, &sch->mux[dst.idx], dst.idx_stream, pkt) :
tq_send(sch->dec[dst.idx].queue, 0, pkt);
if (ret == AVERROR_EOF)
goto finish;
return ret;
finish:
if (dst.type == SCH_NODE_TYPE_MUX)
send_to_mux(sch, &sch->mux[dst.idx], dst.idx_stream, NULL);
else
tq_send_finish(sch->dec[dst.idx].queue, 0);
*dst_finished = 1;
return AVERROR_EOF;
}
int sch_enc_send(Scheduler *sch, unsigned enc_idx, AVPacket *pkt)
{
SchEnc *enc;
int ret;
av_assert0(enc_idx < sch->nb_enc);
enc = &sch->enc[enc_idx];
for (unsigned i = 0; i < enc->nb_dst; i++) {
uint8_t *finished = &enc->dst_finished[i];
AVPacket *to_send = pkt;
// sending a packet consumes it, so make a temporary reference if needed
if (i < enc->nb_dst - 1) {
to_send = enc->send_pkt;
ret = av_packet_ref(to_send, pkt);
if (ret < 0)
return ret;
}
ret = enc_send_to_dst(sch, enc->dst[i], finished, to_send);
if (ret < 0) {
av_packet_unref(to_send);
if (ret == AVERROR_EOF) {
ret = 0;
continue;
}
return ret;
}
}
return ret;
}
static int enc_done(Scheduler *sch, unsigned enc_idx)
{
SchEnc *enc = &sch->enc[enc_idx];
int ret = 0;
tq_receive_finish(enc->queue, 0);
for (unsigned i = 0; i < enc->nb_dst; i++) {
int err = enc_send_to_dst(sch, enc->dst[i], &enc->dst_finished[i], NULL);
if (err < 0 && err != AVERROR_EOF)
ret = err_merge(ret, err);
}
return ret;
}
int sch_filter_receive(Scheduler *sch, unsigned fg_idx,
unsigned *in_idx, AVFrame *frame)
{
SchFilterGraph *fg;
av_assert0(fg_idx < sch->nb_filters);
fg = &sch->filters[fg_idx];
av_assert0(*in_idx <= fg->nb_inputs);
// update scheduling to account for desired input stream, if it changed
//
// this check needs no locking because only the filtering thread
// updates this value
if (*in_idx != fg->best_input) {
pthread_mutex_lock(&sch->schedule_lock);
fg->best_input = *in_idx;
schedule_update_locked(sch);
pthread_mutex_unlock(&sch->schedule_lock);
}
if (*in_idx == fg->nb_inputs) {
int terminate = waiter_wait(sch, &fg->waiter);
return terminate ? AVERROR_EOF : AVERROR(EAGAIN);
}
while (1) {
int ret, idx;
ret = tq_receive(fg->queue, &idx, frame);
if (idx < 0)
return AVERROR_EOF;
else if (ret >= 0) {
*in_idx = idx;
return 0;
}
// disregard EOFs for specific streams - they should always be
// preceded by an EOF frame
}
}
void sch_filter_receive_finish(Scheduler *sch, unsigned fg_idx, unsigned in_idx)
{
SchFilterGraph *fg;
SchFilterIn *fi;
av_assert0(fg_idx < sch->nb_filters);
fg = &sch->filters[fg_idx];
av_assert0(in_idx < fg->nb_inputs);
fi = &fg->inputs[in_idx];
if (!fi->receive_finished) {
fi->receive_finished = 1;
tq_receive_finish(fg->queue, in_idx);
// close the control stream when all actual inputs are done
if (++fg->nb_inputs_finished_receive == fg->nb_inputs)
tq_receive_finish(fg->queue, fg->nb_inputs);
}
}
int sch_filter_send(Scheduler *sch, unsigned fg_idx, unsigned out_idx, AVFrame *frame)
{
SchFilterGraph *fg;
av_assert0(fg_idx < sch->nb_filters);
fg = &sch->filters[fg_idx];
av_assert0(out_idx < fg->nb_outputs);
return send_to_enc(sch, &sch->enc[fg->outputs[out_idx].dst.idx], frame);
}
static int filter_done(Scheduler *sch, unsigned fg_idx)
{
SchFilterGraph *fg = &sch->filters[fg_idx];
int ret = 0;
for (unsigned i = 0; i <= fg->nb_inputs; i++)
tq_receive_finish(fg->queue, i);
for (unsigned i = 0; i < fg->nb_outputs; i++) {
SchEnc *enc = &sch->enc[fg->outputs[i].dst.idx];
int err = send_to_enc(sch, enc, NULL);
if (err < 0 && err != AVERROR_EOF)
ret = err_merge(ret, err);
}
pthread_mutex_lock(&sch->schedule_lock);
fg->task_exited = 1;
schedule_update_locked(sch);
pthread_mutex_unlock(&sch->schedule_lock);
return ret;
}
int sch_filter_command(Scheduler *sch, unsigned fg_idx, AVFrame *frame)
{
SchFilterGraph *fg;
av_assert0(fg_idx < sch->nb_filters);
fg = &sch->filters[fg_idx];
return send_to_filter(sch, fg, fg->nb_inputs, frame);
}
static void *task_wrapper(void *arg)
{
SchTask *task = arg;
Scheduler *sch = task->parent;
int ret;
int err = 0;
ret = task->func(task->func_arg);
if (ret < 0)
av_log(task->func_arg, AV_LOG_ERROR,
"Task finished with error code: %d (%s)\n", ret, av_err2str(ret));
switch (task->node.type) {
case SCH_NODE_TYPE_DEMUX: err = demux_done (sch, task->node.idx); break;
case SCH_NODE_TYPE_MUX: err = mux_done (sch, task->node.idx); break;
case SCH_NODE_TYPE_DEC: err = dec_done (sch, task->node.idx); break;
case SCH_NODE_TYPE_ENC: err = enc_done (sch, task->node.idx); break;
case SCH_NODE_TYPE_FILTER_IN: err = filter_done(sch, task->node.idx); break;
default: av_assert0(0);
}
ret = err_merge(ret, err);
// EOF is considered normal termination
if (ret == AVERROR_EOF)
ret = 0;
if (ret < 0)
atomic_store(&sch->task_failed, 1);
av_log(task->func_arg, ret < 0 ? AV_LOG_ERROR : AV_LOG_VERBOSE,
"Terminating thread with return code %d (%s)\n", ret,
ret < 0 ? av_err2str(ret) : "success");
return (void*)(intptr_t)ret;
}
View Git Blame Copy Permalink