mirror of
https://github.com/FFmpeg/FFmpeg.git
synced 2024-12-07 11:13:41 +02:00
87016e031f
This ensures that tq_receive() will always return EOF after all streams were receive-finished, even though the sending side might not have closed them yet. This may allow the receiver to avoid manually tracking which streams it has already closed.
255 lines
6.0 KiB
C
255 lines
6.0 KiB
C
/*
|
|
* This file is part of FFmpeg.
|
|
*
|
|
* FFmpeg is free software; you can redistribute it and/or
|
|
* modify it under the terms of the GNU Lesser General Public
|
|
* License as published by the Free Software Foundation; either
|
|
* version 2.1 of the License, or (at your option) any later version.
|
|
*
|
|
* FFmpeg is distributed in the hope that it will be useful,
|
|
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
|
|
* Lesser General Public License for more details.
|
|
*
|
|
* You should have received a copy of the GNU Lesser General Public
|
|
* License along with FFmpeg; if not, write to the Free Software
|
|
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
|
|
*/
|
|
|
|
#include <stdint.h>
|
|
#include <string.h>
|
|
|
|
#include "libavutil/avassert.h"
|
|
#include "libavutil/error.h"
|
|
#include "libavutil/fifo.h"
|
|
#include "libavutil/intreadwrite.h"
|
|
#include "libavutil/mem.h"
|
|
#include "libavutil/thread.h"
|
|
|
|
#include "objpool.h"
|
|
#include "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;
|
|
|
|
while (av_fifo_read(tq->fifo, &elem, 1) >= 0) {
|
|
if (tq->finished[elem.stream_idx] & FINISHED_RECV) {
|
|
objpool_release(tq->obj_pool, &elem.obj);
|
|
continue;
|
|
}
|
|
|
|
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])
|
|
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) {
|
|
size_t can_read = av_fifo_can_read(tq->fifo);
|
|
|
|
ret = receive_locked(tq, stream_idx, data);
|
|
|
|
// signal other threads if the fifo state changed
|
|
if (can_read != av_fifo_can_read(tq->fifo))
|
|
pthread_cond_broadcast(&tq->cond);
|
|
|
|
if (ret == AVERROR(EAGAIN)) {
|
|
pthread_cond_wait(&tq->cond, &tq->lock);
|
|
continue;
|
|
}
|
|
|
|
break;
|
|
}
|
|
|
|
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);
|
|
}
|