/* * Copyright (c) 2014 Nicolas George * * 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 <limits.h> #include <stddef.h> #include "error.h" #include "fifo.h" #include "mem.h" #include "threadmessage.h" #include "thread.h" struct AVThreadMessageQueue { #if HAVE_THREADS AVFifo *fifo; pthread_mutex_t lock; pthread_cond_t cond_recv; pthread_cond_t cond_send; int err_send; int err_recv; unsigned elsize; void (*free_func)(void *msg); #else int dummy; #endif }; int av_thread_message_queue_alloc(AVThreadMessageQueue **mq, unsigned nelem, unsigned elsize) { #if HAVE_THREADS AVThreadMessageQueue *rmq; int ret = 0; if (nelem > INT_MAX / elsize) return AVERROR(EINVAL); if (!(rmq = av_mallocz(sizeof(*rmq)))) return AVERROR(ENOMEM); if ((ret = pthread_mutex_init(&rmq->lock, NULL))) { av_free(rmq); return AVERROR(ret); } if ((ret = pthread_cond_init(&rmq->cond_recv, NULL))) { pthread_mutex_destroy(&rmq->lock); av_free(rmq); return AVERROR(ret); } if ((ret = pthread_cond_init(&rmq->cond_send, NULL))) { pthread_cond_destroy(&rmq->cond_recv); pthread_mutex_destroy(&rmq->lock); av_free(rmq); return AVERROR(ret); } if (!(rmq->fifo = av_fifo_alloc2(nelem, elsize, 0))) { pthread_cond_destroy(&rmq->cond_send); pthread_cond_destroy(&rmq->cond_recv); pthread_mutex_destroy(&rmq->lock); av_free(rmq); return AVERROR(ENOMEM); } rmq->elsize = elsize; *mq = rmq; return 0; #else *mq = NULL; return AVERROR(ENOSYS); #endif /* HAVE_THREADS */ } void av_thread_message_queue_set_free_func(AVThreadMessageQueue *mq, void (*free_func)(void *msg)) { #if HAVE_THREADS mq->free_func = free_func; #endif } void av_thread_message_queue_free(AVThreadMessageQueue **mq) { #if HAVE_THREADS if (*mq) { av_thread_message_flush(*mq); av_fifo_freep2(&(*mq)->fifo); pthread_cond_destroy(&(*mq)->cond_send); pthread_cond_destroy(&(*mq)->cond_recv); pthread_mutex_destroy(&(*mq)->lock); av_freep(mq); } #endif } int av_thread_message_queue_nb_elems(AVThreadMessageQueue *mq) { #if HAVE_THREADS int ret; pthread_mutex_lock(&mq->lock); ret = av_fifo_can_read(mq->fifo); pthread_mutex_unlock(&mq->lock); return ret; #else return AVERROR(ENOSYS); #endif } #if HAVE_THREADS static int av_thread_message_queue_send_locked(AVThreadMessageQueue *mq, void *msg, unsigned flags) { while (!mq->err_send && !av_fifo_can_write(mq->fifo)) { if ((flags & AV_THREAD_MESSAGE_NONBLOCK)) return AVERROR(EAGAIN); pthread_cond_wait(&mq->cond_send, &mq->lock); } if (mq->err_send) return mq->err_send; av_fifo_write(mq->fifo, msg, 1); /* one message is sent, signal one receiver */ pthread_cond_signal(&mq->cond_recv); return 0; } static int av_thread_message_queue_recv_locked(AVThreadMessageQueue *mq, void *msg, unsigned flags) { while (!mq->err_recv && !av_fifo_can_read(mq->fifo)) { if ((flags & AV_THREAD_MESSAGE_NONBLOCK)) return AVERROR(EAGAIN); pthread_cond_wait(&mq->cond_recv, &mq->lock); } if (!av_fifo_can_read(mq->fifo)) return mq->err_recv; av_fifo_read(mq->fifo, msg, 1); /* one message space appeared, signal one sender */ pthread_cond_signal(&mq->cond_send); return 0; } #endif /* HAVE_THREADS */ int av_thread_message_queue_send(AVThreadMessageQueue *mq, void *msg, unsigned flags) { #if HAVE_THREADS int ret; pthread_mutex_lock(&mq->lock); ret = av_thread_message_queue_send_locked(mq, msg, flags); pthread_mutex_unlock(&mq->lock); return ret; #else return AVERROR(ENOSYS); #endif /* HAVE_THREADS */ } int av_thread_message_queue_recv(AVThreadMessageQueue *mq, void *msg, unsigned flags) { #if HAVE_THREADS int ret; pthread_mutex_lock(&mq->lock); ret = av_thread_message_queue_recv_locked(mq, msg, flags); pthread_mutex_unlock(&mq->lock); return ret; #else return AVERROR(ENOSYS); #endif /* HAVE_THREADS */ } void av_thread_message_queue_set_err_send(AVThreadMessageQueue *mq, int err) { #if HAVE_THREADS pthread_mutex_lock(&mq->lock); mq->err_send = err; pthread_cond_broadcast(&mq->cond_send); pthread_mutex_unlock(&mq->lock); #endif /* HAVE_THREADS */ } void av_thread_message_queue_set_err_recv(AVThreadMessageQueue *mq, int err) { #if HAVE_THREADS pthread_mutex_lock(&mq->lock); mq->err_recv = err; pthread_cond_broadcast(&mq->cond_recv); pthread_mutex_unlock(&mq->lock); #endif /* HAVE_THREADS */ } #if HAVE_THREADS static int free_func_wrap(void *arg, void *buf, size_t *nb_elems) { AVThreadMessageQueue *mq = arg; uint8_t *msg = buf; for (size_t i = 0; i < *nb_elems; i++) mq->free_func(msg + i * mq->elsize); return 0; } #endif void av_thread_message_flush(AVThreadMessageQueue *mq) { #if HAVE_THREADS size_t used; pthread_mutex_lock(&mq->lock); used = av_fifo_can_read(mq->fifo); if (mq->free_func) av_fifo_read_to_cb(mq->fifo, free_func_wrap, mq, &used); /* only the senders need to be notified since the queue is empty and there * is nothing to read */ pthread_cond_broadcast(&mq->cond_send); pthread_mutex_unlock(&mq->lock); #endif /* HAVE_THREADS */ }