ffmpeg/libavutil/threadmessage.c

   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
  35 struct 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
  48 int 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
  84 void 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
  98 static 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
 114 static 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
 132 int 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
 148 int 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
 164 void 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
 175 void 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 }