parent
be28fa1010
commit
50b13ca2e1
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,165 @@ |
||||
/*
|
||||
* Muxer internal APIs - should not be included outside of ffmpeg_mux* |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of ffmpeg_mux.h file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
* - fftools header names updated |
||||
* - want_sdp made thread-local |
||||
* - EncStatsFile declaration migrated from ffmpeg_mux_init.c |
||||
* - WARN_MULTIPLE_OPT_USAGE, MATCH_PER_STREAM_OPT, MATCH_PER_TYPE_OPT, SPECIFIER_OPT_FMT declarations migrated from |
||||
* ffmpeg.h |
||||
* - ms_from_ost migrated to ffmpeg_mux.c |
||||
*/ |
||||
|
||||
#ifndef FFTOOLS_FFMPEG_MUX_H |
||||
#define FFTOOLS_FFMPEG_MUX_H |
||||
|
||||
#include <stdatomic.h> |
||||
#include <stdint.h> |
||||
|
||||
#include "fftools_thread_queue.h" |
||||
|
||||
#include "libavformat/avformat.h" |
||||
|
||||
#include "libavcodec/packet.h" |
||||
|
||||
#include "libavutil/dict.h" |
||||
#include "libavutil/fifo.h" |
||||
#include "libavutil/thread.h" |
||||
|
||||
#define SPECIFIER_OPT_FMT_str "%s" |
||||
#define SPECIFIER_OPT_FMT_i "%i" |
||||
#define SPECIFIER_OPT_FMT_i64 "%"PRId64 |
||||
#define SPECIFIER_OPT_FMT_ui64 "%"PRIu64 |
||||
#define SPECIFIER_OPT_FMT_f "%f" |
||||
#define SPECIFIER_OPT_FMT_dbl "%lf" |
||||
|
||||
#define WARN_MULTIPLE_OPT_USAGE(name, type, so, st)\ |
||||
{\
|
||||
char namestr[128] = "";\
|
||||
const char *spec = so->specifier && so->specifier[0] ? so->specifier : "";\
|
||||
for (int _i = 0; opt_name_##name[_i]; _i++)\
|
||||
av_strlcatf(namestr, sizeof(namestr), "-%s%s", opt_name_##name[_i], opt_name_##name[_i+1] ? (opt_name_##name[_i+2] ? ", " : " or ") : "");\
|
||||
av_log(NULL, AV_LOG_WARNING, "Multiple %s options specified for stream %d, only the last option '-%s%s%s "SPECIFIER_OPT_FMT_##type"' will be used.\n",\
|
||||
namestr, st->index, opt_name_##name[0], spec[0] ? ":" : "", spec, so->u.type);\
|
||||
} |
||||
|
||||
#define MATCH_PER_STREAM_OPT(name, type, outvar, fmtctx, st)\ |
||||
{\
|
||||
int _ret, _matches = 0;\
|
||||
SpecifierOpt *so;\
|
||||
for (int _i = 0; _i < o->nb_ ## name; _i++) {\
|
||||
char *spec = o->name[_i].specifier;\
|
||||
if ((_ret = check_stream_specifier(fmtctx, st, spec)) > 0) {\
|
||||
outvar = o->name[_i].u.type;\
|
||||
so = &o->name[_i];\
|
||||
_matches++;\
|
||||
} else if (_ret < 0)\
|
||||
exit_program(1);\
|
||||
}\
|
||||
if (_matches > 1)\
|
||||
WARN_MULTIPLE_OPT_USAGE(name, type, so, st);\
|
||||
} |
||||
|
||||
#define MATCH_PER_TYPE_OPT(name, type, outvar, fmtctx, mediatype)\ |
||||
{\
|
||||
int i;\
|
||||
for (i = 0; i < o->nb_ ## name; i++) {\
|
||||
char *spec = o->name[i].specifier;\
|
||||
if (!strcmp(spec, mediatype))\
|
||||
outvar = o->name[i].u.type;\
|
||||
}\
|
||||
} |
||||
|
||||
typedef struct MuxStream { |
||||
OutputStream ost; |
||||
|
||||
// name used for logging
|
||||
char log_name[32]; |
||||
|
||||
/* the packets are buffered here until the muxer is ready to be initialized */ |
||||
AVFifo *muxing_queue; |
||||
|
||||
AVBSFContext *bsf_ctx; |
||||
|
||||
EncStats stats; |
||||
|
||||
int64_t max_frames; |
||||
|
||||
/*
|
||||
* The size of the AVPackets' buffers in queue. |
||||
* Updated when a packet is either pushed or pulled from the queue. |
||||
*/ |
||||
size_t muxing_queue_data_size; |
||||
|
||||
int max_muxing_queue_size; |
||||
|
||||
/* Threshold after which max_muxing_queue_size will be in effect */ |
||||
size_t muxing_queue_data_threshold; |
||||
|
||||
/* dts of the last packet sent to the muxer, in the stream timebase
|
||||
* used for making up missing dts values */ |
||||
int64_t last_mux_dts; |
||||
} MuxStream; |
||||
|
||||
typedef struct Muxer { |
||||
OutputFile of; |
||||
|
||||
// name used for logging
|
||||
char log_name[32]; |
||||
|
||||
AVFormatContext *fc; |
||||
|
||||
pthread_t thread; |
||||
ThreadQueue *tq; |
||||
|
||||
AVDictionary *opts; |
||||
|
||||
int thread_queue_size; |
||||
|
||||
/* filesize limit expressed in bytes */ |
||||
int64_t limit_filesize; |
||||
atomic_int_least64_t last_filesize; |
||||
int header_written; |
||||
|
||||
SyncQueue *sq_mux; |
||||
AVPacket *sq_pkt; |
||||
} Muxer; |
||||
|
||||
typedef struct EncStatsFile { |
||||
char *path; |
||||
AVIOContext *io; |
||||
} EncStatsFile; |
||||
|
||||
/* whether we want to print an SDP, set in of_open() */ |
||||
extern __thread int want_sdp; |
||||
|
||||
int mux_check_init(Muxer *mux); |
||||
|
||||
#endif /* FFTOOLS_FFMPEG_MUX_H */ |
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,145 @@ |
||||
/*
|
||||
* This file is part of FFmpeg. |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of objpool.c file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
* - fftools header names updated |
||||
*/ |
||||
|
||||
#include <stdint.h> |
||||
|
||||
#include "libavcodec/packet.h" |
||||
|
||||
#include "libavutil/common.h" |
||||
#include "libavutil/error.h" |
||||
#include "libavutil/frame.h" |
||||
#include "libavutil/mem.h" |
||||
|
||||
#include "fftools_objpool.h" |
||||
|
||||
struct ObjPool { |
||||
void *pool[32]; |
||||
unsigned int pool_count; |
||||
|
||||
ObjPoolCBAlloc alloc; |
||||
ObjPoolCBReset reset; |
||||
ObjPoolCBFree free; |
||||
}; |
||||
|
||||
ObjPool *objpool_alloc(ObjPoolCBAlloc cb_alloc, ObjPoolCBReset cb_reset, |
||||
ObjPoolCBFree cb_free) |
||||
{ |
||||
ObjPool *op = av_mallocz(sizeof(*op)); |
||||
|
||||
if (!op) |
||||
return NULL; |
||||
|
||||
op->alloc = cb_alloc; |
||||
op->reset = cb_reset; |
||||
op->free = cb_free; |
||||
|
||||
return op; |
||||
} |
||||
|
||||
void objpool_free(ObjPool **pop) |
||||
{ |
||||
ObjPool *op = *pop; |
||||
|
||||
if (!op) |
||||
return; |
||||
|
||||
for (unsigned int i = 0; i < op->pool_count; i++) |
||||
op->free(&op->pool[i]); |
||||
|
||||
av_freep(pop); |
||||
} |
||||
|
||||
int objpool_get(ObjPool *op, void **obj) |
||||
{ |
||||
if (op->pool_count) { |
||||
*obj = op->pool[--op->pool_count]; |
||||
op->pool[op->pool_count] = NULL; |
||||
} else |
||||
*obj = op->alloc(); |
||||
|
||||
return *obj ? 0 : AVERROR(ENOMEM); |
||||
} |
||||
|
||||
void objpool_release(ObjPool *op, void **obj) |
||||
{ |
||||
if (!*obj) |
||||
return; |
||||
|
||||
op->reset(*obj); |
||||
|
||||
if (op->pool_count < FF_ARRAY_ELEMS(op->pool)) |
||||
op->pool[op->pool_count++] = *obj; |
||||
else |
||||
op->free(obj); |
||||
|
||||
*obj = NULL; |
||||
} |
||||
|
||||
static void *alloc_packet(void) |
||||
{ |
||||
return av_packet_alloc(); |
||||
} |
||||
static void *alloc_frame(void) |
||||
{ |
||||
return av_frame_alloc(); |
||||
} |
||||
|
||||
static void reset_packet(void *obj) |
||||
{ |
||||
av_packet_unref(obj); |
||||
} |
||||
static void reset_frame(void *obj) |
||||
{ |
||||
av_frame_unref(obj); |
||||
} |
||||
|
||||
static void free_packet(void **obj) |
||||
{ |
||||
AVPacket *pkt = *obj; |
||||
av_packet_free(&pkt); |
||||
*obj = NULL; |
||||
} |
||||
static void free_frame(void **obj) |
||||
{ |
||||
AVFrame *frame = *obj; |
||||
av_frame_free(&frame); |
||||
*obj = NULL; |
||||
} |
||||
|
||||
ObjPool *objpool_alloc_packets(void) |
||||
{ |
||||
return objpool_alloc(alloc_packet, reset_packet, free_packet); |
||||
} |
||||
ObjPool *objpool_alloc_frames(void) |
||||
{ |
||||
return objpool_alloc(alloc_frame, reset_frame, free_frame); |
||||
} |
||||
@ -0,0 +1,50 @@ |
||||
/*
|
||||
* This file is part of FFmpeg. |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of objpool.h file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
*/ |
||||
|
||||
#ifndef FFTOOLS_OBJPOOL_H |
||||
#define FFTOOLS_OBJPOOL_H |
||||
|
||||
typedef struct ObjPool ObjPool; |
||||
|
||||
typedef void* (*ObjPoolCBAlloc)(void); |
||||
typedef void (*ObjPoolCBReset)(void *); |
||||
typedef void (*ObjPoolCBFree)(void **); |
||||
|
||||
void objpool_free(ObjPool **op); |
||||
ObjPool *objpool_alloc(ObjPoolCBAlloc cb_alloc, ObjPoolCBReset cb_reset, |
||||
ObjPoolCBFree cb_free); |
||||
ObjPool *objpool_alloc_packets(void); |
||||
ObjPool *objpool_alloc_frames(void); |
||||
|
||||
int objpool_get(ObjPool *op, void **obj); |
||||
void objpool_release(ObjPool *op, void **obj); |
||||
|
||||
#endif // FFTOOLS_OBJPOOL_H
|
||||
@ -0,0 +1,462 @@ |
||||
/*
|
||||
* This file is part of FFmpeg. |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of sync_queue.c file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
* - fftools header names updated |
||||
*/ |
||||
|
||||
#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 "fftools_objpool.h" |
||||
#include "fftools_sync_queue.h" |
||||
|
||||
typedef struct SyncQueueStream { |
||||
AVFifo *fifo; |
||||
AVRational tb; |
||||
|
||||
/* stream head: largest timestamp seen */ |
||||
int64_t head_ts; |
||||
int limiting; |
||||
/* no more frames will be sent for this stream */ |
||||
int finished; |
||||
|
||||
uint64_t frames_sent; |
||||
uint64_t frames_max; |
||||
} 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->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->limiting && 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->limiting && 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->limiting && 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 (st->limiting && |
||||
(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); |
||||
|
||||
st->frames_sent++; |
||||
if (st->frames_sent >= st->frames_max) |
||||
finish_stream(sq, stream_idx); |
||||
|
||||
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, int limiting) |
||||
{ |
||||
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; |
||||
st->frames_max = UINT64_MAX; |
||||
st->limiting = limiting; |
||||
|
||||
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; |
||||
} |
||||
|
||||
void sq_limit_frames(SyncQueue *sq, unsigned int stream_idx, uint64_t frames) |
||||
{ |
||||
SyncQueueStream *st; |
||||
|
||||
av_assert0(stream_idx < sq->nb_streams); |
||||
st = &sq->streams[stream_idx]; |
||||
|
||||
st->frames_max = frames; |
||||
if (st->frames_sent >= st->frames_max) |
||||
finish_stream(sq, stream_idx); |
||||
} |
||||
|
||||
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,122 @@ |
||||
/*
|
||||
* This file is part of FFmpeg. |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of sync_queue.h file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
*/ |
||||
|
||||
#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. |
||||
* |
||||
* @param limiting whether the stream is limiting, i.e. no other stream can be |
||||
* longer than this one |
||||
* @return |
||||
* - a non-negative stream index on success |
||||
* - a negative error code on error |
||||
*/ |
||||
int sq_add_stream(SyncQueue *sq, int limiting); |
||||
|
||||
/**
|
||||
* 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); |
||||
|
||||
/**
|
||||
* Limit the number of output frames for stream with index stream_idx |
||||
* to max_frames. |
||||
*/ |
||||
void sq_limit_frames(SyncQueue *sq, unsigned int stream_idx, |
||||
uint64_t max_frames); |
||||
|
||||
/**
|
||||
* 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
|
||||
@ -0,0 +1,259 @@ |
||||
/*
|
||||
* This file is part of FFmpeg. |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of thread_queue.c file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
* - fftools header names updated |
||||
*/ |
||||
|
||||
#include <stdint.h> |
||||
#include <string.h> |
||||
|
||||
#include "libavutil/avassert.h" |
||||
#include "libavutil/error.h" |
||||
#include "libavutil/fifo.h" |
||||
#include "libavutil/intreadwrite.h" |
||||
#include "libavutil/mem.h" |
||||
#include "libavutil/thread.h" |
||||
|
||||
#include "fftools_objpool.h" |
||||
#include "fftools_thread_queue.h" |
||||
|
||||
enum { |
||||
FINISHED_SEND = (1 << 0), |
||||
FINISHED_RECV = (1 << 1), |
||||
}; |
||||
|
||||
typedef struct FifoElem { |
||||
void *obj; |
||||
unsigned int stream_idx; |
||||
} FifoElem; |
||||
|
||||
struct ThreadQueue { |
||||
int *finished; |
||||
unsigned int nb_streams; |
||||
|
||||
AVFifo *fifo; |
||||
|
||||
ObjPool *obj_pool; |
||||
void (*obj_move)(void *dst, void *src); |
||||
|
||||
pthread_mutex_t lock; |
||||
pthread_cond_t cond; |
||||
}; |
||||
|
||||
void tq_free(ThreadQueue **ptq) |
||||
{ |
||||
ThreadQueue *tq = *ptq; |
||||
|
||||
if (!tq) |
||||
return; |
||||
|
||||
if (tq->fifo) { |
||||
FifoElem elem; |
||||
while (av_fifo_read(tq->fifo, &elem, 1) >= 0) |
||||
objpool_release(tq->obj_pool, &elem.obj); |
||||
} |
||||
av_fifo_freep2(&tq->fifo); |
||||
|
||||
objpool_free(&tq->obj_pool); |
||||
|
||||
av_freep(&tq->finished); |
||||
|
||||
pthread_cond_destroy(&tq->cond); |
||||
pthread_mutex_destroy(&tq->lock); |
||||
|
||||
av_freep(ptq); |
||||
} |
||||
|
||||
ThreadQueue *tq_alloc(unsigned int nb_streams, size_t queue_size, |
||||
ObjPool *obj_pool, void (*obj_move)(void *dst, void *src)) |
||||
{ |
||||
ThreadQueue *tq; |
||||
int ret; |
||||
|
||||
tq = av_mallocz(sizeof(*tq)); |
||||
if (!tq) |
||||
return NULL; |
||||
|
||||
ret = pthread_cond_init(&tq->cond, NULL); |
||||
if (ret) { |
||||
av_freep(&tq); |
||||
return NULL; |
||||
} |
||||
|
||||
ret = pthread_mutex_init(&tq->lock, NULL); |
||||
if (ret) { |
||||
pthread_cond_destroy(&tq->cond); |
||||
av_freep(&tq); |
||||
return NULL; |
||||
} |
||||
|
||||
tq->finished = av_calloc(nb_streams, sizeof(*tq->finished)); |
||||
if (!tq->finished) |
||||
goto fail; |
||||
tq->nb_streams = nb_streams; |
||||
|
||||
tq->fifo = av_fifo_alloc2(queue_size, sizeof(FifoElem), 0); |
||||
if (!tq->fifo) |
||||
goto fail; |
||||
|
||||
tq->obj_pool = obj_pool; |
||||
tq->obj_move = obj_move; |
||||
|
||||
return tq; |
||||
fail: |
||||
tq_free(&tq); |
||||
return NULL; |
||||
} |
||||
|
||||
int tq_send(ThreadQueue *tq, unsigned int stream_idx, void *data) |
||||
{ |
||||
int *finished; |
||||
int ret; |
||||
|
||||
av_assert0(stream_idx < tq->nb_streams); |
||||
finished = &tq->finished[stream_idx]; |
||||
|
||||
pthread_mutex_lock(&tq->lock); |
||||
|
||||
if (*finished & FINISHED_SEND) { |
||||
ret = AVERROR(EINVAL); |
||||
goto finish; |
||||
} |
||||
|
||||
while (!(*finished & FINISHED_RECV) && !av_fifo_can_write(tq->fifo)) |
||||
pthread_cond_wait(&tq->cond, &tq->lock); |
||||
|
||||
if (*finished & FINISHED_RECV) { |
||||
ret = AVERROR_EOF; |
||||
*finished |= FINISHED_SEND; |
||||
} else { |
||||
FifoElem elem = { .stream_idx = stream_idx }; |
||||
|
||||
ret = objpool_get(tq->obj_pool, &elem.obj); |
||||
if (ret < 0) |
||||
goto finish; |
||||
|
||||
tq->obj_move(elem.obj, data); |
||||
|
||||
ret = av_fifo_write(tq->fifo, &elem, 1); |
||||
av_assert0(ret >= 0); |
||||
pthread_cond_broadcast(&tq->cond); |
||||
} |
||||
|
||||
finish: |
||||
pthread_mutex_unlock(&tq->lock); |
||||
|
||||
return ret; |
||||
} |
||||
|
||||
static int receive_locked(ThreadQueue *tq, int *stream_idx, |
||||
void *data) |
||||
{ |
||||
FifoElem elem; |
||||
unsigned int nb_finished = 0; |
||||
|
||||
if (av_fifo_read(tq->fifo, &elem, 1) >= 0) { |
||||
tq->obj_move(data, elem.obj); |
||||
objpool_release(tq->obj_pool, &elem.obj); |
||||
*stream_idx = elem.stream_idx; |
||||
return 0; |
||||
} |
||||
|
||||
for (unsigned int i = 0; i < tq->nb_streams; i++) { |
||||
if (!(tq->finished[i] & FINISHED_SEND)) |
||||
continue; |
||||
|
||||
/* return EOF to the consumer at most once for each stream */ |
||||
if (!(tq->finished[i] & FINISHED_RECV)) { |
||||
tq->finished[i] |= FINISHED_RECV; |
||||
*stream_idx = i; |
||||
return AVERROR_EOF; |
||||
} |
||||
|
||||
nb_finished++; |
||||
} |
||||
|
||||
return nb_finished == tq->nb_streams ? AVERROR_EOF : AVERROR(EAGAIN); |
||||
} |
||||
|
||||
int tq_receive(ThreadQueue *tq, int *stream_idx, void *data) |
||||
{ |
||||
int ret; |
||||
|
||||
*stream_idx = -1; |
||||
|
||||
pthread_mutex_lock(&tq->lock); |
||||
|
||||
while (1) { |
||||
ret = receive_locked(tq, stream_idx, data); |
||||
if (ret == AVERROR(EAGAIN)) { |
||||
pthread_cond_wait(&tq->cond, &tq->lock); |
||||
continue; |
||||
} |
||||
|
||||
break; |
||||
} |
||||
|
||||
if (ret == 0) |
||||
pthread_cond_broadcast(&tq->cond); |
||||
|
||||
pthread_mutex_unlock(&tq->lock); |
||||
|
||||
return ret; |
||||
} |
||||
|
||||
void tq_send_finish(ThreadQueue *tq, unsigned int stream_idx) |
||||
{ |
||||
av_assert0(stream_idx < tq->nb_streams); |
||||
|
||||
pthread_mutex_lock(&tq->lock); |
||||
|
||||
/* mark the stream as send-finished;
|
||||
* next time the consumer thread tries to read this stream it will get |
||||
* an EOF and recv-finished flag will be set */ |
||||
tq->finished[stream_idx] |= FINISHED_SEND; |
||||
pthread_cond_broadcast(&tq->cond); |
||||
|
||||
pthread_mutex_unlock(&tq->lock); |
||||
} |
||||
|
||||
void tq_receive_finish(ThreadQueue *tq, unsigned int stream_idx) |
||||
{ |
||||
av_assert0(stream_idx < tq->nb_streams); |
||||
|
||||
pthread_mutex_lock(&tq->lock); |
||||
|
||||
/* mark the stream as recv-finished;
|
||||
* next time the producer thread tries to send for this stream, it will |
||||
* get an EOF and send-finished flag will be set */ |
||||
tq->finished[stream_idx] |= FINISHED_RECV; |
||||
pthread_cond_broadcast(&tq->cond); |
||||
|
||||
pthread_mutex_unlock(&tq->lock); |
||||
} |
||||
@ -0,0 +1,94 @@ |
||||
/*
|
||||
* This file is part of FFmpeg. |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of thread_queue.h file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
*/ |
||||
|
||||
#ifndef FFTOOLS_THREAD_QUEUE_H |
||||
#define FFTOOLS_THREAD_QUEUE_H |
||||
|
||||
#include <string.h> |
||||
|
||||
#include "fftools_objpool.h" |
||||
|
||||
typedef struct ThreadQueue ThreadQueue; |
||||
|
||||
/**
|
||||
* Allocate a queue for sending data between threads. |
||||
* |
||||
* @param nb_streams number of streams for which a distinct EOF state is |
||||
* maintained |
||||
* @param queue_size number of items that can be stored in the queue without |
||||
* blocking |
||||
* @param obj_pool object pool that will be used to allocate items stored in the |
||||
* queue; the pool becomes owned by the queue |
||||
* @param callback that moves the contents between two data pointers |
||||
*/ |
||||
ThreadQueue *tq_alloc(unsigned int nb_streams, size_t queue_size, |
||||
ObjPool *obj_pool, void (*obj_move)(void *dst, void *src)); |
||||
void tq_free(ThreadQueue **tq); |
||||
|
||||
/**
|
||||
* Send an item for the given stream to the queue. |
||||
* |
||||
* @param data the item to send, its contents will be moved using the callback |
||||
* provided to tq_alloc(); on failure the item will be left |
||||
* untouched |
||||
* @return |
||||
* - 0 the item was successfully sent |
||||
* - AVERROR(ENOMEM) could not allocate an item for writing to the FIFO |
||||
* - AVERROR(EINVAL) the sending side has previously been marked as finished |
||||
* - AVERROR_EOF the receiving side has marked the given stream as finished |
||||
*/ |
||||
int tq_send(ThreadQueue *tq, unsigned int stream_idx, void *data); |
||||
/**
|
||||
* Mark the given stream finished from the sending side. |
||||
*/ |
||||
void tq_send_finish(ThreadQueue *tq, unsigned int stream_idx); |
||||
|
||||
/**
|
||||
* Read the next item from the queue. |
||||
* |
||||
* @param stream_idx the index of the stream that was processed or -1 will be |
||||
* written here |
||||
* @param data the data item will be written here on success using the |
||||
* callback provided to tq_alloc() |
||||
* @return |
||||
* - 0 a data item was successfully read; *stream_idx contains a non-negative |
||||
* stream index |
||||
* - AVERROR_EOF When *stream_idx is non-negative, this signals that the sending |
||||
* side has marked the given stream as finished. This will happen at most once |
||||
* for each stream. When *stream_idx is -1, all streams are done. |
||||
*/ |
||||
int tq_receive(ThreadQueue *tq, int *stream_idx, void *data); |
||||
/**
|
||||
* Mark the given stream finished from the receiving side. |
||||
*/ |
||||
void tq_receive_finish(ThreadQueue *tq, unsigned int stream_idx); |
||||
|
||||
#endif // FFTOOLS_THREAD_QUEUE_H
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,165 @@ |
||||
/*
|
||||
* Muxer internal APIs - should not be included outside of ffmpeg_mux* |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of ffmpeg_mux.h file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
* - fftools header names updated |
||||
* - want_sdp made thread-local |
||||
* - EncStatsFile declaration migrated from ffmpeg_mux_init.c |
||||
* - WARN_MULTIPLE_OPT_USAGE, MATCH_PER_STREAM_OPT, MATCH_PER_TYPE_OPT, SPECIFIER_OPT_FMT declarations migrated from |
||||
* ffmpeg.h |
||||
* - ms_from_ost migrated to ffmpeg_mux.c |
||||
*/ |
||||
|
||||
#ifndef FFTOOLS_FFMPEG_MUX_H |
||||
#define FFTOOLS_FFMPEG_MUX_H |
||||
|
||||
#include <stdatomic.h> |
||||
#include <stdint.h> |
||||
|
||||
#include "fftools_thread_queue.h" |
||||
|
||||
#include "libavformat/avformat.h" |
||||
|
||||
#include "libavcodec/packet.h" |
||||
|
||||
#include "libavutil/dict.h" |
||||
#include "libavutil/fifo.h" |
||||
#include "libavutil/thread.h" |
||||
|
||||
#define SPECIFIER_OPT_FMT_str "%s" |
||||
#define SPECIFIER_OPT_FMT_i "%i" |
||||
#define SPECIFIER_OPT_FMT_i64 "%"PRId64 |
||||
#define SPECIFIER_OPT_FMT_ui64 "%"PRIu64 |
||||
#define SPECIFIER_OPT_FMT_f "%f" |
||||
#define SPECIFIER_OPT_FMT_dbl "%lf" |
||||
|
||||
#define WARN_MULTIPLE_OPT_USAGE(name, type, so, st)\ |
||||
{\
|
||||
char namestr[128] = "";\
|
||||
const char *spec = so->specifier && so->specifier[0] ? so->specifier : "";\
|
||||
for (int _i = 0; opt_name_##name[_i]; _i++)\
|
||||
av_strlcatf(namestr, sizeof(namestr), "-%s%s", opt_name_##name[_i], opt_name_##name[_i+1] ? (opt_name_##name[_i+2] ? ", " : " or ") : "");\
|
||||
av_log(NULL, AV_LOG_WARNING, "Multiple %s options specified for stream %d, only the last option '-%s%s%s "SPECIFIER_OPT_FMT_##type"' will be used.\n",\
|
||||
namestr, st->index, opt_name_##name[0], spec[0] ? ":" : "", spec, so->u.type);\
|
||||
} |
||||
|
||||
#define MATCH_PER_STREAM_OPT(name, type, outvar, fmtctx, st)\ |
||||
{\
|
||||
int _ret, _matches = 0;\
|
||||
SpecifierOpt *so;\
|
||||
for (int _i = 0; _i < o->nb_ ## name; _i++) {\
|
||||
char *spec = o->name[_i].specifier;\
|
||||
if ((_ret = check_stream_specifier(fmtctx, st, spec)) > 0) {\
|
||||
outvar = o->name[_i].u.type;\
|
||||
so = &o->name[_i];\
|
||||
_matches++;\
|
||||
} else if (_ret < 0)\
|
||||
exit_program(1);\
|
||||
}\
|
||||
if (_matches > 1)\
|
||||
WARN_MULTIPLE_OPT_USAGE(name, type, so, st);\
|
||||
} |
||||
|
||||
#define MATCH_PER_TYPE_OPT(name, type, outvar, fmtctx, mediatype)\ |
||||
{\
|
||||
int i;\
|
||||
for (i = 0; i < o->nb_ ## name; i++) {\
|
||||
char *spec = o->name[i].specifier;\
|
||||
if (!strcmp(spec, mediatype))\
|
||||
outvar = o->name[i].u.type;\
|
||||
}\
|
||||
} |
||||
|
||||
typedef struct MuxStream { |
||||
OutputStream ost; |
||||
|
||||
// name used for logging
|
||||
char log_name[32]; |
||||
|
||||
/* the packets are buffered here until the muxer is ready to be initialized */ |
||||
AVFifo *muxing_queue; |
||||
|
||||
AVBSFContext *bsf_ctx; |
||||
|
||||
EncStats stats; |
||||
|
||||
int64_t max_frames; |
||||
|
||||
/*
|
||||
* The size of the AVPackets' buffers in queue. |
||||
* Updated when a packet is either pushed or pulled from the queue. |
||||
*/ |
||||
size_t muxing_queue_data_size; |
||||
|
||||
int max_muxing_queue_size; |
||||
|
||||
/* Threshold after which max_muxing_queue_size will be in effect */ |
||||
size_t muxing_queue_data_threshold; |
||||
|
||||
/* dts of the last packet sent to the muxer, in the stream timebase
|
||||
* used for making up missing dts values */ |
||||
int64_t last_mux_dts; |
||||
} MuxStream; |
||||
|
||||
typedef struct Muxer { |
||||
OutputFile of; |
||||
|
||||
// name used for logging
|
||||
char log_name[32]; |
||||
|
||||
AVFormatContext *fc; |
||||
|
||||
pthread_t thread; |
||||
ThreadQueue *tq; |
||||
|
||||
AVDictionary *opts; |
||||
|
||||
int thread_queue_size; |
||||
|
||||
/* filesize limit expressed in bytes */ |
||||
int64_t limit_filesize; |
||||
atomic_int_least64_t last_filesize; |
||||
int header_written; |
||||
|
||||
SyncQueue *sq_mux; |
||||
AVPacket *sq_pkt; |
||||
} Muxer; |
||||
|
||||
typedef struct EncStatsFile { |
||||
char *path; |
||||
AVIOContext *io; |
||||
} EncStatsFile; |
||||
|
||||
/* whether we want to print an SDP, set in of_open() */ |
||||
extern __thread int want_sdp; |
||||
|
||||
int mux_check_init(Muxer *mux); |
||||
|
||||
#endif /* FFTOOLS_FFMPEG_MUX_H */ |
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,145 @@ |
||||
/*
|
||||
* This file is part of FFmpeg. |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of objpool.c file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
* - fftools header names updated |
||||
*/ |
||||
|
||||
#include <stdint.h> |
||||
|
||||
#include "libavcodec/packet.h" |
||||
|
||||
#include "libavutil/common.h" |
||||
#include "libavutil/error.h" |
||||
#include "libavutil/frame.h" |
||||
#include "libavutil/mem.h" |
||||
|
||||
#include "fftools_objpool.h" |
||||
|
||||
struct ObjPool { |
||||
void *pool[32]; |
||||
unsigned int pool_count; |
||||
|
||||
ObjPoolCBAlloc alloc; |
||||
ObjPoolCBReset reset; |
||||
ObjPoolCBFree free; |
||||
}; |
||||
|
||||
ObjPool *objpool_alloc(ObjPoolCBAlloc cb_alloc, ObjPoolCBReset cb_reset, |
||||
ObjPoolCBFree cb_free) |
||||
{ |
||||
ObjPool *op = av_mallocz(sizeof(*op)); |
||||
|
||||
if (!op) |
||||
return NULL; |
||||
|
||||
op->alloc = cb_alloc; |
||||
op->reset = cb_reset; |
||||
op->free = cb_free; |
||||
|
||||
return op; |
||||
} |
||||
|
||||
void objpool_free(ObjPool **pop) |
||||
{ |
||||
ObjPool *op = *pop; |
||||
|
||||
if (!op) |
||||
return; |
||||
|
||||
for (unsigned int i = 0; i < op->pool_count; i++) |
||||
op->free(&op->pool[i]); |
||||
|
||||
av_freep(pop); |
||||
} |
||||
|
||||
int objpool_get(ObjPool *op, void **obj) |
||||
{ |
||||
if (op->pool_count) { |
||||
*obj = op->pool[--op->pool_count]; |
||||
op->pool[op->pool_count] = NULL; |
||||
} else |
||||
*obj = op->alloc(); |
||||
|
||||
return *obj ? 0 : AVERROR(ENOMEM); |
||||
} |
||||
|
||||
void objpool_release(ObjPool *op, void **obj) |
||||
{ |
||||
if (!*obj) |
||||
return; |
||||
|
||||
op->reset(*obj); |
||||
|
||||
if (op->pool_count < FF_ARRAY_ELEMS(op->pool)) |
||||
op->pool[op->pool_count++] = *obj; |
||||
else |
||||
op->free(obj); |
||||
|
||||
*obj = NULL; |
||||
} |
||||
|
||||
static void *alloc_packet(void) |
||||
{ |
||||
return av_packet_alloc(); |
||||
} |
||||
static void *alloc_frame(void) |
||||
{ |
||||
return av_frame_alloc(); |
||||
} |
||||
|
||||
static void reset_packet(void *obj) |
||||
{ |
||||
av_packet_unref(obj); |
||||
} |
||||
static void reset_frame(void *obj) |
||||
{ |
||||
av_frame_unref(obj); |
||||
} |
||||
|
||||
static void free_packet(void **obj) |
||||
{ |
||||
AVPacket *pkt = *obj; |
||||
av_packet_free(&pkt); |
||||
*obj = NULL; |
||||
} |
||||
static void free_frame(void **obj) |
||||
{ |
||||
AVFrame *frame = *obj; |
||||
av_frame_free(&frame); |
||||
*obj = NULL; |
||||
} |
||||
|
||||
ObjPool *objpool_alloc_packets(void) |
||||
{ |
||||
return objpool_alloc(alloc_packet, reset_packet, free_packet); |
||||
} |
||||
ObjPool *objpool_alloc_frames(void) |
||||
{ |
||||
return objpool_alloc(alloc_frame, reset_frame, free_frame); |
||||
} |
||||
@ -0,0 +1,50 @@ |
||||
/*
|
||||
* This file is part of FFmpeg. |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of objpool.h file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
*/ |
||||
|
||||
#ifndef FFTOOLS_OBJPOOL_H |
||||
#define FFTOOLS_OBJPOOL_H |
||||
|
||||
typedef struct ObjPool ObjPool; |
||||
|
||||
typedef void* (*ObjPoolCBAlloc)(void); |
||||
typedef void (*ObjPoolCBReset)(void *); |
||||
typedef void (*ObjPoolCBFree)(void **); |
||||
|
||||
void objpool_free(ObjPool **op); |
||||
ObjPool *objpool_alloc(ObjPoolCBAlloc cb_alloc, ObjPoolCBReset cb_reset, |
||||
ObjPoolCBFree cb_free); |
||||
ObjPool *objpool_alloc_packets(void); |
||||
ObjPool *objpool_alloc_frames(void); |
||||
|
||||
int objpool_get(ObjPool *op, void **obj); |
||||
void objpool_release(ObjPool *op, void **obj); |
||||
|
||||
#endif // FFTOOLS_OBJPOOL_H
|
||||
@ -0,0 +1,462 @@ |
||||
/*
|
||||
* This file is part of FFmpeg. |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of sync_queue.c file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
* - fftools header names updated |
||||
*/ |
||||
|
||||
#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 "fftools_objpool.h" |
||||
#include "fftools_sync_queue.h" |
||||
|
||||
typedef struct SyncQueueStream { |
||||
AVFifo *fifo; |
||||
AVRational tb; |
||||
|
||||
/* stream head: largest timestamp seen */ |
||||
int64_t head_ts; |
||||
int limiting; |
||||
/* no more frames will be sent for this stream */ |
||||
int finished; |
||||
|
||||
uint64_t frames_sent; |
||||
uint64_t frames_max; |
||||
} 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->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->limiting && 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->limiting && 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->limiting && 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 (st->limiting && |
||||
(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); |
||||
|
||||
st->frames_sent++; |
||||
if (st->frames_sent >= st->frames_max) |
||||
finish_stream(sq, stream_idx); |
||||
|
||||
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, int limiting) |
||||
{ |
||||
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; |
||||
st->frames_max = UINT64_MAX; |
||||
st->limiting = limiting; |
||||
|
||||
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; |
||||
} |
||||
|
||||
void sq_limit_frames(SyncQueue *sq, unsigned int stream_idx, uint64_t frames) |
||||
{ |
||||
SyncQueueStream *st; |
||||
|
||||
av_assert0(stream_idx < sq->nb_streams); |
||||
st = &sq->streams[stream_idx]; |
||||
|
||||
st->frames_max = frames; |
||||
if (st->frames_sent >= st->frames_max) |
||||
finish_stream(sq, stream_idx); |
||||
} |
||||
|
||||
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,122 @@ |
||||
/*
|
||||
* This file is part of FFmpeg. |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of sync_queue.h file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
*/ |
||||
|
||||
#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. |
||||
* |
||||
* @param limiting whether the stream is limiting, i.e. no other stream can be |
||||
* longer than this one |
||||
* @return |
||||
* - a non-negative stream index on success |
||||
* - a negative error code on error |
||||
*/ |
||||
int sq_add_stream(SyncQueue *sq, int limiting); |
||||
|
||||
/**
|
||||
* 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); |
||||
|
||||
/**
|
||||
* Limit the number of output frames for stream with index stream_idx |
||||
* to max_frames. |
||||
*/ |
||||
void sq_limit_frames(SyncQueue *sq, unsigned int stream_idx, |
||||
uint64_t max_frames); |
||||
|
||||
/**
|
||||
* 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
|
||||
@ -0,0 +1,259 @@ |
||||
/*
|
||||
* This file is part of FFmpeg. |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of thread_queue.c file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
* - fftools header names updated |
||||
*/ |
||||
|
||||
#include <stdint.h> |
||||
#include <string.h> |
||||
|
||||
#include "libavutil/avassert.h" |
||||
#include "libavutil/error.h" |
||||
#include "libavutil/fifo.h" |
||||
#include "libavutil/intreadwrite.h" |
||||
#include "libavutil/mem.h" |
||||
#include "libavutil/thread.h" |
||||
|
||||
#include "fftools_objpool.h" |
||||
#include "fftools_thread_queue.h" |
||||
|
||||
enum { |
||||
FINISHED_SEND = (1 << 0), |
||||
FINISHED_RECV = (1 << 1), |
||||
}; |
||||
|
||||
typedef struct FifoElem { |
||||
void *obj; |
||||
unsigned int stream_idx; |
||||
} FifoElem; |
||||
|
||||
struct ThreadQueue { |
||||
int *finished; |
||||
unsigned int nb_streams; |
||||
|
||||
AVFifo *fifo; |
||||
|
||||
ObjPool *obj_pool; |
||||
void (*obj_move)(void *dst, void *src); |
||||
|
||||
pthread_mutex_t lock; |
||||
pthread_cond_t cond; |
||||
}; |
||||
|
||||
void tq_free(ThreadQueue **ptq) |
||||
{ |
||||
ThreadQueue *tq = *ptq; |
||||
|
||||
if (!tq) |
||||
return; |
||||
|
||||
if (tq->fifo) { |
||||
FifoElem elem; |
||||
while (av_fifo_read(tq->fifo, &elem, 1) >= 0) |
||||
objpool_release(tq->obj_pool, &elem.obj); |
||||
} |
||||
av_fifo_freep2(&tq->fifo); |
||||
|
||||
objpool_free(&tq->obj_pool); |
||||
|
||||
av_freep(&tq->finished); |
||||
|
||||
pthread_cond_destroy(&tq->cond); |
||||
pthread_mutex_destroy(&tq->lock); |
||||
|
||||
av_freep(ptq); |
||||
} |
||||
|
||||
ThreadQueue *tq_alloc(unsigned int nb_streams, size_t queue_size, |
||||
ObjPool *obj_pool, void (*obj_move)(void *dst, void *src)) |
||||
{ |
||||
ThreadQueue *tq; |
||||
int ret; |
||||
|
||||
tq = av_mallocz(sizeof(*tq)); |
||||
if (!tq) |
||||
return NULL; |
||||
|
||||
ret = pthread_cond_init(&tq->cond, NULL); |
||||
if (ret) { |
||||
av_freep(&tq); |
||||
return NULL; |
||||
} |
||||
|
||||
ret = pthread_mutex_init(&tq->lock, NULL); |
||||
if (ret) { |
||||
pthread_cond_destroy(&tq->cond); |
||||
av_freep(&tq); |
||||
return NULL; |
||||
} |
||||
|
||||
tq->finished = av_calloc(nb_streams, sizeof(*tq->finished)); |
||||
if (!tq->finished) |
||||
goto fail; |
||||
tq->nb_streams = nb_streams; |
||||
|
||||
tq->fifo = av_fifo_alloc2(queue_size, sizeof(FifoElem), 0); |
||||
if (!tq->fifo) |
||||
goto fail; |
||||
|
||||
tq->obj_pool = obj_pool; |
||||
tq->obj_move = obj_move; |
||||
|
||||
return tq; |
||||
fail: |
||||
tq_free(&tq); |
||||
return NULL; |
||||
} |
||||
|
||||
int tq_send(ThreadQueue *tq, unsigned int stream_idx, void *data) |
||||
{ |
||||
int *finished; |
||||
int ret; |
||||
|
||||
av_assert0(stream_idx < tq->nb_streams); |
||||
finished = &tq->finished[stream_idx]; |
||||
|
||||
pthread_mutex_lock(&tq->lock); |
||||
|
||||
if (*finished & FINISHED_SEND) { |
||||
ret = AVERROR(EINVAL); |
||||
goto finish; |
||||
} |
||||
|
||||
while (!(*finished & FINISHED_RECV) && !av_fifo_can_write(tq->fifo)) |
||||
pthread_cond_wait(&tq->cond, &tq->lock); |
||||
|
||||
if (*finished & FINISHED_RECV) { |
||||
ret = AVERROR_EOF; |
||||
*finished |= FINISHED_SEND; |
||||
} else { |
||||
FifoElem elem = { .stream_idx = stream_idx }; |
||||
|
||||
ret = objpool_get(tq->obj_pool, &elem.obj); |
||||
if (ret < 0) |
||||
goto finish; |
||||
|
||||
tq->obj_move(elem.obj, data); |
||||
|
||||
ret = av_fifo_write(tq->fifo, &elem, 1); |
||||
av_assert0(ret >= 0); |
||||
pthread_cond_broadcast(&tq->cond); |
||||
} |
||||
|
||||
finish: |
||||
pthread_mutex_unlock(&tq->lock); |
||||
|
||||
return ret; |
||||
} |
||||
|
||||
static int receive_locked(ThreadQueue *tq, int *stream_idx, |
||||
void *data) |
||||
{ |
||||
FifoElem elem; |
||||
unsigned int nb_finished = 0; |
||||
|
||||
if (av_fifo_read(tq->fifo, &elem, 1) >= 0) { |
||||
tq->obj_move(data, elem.obj); |
||||
objpool_release(tq->obj_pool, &elem.obj); |
||||
*stream_idx = elem.stream_idx; |
||||
return 0; |
||||
} |
||||
|
||||
for (unsigned int i = 0; i < tq->nb_streams; i++) { |
||||
if (!(tq->finished[i] & FINISHED_SEND)) |
||||
continue; |
||||
|
||||
/* return EOF to the consumer at most once for each stream */ |
||||
if (!(tq->finished[i] & FINISHED_RECV)) { |
||||
tq->finished[i] |= FINISHED_RECV; |
||||
*stream_idx = i; |
||||
return AVERROR_EOF; |
||||
} |
||||
|
||||
nb_finished++; |
||||
} |
||||
|
||||
return nb_finished == tq->nb_streams ? AVERROR_EOF : AVERROR(EAGAIN); |
||||
} |
||||
|
||||
int tq_receive(ThreadQueue *tq, int *stream_idx, void *data) |
||||
{ |
||||
int ret; |
||||
|
||||
*stream_idx = -1; |
||||
|
||||
pthread_mutex_lock(&tq->lock); |
||||
|
||||
while (1) { |
||||
ret = receive_locked(tq, stream_idx, data); |
||||
if (ret == AVERROR(EAGAIN)) { |
||||
pthread_cond_wait(&tq->cond, &tq->lock); |
||||
continue; |
||||
} |
||||
|
||||
break; |
||||
} |
||||
|
||||
if (ret == 0) |
||||
pthread_cond_broadcast(&tq->cond); |
||||
|
||||
pthread_mutex_unlock(&tq->lock); |
||||
|
||||
return ret; |
||||
} |
||||
|
||||
void tq_send_finish(ThreadQueue *tq, unsigned int stream_idx) |
||||
{ |
||||
av_assert0(stream_idx < tq->nb_streams); |
||||
|
||||
pthread_mutex_lock(&tq->lock); |
||||
|
||||
/* mark the stream as send-finished;
|
||||
* next time the consumer thread tries to read this stream it will get |
||||
* an EOF and recv-finished flag will be set */ |
||||
tq->finished[stream_idx] |= FINISHED_SEND; |
||||
pthread_cond_broadcast(&tq->cond); |
||||
|
||||
pthread_mutex_unlock(&tq->lock); |
||||
} |
||||
|
||||
void tq_receive_finish(ThreadQueue *tq, unsigned int stream_idx) |
||||
{ |
||||
av_assert0(stream_idx < tq->nb_streams); |
||||
|
||||
pthread_mutex_lock(&tq->lock); |
||||
|
||||
/* mark the stream as recv-finished;
|
||||
* next time the producer thread tries to send for this stream, it will |
||||
* get an EOF and send-finished flag will be set */ |
||||
tq->finished[stream_idx] |= FINISHED_RECV; |
||||
pthread_cond_broadcast(&tq->cond); |
||||
|
||||
pthread_mutex_unlock(&tq->lock); |
||||
} |
||||
@ -0,0 +1,94 @@ |
||||
/*
|
||||
* This file is part of FFmpeg. |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of thread_queue.h file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
*/ |
||||
|
||||
#ifndef FFTOOLS_THREAD_QUEUE_H |
||||
#define FFTOOLS_THREAD_QUEUE_H |
||||
|
||||
#include <string.h> |
||||
|
||||
#include "fftools_objpool.h" |
||||
|
||||
typedef struct ThreadQueue ThreadQueue; |
||||
|
||||
/**
|
||||
* Allocate a queue for sending data between threads. |
||||
* |
||||
* @param nb_streams number of streams for which a distinct EOF state is |
||||
* maintained |
||||
* @param queue_size number of items that can be stored in the queue without |
||||
* blocking |
||||
* @param obj_pool object pool that will be used to allocate items stored in the |
||||
* queue; the pool becomes owned by the queue |
||||
* @param callback that moves the contents between two data pointers |
||||
*/ |
||||
ThreadQueue *tq_alloc(unsigned int nb_streams, size_t queue_size, |
||||
ObjPool *obj_pool, void (*obj_move)(void *dst, void *src)); |
||||
void tq_free(ThreadQueue **tq); |
||||
|
||||
/**
|
||||
* Send an item for the given stream to the queue. |
||||
* |
||||
* @param data the item to send, its contents will be moved using the callback |
||||
* provided to tq_alloc(); on failure the item will be left |
||||
* untouched |
||||
* @return |
||||
* - 0 the item was successfully sent |
||||
* - AVERROR(ENOMEM) could not allocate an item for writing to the FIFO |
||||
* - AVERROR(EINVAL) the sending side has previously been marked as finished |
||||
* - AVERROR_EOF the receiving side has marked the given stream as finished |
||||
*/ |
||||
int tq_send(ThreadQueue *tq, unsigned int stream_idx, void *data); |
||||
/**
|
||||
* Mark the given stream finished from the sending side. |
||||
*/ |
||||
void tq_send_finish(ThreadQueue *tq, unsigned int stream_idx); |
||||
|
||||
/**
|
||||
* Read the next item from the queue. |
||||
* |
||||
* @param stream_idx the index of the stream that was processed or -1 will be |
||||
* written here |
||||
* @param data the data item will be written here on success using the |
||||
* callback provided to tq_alloc() |
||||
* @return |
||||
* - 0 a data item was successfully read; *stream_idx contains a non-negative |
||||
* stream index |
||||
* - AVERROR_EOF When *stream_idx is non-negative, this signals that the sending |
||||
* side has marked the given stream as finished. This will happen at most once |
||||
* for each stream. When *stream_idx is -1, all streams are done. |
||||
*/ |
||||
int tq_receive(ThreadQueue *tq, int *stream_idx, void *data); |
||||
/**
|
||||
* Mark the given stream finished from the receiving side. |
||||
*/ |
||||
void tq_receive_finish(ThreadQueue *tq, unsigned int stream_idx); |
||||
|
||||
#endif // FFTOOLS_THREAD_QUEUE_H
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,165 @@ |
||||
/*
|
||||
* Muxer internal APIs - should not be included outside of ffmpeg_mux* |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of ffmpeg_mux.h file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
* - fftools header names updated |
||||
* - want_sdp made thread-local |
||||
* - EncStatsFile declaration migrated from ffmpeg_mux_init.c |
||||
* - WARN_MULTIPLE_OPT_USAGE, MATCH_PER_STREAM_OPT, MATCH_PER_TYPE_OPT, SPECIFIER_OPT_FMT declarations migrated from |
||||
* ffmpeg.h |
||||
* - ms_from_ost migrated to ffmpeg_mux.c |
||||
*/ |
||||
|
||||
#ifndef FFTOOLS_FFMPEG_MUX_H |
||||
#define FFTOOLS_FFMPEG_MUX_H |
||||
|
||||
#include <stdatomic.h> |
||||
#include <stdint.h> |
||||
|
||||
#include "fftools_thread_queue.h" |
||||
|
||||
#include "libavformat/avformat.h" |
||||
|
||||
#include "libavcodec/packet.h" |
||||
|
||||
#include "libavutil/dict.h" |
||||
#include "libavutil/fifo.h" |
||||
#include "libavutil/thread.h" |
||||
|
||||
#define SPECIFIER_OPT_FMT_str "%s" |
||||
#define SPECIFIER_OPT_FMT_i "%i" |
||||
#define SPECIFIER_OPT_FMT_i64 "%"PRId64 |
||||
#define SPECIFIER_OPT_FMT_ui64 "%"PRIu64 |
||||
#define SPECIFIER_OPT_FMT_f "%f" |
||||
#define SPECIFIER_OPT_FMT_dbl "%lf" |
||||
|
||||
#define WARN_MULTIPLE_OPT_USAGE(name, type, so, st)\ |
||||
{\
|
||||
char namestr[128] = "";\
|
||||
const char *spec = so->specifier && so->specifier[0] ? so->specifier : "";\
|
||||
for (int _i = 0; opt_name_##name[_i]; _i++)\
|
||||
av_strlcatf(namestr, sizeof(namestr), "-%s%s", opt_name_##name[_i], opt_name_##name[_i+1] ? (opt_name_##name[_i+2] ? ", " : " or ") : "");\
|
||||
av_log(NULL, AV_LOG_WARNING, "Multiple %s options specified for stream %d, only the last option '-%s%s%s "SPECIFIER_OPT_FMT_##type"' will be used.\n",\
|
||||
namestr, st->index, opt_name_##name[0], spec[0] ? ":" : "", spec, so->u.type);\
|
||||
} |
||||
|
||||
#define MATCH_PER_STREAM_OPT(name, type, outvar, fmtctx, st)\ |
||||
{\
|
||||
int _ret, _matches = 0;\
|
||||
SpecifierOpt *so;\
|
||||
for (int _i = 0; _i < o->nb_ ## name; _i++) {\
|
||||
char *spec = o->name[_i].specifier;\
|
||||
if ((_ret = check_stream_specifier(fmtctx, st, spec)) > 0) {\
|
||||
outvar = o->name[_i].u.type;\
|
||||
so = &o->name[_i];\
|
||||
_matches++;\
|
||||
} else if (_ret < 0)\
|
||||
exit_program(1);\
|
||||
}\
|
||||
if (_matches > 1)\
|
||||
WARN_MULTIPLE_OPT_USAGE(name, type, so, st);\
|
||||
} |
||||
|
||||
#define MATCH_PER_TYPE_OPT(name, type, outvar, fmtctx, mediatype)\ |
||||
{\
|
||||
int i;\
|
||||
for (i = 0; i < o->nb_ ## name; i++) {\
|
||||
char *spec = o->name[i].specifier;\
|
||||
if (!strcmp(spec, mediatype))\
|
||||
outvar = o->name[i].u.type;\
|
||||
}\
|
||||
} |
||||
|
||||
typedef struct MuxStream { |
||||
OutputStream ost; |
||||
|
||||
// name used for logging
|
||||
char log_name[32]; |
||||
|
||||
/* the packets are buffered here until the muxer is ready to be initialized */ |
||||
AVFifo *muxing_queue; |
||||
|
||||
AVBSFContext *bsf_ctx; |
||||
|
||||
EncStats stats; |
||||
|
||||
int64_t max_frames; |
||||
|
||||
/*
|
||||
* The size of the AVPackets' buffers in queue. |
||||
* Updated when a packet is either pushed or pulled from the queue. |
||||
*/ |
||||
size_t muxing_queue_data_size; |
||||
|
||||
int max_muxing_queue_size; |
||||
|
||||
/* Threshold after which max_muxing_queue_size will be in effect */ |
||||
size_t muxing_queue_data_threshold; |
||||
|
||||
/* dts of the last packet sent to the muxer, in the stream timebase
|
||||
* used for making up missing dts values */ |
||||
int64_t last_mux_dts; |
||||
} MuxStream; |
||||
|
||||
typedef struct Muxer { |
||||
OutputFile of; |
||||
|
||||
// name used for logging
|
||||
char log_name[32]; |
||||
|
||||
AVFormatContext *fc; |
||||
|
||||
pthread_t thread; |
||||
ThreadQueue *tq; |
||||
|
||||
AVDictionary *opts; |
||||
|
||||
int thread_queue_size; |
||||
|
||||
/* filesize limit expressed in bytes */ |
||||
int64_t limit_filesize; |
||||
atomic_int_least64_t last_filesize; |
||||
int header_written; |
||||
|
||||
SyncQueue *sq_mux; |
||||
AVPacket *sq_pkt; |
||||
} Muxer; |
||||
|
||||
typedef struct EncStatsFile { |
||||
char *path; |
||||
AVIOContext *io; |
||||
} EncStatsFile; |
||||
|
||||
/* whether we want to print an SDP, set in of_open() */ |
||||
extern __thread int want_sdp; |
||||
|
||||
int mux_check_init(Muxer *mux); |
||||
|
||||
#endif /* FFTOOLS_FFMPEG_MUX_H */ |
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,145 @@ |
||||
/*
|
||||
* This file is part of FFmpeg. |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of objpool.c file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
* - fftools header names updated |
||||
*/ |
||||
|
||||
#include <stdint.h> |
||||
|
||||
#include "libavcodec/packet.h" |
||||
|
||||
#include "libavutil/common.h" |
||||
#include "libavutil/error.h" |
||||
#include "libavutil/frame.h" |
||||
#include "libavutil/mem.h" |
||||
|
||||
#include "fftools_objpool.h" |
||||
|
||||
struct ObjPool { |
||||
void *pool[32]; |
||||
unsigned int pool_count; |
||||
|
||||
ObjPoolCBAlloc alloc; |
||||
ObjPoolCBReset reset; |
||||
ObjPoolCBFree free; |
||||
}; |
||||
|
||||
ObjPool *objpool_alloc(ObjPoolCBAlloc cb_alloc, ObjPoolCBReset cb_reset, |
||||
ObjPoolCBFree cb_free) |
||||
{ |
||||
ObjPool *op = av_mallocz(sizeof(*op)); |
||||
|
||||
if (!op) |
||||
return NULL; |
||||
|
||||
op->alloc = cb_alloc; |
||||
op->reset = cb_reset; |
||||
op->free = cb_free; |
||||
|
||||
return op; |
||||
} |
||||
|
||||
void objpool_free(ObjPool **pop) |
||||
{ |
||||
ObjPool *op = *pop; |
||||
|
||||
if (!op) |
||||
return; |
||||
|
||||
for (unsigned int i = 0; i < op->pool_count; i++) |
||||
op->free(&op->pool[i]); |
||||
|
||||
av_freep(pop); |
||||
} |
||||
|
||||
int objpool_get(ObjPool *op, void **obj) |
||||
{ |
||||
if (op->pool_count) { |
||||
*obj = op->pool[--op->pool_count]; |
||||
op->pool[op->pool_count] = NULL; |
||||
} else |
||||
*obj = op->alloc(); |
||||
|
||||
return *obj ? 0 : AVERROR(ENOMEM); |
||||
} |
||||
|
||||
void objpool_release(ObjPool *op, void **obj) |
||||
{ |
||||
if (!*obj) |
||||
return; |
||||
|
||||
op->reset(*obj); |
||||
|
||||
if (op->pool_count < FF_ARRAY_ELEMS(op->pool)) |
||||
op->pool[op->pool_count++] = *obj; |
||||
else |
||||
op->free(obj); |
||||
|
||||
*obj = NULL; |
||||
} |
||||
|
||||
static void *alloc_packet(void) |
||||
{ |
||||
return av_packet_alloc(); |
||||
} |
||||
static void *alloc_frame(void) |
||||
{ |
||||
return av_frame_alloc(); |
||||
} |
||||
|
||||
static void reset_packet(void *obj) |
||||
{ |
||||
av_packet_unref(obj); |
||||
} |
||||
static void reset_frame(void *obj) |
||||
{ |
||||
av_frame_unref(obj); |
||||
} |
||||
|
||||
static void free_packet(void **obj) |
||||
{ |
||||
AVPacket *pkt = *obj; |
||||
av_packet_free(&pkt); |
||||
*obj = NULL; |
||||
} |
||||
static void free_frame(void **obj) |
||||
{ |
||||
AVFrame *frame = *obj; |
||||
av_frame_free(&frame); |
||||
*obj = NULL; |
||||
} |
||||
|
||||
ObjPool *objpool_alloc_packets(void) |
||||
{ |
||||
return objpool_alloc(alloc_packet, reset_packet, free_packet); |
||||
} |
||||
ObjPool *objpool_alloc_frames(void) |
||||
{ |
||||
return objpool_alloc(alloc_frame, reset_frame, free_frame); |
||||
} |
||||
@ -0,0 +1,50 @@ |
||||
/*
|
||||
* This file is part of FFmpeg. |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of objpool.h file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
*/ |
||||
|
||||
#ifndef FFTOOLS_OBJPOOL_H |
||||
#define FFTOOLS_OBJPOOL_H |
||||
|
||||
typedef struct ObjPool ObjPool; |
||||
|
||||
typedef void* (*ObjPoolCBAlloc)(void); |
||||
typedef void (*ObjPoolCBReset)(void *); |
||||
typedef void (*ObjPoolCBFree)(void **); |
||||
|
||||
void objpool_free(ObjPool **op); |
||||
ObjPool *objpool_alloc(ObjPoolCBAlloc cb_alloc, ObjPoolCBReset cb_reset, |
||||
ObjPoolCBFree cb_free); |
||||
ObjPool *objpool_alloc_packets(void); |
||||
ObjPool *objpool_alloc_frames(void); |
||||
|
||||
int objpool_get(ObjPool *op, void **obj); |
||||
void objpool_release(ObjPool *op, void **obj); |
||||
|
||||
#endif // FFTOOLS_OBJPOOL_H
|
||||
@ -0,0 +1,462 @@ |
||||
/*
|
||||
* This file is part of FFmpeg. |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of sync_queue.c file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
* - fftools header names updated |
||||
*/ |
||||
|
||||
#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 "fftools_objpool.h" |
||||
#include "fftools_sync_queue.h" |
||||
|
||||
typedef struct SyncQueueStream { |
||||
AVFifo *fifo; |
||||
AVRational tb; |
||||
|
||||
/* stream head: largest timestamp seen */ |
||||
int64_t head_ts; |
||||
int limiting; |
||||
/* no more frames will be sent for this stream */ |
||||
int finished; |
||||
|
||||
uint64_t frames_sent; |
||||
uint64_t frames_max; |
||||
} 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->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->limiting && 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->limiting && 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->limiting && 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 (st->limiting && |
||||
(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); |
||||
|
||||
st->frames_sent++; |
||||
if (st->frames_sent >= st->frames_max) |
||||
finish_stream(sq, stream_idx); |
||||
|
||||
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, int limiting) |
||||
{ |
||||
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; |
||||
st->frames_max = UINT64_MAX; |
||||
st->limiting = limiting; |
||||
|
||||
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; |
||||
} |
||||
|
||||
void sq_limit_frames(SyncQueue *sq, unsigned int stream_idx, uint64_t frames) |
||||
{ |
||||
SyncQueueStream *st; |
||||
|
||||
av_assert0(stream_idx < sq->nb_streams); |
||||
st = &sq->streams[stream_idx]; |
||||
|
||||
st->frames_max = frames; |
||||
if (st->frames_sent >= st->frames_max) |
||||
finish_stream(sq, stream_idx); |
||||
} |
||||
|
||||
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,122 @@ |
||||
/*
|
||||
* This file is part of FFmpeg. |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of sync_queue.h file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
*/ |
||||
|
||||
#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. |
||||
* |
||||
* @param limiting whether the stream is limiting, i.e. no other stream can be |
||||
* longer than this one |
||||
* @return |
||||
* - a non-negative stream index on success |
||||
* - a negative error code on error |
||||
*/ |
||||
int sq_add_stream(SyncQueue *sq, int limiting); |
||||
|
||||
/**
|
||||
* 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); |
||||
|
||||
/**
|
||||
* Limit the number of output frames for stream with index stream_idx |
||||
* to max_frames. |
||||
*/ |
||||
void sq_limit_frames(SyncQueue *sq, unsigned int stream_idx, |
||||
uint64_t max_frames); |
||||
|
||||
/**
|
||||
* 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
|
||||
@ -0,0 +1,259 @@ |
||||
/*
|
||||
* This file is part of FFmpeg. |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of thread_queue.c file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
* - fftools header names updated |
||||
*/ |
||||
|
||||
#include <stdint.h> |
||||
#include <string.h> |
||||
|
||||
#include "libavutil/avassert.h" |
||||
#include "libavutil/error.h" |
||||
#include "libavutil/fifo.h" |
||||
#include "libavutil/intreadwrite.h" |
||||
#include "libavutil/mem.h" |
||||
#include "libavutil/thread.h" |
||||
|
||||
#include "fftools_objpool.h" |
||||
#include "fftools_thread_queue.h" |
||||
|
||||
enum { |
||||
FINISHED_SEND = (1 << 0), |
||||
FINISHED_RECV = (1 << 1), |
||||
}; |
||||
|
||||
typedef struct FifoElem { |
||||
void *obj; |
||||
unsigned int stream_idx; |
||||
} FifoElem; |
||||
|
||||
struct ThreadQueue { |
||||
int *finished; |
||||
unsigned int nb_streams; |
||||
|
||||
AVFifo *fifo; |
||||
|
||||
ObjPool *obj_pool; |
||||
void (*obj_move)(void *dst, void *src); |
||||
|
||||
pthread_mutex_t lock; |
||||
pthread_cond_t cond; |
||||
}; |
||||
|
||||
void tq_free(ThreadQueue **ptq) |
||||
{ |
||||
ThreadQueue *tq = *ptq; |
||||
|
||||
if (!tq) |
||||
return; |
||||
|
||||
if (tq->fifo) { |
||||
FifoElem elem; |
||||
while (av_fifo_read(tq->fifo, &elem, 1) >= 0) |
||||
objpool_release(tq->obj_pool, &elem.obj); |
||||
} |
||||
av_fifo_freep2(&tq->fifo); |
||||
|
||||
objpool_free(&tq->obj_pool); |
||||
|
||||
av_freep(&tq->finished); |
||||
|
||||
pthread_cond_destroy(&tq->cond); |
||||
pthread_mutex_destroy(&tq->lock); |
||||
|
||||
av_freep(ptq); |
||||
} |
||||
|
||||
ThreadQueue *tq_alloc(unsigned int nb_streams, size_t queue_size, |
||||
ObjPool *obj_pool, void (*obj_move)(void *dst, void *src)) |
||||
{ |
||||
ThreadQueue *tq; |
||||
int ret; |
||||
|
||||
tq = av_mallocz(sizeof(*tq)); |
||||
if (!tq) |
||||
return NULL; |
||||
|
||||
ret = pthread_cond_init(&tq->cond, NULL); |
||||
if (ret) { |
||||
av_freep(&tq); |
||||
return NULL; |
||||
} |
||||
|
||||
ret = pthread_mutex_init(&tq->lock, NULL); |
||||
if (ret) { |
||||
pthread_cond_destroy(&tq->cond); |
||||
av_freep(&tq); |
||||
return NULL; |
||||
} |
||||
|
||||
tq->finished = av_calloc(nb_streams, sizeof(*tq->finished)); |
||||
if (!tq->finished) |
||||
goto fail; |
||||
tq->nb_streams = nb_streams; |
||||
|
||||
tq->fifo = av_fifo_alloc2(queue_size, sizeof(FifoElem), 0); |
||||
if (!tq->fifo) |
||||
goto fail; |
||||
|
||||
tq->obj_pool = obj_pool; |
||||
tq->obj_move = obj_move; |
||||
|
||||
return tq; |
||||
fail: |
||||
tq_free(&tq); |
||||
return NULL; |
||||
} |
||||
|
||||
int tq_send(ThreadQueue *tq, unsigned int stream_idx, void *data) |
||||
{ |
||||
int *finished; |
||||
int ret; |
||||
|
||||
av_assert0(stream_idx < tq->nb_streams); |
||||
finished = &tq->finished[stream_idx]; |
||||
|
||||
pthread_mutex_lock(&tq->lock); |
||||
|
||||
if (*finished & FINISHED_SEND) { |
||||
ret = AVERROR(EINVAL); |
||||
goto finish; |
||||
} |
||||
|
||||
while (!(*finished & FINISHED_RECV) && !av_fifo_can_write(tq->fifo)) |
||||
pthread_cond_wait(&tq->cond, &tq->lock); |
||||
|
||||
if (*finished & FINISHED_RECV) { |
||||
ret = AVERROR_EOF; |
||||
*finished |= FINISHED_SEND; |
||||
} else { |
||||
FifoElem elem = { .stream_idx = stream_idx }; |
||||
|
||||
ret = objpool_get(tq->obj_pool, &elem.obj); |
||||
if (ret < 0) |
||||
goto finish; |
||||
|
||||
tq->obj_move(elem.obj, data); |
||||
|
||||
ret = av_fifo_write(tq->fifo, &elem, 1); |
||||
av_assert0(ret >= 0); |
||||
pthread_cond_broadcast(&tq->cond); |
||||
} |
||||
|
||||
finish: |
||||
pthread_mutex_unlock(&tq->lock); |
||||
|
||||
return ret; |
||||
} |
||||
|
||||
static int receive_locked(ThreadQueue *tq, int *stream_idx, |
||||
void *data) |
||||
{ |
||||
FifoElem elem; |
||||
unsigned int nb_finished = 0; |
||||
|
||||
if (av_fifo_read(tq->fifo, &elem, 1) >= 0) { |
||||
tq->obj_move(data, elem.obj); |
||||
objpool_release(tq->obj_pool, &elem.obj); |
||||
*stream_idx = elem.stream_idx; |
||||
return 0; |
||||
} |
||||
|
||||
for (unsigned int i = 0; i < tq->nb_streams; i++) { |
||||
if (!(tq->finished[i] & FINISHED_SEND)) |
||||
continue; |
||||
|
||||
/* return EOF to the consumer at most once for each stream */ |
||||
if (!(tq->finished[i] & FINISHED_RECV)) { |
||||
tq->finished[i] |= FINISHED_RECV; |
||||
*stream_idx = i; |
||||
return AVERROR_EOF; |
||||
} |
||||
|
||||
nb_finished++; |
||||
} |
||||
|
||||
return nb_finished == tq->nb_streams ? AVERROR_EOF : AVERROR(EAGAIN); |
||||
} |
||||
|
||||
int tq_receive(ThreadQueue *tq, int *stream_idx, void *data) |
||||
{ |
||||
int ret; |
||||
|
||||
*stream_idx = -1; |
||||
|
||||
pthread_mutex_lock(&tq->lock); |
||||
|
||||
while (1) { |
||||
ret = receive_locked(tq, stream_idx, data); |
||||
if (ret == AVERROR(EAGAIN)) { |
||||
pthread_cond_wait(&tq->cond, &tq->lock); |
||||
continue; |
||||
} |
||||
|
||||
break; |
||||
} |
||||
|
||||
if (ret == 0) |
||||
pthread_cond_broadcast(&tq->cond); |
||||
|
||||
pthread_mutex_unlock(&tq->lock); |
||||
|
||||
return ret; |
||||
} |
||||
|
||||
void tq_send_finish(ThreadQueue *tq, unsigned int stream_idx) |
||||
{ |
||||
av_assert0(stream_idx < tq->nb_streams); |
||||
|
||||
pthread_mutex_lock(&tq->lock); |
||||
|
||||
/* mark the stream as send-finished;
|
||||
* next time the consumer thread tries to read this stream it will get |
||||
* an EOF and recv-finished flag will be set */ |
||||
tq->finished[stream_idx] |= FINISHED_SEND; |
||||
pthread_cond_broadcast(&tq->cond); |
||||
|
||||
pthread_mutex_unlock(&tq->lock); |
||||
} |
||||
|
||||
void tq_receive_finish(ThreadQueue *tq, unsigned int stream_idx) |
||||
{ |
||||
av_assert0(stream_idx < tq->nb_streams); |
||||
|
||||
pthread_mutex_lock(&tq->lock); |
||||
|
||||
/* mark the stream as recv-finished;
|
||||
* next time the producer thread tries to send for this stream, it will |
||||
* get an EOF and send-finished flag will be set */ |
||||
tq->finished[stream_idx] |= FINISHED_RECV; |
||||
pthread_cond_broadcast(&tq->cond); |
||||
|
||||
pthread_mutex_unlock(&tq->lock); |
||||
} |
||||
@ -0,0 +1,94 @@ |
||||
/*
|
||||
* This file is part of FFmpeg. |
||||
* Copyright (c) 2023 ARTHENICA LTD |
||||
* |
||||
* 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 |
||||
*/ |
||||
|
||||
/*
|
||||
* This file is the modified version of thread_queue.h file living in ffmpeg source code under the fftools folder. We |
||||
* manually update it each time we depend on a new ffmpeg version. Below you can see the list of changes applied |
||||
* by us to develop ffmpeg-kit library. |
||||
* |
||||
* ffmpeg-kit changes by ARTHENICA LTD |
||||
* |
||||
* 07.2023 |
||||
* -------------------------------------------------------- |
||||
* - FFmpeg 6.0 changes migrated |
||||
*/ |
||||
|
||||
#ifndef FFTOOLS_THREAD_QUEUE_H |
||||
#define FFTOOLS_THREAD_QUEUE_H |
||||
|
||||
#include <string.h> |
||||
|
||||
#include "fftools_objpool.h" |
||||
|
||||
typedef struct ThreadQueue ThreadQueue; |
||||
|
||||
/**
|
||||
* Allocate a queue for sending data between threads. |
||||
* |
||||
* @param nb_streams number of streams for which a distinct EOF state is |
||||
* maintained |
||||
* @param queue_size number of items that can be stored in the queue without |
||||
* blocking |
||||
* @param obj_pool object pool that will be used to allocate items stored in the |
||||
* queue; the pool becomes owned by the queue |
||||
* @param callback that moves the contents between two data pointers |
||||
*/ |
||||
ThreadQueue *tq_alloc(unsigned int nb_streams, size_t queue_size, |
||||
ObjPool *obj_pool, void (*obj_move)(void *dst, void *src)); |
||||
void tq_free(ThreadQueue **tq); |
||||
|
||||
/**
|
||||
* Send an item for the given stream to the queue. |
||||
* |
||||
* @param data the item to send, its contents will be moved using the callback |
||||
* provided to tq_alloc(); on failure the item will be left |
||||
* untouched |
||||
* @return |
||||
* - 0 the item was successfully sent |
||||
* - AVERROR(ENOMEM) could not allocate an item for writing to the FIFO |
||||
* - AVERROR(EINVAL) the sending side has previously been marked as finished |
||||
* - AVERROR_EOF the receiving side has marked the given stream as finished |
||||
*/ |
||||
int tq_send(ThreadQueue *tq, unsigned int stream_idx, void *data); |
||||
/**
|
||||
* Mark the given stream finished from the sending side. |
||||
*/ |
||||
void tq_send_finish(ThreadQueue *tq, unsigned int stream_idx); |
||||
|
||||
/**
|
||||
* Read the next item from the queue. |
||||
* |
||||
* @param stream_idx the index of the stream that was processed or -1 will be |
||||
* written here |
||||
* @param data the data item will be written here on success using the |
||||
* callback provided to tq_alloc() |
||||
* @return |
||||
* - 0 a data item was successfully read; *stream_idx contains a non-negative |
||||
* stream index |
||||
* - AVERROR_EOF When *stream_idx is non-negative, this signals that the sending |
||||
* side has marked the given stream as finished. This will happen at most once |
||||
* for each stream. When *stream_idx is -1, all streams are done. |
||||
*/ |
||||
int tq_receive(ThreadQueue *tq, int *stream_idx, void *data); |
||||
/**
|
||||
* Mark the given stream finished from the receiving side. |
||||
*/ |
||||
void tq_receive_finish(ThreadQueue *tq, unsigned int stream_idx); |
||||
|
||||
#endif // FFTOOLS_THREAD_QUEUE_H
|
||||
Loading…
Reference in new issue