mirror of
https://git.ffmpeg.org/ffmpeg.git
synced 2024-12-27 01:42:20 +00:00
ed364444f1
Fixes: CID1592383 Unused value Sponsored-by: Sovereign Tech Fund Signed-off-by: Michael Niedermayer <michael@niedermayer.cc>
2560 lines
66 KiB
C
2560 lines
66 KiB
C
/*
|
|
* 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;
|
|
|
|
enum SchedulerState {
|
|
SCH_STATE_UNINIT,
|
|
SCH_STATE_STARTED,
|
|
SCH_STATE_STOPPED,
|
|
};
|
|
|
|
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;
|
|
|
|
enum SchedulerState state;
|
|
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_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;
|
|
}
|
|
|
|
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;
|
|
|
|
av_assert0(src.idx < sch->nb_filters &&
|
|
src.idx_stream < sch->filters[src.idx].nb_outputs);
|
|
fo = &sch->filters[src.idx].outputs[src.idx_stream];
|
|
|
|
av_assert0(!fo->dst.type);
|
|
fo->dst = dst;
|
|
|
|
// filtered frames go to encoding or another filtergraph
|
|
switch (dst.type) {
|
|
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;
|
|
}
|
|
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;
|
|
}
|
|
default: av_assert0(0);
|
|
}
|
|
|
|
|
|
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->state >= SCH_STATE_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 filtegraph
|
|
for (unsigned i = 0; i < sch->nb_filters; i++) {
|
|
ret = check_acyclic_for_output(sch, (SchedulerNode){ .idx = i },
|
|
filters_visited, filters_stack);
|
|
if (ret < 0) {
|
|
av_log(&sch->filters[i], 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);
|
|
}
|
|
|
|
if (fi->src.type == SCH_NODE_TYPE_FILTER_OUT)
|
|
fi->src_sched = fi->src;
|
|
else {
|
|
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;
|
|
|
|
av_assert0(sch->state == SCH_STATE_UNINIT);
|
|
sch->state = SCH_STATE_STARTED;
|
|
|
|
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)
|
|
goto fail;
|
|
}
|
|
}
|
|
|
|
for (unsigned i = 0; i < sch->nb_enc; i++) {
|
|
SchEnc *enc = &sch->enc[i];
|
|
|
|
ret = task_start(&enc->task);
|
|
if (ret < 0)
|
|
goto fail;
|
|
}
|
|
|
|
for (unsigned i = 0; i < sch->nb_filters; i++) {
|
|
SchFilterGraph *fg = &sch->filters[i];
|
|
|
|
ret = task_start(&fg->task);
|
|
if (ret < 0)
|
|
goto fail;
|
|
}
|
|
|
|
for (unsigned i = 0; i < sch->nb_dec; i++) {
|
|
SchDec *dec = &sch->dec[i];
|
|
|
|
ret = task_start(&dec->task);
|
|
if (ret < 0)
|
|
goto fail;
|
|
}
|
|
|
|
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)
|
|
goto fail;
|
|
}
|
|
|
|
pthread_mutex_lock(&sch->schedule_lock);
|
|
schedule_update_locked(sch);
|
|
pthread_mutex_unlock(&sch->schedule_lock);
|
|
|
|
return 0;
|
|
fail:
|
|
sch_stop(sch, NULL);
|
|
return ret;
|
|
}
|
|
|
|
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;
|
|
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++;
|
|
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)
|
|
continue;
|
|
return ret;
|
|
}
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
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;
|
|
SchedulerNode dst;
|
|
|
|
av_assert0(fg_idx < sch->nb_filters);
|
|
fg = &sch->filters[fg_idx];
|
|
|
|
av_assert0(out_idx < fg->nb_outputs);
|
|
dst = fg->outputs[out_idx].dst;
|
|
|
|
return (dst.type == SCH_NODE_TYPE_ENC) ?
|
|
send_to_enc (sch, &sch->enc[dst.idx], frame) :
|
|
send_to_filter(sch, &sch->filters[dst.idx], dst.idx_stream, 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++) {
|
|
SchedulerNode dst = fg->outputs[i].dst;
|
|
int err = (dst.type == SCH_NODE_TYPE_ENC) ?
|
|
send_to_enc (sch, &sch->enc[dst.idx], NULL) :
|
|
send_to_filter(sch, &sch->filters[dst.idx], dst.idx_stream, 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 int task_cleanup(Scheduler *sch, SchedulerNode node)
|
|
{
|
|
switch (node.type) {
|
|
case SCH_NODE_TYPE_DEMUX: return demux_done (sch, node.idx);
|
|
case SCH_NODE_TYPE_MUX: return mux_done (sch, node.idx);
|
|
case SCH_NODE_TYPE_DEC: return dec_done (sch, node.idx);
|
|
case SCH_NODE_TYPE_ENC: return enc_done (sch, node.idx);
|
|
case SCH_NODE_TYPE_FILTER_IN: return filter_done(sch, node.idx);
|
|
default: av_assert0(0);
|
|
}
|
|
}
|
|
|
|
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));
|
|
|
|
err = task_cleanup(sch, task->node);
|
|
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;
|
|
}
|
|
|
|
static int task_stop(Scheduler *sch, SchTask *task)
|
|
{
|
|
int ret;
|
|
void *thread_ret;
|
|
|
|
if (!task->thread_running)
|
|
return task_cleanup(sch, task->node);
|
|
|
|
ret = pthread_join(task->thread, &thread_ret);
|
|
av_assert0(ret == 0);
|
|
|
|
task->thread_running = 0;
|
|
|
|
return (intptr_t)thread_ret;
|
|
}
|
|
|
|
int sch_stop(Scheduler *sch, int64_t *finish_ts)
|
|
{
|
|
int ret = 0, err;
|
|
|
|
if (sch->state != SCH_STATE_STARTED)
|
|
return 0;
|
|
|
|
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(sch, &d->task);
|
|
ret = err_merge(ret, err);
|
|
}
|
|
|
|
for (unsigned i = 0; i < sch->nb_dec; i++) {
|
|
SchDec *dec = &sch->dec[i];
|
|
|
|
err = task_stop(sch, &dec->task);
|
|
ret = err_merge(ret, err);
|
|
}
|
|
|
|
for (unsigned i = 0; i < sch->nb_filters; i++) {
|
|
SchFilterGraph *fg = &sch->filters[i];
|
|
|
|
err = task_stop(sch, &fg->task);
|
|
ret = err_merge(ret, err);
|
|
}
|
|
|
|
for (unsigned i = 0; i < sch->nb_enc; i++) {
|
|
SchEnc *enc = &sch->enc[i];
|
|
|
|
err = task_stop(sch, &enc->task);
|
|
ret = err_merge(ret, err);
|
|
}
|
|
|
|
for (unsigned i = 0; i < sch->nb_mux; i++) {
|
|
SchMux *mux = &sch->mux[i];
|
|
|
|
err = task_stop(sch, &mux->task);
|
|
ret = err_merge(ret, err);
|
|
}
|
|
|
|
if (finish_ts)
|
|
*finish_ts = trailing_dts(sch, 1);
|
|
|
|
sch->state = SCH_STATE_STOPPED;
|
|
|
|
return ret;
|
|
}
|