FFmpeg
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros Groups Pages
slicethread.c
Go to the documentation of this file.
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 <stdatomic.h>
20 #include "slicethread.h"
21 #include "mem.h"
22 #include "thread.h"
23 #include "avassert.h"
24 
25 #if HAVE_PTHREADS || HAVE_W32THREADS || HAVE_OS2THREADS
26 
27 typedef struct WorkerContext {
31  pthread_t thread;
32  int done;
33 } WorkerContext;
34 
35 struct AVSliceThread {
36  WorkerContext *workers;
37  int nb_threads;
38  int nb_active_threads;
39  int nb_jobs;
40 
41  atomic_uint first_job;
42  atomic_uint current_job;
43  pthread_mutex_t done_mutex;
44  pthread_cond_t done_cond;
45  int done;
46  int finished;
47 
48  void *priv;
49  void (*worker_func)(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads);
50  void (*main_func)(void *priv);
51 };
52 
53 static int run_jobs(AVSliceThread *ctx)
54 {
55  unsigned nb_jobs = ctx->nb_jobs;
56  unsigned nb_active_threads = ctx->nb_active_threads;
57  unsigned first_job = atomic_fetch_add_explicit(&ctx->first_job, 1, memory_order_acq_rel);
58  unsigned current_job = first_job;
59 
60  do {
61  ctx->worker_func(ctx->priv, current_job, first_job, nb_jobs, nb_active_threads);
62  } while ((current_job = atomic_fetch_add_explicit(&ctx->current_job, 1, memory_order_acq_rel)) < nb_jobs);
63 
64  return current_job == nb_jobs + nb_active_threads - 1;
65 }
66 
67 static void *attribute_align_arg thread_worker(void *v)
68 {
69  WorkerContext *w = v;
70  AVSliceThread *ctx = w->ctx;
71 
72  pthread_mutex_lock(&w->mutex);
73  pthread_cond_signal(&w->cond);
74 
75  while (1) {
76  w->done = 1;
77  while (w->done)
78  pthread_cond_wait(&w->cond, &w->mutex);
79 
80  if (ctx->finished) {
81  pthread_mutex_unlock(&w->mutex);
82  return NULL;
83  }
84 
85  if (run_jobs(ctx)) {
86  pthread_mutex_lock(&ctx->done_mutex);
87  ctx->done = 1;
88  pthread_cond_signal(&ctx->done_cond);
89  pthread_mutex_unlock(&ctx->done_mutex);
90  }
91  }
92 }
93 
94 int avpriv_slicethread_create(AVSliceThread **pctx, void *priv,
95  void (*worker_func)(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads),
96  void (*main_func)(void *priv),
97  int nb_threads)
98 {
100  int nb_workers, i;
101 
102 #if HAVE_W32THREADS
103  w32thread_init();
104 #endif
105 
106  av_assert0(nb_threads >= 0);
107  if (!nb_threads) {
108  int nb_cpus = av_cpu_count();
109  if (nb_cpus > 1)
110  nb_threads = nb_cpus + 1;
111  else
112  nb_threads = 1;
113  }
114 
115  nb_workers = nb_threads;
116  if (!main_func)
117  nb_workers--;
118 
119  *pctx = ctx = av_mallocz(sizeof(*ctx));
120  if (!ctx)
121  return AVERROR(ENOMEM);
122 
123  if (nb_workers && !(ctx->workers = av_calloc(nb_workers, sizeof(*ctx->workers)))) {
124  av_freep(pctx);
125  return AVERROR(ENOMEM);
126  }
127 
128  ctx->priv = priv;
129  ctx->worker_func = worker_func;
130  ctx->main_func = main_func;
131  ctx->nb_threads = nb_threads;
132  ctx->nb_active_threads = 0;
133  ctx->nb_jobs = 0;
134  ctx->finished = 0;
135 
136  atomic_init(&ctx->first_job, 0);
137  atomic_init(&ctx->current_job, 0);
138  pthread_mutex_init(&ctx->done_mutex, NULL);
139  pthread_cond_init(&ctx->done_cond, NULL);
140  ctx->done = 0;
141 
142  for (i = 0; i < nb_workers; i++) {
143  WorkerContext *w = &ctx->workers[i];
144  int ret;
145  w->ctx = ctx;
146  pthread_mutex_init(&w->mutex, NULL);
147  pthread_cond_init(&w->cond, NULL);
148  pthread_mutex_lock(&w->mutex);
149  w->done = 0;
150 
151  if (ret = pthread_create(&w->thread, NULL, thread_worker, w)) {
152  ctx->nb_threads = main_func ? i : i + 1;
153  pthread_mutex_unlock(&w->mutex);
154  pthread_cond_destroy(&w->cond);
155  pthread_mutex_destroy(&w->mutex);
157  return AVERROR(ret);
158  }
159 
160  while (!w->done)
161  pthread_cond_wait(&w->cond, &w->mutex);
162  pthread_mutex_unlock(&w->mutex);
163  }
164 
165  return nb_threads;
166 }
167 
168 void avpriv_slicethread_execute(AVSliceThread *ctx, int nb_jobs, int execute_main)
169 {
170  int nb_workers, i, is_last = 0;
171 
172  av_assert0(nb_jobs > 0);
173  ctx->nb_jobs = nb_jobs;
174  ctx->nb_active_threads = FFMIN(nb_jobs, ctx->nb_threads);
175  atomic_store_explicit(&ctx->first_job, 0, memory_order_relaxed);
176  atomic_store_explicit(&ctx->current_job, ctx->nb_active_threads, memory_order_relaxed);
177  nb_workers = ctx->nb_active_threads;
178  if (!ctx->main_func || !execute_main)
179  nb_workers--;
180 
181  for (i = 0; i < nb_workers; i++) {
182  WorkerContext *w = &ctx->workers[i];
183  pthread_mutex_lock(&w->mutex);
184  w->done = 0;
185  pthread_cond_signal(&w->cond);
186  pthread_mutex_unlock(&w->mutex);
187  }
188 
189  if (ctx->main_func && execute_main)
190  ctx->main_func(ctx->priv);
191  else
192  is_last = run_jobs(ctx);
193 
194  if (!is_last) {
195  pthread_mutex_lock(&ctx->done_mutex);
196  while (!ctx->done)
197  pthread_cond_wait(&ctx->done_cond, &ctx->done_mutex);
198  ctx->done = 0;
199  pthread_mutex_unlock(&ctx->done_mutex);
200  }
201 }
202 
204 {
206  int nb_workers, i;
207 
208  if (!pctx || !*pctx)
209  return;
210 
211  ctx = *pctx;
212  nb_workers = ctx->nb_threads;
213  if (!ctx->main_func)
214  nb_workers--;
215 
216  ctx->finished = 1;
217  for (i = 0; i < nb_workers; i++) {
218  WorkerContext *w = &ctx->workers[i];
219  pthread_mutex_lock(&w->mutex);
220  w->done = 0;
221  pthread_cond_signal(&w->cond);
222  pthread_mutex_unlock(&w->mutex);
223  }
224 
225  for (i = 0; i < nb_workers; i++) {
226  WorkerContext *w = &ctx->workers[i];
227  pthread_join(w->thread, NULL);
228  pthread_cond_destroy(&w->cond);
229  pthread_mutex_destroy(&w->mutex);
230  }
231 
232  pthread_cond_destroy(&ctx->done_cond);
233  pthread_mutex_destroy(&ctx->done_mutex);
234  av_freep(&ctx->workers);
235  av_freep(pctx);
236 }
237 
238 #else /* HAVE_PTHREADS || HAVE_W32THREADS || HAVE_OS32THREADS */
239 
241  void (*worker_func)(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads),
242  void (*main_func)(void *priv),
243  int nb_threads)
244 {
245  *pctx = NULL;
246  return AVERROR(EINVAL);
247 }
248 
249 void avpriv_slicethread_execute(AVSliceThread *ctx, int nb_jobs, int execute_main)
250 {
251  av_assert0(0);
252 }
253 
255 {
256  av_assert0(!pctx || !*pctx);
257 }
258 
259 #endif /* HAVE_PTHREADS || HAVE_W32THREADS || HAVE_OS32THREADS */
static av_unused void w32thread_init(void)
Definition: w32pthreads.h:397
#define NULL
Definition: coverity.c:32
static av_always_inline int pthread_mutex_destroy(pthread_mutex_t *mutex)
Definition: os2threads.h:106
#define pthread_mutex_lock(a)
Definition: ffprobe.c:61
static av_always_inline int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex)
Definition: os2threads.h:164
int av_cpu_count(void)
Definition: cpu.c:263
Memory handling functions.
static void worker_func(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads)
Definition: pthread_slice.c:65
void * av_mallocz(size_t size)
Allocate a memory block with alignment suitable for all memory accesses (including vectors if availab...
Definition: mem.c:222
static av_always_inline int pthread_cond_destroy(pthread_cond_t *cond)
Definition: os2threads.h:138
void * av_calloc(size_t nmemb, size_t size)
Non-inlined equivalent of av_mallocz_array().
Definition: mem.c:230
int(* cond)(enum AVPixelFormat pix_fmt)
Definition: pixdesc_query.c:28
#define av_assert0(cond)
assert() equivalent, that is always enabled.
Definition: avassert.h:37
struct AVSliceThread AVSliceThread
Definition: slicethread.h:22
HMTX pthread_mutex_t
Definition: os2threads.h:49
static av_always_inline int pthread_cond_signal(pthread_cond_t *cond)
Definition: os2threads.h:146
#define AVERROR(e)
Definition: error.h:43
simple assert() macros that are a bit more flexible than ISO C assert().
#define FFMIN(a, b)
Definition: common.h:96
void avpriv_slicethread_free(AVSliceThread **pctx)
Destroy slice threading context.
Definition: slicethread.c:254
typedef void(APIENTRY *FF_PFNGLACTIVETEXTUREPROC)(GLenum texture)
AVFormatContext * ctx
Definition: movenc.c:48
static av_always_inline int pthread_join(pthread_t thread, void **value_ptr)
Definition: os2threads.h:88
static av_always_inline int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutexattr_t *attr)
Definition: os2threads.h:98
#define pthread_mutex_unlock(a)
Definition: ffprobe.c:65
static pthread_mutex_t * mutex
Definition: w32pthreads.h:258
#define atomic_fetch_add_explicit(object, operand, order)
Definition: stdatomic.h:149
#define attribute_align_arg
Definition: internal.h:61
static av_always_inline int pthread_create(pthread_t *thread, const pthread_attr_t *attr, void *(*start_routine)(void *), void *arg)
Definition: os2threads.h:74
int avpriv_slicethread_create(AVSliceThread **pctx, void *priv, void(*worker_func)(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads), void(*main_func)(void *priv), int nb_threads)
Create slice threading context.
Definition: slicethread.c:240
void avpriv_slicethread_execute(AVSliceThread *ctx, int nb_jobs, int execute_main)
Execute slice threading.
Definition: slicethread.c:249
intptr_t atomic_uint
Definition: stdatomic.h:56
#define atomic_store_explicit(object, desired, order)
Definition: stdatomic.h:90
int( main_func)(AVCodecContext *c)
Definition: pthread_slice.c:41
static av_always_inline int pthread_cond_init(pthread_cond_t *cond, const pthread_condattr_t *attr)
Definition: os2threads.h:127
#define atomic_init(obj, value)
Definition: stdatomic.h:33
#define av_freep(p)