History

Merge commit '129bb238430ec45a3b5f8f1d384df590ddf7b62f'

* commit '129bb238430ec45a3b5f8f1d384df590ddf7b62f': lavfi: add a slice threading infrastructure Conflicts: Changelog cmdutils.c doc/APIchanges libavfilter/Makefile libavfilter/avfilter.c libavfilter/avfilter.h libavfilter/avfiltergraph.c libavfilter/internal.h libavfilter/version.h

Michael Niedermayer committed 13 years ago

Commit 8d4e9692 parents fe40a9f 129bb2311 files · +447 −14Browse files

Showing the changes to libavfilter/pthread.c (1 of 11 files). Show all

Unified · Split

libavfilter/pthread.c+229−0
@@ -0,0 +1,229 @@
1+/*
2+ *
3+ * This file is part of FFmpeg.
4+ *
5+ * FFmpeg is free software; you can redistribute it and/or
6+ * modify it under the terms of the GNU Lesser General Public
7+ * License as published by the Free Software Foundation; either
8+ * version 2.1 of the License, or (at your option) any later version.
9+ *
10+ * FFmpeg is distributed in the hope that it will be useful,
11+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
12+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13+ * Lesser General Public License for more details.
14+ *
15+ * You should have received a copy of the GNU Lesser General Public
16+ * License along with FFmpeg; if not, write to the Free Software
17+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
18+ */
19+
20+/**
21+ * @file
22+ * Libavfilter multithreading support
23+ */
24+
25+#include "config.h"
26+
27+#include "libavutil/common.h"
28+#include "libavutil/cpu.h"
29+#include "libavutil/mem.h"
30+
31+#include "avfilter.h"
32+#include "internal.h"
33+#include "thread.h"
34+
35+#if HAVE_PTHREADS
36+#include <pthread.h>
37+#elif HAVE_W32THREADS
38+#include "compat/w32pthreads.h"
39+#endif
40+
41+typedef struct ThreadContext {
42+ AVFilterGraph *graph;
43+
44+ int nb_threads;
45+ pthread_t *workers;
46+ action_func *func;
47+
48+ /* per-execute perameters */
49+ AVFilterContext *ctx;
50+ void *arg;
51+ int *rets;
52+ int nb_rets;
53+ int nb_jobs;
54+
55+ pthread_cond_t last_job_cond;
56+ pthread_cond_t current_job_cond;
57+ pthread_mutex_t current_job_lock;
58+ int current_job;
59+ int done;
60+} ThreadContext;
61+
62+static void* attribute_align_arg worker(void *v)
63+{
64+ ThreadContext *c = v;
65+ int our_job = c->nb_jobs;
66+ int nb_threads = c->nb_threads;
67+ int self_id;
68+
69+ pthread_mutex_lock(&c->current_job_lock);
70+ self_id = c->current_job++;
71+ for (;;) {
72+ while (our_job >= c->nb_jobs) {
73+ if (c->current_job == nb_threads + c->nb_jobs)
74+ pthread_cond_signal(&c->last_job_cond);
75+
76+ pthread_cond_wait(&c->current_job_cond, &c->current_job_lock);
77+ our_job = self_id;
78+
79+ if (c->done) {
80+ pthread_mutex_unlock(&c->current_job_lock);
81+ return NULL;
82+ }
83+ }
84+ pthread_mutex_unlock(&c->current_job_lock);
85+
86+ c->rets[our_job % c->nb_rets] = c->func(c->ctx, c->arg, our_job, c->nb_jobs);
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+ pthread_cond_wait(&c->last_job_cond, &c->current_job_lock);
114+ pthread_mutex_unlock(&c->current_job_lock);
115+}
116+
117+static int thread_execute(AVFilterContext *ctx, action_func *func,
118+ void *arg, int *ret, int nb_jobs)
119+{
120+ ThreadContext *c = ctx->graph->internal->thread;
121+ int dummy_ret;
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+ if (ret) {
134+ c->rets = ret;
135+ c->nb_rets = nb_jobs;
136+ } else {
137+ c->rets = &dummy_ret;
138+ c->nb_rets = 1;
139+ }
140+ pthread_cond_broadcast(&c->current_job_cond);
141+
142+ slice_thread_park_workers(c);
143+
144+ return 0;
145+}
146+
147+static int thread_init(ThreadContext *c, int nb_threads)
148+{
149+ int i, ret;
150+
151+ if (!nb_threads) {
152+ int nb_cpus = av_cpu_count();
153+ av_log(c->graph, AV_LOG_DEBUG, "Detected %d logical cores.\n", nb_cpus);
154+ // use number of cores + 1 as thread count if there is more than one
155+ if (nb_cpus > 1)
156+ nb_threads = nb_cpus + 1;
157+ else
158+ nb_threads = 1;
159+ }
160+
161+ if (nb_threads <= 1)
162+ return 1;
163+
164+ c->nb_threads = nb_threads;
165+ c->workers = av_mallocz(sizeof(*c->workers) * nb_threads);
166+ if (!c->workers)
167+ return AVERROR(ENOMEM);
168+
169+ c->current_job = 0;
170+ c->nb_jobs = 0;
171+ c->done = 0;
172+
173+ pthread_cond_init(&c->current_job_cond, NULL);
174+ pthread_cond_init(&c->last_job_cond, NULL);
175+
176+ pthread_mutex_init(&c->current_job_lock, NULL);
177+ pthread_mutex_lock(&c->current_job_lock);
178+ for (i = 0; i < nb_threads; i++) {
179+ ret = pthread_create(&c->workers[i], NULL, worker, c);
180+ if (ret) {
181+ pthread_mutex_unlock(&c->current_job_lock);
182+ c->nb_threads = i;
183+ slice_thread_uninit(c);
184+ return AVERROR(ret);
185+ }
186+ }
187+
188+ slice_thread_park_workers(c);
189+
190+ return c->nb_threads;
191+}
192+
193+int ff_graph_thread_init(AVFilterGraph *graph)
194+{
195+ int ret;
196+
197+#if HAVE_W32THREADS
198+ w32thread_init();
199+#endif
200+
201+ if (graph->nb_threads == 1) {
202+ graph->thread_type = 0;
203+ return 0;
204+ }
205+
206+ graph->internal->thread = av_mallocz(sizeof(ThreadContext));
207+ if (!graph->internal->thread)
208+ return AVERROR(ENOMEM);
209+
210+ ret = thread_init(graph->internal->thread, graph->nb_threads);
211+ if (ret <= 1) {
212+ av_freep(&graph->internal->thread);
213+ graph->thread_type = 0;
214+ graph->nb_threads = 1;
215+ return (ret < 0) ? ret : 0;
216+ }
217+ graph->nb_threads = ret;
218+
219+ graph->internal->thread_execute = thread_execute;
220+
221+ return 0;
222+}
223+
224+void ff_graph_thread_free(AVFilterGraph *graph)
225+{
226+ if (graph->internal->thread)
227+ slice_thread_uninit(graph->internal->thread);
228+ av_freep(&graph->internal->thread);
229+}

Keyboard shortcuts

General

Show keyboard shortcuts
?
Focus the current search
/

Issues and pull requests

Open filters
f

Issues, pull requests and Inbox

Select next item
j↓
Select previous item
k↑
Open selected item
Enter
Dismiss filters or clear selection
Esc

Repository code

Go to file
t