fftools/thread_queue.c - FFmpeg coverage

fftools/thread_queue.c

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 choked;
42			int *finished;
43			unsigned int nb_streams;
44			size_t queue_size;
45
46			enum ThreadQueueType type;
47
48			AVContainerFifo *fifo;
49			AVFifo *fifo_stream_index;
50			size_t *stream_count;
51
52			pthread_mutex_t lock;
53			pthread_cond_t cond_read;
54			pthread_cond_t *cond_write;
55			unsigned int nb_cond_write;
56			};
57
58		33698	void tq_free(ThreadQueue **ptq)
59			{
60		33698	ThreadQueue tq = ptq;
61
62	2/2 ✓ Branch 2 → 3 taken 11 times. ✓ Branch 2 → 4 taken 33687 times.	33698	if (!tq)
63		11	return;
64
65		33687	av_container_fifo_free(&tq->fifo);
66		33687	av_fifo_freep2(&tq->fifo_stream_index);
67		33687	av_freep(&tq->stream_count);
68
69		33687	av_freep(&tq->finished);
70
71	2/2 ✓ Branch 11 → 9 taken 41764 times. ✓ Branch 11 → 12 taken 33687 times.	75451	for (unsigned i = 0; i < tq->nb_cond_write; i++)
72		41764	pthread_cond_destroy(&tq->cond_write[i]);
73		33687	av_freep(&tq->cond_write);
74
75		33687	pthread_cond_destroy(&tq->cond_read);
76		33687	pthread_mutex_destroy(&tq->lock);
77
78		33687	av_freep(ptq);
79			}
80
81		33687	ThreadQueue *tq_alloc(unsigned int nb_streams, size_t queue_size,
82			enum ThreadQueueType type)
83			{
84			ThreadQueue *tq;
85			int ret;
86
87		33687	tq = av_mallocz(sizeof(*tq));
88	1/2 ✗ Branch 3 → 4 not taken. ✓ Branch 3 → 5 taken 33687 times.	33687	if (!tq)
89		✗	return NULL;
90
91		33687	ret = pthread_cond_init(&tq->cond_read, NULL);
92	1/2 ✗ Branch 6 → 7 not taken. ✓ Branch 6 → 9 taken 33687 times.	33687	if (ret) {
93		✗	av_freep(&tq);
94		✗	return NULL;
95			}
96
97		33687	ret = pthread_mutex_init(&tq->lock, NULL);
98	1/2 ✗ Branch 10 → 11 not taken. ✓ Branch 10 → 14 taken 33687 times.	33687	if (ret) {
99		✗	pthread_cond_destroy(&tq->cond_read);
100		✗	av_freep(&tq);
101		✗	return NULL;
102			}
103
104		33687	tq->cond_write = av_calloc(nb_streams, sizeof(*tq->cond_write));
105	1/2 ✗ Branch 15 → 16 not taken. ✓ Branch 15 → 17 taken 33687 times.	33687	if (!tq->cond_write)
106		✗	goto fail;
107	2/2 ✓ Branch 22 → 18 taken 41764 times. ✓ Branch 22 → 23 taken 33687 times.	75451	for (tq->nb_cond_write = 0; tq->nb_cond_write < nb_streams; tq->nb_cond_write++) {
108		41764	ret = pthread_cond_init(&tq->cond_write[tq->nb_cond_write], NULL);
109	1/2 ✗ Branch 19 → 20 not taken. ✓ Branch 19 → 21 taken 41764 times.	41764	if (ret)
110		✗	goto fail;
111			}
112
113		33687	tq->finished = av_calloc(nb_streams, sizeof(*tq->finished));
114	1/2 ✗ Branch 24 → 25 not taken. ✓ Branch 24 → 26 taken 33687 times.	33687	if (!tq->finished)
115		✗	goto fail;
116		33687	tq->nb_streams = nb_streams;
117		33687	tq->queue_size = queue_size;
118		33687	tq->type = type;
119
120		67374	tq->fifo = (type == THREAD_QUEUE_FRAMES) ?
121	2/2 ✓ Branch 26 → 27 taken 17251 times. ✓ Branch 26 → 28 taken 16436 times.	33687	av_container_fifo_alloc_avframe(0) : av_container_fifo_alloc_avpacket(0);
122	1/2 ✗ Branch 29 → 30 not taken. ✓ Branch 29 → 31 taken 33687 times.	33687	if (!tq->fifo)
123		✗	goto fail;
124
125	1/2 ✗ Branch 31 → 32 not taken. ✓ Branch 31 → 34 taken 33687 times.	33687	av_assert0(queue_size);
126	1/2 ✗ Branch 34 → 35 not taken. ✓ Branch 34 → 36 taken 33687 times.	33687	if (nb_streams > SIZE_MAX / queue_size)
127		✗	goto fail; // treat like OOM
128
129		33687	tq->fifo_stream_index = av_fifo_alloc2(queue_size * nb_streams, sizeof(unsigned), 0);
130	1/2 ✗ Branch 37 → 38 not taken. ✓ Branch 37 → 39 taken 33687 times.	33687	if (!tq->fifo_stream_index)
131		✗	goto fail;
132
133		33687	tq->stream_count = av_calloc(nb_streams, sizeof(*tq->stream_count));
134	1/2 ✗ Branch 40 → 41 not taken. ✓ Branch 40 → 42 taken 33687 times.	33687	if (!tq->stream_count)
135		✗	goto fail;
136
137		33687	return tq;
138		✗	fail:
139		✗	tq_free(&tq);
140		✗	return NULL;
141			}
142
143		2796095	static int can_write(ThreadQueue *tq, unsigned int stream_idx)
144			{
145		2796095	return tq->stream_count[stream_idx] < tq->queue_size;
146			}
147
148		2115713	int tq_send(ThreadQueue tq, unsigned int stream_idx, void data)
149			{
150			int *finished;
151			int ret;
152
153	1/2 ✗ Branch 2 → 3 not taken. ✓ Branch 2 → 5 taken 2115713 times.	2115713	av_assert0(stream_idx < tq->nb_streams);
154		2115713	finished = &tq->finished[stream_idx];
155
156		2115713	pthread_mutex_lock(&tq->lock);
157
158	1/2 ✗ Branch 6 → 7 not taken. ✓ Branch 6 → 8 taken 2115713 times.	2115713	if (*finished & FINISHED_SEND) {
159		✗	ret = AVERROR(EINVAL);
160		✗	goto finish;
161			}
162
163	4/4 ✓ Branch 10 → 11 taken 2796095 times. ✓ Branch 10 → 13 taken 7159 times. ✓ Branch 12 → 9 taken 687541 times. ✓ Branch 12 → 13 taken 2108554 times.	2803254	while (!(*finished & FINISHED_RECV) && !can_write(tq, stream_idx))
164		687541	pthread_cond_wait(&tq->cond_write[stream_idx], &tq->lock);
165
166	2/2 ✓ Branch 13 → 14 taken 7159 times. ✓ Branch 13 → 15 taken 2108554 times.	2115713	if (*finished & FINISHED_RECV) {
167		7159	ret = AVERROR_EOF;
168		7159	*finished \|= FINISHED_SEND;
169			} else {
170		2108554	ret = av_fifo_write(tq->fifo_stream_index, &stream_idx, 1);
171	1/2 ✗ Branch 16 → 17 not taken. ✓ Branch 16 → 18 taken 2108554 times.	2108554	if (ret < 0)
172		✗	goto finish;
173
174		2108554	ret = av_container_fifo_write(tq->fifo, data, 0);
175	1/2 ✗ Branch 19 → 20 not taken. ✓ Branch 19 → 21 taken 2108554 times.	2108554	if (ret < 0)
176		✗	goto finish;
177
178		2108554	tq->stream_count[stream_idx]++;
179		2108554	pthread_cond_broadcast(&tq->cond_read); // signal downstream
180			}
181
182		2115713	finish:
183		2115713	pthread_mutex_unlock(&tq->lock);
184
185		2115713	return ret;
186			}
187
188		3330165	static int receive_locked(ThreadQueue tq, int stream_idx,
189			void *data)
190			{
191		3330165	unsigned int nb_finished = 0;
192
193	2/2 ✓ Branch 2 → 3 taken 7750 times. ✓ Branch 2 → 4 taken 3322415 times.	3330165	if (tq->choked)
194		7750	return AVERROR(EAGAIN);
195
196	2/2 ✓ Branch 19 → 5 taken 2095072 times. ✓ Branch 19 → 20 taken 1228030 times.	3323102	while (av_container_fifo_read(tq->fifo, data, 0) >= 0) {
197			unsigned idx;
198			int ret;
199
200		2095072	ret = av_fifo_read(tq->fifo_stream_index, &idx, 1);
201	1/2 ✗ Branch 6 → 7 not taken. ✓ Branch 6 → 9 taken 2095072 times.	2095072	av_assert0(ret >= 0);
202
203			// signal upstream if the fifo is no longer full
204	2/2 ✓ Branch 9 → 10 taken 833565 times. ✓ Branch 9 → 11 taken 1261507 times.	2095072	if (tq->stream_count[idx]-- == tq->queue_size)
205		833565	pthread_cond_broadcast(&tq->cond_write[idx]);
206
207	2/2 ✓ Branch 11 → 12 taken 687 times. ✓ Branch 11 → 16 taken 2094385 times.	2095072	if (tq->finished[idx] & FINISHED_RECV) {
208		687	(tq->type == THREAD_QUEUE_FRAMES) ?
209	2/2 ✓ Branch 12 → 13 taken 557 times. ✓ Branch 12 → 14 taken 130 times.	687	av_frame_unref(data) : av_packet_unref(data);
210		687	continue;
211			}
212
213		2094385	*stream_idx = idx;
214		2094385	return 0;
215			}
216
217	2/2 ✓ Branch 27 → 21 taken 1620916 times. ✓ Branch 27 → 28 taken 1205940 times.	2826856	for (unsigned int i = 0; i < tq->nb_streams; i++) {
218	2/2 ✓ Branch 21 → 22 taken 1576665 times. ✓ Branch 21 → 23 taken 44251 times.	1620916	if (!tq->finished[i])
219		1576665	continue;
220
221			/* return EOF to the consumer at most once for each stream */
222	2/2 ✓ Branch 23 → 24 taken 22090 times. ✓ Branch 23 → 25 taken 22161 times.	44251	if (!(tq->finished[i] & FINISHED_RECV)) {
223		22090	tq->finished[i] \|= FINISHED_RECV;
224		22090	*stream_idx = i;
225		22090	return AVERROR_EOF;
226			}
227
228		22161	nb_finished++;
229			}
230
231	2/2 ✓ Branch 28 → 29 taken 9424 times. ✓ Branch 28 → 30 taken 1196516 times.	1205940	return nb_finished == tq->nb_streams ? AVERROR_EOF : AVERROR(EAGAIN);
232			}
233
234		2159500	int tq_receive(ThreadQueue tq, int stream_idx, void *data, int flags)
235			{
236			int ret;
237
238		2159500	*stream_idx = -1;
239
240		2159500	pthread_mutex_lock(&tq->lock);
241
242			while (1) {
243		3330165	ret = receive_locked(tq, stream_idx, data);
244
245	4/4 ✓ Branch 4 → 5 taken 1204266 times. ✓ Branch 4 → 8 taken 2125899 times. ✓ Branch 5 → 6 taken 1170665 times. ✓ Branch 5 → 8 taken 33601 times.	3330165	if (ret == AVERROR(EAGAIN) && !(flags & THREAD_QUEUE_FLAG_NO_BLOCK)) {
246		1170665	pthread_cond_wait(&tq->cond_read, &tq->lock);
247		1170665	continue;
248			}
249
250		2159500	break;
251			}
252
253		2159500	pthread_mutex_unlock(&tq->lock);
254
255		2159500	return ret;
256			}
257
258		62925	void tq_send_finish(ThreadQueue *tq, unsigned int stream_idx)
259			{
260	1/2 ✗ Branch 2 → 3 not taken. ✓ Branch 2 → 5 taken 62925 times.	62925	av_assert0(stream_idx < tq->nb_streams);
261
262		62925	pthread_mutex_lock(&tq->lock);
263
264			/* mark the stream as send-finished;
265			* next time the consumer thread tries to read this stream it will get
266			* an EOF and recv-finished flag will be set */
267		62925	tq->finished[stream_idx] \|= FINISHED_SEND;
268		62925	tq->choked = 0;
269		62925	pthread_cond_broadcast(&tq->cond_read);
270
271		62925	pthread_mutex_unlock(&tq->lock);
272		62925	}
273
274		50360	void tq_receive_finish(ThreadQueue *tq, unsigned int stream_idx)
275			{
276	1/2 ✗ Branch 2 → 3 not taken. ✓ Branch 2 → 5 taken 50360 times.	50360	av_assert0(stream_idx < tq->nb_streams);
277
278		50360	pthread_mutex_lock(&tq->lock);
279
280			/* mark the stream as recv-finished;
281			* next time the producer thread tries to send for this stream, it will
282			* get an EOF and send-finished flag will be set */
283		50360	tq->finished[stream_idx] \|= FINISHED_RECV;
284		50360	pthread_cond_broadcast(&tq->cond_write[stream_idx]);
285
286		50360	pthread_mutex_unlock(&tq->lock);
287		50360	}
288
289		28323	void tq_choke(ThreadQueue *tq, int choked)
290			{
291		28323	pthread_mutex_lock(&tq->lock);
292
293		28323	int prev_choked = tq->choked;
294		28323	tq->choked = choked;
295	2/2 ✓ Branch 3 → 4 taken 25097 times. ✓ Branch 3 → 5 taken 3226 times.	28323	if (choked != prev_choked)
296		25097	pthread_cond_broadcast(&tq->cond_read);
297
298		28323	pthread_mutex_unlock(&tq->lock);
299		28323	}
300