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
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
|
#ifndef SRC_NODE_PLATFORM_H_
#define SRC_NODE_PLATFORM_H_
#if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
#include <queue>
#include <unordered_map>
#include <vector>
#include <functional>
#include "libplatform/libplatform.h"
#include "node.h"
#include "node_mutex.h"
#include "tracing/agent.h"
#include "uv.h"
namespace node {
class NodePlatform;
class IsolateData;
class PerIsolatePlatformData;
template <class T>
class TaskQueue {
public:
TaskQueue();
~TaskQueue() = default;
void Push(std::unique_ptr<T> task);
std::unique_ptr<T> Pop();
std::unique_ptr<T> BlockingPop();
std::queue<std::unique_ptr<T>> PopAll();
void NotifyOfCompletion();
void BlockingDrain();
void Stop();
private:
Mutex lock_;
ConditionVariable tasks_available_;
ConditionVariable tasks_drained_;
int outstanding_tasks_;
bool stopped_;
std::queue<std::unique_ptr<T>> task_queue_;
};
struct DelayedTask {
std::unique_ptr<v8::Task> task;
uv_timer_t timer;
double timeout;
std::shared_ptr<PerIsolatePlatformData> platform_data;
};
// This acts as the foreground task runner for a given Isolate.
class PerIsolatePlatformData :
public IsolatePlatformDelegate,
public v8::TaskRunner,
public std::enable_shared_from_this<PerIsolatePlatformData> {
public:
PerIsolatePlatformData(v8::Isolate* isolate, uv_loop_t* loop);
~PerIsolatePlatformData() override;
std::shared_ptr<v8::TaskRunner> GetForegroundTaskRunner() override;
void PostTask(std::unique_ptr<v8::Task> task) override;
void PostIdleTask(std::unique_ptr<v8::IdleTask> task) override;
void PostDelayedTask(std::unique_ptr<v8::Task> task,
double delay_in_seconds) override;
bool IdleTasksEnabled() override { return false; }
// Non-nestable tasks are treated like regular tasks.
bool NonNestableTasksEnabled() const override { return true; }
bool NonNestableDelayedTasksEnabled() const override { return true; }
void PostNonNestableTask(std::unique_ptr<v8::Task> task) override;
void PostNonNestableDelayedTask(std::unique_ptr<v8::Task> task,
double delay_in_seconds) override;
void AddShutdownCallback(void (*callback)(void*), void* data);
void Shutdown();
// Returns true if work was dispatched or executed. New tasks that are
// posted during flushing of the queue are postponed until the next
// flushing.
bool FlushForegroundTasksInternal();
const uv_loop_t* event_loop() const { return loop_; }
private:
void DeleteFromScheduledTasks(DelayedTask* task);
void DecreaseHandleCount();
static void FlushTasks(uv_async_t* handle);
static void RunForegroundTask(std::unique_ptr<v8::Task> task);
static void RunForegroundTask(uv_timer_t* timer);
struct ShutdownCallback {
void (*cb)(void*);
void* data;
};
typedef std::vector<ShutdownCallback> ShutdownCbList;
ShutdownCbList shutdown_callbacks_;
// shared_ptr to self to keep this object alive during shutdown.
std::shared_ptr<PerIsolatePlatformData> self_reference_;
uint32_t uv_handle_count_ = 1; // 1 = flush_tasks_
uv_loop_t* const loop_;
uv_async_t* flush_tasks_ = nullptr;
TaskQueue<v8::Task> foreground_tasks_;
TaskQueue<DelayedTask> foreground_delayed_tasks_;
// Use a custom deleter because libuv needs to close the handle first.
typedef std::unique_ptr<DelayedTask, void(*)(DelayedTask*)>
DelayedTaskPointer;
std::vector<DelayedTaskPointer> scheduled_delayed_tasks_;
};
// This acts as the single worker thread task runner for all Isolates.
class WorkerThreadsTaskRunner {
public:
explicit WorkerThreadsTaskRunner(int thread_pool_size);
void PostTask(std::unique_ptr<v8::Task> task);
void PostDelayedTask(std::unique_ptr<v8::Task> task,
double delay_in_seconds);
void BlockingDrain();
void Shutdown();
int NumberOfWorkerThreads() const;
private:
TaskQueue<v8::Task> pending_worker_tasks_;
class DelayedTaskScheduler;
std::unique_ptr<DelayedTaskScheduler> delayed_task_scheduler_;
std::vector<std::unique_ptr<uv_thread_t>> threads_;
};
class NodePlatform : public MultiIsolatePlatform {
public:
NodePlatform(int thread_pool_size,
node::tracing::TracingController* tracing_controller);
~NodePlatform() override = default;
void DrainTasks(v8::Isolate* isolate) override;
void Shutdown();
// v8::Platform implementation.
int NumberOfWorkerThreads() override;
void CallOnWorkerThread(std::unique_ptr<v8::Task> task) override;
void CallDelayedOnWorkerThread(std::unique_ptr<v8::Task> task,
double delay_in_seconds) override;
void CallOnForegroundThread(v8::Isolate* isolate, v8::Task* task) override {
UNREACHABLE();
}
void CallDelayedOnForegroundThread(v8::Isolate* isolate,
v8::Task* task,
double delay_in_seconds) override {
UNREACHABLE();
}
bool IdleTasksEnabled(v8::Isolate* isolate) override;
double MonotonicallyIncreasingTime() override;
double CurrentClockTimeMillis() override;
node::tracing::TracingController* GetTracingController() override;
bool FlushForegroundTasks(v8::Isolate* isolate) override;
void RegisterIsolate(v8::Isolate* isolate, uv_loop_t* loop) override;
void RegisterIsolate(v8::Isolate* isolate,
IsolatePlatformDelegate* delegate) override;
void UnregisterIsolate(v8::Isolate* isolate) override;
void AddIsolateFinishedCallback(v8::Isolate* isolate,
void (*callback)(void*), void* data) override;
std::shared_ptr<v8::TaskRunner> GetForegroundTaskRunner(
v8::Isolate* isolate) override;
private:
IsolatePlatformDelegate* ForIsolate(v8::Isolate* isolate);
std::shared_ptr<PerIsolatePlatformData> ForNodeIsolate(v8::Isolate* isolate);
Mutex per_isolate_mutex_;
using DelegatePair = std::pair<
IsolatePlatformDelegate*, std::shared_ptr<PerIsolatePlatformData>>;
std::unordered_map<v8::Isolate*, DelegatePair> per_isolate_;
node::tracing::TracingController* tracing_controller_;
std::shared_ptr<WorkerThreadsTaskRunner> worker_thread_task_runner_;
};
} // namespace node
#endif // defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS
#endif // SRC_NODE_PLATFORM_H_
|
