1/* 2 * Copyright (c) 2014 Nicolas George 3 * 4 * This file is part of FFmpeg. 5 * 6 * FFmpeg is free software; you can redistribute it and/or 7 * modify it under the terms of the GNU Lesser General Public License 8 * as published by the Free Software Foundation; either 9 * version 2.1 of the License, or (at your option) any later version. 10 * 11 * FFmpeg is distributed in the hope that it will be useful, 12 * but WITHOUT ANY WARRANTY; without even the implied warranty of 13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 14 * GNU Lesser General Public License for more details. 15 * 16 * You should have received a copy of the GNU Lesser General Public License 17 * along with FFmpeg; if not, write to the Free Software Foundation, Inc., 18 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA 19 */ 20 21#include "fifo.h" 22#include "threadmessage.h" 23#if HAVE_THREADS 24#if HAVE_PTHREADS 25#include <pthread.h> 26#elif HAVE_W32THREADS 27#include "compat/w32pthreads.h" 28#elif HAVE_OS2THREADS 29#include "compat/os2threads.h" 30#else 31#error "Unknown threads implementation" 32#endif 33#endif 34 35struct AVThreadMessageQueue { 36#if HAVE_THREADS 37 AVFifoBuffer *fifo; 38 pthread_mutex_t lock; 39 pthread_cond_t cond; 40 int err_send; 41 int err_recv; 42 unsigned elsize; 43#else 44 int dummy; 45#endif 46}; 47 48int av_thread_message_queue_alloc(AVThreadMessageQueue **mq, 49 unsigned nelem, 50 unsigned elsize) 51{ 52#if HAVE_THREADS 53 AVThreadMessageQueue *rmq; 54 int ret = 0; 55 56 if (nelem > INT_MAX / elsize) 57 return AVERROR(EINVAL); 58 if (!(rmq = av_mallocz(sizeof(*rmq)))) 59 return AVERROR(ENOMEM); 60 if ((ret = pthread_mutex_init(&rmq->lock, NULL))) { 61 av_free(rmq); 62 return AVERROR(ret); 63 } 64 if ((ret = pthread_cond_init(&rmq->cond, NULL))) { 65 pthread_mutex_destroy(&rmq->lock); 66 av_free(rmq); 67 return AVERROR(ret); 68 } 69 if (!(rmq->fifo = av_fifo_alloc(elsize * nelem))) { 70 pthread_cond_destroy(&rmq->cond); 71 pthread_mutex_destroy(&rmq->lock); 72 av_free(rmq); 73 return AVERROR(ret); 74 } 75 rmq->elsize = elsize; 76 *mq = rmq; 77 return 0; 78#else 79 *mq = NULL; 80 return AVERROR(ENOSYS); 81#endif /* HAVE_THREADS */ 82} 83 84void av_thread_message_queue_free(AVThreadMessageQueue **mq) 85{ 86#if HAVE_THREADS 87 if (*mq) { 88 av_fifo_freep(&(*mq)->fifo); 89 pthread_cond_destroy(&(*mq)->cond); 90 pthread_mutex_destroy(&(*mq)->lock); 91 av_freep(mq); 92 } 93#endif 94} 95 96#if HAVE_THREADS 97 98static int av_thread_message_queue_send_locked(AVThreadMessageQueue *mq, 99 void *msg, 100 unsigned flags) 101{ 102 while (!mq->err_send && av_fifo_space(mq->fifo) < mq->elsize) { 103 if ((flags & AV_THREAD_MESSAGE_NONBLOCK)) 104 return AVERROR(EAGAIN); 105 pthread_cond_wait(&mq->cond, &mq->lock); 106 } 107 if (mq->err_send) 108 return mq->err_send; 109 av_fifo_generic_write(mq->fifo, msg, mq->elsize, NULL); 110 pthread_cond_signal(&mq->cond); 111 return 0; 112} 113 114static int av_thread_message_queue_recv_locked(AVThreadMessageQueue *mq, 115 void *msg, 116 unsigned flags) 117{ 118 while (!mq->err_recv && av_fifo_size(mq->fifo) < mq->elsize) { 119 if ((flags & AV_THREAD_MESSAGE_NONBLOCK)) 120 return AVERROR(EAGAIN); 121 pthread_cond_wait(&mq->cond, &mq->lock); 122 } 123 if (av_fifo_size(mq->fifo) < mq->elsize) 124 return mq->err_recv; 125 av_fifo_generic_read(mq->fifo, msg, mq->elsize, NULL); 126 pthread_cond_signal(&mq->cond); 127 return 0; 128} 129 130#endif /* HAVE_THREADS */ 131 132int av_thread_message_queue_send(AVThreadMessageQueue *mq, 133 void *msg, 134 unsigned flags) 135{ 136#if HAVE_THREADS 137 int ret; 138 139 pthread_mutex_lock(&mq->lock); 140 ret = av_thread_message_queue_send_locked(mq, msg, flags); 141 pthread_mutex_unlock(&mq->lock); 142 return ret; 143#else 144 return AVERROR(ENOSYS); 145#endif /* HAVE_THREADS */ 146} 147 148int av_thread_message_queue_recv(AVThreadMessageQueue *mq, 149 void *msg, 150 unsigned flags) 151{ 152#if HAVE_THREADS 153 int ret; 154 155 pthread_mutex_lock(&mq->lock); 156 ret = av_thread_message_queue_recv_locked(mq, msg, flags); 157 pthread_mutex_unlock(&mq->lock); 158 return ret; 159#else 160 return AVERROR(ENOSYS); 161#endif /* HAVE_THREADS */ 162} 163 164void av_thread_message_queue_set_err_send(AVThreadMessageQueue *mq, 165 int err) 166{ 167#if HAVE_THREADS 168 pthread_mutex_lock(&mq->lock); 169 mq->err_send = err; 170 pthread_cond_broadcast(&mq->cond); 171 pthread_mutex_unlock(&mq->lock); 172#endif /* HAVE_THREADS */ 173} 174 175void av_thread_message_queue_set_err_recv(AVThreadMessageQueue *mq, 176 int err) 177{ 178#if HAVE_THREADS 179 pthread_mutex_lock(&mq->lock); 180 mq->err_recv = err; 181 pthread_cond_broadcast(&mq->cond); 182 pthread_mutex_unlock(&mq->lock); 183#endif /* HAVE_THREADS */ 184} 185