/* * 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); }