FFmpeg
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros Groups Pages
pthread_slice.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 /**
20  * @file
21  * Slice multithreading support functions
22  * @see doc/multithreading.txt
23  */
24 
25 #include "config.h"
26 
27 #if HAVE_PTHREADS
28 #include <pthread.h>
29 #elif HAVE_W32THREADS
30 #include "compat/w32pthreads.h"
31 #elif HAVE_OS2THREADS
32 #include "compat/os2threads.h"
33 #endif
34 
35 #include "avcodec.h"
36 #include "internal.h"
37 #include "pthread_internal.h"
38 #include "thread.h"
39 
40 #include "libavutil/common.h"
41 #include "libavutil/cpu.h"
42 #include "libavutil/mem.h"
43 
44 typedef int (action_func)(AVCodecContext *c, void *arg);
45 typedef int (action_func2)(AVCodecContext *c, void *arg, int jobnr, int threadnr);
46 
47 typedef struct SliceThreadContext {
51  void *args;
52  int *rets;
54  int job_count;
55  int job_size;
56 
60  unsigned current_execute;
62  int done;
63 
64  int *entries;
70 
71 static void* attribute_align_arg worker(void *v)
72 {
73  AVCodecContext *avctx = v;
75  unsigned last_execute = 0;
76  int our_job = c->job_count;
77  int thread_count = avctx->thread_count;
78  int self_id;
79 
81  self_id = c->current_job++;
82  for (;;){
83  while (our_job >= c->job_count) {
84  if (c->current_job == thread_count + c->job_count)
86 
87  while (last_execute == c->current_execute && !c->done)
89  last_execute = c->current_execute;
90  our_job = self_id;
91 
92  if (c->done) {
94  return NULL;
95  }
96  }
98 
99  c->rets[our_job%c->rets_count] = c->func ? c->func(avctx, (char*)c->args + our_job*c->job_size):
100  c->func2(avctx, c->args, our_job, self_id);
101 
103  our_job = c->current_job++;
104  }
105 }
106 
108 {
110  int i;
111 
113  c->done = 1;
115  for (i = 0; i < c->thread_count; i++)
118 
119  for (i=0; i<avctx->thread_count; i++)
120  pthread_join(c->workers[i], NULL);
121 
122  for (i = 0; i < c->thread_count; i++) {
125  }
126 
130 
131  av_freep(&c->entries);
133  av_freep(&c->progress_cond);
134 
135  av_freep(&c->workers);
136  av_freep(&avctx->internal->thread_ctx);
137 }
138 
140 {
141  while (c->current_job != thread_count + c->job_count)
144 }
145 
146 static int thread_execute(AVCodecContext *avctx, action_func* func, void *arg, int *ret, int job_count, int job_size)
147 {
149  int dummy_ret;
150 
151  if (!(avctx->active_thread_type&FF_THREAD_SLICE) || avctx->thread_count <= 1)
152  return avcodec_default_execute(avctx, func, arg, ret, job_count, job_size);
153 
154  if (job_count <= 0)
155  return 0;
156 
158 
159  c->current_job = avctx->thread_count;
160  c->job_count = job_count;
161  c->job_size = job_size;
162  c->args = arg;
163  c->func = func;
164  if (ret) {
165  c->rets = ret;
166  c->rets_count = job_count;
167  } else {
168  c->rets = &dummy_ret;
169  c->rets_count = 1;
170  }
171  c->current_execute++;
173 
175 
176  return 0;
177 }
178 
179 static int thread_execute2(AVCodecContext *avctx, action_func2* func2, void *arg, int *ret, int job_count)
180 {
182  c->func2 = func2;
183  return thread_execute(avctx, NULL, arg, ret, job_count, 0);
184 }
185 
187 {
188  int i;
190  int thread_count = avctx->thread_count;
191 
192 #if HAVE_W32THREADS
193  w32thread_init();
194 #endif
195 
196  if (!thread_count) {
197  int nb_cpus = av_cpu_count();
198  if (avctx->height)
199  nb_cpus = FFMIN(nb_cpus, (avctx->height+15)/16);
200  // use number of cores + 1 as thread count if there is more than one
201  if (nb_cpus > 1)
202  thread_count = avctx->thread_count = FFMIN(nb_cpus + 1, MAX_AUTO_THREADS);
203  else
204  thread_count = avctx->thread_count = 1;
205  }
206 
207  if (thread_count <= 1) {
208  avctx->active_thread_type = 0;
209  return 0;
210  }
211 
212  c = av_mallocz(sizeof(SliceThreadContext));
213  if (!c)
214  return -1;
215 
216  c->workers = av_mallocz_array(thread_count, sizeof(pthread_t));
217  if (!c->workers) {
218  av_free(c);
219  return -1;
220  }
221 
222  avctx->internal->thread_ctx = c;
223  c->current_job = 0;
224  c->job_count = 0;
225  c->job_size = 0;
226  c->done = 0;
231  for (i=0; i<thread_count; i++) {
232  if(pthread_create(&c->workers[i], NULL, worker, avctx)) {
233  avctx->thread_count = i;
235  ff_thread_free(avctx);
236  return -1;
237  }
238  }
239 
240  thread_park_workers(c, thread_count);
241 
242  avctx->execute = thread_execute;
243  avctx->execute2 = thread_execute2;
244  return 0;
245 }
246 
247 void ff_thread_report_progress2(AVCodecContext *avctx, int field, int thread, int n)
248 {
250  int *entries = p->entries;
251 
252  pthread_mutex_lock(&p->progress_mutex[thread]);
253  entries[field] +=n;
254  pthread_cond_signal(&p->progress_cond[thread]);
256 }
257 
258 void ff_thread_await_progress2(AVCodecContext *avctx, int field, int thread, int shift)
259 {
261  int *entries = p->entries;
262 
263  if (!entries || !field) return;
264 
265  thread = thread ? thread - 1 : p->thread_count - 1;
266 
267  pthread_mutex_lock(&p->progress_mutex[thread]);
268  while ((entries[field - 1] - entries[field]) < shift){
269  pthread_cond_wait(&p->progress_cond[thread], &p->progress_mutex[thread]);
270  }
272 }
273 
275 {
276  int i;
277 
278  if (avctx->active_thread_type & FF_THREAD_SLICE) {
280  p->thread_count = avctx->thread_count;
281  p->entries = av_mallocz_array(count, sizeof(int));
282 
285 
286  if (!p->entries || !p->progress_mutex || !p->progress_cond) {
287  av_freep(&p->entries);
289  av_freep(&p->progress_cond);
290  return AVERROR(ENOMEM);
291  }
292  p->entries_count = count;
293 
294  for (i = 0; i < p->thread_count; i++) {
297  }
298  }
299 
300  return 0;
301 }
302 
304 {
306  memset(p->entries, 0, p->entries_count * sizeof(int));
307 }
static av_unused void w32thread_init(void)
Definition: w32pthreads.h:299
#define NULL
Definition: coverity.c:32
int ff_slice_thread_init(AVCodecContext *avctx)
float v
static int shift(int a, int b)
Definition: sonic.c:82
static av_always_inline int pthread_mutex_destroy(pthread_mutex_t *mutex)
Definition: os2threads.h:94
static av_always_inline int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex)
Definition: os2threads.h:153
int av_cpu_count(void)
Definition: cpu.c:251
memory handling functions
os2threads to pthreads wrapper
int( action_func)(AVCodecContext *c, void *arg)
Definition: pthread_slice.c:44
void ff_thread_await_progress2(AVCodecContext *avctx, int field, int thread, int shift)
static av_always_inline int pthread_cond_destroy(pthread_cond_t *cond)
Definition: os2threads.h:124
pthread_cond_t current_job_cond
Definition: pthread_slice.c:58
HMTX pthread_mutex_t
Definition: os2threads.h:40
void * thread_ctx
Definition: internal.h:135
Multithreading support functions.
pthread_mutex_t * progress_mutex
Definition: pthread_slice.c:68
static av_always_inline int pthread_cond_signal(pthread_cond_t *cond)
Definition: os2threads.h:131
static void *attribute_align_arg worker(void *v)
Definition: pthread_slice.c:71
#define AVERROR(e)
Definition: error.h:43
pthread_mutex_t current_job_lock
Definition: pthread_slice.c:59
int active_thread_type
Which multithreading methods are in use by the codec.
Definition: avcodec.h:3052
action_func2 * func2
Definition: pthread_slice.c:50
const char * arg
Definition: jacosubdec.c:66
GLsizei count
Definition: opengl_enc.c:109
void ff_thread_free(AVCodecContext *avctx)
Definition: pthread.c:82
action_func * func
Definition: pthread_slice.c:49
Libavcodec external API header.
int ff_alloc_entries(AVCodecContext *avctx, int count)
static int thread_execute2(AVCodecContext *avctx, action_func2 *func2, void *arg, int *ret, int job_count)
#define FFMIN(a, b)
Definition: common.h:81
void ff_slice_thread_free(AVCodecContext *avctx)
static av_always_inline int pthread_join(pthread_t thread, void **value_ptr)
Definition: os2threads.h:80
static av_always_inline int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutexattr_t *attr)
Definition: os2threads.h:87
pthread_cond_t * progress_cond
Definition: pthread_slice.c:67
int n
Definition: avisynth_c.h:547
int thread_count
thread count is used to decide how many independent tasks should be passed to execute() ...
Definition: avcodec.h:3033
void ff_reset_entries(AVCodecContext *avctx)
static av_always_inline int pthread_create(pthread_t *thread, const pthread_attr_t *attr, void *(*start_routine)(void *), void *arg)
Definition: os2threads.h:64
unsigned current_execute
Definition: pthread_slice.c:60
void ff_thread_report_progress2(AVCodecContext *avctx, int field, int thread, int n)
#define FF_THREAD_SLICE
Decode more than one part of a single frame at once.
Definition: avcodec.h:3045
main external API structure.
Definition: avcodec.h:1502
static av_always_inline void thread_park_workers(SliceThreadContext *c, int thread_count)
int(* func)(AVBPrint *dst, const char *in, const char *arg)
Definition: jacosubdec.c:67
#define MAX_AUTO_THREADS
common internal api header.
common internal and external API header
static double c[64]
static av_always_inline int pthread_cond_init(pthread_cond_t *cond, const pthread_condattr_t *attr)
Definition: os2threads.h:115
#define av_free(p)
int(* execute)(struct AVCodecContext *c, int(*func)(struct AVCodecContext *c2, void *arg), void *arg2, int *ret, int count, int size)
The codec may call this to execute several independent things.
Definition: avcodec.h:3073
int(* execute2)(struct AVCodecContext *c, int(*func)(struct AVCodecContext *c2, void *arg, int jobnr, int threadnr), void *arg2, int *ret, int count)
The codec may call this to execute several independent things.
Definition: avcodec.h:3093
static av_always_inline int pthread_cond_broadcast(pthread_cond_t *cond)
Definition: os2threads.h:142
static av_always_inline int pthread_mutex_unlock(pthread_mutex_t *mutex)
Definition: os2threads.h:108
struct AVCodecInternal * internal
Private context used for internal data.
Definition: avcodec.h:1552
pthread_t * workers
Definition: pthread_slice.c:48
w32threads to pthreads wrapper
static void * av_mallocz_array(size_t nmemb, size_t size)
Definition: mem.h:228
#define av_freep(p)
#define av_always_inline
Definition: attributes.h:37
static av_always_inline int pthread_mutex_lock(pthread_mutex_t *mutex)
Definition: os2threads.h:101
#define av_malloc_array(a, b)
int( action_func2)(AVCodecContext *c, void *arg, int jobnr, int threadnr)
Definition: pthread_slice.c:45
static int thread_execute(AVCodecContext *avctx, action_func *func, void *arg, int *ret, int job_count, int job_size)
pthread_cond_t last_job_cond
Definition: pthread_slice.c:57
void * av_mallocz(size_t size)
Allocate a block of size bytes with alignment suitable for all memory accesses (including vectors if ...
Definition: mem.c:252
int avcodec_default_execute(AVCodecContext *c, int(*func)(AVCodecContext *c2, void *arg2), void *arg, int *ret, int count, int size)
Definition: utils.c:1110