Coverage Report

Created: 2024-09-08 07:14

/src/libjxl/lib/threads/thread_parallel_runner_internal.cc
Line
Count
Source (jump to first uncovered line)
1
// Copyright (c) the JPEG XL Project Authors. All rights reserved.
2
//
3
// Use of this source code is governed by a BSD-style
4
// license that can be found in the LICENSE file.
5
6
#include "lib/threads/thread_parallel_runner_internal.h"
7
8
#include <jxl/parallel_runner.h>
9
#include <jxl/types.h>
10
11
#include <algorithm>
12
#include <atomic>
13
#include <cstddef>
14
#include <cstdint>
15
#include <mutex>
16
#include <thread>
17
18
#include "lib/jxl/base/compiler_specific.h"
19
20
namespace jpegxl {
21
22
// static
23
JxlParallelRetCode ThreadParallelRunner::Runner(
24
    void* runner_opaque, void* jpegxl_opaque, JxlParallelRunInit init,
25
100k
    JxlParallelRunFunction func, uint32_t start_range, uint32_t end_range) {
26
100k
  ThreadParallelRunner* self =
27
100k
      static_cast<ThreadParallelRunner*>(runner_opaque);
28
100k
  if (start_range > end_range) return JXL_PARALLEL_RET_RUNNER_ERROR;
29
100k
  if (start_range == end_range) return JXL_PARALLEL_RET_SUCCESS;
30
31
100k
  int ret = init(jpegxl_opaque, std::max<size_t>(self->num_worker_threads_, 1));
32
100k
  if (ret != JXL_PARALLEL_RET_SUCCESS) return ret;
33
34
  // Use a sequential run when num_worker_threads_ is zero since we have no
35
  // worker threads.
36
100k
  if (self->num_worker_threads_ == 0) {
37
3.94k
    const size_t thread = 0;
38
207k
    for (uint32_t task = start_range; task < end_range; ++task) {
39
203k
      func(jpegxl_opaque, task, thread);
40
203k
    }
41
3.94k
    return JXL_PARALLEL_RET_SUCCESS;
42
3.94k
  }
43
44
96.9k
  if (self->depth_.fetch_add(1, std::memory_order_acq_rel) != 0) {
45
0
    return JXL_PARALLEL_RET_RUNNER_ERROR;  // Must not re-enter.
46
0
  }
47
48
96.9k
  const WorkerCommand worker_command =
49
96.9k
      (static_cast<WorkerCommand>(start_range) << 32) + end_range;
50
  // Ensure the inputs do not result in a reserved command.
51
96.9k
  if ((worker_command == kWorkerWait) || (worker_command == kWorkerOnce) ||
52
96.9k
      (worker_command == kWorkerExit)) {
53
0
    return JXL_PARALLEL_RET_RUNNER_ERROR;
54
0
  }
55
56
96.9k
  self->data_func_ = func;
57
96.9k
  self->jpegxl_opaque_ = jpegxl_opaque;
58
96.9k
  self->num_reserved_.store(0, std::memory_order_relaxed);
59
60
96.9k
  self->StartWorkers(worker_command);
61
96.9k
  self->WorkersReadyBarrier();
62
63
96.9k
  if (self->depth_.fetch_add(-1, std::memory_order_acq_rel) != 1) {
64
0
    return JXL_PARALLEL_RET_RUNNER_ERROR;
65
0
  }
66
96.9k
  return JXL_PARALLEL_RET_SUCCESS;
67
96.9k
}
68
69
// static
70
void ThreadParallelRunner::RunRange(ThreadParallelRunner* self,
71
                                    const WorkerCommand command,
72
193k
                                    const int thread) {
73
193k
  const uint32_t begin = command >> 32;
74
193k
  const uint32_t end = command & 0xFFFFFFFF;
75
193k
  const uint32_t num_tasks = end - begin;
76
193k
  const uint32_t num_worker_threads = self->num_worker_threads_;
77
78
  // OpenMP introduced several "schedule" strategies:
79
  // "single" (static assignment of exactly one chunk per thread): slower.
80
  // "dynamic" (allocates k tasks at a time): competitive for well-chosen k.
81
  // "guided" (allocates k tasks, decreases k): computing k = remaining/n
82
  //   is faster than halving k each iteration. We prefer this strategy
83
  //   because it avoids user-specified parameters.
84
85
489k
  for (;;) {
86
#if JXL_FALSE
87
    // dynamic
88
    const uint32_t my_size = std::max(num_tasks / (num_worker_threads * 4), 1);
89
#else
90
    // guided
91
489k
    const uint32_t num_reserved =
92
489k
        self->num_reserved_.load(std::memory_order_relaxed);
93
    // It is possible that more tasks are reserved than ready to run.
94
489k
    const uint32_t num_remaining =
95
489k
        num_tasks - std::min(num_reserved, num_tasks);
96
489k
    const uint32_t my_size =
97
489k
        std::max(num_remaining / (num_worker_threads * 4), 1u);
98
489k
#endif
99
489k
    const uint32_t my_begin = begin + self->num_reserved_.fetch_add(
100
489k
                                          my_size, std::memory_order_relaxed);
101
489k
    const uint32_t my_end = std::min(my_begin + my_size, begin + num_tasks);
102
    // Another thread already reserved the last task.
103
489k
    if (my_begin >= my_end) {
104
193k
      break;
105
193k
    }
106
759k
    for (uint32_t task = my_begin; task < my_end; ++task) {
107
463k
      self->data_func_(self->jpegxl_opaque_, task, thread);
108
463k
    }
109
296k
  }
110
193k
}
111
112
// static
113
void ThreadParallelRunner::ThreadFunc(ThreadParallelRunner* self,
114
26.0k
                                      const int thread) {
115
  // Until kWorkerExit command received:
116
219k
  for (;;) {
117
219k
    std::unique_lock<std::mutex> lock(self->mutex_);
118
    // Notify main thread that this thread is ready.
119
219k
    if (++self->workers_ready_ == self->num_threads_) {
120
109k
      self->workers_ready_cv_.notify_one();
121
109k
    }
122
219k
  RESUME_WAIT:
123
    // Wait for a command.
124
219k
    self->worker_start_cv_.wait(lock);
125
219k
    const WorkerCommand command = self->worker_start_command_;
126
219k
    switch (command) {
127
0
      case kWorkerWait:    // spurious wakeup:
128
0
        goto RESUME_WAIT;  // lock still held, avoid incrementing ready.
129
0
      case kWorkerOnce:
130
0
        lock.unlock();
131
0
        self->data_func_(self->jpegxl_opaque_, thread, thread);
132
0
        break;
133
26.1k
      case kWorkerExit:
134
26.1k
        return;  // exits thread
135
193k
      default:
136
193k
        lock.unlock();
137
193k
        RunRange(self, command, thread);
138
193k
        break;
139
219k
    }
140
219k
  }
141
26.0k
}
142
143
ThreadParallelRunner::ThreadParallelRunner(const int num_worker_threads)
144
    : num_worker_threads_(num_worker_threads),
145
23.5k
      num_threads_(std::max(num_worker_threads, 1)) {
146
23.5k
  threads_.reserve(num_worker_threads_);
147
148
  // Suppress "unused-private-field" warning.
149
23.5k
  (void)padding1;
150
23.5k
  (void)padding2;
151
152
  // Safely handle spurious worker wakeups.
153
23.5k
  worker_start_command_ = kWorkerWait;
154
155
49.6k
  for (uint32_t i = 0; i < num_worker_threads_; ++i) {
156
26.1k
    threads_.emplace_back(ThreadFunc, this, i);
157
26.1k
  }
158
159
23.5k
  if (num_worker_threads_ != 0) {
160
13.0k
    WorkersReadyBarrier();
161
13.0k
  }
162
23.5k
}
163
164
23.5k
ThreadParallelRunner::~ThreadParallelRunner() {
165
23.5k
  if (num_worker_threads_ != 0) {
166
13.0k
    StartWorkers(kWorkerExit);
167
13.0k
  }
168
169
26.1k
  for (std::thread& thread : threads_) {
170
26.1k
    if (thread.joinable()) {
171
26.1k
      thread.join();
172
26.1k
    } else {
173
0
#if JXL_IS_DEBUG_BUILD
174
0
      JXL_PRINT_STACK_TRACE();
175
0
      JXL_CRASH();
176
0
#endif
177
0
    }
178
26.1k
  }
179
23.5k
}
180
}  // namespace jpegxl