FFmpeg coverage

Line	Branch	Exec	Source
1			/*
2			* Copyright (C) 2023 Nuo Mi
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
8			* License 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 GNU
14			* Lesser General Public License for more details.
15			*
16			* You should have received a copy of the GNU Lesser General Public
17			* License along with FFmpeg; if not, write to the Free Software
18			* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
19			*/
20
21			#include "config.h"
22
23			#include <stdbool.h>
24
25			#include "mem.h"
26			#include "thread.h"
27
28			#include "executor.h"
29
30			#if !HAVE_THREADS
31
32			#define ExecutorThread char
33
34			#define executor_thread_create(t, a, s, ar) 0
35			#define executor_thread_join(t, r) do {} while(0)
36
37			#else
38
39			#define ExecutorThread pthread_t
40
41			#define executor_thread_create(t, a, s, ar) pthread_create(t, a, s, ar)
42			#define executor_thread_join(t, r) pthread_join(t, r)
43
44			#endif //!HAVE_THREADS
45
46			typedef struct ThreadInfo {
47			AVExecutor *e;
48			ExecutorThread thread;
49			} ThreadInfo;
50
51			struct AVExecutor {
52			AVTaskCallbacks cb;
53			int thread_count;
54			bool recursive;
55
56			ThreadInfo *threads;
57			uint8_t *local_contexts;
58
59			AVMutex lock;
60			AVCond cond;
61			int die;
62
63			AVTask *tasks;
64			};
65
66		✗	static AVTask* remove_task(AVTask **prev, AVTask *t)
67			{
68		✗	*prev = t->next;
69		✗	t->next = NULL;
70		✗	return t;
71			}
72
73		✗	static void add_task(AVTask **prev, AVTask *t)
74			{
75		✗	t->next = *prev;
76		✗	*prev = t;
77		✗	}
78
79		✗	static int run_one_task(AVExecutor *e, void *lc)
80			{
81		✗	AVTaskCallbacks *cb = &e->cb;
82			AVTask **prev;
83
84		✗	for (prev = &e->tasks; *prev && !cb->ready(*prev, cb->user_data); prev = &(*prev)->next)
85			/* nothing */;
86		✗	if (*prev) {
87		✗	AVTask *t = remove_task(prev, *prev);
88		✗	if (e->thread_count > 0)
89		✗	ff_mutex_unlock(&e->lock);
90		✗	cb->run(t, lc, cb->user_data);
91		✗	if (e->thread_count > 0)
92		✗	ff_mutex_lock(&e->lock);
93		✗	return 1;
94			}
95		✗	return 0;
96			}
97
98			#if HAVE_THREADS
99		✗	static void *executor_worker_task(void *data)
100			{
101		✗	ThreadInfo *ti = (ThreadInfo*)data;
102		✗	AVExecutor *e = ti->e;
103		✗	void *lc = e->local_contexts + (ti - e->threads) * e->cb.local_context_size;
104
105		✗	ff_mutex_lock(&e->lock);
106			while (1) {
107		✗	if (e->die) break;
108
109		✗	if (!run_one_task(e, lc)) {
110			//no task in one loop
111		✗	ff_cond_wait(&e->cond, &e->lock);
112			}
113			}
114		✗	ff_mutex_unlock(&e->lock);
115		✗	return NULL;
116			}
117			#endif
118
119		✗	static void executor_free(AVExecutor *e, const int has_lock, const int has_cond)
120			{
121		✗	if (e->thread_count) {
122			//signal die
123		✗	ff_mutex_lock(&e->lock);
124		✗	e->die = 1;
125		✗	ff_cond_broadcast(&e->cond);
126		✗	ff_mutex_unlock(&e->lock);
127
128		✗	for (int i = 0; i < e->thread_count; i++)
129		✗	executor_thread_join(e->threads[i].thread, NULL);
130			}
131		✗	if (has_cond)
132		✗	ff_cond_destroy(&e->cond);
133		✗	if (has_lock)
134		✗	ff_mutex_destroy(&e->lock);
135
136		✗	av_free(e->threads);
137		✗	av_free(e->local_contexts);
138
139		✗	av_free(e);
140		✗	}
141
142		✗	AVExecutor* av_executor_alloc(const AVTaskCallbacks *cb, int thread_count)
143			{
144			AVExecutor *e;
145		✗	int has_lock = 0, has_cond = 0;
146		✗	if (!cb || !cb->user_data || !cb->ready || !cb->run || !cb->priority_higher)
147		✗	return NULL;
148
149		✗	e = av_mallocz(sizeof(*e));
150		✗	if (!e)
151		✗	return NULL;
152		✗	e->cb = *cb;
153
154		✗	e->local_contexts = av_calloc(FFMAX(thread_count, 1), e->cb.local_context_size);
155		✗	if (!e->local_contexts)
156		✗	goto free_executor;
157
158		✗	e->threads = av_calloc(FFMAX(thread_count, 1), sizeof(*e->threads));
159		✗	if (!e->threads)
160		✗	goto free_executor;
161
162		✗	if (!thread_count)
163		✗	return e;
164
165		✗	has_lock = !ff_mutex_init(&e->lock, NULL);
166		✗	has_cond = !ff_cond_init(&e->cond, NULL);
167
168		✗	if (!has_lock || !has_cond)
169		✗	goto free_executor;
170
171		✗	for (/* nothing */; e->thread_count < thread_count; e->thread_count++) {
172		✗	ThreadInfo *ti = e->threads + e->thread_count;
173		✗	ti->e = e;
174		✗	if (executor_thread_create(&ti->thread, NULL, executor_worker_task, ti))
175		✗	goto free_executor;
176			}
177		✗	return e;
178
179		✗	free_executor:
180		✗	executor_free(e, has_lock, has_cond);
181		✗	return NULL;
182			}
183
184		✗	void av_executor_free(AVExecutor **executor)
185			{
186			int thread_count;
187
188		✗	if (!executor || !*executor)
189		✗	return;
190		✗	thread_count = (*executor)->thread_count;
191		✗	executor_free(*executor, thread_count, thread_count);
192		✗	*executor = NULL;
193			}
194
195		✗	void av_executor_execute(AVExecutor *e, AVTask *t)
196			{
197		✗	AVTaskCallbacks *cb = &e->cb;
198			AVTask **prev;
199
200		✗	if (e->thread_count)
201		✗	ff_mutex_lock(&e->lock);
202		✗	if (t) {
203		✗	for (prev = &e->tasks; *prev && cb->priority_higher(*prev, t); prev = &(*prev)->next)
204			/* nothing */;
205		✗	add_task(prev, t);
206			}
207		✗	if (e->thread_count) {
208		✗	ff_cond_signal(&e->cond);
209		✗	ff_mutex_unlock(&e->lock);
210			}
211
212		✗	if (!e->thread_count || !HAVE_THREADS) {
213		✗	if (e->recursive)
214		✗	return;
215		✗	e->recursive = true;
216			// We are running in a single-threaded environment, so we must handle all tasks ourselves
217		✗	while (run_one_task(e, e->local_contexts))
218			/* nothing */;
219		✗	e->recursive = false;
220			}
221			}
222