diff --git a/libavcodec/pthread_slice.c b/libavcodec/pthread_slice.c index f9da670735..69a4316a86 100644 --- a/libavcodec/pthread_slice.c +++ b/libavcodec/pthread_slice.c @@ -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; diff --git a/libavfilter/pthread.c b/libavfilter/pthread.c index 06590fe65a..22f4242f47 100644 --- a/libavfilter/pthread.c +++ b/libavfilter/pthread.c @@ -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); diff --git a/libavutil/slicethread.c b/libavutil/slicethread.c index 0a2d6662e1..65566f0169 100644 --- a/libavutil/slicethread.c +++ b/libavutil/slicethread.c @@ -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 */ diff --git a/libavutil/slicethread.h b/libavutil/slicethread.h index f6f6f302c4..9b560b78a7 100644 --- a/libavutil/slicethread.h +++ b/libavutil/slicethread.h @@ -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 diff --git a/libswscale/cms.c b/libswscale/cms.c index c7b196616e..6d283ea1d3 100644 --- a/libswscale/cms.c +++ b/libswscale/cms.c @@ -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; } diff --git a/libswscale/graph.c b/libswscale/graph.c index 0f444199f7..558d422531 100644 --- a/libswscale/graph.c +++ b/libswscale/graph.c @@ -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); } } diff --git a/libswscale/swscale.c b/libswscale/swscale.c index a01354f4c6..ad0ccf63d0 100644 --- a/libswscale/swscale.c +++ b/libswscale/swscale.c @@ -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; } diff --git a/libswscale/swscale_internal.h b/libswscale/swscale_internal.h index 9a822cb8e4..58d441822e 100644 --- a/libswscale/swscale_internal.h +++ b/libswscale/swscale_internal.h @@ -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[], diff --git a/libswscale/utils.c b/libswscale/utils.c index 08424aa56b..d96ece2ac1 100644 --- a/libswscale/utils.c +++ b/libswscale/utils.c @@ -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;