avutil/threadmessage: split the pthread condition in two
[ffmpeg.git] / 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_recv;
40     pthread_cond_t cond_send;
41     int err_send;
42     int err_recv;
43     unsigned elsize;
44     void (*free_func)(void *msg);
45 #else
46     int dummy;
47 #endif
48 };
49
50 int av_thread_message_queue_alloc(AVThreadMessageQueue **mq,
51                                   unsigned nelem,
52                                   unsigned elsize)
53 {
54 #if HAVE_THREADS
55     AVThreadMessageQueue *rmq;
56     int ret = 0;
57
58     if (nelem > INT_MAX / elsize)
59         return AVERROR(EINVAL);
60     if (!(rmq = av_mallocz(sizeof(*rmq))))
61         return AVERROR(ENOMEM);
62     if ((ret = pthread_mutex_init(&rmq->lock, NULL))) {
63         av_free(rmq);
64         return AVERROR(ret);
65     }
66     if ((ret = pthread_cond_init(&rmq->cond_recv, NULL))) {
67         pthread_mutex_destroy(&rmq->lock);
68         av_free(rmq);
69         return AVERROR(ret);
70     }
71     if ((ret = pthread_cond_init(&rmq->cond_send, NULL))) {
72         pthread_cond_destroy(&rmq->cond_recv);
73         pthread_mutex_destroy(&rmq->lock);
74         av_free(rmq);
75         return AVERROR(ret);
76     }
77     if (!(rmq->fifo = av_fifo_alloc(elsize * nelem))) {
78         pthread_cond_destroy(&rmq->cond_send);
79         pthread_cond_destroy(&rmq->cond_recv);
80         pthread_mutex_destroy(&rmq->lock);
81         av_free(rmq);
82         return AVERROR(ret);
83     }
84     rmq->elsize = elsize;
85     *mq = rmq;
86     return 0;
87 #else
88     *mq = NULL;
89     return AVERROR(ENOSYS);
90 #endif /* HAVE_THREADS */
91 }
92
93 void av_thread_message_queue_set_free_func(AVThreadMessageQueue *mq,
94                                            void (*free_func)(void *msg))
95 {
96     mq->free_func = free_func;
97 }
98
99 void av_thread_message_queue_free(AVThreadMessageQueue **mq)
100 {
101 #if HAVE_THREADS
102     if (*mq) {
103         av_thread_message_flush(*mq);
104         av_fifo_freep(&(*mq)->fifo);
105         pthread_cond_destroy(&(*mq)->cond_send);
106         pthread_cond_destroy(&(*mq)->cond_recv);
107         pthread_mutex_destroy(&(*mq)->lock);
108         av_freep(mq);
109     }
110 #endif
111 }
112
113 #if HAVE_THREADS
114
115 static int av_thread_message_queue_send_locked(AVThreadMessageQueue *mq,
116                                                void *msg,
117                                                unsigned flags)
118 {
119     while (!mq->err_send && av_fifo_space(mq->fifo) < mq->elsize) {
120         if ((flags & AV_THREAD_MESSAGE_NONBLOCK))
121             return AVERROR(EAGAIN);
122         pthread_cond_wait(&mq->cond_send, &mq->lock);
123     }
124     if (mq->err_send)
125         return mq->err_send;
126     av_fifo_generic_write(mq->fifo, msg, mq->elsize, NULL);
127     /* one message is sent, signal one receiver */
128     pthread_cond_signal(&mq->cond_recv);
129     return 0;
130 }
131
132 static int av_thread_message_queue_recv_locked(AVThreadMessageQueue *mq,
133                                                void *msg,
134                                                unsigned flags)
135 {
136     while (!mq->err_recv && av_fifo_size(mq->fifo) < mq->elsize) {
137         if ((flags & AV_THREAD_MESSAGE_NONBLOCK))
138             return AVERROR(EAGAIN);
139         pthread_cond_wait(&mq->cond_recv, &mq->lock);
140     }
141     if (av_fifo_size(mq->fifo) < mq->elsize)
142         return mq->err_recv;
143     av_fifo_generic_read(mq->fifo, msg, mq->elsize, NULL);
144     /* one message space appeared, signal one sender */
145     pthread_cond_signal(&mq->cond_send);
146     return 0;
147 }
148
149 #endif /* HAVE_THREADS */
150
151 int av_thread_message_queue_send(AVThreadMessageQueue *mq,
152                                  void *msg,
153                                  unsigned flags)
154 {
155 #if HAVE_THREADS
156     int ret;
157
158     pthread_mutex_lock(&mq->lock);
159     ret = av_thread_message_queue_send_locked(mq, msg, flags);
160     pthread_mutex_unlock(&mq->lock);
161     return ret;
162 #else
163     return AVERROR(ENOSYS);
164 #endif /* HAVE_THREADS */
165 }
166
167 int av_thread_message_queue_recv(AVThreadMessageQueue *mq,
168                                  void *msg,
169                                  unsigned flags)
170 {
171 #if HAVE_THREADS
172     int ret;
173
174     pthread_mutex_lock(&mq->lock);
175     ret = av_thread_message_queue_recv_locked(mq, msg, flags);
176     pthread_mutex_unlock(&mq->lock);
177     return ret;
178 #else
179     return AVERROR(ENOSYS);
180 #endif /* HAVE_THREADS */
181 }
182
183 void av_thread_message_queue_set_err_send(AVThreadMessageQueue *mq,
184                                           int err)
185 {
186 #if HAVE_THREADS
187     pthread_mutex_lock(&mq->lock);
188     mq->err_send = err;
189     pthread_cond_broadcast(&mq->cond_send);
190     pthread_mutex_unlock(&mq->lock);
191 #endif /* HAVE_THREADS */
192 }
193
194 void av_thread_message_queue_set_err_recv(AVThreadMessageQueue *mq,
195                                           int err)
196 {
197 #if HAVE_THREADS
198     pthread_mutex_lock(&mq->lock);
199     mq->err_recv = err;
200     pthread_cond_broadcast(&mq->cond_recv);
201     pthread_mutex_unlock(&mq->lock);
202 #endif /* HAVE_THREADS */
203 }
204
205 static void free_func_wrap(void *arg, void *msg, int size)
206 {
207     AVThreadMessageQueue *mq = arg;
208     mq->free_func(msg);
209 }
210
211 void av_thread_message_flush(AVThreadMessageQueue *mq)
212 {
213 #if HAVE_THREADS
214     int used, off;
215     void *free_func = mq->free_func;
216
217     pthread_mutex_lock(&mq->lock);
218     used = av_fifo_size(mq->fifo);
219     if (free_func)
220         for (off = 0; off < used; off += mq->elsize)
221             av_fifo_generic_peek_at(mq->fifo, mq, off, mq->elsize, free_func_wrap);
222     av_fifo_drain(mq->fifo, used);
223     /* only the senders need to be notified since the queue is empty and there
224      * is nothing to read */
225     pthread_cond_broadcast(&mq->cond_send);
226     pthread_mutex_unlock(&mq->lock);
227 #endif /* HAVE_THREADS */
228 }