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
|
#ifndef SRC_NODE_PLATFORM_H_
#define SRC_NODE_PLATFORM_H_
#include <queue>
#include <unordered_map>
#include <vector>
#include <functional>
#include "libplatform/libplatform.h"
#include "node.h"
#include "node_mutex.h"
#include "uv.h"
namespace node {
class NodePlatform;
class IsolateData;
class PerIsolatePlatformData;
template <class T>
class TaskQueue {
public:
TaskQueue();
~TaskQueue() {}
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 v8::TaskRunner,
public std::enable_shared_from_this<PerIsolatePlatformData> {
public:
PerIsolatePlatformData(v8::Isolate* isolate, uv_loop_t* loop);
~PerIsolatePlatformData();
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; };
void Shutdown();
void ref();
int unref();
// 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();
void CancelPendingDelayedTasks();
private:
void DeleteFromScheduledTasks(DelayedTask* task);
static void FlushTasks(uv_async_t* handle);
static void RunForegroundTask(std::unique_ptr<v8::Task> task);
static void RunForegroundTask(uv_timer_t* timer);
int ref_count_ = 1;
v8::Isolate* isolate_;
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, std::function<void(DelayedTask*)>>
DelayedTaskPointer;
std::vector<DelayedTaskPointer> scheduled_delayed_tasks_;
};
// This acts as the single background task runner for all Isolates.
class BackgroundTaskRunner : public v8::TaskRunner {
public:
explicit BackgroundTaskRunner(int thread_pool_size);
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; };
void BlockingDrain();
void Shutdown();
size_t NumberOfAvailableBackgroundThreads() const;
private:
TaskQueue<v8::Task> background_tasks_;
std::vector<std::unique_ptr<uv_thread_t>> threads_;
};
class NodePlatform : public MultiIsolatePlatform {
public:
NodePlatform(int thread_pool_size, v8::TracingController* tracing_controller);
virtual ~NodePlatform() {}
void DrainBackgroundTasks(v8::Isolate* isolate) override;
void CancelPendingDelayedTasks(v8::Isolate* isolate) override;
void Shutdown();
// v8::Platform implementation.
size_t NumberOfAvailableBackgroundThreads() override;
void CallOnBackgroundThread(v8::Task* task,
ExpectedRuntime expected_runtime) override;
void CallOnForegroundThread(v8::Isolate* isolate, v8::Task* task) override;
void CallDelayedOnForegroundThread(v8::Isolate* isolate, v8::Task* task,
double delay_in_seconds) override;
bool IdleTasksEnabled(v8::Isolate* isolate) override;
double MonotonicallyIncreasingTime() override;
double CurrentClockTimeMillis() override;
v8::TracingController* GetTracingController() override;
// 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 FlushForegroundTasks(v8::Isolate* isolate);
void RegisterIsolate(IsolateData* isolate_data, uv_loop_t* loop) override;
void UnregisterIsolate(IsolateData* isolate_data) override;
std::shared_ptr<v8::TaskRunner> GetBackgroundTaskRunner(
v8::Isolate* isolate) override;
std::shared_ptr<v8::TaskRunner> GetForegroundTaskRunner(
v8::Isolate* isolate) override;
private:
std::shared_ptr<PerIsolatePlatformData> ForIsolate(v8::Isolate* isolate);
Mutex per_isolate_mutex_;
std::unordered_map<v8::Isolate*,
std::shared_ptr<PerIsolatePlatformData>> per_isolate_;
std::unique_ptr<v8::TracingController> tracing_controller_;
std::shared_ptr<BackgroundTaskRunner> background_task_runner_;
};
} // namespace node
#endif // SRC_NODE_PLATFORM_H_
|
