Line | Branch | Exec | Source |
---|---|---|---|
1 | /* | ||
2 | * This file is part of FFmpeg. | ||
3 | * | ||
4 | * FFmpeg is free software; you can redistribute it and/or | ||
5 | * modify it under the terms of the GNU Lesser General Public | ||
6 | * License as published by the Free Software Foundation; either | ||
7 | * version 2.1 of the License, or (at your option) any later version. | ||
8 | * | ||
9 | * FFmpeg is distributed in the hope that it will be useful, | ||
10 | * but WITHOUT ANY WARRANTY; without even the implied warranty of | ||
11 | * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU | ||
12 | * Lesser General Public License for more details. | ||
13 | * | ||
14 | * You should have received a copy of the GNU Lesser General Public | ||
15 | * License along with FFmpeg; if not, write to the Free Software | ||
16 | * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA | ||
17 | */ | ||
18 | |||
19 | #include <stdint.h> | ||
20 | #include <string.h> | ||
21 | |||
22 | #include "libavutil/avassert.h" | ||
23 | #include "libavutil/error.h" | ||
24 | #include "libavutil/fifo.h" | ||
25 | #include "libavutil/intreadwrite.h" | ||
26 | #include "libavutil/mem.h" | ||
27 | #include "libavutil/thread.h" | ||
28 | |||
29 | #include "objpool.h" | ||
30 | #include "thread_queue.h" | ||
31 | |||
32 | enum { | ||
33 | FINISHED_SEND = (1 << 0), | ||
34 | FINISHED_RECV = (1 << 1), | ||
35 | }; | ||
36 | |||
37 | typedef struct FifoElem { | ||
38 | void *obj; | ||
39 | unsigned int stream_idx; | ||
40 | } FifoElem; | ||
41 | |||
42 | struct ThreadQueue { | ||
43 | int *finished; | ||
44 | unsigned int nb_streams; | ||
45 | |||
46 | AVFifo *fifo; | ||
47 | |||
48 | ObjPool *obj_pool; | ||
49 | void (*obj_move)(void *dst, void *src); | ||
50 | |||
51 | pthread_mutex_t lock; | ||
52 | pthread_cond_t cond; | ||
53 | }; | ||
54 | |||
55 | 6566 | void tq_free(ThreadQueue **ptq) | |
56 | { | ||
57 | 6566 | ThreadQueue *tq = *ptq; | |
58 | |||
59 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 6566 times.
|
6566 | if (!tq) |
60 | ✗ | return; | |
61 | |||
62 |
1/2✓ Branch 0 taken 6566 times.
✗ Branch 1 not taken.
|
6566 | if (tq->fifo) { |
63 | FifoElem elem; | ||
64 |
1/2✗ Branch 1 not taken.
✓ Branch 2 taken 6566 times.
|
6566 | while (av_fifo_read(tq->fifo, &elem, 1) >= 0) |
65 | ✗ | objpool_release(tq->obj_pool, &elem.obj); | |
66 | } | ||
67 | 6566 | av_fifo_freep2(&tq->fifo); | |
68 | |||
69 | 6566 | objpool_free(&tq->obj_pool); | |
70 | |||
71 | 6566 | av_freep(&tq->finished); | |
72 | |||
73 | 6566 | pthread_cond_destroy(&tq->cond); | |
74 | 6566 | pthread_mutex_destroy(&tq->lock); | |
75 | |||
76 | 6566 | av_freep(ptq); | |
77 | } | ||
78 | |||
79 | 6566 | ThreadQueue *tq_alloc(unsigned int nb_streams, size_t queue_size, | |
80 | ObjPool *obj_pool, void (*obj_move)(void *dst, void *src)) | ||
81 | { | ||
82 | ThreadQueue *tq; | ||
83 | int ret; | ||
84 | |||
85 | 6566 | tq = av_mallocz(sizeof(*tq)); | |
86 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 6566 times.
|
6566 | if (!tq) |
87 | ✗ | return NULL; | |
88 | |||
89 | 6566 | ret = pthread_cond_init(&tq->cond, NULL); | |
90 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 6566 times.
|
6566 | if (ret) { |
91 | ✗ | av_freep(&tq); | |
92 | ✗ | return NULL; | |
93 | } | ||
94 | |||
95 | 6566 | ret = pthread_mutex_init(&tq->lock, NULL); | |
96 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 6566 times.
|
6566 | if (ret) { |
97 | ✗ | pthread_cond_destroy(&tq->cond); | |
98 | ✗ | av_freep(&tq); | |
99 | ✗ | return NULL; | |
100 | } | ||
101 | |||
102 | 6566 | tq->finished = av_calloc(nb_streams, sizeof(*tq->finished)); | |
103 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 6566 times.
|
6566 | if (!tq->finished) |
104 | ✗ | goto fail; | |
105 | 6566 | tq->nb_streams = nb_streams; | |
106 | |||
107 | 6566 | tq->fifo = av_fifo_alloc2(queue_size, sizeof(FifoElem), 0); | |
108 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 6566 times.
|
6566 | if (!tq->fifo) |
109 | ✗ | goto fail; | |
110 | |||
111 | 6566 | tq->obj_pool = obj_pool; | |
112 | 6566 | tq->obj_move = obj_move; | |
113 | |||
114 | 6566 | return tq; | |
115 | ✗ | fail: | |
116 | ✗ | tq_free(&tq); | |
117 | ✗ | return NULL; | |
118 | } | ||
119 | |||
120 | 483973 | int tq_send(ThreadQueue *tq, unsigned int stream_idx, void *data) | |
121 | { | ||
122 | int *finished; | ||
123 | int ret; | ||
124 | |||
125 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 483973 times.
|
483973 | av_assert0(stream_idx < tq->nb_streams); |
126 | 483973 | finished = &tq->finished[stream_idx]; | |
127 | |||
128 | 483973 | pthread_mutex_lock(&tq->lock); | |
129 | |||
130 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 483973 times.
|
483973 | if (*finished & FINISHED_SEND) { |
131 | ✗ | ret = AVERROR(EINVAL); | |
132 | ✗ | goto finish; | |
133 | } | ||
134 | |||
135 |
4/4✓ Branch 0 taken 518292 times.
✓ Branch 1 taken 18 times.
✓ Branch 3 taken 34337 times.
✓ Branch 4 taken 483955 times.
|
518310 | while (!(*finished & FINISHED_RECV) && !av_fifo_can_write(tq->fifo)) |
136 | 34337 | pthread_cond_wait(&tq->cond, &tq->lock); | |
137 | |||
138 |
2/2✓ Branch 0 taken 18 times.
✓ Branch 1 taken 483955 times.
|
483973 | if (*finished & FINISHED_RECV) { |
139 | 18 | ret = AVERROR_EOF; | |
140 | 18 | *finished |= FINISHED_SEND; | |
141 | } else { | ||
142 | 483955 | FifoElem elem = { .stream_idx = stream_idx }; | |
143 | |||
144 | 483955 | ret = objpool_get(tq->obj_pool, &elem.obj); | |
145 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 483955 times.
|
483955 | if (ret < 0) |
146 | ✗ | goto finish; | |
147 | |||
148 | 483955 | tq->obj_move(elem.obj, data); | |
149 | |||
150 | 483955 | ret = av_fifo_write(tq->fifo, &elem, 1); | |
151 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 483955 times.
|
483955 | av_assert0(ret >= 0); |
152 | 483955 | pthread_cond_broadcast(&tq->cond); | |
153 | } | ||
154 | |||
155 | 483973 | finish: | |
156 | 483973 | pthread_mutex_unlock(&tq->lock); | |
157 | |||
158 | 483973 | return ret; | |
159 | } | ||
160 | |||
161 | 908020 | static int receive_locked(ThreadQueue *tq, int *stream_idx, | |
162 | void *data) | ||
163 | { | ||
164 | FifoElem elem; | ||
165 | 908020 | unsigned int nb_finished = 0; | |
166 | |||
167 |
2/2✓ Branch 1 taken 483955 times.
✓ Branch 2 taken 424065 times.
|
908020 | if (av_fifo_read(tq->fifo, &elem, 1) >= 0) { |
168 | 483955 | tq->obj_move(data, elem.obj); | |
169 | 483955 | objpool_release(tq->obj_pool, &elem.obj); | |
170 | 483955 | *stream_idx = elem.stream_idx; | |
171 | 483955 | return 0; | |
172 | } | ||
173 | |||
174 |
2/2✓ Branch 0 taken 453797 times.
✓ Branch 1 taken 417263 times.
|
871060 | for (unsigned int i = 0; i < tq->nb_streams; i++) { |
175 |
2/2✓ Branch 0 taken 438227 times.
✓ Branch 1 taken 15570 times.
|
453797 | if (!(tq->finished[i] & FINISHED_SEND)) |
176 | 438227 | continue; | |
177 | |||
178 | /* return EOF to the consumer at most once for each stream */ | ||
179 |
2/2✓ Branch 0 taken 6802 times.
✓ Branch 1 taken 8768 times.
|
15570 | if (!(tq->finished[i] & FINISHED_RECV)) { |
180 | 6802 | tq->finished[i] |= FINISHED_RECV; | |
181 | 6802 | *stream_idx = i; | |
182 | 6802 | return AVERROR_EOF; | |
183 | } | ||
184 | |||
185 | 8768 | nb_finished++; | |
186 | } | ||
187 | |||
188 |
2/2✓ Branch 0 taken 6566 times.
✓ Branch 1 taken 410697 times.
|
417263 | return nb_finished == tq->nb_streams ? AVERROR_EOF : AVERROR(EAGAIN); |
189 | } | ||
190 | |||
191 | 497323 | int tq_receive(ThreadQueue *tq, int *stream_idx, void *data) | |
192 | { | ||
193 | int ret; | ||
194 | |||
195 | 497323 | *stream_idx = -1; | |
196 | |||
197 | 497323 | pthread_mutex_lock(&tq->lock); | |
198 | |||
199 | while (1) { | ||
200 | 908020 | ret = receive_locked(tq, stream_idx, data); | |
201 |
2/2✓ Branch 0 taken 410697 times.
✓ Branch 1 taken 497323 times.
|
908020 | if (ret == AVERROR(EAGAIN)) { |
202 | 410697 | pthread_cond_wait(&tq->cond, &tq->lock); | |
203 | 410697 | continue; | |
204 | } | ||
205 | |||
206 | 497323 | break; | |
207 | } | ||
208 | |||
209 |
2/2✓ Branch 0 taken 483955 times.
✓ Branch 1 taken 13368 times.
|
497323 | if (ret == 0) |
210 | 483955 | pthread_cond_broadcast(&tq->cond); | |
211 | |||
212 | 497323 | pthread_mutex_unlock(&tq->lock); | |
213 | |||
214 | 497323 | return ret; | |
215 | } | ||
216 | |||
217 | 18265 | void tq_send_finish(ThreadQueue *tq, unsigned int stream_idx) | |
218 | { | ||
219 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 18265 times.
|
18265 | av_assert0(stream_idx < tq->nb_streams); |
220 | |||
221 | 18265 | pthread_mutex_lock(&tq->lock); | |
222 | |||
223 | /* mark the stream as send-finished; | ||
224 | * next time the consumer thread tries to read this stream it will get | ||
225 | * an EOF and recv-finished flag will be set */ | ||
226 | 18265 | tq->finished[stream_idx] |= FINISHED_SEND; | |
227 | 18265 | pthread_cond_broadcast(&tq->cond); | |
228 | |||
229 | 18265 | pthread_mutex_unlock(&tq->lock); | |
230 | 18265 | } | |
231 | |||
232 | 6858 | void tq_receive_finish(ThreadQueue *tq, unsigned int stream_idx) | |
233 | { | ||
234 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 6858 times.
|
6858 | av_assert0(stream_idx < tq->nb_streams); |
235 | |||
236 | 6858 | pthread_mutex_lock(&tq->lock); | |
237 | |||
238 | /* mark the stream as recv-finished; | ||
239 | * next time the producer thread tries to send for this stream, it will | ||
240 | * get an EOF and send-finished flag will be set */ | ||
241 | 6858 | tq->finished[stream_idx] |= FINISHED_RECV; | |
242 | 6858 | pthread_cond_broadcast(&tq->cond); | |
243 | |||
244 | 6858 | pthread_mutex_unlock(&tq->lock); | |
245 | 6858 | } | |
246 |