1 |
|
|
#ifndef SRC_NODE_MESSAGING_H_ |
2 |
|
|
#define SRC_NODE_MESSAGING_H_ |
3 |
|
|
|
4 |
|
|
#if defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS |
5 |
|
|
|
6 |
|
|
#include "env.h" |
7 |
|
|
#include "node_mutex.h" |
8 |
|
|
#include "sharedarraybuffer_metadata.h" |
9 |
|
|
#include <list> |
10 |
|
|
|
11 |
|
|
namespace node { |
12 |
|
|
namespace worker { |
13 |
|
|
|
14 |
|
|
class MessagePortData; |
15 |
|
|
class MessagePort; |
16 |
|
|
|
17 |
|
|
typedef MaybeStackBuffer<v8::Local<v8::Value>, 8> TransferList; |
18 |
|
|
|
19 |
|
|
// Represents a single communication message. |
20 |
✗✓ |
329486 |
class Message : public MemoryRetainer { |
21 |
|
|
public: |
22 |
|
|
// Create a Message with a specific underlying payload, in the format of the |
23 |
|
|
// V8 ValueSerializer API. If `payload` is empty, this message indicates |
24 |
|
|
// that the receiving message port should close itself. |
25 |
|
|
explicit Message(MallocedBuffer<char>&& payload = MallocedBuffer<char>()); |
26 |
|
|
|
27 |
|
103835 |
Message(Message&& other) = default; |
28 |
|
|
Message& operator=(Message&& other) = default; |
29 |
|
|
Message& operator=(const Message&) = delete; |
30 |
|
|
Message(const Message&) = delete; |
31 |
|
|
|
32 |
|
|
// Whether this is a message indicating that the port is to be closed. |
33 |
|
|
// This is the last message to be received by a MessagePort. |
34 |
|
|
bool IsCloseMessage() const; |
35 |
|
|
|
36 |
|
|
// Deserialize the contained JS value. May only be called once, and only |
37 |
|
|
// after Serialize() has been called (e.g. by another thread). |
38 |
|
|
v8::MaybeLocal<v8::Value> Deserialize(Environment* env, |
39 |
|
|
v8::Local<v8::Context> context); |
40 |
|
|
|
41 |
|
|
// Serialize a JS value, and optionally transfer objects, into this message. |
42 |
|
|
// The Message object retains ownership of all transferred objects until |
43 |
|
|
// deserialization. |
44 |
|
|
// The source_port parameter, if provided, will make Serialize() throw a |
45 |
|
|
// "DataCloneError" DOMException if source_port is found in transfer_list. |
46 |
|
|
v8::Maybe<bool> Serialize(Environment* env, |
47 |
|
|
v8::Local<v8::Context> context, |
48 |
|
|
v8::Local<v8::Value> input, |
49 |
|
|
const TransferList& transfer_list, |
50 |
|
|
v8::Local<v8::Object> source_port = |
51 |
|
|
v8::Local<v8::Object>()); |
52 |
|
|
|
53 |
|
|
// Internal method of Message that is called when a new SharedArrayBuffer |
54 |
|
|
// object is encountered in the incoming value's structure. |
55 |
|
|
void AddSharedArrayBuffer(const SharedArrayBufferMetadataReference& ref); |
56 |
|
|
// Internal method of Message that is called once serialization finishes |
57 |
|
|
// and that transfers ownership of `data` to this message. |
58 |
|
|
void AddMessagePort(std::unique_ptr<MessagePortData>&& data); |
59 |
|
|
// Internal method of Message that is called when a new WebAssembly.Module |
60 |
|
|
// object is encountered in the incoming value's structure. |
61 |
|
|
uint32_t AddWASMModule(v8::WasmModuleObject::TransferrableModule&& mod); |
62 |
|
|
|
63 |
|
|
// The MessagePorts that will be transferred, as recorded by Serialize(). |
64 |
|
|
// Used for warning user about posting the target MessagePort to itself, |
65 |
|
|
// which will as a side effect destroy the communication channel. |
66 |
|
41139 |
const std::vector<std::unique_ptr<MessagePortData>>& message_ports() const { |
67 |
|
41139 |
return message_ports_; |
68 |
|
|
} |
69 |
|
|
|
70 |
|
|
void MemoryInfo(MemoryTracker* tracker) const override; |
71 |
|
|
|
72 |
|
4 |
SET_MEMORY_INFO_NAME(Message) |
73 |
|
4 |
SET_SELF_SIZE(Message) |
74 |
|
|
|
75 |
|
|
private: |
76 |
|
|
MallocedBuffer<char> main_message_buf_; |
77 |
|
|
std::vector<MallocedBuffer<char>> array_buffer_contents_; |
78 |
|
|
std::vector<SharedArrayBufferMetadataReference> shared_array_buffers_; |
79 |
|
|
std::vector<std::unique_ptr<MessagePortData>> message_ports_; |
80 |
|
|
std::vector<v8::WasmModuleObject::TransferrableModule> wasm_modules_; |
81 |
|
|
|
82 |
|
|
friend class MessagePort; |
83 |
|
|
}; |
84 |
|
|
|
85 |
|
|
// This contains all data for a `MessagePort` instance that is not tied to |
86 |
|
|
// a specific Environment/Isolate/event loop, for easier transfer between those. |
87 |
|
|
class MessagePortData : public MemoryRetainer { |
88 |
|
|
public: |
89 |
|
|
explicit MessagePortData(MessagePort* owner); |
90 |
|
|
~MessagePortData() override; |
91 |
|
|
|
92 |
|
|
MessagePortData(MessagePortData&& other) = delete; |
93 |
|
|
MessagePortData& operator=(MessagePortData&& other) = delete; |
94 |
|
|
MessagePortData(const MessagePortData& other) = delete; |
95 |
|
|
MessagePortData& operator=(const MessagePortData& other) = delete; |
96 |
|
|
|
97 |
|
|
// Add a message to the incoming queue and notify the receiver. |
98 |
|
|
// This may be called from any thread. |
99 |
|
|
void AddToIncomingQueue(Message&& message); |
100 |
|
|
|
101 |
|
|
// Turns `a` and `b` into siblings, i.e. connects the sending side of one |
102 |
|
|
// to the receiving side of the other. This is not thread-safe. |
103 |
|
|
static void Entangle(MessagePortData* a, MessagePortData* b); |
104 |
|
|
|
105 |
|
|
// Removes any possible sibling. This is thread-safe (it acquires both |
106 |
|
|
// `sibling_mutex_` and `mutex_`), and has to be because it is called once |
107 |
|
|
// the corresponding JS handle handle wants to close |
108 |
|
|
// which can happen on either side of a worker. |
109 |
|
|
void Disentangle(); |
110 |
|
|
|
111 |
|
|
void MemoryInfo(MemoryTracker* tracker) const override; |
112 |
|
|
|
113 |
|
8 |
SET_MEMORY_INFO_NAME(MessagePortData) |
114 |
|
8 |
SET_SELF_SIZE(MessagePortData) |
115 |
|
|
|
116 |
|
|
private: |
117 |
|
|
// This mutex protects all fields below it, with the exception of |
118 |
|
|
// sibling_. |
119 |
|
|
mutable Mutex mutex_; |
120 |
|
|
std::list<Message> incoming_messages_; |
121 |
|
|
MessagePort* owner_ = nullptr; |
122 |
|
|
// This mutex protects the sibling_ field and is shared between two entangled |
123 |
|
|
// MessagePorts. If both mutexes are acquired, this one needs to be |
124 |
|
|
// acquired first. |
125 |
|
|
std::shared_ptr<Mutex> sibling_mutex_ = std::make_shared<Mutex>(); |
126 |
|
|
MessagePortData* sibling_ = nullptr; |
127 |
|
|
|
128 |
|
|
friend class MessagePort; |
129 |
|
|
}; |
130 |
|
|
|
131 |
|
|
// A message port that receives messages from other threads, including |
132 |
|
|
// the uv_async_t handle that is used to notify the current event loop of |
133 |
|
|
// new incoming messages. |
134 |
|
|
class MessagePort : public HandleWrap { |
135 |
|
|
public: |
136 |
|
|
// Create a new MessagePort. The `context` argument specifies the Context |
137 |
|
|
// instance that is used for creating the values emitted from this port. |
138 |
|
|
MessagePort(Environment* env, |
139 |
|
|
v8::Local<v8::Context> context, |
140 |
|
|
v8::Local<v8::Object> wrap); |
141 |
|
|
~MessagePort() override; |
142 |
|
|
|
143 |
|
|
// Create a new message port instance, optionally over an existing |
144 |
|
|
// `MessagePortData` object. |
145 |
|
|
static MessagePort* New(Environment* env, |
146 |
|
|
v8::Local<v8::Context> context, |
147 |
|
|
std::unique_ptr<MessagePortData> data = nullptr); |
148 |
|
|
|
149 |
|
|
// Send a message, i.e. deliver it into the sibling's incoming queue. |
150 |
|
|
// If this port is closed, or if there is no sibling, this message is |
151 |
|
|
// serialized with transfers, then silently discarded. |
152 |
|
|
v8::Maybe<bool> PostMessage(Environment* env, |
153 |
|
|
v8::Local<v8::Value> message, |
154 |
|
|
const TransferList& transfer); |
155 |
|
|
|
156 |
|
|
// Start processing messages on this port as a receiving end. |
157 |
|
|
void Start(); |
158 |
|
|
// Stop processing messages on this port as a receiving end. |
159 |
|
|
void Stop(); |
160 |
|
|
|
161 |
|
|
/* constructor */ |
162 |
|
|
static void New(const v8::FunctionCallbackInfo<v8::Value>& args); |
163 |
|
|
/* prototype methods */ |
164 |
|
|
static void PostMessage(const v8::FunctionCallbackInfo<v8::Value>& args); |
165 |
|
|
static void Start(const v8::FunctionCallbackInfo<v8::Value>& args); |
166 |
|
|
static void Stop(const v8::FunctionCallbackInfo<v8::Value>& args); |
167 |
|
|
static void Drain(const v8::FunctionCallbackInfo<v8::Value>& args); |
168 |
|
|
static void ReceiveMessage(const v8::FunctionCallbackInfo<v8::Value>& args); |
169 |
|
|
|
170 |
|
|
/* static */ |
171 |
|
|
static void MoveToContext(const v8::FunctionCallbackInfo<v8::Value>& args); |
172 |
|
|
|
173 |
|
|
// Turns `a` and `b` into siblings, i.e. connects the sending side of one |
174 |
|
|
// to the receiving side of the other. This is not thread-safe. |
175 |
|
|
static void Entangle(MessagePort* a, MessagePort* b); |
176 |
|
|
static void Entangle(MessagePort* a, MessagePortData* b); |
177 |
|
|
|
178 |
|
|
// Detach this port's data for transferring. After this, the MessagePortData |
179 |
|
|
// is no longer associated with this handle, although it can still receive |
180 |
|
|
// messages. |
181 |
|
|
std::unique_ptr<MessagePortData> Detach(); |
182 |
|
|
|
183 |
|
|
void Close( |
184 |
|
|
v8::Local<v8::Value> close_callback = v8::Local<v8::Value>()) override; |
185 |
|
|
|
186 |
|
|
// Returns true if either data_ has been freed, or if the handle is being |
187 |
|
|
// closed. Equivalent to the [[Detached]] internal slot in the HTML Standard. |
188 |
|
|
// |
189 |
|
|
// If checking if a JavaScript MessagePort object is detached, this method |
190 |
|
|
// alone is often not enough, since the backing C++ MessagePort object may |
191 |
|
|
// have been deleted already. For all intents and purposes, an object with a |
192 |
|
|
// NULL pointer to the C++ MessagePort object is also detached. |
193 |
|
|
inline bool IsDetached() const; |
194 |
|
|
|
195 |
|
12 |
void MemoryInfo(MemoryTracker* tracker) const override { |
196 |
|
12 |
tracker->TrackField("data", data_); |
197 |
|
12 |
} |
198 |
|
|
|
199 |
|
12 |
SET_MEMORY_INFO_NAME(MessagePort) |
200 |
|
12 |
SET_SELF_SIZE(MessagePort) |
201 |
|
|
|
202 |
|
|
private: |
203 |
|
|
void OnClose() override; |
204 |
|
|
void OnMessage(); |
205 |
|
|
void TriggerAsync(); |
206 |
|
|
v8::MaybeLocal<v8::Value> ReceiveMessage(v8::Local<v8::Context> context, |
207 |
|
|
bool only_if_receiving); |
208 |
|
|
|
209 |
|
|
std::unique_ptr<MessagePortData> data_ = nullptr; |
210 |
|
|
bool receiving_messages_ = false; |
211 |
|
|
uv_async_t async_; |
212 |
|
|
|
213 |
|
|
friend class MessagePortData; |
214 |
|
|
}; |
215 |
|
|
|
216 |
|
|
v8::Local<v8::FunctionTemplate> GetMessagePortConstructorTemplate( |
217 |
|
|
Environment* env); |
218 |
|
|
|
219 |
|
|
} // namespace worker |
220 |
|
|
} // namespace node |
221 |
|
|
|
222 |
|
|
#endif // defined(NODE_WANT_INTERNALS) && NODE_WANT_INTERNALS |
223 |
|
|
|
224 |
|
|
|
225 |
|
|
#endif // SRC_NODE_MESSAGING_H_ |