Coverage Report

Created: 2026-08-17 07:50

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
/src/ffmpeg/libavutil/slicethread.c
Line
Count
Source
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 "cpu.h"
21
#include "internal.h"
22
#include "slicethread.h"
23
#include "mem.h"
24
#include "thread.h"
25
#include "avassert.h"
26
27
#define MAX_AUTO_THREADS 16
28
29
#if HAVE_PTHREADS || HAVE_W32THREADS || HAVE_OS2THREADS
30
31
typedef struct WorkerContext {
32
    AVSliceThread   *ctx;
33
    pthread_mutex_t mutex;
34
    pthread_cond_t  cond;
35
    pthread_t       thread;
36
    int             done;
37
} WorkerContext;
38
39
struct AVSliceThread {
40
    WorkerContext   *workers;
41
    int             nb_threads;
42
    int             nb_active_threads;
43
    int             nb_jobs;
44
45
    atomic_uint     first_job;
46
    atomic_uint     current_job;
47
    pthread_mutex_t done_mutex;
48
    pthread_cond_t  done_cond;
49
    int             done;
50
    int             finished;
51
    atomic_int      error;
52
53
    void            *priv;
54
    int            (*worker_func)(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads);
55
    int            (*main_func)(void *priv);
56
57
#if LIBAVUTIL_VERSION_MAJOR < 62
58
    void           (*worker_func_v1)(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads);
59
    void           (*main_func_v1)(void *priv);
60
    void            *priv_v1;
61
#endif
62
};
63
64
static int run_jobs(AVSliceThread *ctx)
65
0
{
66
0
    unsigned nb_jobs    = ctx->nb_jobs;
67
0
    unsigned nb_active_threads = ctx->nb_active_threads;
68
0
    unsigned first_job    = atomic_fetch_add_explicit(&ctx->first_job, 1, memory_order_acq_rel);
69
0
    unsigned current_job  = first_job;
70
71
0
    do {
72
0
        int ret = atomic_load_explicit(&ctx->error, memory_order_relaxed);
73
0
        if (ret)
74
0
            continue;
75
0
        ret = ctx->worker_func(ctx->priv, current_job, first_job, nb_jobs, nb_active_threads);
76
0
        if (ret) {
77
0
            int prev = 0;
78
0
            atomic_compare_exchange_strong_explicit(&ctx->error, &prev, ret,
79
0
                                                    memory_order_relaxed,
80
0
                                                    memory_order_relaxed);
81
0
        }
82
0
    } while ((current_job = atomic_fetch_add_explicit(&ctx->current_job, 1, memory_order_acq_rel)) < nb_jobs);
83
84
0
    return current_job == nb_jobs + nb_active_threads - 1;
85
0
}
86
87
static void *attribute_align_arg thread_worker(void *v)
88
0
{
89
0
    WorkerContext *w = v;
90
0
    AVSliceThread *ctx = w->ctx;
91
92
0
    pthread_mutex_lock(&w->mutex);
93
0
    pthread_cond_signal(&w->cond);
94
95
0
    while (1) {
96
0
        w->done = 1;
97
0
        while (w->done)
98
0
            pthread_cond_wait(&w->cond, &w->mutex);
99
100
0
        if (ctx->finished) {
101
0
            pthread_mutex_unlock(&w->mutex);
102
0
            return NULL;
103
0
        }
104
105
0
        if (run_jobs(ctx)) {
106
0
            pthread_mutex_lock(&ctx->done_mutex);
107
0
            ctx->done = 1;
108
0
            pthread_cond_signal(&ctx->done_cond);
109
0
            pthread_mutex_unlock(&ctx->done_mutex);
110
0
        }
111
0
    }
112
0
}
113
114
av_cold
115
int avpriv_slicethread_create2(AVSliceThread **pctx, void *priv,
116
                               int (*worker_func)(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads),
117
                               int (*main_func)(void *priv),
118
                               int nb_threads)
