ffmpeg/fftools/ffmpeg_sched.h

490 lines
20 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
*/
#ifndef FFTOOLS_FFMPEG_SCHED_H
#define FFTOOLS_FFMPEG_SCHED_H
#include <stddef.h>
#include <stdint.h>
#include "ffmpeg_utils.h"
/*
* This file contains the API for the transcode scheduler.
*
* Overall architecture of the transcoding process involves instances of the
* following components:
* - demuxers, each containing any number of demuxed streams; demuxed packets
* belonging to some stream are sent to any number of decoders (transcoding)
* and/or muxers (streamcopy);
* - decoders, which receive encoded packets from some demuxed stream or
* encoder, decode them, and send decoded frames to any number of filtergraph
* inputs (audio/video) or encoders (subtitles);
* - filtergraphs, each containing zero or more inputs (0 in case the
* filtergraph contains a lavfi source filter), and one or more outputs; the
* inputs and outputs need not have matching media types;
* each filtergraph input receives decoded frames from some decoder;
* filtered frames from each output are sent to some encoder;
* - encoders, which receive decoded frames from some decoder (subtitles) or
* some filtergraph output (audio/video), encode them, and send encoded
* packets to any number of muxed streams or decoders;
* - muxers, each containing any number of muxed streams; each muxed stream
* receives encoded packets from some demuxed stream (streamcopy) or some
* encoder (transcoding); those packets are interleaved and written out by the
* muxer.
*
* There must be at least one muxer instance, otherwise the transcode produces
* no output and is meaningless. Otherwise, in a generic transcoding scenario
* there may be arbitrary number of instances of any of the above components,
* interconnected in various ways.
*
* The code tries to keep all the output streams across all the muxers in sync
* (i.e. at the same DTS), which is accomplished by varying the rates at which
* packets are read from different demuxers and lavfi sources. Note that the
* degree of control we have over synchronization is fundamentally limited - if
* some demuxed streams in the same input are interleaved at different rates
* than that at which they are to be muxed (e.g. because an input file is badly
* interleaved, or the user changed their speed by mismatching amounts), then
* there will be increasing amounts of buffering followed by eventual
* transcoding failure.
*
* N.B. 1: there are meaningful transcode scenarios with no demuxers, e.g.
* - encoding and muxing output from filtergraph(s) that have no inputs;
* - creating a file that contains nothing but attachments and/or metadata.
*
* N.B. 2: a filtergraph output could, in principle, feed multiple encoders, but
* this is unnecessary because the (a)split filter provides the same
* functionality.
*
* The scheduler, in the above model, is the master object that oversees and
* facilitates the transcoding process. The basic idea is that all instances
* of the abovementioned components communicate only with the scheduler and not
* with each other. The scheduler is then the single place containing the
* knowledge about the whole transcoding pipeline.
*/
struct AVFrame;
struct AVPacket;
typedef struct Scheduler Scheduler;
enum SchedulerNodeType {
SCH_NODE_TYPE_NONE = 0,
SCH_NODE_TYPE_DEMUX,
SCH_NODE_TYPE_MUX,
SCH_NODE_TYPE_DEC,
SCH_NODE_TYPE_ENC,
SCH_NODE_TYPE_FILTER_IN,
SCH_NODE_TYPE_FILTER_OUT,
};
typedef struct SchedulerNode {
enum SchedulerNodeType type;
unsigned idx;
unsigned idx_stream;
} SchedulerNode;
typedef int (*SchThreadFunc)(void *arg);
#define SCH_DSTREAM(file, stream) \
(SchedulerNode){ .type = SCH_NODE_TYPE_DEMUX, \
.idx = file, .idx_stream = stream }
#define SCH_MSTREAM(file, stream) \
(SchedulerNode){ .type = SCH_NODE_TYPE_MUX, \
.idx = file, .idx_stream = stream }
#define SCH_DEC(decoder) \
(SchedulerNode){ .type = SCH_NODE_TYPE_DEC, \
.idx = decoder }
#define SCH_ENC(encoder) \
(SchedulerNode){ .type = SCH_NODE_TYPE_ENC, \
.idx = encoder }
#define SCH_FILTER_IN(filter, input) \
(SchedulerNode){ .type = SCH_NODE_TYPE_FILTER_IN, \
.idx = filter, .idx_stream = input }
#define SCH_FILTER_OUT(filter, output) \
(SchedulerNode){ .type = SCH_NODE_TYPE_FILTER_OUT, \
.idx = filter, .idx_stream = output }
Scheduler *sch_alloc(void);
void sch_free(Scheduler **sch);
int sch_start(Scheduler *sch);
int sch_stop(Scheduler *sch, int64_t *finish_ts);
/**
* Wait until transcoding terminates or the specified timeout elapses.
*
* @param timeout_us Amount of time in microseconds after which this function
* will timeout.
* @param transcode_ts Current transcode timestamp in AV_TIME_BASE_Q, for
* informational purposes only.
*
* @retval 0 waiting timed out, transcoding is not finished
* @retval 1 transcoding is finished
*/
int sch_wait(Scheduler *sch, uint64_t timeout_us, int64_t *transcode_ts);
/**
* Add a demuxer to the scheduler.
*
* @param func Function executed as the demuxer task.
* @param ctx Demuxer state; will be passed to func and used for logging.
*
* @retval ">=0" Index of the newly-created demuxer.
* @retval "<0" Error code.
*/
int sch_add_demux(Scheduler *sch, SchThreadFunc func, void *ctx);
/**
* Add a demuxed stream for a previously added demuxer.
*
* @param demux_idx index previously returned by sch_add_demux()
*
* @retval ">=0" Index of the newly-created demuxed stream.
* @retval "<0" Error code.
*/
int sch_add_demux_stream(Scheduler *sch, unsigned demux_idx);
/**
* Add a decoder to the scheduler.
*
* @param func Function executed as the decoder task.
* @param ctx Decoder state; will be passed to func and used for logging.
* @param send_end_ts The decoder will return an end timestamp after flush packets
* are delivered to it. See documentation for
* sch_dec_receive() for more details.
*
* @retval ">=0" Index of the newly-created decoder.
* @retval "<0" Error code.
*/
int sch_add_dec(Scheduler *sch, SchThreadFunc func, void *ctx,
int send_end_ts);
/**
* Add a filtergraph to the scheduler.
*
* @param nb_inputs Number of filtergraph inputs.
* @param nb_outputs number of filtergraph outputs
* @param func Function executed as the filtering task.
* @param ctx Filter state; will be passed to func and used for logging.
*
* @retval ">=0" Index of the newly-created filtergraph.
* @retval "<0" Error code.
*/
int sch_add_filtergraph(Scheduler *sch, unsigned nb_inputs, unsigned nb_outputs,
SchThreadFunc func, void *ctx);
/**
* Add a muxer to the scheduler.
*
* Note that muxer thread startup is more complicated than for other components,
* because
* - muxer streams fed by audio/video encoders become initialized dynamically at
* runtime, after those encoders receive their first frame and initialize
* themselves, followed by calling sch_mux_stream_ready()
* - the header can be written after all the streams for a muxer are initialized
* - we may need to write an SDP, which must happen
* - AFTER all the headers are written
* - BEFORE any packets are written by any muxer
* - with all the muxers quiescent
* To avoid complicated muxer-thread synchronization dances, we postpone
* starting the muxer threads until after the SDP is written. The sequence of
* events is then as follows:
* - After sch_mux_stream_ready() is called for all the streams in a given muxer,
* the header for that muxer is written (care is taken that headers for
* different muxers are not written concurrently, since they write file
* information to stderr). If SDP is not wanted, the muxer thread then starts
* and muxing begins.
* - When SDP _is_ wanted, no muxer threads start until the header for the last
* muxer is written. After that, the SDP is written, after which all the muxer
* threads are started at once.
*
* In order for the above to work, the scheduler needs to be able to invoke
* just writing the header, which is the reason the init parameter exists.
*
* @param func Function executed as the muxing task.
* @param init Callback that is called to initialize the muxer and write the
* header. Called after sch_mux_stream_ready() is called for all the
* streams in the muxer.
* @param ctx Muxer state; will be passed to func/init and used for logging.
* @param sdp_auto Determines automatic SDP writing - see sch_sdp_filename().
* @param thread_queue_size number of packets that can be buffered before
* sending to the muxer blocks
*
* @retval ">=0" Index of the newly-created muxer.
* @retval "<0" Error code.
*/
int sch_add_mux(Scheduler *sch, SchThreadFunc func, int (*init)(void *),
void *ctx, int sdp_auto, unsigned thread_queue_size);
/**
* Default size of a packet thread queue. For muxing this can be overridden by
* the thread_queue_size option as passed to a call to sch_add_mux().
*/
#define DEFAULT_PACKET_THREAD_QUEUE_SIZE 8
/**
* Default size of a frame thread queue.
*/
#define DEFAULT_FRAME_THREAD_QUEUE_SIZE 8
/**
* Add a muxed stream for a previously added muxer.
*
* @param mux_idx index previously returned by sch_add_mux()
*
* @retval ">=0" Index of the newly-created muxed stream.
* @retval "<0" Error code.
*/
int sch_add_mux_stream(Scheduler *sch, unsigned mux_idx);
/**
* Configure limits on packet buffering performed before the muxer task is
* started.
*
* @param mux_idx index previously returned by sch_add_mux()
* @param stream_idx_idx index previously returned by sch_add_mux_stream()
* @param data_threshold Total size of the buffered packets' data after which
* max_packets applies.
* @param max_packets maximum Maximum number of buffered packets after
* data_threshold is reached.
*/
void sch_mux_stream_buffering(Scheduler *sch, unsigned mux_idx, unsigned stream_idx,
size_t data_threshold, int max_packets);
/**
* Signal to the scheduler that the specified muxed stream is initialized and
* ready. Muxing is started once all the streams are ready.
*/
int sch_mux_stream_ready(Scheduler *sch, unsigned mux_idx, unsigned stream_idx);
/**
* Set the file path for the SDP.
*
* The SDP is written when either of the following is true:
* - this function is called at least once
* - sdp_auto=1 is passed to EVERY call of sch_add_mux()
*/
int sch_sdp_filename(Scheduler *sch, const char *sdp_filename);
/**
* Add an encoder to the scheduler.
*
* @param func Function executed as the encoding task.
* @param ctx Encoder state; will be passed to func and used for logging.
* @param open_cb This callback, if specified, will be called when the first
* frame is obtained for this encoder. For audio encoders with a
* fixed frame size (which use a sync queue in the scheduler to
* rechunk frames), it must return that frame size on success.
* Otherwise (non-audio, variable frame size) it should return 0.
*
* @retval ">=0" Index of the newly-created encoder.
* @retval "<0" Error code.
*/
int sch_add_enc(Scheduler *sch, SchThreadFunc func, void *ctx,
int (*open_cb)(void *func_arg, const struct AVFrame *frame));
/**
* Add an pre-encoding sync queue to the scheduler.
*
* @param buf_size_us Sync queue buffering size, passed to sq_alloc().
* @param logctx Logging context for the sync queue. passed to sq_alloc().
*
* @retval ">=0" Index of the newly-created sync queue.
* @retval "<0" Error code.
*/
int sch_add_sq_enc(Scheduler *sch, uint64_t buf_size_us, void *logctx);
int sch_sq_add_enc(Scheduler *sch, unsigned sq_idx, unsigned enc_idx,
int limiting, uint64_t max_frames);
int sch_connect(Scheduler *sch, SchedulerNode src, SchedulerNode dst);
enum DemuxSendFlags {
/**
* Treat the packet as an EOF for SCH_NODE_TYPE_MUX destinations
* send normally to other types.
*/
DEMUX_SEND_STREAMCOPY_EOF = (1 << 0),
};
/**
* Called by demuxer tasks to communicate with their downstreams. The following
* may be sent:
* - a demuxed packet for the stream identified by pkt->stream_index;
* - demuxer discontinuity/reset (e.g. after a seek) - this is signalled by an
* empty packet with stream_index=-1.
*
* @param demux_idx demuxer index
* @param pkt A demuxed packet to send.
* When flushing (i.e. pkt->stream_index=-1 on entry to this
* function), on successful return pkt->pts/pkt->time_base will be
* set to the maximum end timestamp of any decoded audio stream, or
* AV_NOPTS_VALUE if no decoded audio streams are present.
*
* @retval "non-negative value" success
* @retval AVERROR_EOF all consumers for the stream are done
* @retval AVERROR_EXIT all consumers are done, should terminate demuxing
* @retval "anoter negative error code" other failure
*/
int sch_demux_send(Scheduler *sch, unsigned demux_idx, struct AVPacket *pkt,
unsigned flags);
/**
* Called by decoder tasks to receive a packet for decoding.
*
* @param dec_idx decoder index
* @param pkt Input packet will be written here on success.
*
* An empty packet signals that the decoder should be flushed, but
* more packets will follow (e.g. after seeking). When a decoder
* created with send_end_ts=1 receives a flush packet, it must write
* the end timestamp of the stream after flushing to
* pkt->pts/time_base on the next call to this function (if any).
*
* @retval "non-negative value" success
* @retval AVERROR_EOF no more packets will arrive, should terminate decoding
* @retval "another negative error code" other failure
*/
int sch_dec_receive(Scheduler *sch, unsigned dec_idx, struct AVPacket *pkt);
/**
* Called by decoder tasks to send a decoded frame downstream.
*
* @param dec_idx Decoder index previously returned by sch_add_dec().
* @param frame Decoded frame; on success it is consumed and cleared by this
* function
*
* @retval ">=0" success
* @retval AVERROR_EOF all consumers are done, should terminate decoding
* @retval "another negative error code" other failure
*/
int sch_dec_send(Scheduler *sch, unsigned dec_idx, struct AVFrame *frame);
/**
* Called by filtergraph tasks to obtain frames for filtering. Will wait for a
* frame to become available and return it in frame.
*
* Filtergraphs that contain lavfi sources and do not currently require new
* input frames should call this function as a means of rate control - then
* in_idx should be set equal to nb_inputs on entry to this function.
*
* @param fg_idx Filtergraph index previously returned by sch_add_filtergraph().
* @param[in,out] in_idx On input contains the index of the input on which a frame
* is most desired. May be set to nb_inputs to signal that
* the filtergraph does not need more input currently.
*
* On success, will be replaced with the input index of
* the actually returned frame or EOF timestamp.
*
* @retval ">=0" Frame data or EOF timestamp was delivered into frame, in_idx
* contains the index of the input it belongs to.
* @retval AVERROR(EAGAIN) No frame was returned, the filtergraph should
* resume filtering. May only be returned when
* in_idx=nb_inputs on entry to this function.
* @retval AVERROR_EOF No more frames will arrive, should terminate filtering.
*/
int sch_filter_receive(Scheduler *sch, unsigned fg_idx,
unsigned *in_idx, struct AVFrame *frame);
/**
* Called by filter tasks to signal that a filter input will no longer accept input.
*
* @param fg_idx Filtergraph index previously returned from sch_add_filtergraph().
* @param in_idx Index of the input to finish.
*/
void sch_filter_receive_finish(Scheduler *sch, unsigned fg_idx, unsigned in_idx);
/**
* Called by filtergraph tasks to send a filtered frame or EOF to consumers.
*
* @param fg_idx Filtergraph index previously returned by sch_add_filtergraph().
* @param out_idx Index of the output which produced the frame.
* @param frame The frame to send to consumers. When NULL, signals that no more
* frames will be produced for the specified output. When non-NULL,
* the frame is consumed and cleared by this function on success.
*
* @retval "non-negative value" success
* @retval AVERROR_EOF all consumers are done
* @retval "anoter negative error code" other failure
*/
int sch_filter_send(Scheduler *sch, unsigned fg_idx, unsigned out_idx,
struct AVFrame *frame);
int sch_filter_command(Scheduler *sch, unsigned fg_idx, struct AVFrame *frame);
/**
* Called by encoder tasks to obtain frames for encoding. Will wait for a frame
* to become available and return it in frame.
*
* @param enc_idx Encoder index previously returned by sch_add_enc().
* @param frame Newly-received frame will be stored here on success. Must be
* clean on entrance to this function.
*
* @retval 0 A frame was successfully delivered into frame.
* @retval AVERROR_EOF No more frames will be delivered, the encoder should
* flush everything and terminate.
*
*/
int sch_enc_receive(Scheduler *sch, unsigned enc_idx, struct AVFrame *frame);
/**
* Called by encoder tasks to send encoded packets downstream.
*
* @param enc_idx Encoder index previously returned by sch_add_enc().
* @param pkt An encoded packet; it will be consumed and cleared by this
* function on success.
*
* @retval 0 success
* @retval "<0" Error code.
*/
int sch_enc_send (Scheduler *sch, unsigned enc_idx, struct AVPacket *pkt);
/**
* Called by muxer tasks to obtain packets for muxing. Will wait for a packet
* for any muxed stream to become available and return it in pkt.
*
* @param mux_idx Muxer index previously returned by sch_add_mux().
* @param pkt Newly-received packet will be stored here on success. Must be
* clean on entrance to this function.
*
* @retval 0 A packet was successfully delivered into pkt. Its stream_index
* corresponds to a stream index previously returned from
* sch_add_mux_stream().
* @retval AVERROR_EOF When pkt->stream_index is non-negative, this signals that
* no more packets will be delivered for this stream index.
* Otherwise this indicates that no more packets will be
* delivered for any stream and the muxer should therefore
* flush everything and terminate.
*/
int sch_mux_receive(Scheduler *sch, unsigned mux_idx, struct AVPacket *pkt);
/**
* Called by muxer tasks to signal that a stream will no longer accept input.
*
* @param stream_idx Stream index previously returned from sch_add_mux_stream().
*/
void sch_mux_receive_finish(Scheduler *sch, unsigned mux_idx, unsigned stream_idx);
int sch_mux_sub_heartbeat_add(Scheduler *sch, unsigned mux_idx, unsigned stream_idx,
unsigned dec_idx);
int sch_mux_sub_heartbeat(Scheduler *sch, unsigned mux_idx, unsigned stream_idx,
const AVPacket *pkt);
#endif /* FFTOOLS_FFMPEG_SCHED_H */