git.sesse.net Git - ffmpeg/blob - libavutil/slicethread.c

   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 {
  28     AVSliceThread   *ctx;
  29     pthread_mutex_t mutex;
  30     pthread_cond_t  cond;
  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 {
  99     AVSliceThread *ctx;
 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);
 152             avpriv_slicethread_free(pctx);
 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
 199 void avpriv_slicethread_free(AVSliceThread **pctx)
 200 {
 201     AVSliceThread *ctx;
 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
 236 int avpriv_slicethread_create(AVSliceThread **pctx, void *priv,
 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
 250 void avpriv_slicethread_free(AVSliceThread **pctx)
 251 {
 252     av_assert0(!pctx || !*pctx);
 253 }
 254
 255 #endif /* HAVE_PTHREADS || HAVE_W32THREADS || HAVE_OS32THREADS */