Google Git Google Git
Sign in
chromium/external/code.videolan.org/videolan/dav2d/refs/heads/upstream/main/./src/thread_task.c
blob: a708a2a935c3bea844ca39cdeec5c22e466075d8 [file] [edit]
/*
* Copyright © 2018-2026, VideoLAN and dav2d authors
* Copyright © 2018-2026, Two Orioles, LLC
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
*
* 1. Redistributions of source code must retain the above copyright notice, this
* list of conditions and the following disclaimer.
*
* 2. Redistributions in binary form must reproduce the above copyright notice,
* this list of conditions and the following disclaimer in the documentation
* and/or other materials provided with the distribution.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
* DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
* ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
* (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
* LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
* ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
* SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/
#include "config.h"
#include "common/frame.h"
#include "src/thread_task.h"
#include "src/fg_apply.h"
// This function resets the cur pointer to the first frame theoretically
// executable after a task completed (ie. each time we update some progress or
// insert some tasks in the queue).
// When frame_idx is set, it can be either from a completed task, or from tasks
// inserted in the queue, in which case we have to make sure the cur pointer
// isn't past this insert.
// The special case where frame_idx is UINT_MAX is to handle the reset after
// completing a task and locklessly signaling progress. In this case we don't
// enter a critical section, which is needed for this function, so we set an
// atomic for a delayed handling, happening here. Meaning we can call this
// function without any actual update other than what's in the atomic, hence
// this special case.
static inline int reset_task_cur(const Dav2dContext *const c,
struct TaskThreadData *const ttd,
unsigned frame_idx)
{
const unsigned first = atomic_load(&ttd->first);
unsigned reset_frame_idx = atomic_exchange(&ttd->reset_task_cur, UINT_MAX);
if (reset_frame_idx < first) {
if (frame_idx == UINT_MAX) return 0;
reset_frame_idx = UINT_MAX;
}
if (!ttd->cur && c->fc[first].task_thread.task_cur_prev == NULL)
return 0;
if (reset_frame_idx != UINT_MAX) {
if (frame_idx == UINT_MAX) {
if (reset_frame_idx > first + ttd->cur)
return 0;
ttd->cur = reset_frame_idx - first;
goto cur_found;
}
} else if (frame_idx == UINT_MAX)
return 0;
if (frame_idx < first) frame_idx += c->n_fc;
const unsigned min_frame_idx = umin(reset_frame_idx, frame_idx);
const unsigned cur_frame_idx = first + ttd->cur;
if (ttd->cur < c->n_fc && cur_frame_idx < min_frame_idx)
return 0;
for (ttd->cur = min_frame_idx - first; ttd->cur < c->n_fc; ttd->cur++)
if (c->fc[(first + ttd->cur) % c->n_fc].task_thread.task_head)
break;
cur_found:
for (unsigned i = ttd->cur; i < c->n_fc; i++)
c->fc[(first + i) % c->n_fc].task_thread.task_cur_prev = NULL;
return 1;
}
static inline void reset_task_cur_async(struct TaskThreadData *const ttd,
unsigned frame_idx, unsigned n_frames)
{
const unsigned first = atomic_load(&ttd->first);
if (frame_idx < first) frame_idx += n_frames;
unsigned last_idx = frame_idx;
do {
frame_idx = last_idx;
last_idx = atomic_exchange(&ttd->reset_task_cur, frame_idx);
} while (last_idx < frame_idx);
if (frame_idx == first && atomic_load(&ttd->first) != first) {
unsigned expected = frame_idx;
atomic_compare_exchange_strong(&ttd->reset_task_cur, &expected, UINT_MAX);
}
}
static void insert_tasks_between(Dav2dFrameContext *const f,
Dav2dTask *const first, Dav2dTask *const last,
Dav2dTask *const a, Dav2dTask *const b,
const int cond_signal)
{
struct TaskThreadData *const ttd = f->task_thread.ttd;
if (atomic_load(f->c->flush)) return;
assert(!a || a->next == b);
if (!a) f->task_thread.task_head = first;
else a->next = first;
if (!b) f->task_thread.task_tail = last;
last->next = b;
reset_task_cur(f->c, ttd, first->frame_idx);
if (cond_signal && !atomic_fetch_or(&ttd->cond_signaled, 1))
pthread_cond_signal(&ttd->cond);
}
static void insert_tasks(Dav2dFrameContext *const f,
Dav2dTask *const first, Dav2dTask *const last,
const int cond_signal)
{
// insert task back into task queue
Dav2dTask *t_ptr, *prev_t = NULL;
for (t_ptr = f->task_thread.task_head;
t_ptr; prev_t = t_ptr, t_ptr = t_ptr->next)
{
// entropy coding precedes other steps
if (t_ptr->type == DAV2D_TASK_TYPE_TILE_ENTROPY) {
if (first->type > DAV2D_TASK_TYPE_TILE_ENTROPY) continue;
// both are entropy
if (first->by64 > t_ptr->by64) continue;
if (first->by64 < t_ptr->by64) {
insert_tasks_between(f, first, last, prev_t, t_ptr, cond_signal);
return;
}
// same sby
} else {
if (first->type == DAV2D_TASK_TYPE_TILE_ENTROPY) {
insert_tasks_between(f, first, last, prev_t, t_ptr, cond_signal);
return;
}
if (first->by64 > t_ptr->by64) continue;
if (first->by64 < t_ptr->by64) {
insert_tasks_between(f, first, last, prev_t, t_ptr, cond_signal);
return;
}
// same sby
if (first->type > t_ptr->type) continue;
if (first->type < t_ptr->type) {
insert_tasks_between(f, first, last, prev_t, t_ptr, cond_signal);
return;
}
// same task type
}
// sort by tile-id
assert(first->type == DAV2D_TASK_TYPE_TILE_RECONSTRUCTION ||
first->type == DAV2D_TASK_TYPE_TILE_MV_RESOLUTION ||
first->type == DAV2D_TASK_TYPE_TILE_ENTROPY);
assert(first->type == t_ptr->type);
assert(t_ptr->by64 == first->by64);
const int pass = first->type == DAV2D_TASK_TYPE_TILE_RECONSTRUCTION ?
f->c->task_thread.n_passes - 1 :
first->type != DAV2D_TASK_TYPE_TILE_ENTROPY;
const int t_tile_idx = (int) (first - f->task_thread.tile_tasks[pass]);
const int p_tile_idx = (int) (t_ptr - f->task_thread.tile_tasks[pass]);
assert(t_tile_idx != p_tile_idx);
if (t_tile_idx > p_tile_idx) continue;
insert_tasks_between(f, first, last, prev_t, t_ptr, cond_signal);
return;
}
// append at the end
insert_tasks_between(f, first, last, prev_t, NULL, cond_signal);
}
static inline void insert_task(Dav2dFrameContext *const f,
Dav2dTask *const t, const int cond_signal)
{
insert_tasks(f, t, t, cond_signal);
}
static inline void add_pending(Dav2dFrameContext *const f, Dav2dTask *const t) {
pthread_mutex_lock(&f->task_thread.pending_tasks.lock);
t->next = NULL;
if (!f->task_thread.pending_tasks.head)
f->task_thread.pending_tasks.head = t;
else
f->task_thread.pending_tasks.tail->next = t;
f->task_thread.pending_tasks.tail = t;
atomic_store(&f->task_thread.pending_tasks.merge, 1);
pthread_mutex_unlock(&f->task_thread.pending_tasks.lock);
}
static inline int merge_pending_frame(Dav2dFrameContext *const f) {
int const merge = atomic_load(&f->task_thread.pending_tasks.merge);
if (merge) {
pthread_mutex_lock(&f->task_thread.pending_tasks.lock);
Dav2dTask *t = f->task_thread.pending_tasks.head;
f->task_thread.pending_tasks.head = NULL;
f->task_thread.pending_tasks.tail = NULL;
atomic_store(&f->task_thread.pending_tasks.merge, 0);
pthread_mutex_unlock(&f->task_thread.pending_tasks.lock);
while (t) {
Dav2dTask *const tmp = t->next;
insert_task(f, t, 0);
t = tmp;
}
}
return merge;
}
static inline int merge_pending(const Dav2dContext *const c) {
int res = 0;
for (unsigned i = 0; i < c->n_fc; i++)
res |= merge_pending_frame(&c->fc[i]);
return res;
}
static int create_filter_sbrow(Dav2dFrameContext *const f,
const int is_entropy_pass, Dav2dTask **res_t)
{
const int has_deblock = f->frame_hdr->deblock.level_y[0] ||
f->frame_hdr->deblock.level_y[1];
const int has_cdef = f->lf.cdef_planes;
const int has_lr = f->lf.restore_planes;
Dav2dTask *tasks = f->task_thread.tasks;
const int n_passes = f->c->task_thread.n_passes;
int num_tasks = f->sb64h + (n_passes > 1) * f->sbh;
if (num_tasks > f->task_thread.num_tasks) {
const size_t size = sizeof(Dav2dTask) * num_tasks;
tasks = dav2d_realloc(ALLOC_COMMON_CTX, f->task_thread.tasks, size);
if (!tasks) return -1;
memset(tasks, 0, size);
f->task_thread.tasks = tasks;
f->task_thread.num_tasks = num_tasks;
}
tasks += f->sb64h * is_entropy_pass;
if (is_entropy_pass) {
f->frame_thread.entropy_progress = 0;
} else {
const int prog_sz = ((f->sb64h + 31) & ~31) >> 5;
if (prog_sz > f->frame_thread.prog_sz) {
atomic_uint *const prog = dav2d_realloc(ALLOC_COMMON_CTX, f->frame_thread.frame_progress,
2 * prog_sz * sizeof(*prog));
if (!prog) return -1;
f->frame_thread.frame_progress = prog;
f->frame_thread.copy_db_progress = prog + prog_sz;
}
f->frame_thread.prog_sz = prog_sz;
memset(f->frame_thread.frame_progress, 0, prog_sz * sizeof(atomic_uint));
memset(f->frame_thread.copy_db_progress, 0, prog_sz * sizeof(atomic_uint));
atomic_store(&f->frame_thread.deblock_progress, 0);
}
f->frame_thread.next_tile_row[is_entropy_pass] = 0;
Dav2dTask *t = &tasks[0];
t->by64 = 0;
t->recon_progress = 1;
t->deblock_progress = 0;
t->type = is_entropy_pass ? DAV2D_TASK_TYPE_ENTROPY_PROGRESS :
has_deblock ? DAV2D_TASK_TYPE_DEBLOCK_COLS :
has_cdef || has_lr /* i.e. LR backup */ ? DAV2D_TASK_TYPE_DEBLOCK_ROWS :
DAV2D_TASK_TYPE_RECONSTRUCTION_PROGRESS;
t->frame_idx = (int)(f - f->c->fc);
*res_t = t;
return 0;
}
int dav2d_task_create_tile_sbrow(Dav2dFrameContext *const f, const int pass,
const int cond_signal)
{
Dav2dTask *tasks = f->task_thread.tile_tasks[0];
const int n_passes = f->c->task_thread.n_passes;
const int num_tasks = f->frame_hdr->tiling.t.cols * f->frame_hdr->tiling.t.rows;
if (!pass) {
int alloc_num_tasks = num_tasks * n_passes;
if (alloc_num_tasks > f->task_thread.num_tile_tasks) {
const size_t size = sizeof(Dav2dTask) * alloc_num_tasks;
tasks = dav2d_realloc(ALLOC_COMMON_CTX, f->task_thread.tile_tasks[0], size);
if (!tasks) return -1;
memset(tasks, 0, size);
f->task_thread.tile_tasks[0] = tasks;
f->task_thread.num_tile_tasks = alloc_num_tasks;
}
f->task_thread.tile_tasks[1] = tasks + num_tasks;
f->task_thread.tile_tasks[2] = tasks + num_tasks * 2;
}
tasks += num_tasks * pass;
Dav2dTask *pf_t = NULL;
if (!(n_passes == 3 && pass == 1) &&
create_filter_sbrow(f, pass + 1 != n_passes, &pf_t))
{
return -1;
}
Dav2dTask *prev_t = NULL;
for (int tile_idx = 0; tile_idx < num_tasks; tile_idx++) {
Dav2dTileState *const ts = &f->ts[tile_idx];
Dav2dTask *t = &tasks[tile_idx];
t->by64 = ts->tiling.row_start >> 4;
if (pf_t && t->by64) {
prev_t->next = pf_t;
prev_t = pf_t;
pf_t = NULL;
}
t->recon_progress = 0;
t->deblock_progress = 0;
t->deps_skip = 0;
t->type = pass + 1 == n_passes ? DAV2D_TASK_TYPE_TILE_RECONSTRUCTION :
!pass ? DAV2D_TASK_TYPE_TILE_ENTROPY :
DAV2D_TASK_TYPE_TILE_MV_RESOLUTION;
t->frame_idx = (int)(f - f->c->fc);
if (prev_t) prev_t->next = t;
prev_t = t;
}
if (pf_t) {
prev_t->next = pf_t;
prev_t = pf_t;
}
prev_t->next = NULL;
if (!pass || pass + 1 == n_passes)
atomic_store(&f->task_thread.done[pass + 1 != n_passes], 0);
// XXX in theory this could be done locklessly, at this point they are no
// tasks in the frameQ, so no other runner should be using this lock, but
// we must add both passes at once
pthread_mutex_lock(&f->task_thread.pending_tasks.lock);
assert(f->task_thread.pending_tasks.head == NULL || pass > 0);
if (!f->task_thread.pending_tasks.head)
f->task_thread.pending_tasks.head = &tasks[0];
else
f->task_thread.pending_tasks.tail->next = &tasks[0];
f->task_thread.pending_tasks.tail = prev_t;
atomic_store(&f->task_thread.pending_tasks.merge, 1);
atomic_store(&f->task_thread.init_done, 1);
pthread_mutex_unlock(&f->task_thread.pending_tasks.lock);
return 0;
}
void dav2d_task_frame_init(Dav2dFrameContext *const f) {
const Dav2dContext *const c = f->c;
atomic_store(&f->task_thread.init_done, 0);
// schedule init task, which will schedule the remaining tasks
Dav2dTask *const t = &f->task_thread.init_task;
t->type = DAV2D_TASK_TYPE_INIT;
t->frame_idx = (int)(f - c->fc);
t->by64 = 0;
t->recon_progress = t->deblock_progress = 0;
insert_task(f, t, 1);
}
void dav2d_task_delayed_fg(Dav2dContext *const c, Dav2dPicture *const out,
const Dav2dPicture *const in)
{
struct TaskThreadData *const ttd = &c->task_thread;
ttd->delayed_fg.in = in;
ttd->delayed_fg.out = out;
ttd->delayed_fg.type = DAV2D_TASK_TYPE_FG_PREP;
atomic_init(&ttd->delayed_fg.progress[0], 0);
atomic_init(&ttd->delayed_fg.progress[1], 0);
pthread_mutex_lock(&ttd->lock);
ttd->delayed_fg.exec = 1;
ttd->delayed_fg.finished = 0;
pthread_cond_signal(&ttd->cond);
do {
pthread_cond_wait(&ttd->delayed_fg.cond, &ttd->lock);
} while (!ttd->delayed_fg.finished);
pthread_mutex_unlock(&ttd->lock);
}
static inline int ensure_progress(struct TaskThreadData *const ttd,
Dav2dFrameContext *const f,
Dav2dTask *const t, const enum TaskType type,
atomic_int *const state, int *const target)
{
// deblock_rows (non-LR portion) depends on deblock of previous sbrow,
// so ensure that completed. if not, re-add to task-queue; else, fall-through
int p1 = atomic_load(state);
if (p1 < t->by64) {
t->type = type;
t->recon_progress = t->deblock_progress = 0;
*target = t->by64;
add_pending(f, t);
pthread_mutex_lock(&ttd->lock);
return 1;
}
return 0;
}
static inline int check_tile(Dav2dTask *const t, Dav2dFrameContext *const f) {
const int n_passes = f->c->task_thread.n_passes;
const int pass = t->type == DAV2D_TASK_TYPE_TILE_RECONSTRUCTION ?
n_passes - 1 : t->type != DAV2D_TASK_TYPE_TILE_ENTROPY;
const int tile_idx = (int)(t - f->task_thread.tile_tasks[pass]);
Dav2dTileState *const ts = &f->ts[tile_idx];
const int p1 = atomic_load(&ts->progress[pass]);
if (p1 < t->by64) return 1;
int error = p1 == TILE_ERROR;
error |= atomic_fetch_or(&f->task_thread.error, error);
if (!error && pass > 0) {
const int p2 = atomic_load(&ts->progress[pass - 1]);
if (p2 <= t->by64) return 1;
error = p2 == TILE_ERROR;
error |= atomic_fetch_or(&f->task_thread.error, error);
}
if (!error && f->c->n_fc > 1 && !IS_KEY_OR_INTRA(f->frame_hdr) &&
!(t->type == DAV2D_TASK_TYPE_TILE_ENTROPY && !f->entropyref_mask) &&
!(t->type == DAV2D_TASK_TYPE_TILE_MV_RESOLUTION && !f->rf.mfmv_mask))
{
// check reference state
const Dav2dThreadPicture *p = &f->cur;
const int sbsz = f->frame_hdr->sbsz, sb64sz = 1 << sbsz;
const int ss_ver = p->p.p.layout == DAV2D_PIXEL_LAYOUT_I420;
const unsigned p_b = (t->by64 + sb64sz) << 6;
const int tile_sby = (t->by64 >> f->frame_hdr->sbsz) -
(ts->tiling.row_start >> f->sb_shift);
const int (*const lowest_px)[2] = ts->lowest_pixel[tile_sby];
for (int n = t->deps_skip; n < f->frame_hdr->n_ref_frames;
n++, t->deps_skip++)
{
unsigned lowest;
if (t->type != DAV2D_TASK_TYPE_TILE_RECONSTRUCTION) {
if (!((t->type == DAV2D_TASK_TYPE_TILE_ENTROPY ?
f->entropyref_mask : f->rf.mfmv_mask) & (1 << n)))
{
continue;
}
lowest = p_b;
} else {
// +8 is postfilter-induced delay
const int y = lowest_px[n][0] == INT_MIN ? INT_MIN :
lowest_px[n][0] + 8;
const int uv = lowest_px[n][1] == INT_MIN ? INT_MIN :
lowest_px[n][1] * (1 << ss_ver) + 8;
int max = imax(y, uv);
if (n_passes == 2 && f->rf.mfmv_mask & (1 << n))
max = imax(max, p_b);
if (max == INT_MIN) continue;
lowest = iclip(max, 1, f->refp[n].p.p.h);
}
const int p_idx = pass * (1 + (n_passes == 2));
const unsigned p3 = atomic_load(&f->refp[n].progress[p_idx]);
if (p3 < lowest) return 1;
atomic_fetch_or(&f->task_thread.error, p3 == FRAME_ERROR);
}
}
return 0;
}
static inline int get_frame_progress(const Dav2dContext *const c,
const Dav2dFrameContext *const f)
{
unsigned frame_prog = c->n_fc > 1 ? atomic_load(&f->cur.progress[2]) : 0;
if (frame_prog >= FRAME_ERROR)
return f->sb64h - 1;
int idx = frame_prog >> 11;
int prog;
do {
atomic_uint *state = &f->frame_thread.frame_progress[idx];
const unsigned val = ~atomic_load(state);
prog = val ? ctz(val) : 32;
if (prog != 32) break;
prog = 0;
} while (++idx < f->frame_thread.prog_sz);
return ((idx << 5) | prog) - 1;
}
static inline void abort_frame(Dav2dFrameContext *const f, const int error) {
atomic_store(&f->task_thread.error, error == DAV2D_ERR(EINVAL) ? 1 : -1);
atomic_store(&f->task_thread.task_counter, 0);
atomic_store(&f->task_thread.done[0], 1);
atomic_store(&f->task_thread.done[1], 1);
atomic_store(&f->cur.progress[0], FRAME_ERROR);
atomic_store(&f->cur.progress[1], FRAME_ERROR);
atomic_store(&f->cur.progress[2], FRAME_ERROR);
dav2d_decode_frame_exit(f, error);
f->n_tile_data = 0;
pthread_cond_signal(&f->task_thread.cond);
}
static inline void delayed_fg_task(const Dav2dContext *const c,
struct TaskThreadData *const ttd)
{
const Dav2dPicture *const in = ttd->delayed_fg.in;
Dav2dPicture *const out = ttd->delayed_fg.out;
#if CONFIG_16BPC
int off;
if (out->p.bpc != 8)
off = (out->p.bpc >> 1) - 4;
#endif
switch (ttd->delayed_fg.type) {
case DAV2D_TASK_TYPE_FG_PREP:
ttd->delayed_fg.exec = 0;
if (atomic_load(&ttd->cond_signaled))
pthread_cond_signal(&ttd->cond);
pthread_mutex_unlock(&ttd->lock);
switch (out->p.bpc) {
#if CONFIG_8BPC
case 8:
dav2d_prep_grain_8bpc(&c->dsp[0].fg, out, in,
ttd->delayed_fg.scaling_8bpc,
ttd->delayed_fg.grain_lut_8bpc);
break;
#endif
#if CONFIG_16BPC
case 10:
case 12:
dav2d_prep_grain_16bpc(&c->dsp[off].fg, out, in,
ttd->delayed_fg.scaling_16bpc,
ttd->delayed_fg.grain_lut_16bpc);
break;
#endif
default: abort();
}
ttd->delayed_fg.type = DAV2D_TASK_TYPE_FG_APPLY;
pthread_mutex_lock(&ttd->lock);
ttd->delayed_fg.exec = 1;
// fall-through
case DAV2D_TASK_TYPE_FG_APPLY:;
int row = atomic_fetch_add(&ttd->delayed_fg.progress[0], 1);
pthread_mutex_unlock(&ttd->lock);
const int bs = 16 << out->fgm->block_size;
int progmax = (out->p.h + bs - 1) / bs;
while (row < progmax) {
if (row + 1 < progmax)
pthread_cond_signal(&ttd->cond);
else {
pthread_mutex_lock(&ttd->lock);
ttd->delayed_fg.exec = 0;
pthread_mutex_unlock(&ttd->lock);
}
switch (out->p.bpc) {
#if CONFIG_8BPC
case 8:
dav2d_apply_grain_row_8bpc(&c->dsp[0].fg, out, in,
ttd->delayed_fg.scaling_8bpc,
ttd->delayed_fg.grain_lut_8bpc, row);
break;
#endif
#if CONFIG_16BPC
case 10:
case 12:
dav2d_apply_grain_row_16bpc(&c->dsp[off].fg, out, in,
ttd->delayed_fg.scaling_16bpc,
ttd->delayed_fg.grain_lut_16bpc, row);
break;
#endif
default: abort();
}
row = atomic_fetch_add(&ttd->delayed_fg.progress[0], 1);
atomic_fetch_add(&ttd->delayed_fg.progress[1], 1);
}
pthread_mutex_lock(&ttd->lock);
ttd->delayed_fg.exec = 0;
int done = atomic_fetch_add(&ttd->delayed_fg.progress[1], 1) + 1;
progmax = atomic_load(&ttd->delayed_fg.progress[0]);
// signal for completion only once the last runner reaches this
if (done >= progmax) {
ttd->delayed_fg.finished = 1;
pthread_cond_signal(&ttd->delayed_fg.cond);
}
break;
default: abort();
}
}
void *dav2d_worker_task(void *data) {
Dav2dTaskContext *const tc = data;
const Dav2dContext *const c = tc->c;
struct TaskThreadData *const ttd = tc->task_thread.ttd;
dav2d_set_thread_name("dav2d-worker");
pthread_mutex_lock(&ttd->lock);
for (;;) {
if (tc->task_thread.die) break;
if (atomic_load(c->flush)) goto park;
merge_pending(c);
if (ttd->delayed_fg.exec) { // run delayed film grain first
delayed_fg_task(c, ttd);
continue;
}
Dav2dFrameContext *f;
Dav2dTask *t, *prev_t = NULL;
if (c->n_fc > 1) { // run init tasks second
for (unsigned i = 0; i < c->n_fc; i++) {
const unsigned first = atomic_load(&ttd->first);
f = &c->fc[(first + i) % c->n_fc];
if (atomic_load(&f->task_thread.init_done)) continue;
t = f->task_thread.task_head;
if (!t) continue;
if (t->type == DAV2D_TASK_TYPE_INIT) goto found;
if (t->type == DAV2D_TASK_TYPE_INIT_CDF) {
// XXX This can be a simple else, if adding tasks of both
// passes at once (in dav2d_task_create_tile_sbrow).
// Adding the tasks to the pending Q can result in a
// thread merging them before setting init_done.
// We will need to set init_done before adding to the
// pending Q, so maybe return the tasks, set init_done,
// and add to pending Q only then.
int p1;
if (!f->use_pri_sec_cdf) {
p1 = f->in_cdf.progress ?
atomic_load(f->in_cdf.progress) : 1;
} else {
p1 = f->src_cdf[0].progress ?
atomic_load(f->src_cdf[0].progress) : 1;
if (p1 == 1)
p1 = f->src_cdf[1].progress ?
atomic_load(f->src_cdf[1].progress) : 1;
}
if (p1) {
atomic_fetch_or(&f->task_thread.error, p1 == TILE_ERROR);
goto found;
}
}
}
}
while (ttd->cur < c->n_fc) { // run decoding tasks last
const unsigned first = atomic_load(&ttd->first);
f = &c->fc[(first + ttd->cur) % c->n_fc];
merge_pending_frame(f);
prev_t = f->task_thread.task_cur_prev;
t = prev_t ? prev_t->next : f->task_thread.task_head;
while (t) {
if (t->type == DAV2D_TASK_TYPE_INIT_CDF) goto next;
else if (t->type == DAV2D_TASK_TYPE_TILE_ENTROPY ||
t->type == DAV2D_TASK_TYPE_TILE_MV_RESOLUTION ||
t->type == DAV2D_TASK_TYPE_TILE_RECONSTRUCTION)
{
// if not bottom sbrow of tile, this task will be re-added
// after it's finished
const int res = check_tile(t, f);
if (!res)
goto found;
} else if (t->recon_progress) {
const int p = t->type == DAV2D_TASK_TYPE_ENTROPY_PROGRESS;
int error = atomic_load(&f->task_thread.error);
assert(!atomic_load(&f->task_thread.done[p]) || error);
const int tile_row_base = f->frame_hdr->tiling.t.cols *
f->frame_thread.next_tile_row[p];
if (t->type == DAV2D_TASK_TYPE_ENTROPY_PROGRESS) {
atomic_int *const prog = &f->frame_thread.entropy_progress;
const int p1 = atomic_load(prog);
if (p1 < t->by64) goto next;
atomic_fetch_or(&f->task_thread.error, p1 == TILE_ERROR);
}
const int pass = t->type == DAV2D_TASK_TYPE_ENTROPY_PROGRESS ?
0 : f->c->task_thread.n_passes - 1;
for (int tc = 0; tc < f->frame_hdr->tiling.t.cols; tc++) {
Dav2dTileState *const ts = &f->ts[tile_row_base + tc];
const int p2 = atomic_load(&ts->progress[pass]);
if (p2 < t->recon_progress) goto next;
atomic_fetch_or(&f->task_thread.error, p2 == TILE_ERROR);
}
const int sbsz = f->frame_hdr->sbsz;
const int inc = p ? 1 << sbsz : 1;
if (t->type != DAV2D_TASK_TYPE_ENTROPY_PROGRESS && c->n_fc > 1) {
atomic_store(&f->cur.progress[1],
atomic_load(&f->task_thread.error) ?
FRAME_ERROR :
t->by64 + inc < f->sb64h ?
((unsigned) t->recon_progress) << 6 :
UINT_MAX);
}
if (t->by64 + inc < f->sb64h) {
// add sby+1 to list to replace this one
Dav2dTask *next_t = &t[1];
*next_t = *t;
next_t->by64 += inc;
const int ntr = f->frame_thread.next_tile_row[p] + 1;
const int start = f->frame_hdr->tiling.t.row_start_sb[ntr] << sbsz;
if (next_t->by64 == start)
f->frame_thread.next_tile_row[p] = ntr;
next_t->recon_progress = next_t->by64 + inc;
insert_task(f, next_t, 0);
}
goto found;
} else if (t->type == DAV2D_TASK_TYPE_CDEF) {
atomic_uint *prog = f->frame_thread.copy_db_progress;
const int p1 = atomic_load(&prog[(t->by64 - 1) >> 5]);
if (p1 & (1U << ((t->by64 - 1) & 31)))
goto found;
} else {
assert(t->deblock_progress);
const int p1 = atomic_load(&f->frame_thread.deblock_progress);
if (p1 >= t->deblock_progress) {
atomic_fetch_or(&f->task_thread.error, p1 == TILE_ERROR);
goto found;
}
}
next:
prev_t = t;
t = t->next;
f->task_thread.task_cur_prev = prev_t;
}
ttd->cur++;
}
if (reset_task_cur(c, ttd, UINT_MAX)) continue;
if (merge_pending(c)) continue;
park:
tc->task_thread.flushed = 1;
pthread_cond_signal(&tc->task_thread.td.cond);
// we want to be woken up next time progress is signaled
atomic_store(&ttd->cond_signaled, 0);
pthread_cond_wait(&ttd->cond, &ttd->lock);
tc->task_thread.flushed = 0;
reset_task_cur(c, ttd, UINT_MAX);
continue;
found:
// remove t from list
if (prev_t) prev_t->next = t->next;
else f->task_thread.task_head = t->next;
if (!t->next) f->task_thread.task_tail = prev_t;
if (t->type > DAV2D_TASK_TYPE_INIT_CDF && !f->task_thread.task_head)
ttd->cur++;
t->next = NULL;
// we don't need to check cond_signaled here, since we found a task
// after the last signal so we want to re-signal the next waiting thread
// and again won't need to signal after that
atomic_store(&ttd->cond_signaled, 1);
pthread_cond_signal(&ttd->cond);
pthread_mutex_unlock(&ttd->lock);
found_unlocked:;
const int flush = atomic_load(c->flush);
int error = atomic_fetch_or(&f->task_thread.error, flush) | flush;
// run it
tc->f = f;
int by64 = t->by64;
const int sbsz = f->frame_hdr->sbsz, sb64sz = 1 << sbsz;
switch (t->type) {
case DAV2D_TASK_TYPE_INIT: {
assert(c->n_fc > 1);
int res = dav2d_decode_frame_init(f);
int p1;
if (!f->use_pri_sec_cdf) {
p1 = f->in_cdf.progress ? atomic_load(f->in_cdf.progress) : 1;
} else {
p1 = f->src_cdf[0].progress ?
atomic_load(f->src_cdf[0].progress) : 1;
if (p1 == 1)
p1 = f->src_cdf[1].progress ?
atomic_load(f->src_cdf[1].progress) : 1;
}
if (res || p1 == TILE_ERROR) {
pthread_mutex_lock(&ttd->lock);
abort_frame(f, res ? res : DAV2D_ERR(EINVAL));
reset_task_cur(c, ttd, t->frame_idx);
} else {
t->type = DAV2D_TASK_TYPE_INIT_CDF;
if (p1) goto found_unlocked;
add_pending(f, t);
pthread_mutex_lock(&ttd->lock);
}
continue;
}
case DAV2D_TASK_TYPE_INIT_CDF: {
assert(c->n_fc > 1);
int res = DAV2D_ERR(EINVAL);
if (!atomic_load(&f->task_thread.error)) {
if (f->frame_hdr->tip.frame_mode != DAV2D_TIP_FRAME_AS_OUTPUT) {
res = dav2d_decode_frame_init_cdf(f);
} else {
dav2d_decode_tip_frame_init(f);
res = 0;
}
}
if (f->frame_hdr->tip.frame_mode != DAV2D_TIP_FRAME_AS_OUTPUT &&
(f->frame_hdr->disable_cdf_update ||
(!f->task_thread.update_set && !f->seq_hdr->avg_cdf_type)))
{
atomic_store(f->out_cdf.progress, res < 0 ? TILE_ERROR : 1);
}
if (!res) {
assert(c->n_fc > 1);
const int n_passes = c->task_thread.n_passes;
for (int p = 0; p < n_passes; p++) {
const int res = dav2d_task_create_tile_sbrow(f, p, 0);
if (res) {
pthread_mutex_lock(&ttd->lock);
// memory allocation failed
atomic_store(&f->task_thread.done[p + 1 != n_passes], 1);
atomic_store(&f->task_thread.error, -1);
atomic_fetch_sub(&f->task_thread.task_counter,
f->frame_hdr->tiling.t.cols *
f->frame_hdr->tiling.t.rows + f->sbh);
atomic_store(&f->cur.progress[p], FRAME_ERROR);
if (p == 2 && atomic_load(&f->task_thread.done[1])) {
assert(!atomic_load(&f->task_thread.task_counter));
dav2d_decode_frame_exit(f, DAV2D_ERR(ENOMEM));
f->n_tile_data = 0;
pthread_cond_signal(&f->task_thread.cond);
} else {
pthread_mutex_unlock(&ttd->lock);
}
}
}
pthread_mutex_lock(&ttd->lock);
} else {
pthread_mutex_lock(&ttd->lock);
abort_frame(f, res);
reset_task_cur(c, ttd, t->frame_idx);
atomic_store(&f->task_thread.init_done, 1);
}
continue;
}
case DAV2D_TASK_TYPE_TILE_ENTROPY:
case DAV2D_TASK_TYPE_TILE_MV_RESOLUTION:
case DAV2D_TASK_TYPE_TILE_RECONSTRUCTION: {
const int n_passes = c->task_thread.n_passes;
const int pass = t->type == DAV2D_TASK_TYPE_TILE_RECONSTRUCTION ?
n_passes - 1 : t->type != DAV2D_TASK_TYPE_TILE_ENTROPY;
const int tile_idx = (int)(t - f->task_thread.tile_tasks[pass]);
Dav2dTileState *const ts = &f->ts[tile_idx];
tc->ts = ts;
tc->by = by64 << 4;
switch (n_passes) {
case 1: // this should only happen in single-threaded mode, not here
default: abort();
case 2:
assert(t->type != DAV2D_TASK_TYPE_TILE_MV_RESOLUTION);
tc->task_thread.pass = t->type == DAV2D_TASK_TYPE_TILE_ENTROPY ?
PASS_ENTROPY : PASS_MVRES | PASS_RECON;
break;
case 3:
tc->task_thread.pass =
t->type == DAV2D_TASK_TYPE_TILE_ENTROPY ? PASS_ENTROPY :
t->type == DAV2D_TASK_TYPE_TILE_MV_RESOLUTION ? PASS_MVRES : PASS_RECON;
break;
}
if (!error) error = dav2d_decode_tile_sbrow(tc);
const int progress = error ? TILE_ERROR : sb64sz + by64;
// signal progress
atomic_fetch_or(&f->task_thread.error, error);
if (((by64 + sb64sz) << 4) < ts->tiling.row_end) {
t->by64 += sb64sz;
t->deps_skip = 0;
if (!check_tile(t, f)) {
atomic_store(&ts->progress[pass], progress);
reset_task_cur_async(ttd, t->frame_idx, c->n_fc);
if (!atomic_fetch_or(&ttd->cond_signaled, 1))
pthread_cond_signal(&ttd->cond);
goto found_unlocked;
}
atomic_store(&ts->progress[pass], progress);
add_pending(f, t);
pthread_mutex_lock(&ttd->lock);
} else {
pthread_mutex_lock(&ttd->lock);
atomic_store(&ts->progress[pass], progress);
reset_task_cur(c, ttd, t->frame_idx);
error = atomic_load(&f->task_thread.error);
if (!f->frame_hdr->disable_cdf_update &&
tc->task_thread.pass & PASS_ENTROPY &&
((f->task_thread.update_set &&
f->frame_hdr->tiling.update == tile_idx) ||
(f->seq_hdr->avg_cdf_type &&
atomic_fetch_add(&f->task_thread.entropy_task_counter, -1) == 1)))
{
if (!error)
dav2d_decode_frame_update_cdf(f);
if (c->n_fc > 1)
atomic_store(f->out_cdf.progress, error ? TILE_ERROR : 1);
}
if (atomic_fetch_sub(&f->task_thread.task_counter, 1) - 1 == 0 &&
atomic_load(&f->task_thread.done[0]) &&
(n_passes == 1 || atomic_load(&f->task_thread.done[1])))
{
error = atomic_load(&f->task_thread.error);
dav2d_decode_frame_exit(f, error == 1 ? DAV2D_ERR(EINVAL) :
error ? DAV2D_ERR(ENOMEM) : 0);
f->n_tile_data = 0;
pthread_cond_signal(&f->task_thread.cond);
}
assert(atomic_load(&f->task_thread.task_counter) >= 0);
if (!atomic_fetch_or(&ttd->cond_signaled, 1))
pthread_cond_signal(&ttd->cond);
}
continue;
}
case DAV2D_TASK_TYPE_DEBLOCK_COLS:
if (!atomic_load(&f->task_thread.error))
f->bd_fn.filter_slice_deblock_cols(f, by64);
if (ensure_progress(ttd, f, t, DAV2D_TASK_TYPE_DEBLOCK_ROWS,
&f->frame_thread.deblock_progress,
&t->deblock_progress)) continue;
// fall-through
case DAV2D_TASK_TYPE_DEBLOCK_ROWS:
if (!atomic_load(&f->task_thread.error))
f->bd_fn.filter_slice_deblock_rows(f, by64);
// signal deblock progress
if (f->frame_hdr->deblock.level_y[0] ||
f->frame_hdr->deblock.level_y[1])
{
error = atomic_load(&f->task_thread.error);
atomic_store(&f->frame_thread.deblock_progress,
error ? TILE_ERROR : by64 + 1);
reset_task_cur_async(ttd, t->frame_idx, c->n_fc);
if (!atomic_fetch_or(&ttd->cond_signaled, 1))
pthread_cond_signal(&ttd->cond);
} else if (f->lf.cdef_planes || f->lf.restore_planes) {
atomic_fetch_or(&f->frame_thread.copy_db_progress[by64 >> 5],
1U << (by64 & 31));
// CDEF needs the top buffer to be saved by lr_copy_lpf of the
// previous sbrow
if (by64) {
int prog = atomic_load(&f->frame_thread.copy_db_progress[(by64 - 1) >> 5]);
if (~prog & (1U << ((by64 - 1) & 31))) {
t->type = DAV2D_TASK_TYPE_CDEF;
t->recon_progress = t->deblock_progress = 0;
add_pending(f, t);
pthread_mutex_lock(&ttd->lock);
continue;
}
}
}
// fall-through
case DAV2D_TASK_TYPE_CDEF:
if (f->lf.cdef_planes) {
if (!atomic_load(&f->task_thread.error))
f->bd_fn.filter_slice_cdef(tc, by64);
reset_task_cur_async(ttd, t->frame_idx, c->n_fc);
if (!atomic_fetch_or(&ttd->cond_signaled, 1))
pthread_cond_signal(&ttd->cond);
}
// fall-through
case DAV2D_TASK_TYPE_LOOP_RESTORATION:
if (!atomic_load(&f->task_thread.error) && f->lf.restore_planes)
f->bd_fn.filter_slice_lr(f, by64);
// fall-through
case DAV2D_TASK_TYPE_RECONSTRUCTION_PROGRESS:
// dummy to cover for no post-filters
case DAV2D_TASK_TYPE_ENTROPY_PROGRESS:
// dummy to convert tile progress to frame
break;
default: abort();
}
// if task completed [typically LR], signal picture progress as per below
const int n_passes = c->task_thread.n_passes;
const int sb64h = f->sb64h;
if (t->type == DAV2D_TASK_TYPE_ENTROPY_PROGRESS) {
error = atomic_load(&f->task_thread.error);
const unsigned y = by64 + sb64sz >= sb64h ? UINT_MAX :
(unsigned)(by64 + sb64sz) * 64;
assert(c->task_thread.n_passes > 1);
if (c->n_fc > 1 &&
f->cur.p.data[0] /* upon flush, this can be free'ed already */)
{
atomic_store(&f->cur.progress[0], error ? FRAME_ERROR : y);
}
atomic_store(&f->frame_thread.entropy_progress,
error ? TILE_ERROR : by64 + sb64sz);
if (by64 + sb64sz >= sb64h)
atomic_store(&f->task_thread.done[1], 1);
pthread_mutex_lock(&ttd->lock);
const int num_tasks = atomic_fetch_sub(&f->task_thread.task_counter, 1) - 1;
if (by64 + sb64sz < sb64h && num_tasks) {
reset_task_cur(c, ttd, t->frame_idx);
continue;
}
if (!num_tasks && atomic_load(&f->task_thread.done[0]) &&
atomic_load(&f->task_thread.done[1]))
{
error = atomic_load(&f->task_thread.error);
dav2d_decode_frame_exit(f, error == 1 ? DAV2D_ERR(EINVAL) :
error ? DAV2D_ERR(ENOMEM) : 0);
f->n_tile_data = 0;
pthread_cond_signal(&f->task_thread.cond);
}
reset_task_cur(c, ttd, t->frame_idx);
continue;
}
// t->type != DAV2D_TASK_TYPE_ENTROPY_PROGRESS
atomic_fetch_or(&f->frame_thread.frame_progress[by64 >> 5],
1U << (by64 & 31));
pthread_mutex_lock(&f->task_thread.lock);
by64 = get_frame_progress(c, f);
error = atomic_load(&f->task_thread.error);
const unsigned y = by64 + 1 == sb64h ? UINT_MAX : (unsigned)(by64 + 1) * 64;
if (c->n_fc > 1 && f->cur.p.data[0] /* upon flush, this can be free'ed already */)
atomic_store(&f->cur.progress[2], error ? FRAME_ERROR : y);
pthread_mutex_unlock(&f->task_thread.lock);
if (by64 + 1 == sb64h)
atomic_store(&f->task_thread.done[0], 1);
pthread_mutex_lock(&ttd->lock);
const int num_tasks = atomic_fetch_sub(&f->task_thread.task_counter, 1) - 1;
if (by64 + 1 < sb64h && num_tasks) {
reset_task_cur(c, ttd, t->frame_idx);
continue;
}
if (!num_tasks && atomic_load(&f->task_thread.done[0]) &&
(n_passes == 1 || atomic_load(&f->task_thread.done[1])))
{
error = atomic_load(&f->task_thread.error);
dav2d_decode_frame_exit(f, error == 1 ? DAV2D_ERR(EINVAL) :
error ? DAV2D_ERR(ENOMEM) : 0);
f->n_tile_data = 0;
pthread_cond_signal(&f->task_thread.cond);
}
reset_task_cur(c, ttd, t->frame_idx);
}
pthread_mutex_unlock(&ttd->lock);
return NULL;
}