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
|
#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() {}
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();
const uv_loop_t* event_loop() const { return loop_; }
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;
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 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);
virtual ~NodePlatform() {}
void DrainTasks(v8::Isolate* isolate) override;
void CancelPendingDelayedTasks(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;
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;
node::tracing::TracingController* GetTracingController() override;
bool FlushForegroundTasks(v8::Isolate* isolate) override;
void RegisterIsolate(v8::Isolate* isolate, uv_loop_t* loop) override;
void UnregisterIsolate(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_;
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_
|