119
0
{
120
0
    AVSliceThread *ctx;
121
0
    int nb_workers, i;
122
0
    int ret;
123
124
0
    av_assert0(nb_threads >= 0);
125
0
    if (!nb_threads) {
126
0
        int nb_cpus = av_cpu_count();
127
0
        if (nb_cpus > 1)
128
0
            nb_threads = FFMIN(nb_cpus + 1, MAX_AUTO_THREADS);
129
0
        else
130
0
            nb_threads = 1;
131
0
    }
132
133
0
    nb_workers = nb_threads;
134
0
    if (!main_func)
135
0
        nb_workers--;
136
137
0
    *pctx = ctx = av_mallocz(sizeof(*ctx));
138
0
    if (!ctx)
139
0
        return AVERROR(ENOMEM);
140
141
0
    if (nb_workers && !(ctx->workers = av_calloc(nb_workers, sizeof(*ctx->workers)))) {
142
0
        av_freep(pctx);
143
0
        return AVERROR(ENOMEM);
144
0
    }
145
146
0
    ctx->priv        = priv;
147
0
    ctx->worker_func = worker_func;
148
0
    ctx->main_func   = main_func;
149
0
    ctx->nb_threads  = nb_threads;
150
0
    ctx->nb_active_threads = 0;
151
0
    ctx->nb_jobs     = 0;
152
0
    ctx->finished    = 0;
153
154
0
    atomic_init(&ctx->first_job, 0);
155
0
    atomic_init(&ctx->current_job, 0);
156
0
    ret = pthread_mutex_init(&ctx->done_mutex, NULL);
157
0
    if (ret) {
158
0
        av_freep(&ctx->workers);
159
0
        av_freep(pctx);
160
0
        return AVERROR(ret);
161
0
    }
162
0
    ret = pthread_cond_init(&ctx->done_cond, NULL);
163
0
    if (ret) {
164
0
        ctx->nb_threads = main_func ? 0 : 1;
165
0
        avpriv_slicethread_free(pctx);
166
0
        return AVERROR(ret);
167
0
    }
168
0
    ctx->done        = 0;
169
170
0
    for (i = 0; i < nb_workers; i++) {
171
0
        WorkerContext *w = &ctx->workers[i];
172
0
        w->ctx = ctx;
173
0
        ret = pthread_mutex_init(&w->mutex, NULL);
174
0
        if (ret) {
175
0
            ctx->nb_threads = main_func ? i : i + 1;
176
0
            avpriv_slicethread_free(pctx);
177
0
            return AVERROR(ret);
178
0
        }
179
0
        ret = pthread_cond_init(&w->cond, NULL);
180
0
        if (ret) {
181
0
            pthread_mutex_destroy(&w->mutex);
182
0
            ctx->nb_threads = main_func ? i : i + 1;
183
0
            avpriv_slicethread_free(pctx);
184
0
            return AVERROR(ret);
185
0
        }
186
0
        pthread_mutex_lock(&w->mutex);
187
0
        w->done = 0;
188
189
0
        if (ret = pthread_create(&w->thread, NULL, thread_worker, w)) {
190
0
            ctx->nb_threads = main_func ? i : i + 1;
191
0
            pthread_mutex_unlock(&w->mutex);
192
0
            pthread_cond_destroy(&w->cond);
193
0
            pthread_mutex_destroy(&w->mutex);
194
0
            avpriv_slicethread_free(pctx);
195
0
            return AVERROR(ret);
196
0
        }
197
198
0
        while (!w->done)
199
0
            pthread_cond_wait(&w->cond, &w->mutex);
200
0
        pthread_mutex_unlock(&w->mutex);
201
0
    }
202
203
0
    return nb_threads;
204
0
}
205
206
int avpriv_slicethread_execute2(AVSliceThread *ctx, int nb_jobs, int execute_main)
207
0
{
208
0
    int nb_workers, i, is_last = 0, ret = 0;
209
210
0
    av_assert0(nb_jobs > 0);
211
0
    ctx->nb_jobs           = nb_jobs;
212
0
    ctx->nb_active_threads = FFMIN(nb_jobs, ctx->nb_threads);
213
0
    atomic_store_explicit(&ctx->error, 0, memory_order_relaxed);
214
0
    atomic_store_explicit(&ctx->first_job, 0, memory_order_relaxed);
215
0
    atomic_store_explicit(&ctx->current_job, ctx->nb_active_threads, memory_order_relaxed);
216
0
    nb_workers             = ctx->nb_active_threads;
217
0
    if (!ctx->main_func || !execute_main)
218
0
        nb_workers--;
219
220
0
    for (i = 0; i < nb_workers; i++) {
221
0
        WorkerContext *w = &ctx->workers[i];
222
0
        pthread_mutex_lock(&w->mutex);
223
0
        w->done = 0;
224
0
        pthread_cond_signal(&w->cond);
225
0
        pthread_mutex_unlock(&w->mutex);
226
0
    }
227
228
0
    if (ctx->main_func && execute_main) {
229
0
        ret = ctx->main_func(ctx->priv);
230
0
    } else
231
0
        is_last = run_jobs(ctx);
232
233
0
    if (!is_last) {
234
0
        pthread_mutex_lock(&ctx->done_mutex);
235
0
        while (!ctx->done)
236
0
            pthread_cond_wait(&ctx->done_cond, &ctx->done_mutex);
237
0
        ctx->done = 0;
238
0
        pthread_mutex_unlock(&ctx->done_mutex);
239
0
    }
240
241
0
    if (!ret)
242
0
        ret = atomic_load_explicit(&ctx->error, memory_order_relaxed);
243
244
0
    return ret;
245
0
}
246
247
av_cold void avpriv_slicethread_free(AVSliceThread **pctx)
248
0
{
249
0
    AVSliceThread *ctx = *pctx;
250
0
    int nb_workers, i;
251
252
0
    if (!ctx)
253
0
        return;
254
255
0
    nb_workers = ctx->nb_threads;
256
0
    if (!ctx->main_func)
257
0
        nb_workers--;
258
259
0
    ctx->finished = 1;
260
0
    for (i = 0; i < nb_workers; i++) {
261
0
        WorkerContext *w = &ctx->workers[i];
262
0
        pthread_mutex_lock(&w->mutex);
263
0
        w->done = 0;
264
0
        pthread_cond_signal(&w->cond);
265
0
        pthread_mutex_unlock(&w->mutex);
266
0
    }
267
268
0
    for (i = 0; i < nb_workers; i++) {
269
0
        WorkerContext *w = &ctx->workers[i];
270
0
        pthread_join(w->thread, NULL);
271
0
        pthread_cond_destroy(&w->cond);
272
0
        pthread_mutex_destroy(&w->mutex);
273
0
    }
274
275
0
    pthread_cond_destroy(&ctx->done_cond);
276
0
    pthread_mutex_destroy(&ctx->done_mutex);
277
0
    av_freep(&ctx->workers);
278
0
    av_freep(pctx);
279
0
}
280
281
#else /* HAVE_PTHREADS || HAVE_W32THREADS || HAVE_OS32THREADS */
282
283
int avpriv_slicethread_create2(AVSliceThread **pctx, void *priv,
284
                               int (*worker_func)(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads),
285
                               int (*main_func)(void *priv),
286
                               int nb_threads)
