FFmpeg coverage

Directory: ../../../ffmpeg/
File: src/fftools/thread_queue.c
Date: 2025-08-19 23:55:23
Exec Total Coverage
Lines: 94 109 86.2%
Functions: 7 7 100.0%
Branches: 41 54 75.9%
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/container_fifo.h"
24 #include "libavutil/error.h"
25 #include "libavutil/fifo.h"
26 #include "libavutil/frame.h"
27 #include "libavutil/intreadwrite.h"
28 #include "libavutil/mem.h"
29 #include "libavutil/thread.h"
30
31 #include "libavcodec/packet.h"
32
33 #include "thread_queue.h"
34
35 enum {
36 FINISHED_SEND = (1 << 0),
37 FINISHED_RECV = (1 << 1),
38 };
39
40 struct ThreadQueue {
41 int *finished;
42 unsigned int nb_streams;
43
44 enum ThreadQueueType type;
45
46 AVContainerFifo *fifo;
47 AVFifo *fifo_stream_index;
48
49 pthread_mutex_t lock;
50 pthread_cond_t cond;
51 };
52
53 31455 void tq_free(ThreadQueue **ptq)
54 {
55 31455 ThreadQueue *tq = *ptq;
56
57
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 31455 times.
 31455 if (!tq)
58 ✗ return;
59
60 31455 av_container_fifo_free(&tq->fifo);
61 31455 av_fifo_freep2(&tq->fifo_stream_index);
62
63 31455 av_freep(&tq->finished);
64
65 31455 pthread_cond_destroy(&tq->cond);
66 31455 pthread_mutex_destroy(&tq->lock);
67
68 31455 av_freep(ptq);
69 }
70
71 31455 ThreadQueue *tq_alloc(unsigned int nb_streams, size_t queue_size,
72 enum ThreadQueueType type)
73 {
74 ThreadQueue *tq;
75 int ret;
76
77 31455 tq = av_mallocz(sizeof(*tq));
78
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 31455 times.
 31455 if (!tq)
79 ✗ return NULL;
80
81 31455 ret = pthread_cond_init(&tq->cond, NULL);
82
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 31455 times.
 31455 if (ret) {
83 ✗ av_freep(&tq);
84 ✗ return NULL;
85 }
86
87 31455 ret = pthread_mutex_init(&tq->lock, NULL);
88
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 31455 times.
 31455 if (ret) {
89 ✗ pthread_cond_destroy(&tq->cond);
90 ✗ av_freep(&tq);
91 ✗ return NULL;
92 }
93
94 31455 tq->finished = av_calloc(nb_streams, sizeof(*tq->finished));
95
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 31455 times.
 31455 if (!tq->finished)
96 ✗ goto fail;
97 31455 tq->nb_streams = nb_streams;
98
99 31455 tq->type = type;
100
101 31455 tq->fifo = (type == THREAD_QUEUE_FRAMES) ?
102
2/2
✓ Branch 0 taken 16152 times.
✓ Branch 1 taken 15303 times.
 31455 av_container_fifo_alloc_avframe(0) : av_container_fifo_alloc_avpacket(0);
103
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 31455 times.
 31455 if (!tq->fifo)
104 ✗ goto fail;
105
106 31455 tq->fifo_stream_index = av_fifo_alloc2(queue_size, sizeof(unsigned), 0);
107
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 31455 times.
 31455 if (!tq->fifo_stream_index)
108 ✗ goto fail;
109
110 31455 return tq;
111 ✗ fail:
112 ✗ tq_free(&tq);
113 ✗ return NULL;
114 }
115
116 1764783 int tq_send(ThreadQueue *tq, unsigned int stream_idx, void *data)
117 {
118 int *finished;
119 int ret;
120
121
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1764783 times.
 1764783 av_assert0(stream_idx < tq->nb_streams);
122 1764783 finished = &tq->finished[stream_idx];
123
124 1764783 pthread_mutex_lock(&tq->lock);
125
126
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 1764782 times.
 1764783 if (*finished & FINISHED_SEND) {
127 1 ret = AVERROR(EINVAL);
128 1 goto finish;
129 }
130
131
4/4
✓ Branch 0 taken 2181046 times.
✓ Branch 1 taken 6898 times.
✓ Branch 3 taken 423162 times.
✓ Branch 4 taken 1757884 times.
 2187944 while (!(*finished & FINISHED_RECV) && !av_fifo_can_write(tq->fifo_stream_index))
132 423162 pthread_cond_wait(&tq->cond, &tq->lock);
133
134
2/2
✓ Branch 0 taken 6898 times.
✓ Branch 1 taken 1757884 times.
 1764782 if (*finished & FINISHED_RECV) {
135 6898 ret = AVERROR_EOF;
136 6898 *finished |= FINISHED_SEND;
137 } else {
138 1757884 ret = av_fifo_write(tq->fifo_stream_index, &stream_idx, 1);
139
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1757884 times.
 1757884 if (ret < 0)
140 ✗ goto finish;
141
142 1757884 ret = av_container_fifo_write(tq->fifo, data, 0);
143
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1757884 times.
 1757884 if (ret < 0)
144 ✗ goto finish;
145
146 1757884 pthread_cond_broadcast(&tq->cond);
147 }
148
149 1764783 finish:
150 1764783 pthread_mutex_unlock(&tq->lock);
151
152 1764783 return ret;
153 }
154
155 2534334 static int receive_locked(ThreadQueue *tq, int *stream_idx,
156 void *data)
157 {
158 2534334 unsigned int nb_finished = 0;
159
160
2/2
✓ Branch 1 taken 1702358 times.
✓ Branch 2 taken 832154 times.
 2534512 while (av_container_fifo_read(tq->fifo, data, 0) >= 0) {
161 unsigned idx;
162 int ret;
163
164 1702358 ret = av_fifo_read(tq->fifo_stream_index, &idx, 1);
165
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1702358 times.
 1702358 av_assert0(ret >= 0);
166
2/2
✓ Branch 0 taken 178 times.
✓ Branch 1 taken 1702180 times.
 1702358 if (tq->finished[idx] & FINISHED_RECV) {
167 178 (tq->type == THREAD_QUEUE_FRAMES) ?
168
2/2
✓ Branch 0 taken 11 times.
✓ Branch 1 taken 167 times.
 178 av_frame_unref(data) : av_packet_unref(data);
169 178 continue;
170 }
171
172 1702180 *stream_idx = idx;
173 1702180 return 0;
174 }
175
176
2/2
✓ Branch 0 taken 1041418 times.
✓ Branch 1 taken 811800 times.
 1853218 for (unsigned int i = 0; i < tq->nb_streams; i++) {
177
2/2
✓ Branch 0 taken 1005308 times.
✓ Branch 1 taken 36110 times.
 1041418 if (!tq->finished[i])
178 1005308 continue;
179
180 /* return EOF to the consumer at most once for each stream */
181
2/2
✓ Branch 0 taken 20354 times.
✓ Branch 1 taken 15756 times.
 36110 if (!(tq->finished[i] & FINISHED_RECV)) {
182 20354 tq->finished[i] |= FINISHED_RECV;
183 20354 *stream_idx = i;
184 20354 return AVERROR_EOF;
185 }
186
187 15756 nb_finished++;
188 }
189
190
2/2
✓ Branch 0 taken 8389 times.
✓ Branch 1 taken 803411 times.
 811800 return nb_finished == tq->nb_streams ? AVERROR_EOF : AVERROR(EAGAIN);
191 }
192
193 1730923 int tq_receive(ThreadQueue *tq, int *stream_idx, void *data)
194 {
195 int ret;
196
197 1730923 *stream_idx = -1;
198
199 1730923 pthread_mutex_lock(&tq->lock);
200
201 803411 while (1) {
202 2534334 size_t can_read = av_container_fifo_can_read(tq->fifo);
203
204 2534334 ret = receive_locked(tq, stream_idx, data);
205
206 // signal other threads if the fifo state changed
207
2/2
✓ Branch 1 taken 1702211 times.
✓ Branch 2 taken 832123 times.
 2534334 if (can_read != av_container_fifo_can_read(tq->fifo))
208 1702211 pthread_cond_broadcast(&tq->cond);
209
210
2/2
✓ Branch 0 taken 803411 times.
✓ Branch 1 taken 1730923 times.
 2534334 if (ret == AVERROR(EAGAIN)) {
211 803411 pthread_cond_wait(&tq->cond, &tq->lock);
212 803411 continue;
213 }
214
215 1730923 break;
216 }
217
218 1730923 pthread_mutex_unlock(&tq->lock);
219
220 1730923 return ret;
221 }
222
223 46174 void tq_send_finish(ThreadQueue *tq, unsigned int stream_idx)
224 {
225
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 46174 times.
 46174 av_assert0(stream_idx < tq->nb_streams);
226
227 46174 pthread_mutex_lock(&tq->lock);
228
229 /* mark the stream as send-finished;
230 * next time the consumer thread tries to read this stream it will get
231 * an EOF and recv-finished flag will be set */
232 46174 tq->finished[stream_idx] |= FINISHED_SEND;
233 46174 pthread_cond_broadcast(&tq->cond);
234
235 46174 pthread_mutex_unlock(&tq->lock);
236 46174 }
237
238 39087 void tq_receive_finish(ThreadQueue *tq, unsigned int stream_idx)
239 {
240
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 39087 times.
 39087 av_assert0(stream_idx < tq->nb_streams);
241
242 39087 pthread_mutex_lock(&tq->lock);
243
244 /* mark the stream as recv-finished;
245 * next time the producer thread tries to send for this stream, it will
246 * get an EOF and send-finished flag will be set */
247 39087 tq->finished[stream_idx] |= FINISHED_RECV;
248 39087 pthread_cond_broadcast(&tq->cond);
249
250 39087 pthread_mutex_unlock(&tq->lock);
251 39087 }
252