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