FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

worker: support MessagePort passing in messages · nodejs/node@f447acd · GitHub

/ node Public

Commit f447acd

Browse files
addaleax
authored and
targos
committed
worker: support MessagePort passing in messages
Support passing `MessagePort` instances through other `MessagePort`s, as expected by the `MessagePort` spec. Thanks to Stephen Belanger for reviewing this change in its original PR. Refs: ayojs/ayo#106 PR-URL: #20876 Reviewed-By: Gireesh Punathil <gpunathi@in.ibm.com> Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com> Reviewed-By: Shingo Inoue <leko.noor@gmail.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Tiancheng "Timothy" Gu <timothygu99@gmail.com> Reviewed-By: John-David Dalton <john.david.dalton@gmail.com> Reviewed-By: Gus Caplan <me@gus.host>
1 parent 337be58 commit f447acd

6 files changed

Lines changed: 123 additions & 6 deletions

File tree

‎doc/api/errors.md‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -629,6 +629,12 @@ An operation outside the bounds of a `Buffer` was attempted.
629629
An attempt has been made to create a `Buffer` larger than the maximum allowed
630630
size.
631631

632+
<a id="ERR_CANNOT_TRANSFER_OBJECT"></a>
633+
### ERR_CANNOT_TRANSFER_OBJECT
634+
635+
The value passed to `postMessage()` contained an object that is not supported
636+
for transferring.
637+
632638
<a id="ERR_CANNOT_WATCH_SIGINT"></a>
633639
### ERR_CANNOT_WATCH_SIGINT
634640

