sysutils/ffmpeg
1
0
Fork 0
mirror of https://git.ffmpeg.org/ffmpeg.git synced 2026-10-24 02:09:03 +03:00
Code Activity

avutil/slicethread: add return value to slicethread callbacks

Browse Source 
Several places in the code currently manually roll the return value
saving. This small signature change allows slicethread.c to collect
return values directly. Unlike the existing solutions which mostly
rely on collecting the return values in an array, we can just use an
atomic CAS to set the error iff no error has been set already.

We need a slight bit of extra complication to avoid breaking the ABI,
which requires keeping around the old names as deprecated aliases that
just set up the appropriate wrappers. However, fortunately, the way I
implemented the deprecated signatures allows callers to freely mix
avpriv_slicethread_execute() and avpriv_slicethread_execute2(), even
on contexts created with one or the other.

It's worth pointing out that this commit just changes the signatures,
but doesn't yet update the various to use the return value.

Signed-off-by: Niklas Haas <git@haasn.dev>
This commit is contained in:
Niklas Haas
2026-07-28 14:44:33 +02:00
parent 1d7b14f61d
commit dab32dcc21
9 changed files with 152 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
libavcodec/pthread_slice.c
View File

@@ -48,13 +48,14 @@ typedef struct SliceThreadContext {
int job_size;
} SliceThreadContext;
static void main_function(void *priv) {
static int main_function(void *priv) {
AVCodecContext *avctx = priv;
SliceThreadContext *c = avctx->internal->thread_ctx;
c->mainfunc(avctx);
return 0;
}
static void worker_func(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads)
static int worker_func(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads)
{
AVCodecContext *avctx = priv;
SliceThreadContext *c = avctx->internal->thread_ctx;
@@ -64,6 +65,8 @@ static void worker_func(void *priv, int jobnr, int threadnr, int nb_jobs, int nb
: c->func2(avctx, c->args, jobnr, threadnr);
if (c->rets)
c->rets[jobnr] = ret;
return 0;
}
av_cold void ff_slice_thread_free(AVCodecContext *avctx)
@@ -90,7 +93,7 @@ static int thread_execute(AVCodecContext *avctx, action_func* func, void *arg, i
c->func = func;
c->rets = ret;
avpriv_slicethread_execute(c->thread, job_count, !!c->mainfunc );
avpriv_slicethread_execute2(c->thread, job_count, !!c->mainfunc );
return 0;
}
@@ -113,7 +116,7 @@ av_cold int ff_slice_thread_init(AVCodecContext *avctx)
{
SliceThreadContext *c;
int thread_count = avctx->thread_count;
void (*mainfunc)(void *);
int (*mainfunc)(void *);
if (!thread_count) {
int nb_cpus = av_cpu_count();
@@ -135,8 +138,8 @@ av_cold int ff_slice_thread_init(AVCodecContext *avctx)
if (!c)
return AVERROR(ENOMEM);
mainfunc = ffcodec(avctx->codec)->caps_internal & FF_CODEC_CAP_SLICE_THREAD_HAS_MF ? &main_function : NULL;
thread_count = avpriv_slicethread_create(&c->thread, avctx, worker_func,
mainfunc, thread_count);
thread_count = avpriv_slicethread_create2(&c->thread, avctx, worker_func,
mainfunc, thread_count);
if (thread_count <= 1) {
ff_slice_thread_free(avctx);
avctx->thread_count = 1;

7
libavfilter/pthread.c
View File

@@ -42,12 +42,13 @@ typedef struct ThreadContext {
int *rets;
} ThreadContext;
static void worker_func(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads)
static int worker_func(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads)
{
ThreadContext *c = priv;
int ret = c->func(c->ctx, c->arg, jobnr, nb_jobs);
if (c->rets)
c->rets[jobnr] = ret;
return 0;
}
static void slice_thread_uninit(ThreadContext *c)
@@ -67,13 +68,13 @@ static int thread_execute(AVFilterContext *ctx, avfilter_action_func *func,
c->func = func;
c->rets = ret;
avpriv_slicethread_execute(c->thread, nb_jobs, 0);
avpriv_slicethread_execute2(c->thread, nb_jobs, 0);
return 0;
}
static int thread_init_internal(ThreadContext *c, int nb_threads)
{
nb_threads = avpriv_slicethread_create(&c->thread, c, worker_func, NULL, nb_threads);
nb_threads = avpriv_slicethread_create2(&c->thread, c, worker_func, NULL, nb_threads);
if (nb_threads <= 1)
avpriv_slicethread_free(&c->thread);
return FFMAX(nb_threads, 1);

101
libavutil/slicethread.c
View File

@@ -48,10 +48,17 @@ struct AVSliceThread {
pthread_cond_t done_cond;
int done;
int finished;
atomic_int error;
void *priv;
void (*worker_func)(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads);
void (*main_func)(void *priv);
int (*worker_func)(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads);
int (*main_func)(void *priv);
#if LIBAVUTIL_VERSION_MAJOR < 62
void (*worker_func_v1)(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads);
void (*main_func_v1)(void *priv);
void *priv_v1;
#endif
};
static int run_jobs(AVSliceThread *ctx)
@@ -62,7 +69,16 @@ static int run_jobs(AVSliceThread *ctx)
unsigned current_job = first_job;
do {
ctx->worker_func(ctx->priv, current_job, first_job, nb_jobs, nb_active_threads);
int ret = atomic_load_explicit(&ctx->error, memory_order_relaxed);
if (ret)
continue;
ret = ctx->worker_func(ctx->priv, current_job, first_job, nb_jobs, nb_active_threads);
if (ret) {
int prev = 0;
atomic_compare_exchange_strong_explicit(&ctx->error, &prev, ret,
memory_order_relaxed,
memory_order_relaxed);
}
} 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;
@@ -96,10 +112,10 @@ static void *attribute_align_arg thread_worker(void *v)
}
av_cold
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)
int avpriv_slicethread_create2(AVSliceThread **pctx, void *priv,
int (*worker_func)(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads),
int (*main_func)(void *priv),
int nb_threads)
{
AVSliceThread *ctx;
int nb_workers, i;
@@ -187,13 +203,14 @@ int avpriv_slicethread_create(AVSliceThread **pctx, void *priv,
return nb_threads;
}
void avpriv_slicethread_execute(AVSliceThread *ctx, int nb_jobs, int execute_main)
int avpriv_slicethread_execute2(AVSliceThread *ctx, int nb_jobs, int execute_main)
{
int nb_workers, i, is_last = 0;
int nb_workers, i, is_last = 0, ret = 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->error, 0, memory_order_relaxed);
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;
@@ -208,9 +225,9 @@ void avpriv_slicethread_execute(AVSliceThread *ctx, int nb_jobs, int execute_mai
pthread_mutex_unlock(&w->mutex);
}
if (ctx->main_func && execute_main)
ctx->main_func(ctx->priv);
else
if (ctx->main_func && execute_main) {
ret = ctx->main_func(ctx->priv);
} else
is_last = run_jobs(ctx);
if (!is_last) {
@@ -220,6 +237,11 @@ void avpriv_slicethread_execute(AVSliceThread *ctx, int nb_jobs, int execute_mai
ctx->done = 0;
pthread_mutex_unlock(&ctx->done_mutex);
}
if (!ret)
ret = atomic_load_explicit(&ctx->error, memory_order_relaxed);
return ret;
}
av_cold void avpriv_slicethread_free(AVSliceThread **pctx)
@@ -258,16 +280,16 @@ av_cold void avpriv_slicethread_free(AVSliceThread **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)
int avpriv_slicethread_create2(AVSliceThread **pctx, void *priv,
int (*worker_func)(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads),
int (*main_func)(void *priv),
int nb_threads)
{
*pctx = NULL;
return AVERROR(ENOSYS);
}
void avpriv_slicethread_execute(AVSliceThread *ctx, int nb_jobs, int execute_main)
int avpriv_slicethread_execute2(AVSliceThread *ctx, int nb_jobs, int execute_main)
{
av_assert0(0);
}
@@ -278,3 +300,48 @@ void avpriv_slicethread_free(AVSliceThread **pctx)
}
#endif /* HAVE_PTHREADS || HAVE_W32THREADS || HAVE_OS32THREADS */
/**
* Backwards compatibility wrapper for the deprecated avpriv_ slicethread API.
*/
#if LIBAVUTIL_VERSION_MAJOR < 62
static int wrapper_worker(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads)
{
AVSliceThread *ctx = priv;
ctx->worker_func_v1(ctx->priv_v1, jobnr, threadnr, nb_jobs, nb_threads);
return 0;
}
static int wrapper_main(void *priv)
{
AVSliceThread *ctx = priv;
ctx->main_func_v1(ctx->priv_v1);
return 0;
}
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)
{
int ret = avpriv_slicethread_create2(pctx, NULL, wrapper_worker,
main_func ? wrapper_main : NULL,
nb_threads);
if (ret < 0)
return ret;
(*pctx)->priv = *pctx;
(*pctx)->priv_v1 = priv;
(*pctx)->worker_func_v1 = worker_func;
(*pctx)->main_func_v1 = main_func;
return ret;
}
void avpriv_slicethread_execute(AVSliceThread *ctx, int nb_jobs, int execute_main)
{
avpriv_slicethread_execute2(ctx, nb_jobs, execute_main);
}
#endif /* LIBAVUTIL_VERSION_MAJOR < 62 */

40
libavutil/slicethread.h
View File

@@ -19,6 +19,9 @@
#ifndef AVUTIL_SLICETHREAD_H
#define AVUTIL_SLICETHREAD_H
#include "attributes.h"
#include "version.h"
typedef struct AVSliceThread AVSliceThread;
/**
@@ -30,18 +33,24 @@ typedef struct AVSliceThread AVSliceThread;
* @param nb_threads number of threads, 0 for automatic, must be >= 0
* @return return number of threads or negative AVERROR on failure
*/
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);
int avpriv_slicethread_create2(AVSliceThread **pctx, void *priv,
int (*worker_func)(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads),
int (*main_func)(void *priv),
int nb_threads);
/**
* Execute slice threading.
* Execute slice threading. If any job returns a nonzero value,
* all remaining jobs will be aborted and that value returned.
*
* If `main_func` itself returns a nonzero value, that value is guaranteed
* to be returned, even if a worker thread returns a nonzero value first.
*
* @param ctx slice threading context
* @param nb_jobs number of jobs, must be > 0
* @param execute_main also execute main_func
* @return 0 on success, else first seen return value
*/
void avpriv_slicethread_execute(AVSliceThread *ctx, int nb_jobs, int execute_main);
int avpriv_slicethread_execute2(AVSliceThread *ctx, int nb_jobs, int execute_main);
/**
* Destroy slice threading context.
@@ -49,4 +58,23 @@ void avpriv_slicethread_execute(AVSliceThread *ctx, int nb_jobs, int execute_mai
*/
void avpriv_slicethread_free(AVSliceThread **pctx);
#if LIBAVUTIL_VERSION_MAJOR < 62
/**
* @deprecated
* Wrapper around avpriv_slicethread_create2() for ABI backwards compatibility.
*/
attribute_deprecated
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);
/**
* @deprecated
* Wrapper around avpriv_slicethread_execute2() for ABI backwards compatibility.
*/
attribute_deprecated
void avpriv_slicethread_execute(AVSliceThread *ctx, int nb_jobs, int execute_main);
#endif /* LIBAVUTIL_VERSION_MAJOR < 62 */
#endif

12
libswscale/cms.c
View File

@@ -589,8 +589,8 @@ static av_always_inline void update_hue_peaks(CmsCtx *ctx, float P, float T)
}
}
static void generate_slice(void *priv, int jobnr, int threadnr, int nb_jobs,
int nb_threads)
static int generate_slice(void *priv, int jobnr, int threadnr, int nb_jobs,
int nb_threads)
{
CmsCtx ctx = *(const CmsCtx *) priv;
@@ -655,7 +655,7 @@ static void generate_slice(void *priv, int jobnr, int threadnr, int nb_jobs,
}
if (!output)
return;
return 0;
/* Generate split gamut mapping LUT */
for (int Tx = output_start; Tx < output_end; Tx++) {
@@ -678,6 +678,8 @@ static void generate_slice(void *priv, int jobnr, int threadnr, int nb_jobs,
}
}
}
return 0;
}
int ff_sws_color_map_generate_static(v3u16_t *lut, int size, const SwsColorMap *map)
@@ -733,13 +735,13 @@ int ff_sws_color_map_generate_dynamic(v3u16_t *input, v3u16_t *output,
ctx.dst.wp, ctx.src.wp);
}
ret = avpriv_slicethread_create(&slicethread, &ctx, generate_slice, NULL, 0);
ret = avpriv_slicethread_create2(&slicethread, &ctx, generate_slice, NULL, 0);
if (ret < 0)
return ret;
ctx.slice_size = (ctx.size_input + ret - 1) / ret;
num_slices = (ctx.size_input + ctx.slice_size - 1) / ctx.slice_size;
avpriv_slicethread_execute(slicethread, num_slices, 0);
avpriv_slicethread_execute2(slicethread, num_slices, 0);
avpriv_slicethread_free(&slicethread);
return 0;
}

11
libswscale/graph.c
View File

@@ -851,8 +851,8 @@ static int init_passes(SwsGraph *graph)
return 0;
}
static void sws_graph_worker(void *priv, int jobnr, int threadnr, int nb_jobs,
int nb_threads)
static int sws_graph_worker(void *priv, int jobnr, int threadnr, int nb_jobs,
int nb_threads)
{
SwsGraph *graph = priv;
const SwsPass *pass = graph->exec.pass;
@@ -860,6 +860,7 @@ static void sws_graph_worker(void *priv, int jobnr, int threadnr, int nb_jobs,
const int slice_h = FFMIN(pass->slice_h, pass->lines - slice_y);
pass->run(graph->exec.output, graph->exec.input, slice_y, slice_h, pass);
return 0;
}
SwsGraph *ff_sws_graph_alloc(void)
@@ -900,8 +901,8 @@ int ff_sws_graph_init(SwsGraph *graph, SwsContext *ctx, const SwsFormat *dst,
if (ctx->threads == 1) {
graph->num_threads = 1;
} else {
ret = avpriv_slicethread_create(&graph->slicethread, (void *) graph,
sws_graph_worker, NULL, ctx->threads);
ret = avpriv_slicethread_create2(&graph->slicethread, (void *) graph,
sws_graph_worker, NULL, ctx->threads);
if (ret == AVERROR(ENOSYS)) {
/* Fall back to single threaded operation */
graph->num_threads = 1;
@@ -1045,7 +1046,7 @@ int ff_sws_graph_run(SwsGraph *graph, const AVFrame *dst, const AVFrame *src)
if (pass->num_slices == 1) {
pass->run(graph->exec.output, graph->exec.input, 0, pass->lines, pass);
} else {
avpriv_slicethread_execute(graph->slicethread, pass->num_slices, 0);
avpriv_slicethread_execute2(graph->slicethread, pass->num_slices, 0);
}
}

7
libswscale/swscale.c
View File

@@ -1391,7 +1391,7 @@ int sws_receive_slice(SwsContext *sws, unsigned int slice_start,
c->dst_slice_start = slice_start;
c->dst_slice_height = slice_height;
avpriv_slicethread_execute(c->slicethread, nb_jobs, 0);
avpriv_slicethread_execute2(c->slicethread, nb_jobs, 0);
for (int i = 0; i < c->nb_slice_ctx; i++) {
if (c->slice_err[i] < 0) {
@@ -1655,8 +1655,8 @@ int attribute_align_arg sws_scale(SwsContext *sws,
dst, dstStride, 0, sws->dst_h);
}
void ff_sws_slice_worker(void *priv, int jobnr, int threadnr,
int nb_jobs, int nb_threads)
int ff_sws_slice_worker(void *priv, int jobnr, int threadnr,
int nb_jobs, int nb_threads)
{
SwsInternal *parent = priv;
SwsContext *sws = parent->slice_ctx[threadnr];
@@ -1686,4 +1686,5 @@ void ff_sws_slice_worker(void *priv, int jobnr, int threadnr,
}
parent->slice_err[threadnr] = err;
return 0;
}

4
libswscale/swscale_internal.h
View File

@@ -1189,8 +1189,8 @@ void ff_init_vscale_pfn(SwsInternal *c, yuv2planar1_fn yuv2plane1, yuv2planarX_f
yuv2interleavedX_fn yuv2nv12cX, yuv2packed1_fn yuv2packed1, yuv2packed2_fn yuv2packed2,
yuv2packedX_fn yuv2packedX, yuv2anyX_fn yuv2anyX, int use_mmx);
void ff_sws_slice_worker(void *priv, int jobnr, int threadnr,
int nb_jobs, int nb_threads);
int ff_sws_slice_worker(void *priv, int jobnr, int threadnr,
int nb_jobs, int nb_threads);
int ff_swscale(SwsInternal *c, const uint8_t *const src[], const int srcStride[],
int srcSliceY, int srcSliceH, uint8_t *const dst[],

4
libswscale/utils.c
View File

@@ -1840,8 +1840,8 @@ static int context_init_threaded(SwsContext *sws,
SwsInternal *c = sws_internal(sws);
int ret;
ret = avpriv_slicethread_create(&c->slicethread, (void*) sws,
ff_sws_slice_worker, NULL, sws->threads);
ret = avpriv_slicethread_create2(&c->slicethread, (void*) sws,
ff_sws_slice_worker, NULL, sws->threads);
if (ret == AVERROR(ENOSYS)) {
sws->threads = 1;
return 0;