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
|
// 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
: NON_EXPORTED_BASE(public JobDelegate),
public std::enable_shared_from_this<DefaultJobState> {
public:
DefaultJobState(Platform* platform, std::unique_ptr<JobTask> job_task,
TaskPriority priority, size_t num_worker_threads)
: platform_(platform),
job_task_(std::move(job_task)),
priority_(priority),
num_worker_threads_(num_worker_threads) {}
virtual ~DefaultJobState();
void NotifyConcurrencyIncrease() override;
bool ShouldYield() override {
// Thread-safe but may return an outdated result.
return is_canceled_.load(std::memory_order_relaxed);
}
void Join();
void CancelAndWait();
// 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();
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() const;
void CallOnWorkerThread(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_;
};
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;
bool IsRunning() override { return state_ != nullptr; }
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 {
job_task_->Run(shared_state.get());
} 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_
|