@@ -1304,6 +1310,12 @@ strict compliance with the API specification (which in some cases may accept
13041310
`func(undefined)` and `func()` are treated identically, and the
13051311
[`ERR_INVALID_ARG_TYPE`][] error code may be used instead.
13061312

1313+
<a id="ERR_MISSING_MESSAGE_PORT_IN_TRANSFER_LIST"></a>
1314+
### ERR_MISSING_MESSAGE_PORT_IN_TRANSFER_LIST
1315+
1316+
A `MessagePort` was found in the object passed to a `postMessage()` call,
1317+
but not provided in the `transferList` for that call.
1318+
13071319
<a id="ERR_MISSING_MODULE"></a>
13081320
### ERR_MISSING_MODULE
13091321

‎doc/api/worker.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -83,7 +83,7 @@ the [HTML structured clone algorithm][]. In particular, it may contain circular
8383
references and objects like typed arrays that the `JSON` API is not able
8484
to stringify.
8585

86-
`transferList` may be a list of `ArrayBuffer` objects.
86+
`transferList` may be a list of `ArrayBuffer` and `MessagePort` objects.
8787
After transferring, they will not be usable on the sending side of the channel
8888
anymore (even if they are not contained in `value`).
8989

‎src/node_errors.h‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ namespace node {
1919
#define ERRORS_WITH_CODE(V) \
2020
V(ERR_BUFFER_OUT_OF_BOUNDS, RangeError) \
2121
V(ERR_BUFFER_TOO_LARGE, Error) \
22+
V(ERR_CANNOT_TRANSFER_OBJECT, TypeError) \
2223
V(ERR_CLOSED_MESSAGE_PORT, Error) \
2324
V(ERR_CONSTRUCT_CALL_REQUIRED, Error) \
2425
V(ERR_INDEX_OUT_OF_RANGE, RangeError) \
@@ -27,6 +28,7 @@ namespace node {
2728
V(ERR_INVALID_TRANSFER_OBJECT, TypeError) \
2829
V(ERR_MEMORY_ALLOCATION_FAILED, Error) \
2930
V(ERR_MISSING_ARGS, TypeError) \
31+
V(ERR_MISSING_MESSAGE_PORT_IN_TRANSFER_LIST, TypeError) \
3032
V(ERR_MISSING_MODULE, Error) \
3133
V(ERR_STRING_TOO_LONG, Error) \
3234

@@ -51,11 +53,14 @@ namespace node {
5153
// Errors with predefined static messages
5254

5355
#define PREDEFINED_ERROR_MESSAGES(V) \
56+
V(ERR_CANNOT_TRANSFER_OBJECT, "Cannot transfer object of unsupported type")\
5457
V(ERR_CLOSED_MESSAGE_PORT, "Cannot send data on closed MessagePort") \
5558
V(ERR_CONSTRUCT_CALL_REQUIRED, "Cannot call constructor without `new`") \
5659
V(ERR_INDEX_OUT_OF_RANGE, "Index out of range") \
5760
V(ERR_INVALID_TRANSFER_OBJECT, "Found invalid object in transferList") \
58-
V(ERR_MEMORY_ALLOCATION_FAILED, "Failed to allocate memory")
61+
V(ERR_MEMORY_ALLOCATION_FAILED, "Failed to allocate memory") \
62+
V(ERR_MISSING_MESSAGE_PORT_IN_TRANSFER_LIST, \
63+
"MessagePort was found in message but not listed in transferList")
5964

6065
#define V(code, message) \
6166
inline v8::Local<v8::Value> code(v8::Isolate* isolate) { \

‎src/node_messaging.cc‎

Lines changed: 76 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -41,14 +41,27 @@ namespace {
4141
// `MessagePort`s and `SharedArrayBuffer`s, and make new JS objects out of them.
4242
class DeserializerDelegate : public ValueDeserializer::Delegate {
4343
public:
44-
DeserializerDelegate(Message* m, Environment* env)
45-
: env_(env), msg_(m) {}
44+
DeserializerDelegate(Message* m,
45+
Environment* env,
46+
const std::vector<MessagePort*>& message_ports)
47+
: env_(env), msg_(m), message_ports_(message_ports) {}
48+
49+
MaybeLocal<Object> ReadHostObject(Isolate* isolate) override {
50+
// Currently, only MessagePort hosts objects are supported, so identifying
51+
// by the index in the message's MessagePort array is sufficient.
52+
uint32_t id;
53+
if (!deserializer->ReadUint32(&id))
54+
return MaybeLocal<Object>();
55+
CHECK_LE(id, message_ports_.size());
56+
return message_ports_[id]->object();
57+
};
4658

4759
ValueDeserializer* deserializer = nullptr;
4860

4961
private:
5062
Environment* env_;
5163
Message* msg_;
64+
const std::vector<MessagePort*>& message_ports_;
5265
};
5366

5467
} // anonymous namespace
@@ -58,7 +71,23 @@ MaybeLocal<Value> Message::Deserialize(Environment* env,
5871
EscapableHandleScope handle_scope(env->isolate());
5972
Context::Scope context_scope(context);
6073

61-
DeserializerDelegate delegate(this, env);
74+
// Create all necessary MessagePort handles.
75+
std::vector<MessagePort*> ports(message_ports_.size());
76+
for (uint32_t i = 0; i < message_ports_.size(); ++i) {
77+
ports[i] = MessagePort::New(env,
78+
context,
79+
std::move(message_ports_[i]));
80+
if (ports[i] == nullptr) {
81+
for (MessagePort* port : ports) {
82+
// This will eventually release the MessagePort object itself.
83+
port->Close();
84+
}
85+
return MaybeLocal<Value>();
86+
}
87+
}
88+
message_ports_.clear();
89+
90+
DeserializerDelegate delegate(this, env, ports);
6291
ValueDeserializer deserializer(
6392
env->isolate(),
6493
reinterpret_cast<const uint8_t*>(main_message_buf_.data),
@@ -83,6 +112,10 @@ MaybeLocal<Value> Message::Deserialize(Environment* env,
83112
deserializer.ReadValue(context).FromMaybe(Local<Value>()));
84113
}
85114

115+
void Message::AddMessagePort(std::unique_ptr<MessagePortData>&& data) {
116+
message_ports_.emplace_back(std::move(data));
117+
}
118+
86119
namespace {
87120

88121
// This tells V8 how to serialize objects that it does not understand
@@ -97,12 +130,43 @@ class SerializerDelegate : public ValueSerializer::Delegate {
97130
env_->isolate()->ThrowException(Exception::Error(message));
98131
}
99132

133+
Maybe<bool> WriteHostObject(Isolate* isolate, Local<Object> object) override {
134+
if (env_->message_port_constructor_template()->HasInstance(object)) {
135+
return WriteMessagePort(Unwrap<MessagePort>(object));
136+
}
137+
138+
THROW_ERR_CANNOT_TRANSFER_OBJECT(env_);
139+
return Nothing<bool>();
140+
}
141+
142+
void Finish() {
143+
// Only close the MessagePort handles and actually transfer them
144+
// once we know that serialization succeeded.
145+
for (MessagePort* port : ports_) {
146+
port->Close();
147+
msg_->AddMessagePort(port->Detach());
148+
}
149+
}
150+
100151
ValueSerializer* serializer = nullptr;
101152

102153
private:
154+
Maybe<bool> WriteMessagePort(MessagePort* port) {
155+
for (uint32_t i = 0; i < ports_.size(); i++) {
156+
if (ports_[i] == port) {
157+
serializer->WriteUint32(i);
158+
return Just(true);
159+
}
160+
}
161+
162+
THROW_ERR_MISSING_MESSAGE_PORT_IN_TRANSFER_LIST(env_);
163+
return Nothing<bool>();
164+
}
165+
103166
Environment* env_;
104167
Local<Context> context_;
105168
Message* msg_;
169+
std::vector<MessagePort*> ports_;
106170

107171
friend class worker::Message;
108172
};
@@ -131,7 +195,7 @@ Maybe<bool> Message::Serialize(Environment* env,
131195
Local<Value> entry;
132196
if (!transfer_list->Get(context, i).ToLocal(&entry))
133197
return Nothing<bool>();
134-
// Currently, we support ArrayBuffers.
198+
// Currently, we support ArrayBuffers and MessagePorts.
135199
if (entry->IsArrayBuffer()) {
136200
Local<ArrayBuffer> ab = entry.As<ArrayBuffer>();
137201
// If we cannot render the ArrayBuffer unusable in this Isolate and
@@ -144,6 +208,12 @@ Maybe<bool> Message::Serialize(Environment* env,
144208
array_buffers.push_back(ab);
145209
serializer.TransferArrayBuffer(id, ab);
146210
continue;
211+
} else if (env->message_port_constructor_template()
212+
->HasInstance(entry)) {
213+
MessagePort* port = Unwrap<MessagePort>(entry.As<Object>());
214+
CHECK_NE(port, nullptr);
215+
delegate.ports_.push_back(port);
216+
continue;
147217
}
148218

149219
THROW_ERR_INVALID_TRANSFER_OBJECT(env);
@@ -167,6 +237,8 @@ Maybe<bool> Message::Serialize(Environment* env,
167237
contents.ByteLength() });
168238
}
169239

240+
delegate.Finish();
241+
170242
// The serializer gave us a buffer allocated using `malloc()`.
171243
std::pair<uint8_t*, size_t> data = serializer.Release();
172244
main_message_buf_ =

‎src/node_messaging.h‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,9 +37,14 @@ class Message {
3737
v8::Local<v8::Value> input,
3838
v8::Local<v8::Value> transfer_list);
3939

40+
// Internal method of Message that is called once serialization finishes
41+
// and that transfers ownership of `data` to this message.
42+
void AddMessagePort(std::unique_ptr<MessagePortData>&& data);
43+
4044
private:
4145
MallocedBuffer<char> main_message_buf_;
4246
std::vector<MallocedBuffer<char>> array_buffer_contents_;
47+
std::vector<std::unique_ptr<MessagePortData>> message_ports_;
4348

4449
friend class MessagePort;
4550
};

‎test/parallel/test-message-port-message-port-transferring.js‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
// Flags: --experimental-worker
2+
'use strict';
3+
const common = require('../common');
4+
const assert = require('assert');
5+
6+
const { MessageChannel } = require('worker');
7+
8+
{
9+
const { port1: basePort1, port2: basePort2 } = new MessageChannel();
10+
const {
11+
port1: transferredPort1, port2: transferredPort2
12+
} = new MessageChannel();
13+
14+
basePort1.postMessage({ transferredPort1 }, [ transferredPort1 ]);
15+
basePort2.on('message', common.mustCall(({ transferredPort1 }) => {
16+
transferredPort1.postMessage('foobar');
17+
transferredPort2.on('message', common.mustCall((msg) => {
18+
assert.strictEqual(msg, 'foobar');
19+
transferredPort1.close(common.mustCall());
20+
basePort1.close(common.mustCall());
21+
}));
22+
}));
23+
}

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL