mirror of https://github.com/FFmpeg/FFmpeg.git
The -shortest option (which finishes the output file at the time the shortest stream ends) is currently implemented by faking the -t option when an output stream ends. This approach is fragile, since it depends on the frames/packets being processed in a specific order. E.g. there are currently some situations in which the output file length will depend unpredictably on unrelated factors like encoder delay. More importantly, the present work aiming at splitting various ffmpeg components into different threads will make this approach completely unworkable, since the frames/packets will arrive in effectively random order. This commit introduces a "sync queue", which is essentially a collection of FIFOs, one per stream. Frames/packets are submitted to these FIFOs and are then released for further processing (encoding or muxing) when it is ensured that the frame in question will not cause its stream to get ahead of the other streams (the logic is similar to libavformat's interleaving queue). These sync queues are then used for encoding and/or muxing when the -shortest option is specified. A new option – -shortest_buf_duration – controls the maximum number of queued packets, to avoid runaway memory usage. This commit changes the results of the following tests: - copy-shortest[12]: the last audio frame is now gone. This is correct, since it actually outlasts the last video frame. - shortest-sub: the video packets following the last subtitle packet are now gone. This is also correct.pull/388/head
Anton Khirnov
2 years ago
parent
9ac78fb347
commit
4740fea7dd
13 changed files with 778 additions and 41 deletions
@ -0,0 +1,425 @@ |
||||
/*
|
||||
* 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 <stdint.h> |
||||
#include <string.h> |
||||
|
||||
#include "libavutil/avassert.h" |
||||
#include "libavutil/error.h" |
||||
#include "libavutil/fifo.h" |
||||
#include "libavutil/mathematics.h" |
||||
#include "libavutil/mem.h" |
||||
|
||||
#include "objpool.h" |
||||
#include "sync_queue.h" |
||||
|
||||
typedef struct SyncQueueStream { |
||||
AVFifo *fifo; |
||||
AVRational tb; |
||||
|
||||
/* stream head: largest timestamp seen */ |
||||
int64_t head_ts; |
||||
/* no more frames will be sent for this stream */ |
||||
int finished; |
||||
} SyncQueueStream; |
||||
|
||||
struct SyncQueue { |
||||
enum SyncQueueType type; |
||||
|
||||
/* no more frames will be sent for any stream */ |
||||
int finished; |
||||
/* sync head: the stream with the _smallest_ head timestamp
|
||||
* this stream determines which frames can be output */ |
||||
int head_stream; |
||||
/* the finished stream with the smallest finish timestamp or -1 */ |
||||
int head_finished_stream; |
||||
|
||||
// maximum buffering duration in microseconds
|
||||
int64_t buf_size_us; |
||||
|
||||
SyncQueueStream *streams; |
||||
unsigned int nb_streams; |
||||
|
||||
// pool of preallocated frames to avoid constant allocations
|
||||
ObjPool *pool; |
||||
}; |
||||
|
||||
static void frame_move(const SyncQueue *sq, SyncQueueFrame dst, |
||||
SyncQueueFrame src) |
||||
{ |
||||
if (sq->type == SYNC_QUEUE_PACKETS) |
||||
av_packet_move_ref(dst.p, src.p); |
||||
else |
||||
av_frame_move_ref(dst.f, src.f); |
||||
} |
||||
|
||||
static int64_t frame_ts(const SyncQueue *sq, SyncQueueFrame frame) |
||||
{ |
||||
return (sq->type == SYNC_QUEUE_PACKETS) ? |
||||
frame.p->pts + frame.p->duration : |
||||
frame.f->pts + frame.f->pkt_duration; |
||||
} |
||||
|
||||
static int frame_null(const SyncQueue *sq, SyncQueueFrame frame) |
||||
{ |
||||
return (sq->type == SYNC_QUEUE_PACKETS) ? (frame.p == NULL) : (frame.f == NULL); |
||||
} |
||||
|
||||
static void finish_stream(SyncQueue *sq, unsigned int stream_idx) |
||||
{ |
||||
SyncQueueStream *st = &sq->streams[stream_idx]; |
||||
|
||||
st->finished = 1; |
||||
|
||||
if (st->head_ts != AV_NOPTS_VALUE) { |
||||
/* check if this stream is the new finished head */ |
||||
if (sq->head_finished_stream < 0 || |
||||
av_compare_ts(st->head_ts, st->tb, |
||||
sq->streams[sq->head_finished_stream].head_ts, |
||||
sq->streams[sq->head_finished_stream].tb) < 0) { |
||||
sq->head_finished_stream = stream_idx; |
||||
} |
||||
|
||||
/* mark as finished all streams that should no longer receive new frames,
|
||||
* due to them being ahead of some finished stream */ |
||||
st = &sq->streams[sq->head_finished_stream]; |
||||
for (unsigned int i = 0; i < sq->nb_streams; i++) { |
||||
SyncQueueStream *st1 = &sq->streams[i]; |
||||
if (st != st1 && st1->head_ts != AV_NOPTS_VALUE && |
||||
av_compare_ts(st->head_ts, st->tb, st1->head_ts, st1->tb) <= 0) |
||||
st1->finished = 1; |
||||
} |
||||
} |
||||
|
||||
/* mark the whole queue as finished if all streams are finished */ |
||||
for (unsigned int i = 0; i < sq->nb_streams; i++) { |
||||
if (!sq->streams[i].finished) |
||||
return; |
||||
} |
||||
sq->finished = 1; |
||||
} |
||||
|
||||
static void queue_head_update(SyncQueue *sq) |
||||
{ |
||||
if (sq->head_stream < 0) { |
||||
/* wait for one timestamp in each stream before determining
|
||||
* the queue head */ |
||||
for (unsigned int i = 0; i < sq->nb_streams; i++) { |
||||
SyncQueueStream *st = &sq->streams[i]; |
||||
if (st->head_ts == AV_NOPTS_VALUE) |
||||
return; |
||||
} |
||||
|
||||
// placeholder value, correct one will be found below
|
||||
sq->head_stream = 0; |
||||
} |
||||
|
||||
for (unsigned int i = 0; i < sq->nb_streams; i++) { |
||||
SyncQueueStream *st_head = &sq->streams[sq->head_stream]; |
||||
SyncQueueStream *st_other = &sq->streams[i]; |
||||
if (st_other->head_ts != AV_NOPTS_VALUE && |
||||
av_compare_ts(st_other->head_ts, st_other->tb, |
||||
st_head->head_ts, st_head->tb) < 0) |
||||
sq->head_stream = i; |
||||
} |
||||
} |
||||
|
||||
/* update this stream's head timestamp */ |
||||
static void stream_update_ts(SyncQueue *sq, unsigned int stream_idx, int64_t ts) |
||||
{ |
||||
SyncQueueStream *st = &sq->streams[stream_idx]; |
||||
|
||||
if (ts == AV_NOPTS_VALUE || |
||||
(st->head_ts != AV_NOPTS_VALUE && st->head_ts >= ts)) |
||||
return; |
||||
|
||||
st->head_ts = ts; |
||||
|
||||
/* if this stream is now ahead of some finished stream, then
|
||||
* this stream is also finished */ |
||||
if (sq->head_finished_stream >= 0 && |
||||
av_compare_ts(sq->streams[sq->head_finished_stream].head_ts, |
||||
sq->streams[sq->head_finished_stream].tb, |
||||
ts, st->tb) <= 0) |
||||
finish_stream(sq, stream_idx); |
||||
|
||||
/* update the overall head timestamp if it could have changed */ |
||||
if (sq->head_stream < 0 || sq->head_stream == stream_idx) |
||||
queue_head_update(sq); |
||||
} |
||||
|
||||
/* If the queue for the given stream (or all streams when stream_idx=-1)
|
||||
* is overflowing, trigger a fake heartbeat on lagging streams. |
||||
* |
||||
* @return 1 if heartbeat triggered, 0 otherwise |
||||
*/ |
||||
static int overflow_heartbeat(SyncQueue *sq, int stream_idx) |
||||
{ |
||||
SyncQueueStream *st; |
||||
SyncQueueFrame frame; |
||||
int64_t tail_ts = AV_NOPTS_VALUE; |
||||
|
||||
/* if no stream specified, pick the one that is most ahead */ |
||||
if (stream_idx < 0) { |
||||
int64_t ts = AV_NOPTS_VALUE; |
||||
|
||||
for (int i = 0; i < sq->nb_streams; i++) { |
||||
st = &sq->streams[i]; |
||||
if (st->head_ts != AV_NOPTS_VALUE && |
||||
(ts == AV_NOPTS_VALUE || |
||||
av_compare_ts(ts, sq->streams[stream_idx].tb, |
||||
st->head_ts, st->tb) < 0)) { |
||||
ts = st->head_ts; |
||||
stream_idx = i; |
||||
} |
||||
} |
||||
/* no stream has a timestamp yet -> nothing to do */ |
||||
if (stream_idx < 0) |
||||
return 0; |
||||
} |
||||
|
||||
st = &sq->streams[stream_idx]; |
||||
|
||||
/* get the chosen stream's tail timestamp */ |
||||
for (size_t i = 0; tail_ts == AV_NOPTS_VALUE && |
||||
av_fifo_peek(st->fifo, &frame, 1, i) >= 0; i++) |
||||
tail_ts = frame_ts(sq, frame); |
||||
|
||||
/* overflow triggers when the tail is over specified duration behind the head */ |
||||
if (tail_ts == AV_NOPTS_VALUE || tail_ts >= st->head_ts || |
||||
av_rescale_q(st->head_ts - tail_ts, st->tb, AV_TIME_BASE_Q) < sq->buf_size_us) |
||||
return 0; |
||||
|
||||
/* signal a fake timestamp for all streams that prevent tail_ts from being output */ |
||||
tail_ts++; |
||||
for (unsigned int i = 0; i < sq->nb_streams; i++) { |
||||
SyncQueueStream *st1 = &sq->streams[i]; |
||||
int64_t ts; |
||||
|
||||
if (st == st1 || st1->finished || |
||||
(st1->head_ts != AV_NOPTS_VALUE && |
||||
av_compare_ts(tail_ts, st->tb, st1->head_ts, st1->tb) <= 0)) |
||||
continue; |
||||
|
||||
ts = av_rescale_q(tail_ts, st->tb, st1->tb); |
||||
if (st1->head_ts != AV_NOPTS_VALUE) |
||||
ts = FFMAX(st1->head_ts + 1, ts); |
||||
|
||||
stream_update_ts(sq, i, ts); |
||||
} |
||||
|
||||
return 1; |
||||
} |
||||
|
||||
int sq_send(SyncQueue *sq, unsigned int stream_idx, SyncQueueFrame frame) |
||||
{ |
||||
SyncQueueStream *st; |
||||
SyncQueueFrame dst; |
||||
int64_t ts; |
||||
int ret; |
||||
|
||||
av_assert0(stream_idx < sq->nb_streams); |
||||
st = &sq->streams[stream_idx]; |
||||
|
||||
av_assert0(st->tb.num > 0 && st->tb.den > 0); |
||||
|
||||
if (frame_null(sq, frame)) { |
||||
finish_stream(sq, stream_idx); |
||||
return 0; |
||||
} |
||||
if (st->finished) |
||||
return AVERROR_EOF; |
||||
|
||||
ret = objpool_get(sq->pool, (void**)&dst); |
||||
if (ret < 0) |
||||
return ret; |
||||
|
||||
frame_move(sq, dst, frame); |
||||
|
||||
ts = frame_ts(sq, dst); |
||||
|
||||
ret = av_fifo_write(st->fifo, &dst, 1); |
||||
if (ret < 0) { |
||||
frame_move(sq, frame, dst); |
||||
objpool_release(sq->pool, (void**)&dst); |
||||
return ret; |
||||
} |
||||
|
||||
stream_update_ts(sq, stream_idx, ts); |
||||
|
||||
return 0; |
||||
} |
||||
|
||||
static int receive_for_stream(SyncQueue *sq, unsigned int stream_idx, |
||||
SyncQueueFrame frame) |
||||
{ |
||||
SyncQueueStream *st_head = sq->head_stream >= 0 ? |
||||
&sq->streams[sq->head_stream] : NULL; |
||||
SyncQueueStream *st; |
||||
|
||||
av_assert0(stream_idx < sq->nb_streams); |
||||
st = &sq->streams[stream_idx]; |
||||
|
||||
if (av_fifo_can_read(st->fifo)) { |
||||
SyncQueueFrame peek; |
||||
int64_t ts; |
||||
int cmp = 1; |
||||
|
||||
av_fifo_peek(st->fifo, &peek, 1, 0); |
||||
ts = frame_ts(sq, peek); |
||||
|
||||
/* check if this stream's tail timestamp does not overtake
|
||||
* the overall queue head */ |
||||
if (ts != AV_NOPTS_VALUE && st_head) |
||||
cmp = av_compare_ts(ts, st->tb, st_head->head_ts, st_head->tb); |
||||
|
||||
/* We can release frames that do not end after the queue head.
|
||||
* Frames with no timestamps are just passed through with no conditions. |
||||
*/ |
||||
if (cmp <= 0 || ts == AV_NOPTS_VALUE) { |
||||
frame_move(sq, frame, peek); |
||||
objpool_release(sq->pool, (void**)&peek); |
||||
av_fifo_drain2(st->fifo, 1); |
||||
return 0; |
||||
} |
||||
} |
||||
|
||||
return (sq->finished || (st->finished && !av_fifo_can_read(st->fifo))) ? |
||||
AVERROR_EOF : AVERROR(EAGAIN); |
||||
} |
||||
|
||||
static int receive_internal(SyncQueue *sq, int stream_idx, SyncQueueFrame frame) |
||||
{ |
||||
int nb_eof = 0; |
||||
int ret; |
||||
|
||||
/* read a frame for a specific stream */ |
||||
if (stream_idx >= 0) { |
||||
ret = receive_for_stream(sq, stream_idx, frame); |
||||
return (ret < 0) ? ret : stream_idx; |
||||
} |
||||
|
||||
/* read a frame for any stream with available output */ |
||||
for (unsigned int i = 0; i < sq->nb_streams; i++) { |
||||
ret = receive_for_stream(sq, i, frame); |
||||
if (ret == AVERROR_EOF || ret == AVERROR(EAGAIN)) { |
||||
nb_eof += (ret == AVERROR_EOF); |
||||
continue; |
||||
} |
||||
return (ret < 0) ? ret : i; |
||||
} |
||||
|
||||
return (nb_eof == sq->nb_streams) ? AVERROR_EOF : AVERROR(EAGAIN); |
||||
} |
||||
|
||||
int sq_receive(SyncQueue *sq, int stream_idx, SyncQueueFrame frame) |
||||
{ |
||||
int ret = receive_internal(sq, stream_idx, frame); |
||||
|
||||
/* try again if the queue overflowed and triggered a fake heartbeat
|
||||
* for lagging streams */ |
||||
if (ret == AVERROR(EAGAIN) && overflow_heartbeat(sq, stream_idx)) |
||||
ret = receive_internal(sq, stream_idx, frame); |
||||
|
||||
return ret; |
||||
} |
||||
|
||||
int sq_add_stream(SyncQueue *sq) |
||||
{ |
||||
SyncQueueStream *tmp, *st; |
||||
|
||||
tmp = av_realloc_array(sq->streams, sq->nb_streams + 1, sizeof(*sq->streams)); |
||||
if (!tmp) |
||||
return AVERROR(ENOMEM); |
||||
sq->streams = tmp; |
||||
|
||||
st = &sq->streams[sq->nb_streams]; |
||||
memset(st, 0, sizeof(*st)); |
||||
|
||||
st->fifo = av_fifo_alloc2(1, sizeof(SyncQueueFrame), AV_FIFO_FLAG_AUTO_GROW); |
||||
if (!st->fifo) |
||||
return AVERROR(ENOMEM); |
||||
|
||||
/* we set a valid default, so that a pathological stream that never
|
||||
* receives even a real timebase (and no frames) won't stall all other |
||||
* streams forever; cf. overflow_heartbeat() */ |
||||
st->tb = (AVRational){ 1, 1 }; |
||||
st->head_ts = AV_NOPTS_VALUE; |
||||
|
||||
return sq->nb_streams++; |
||||
} |
||||
|
||||
void sq_set_tb(SyncQueue *sq, unsigned int stream_idx, AVRational tb) |
||||
{ |
||||
SyncQueueStream *st; |
||||
|
||||
av_assert0(stream_idx < sq->nb_streams); |
||||
st = &sq->streams[stream_idx]; |
||||
|
||||
av_assert0(!av_fifo_can_read(st->fifo)); |
||||
|
||||
if (st->head_ts != AV_NOPTS_VALUE) |
||||
st->head_ts = av_rescale_q(st->head_ts, st->tb, tb); |
||||
|
||||
st->tb = tb; |
||||
} |
||||
|
||||
SyncQueue *sq_alloc(enum SyncQueueType type, int64_t buf_size_us) |
||||
{ |
||||
SyncQueue *sq = av_mallocz(sizeof(*sq)); |
||||
|
||||
if (!sq) |
||||
return NULL; |
||||
|
||||
sq->type = type; |
||||
sq->buf_size_us = buf_size_us; |
||||
|
||||
sq->head_stream = -1; |
||||
sq->head_finished_stream = -1; |
||||
|
||||
sq->pool = (type == SYNC_QUEUE_PACKETS) ? objpool_alloc_packets() : |
||||
objpool_alloc_frames(); |
||||
if (!sq->pool) { |
||||
av_freep(&sq); |
||||
return NULL; |
||||
} |
||||
|
||||
return sq; |
||||
} |
||||
|
||||
void sq_free(SyncQueue **psq) |
||||
{ |
||||
SyncQueue *sq = *psq; |
||||
|
||||
if (!sq) |
||||
return; |
||||
|
||||
for (unsigned int i = 0; i < sq->nb_streams; i++) { |
||||
SyncQueueFrame frame; |
||||
while (av_fifo_read(sq->streams[i].fifo, &frame, 1) >= 0) |
||||
objpool_release(sq->pool, (void**)&frame); |
||||
|
||||
av_fifo_freep2(&sq->streams[i].fifo); |
||||
} |
||||
|
||||
av_freep(&sq->streams); |
||||
|
||||
objpool_free(&sq->pool); |
||||
|
||||
av_freep(psq); |
||||
} |
||||
@ -0,0 +1,100 @@ |
||||
/*
|
||||
* 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_SYNC_QUEUE_H |
||||
#define FFTOOLS_SYNC_QUEUE_H |
||||
|
||||
#include <stdint.h> |
||||
|
||||
#include "libavcodec/packet.h" |
||||
|
||||
#include "libavutil/frame.h" |
||||
|
||||
enum SyncQueueType { |
||||
SYNC_QUEUE_PACKETS, |
||||
SYNC_QUEUE_FRAMES, |
||||
}; |
||||
|
||||
typedef union SyncQueueFrame { |
||||
AVFrame *f; |
||||
AVPacket *p; |
||||
} SyncQueueFrame; |
||||
|
||||
#define SQFRAME(frame) ((SyncQueueFrame){ .f = (frame) }) |
||||
#define SQPKT(pkt) ((SyncQueueFrame){ .p = (pkt) }) |
||||
|
||||
typedef struct SyncQueue SyncQueue; |
||||
|
||||
/**
|
||||
* Allocate a sync queue of the given type. |
||||
* |
||||
* @param buf_size_us maximum duration that will be buffered in microseconds |
||||
*/ |
||||
SyncQueue *sq_alloc(enum SyncQueueType type, int64_t buf_size_us); |
||||
void sq_free(SyncQueue **sq); |
||||
|
||||
/**
|
||||
* Add a new stream to the sync queue. |
||||
* |
||||
* @return |
||||
* - a non-negative stream index on success |
||||
* - a negative error code on error |
||||
*/ |
||||
int sq_add_stream(SyncQueue *sq); |
||||
|
||||
/**
|
||||
* Set the timebase for the stream with index stream_idx. Should be called |
||||
* before sending any frames for this stream. |
||||
*/ |
||||
void sq_set_tb(SyncQueue *sq, unsigned int stream_idx, AVRational tb); |
||||
|
||||
/**
|
||||
* Submit a frame for the stream with index stream_idx. |
||||
* |
||||
* On success, the sync queue takes ownership of the frame and will reset the |
||||
* contents of the supplied frame. On failure, the frame remains owned by the |
||||
* caller. |
||||
* |
||||
* Sending a frame with NULL contents marks the stream as finished. |
||||
* |
||||
* @return |
||||
* - 0 on success |
||||
* - AVERROR_EOF when no more frames should be submitted for this stream |
||||
* - another a negative error code on failure |
||||
*/ |
||||
int sq_send(SyncQueue *sq, unsigned int stream_idx, SyncQueueFrame frame); |
||||
|
||||
/**
|
||||
* Read a frame from the queue. |
||||
* |
||||
* @param stream_idx index of the stream to read a frame for. May be -1, then |
||||
* try to read a frame from any stream that is ready for |
||||
* output. |
||||
* @param frame output frame will be written here on success. The frame is owned |
||||
* by the caller. |
||||
* |
||||
* @return |
||||
* - a non-negative index of the stream to which the returned frame belongs |
||||
* - AVERROR(EAGAIN) when more frames need to be submitted to the queue |
||||
* - AVERROR_EOF when no more frames will be available for this stream (for any |
||||
* stream if stream_idx is -1) |
||||
* - another negative error code on failure |
||||
*/ |
||||
int sq_receive(SyncQueue *sq, int stream_idx, SyncQueueFrame frame); |
||||
|
||||
#endif // FFTOOLS_SYNC_QUEUE_H
|
||||
@ -1,4 +1,4 @@ |
||||
145b9b48d56f9c966bf41657f7569954 *tests/data/fate/shortest-sub.matroska |
||||
139232 tests/data/fate/shortest-sub.matroska |
||||
d71f5d359ef788ea689415bc1e4a90df *tests/data/fate/shortest-sub.out.framecrc |
||||
stddev:11541.12 PSNR: 15.08 MAXDIFF:22854 bytes: 2591/ 26055 |
||||
876ac3fa52e467050ab843969d4cf343 *tests/data/fate/shortest-sub.out.framecrc |
||||
stddev:11541.12 PSNR: 15.08 MAXDIFF:22854 bytes: 2591/ 23735 |
||||
|
||||
Loading…
Reference in new issue