path: root/libavfilter/pthread.c (plain)
blob: c7a00210d6648379547726344faf20852a0ce496

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	/**
20	* @file
21	* Libavfilter multithreading support
22	*/
23
24	#include "config.h"
25
26	#include "libavutil/common.h"
27	#include "libavutil/cpu.h"
28	#include "libavutil/mem.h"
29	#include "libavutil/thread.h"
30
31	#include "avfilter.h"
32	#include "internal.h"
33	#include "thread.h"
34
35	typedef struct ThreadContext {
36	AVFilterGraph *graph;
37
38	int nb_threads;
39	pthread_t *workers;
40	avfilter_action_func *func;
41
42	/* per-execute parameters */
43	AVFilterContext *ctx;
44	void *arg;
45	int *rets;
46	int nb_jobs;
47
48	pthread_cond_t last_job_cond;
49	pthread_cond_t current_job_cond;
50	pthread_mutex_t current_job_lock;
51	int current_job;
52	unsigned int current_execute;
53	int done;
54	} ThreadContext;
55
56	static void* attribute_align_arg worker(void *v)
57	{
58	ThreadContext *c = v;
59	int our_job = c->nb_jobs;
60	int nb_threads = c->nb_threads;
61	unsigned int last_execute = 0;
62	int ret, self_id;
63
64	pthread_mutex_lock(&c->current_job_lock);
65	self_id = c->current_job++;
66
67	for (;;) {
68	while (our_job >= c->nb_jobs) {
69	if (c->current_job == nb_threads + c->nb_jobs)
70	pthread_cond_signal(&c->last_job_cond);
71
72	while (last_execute == c->current_execute && !c->done)
73	pthread_cond_wait(&c->current_job_cond, &c->current_job_lock);
74	last_execute = c->current_execute;
75	our_job = self_id;
76
77	if (c->done) {
78	pthread_mutex_unlock(&c->current_job_lock);
79	return NULL;
80	}
81	}
82	pthread_mutex_unlock(&c->current_job_lock);
83
84	ret = c->func(c->ctx, c->arg, our_job, c->nb_jobs);
85	if (c->rets)
86	c->rets[our_job % c->nb_jobs] = ret;
87
88	pthread_mutex_lock(&c->current_job_lock);
89	our_job = c->current_job++;
90	}
91	}
92
93	static void slice_thread_uninit(ThreadContext *c)
94	{
95	int i;
96
97	pthread_mutex_lock(&c->current_job_lock);
98	c->done = 1;
99	pthread_cond_broadcast(&c->current_job_cond);
100	pthread_mutex_unlock(&c->current_job_lock);
101
102	for (i = 0; i < c->nb_threads; i++)
103	pthread_join(c->workers[i], NULL);
104
105	pthread_mutex_destroy(&c->current_job_lock);
106	pthread_cond_destroy(&c->current_job_cond);
107	pthread_cond_destroy(&c->last_job_cond);
108	av_freep(&c->workers);
109	}
110
111	static void slice_thread_park_workers(ThreadContext *c)
112	{
113	while (c->current_job != c->nb_threads + c->nb_jobs)
114	pthread_cond_wait(&c->last_job_cond, &c->current_job_lock);
115	pthread_mutex_unlock(&c->current_job_lock);
116	}
117
118	static int thread_execute(AVFilterContext *ctx, avfilter_action_func *func,
119	void *arg, int *ret, int nb_jobs)
120	{
121	ThreadContext *c = ctx->graph->internal->thread;
122
123	if (nb_jobs <= 0)
124	return 0;
125
126	pthread_mutex_lock(&c->current_job_lock);
127
128	c->current_job = c->nb_threads;
129	c->nb_jobs = nb_jobs;
130	c->ctx = ctx;
131	c->arg = arg;
132	c->func = func;
133	c->rets = ret;
134	c->current_execute++;
135
136	pthread_cond_broadcast(&c->current_job_cond);
137
138	slice_thread_park_workers(c);
139
140	return 0;
141	}
142
143	static int thread_init_internal(ThreadContext *c, int nb_threads)
144	{
145	int i, ret;
146
147	if (!nb_threads) {
148	int nb_cpus = av_cpu_count();
149	// use number of cores + 1 as thread count if there is more than one
150	if (nb_cpus > 1)
151	nb_threads = nb_cpus + 1;
152	else
153	nb_threads = 1;
154	}
155
156	if (nb_threads <= 1)
157	return 1;
158
159	c->nb_threads = nb_threads;
160	c->workers = av_mallocz_array(sizeof(*c->workers), nb_threads);
161	if (!c->workers)
162	return AVERROR(ENOMEM);
163
164	c->current_job = 0;
165	c->nb_jobs = 0;
166	c->done = 0;
167
168	pthread_cond_init(&c->current_job_cond, NULL);
169	pthread_cond_init(&c->last_job_cond, NULL);
170
171	pthread_mutex_init(&c->current_job_lock, NULL);
172	pthread_mutex_lock(&c->current_job_lock);
173	for (i = 0; i < nb_threads; i++) {
174	ret = pthread_create(&c->workers[i], NULL, worker, c);
175	if (ret) {
176	pthread_mutex_unlock(&c->current_job_lock);
177	c->nb_threads = i;
178	slice_thread_uninit(c);
179	return AVERROR(ret);
180	}
181	}
182
183	slice_thread_park_workers(c);
184
185	return c->nb_threads;
186	}
187
188	int ff_graph_thread_init(AVFilterGraph *graph)
189	{
190	int ret;
191
192	#if HAVE_W32THREADS
193	w32thread_init();
194	#endif
195
196	if (graph->nb_threads == 1) {
197	graph->thread_type = 0;
198	return 0;
199	}
200
201	graph->internal->thread = av_mallocz(sizeof(ThreadContext));
202	if (!graph->internal->thread)
203	return AVERROR(ENOMEM);
204
205	ret = thread_init_internal(graph->internal->thread, graph->nb_threads);
206	if (ret <= 1) {
207	av_freep(&graph->internal->thread);
208	graph->thread_type = 0;
209	graph->nb_threads = 1;
210	return (ret < 0) ? ret : 0;
211	}
212	graph->nb_threads = ret;
213
214	graph->internal->thread_execute = thread_execute;
215
216	return 0;
217	}
218
219	void ff_graph_thread_free(AVFilterGraph *graph)
220	{
221	if (graph->internal->thread)
222	slice_thread_uninit(graph->internal->thread);
223	av_freep(&graph->internal->thread);
224	}
225