/*
* This file is part of FFmpeg.
*
* FFmpeg is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* FFmpeg is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with FFmpeg; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#include <stdatomic.h>
#include "slicethread.h"
#include "mem.h"
#include "thread.h"
#include "avassert.h"
#if HAVE_PTHREADS || HAVE_W32THREADS || HAVE_OS2THREADS
typedef struct WorkerContext {
AVSliceThread *ctx;
pthread_mutex_t mutex;
pthread_cond_t cond;
pthread_t thread;
int done;
} WorkerContext;
struct AVSliceThread {
WorkerContext *workers;
int nb_threads;
int nb_active_threads;
int nb_jobs;
atomic_uint first_job;
atomic_uint current_job;
pthread_mutex_t done_mutex;
pthread_cond_t done_cond;
int done;
int finished;
void *priv;
void (*worker_func)(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads);
void (*main_func)(void *priv);
};
static int run_jobs(AVSliceThread *ctx)
{
unsigned nb_jobs = ctx->nb_jobs;
unsigned nb_active_threads = ctx->nb_active_threads;
unsigned first_job = atomic_fetch_add_explicit(&ctx->first_job, 1, memory_order_acq_rel);
unsigned current_job = first_job;
do {
ctx->worker_func(ctx->priv, current_job, first_job, nb_jobs, nb_active_threads);
} while ((current_job = atomic_fetch_add_explicit(&ctx->current_job, 1, memory_order_acq_rel)) < nb_jobs);
return current_job == nb_jobs + nb_active_threads - 1;
}
static void *attribute_align_arg thread_worker(void *v)
{
WorkerContext *w = v;
AVSliceThread *ctx = w->ctx;
pthread_mutex_lock(&w->mutex);
pthread_cond_signal(&w->cond);
while (1) {
w->done = 1;
while (w->done)
pthread_cond_wait(&w->cond, &w->mutex);
if (ctx->finished) {
pthread_mutex_unlock(&w->mutex);
return NULL;
}
if (run_jobs(ctx)) {
pthread_mutex_lock(&ctx->done_mutex);
ctx->done = 1;
pthread_cond_signal(&ctx->done_cond);
pthread_mutex_unlock(&ctx->done_mutex);
}
}
}
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)
{
AVSliceThread *ctx;
int nb_workers, i;
#if HAVE_W32THREADS
w32thread_init();
#endif
av_assert0(nb_threads >= 0);
if (!nb_threads) {
int nb_cpus = av_cpu_count();
if (nb_cpus > 1)
nb_threads = nb_cpus + 1;
else
nb_threads = 1;
}
nb_workers = nb_threads;
if (!main_func)
nb_workers--;
*pctx = ctx = av_mallocz(sizeof(*ctx));
if (!ctx)
return AVERROR(ENOMEM);
if (nb_workers && !(ctx->workers = av_calloc(nb_workers, sizeof(*ctx->workers)))) {
av_freep(pctx);
return AVERROR(ENOMEM);
}
ctx->priv = priv;
ctx->worker_func = worker_func;
ctx->main_func = main_func;
ctx->nb_threads = nb_threads;
ctx->nb_active_threads = 0;
ctx->nb_jobs = 0;
ctx->finished = 0;
atomic_init(&ctx->first_job, 0);
atomic_init(&ctx->current_job, 0);
pthread_mutex_init(&ctx->done_mutex, NULL);
pthread_cond_init(&ctx->done_cond, NULL);
ctx->done = 0;
for (i = 0; i < nb_workers; i++) {
WorkerContext *w = &ctx->workers[i];
int ret;
w->ctx = ctx;
pthread_mutex_init(&w->mutex, NULL);
pthread_cond_init(&w->cond, NULL);
pthread_mutex_lock(&w->mutex);
w->done = 0;
if (ret = pthread_create(&w->thread, NULL, thread_worker, w)) {
ctx->nb_threads = main_func ? i : i + 1;
pthread_mutex_unlock(&w->mutex);
pthread_cond_destroy(&w->cond);
pthread_mutex_destroy(&w->mutex);
avpriv_slicethread_free(pctx);
return AVERROR(ret);
}
while (!w->done)
pthread_cond_wait(&w->cond, &w->mutex);
pthread_mutex_unlock(&w->mutex);
}
return nb_threads;
}
void avpriv_slicethread_execute(AVSliceThread *ctx, int nb_jobs, int execute_main)
{
int nb_workers, i, is_last = 0;
av_assert0(nb_jobs > 0);
ctx->nb_jobs = nb_jobs;
ctx->nb_active_threads = FFMIN(nb_jobs, ctx->nb_threads);
atomic_store_explicit(&ctx->first_job, 0, memory_order_relaxed);
atomic_store_explicit(&ctx->current_job, ctx->nb_active_threads, memory_order_relaxed);
nb_workers = ctx->nb_active_threads;
if (!ctx->main_func || !execute_main)
nb_workers--;
for (i = 0; i < nb_workers; i++) {
WorkerContext *w = &ctx->workers[i];
pthread_mutex_lock(&w->mutex);
w->done = 0;
pthread_cond_signal(&w->cond);
pthread_mutex_unlock(&w->mutex);
}
if (ctx->main_func && execute_main)
ctx->main_func(ctx->priv);
else
is_last = run_jobs(ctx);
if (!is_last) {
pthread_mutex_lock(&ctx->done_mutex);
while (!ctx->done)
pthread_cond_wait(&ctx->done_cond, &ctx->done_mutex);
ctx->done = 0;
pthread_mutex_unlock(&ctx->done_mutex);
}
}
void avpriv_slicethread_free(AVSliceThread **pctx)
{
AVSliceThread *ctx;
int nb_workers, i;
if (!pctx || !*pctx)
return;
ctx = *pctx;
nb_workers = ctx->nb_threads;
if (!ctx->main_func)
nb_workers--;
ctx->finished = 1;
for (i = 0; i < nb_workers; i++) {
WorkerContext *w = &ctx->workers[i];
pthread_mutex_lock(&w->mutex);
w->done = 0;
pthread_cond_signal(&w->cond);
pthread_mutex_unlock(&w->mutex);
}
for (i = 0; i < nb_workers; i++) {
WorkerContext *w = &ctx->workers[i];
pthread_join(w->thread, NULL);
pthread_cond_destroy(&w->cond);
pthread_mutex_destroy(&w->mutex);
}
pthread_cond_destroy(&ctx->done_cond);
pthread_mutex_destroy(&ctx->done_mutex);
av_freep(&ctx->workers);
av_freep(pctx);
}
#else /* HAVE_PTHREADS || HAVE_W32THREADS || HAVE_OS32THREADS */
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)
{
*pctx = NULL;
return AVERROR(EINVAL);
}
void avpriv_slicethread_execute(AVSliceThread *ctx, int nb_jobs, int execute_main)
{
av_assert0(0);
}
void avpriv_slicethread_free(AVSliceThread **pctx)
{
av_assert0(!pctx || !*pctx);
}
#endif /* HAVE_PTHREADS || HAVE_W32THREADS || HAVE_OS32THREADS */