1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
|
// Copyright 2020 the V8 project authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#ifndef V8_LIBPLATFORM_DEFAULT_JOB_H_
#define V8_LIBPLATFORM_DEFAULT_JOB_H_
#include <atomic>
#include <memory>
#include "include/libplatform/libplatform-export.h"
#include "include/v8-platform.h"
#include "src/base/platform/condition-variable.h"
#include "src/base/platform/mutex.h"
namespace v8 {
namespace platform {
class V8_PLATFORM_EXPORT DefaultJobState
: public std::enable_shared_from_this<DefaultJobState> {
public:
class JobDelegate : public v8::JobDelegate {
public:
explicit JobDelegate(DefaultJobState* outer, bool is_joining_thread = false)
: outer_(outer), is_joining_thread_(is_joining_thread) {}
~JobDelegate();
void NotifyConcurrencyIncrease() override {
outer_->NotifyConcurrencyIncrease();
}
bool ShouldYield() override {
// Thread-safe but may return an outdated result.
return outer_->is_canceled_.load(std::memory_order_relaxed);
}
uint8_t GetTaskId() override;
bool IsJoiningThread() const override { return is_joining_thread_; }
private:
static constexpr uint8_t kInvalidTaskId =
std::numeric_limits<uint8_t>::max();
DefaultJobState* outer_;
uint8_t task_id_ = kInvalidTaskId;
bool is_joining_thread_;
};
DefaultJobState(Platform* platform, std::unique_ptr<JobTask> job_task,
TaskPriority priority, size_t num_worker_threads);
virtual ~DefaultJobState();
void NotifyConcurrencyIncrease();
uint8_t AcquireTaskId();
void ReleaseTaskId(uint8_t task_id);
void Join();
void CancelAndWait();
void CancelAndDetach();
bool IsActive();
// Must be called before running |job_task_| for the first time. If it returns
// true, then the worker thread must contribute and must call DidRunTask(), or
// false if it should return.
bool CanRunFirstTask();
// Must be called after running |job_task_|. Returns true if the worker thread
// must contribute again, or false if it should return.
bool DidRunTask();
void UpdatePriority(TaskPriority);
private:
// Called from the joining thread. Waits for the worker count to be below or
// equal to max concurrency (will happen when a worker calls
// DidRunTask()). Returns true if the joining thread should run a task, or
// false if joining was completed and all other workers returned because
// there's no work remaining.
bool WaitForParticipationOpportunityLockRequired();
// Returns GetMaxConcurrency() capped by the number of threads used by this
// job.
size_t CappedMaxConcurrency(size_t worker_count) const;
void CallOnWorkerThread(TaskPriority priority, std::unique_ptr<Task> task);
Platform* const platform_;
std::unique_ptr<JobTask> job_task_;
// All members below are protected by |mutex_|.
base::Mutex mutex_;
TaskPriority priority_;
// Number of workers running this job.
size_t active_workers_ = 0;
// Number of posted tasks that aren't running this job yet.
size_t pending_tasks_ = 0;
// Indicates if the job is canceled.
std::atomic_bool is_canceled_{false};
// Number of worker threads available to schedule the worker task.
size_t num_worker_threads_;
// Signaled when a worker returns.
base::ConditionVariable worker_released_condition_;
std::atomic<uint32_t> assigned_task_ids_{0};
};
class V8_PLATFORM_EXPORT DefaultJobHandle : public JobHandle {
public:
explicit DefaultJobHandle(std::shared_ptr<DefaultJobState> state);
~DefaultJobHandle() override;
void NotifyConcurrencyIncrease() override {
state_->NotifyConcurrencyIncrease();
}
void Join() override;
void Cancel() override;
void CancelAndDetach() override;
bool IsCompleted() override { return !IsActive(); }
bool IsActive() override;
bool IsRunning() override { return IsValid(); }
bool IsValid() override { return state_ != nullptr; }
bool UpdatePriorityEnabled() const override { return true; }
void UpdatePriority(TaskPriority) override;
private:
std::shared_ptr<DefaultJobState> state_;
DISALLOW_COPY_AND_ASSIGN(DefaultJobHandle);
};
class DefaultJobWorker : public Task {
public:
DefaultJobWorker(std::weak_ptr<DefaultJobState> state, JobTask* job_task)
: state_(std::move(state)), job_task_(job_task) {}
~DefaultJobWorker() override = default;
void Run() override {
auto shared_state = state_.lock();
if (!shared_state) return;
if (!shared_state->CanRunFirstTask()) return;
do {
// Scope of |delegate| must not outlive DidRunTask() so that associated
// state is freed before the worker becomes inactive.
DefaultJobState::JobDelegate delegate(shared_state.get());
job_task_->Run(&delegate);
} while (shared_state->DidRunTask());
}
private:
friend class DefaultJob;
std::weak_ptr<DefaultJobState> state_;
JobTask* job_task_;
DISALLOW_COPY_AND_ASSIGN(DefaultJobWorker);
};
} // namespace platform
} // namespace v8
#endif // V8_LIBPLATFORM_DEFAULT_JOB_H_
|