287
{
288
    *pctx = NULL;
289
    return AVERROR(ENOSYS);
290
}
291
292
int avpriv_slicethread_execute2(AVSliceThread *ctx, int nb_jobs, int execute_main)
293
{
294
    av_assert0(0);
295
}
296
297
void avpriv_slicethread_free(AVSliceThread **pctx)
298
{
299
    av_assert0(!pctx || !*pctx);
300
}
301
302
#endif /* HAVE_PTHREADS || HAVE_W32THREADS || HAVE_OS32THREADS */
303
304
/**
305
 * Backwards compatibility wrapper for the deprecated avpriv_ slicethread API.
306
 */
307
308
#if LIBAVUTIL_VERSION_MAJOR < 62
309
310
static int wrapper_worker(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads)
311
0
{
312
0
    AVSliceThread *ctx = priv;
313
0
    ctx->worker_func_v1(ctx->priv_v1, jobnr, threadnr, nb_jobs, nb_threads);
314
0
    return 0;
315
0
}
316
317
static int wrapper_main(void *priv)
318
0
{
319
0
    AVSliceThread *ctx = priv;
320
0
    ctx->main_func_v1(ctx->priv_v1);
321
0
    return 0;
322
0
}
323
324
int avpriv_slicethread_create(AVSliceThread **pctx, void *priv,
325
                              void (*worker_func)(void *priv, int jobnr, int threadnr, int nb_jobs, int nb_threads),
326
                              void (*main_func)(void *priv),
327
                              int nb_threads)
328
0
{
329
0
    int ret = avpriv_slicethread_create2(pctx, NULL, wrapper_worker,
330
0
                                         main_func ? wrapper_main : NULL,
331
0
                                         nb_threads);
332
0
    if (ret < 0)
333
0
        return ret;
334
335
0
    (*pctx)->priv           = *pctx;
336
0
    (*pctx)->priv_v1        = priv;
337
0
    (*pctx)->worker_func_v1 = worker_func;
338
0
    (*pctx)->main_func_v1   = main_func;
339
0
    return ret;
340
0
}
341
342
void avpriv_slicethread_execute(AVSliceThread *ctx, int nb_jobs, int execute_main)
343
0
{
344
0
    avpriv_slicethread_execute2(ctx, nb_jobs, execute_main);
345
0
}
346
347
#endif /* LIBAVUTIL_VERSION_MAJOR < 62 */