fftools/ffmpeg_sched.h

   1 /*
   2  * Inter-thread scheduling/synchronization.
   3  * Copyright (c) 2023 Anton Khirnov
   4  *
   5  * This file is part of FFmpeg.
   6  *
   7  * FFmpeg is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Lesser General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2.1 of the License, or (at your option) any later version.
  11  *
  12  * FFmpeg is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Lesser General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Lesser General Public
  18  * License along with FFmpeg; if not, write to the Free Software
  19  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  20  */
  21
  22 #ifndef FFTOOLS_FFMPEG_SCHED_H
  23 #define FFTOOLS_FFMPEG_SCHED_H
  24
  25 #include <stddef.h>
  26 #include <stdint.h>
  27
  28 #include "ffmpeg_utils.h"
  29
  30 /*
  31  * This file contains the API for the transcode scheduler.
  32  *
  33  * Overall architecture of the transcoding process involves instances of the
  34  * following components:
  35  * - demuxers, each containing any number of demuxed streams; demuxed packets
  36  *   belonging to some stream are sent to any number of decoders (transcoding)
  37  *   and/or muxers (streamcopy);
  38  * - decoders, which receive encoded packets from some demuxed stream or
  39  *   encoder, decode them, and send decoded frames to any number of filtergraph
  40  *   inputs (audio/video) or encoders (subtitles);
  41  * - filtergraphs, each containing zero or more inputs (0 in case the
  42  *   filtergraph contains a lavfi source filter), and one or more outputs; the
  43  *   inputs and outputs need not have matching media types;
  44  *   each filtergraph input receives decoded frames from some decoder or another
  45  *   filtergraph output;
  46  *   filtered frames from each output are sent to some encoder;
  47  * - encoders, which receive decoded frames from some decoder (subtitles) or
  48  *   some filtergraph output (audio/video), encode them, and send encoded
  49  *   packets to any number of muxed streams or decoders;
  50  * - muxers, each containing any number of muxed streams; each muxed stream
  51  *   receives encoded packets from some demuxed stream (streamcopy) or some
  52  *   encoder (transcoding); those packets are interleaved and written out by the
  53  *   muxer.
  54  *
  55  * The structure formed by the above components is a directed acyclic graph
  56  * (absence of cycles is checked at startup).
  57  *
  58  * There must be at least one muxer instance, otherwise the transcode produces
  59  * no output and is meaningless. Otherwise, in a generic transcoding scenario
  60  * there may be arbitrary number of instances of any of the above components,
  61  * interconnected in various ways.
  62  *
  63  * The code tries to keep all the output streams across all the muxers in sync
  64  * (i.e. at the same DTS), which is accomplished by varying the rates at which
  65  * packets are read from different demuxers and lavfi sources. Note that the
  66  * degree of control we have over synchronization is fundamentally limited - if
  67  * some demuxed streams in the same input are interleaved at different rates
  68  * than that at which they are to be muxed (e.g. because an input file is badly
  69  * interleaved, or the user changed their speed by mismatching amounts), then
  70  * there will be increasing amounts of buffering followed by eventual
  71  * transcoding failure.
  72  *
  73  * N.B. 1: there are meaningful transcode scenarios with no demuxers, e.g.
  74  * - encoding and muxing output from filtergraph(s) that have no inputs;
  75  * - creating a file that contains nothing but attachments and/or metadata.
  76  *
  77  * N.B. 2: a filtergraph output could, in principle, feed multiple encoders, but
  78  * this is unnecessary because the (a)split filter provides the same
  79  * functionality.
  80  *
  81  * The scheduler, in the above model, is the master object that oversees and
  82  * facilitates the transcoding process. The basic idea is that all instances
  83  * of the abovementioned components communicate only with the scheduler and not
  84  * with each other. The scheduler is then the single place containing the
  85  * knowledge about the whole transcoding pipeline.
  86  */
  87
  88 struct AVFrame;
  89 struct AVPacket;
  90
  91 typedef struct Scheduler Scheduler;
  92
  93 enum SchedulerNodeType {
  94     SCH_NODE_TYPE_NONE = 0,
  95     SCH_NODE_TYPE_DEMUX,
  96     SCH_NODE_TYPE_MUX,
  97     SCH_NODE_TYPE_DEC,
  98     SCH_NODE_TYPE_ENC,
  99     SCH_NODE_TYPE_FILTER_IN,
 100     SCH_NODE_TYPE_FILTER_OUT,
 101 };
 102
 103 typedef struct SchedulerNode {
 104     enum SchedulerNodeType  type;
 105     unsigned                idx;
 106     unsigned                idx_stream;
 107 } SchedulerNode;
 108
 109 typedef int (*SchThreadFunc)(void *arg);
 110
 111 #define SCH_DSTREAM(file, stream)                           \
 112     (SchedulerNode){ .type = SCH_NODE_TYPE_DEMUX,           \
 113                      .idx = file, .idx_stream = stream }
 114 #define SCH_MSTREAM(file, stream)                           \
 115     (SchedulerNode){ .type = SCH_NODE_TYPE_MUX,             \
 116                      .idx = file, .idx_stream = stream }
 117 #define SCH_DEC_IN(decoder)                                 \
 118     (SchedulerNode){ .type = SCH_NODE_TYPE_DEC,             \
 119                      .idx = decoder }
 120 #define SCH_DEC_OUT(decoder, out_idx)                       \
 121     (SchedulerNode){ .type = SCH_NODE_TYPE_DEC,             \
 122                      .idx = decoder, .idx_stream = out_idx }
 123 #define SCH_ENC(encoder)                                    \
 124     (SchedulerNode){ .type = SCH_NODE_TYPE_ENC,             \
 125                     .idx = encoder }
 126 #define SCH_FILTER_IN(filter, input)                        \
 127     (SchedulerNode){ .type = SCH_NODE_TYPE_FILTER_IN,       \
 128                      .idx = filter, .idx_stream = input }
 129 #define SCH_FILTER_OUT(filter, output)                      \
 130     (SchedulerNode){ .type = SCH_NODE_TYPE_FILTER_OUT,      \
 131                      .idx = filter, .idx_stream = output }
 132
 133 Scheduler *sch_alloc(void);
 134 void sch_free(Scheduler **sch);
 135
 136 int sch_start(Scheduler *sch);
 137 int sch_stop(Scheduler *sch, int64_t *finish_ts);
 138
 139 /**
 140  * Wait until transcoding terminates or the specified timeout elapses.
 141  *
 142  * @param timeout_us Amount of time in microseconds after which this function
 143  *                   will timeout.
 144  * @param transcode_ts Current transcode timestamp in AV_TIME_BASE_Q, for
 145  *                     informational purposes only.
 146  *
 147  * @retval 0 waiting timed out, transcoding is not finished
 148  * @retval 1 transcoding is finished
 149  */
 150 int sch_wait(Scheduler *sch, uint64_t timeout_us, int64_t *transcode_ts);
 151
 152 /**
 153  * Add a demuxer to the scheduler.
 154  *
 155  * @param func Function executed as the demuxer task.
 156  * @param ctx Demuxer state; will be passed to func and used for logging.
 157  *
 158  * @retval ">=0" Index of the newly-created demuxer.
 159  * @retval "<0"  Error code.
 160  */
 161 int sch_add_demux(Scheduler *sch, SchThreadFunc func, void *ctx);
 162 /**
 163  * Add a demuxed stream for a previously added demuxer.
 164  *
 165  * @param demux_idx index previously returned by sch_add_demux()
 166  *
 167  * @retval ">=0" Index of the newly-created demuxed stream.
 168  * @retval "<0"  Error code.
 169  */
 170 int sch_add_demux_stream(Scheduler *sch, unsigned demux_idx);
 171
 172 /**
 173  * Add a decoder to the scheduler.
 174  *
 175  * @param func Function executed as the decoder task.
 176  * @param ctx Decoder state; will be passed to func and used for logging.
 177  * @param send_end_ts The decoder will return an end timestamp after flush packets
 178  *                    are delivered to it. See documentation for
 179  *                    sch_dec_receive() for more details.
 180  *
 181  * @retval ">=0" Index of the newly-created decoder.
 182  * @retval "<0"  Error code.
 183  */
 184 int sch_add_dec(Scheduler *sch, SchThreadFunc func, void *ctx, int send_end_ts);
 185
 186 /**
 187  * Add another output to decoder (e.g. for multiview video).
 188  *
 189  * @retval ">=0" Index of the newly-added decoder output.
 190  * @retval "<0"  Error code.
 191  */
 192 int sch_add_dec_output(Scheduler *sch, unsigned dec_idx);
 193
 194 /**
 195  * Add a filtergraph to the scheduler.
 196  *
 197  * @param nb_inputs Number of filtergraph inputs.
 198  * @param nb_outputs number of filtergraph outputs
 199  * @param func Function executed as the filtering task.
 200  * @param ctx Filter state; will be passed to func and used for logging.
 201  *
 202  * @retval ">=0" Index of the newly-created filtergraph.
 203  * @retval "<0"  Error code.
 204  */
 205 int sch_add_filtergraph(Scheduler *sch, unsigned nb_inputs, unsigned nb_outputs,
 206                         SchThreadFunc func, void *ctx);
 207
 208 void sch_remove_filtergraph(Scheduler *sch, int idx);
 209
 210 /**
 211  * Add a muxer to the scheduler.
 212  *
 213  * Note that muxer thread startup is more complicated than for other components,
 214  * because
 215  * - muxer streams fed by audio/video encoders become initialized dynamically at
 216  *   runtime, after those encoders receive their first frame and initialize
 217  *   themselves, followed by calling sch_mux_stream_ready()
 218  * - the header can be written after all the streams for a muxer are initialized
 219  * - we may need to write an SDP, which must happen
 220  *      - AFTER all the headers are written
 221  *      - BEFORE any packets are written by any muxer
 222  *      - with all the muxers quiescent
 223  * To avoid complicated muxer-thread synchronization dances, we postpone
 224  * starting the muxer threads until after the SDP is written. The sequence of
 225  * events is then as follows:
 226  * - After sch_mux_stream_ready() is called for all the streams in a given muxer,
 227  *   the header for that muxer is written (care is taken that headers for
 228  *   different muxers are not written concurrently, since they write file
 229  *   information to stderr). If SDP is not wanted, the muxer thread then starts
 230  *   and muxing begins.
 231  * - When SDP _is_ wanted, no muxer threads start until the header for the last
 232  *   muxer is written. After that, the SDP is written, after which all the muxer
 233  *   threads are started at once.
 234  *
 235  * In order for the above to work, the scheduler needs to be able to invoke
 236  * just writing the header, which is the reason the init parameter exists.
 237  *
 238  * @param func Function executed as the muxing task.
 239  * @param init Callback that is called to initialize the muxer and write the
 240  *             header. Called after sch_mux_stream_ready() is called for all the
 241  *             streams in the muxer.
 242  * @param ctx Muxer state; will be passed to func/init and used for logging.
 243  * @param sdp_auto Determines automatic SDP writing - see sch_sdp_filename().
 244  * @param thread_queue_size number of packets that can be buffered before
 245  *                          sending to the muxer blocks
 246  *
 247  * @retval ">=0" Index of the newly-created muxer.
 248  * @retval "<0"  Error code.
 249  */
 250 int sch_add_mux(Scheduler *sch, SchThreadFunc func, int (*init)(void *),
 251                 void *ctx, int sdp_auto, unsigned thread_queue_size);
 252
 253 /**
 254  * Default size of a packet thread queue.  For muxing this can be overridden by
 255  * the thread_queue_size option as passed to a call to sch_add_mux().
 256  */
 257 #define DEFAULT_PACKET_THREAD_QUEUE_SIZE 8
 258
 259 /**
 260  * Default size of a frame thread queue.
 261  */
 262 #define DEFAULT_FRAME_THREAD_QUEUE_SIZE 2
 263
 264 /**
 265  * Add a muxed stream for a previously added muxer.
 266  *
 267  * @param mux_idx index previously returned by sch_add_mux()
 268  *
 269  * @retval ">=0" Index of the newly-created muxed stream.
 270  * @retval "<0"  Error code.
 271  */
 272 int sch_add_mux_stream(Scheduler *sch, unsigned mux_idx);
 273
 274 /**
 275  * Configure limits on packet buffering performed before the muxer task is
 276  * started.
 277  *
 278  * @param mux_idx index previously returned by sch_add_mux()
 279  * @param stream_idx_idx index previously returned by sch_add_mux_stream()
 280  * @param data_threshold Total size of the buffered packets' data after which
 281  *                       max_packets applies.
 282  * @param max_packets maximum Maximum number of buffered packets after
 283  *                            data_threshold is reached.
 284  */
 285 void sch_mux_stream_buffering(Scheduler *sch, unsigned mux_idx, unsigned stream_idx,
 286                               size_t data_threshold, int max_packets);
 287
 288 /**
 289  * Signal to the scheduler that the specified muxed stream is initialized and
 290  * ready. Muxing is started once all the streams are ready.
 291  */
 292 int sch_mux_stream_ready(Scheduler *sch, unsigned mux_idx, unsigned stream_idx);
 293
 294 /**
 295  * Set the file path for the SDP.
 296  *
 297  * The SDP is written when either of the following is true:
 298  * - this function is called at least once
 299  * - sdp_auto=1 is passed to EVERY call of sch_add_mux()
 300  */
 301 int sch_sdp_filename(Scheduler *sch, const char *sdp_filename);
 302
 303 /**
 304  * Add an encoder to the scheduler.
 305  *
 306  * @param func Function executed as the encoding task.
 307  * @param ctx Encoder state; will be passed to func and used for logging.
 308  * @param open_cb This callback, if specified, will be called when the first
 309  *                frame is obtained for this encoder. For audio encoders with a
 310  *                fixed frame size (which use a sync queue in the scheduler to
 311  *                rechunk frames), it must return that frame size on success.
 312  *                Otherwise (non-audio, variable frame size) it should return 0.
 313  *
 314  * @retval ">=0" Index of the newly-created encoder.
 315  * @retval "<0"  Error code.
 316  */
 317 int sch_add_enc(Scheduler *sch, SchThreadFunc func, void *ctx,
 318                 int (*open_cb)(void *func_arg, const struct AVFrame *frame));
 319
 320 /**
 321  * Add an pre-encoding sync queue to the scheduler.
 322  *
 323  * @param buf_size_us Sync queue buffering size, passed to sq_alloc().
 324  * @param logctx Logging context for the sync queue. passed to sq_alloc().
 325  *
 326  * @retval ">=0" Index of the newly-created sync queue.
 327  * @retval "<0"  Error code.
 328  */
 329 int sch_add_sq_enc(Scheduler *sch, uint64_t buf_size_us, void *logctx);
 330 int sch_sq_add_enc(Scheduler *sch, unsigned sq_idx, unsigned enc_idx,
 331                    int limiting, uint64_t max_frames);
 332
 333 int sch_connect(Scheduler *sch, SchedulerNode src, SchedulerNode dst);
 334
 335 enum DemuxSendFlags {
 336     /**
 337      * Treat the packet as an EOF for SCH_NODE_TYPE_MUX destinations
 338      * send normally to other types.
 339      */
 340     DEMUX_SEND_STREAMCOPY_EOF = (1 << 0),
 341 };
 342
 343 /**
 344  * Called by demuxer tasks to communicate with their downstreams. The following
 345  * may be sent:
 346  * - a demuxed packet for the stream identified by pkt->stream_index;
 347  * - demuxer discontinuity/reset (e.g. after a seek) - this is signalled by an
 348  *   empty packet with stream_index=-1.
 349  *
 350  * @param demux_idx demuxer index
 351  * @param pkt A demuxed packet to send.
 352  *            When flushing (i.e. pkt->stream_index=-1 on entry to this
 353  *            function), on successful return pkt->pts/pkt->time_base will be
 354  *            set to the maximum end timestamp of any decoded audio stream, or
 355  *            AV_NOPTS_VALUE if no decoded audio streams are present.
 356  *
 357  * @retval "non-negative value" success
 358  * @retval AVERROR_EOF all consumers for the stream are done
 359  * @retval AVERROR_EXIT all consumers are done, should terminate demuxing
 360  * @retval "another negative error code" other failure
 361  */
 362 int sch_demux_send(Scheduler *sch, unsigned demux_idx, struct AVPacket *pkt,
 363                    unsigned flags);
 364
 365 /**
 366  * Called by decoder tasks to receive a packet for decoding.
 367  *
 368  * @param dec_idx decoder index
 369  * @param pkt Input packet will be written here on success.
 370  *
 371  *            An empty packet signals that the decoder should be flushed, but
 372  *            more packets will follow (e.g. after seeking). When a decoder
 373  *            created with send_end_ts=1 receives a flush packet, it must write
 374  *            the end timestamp of the stream after flushing to
 375  *            pkt->pts/time_base on the next call to this function (if any).
 376  *
 377  * @retval "non-negative value" success
 378  * @retval AVERROR_EOF no more packets will arrive, should terminate decoding
 379  * @retval "another negative error code" other failure
 380  */
 381 int sch_dec_receive(Scheduler *sch, unsigned dec_idx, struct AVPacket *pkt);
 382
 383 /**
 384  * Called by decoder tasks to send a decoded frame downstream.
 385  *
 386  * @param dec_idx Decoder index previously returned by sch_add_dec().
 387  * @param frame Decoded frame; on success it is consumed and cleared by this
 388  *              function
 389  *
 390  * @retval ">=0" success
 391  * @retval AVERROR_EOF all consumers are done, should terminate decoding
 392  * @retval "another negative error code" other failure
 393  */
 394 int sch_dec_send(Scheduler *sch, unsigned dec_idx,
 395                  unsigned out_idx, struct AVFrame *frame);
 396
 397 /**
 398  * Called by filtergraph tasks to obtain frames for filtering. Will wait for a
 399  * frame to become available and return it in frame.
 400  *
 401  * Filtergraphs that contain lavfi sources and do not currently require new
 402  * input frames should call this function as a means of rate control - then
 403  * in_idx should be set equal to nb_inputs on entry to this function.
 404  *
 405  * @param fg_idx Filtergraph index previously returned by sch_add_filtergraph().
 406  * @param[in,out] in_idx On input contains the index of the input on which a frame
 407  *                       is most desired. May be set to nb_inputs to signal that
 408  *                       the filtergraph does not need more input currently.
 409  *
 410  *                       On success, will be replaced with the input index of
 411  *                       the actually returned frame or EOF timestamp.
 412  *
 413  * @retval ">=0" Frame data or EOF timestamp was delivered into frame, in_idx
 414  *               contains the index of the input it belongs to.
 415  * @retval AVERROR(EAGAIN) No frame was returned, the filtergraph should
 416  *                         resume filtering. May only be returned when
 417  *                         in_idx=nb_inputs on entry to this function.
 418  * @retval AVERROR_EOF No more frames will arrive, should terminate filtering.
 419  */
 420 int sch_filter_receive(Scheduler *sch, unsigned fg_idx,
 421                        unsigned *in_idx, struct AVFrame *frame);
 422 /**
 423  * Called by filter tasks to signal that a filter input will no longer accept input.
 424  *
 425  * @param fg_idx Filtergraph index previously returned from sch_add_filtergraph().
 426  * @param in_idx Index of the input to finish.
 427  */
 428 void sch_filter_receive_finish(Scheduler *sch, unsigned fg_idx, unsigned in_idx);
 429
 430 /**
 431  * Called by filtergraph tasks to send a filtered frame or EOF to consumers.
 432  *
 433  * @param fg_idx Filtergraph index previously returned by sch_add_filtergraph().
 434  * @param out_idx Index of the output which produced the frame.
 435  * @param frame The frame to send to consumers. When NULL, signals that no more
 436  *              frames will be produced for the specified output. When non-NULL,
 437  *              the frame is consumed and cleared by this function on success.
 438  *
 439  * @retval "non-negative value" success
 440  * @retval AVERROR_EOF all consumers are done
 441  * @retval "another negative error code" other failure
 442  */
 443 int sch_filter_send(Scheduler *sch, unsigned fg_idx, unsigned out_idx,
 444                     struct AVFrame *frame);
 445
 446 int sch_filter_command(Scheduler *sch, unsigned fg_idx, struct AVFrame *frame);
 447
 448 /**
 449  * Called by filtergraph tasks to choke all filter inputs, preventing them from
 450  * receiving more frames until woken up again by the scheduler. Used during
 451  * initial graph configuration to avoid unnecessary buffering.
 452  */
 453 void sch_filter_choke_inputs(Scheduler *sch, unsigned fg_idx);
 454
 455 /**
 456  * Called by encoder tasks to obtain frames for encoding. Will wait for a frame
 457  * to become available and return it in frame.
 458  *
 459  * @param enc_idx Encoder index previously returned by sch_add_enc().
 460  * @param frame   Newly-received frame will be stored here on success. Must be
 461  *                clean on entrance to this function.
 462  *
 463  * @retval 0 A frame was successfully delivered into frame.
 464  * @retval AVERROR_EOF No more frames will be delivered, the encoder should
 465  *                     flush everything and terminate.
 466  *
 467  */
 468 int sch_enc_receive(Scheduler *sch, unsigned enc_idx, struct AVFrame *frame);
 469
 470 /**
 471  * Called by encoder tasks to send encoded packets downstream.
 472  *
 473  * @param enc_idx Encoder index previously returned by sch_add_enc().
 474  * @param pkt     An encoded packet; it will be consumed and cleared by this
 475  *                function on success.
 476  *
 477  * @retval 0     success
 478  * @retval "<0"  Error code.
 479  */
 480 int sch_enc_send   (Scheduler *sch, unsigned enc_idx, struct AVPacket *pkt);
 481
 482 /**
 483  * Called by muxer tasks to obtain packets for muxing. Will wait for a packet
 484  * for any muxed stream to become available and return it in pkt.
 485  *
 486  * @param mux_idx Muxer index previously returned by sch_add_mux().
 487  * @param pkt     Newly-received packet will be stored here on success. Must be
 488  *                clean on entrance to this function.
 489  *
 490  * @retval 0 A packet was successfully delivered into pkt. Its stream_index
 491  *           corresponds to a stream index previously returned from
 492  *           sch_add_mux_stream().
 493  * @retval AVERROR_EOF When pkt->stream_index is non-negative, this signals that
 494  *                     no more packets will be delivered for this stream index.
 495  *                     Otherwise this indicates that no more packets will be
 496  *                     delivered for any stream and the muxer should therefore
 497  *                     flush everything and terminate.
 498  */
 499 int sch_mux_receive(Scheduler *sch, unsigned mux_idx, struct AVPacket *pkt);
 500
 501 /**
 502  * Called by muxer tasks to signal that a stream will no longer accept input.
 503  *
 504  * @param stream_idx Stream index previously returned from sch_add_mux_stream().
 505  */
 506 void sch_mux_receive_finish(Scheduler *sch, unsigned mux_idx, unsigned stream_idx);
 507
 508 int sch_mux_sub_heartbeat_add(Scheduler *sch, unsigned mux_idx, unsigned stream_idx,
 509                               unsigned dec_idx);
 510 int sch_mux_sub_heartbeat(Scheduler *sch, unsigned mux_idx, unsigned stream_idx,
 511                           const AVPacket *pkt);
 512
 513 #endif /* FFTOOLS_FFMPEG_SCHED_H */