FFmpeg
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  av_assert0(nb_threads >= 0);
103  if (!nb_threads) {
104  int nb_cpus = av_cpu_count();
105  if (nb_cpus > 1)
106  nb_threads = nb_cpus + 1;
107  else
108  nb_threads = 1;
109  }
110 
111  nb_workers = nb_threads;
112  if (!main_func)
113  nb_workers--;
114 
115  *pctx = ctx = av_mallocz(sizeof(*ctx));
116  if (!ctx)
117  return AVERROR(ENOMEM);
118 
119  if (nb_workers && !(ctx->workers = av_calloc(nb_workers, sizeof(*ctx->workers)))) {
120  av_freep(pctx);
121  return AVERROR(ENOMEM);
122  }
123 
124  ctx->priv = priv;
125  ctx->worker_func = worker_func;
126  ctx->main_func = main_func;
127  ctx->nb_threads = nb_threads;
128  ctx->nb_active_threads = 0;
129  ctx->nb_jobs = 0;
130  ctx->finished = 0;
131 
132  atomic_init(&ctx->first_job, 0);
133  atomic_init(&ctx->current_job, 0);
134  pthread_mutex_init(&ctx->done_mutex, NULL);
135  pthread_cond_init(&ctx->done_cond, NULL);
136  ctx->done = 0;
137 
138  for (i = 0; i < nb_workers; i++) {
139  WorkerContext *w = &ctx->workers[i];
140  int ret;
141  w->ctx = ctx;
142  pthread_mutex_init(&w->mutex, NULL);
143  pthread_cond_init(&w->cond, NULL);
144  pthread_mutex_lock(&w->mutex);
145  w->done = 0;
146 
147  if (ret = pthread_create(&w->thread, NULL, thread_worker, w)) {
148  ctx->nb_threads = main_func ? i : i + 1;
149  pthread_mutex_unlock(&w->mutex);
150  pthread_cond_destroy(&w->cond);
151  pthread_mutex_destroy(&w->mutex);
153  return AVERROR(ret);
154  }
155 
156  while (!w->done)
157  pthread_cond_wait(&w->cond, &w->mutex);
158  pthread_mutex_unlock(&w->mutex);
159  }
160 
161  return nb_threads;
162 }
163 
164 void avpriv_slicethread_execute(AVSliceThread *ctx, int nb_jobs, int execute_main)
165 {
166  int nb_workers, i, is_last = 0;
167 
168  av_assert0(nb_jobs > 0);
169  ctx->nb_jobs = nb_jobs;
170  ctx->nb_active_threads = FFMIN(nb_jobs, ctx->nb_threads);
171  atomic_store_explicit(&ctx->first_job, 0, memory_order_relaxed);
172  atomic_store_explicit(&ctx->current_job, ctx->nb_active_threads, memory_order_relaxed);
173  nb_workers = ctx->nb_active_threads;
174  if (!ctx->main_func || !execute_main)
175  nb_workers--;
176 
177  for (i = 0; i < nb_workers; i++) {
178  WorkerContext *w = &ctx->workers[i];
179  pthread_mutex_lock(&w->mutex);
180  w->done = 0;
181  pthread_cond_signal(&w->cond);
182  pthread_mutex_unlock(&w->mutex);
183  }
184 
185  if (ctx->main_func && execute_main)
186  ctx->main_func(ctx->priv);
187  else
188  is_last = run_jobs(ctx);
189 
190  if (!is_last) {
191  pthread_mutex_lock(&ctx->done_mutex);
192  while (!ctx->done)
193  pthread_cond_wait(&ctx->done_cond, &ctx->done_mutex);
194  ctx->done = 0;
195  pthread_mutex_unlock(&ctx->done_mutex);
196  }
197 }
198 
200 {
202  int nb_workers, i;
203 
204  if (!pctx || !*pctx)
205  return;
206 
207  ctx = *pctx;
208  nb_workers = ctx->nb_threads;
209  if (!ctx->main_func)
210  nb_workers--;
211 
212  ctx->finished = 1;
213  for (i = 0; i < nb_workers; i++) {
214  WorkerContext *w = &ctx->workers[i];
215  pthread_mutex_lock(&w->mutex);
216  w->done = 0;
217  pthread_cond_signal(&w->cond);
218  pthread_mutex_unlock(&w->mutex);
219  }
220 
221  for (i = 0; i < nb_workers; i++) {
222  WorkerContext *w = &ctx->workers[i];
223  pthread_join(w->thread, NULL);
224  pthread_cond_destroy(&w->cond);
225  pthread_mutex_destroy(&w->mutex);
226  }
227 
228  pthread_cond_destroy(&ctx->done_cond);
229  pthread_mutex_destroy(&ctx->done_mutex);
230  av_freep(&ctx->workers);
231  av_freep(pctx);
232 }
233 
234 #else /* HAVE_PTHREADS || HAVE_W32THREADS || HAVE_OS32THREADS */
235 
237  void (*worker_func)(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads),
238  void (*main_func)(void *priv),
239  int nb_threads)
240 {
241  *pctx = NULL;
242  return AVERROR(EINVAL);
243 }
244 
245 void avpriv_slicethread_execute(AVSliceThread *ctx, int nb_jobs, int execute_main)
246 {
247  av_assert0(0);
248 }
249 
251 {
252  av_assert0(!pctx || !*pctx);
253 }
254 
255 #endif /* HAVE_PTHREADS || HAVE_W32THREADS || HAVE_OS32THREADS */
#define NULL
Definition: coverity.c:32
static av_always_inline int pthread_mutex_destroy(pthread_mutex_t *mutex)
Definition: os2threads.h:108
static AVMutex mutex
Definition: log.c:44
#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:166
int av_cpu_count(void)
Definition: cpu.c:267
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:236
static av_always_inline int pthread_cond_destroy(pthread_cond_t *cond)
Definition: os2threads.h:140
void * av_calloc(size_t nmemb, size_t size)
Non-inlined equivalent of av_mallocz_array().
Definition: mem.c:244
#define av_assert0(cond)
assert() equivalent, that is always enabled.
Definition: avassert.h:37
struct AVSliceThread AVSliceThread
Definition: slicethread.h:22
static av_always_inline int pthread_cond_signal(pthread_cond_t *cond)
Definition: os2threads.h:148
#define i(width, name, range_min, range_max)
Definition: cbs_h2645.c:259
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:250
uint8_t w
Definition: llviddspenc.c:38
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:90
int( main_func)(AVCodecContext *c)
Definition: pthread_slice.c:41
static av_always_inline int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutexattr_t *attr)
Definition: os2threads.h:100
#define pthread_mutex_unlock(a)
Definition: ffprobe.c:65
#define atomic_fetch_add_explicit(object, operand, order)
Definition: stdatomic.h:149
static av_always_inline int pthread_create(pthread_t *thread, const pthread_attr_t *attr, void *(*start_routine)(void *), void *arg)
Definition: os2threads.h:76
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:236
void avpriv_slicethread_execute(AVSliceThread *ctx, int nb_jobs, int execute_main)
Execute slice threading.
Definition: slicethread.c:245
intptr_t atomic_uint
Definition: stdatomic.h:56
#define atomic_store_explicit(object, desired, order)
Definition: stdatomic.h:90
int(* cond)(enum AVPixelFormat pix_fmt)
Definition: pixdesc_query.c:28
_fmutex pthread_mutex_t
Definition: os2threads.h:49
static av_always_inline int pthread_cond_init(pthread_cond_t *cond, const pthread_condattr_t *attr)
Definition: os2threads.h:129
#define atomic_init(obj, value)
Definition: stdatomic.h:33
#define av_freep(p)
Filter the word “frame” indicates either a video frame or a group of audio as stored in an AVFrame structure Format for each input and each output the list of supported formats For video that means pixel format For audio that means channel sample they are references to shared objects When the negotiation mechanism computes the intersection of the formats supported at each end of a all references to both lists are replaced with a reference to the intersection And when a single format is eventually chosen for a link amongst the remaining all references to the list are updated That means that if a filter requires that its input and output have the same format amongst a supported all it has to do is use a reference to the same list of formats query_formats can leave some formats unset and return AVERROR(EAGAIN) to cause the negotiation mechanism toagain later.That can be used by filters with complex requirements to use the format negotiated on one link to set the formats supported on another.Frame references ownership and permissions