Loading...
Searching...
No Matches
fftools_thread_queue.c
Go to the documentation of this file.
/*
* 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);
}
pthread_mutex_lock
#define pthread_mutex_lock(a)
Definition fftools_ffprobe.c:102
pthread_mutex_unlock
#define pthread_mutex_unlock(a)
Definition fftools_ffprobe.c:106
nb_streams
__thread int nb_streams
Definition fftools_ffprobe.c:336
objpool_release
void objpool_release(ObjPool *op, void **obj)
Definition fftools_objpool.c:92
objpool_free
void objpool_free(ObjPool **pop)
Definition fftools_objpool.c:68
objpool_get
int objpool_get(ObjPool *op, void **obj)
Definition fftools_objpool.c:81
fftools_objpool.h
tq_send_finish
void tq_send_finish(ThreadQueue *tq, unsigned int stream_idx)
Definition fftools_thread_queue.c:231
tq_send
int tq_send(ThreadQueue *tq, unsigned int stream_idx, void *data)
Definition fftools_thread_queue.c:134
FINISHED_SEND
@ FINISHED_SEND
Definition fftools_thread_queue.c:47
FINISHED_RECV
@ FINISHED_RECV
Definition fftools_thread_queue.c:48
tq_alloc
ThreadQueue * tq_alloc(unsigned int nb_streams, size_t queue_size, ObjPool *obj_pool, void(*obj_move)(void *dst, void *src))
Definition fftools_thread_queue.c:93
receive_locked
static int receive_locked(ThreadQueue *tq, int *stream_idx, void *data)
Definition fftools_thread_queue.c:175
tq_receive
int tq_receive(ThreadQueue *tq, int *stream_idx, void *data)
Definition fftools_thread_queue.c:205
tq_free
void tq_free(ThreadQueue **ptq)
Definition fftools_thread_queue.c:69
tq_receive_finish
void tq_receive_finish(ThreadQueue *tq, unsigned int stream_idx)
Definition fftools_thread_queue.c:246
fftools_thread_queue.h
FifoElem
Definition fftools_thread_queue.c:51
FifoElem::stream_idx
unsigned int stream_idx
Definition fftools_thread_queue.c:53
FifoElem::obj
void * obj
Definition fftools_thread_queue.c:52
ObjPool
Definition fftools_objpool.c:44
ThreadQueue
Definition fftools_thread_queue.c:56
ThreadQueue::lock
pthread_mutex_t lock
Definition fftools_thread_queue.c:65
ThreadQueue::finished
int * finished
Definition fftools_thread_queue.c:57
ThreadQueue::nb_streams
unsigned int nb_streams
Definition fftools_thread_queue.c:58
ThreadQueue::fifo
AVFifo * fifo
Definition fftools_thread_queue.c:60
ThreadQueue::obj_pool
ObjPool * obj_pool
Definition fftools_thread_queue.c:62
ThreadQueue::cond
pthread_cond_t cond
Definition fftools_thread_queue.c:66
ThreadQueue::obj_move
void(* obj_move)(void *dst, void *src)
Definition fftools_thread_queue.c:63