FFmpeg coverage

Directory: ../../../ffmpeg/
File: src/fftools/thread_queue.c
Date: 2023-12-04 05:51:44
Exec Total Coverage
Lines: 94 109 86.2%
Functions: 7 7 100.0%
Branches: 37 50 74.0%
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 19361 void tq_free(ThreadQueue **ptq)
56 {
57 19361 ThreadQueue *tq = *ptq;
58
59
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 19361 times.
 19361 if (!tq)
60 ✗ return;
61
62
1/2
✓ Branch 0 taken 19361 times.
✗ Branch 1 not taken.
 19361 if (tq->fifo) {
63 FifoElem elem;
64
2/2
✓ Branch 1 taken 18 times.
✓ Branch 2 taken 19361 times.
 19379 while (av_fifo_read(tq->fifo, &elem, 1) >= 0)
65 18 objpool_release(tq->obj_pool, &elem.obj);
66 }
67 19361 av_fifo_freep2(&tq->fifo);
68
69 19361 objpool_free(&tq->obj_pool);
70
71 19361 av_freep(&tq->finished);
72
73 19361 pthread_cond_destroy(&tq->cond);
74 19361 pthread_mutex_destroy(&tq->lock);
75
76 19361 av_freep(ptq);
77 }
78
79 19361 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 19361 tq = av_mallocz(sizeof(*tq));
86
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 19361 times.
 19361 if (!tq)
87 ✗ return NULL;
88
89 19361 ret = pthread_cond_init(&tq->cond, NULL);
90
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 19361 times.
 19361 if (ret) {
91 ✗ av_freep(&tq);
92 ✗ return NULL;
93 }
94
95 19361 ret = pthread_mutex_init(&tq->lock, NULL);
96
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 19361 times.
 19361 if (ret) {
97 ✗ pthread_cond_destroy(&tq->cond);
98 ✗ av_freep(&tq);
99 ✗ return NULL;
100 }
101
102 19361 tq->finished = av_calloc(nb_streams, sizeof(*tq->finished));
103
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 19361 times.
 19361 if (!tq->finished)
104 ✗ goto fail;
105 19361 tq->nb_streams = nb_streams;
106
107 19361 tq->fifo = av_fifo_alloc2(queue_size, sizeof(FifoElem), 0);
108
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 19361 times.
 19361 if (!tq->fifo)
109 ✗ goto fail;
110
111 19361 tq->obj_pool = obj_pool;
112 19361 tq->obj_move = obj_move;
113
114 19361 return tq;
115 ✗ fail:
116 ✗ tq_free(&tq);
117 ✗ return NULL;
118 }
119
120 1660557 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 1660557 times.
 1660557 av_assert0(stream_idx < tq->nb_streams);
126 1660557 finished = &tq->finished[stream_idx];
127
128 1660557 pthread_mutex_lock(&tq->lock);
129
130
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1660557 times.
 1660557 if (*finished & FINISHED_SEND) {
131 ✗ ret = AVERROR(EINVAL);
132 ✗ goto finish;
133 }
134
135
4/4
✓ Branch 0 taken 1718041 times.
✓ Branch 1 taken 19 times.
✓ Branch 3 taken 57503 times.
✓ Branch 4 taken 1660538 times.
 1718060 while (!(*finished & FINISHED_RECV) && !av_fifo_can_write(tq->fifo))
136 57503 pthread_cond_wait(&tq->cond, &tq->lock);
137
138
2/2
✓ Branch 0 taken 19 times.
✓ Branch 1 taken 1660538 times.
 1660557 if (*finished & FINISHED_RECV) {
139 19 ret = AVERROR_EOF;
140 19 *finished |= FINISHED_SEND;
141 } else {
142 1660538 FifoElem elem = { .stream_idx = stream_idx };
143
144 1660538 ret = objpool_get(tq->obj_pool, &elem.obj);
145
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1660538 times.
 1660538 if (ret < 0)
146 ✗ goto finish;
147
148 1660538 tq->obj_move(elem.obj, data);
149
150 1660538 ret = av_fifo_write(tq->fifo, &elem, 1);
151
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1660538 times.
 1660538 av_assert0(ret >= 0);
152 1660538 pthread_cond_broadcast(&tq->cond);
153 }
154
155 1660557 finish:
156 1660557 pthread_mutex_unlock(&tq->lock);
157
158 1660557 return ret;
159 }
160
161 2780562 static int receive_locked(ThreadQueue *tq, int *stream_idx,
162 void *data)
163 {
164 FifoElem elem;
165 2780562 unsigned int nb_finished = 0;
166
167
2/2
✓ Branch 1 taken 1660520 times.
✓ Branch 2 taken 1120051 times.
 2780571 while (av_fifo_read(tq->fifo, &elem, 1) >= 0) {
168
2/2
✓ Branch 0 taken 9 times.
✓ Branch 1 taken 1660511 times.
 1660520 if (tq->finished[elem.stream_idx] & FINISHED_RECV) {
169 9 objpool_release(tq->obj_pool, &elem.obj);
170 9 continue;
171 }
172
173 1660511 tq->obj_move(data, elem.obj);
174 1660511 objpool_release(tq->obj_pool, &elem.obj);
175 1660511 *stream_idx = elem.stream_idx;
176 1660511 return 0;
177 }
178
179
2/2
✓ Branch 0 taken 1150769 times.
✓ Branch 1 taken 1100440 times.
 2251209 for (unsigned int i = 0; i < tq->nb_streams; i++) {
180
2/2
✓ Branch 0 taken 1122008 times.
✓ Branch 1 taken 28761 times.
 1150769 if (!tq->finished[i])
181 1122008 continue;
182
183 /* return EOF to the consumer at most once for each stream */
184
2/2
✓ Branch 0 taken 19611 times.
✓ Branch 1 taken 9150 times.
 28761 if (!(tq->finished[i] & FINISHED_RECV)) {
185 19611 tq->finished[i] |= FINISHED_RECV;
186 19611 *stream_idx = i;
187 19611 return AVERROR_EOF;
188 }
189
190 9150 nb_finished++;
191 }
192
193
2/2
✓ Branch 0 taken 6669 times.
✓ Branch 1 taken 1093771 times.
 1100440 return nb_finished == tq->nb_streams ? AVERROR_EOF : AVERROR(EAGAIN);
194 }
195
196 1686791 int tq_receive(ThreadQueue *tq, int *stream_idx, void *data)
197 {
198 int ret;
199
200 1686791 *stream_idx = -1;
201
202 1686791 pthread_mutex_lock(&tq->lock);
203
204 1093771 while (1) {
205 2780562 size_t can_read = av_fifo_can_read(tq->fifo);
206
207 2780562 ret = receive_locked(tq, stream_idx, data);
208
209 // signal other threads if the fifo state changed
210
2/2
✓ Branch 1 taken 1660513 times.
✓ Branch 2 taken 1120049 times.
 2780562 if (can_read != av_fifo_can_read(tq->fifo))
211 1660513 pthread_cond_broadcast(&tq->cond);
212
213
2/2
✓ Branch 0 taken 1093771 times.
✓ Branch 1 taken 1686791 times.
 2780562 if (ret == AVERROR(EAGAIN)) {
214 1093771 pthread_cond_wait(&tq->cond, &tq->lock);
215 1093771 continue;
216 }
217
218 1686791 break;
219 }
220
221 1686791 pthread_mutex_unlock(&tq->lock);
222
223 1686791 return ret;
224 }
225
226 37643 void tq_send_finish(ThreadQueue *tq, unsigned int stream_idx)
227 {
228
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 37643 times.
 37643 av_assert0(stream_idx < tq->nb_streams);
229
230 37643 pthread_mutex_lock(&tq->lock);
231
232 /* mark the stream as send-finished;
233 * next time the consumer thread tries to read this stream it will get
234 * an EOF and recv-finished flag will be set */
235 37643 tq->finished[stream_idx] |= FINISHED_SEND;
236 37643 pthread_cond_broadcast(&tq->cond);
237
238 37643 pthread_mutex_unlock(&tq->lock);
239 37643 }
240
241 19648 void tq_receive_finish(ThreadQueue *tq, unsigned int stream_idx)
242 {
243
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 19648 times.
 19648 av_assert0(stream_idx < tq->nb_streams);
244
245 19648 pthread_mutex_lock(&tq->lock);
246
247 /* mark the stream as recv-finished;
248 * next time the producer thread tries to send for this stream, it will
249 * get an EOF and send-finished flag will be set */
250 19648 tq->finished[stream_idx] |= FINISHED_RECV;
251 19648 pthread_cond_broadcast(&tq->cond);
252
253 19648 pthread_mutex_unlock(&tq->lock);
254 19648 }
255