GV

GeezerVison

The World Wide Web, restored to 800 pixels of glory.

Home|About|Gaming cabinet|Archives|How it works|Starter sites

On this site

  • Sign in
  • ActionsCI/CD and automation
  • CodespacesCloud development environments
  • All Features
  • Documentation
  • Training
  • Certifications
  • Build with Copilot
  • Customize Copilot
  • Integrations
  • Trending repos
  • GitHub Sponsors
  • Developer articles
  • Enterprises

This view

  • Original page
  • New address

GeezerVison is showing github.com — perf(runtime): skip unread BroadcastChannel deliveries and let idle c… · NativeScript/android@71719d1

File tree

Expand file treeCollapse file tree
Open diff view settings
Filter options
  • test-app
    • app/src/main/assets/app/tests
      • testMessaging.js
    • runtime/src/main/cpp
      • Messaging.cpp
      • Messaging.h
      • js
        • broadcast-channel.js
        • message-channel.js
Expand file treeCollapse file tree
Open diff view settings
Collapse file

‎test-app/app/src/main/assets/app/tests/testMessaging.js‎

Copy file name to clipboardExpand all lines: test-app/app/src/main/assets/app/tests/testMessaging.js
+200Lines changed: 200 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -123,6 +123,206 @@ describe("Messaging runtime edges", function () {
123123
}, SETTLE);
124124
}, SETTLE);
125125
});
126+
127+
var channelId = 0;
128+
function channelName() {
129+
return "messaging-edges-" + (channelId++);
130+
}
131+
132+
function pollGC(predicate, cb) {
133+
var turns = 0;
134+
(function poll() {
135+
__collect();
136+
if (predicate() || turns >= 100) {
137+
cb();
138+
return;
139+
}
140+
turns++;
141+
setTimeout(poll, 20);
142+
})();
143+
}
144+
145+
it("delivers a message posted before the listener was attached in the same turn", function (done) {
146+
var name = channelName();
147+
var sender = new BroadcastChannel(name);
148+
var receiver = new BroadcastChannel(name);
149+
sender.postMessage("early");
150+
receiver.onmessage = function (event) {
151+
expect(event.data).toBe("early");
152+
sender.close();
153+
receiver.close();
154+
done();
155+
};
156+
});
157+
158+
it("discards a message that arrived while the channel had no listener", function (done) {
159+
var name = channelName();
160+
var sender = new BroadcastChannel(name);
161+
var receiver = new BroadcastChannel(name);
162+
var got = [];
163+
sender.postMessage("unheard");
164+
setTimeout(function () {
165+
receiver.onmessage = function (event) { got.push(event.data); };
166+
sender.postMessage("heard");
167+
setTimeout(function () {
168+
expect(got).toEqual(["heard"]);
169+
sender.close();
170+
receiver.close();
171+
done();
172+
}, SETTLE);
173+
}, SETTLE);
174+
});
175+
176+
it("stops delivering when the last listener goes and resumes when one returns", function (done) {
177+
var name = channelName();
178+
var sender = new BroadcastChannel(name);
179+
var receiver = new BroadcastChannel(name);
180+
var got = [];
181+
var listener = function (event) { got.push(event.data); };
182+
receiver.addEventListener("message", listener);
183+
receiver.removeEventListener("message", listener);
184+
sender.postMessage("gone");
185+
setTimeout(function () {
186+
receiver.addEventListener("message", listener);
187+
sender.postMessage("back");
188+
setTimeout(function () {
189+
expect(got).toEqual(["back"]);
190+
sender.close();
191+
receiver.close();
192+
done();
193+
}, SETTLE);
194+
}, SETTLE);
195+
});
196+
197+
it("dispatches a single MessageEvent at the channel itself", function (done) {
198+
var name = channelName();
199+
var sender = new BroadcastChannel(name);
200+
var receiver = new BroadcastChannel(name);
201+
var events = [];
202+
receiver.addEventListener("message", function (event) {
203+
expect(this).toBe(receiver);
204+
expect(event.currentTarget).toBe(receiver);
205+
events.push(event);
206+
});
207+
receiver.onmessage = function (event) { events.push(event); };
208+
sender.postMessage("once");
209+
setTimeout(function () {
210+
expect(events.length).toBe(2);
211+
expect(events[0]).toBe(events[1]);
212+
expect(events[0] instanceof MessageEvent).toBe(true);
213+
expect(events[0].target).toBe(receiver);
214+
sender.close();
215+
receiver.close();
216+
done();
217+
}, SETTLE);
218+
});
219+
220+
it("delivers to the channels of a group in creation order", function (done) {
221+
var name = channelName();
222+
var sender = new BroadcastChannel(name);
223+
var receivers = [];
224+
var order = [];
225+
for (var i = 0; i < 16; i++) {
226+
(function (index) {
227+
var channel = new BroadcastChannel(name);
228+
channel.onmessage = function () { order.push(index); };
229+
receivers.push(channel);
230+
})(i);
231+
}
232+
sender.postMessage("fan");
233+
setTimeout(function () {
234+
var expected = [];
235+
for (var j = 0; j < receivers.length; j++) { expected.push(j); }
236+
expect(order).toEqual(expected);
237+
sender.close();
238+
receivers.forEach(function (channel) { channel.close(); });
239+
done();
240+
}, SETTLE);
241+
});
242+
243+
it("lets an unreferenced channel without a listener be collected", function (done) {
244+
var wr = (function () {
245+
return new WeakRef(new BroadcastChannel(channelName()));
246+
})();
247+
pollGC(function () { return wr.deref() === undefined; }, function () {
248+
expect(wr.deref()).toBeUndefined();
249+
done();
250+
});
251+
});
252+
253+
it("lets a channel be collected once its last listener is removed", function (done) {
254+
var wr = (function () {
255+
var channel = new BroadcastChannel(channelName());
256+
channel.onmessage = function () {};
257+
channel.onmessage = null;
258+
return new WeakRef(channel);
259+
})();
260+
pollGC(function () { return wr.deref() === undefined; }, function () {
261+
expect(wr.deref()).toBeUndefined();
262+
done();
263+
});
264+
});
265+
266+
it("lets a closed channel be collected even with a listener", function (done) {
267+
var wr = (function () {
268+
var channel = new BroadcastChannel(channelName());
269+
channel.onmessage = function () {};
270+
channel.close();
271+
return new WeakRef(channel);
272+
})();
273+
pollGC(function () { return wr.deref() === undefined; }, function () {
274+
expect(wr.deref()).toBeUndefined();
275+
done();
276+
});
277+
});
278+
279+
it("keeps an unreferenced channel with a messageerror listener alive", function (done) {
280+
var wr = (function () {
281+
var channel = new BroadcastChannel(channelName());
282+
channel.onmessageerror = function () {};
283+
return new WeakRef(channel);
284+
})();
285+
var turns = 0;
286+
(function poll() {
287+
__collect();
288+
if (++turns < 10) {
289+
setTimeout(poll, 20);
290+
return;
291+
}
292+
var channel = wr.deref();
293+
expect(channel instanceof BroadcastChannel).toBe(true);
294+
if (channel) { channel.close(); }
295+
done();
296+
})();
297+
});
298+
299+
it("keeps delivering to an unreferenced channel that has a listener", function (done) {
300+
var name = channelName();
301+
var sender = new BroadcastChannel(name);
302+
var got = null;
303+
var receiver = null;
304+
(function () {
305+
new BroadcastChannel(name).onmessage = function (event) {
306+
got = event.data;
307+
receiver = event.target;
308+
};
309+
})();
310+
var turns = 0;
311+
(function poll() {
312+
__collect();
313+
if (++turns < 10) {
314+
setTimeout(poll, 20);
315+
return;
316+
}
317+
sender.postMessage("still here");
318+
setTimeout(function () {
319+
expect(got).toBe("still here");
320+
sender.close();
321+
if (receiver) { receiver.close(); }
322+
done();
323+
}, SETTLE);
324+
})();
325+
});
126326
});
127327

128328
describe("node:worker_threads", function () {
Collapse file

‎test-app/runtime/src/main/cpp/Messaging.cpp‎

Copy file name to clipboardExpand all lines: test-app/runtime/src/main/cpp/Messaging.cpp
+58-12Lines changed: 58 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -2,11 +2,11 @@
22

33
#include <algorithm>
44
#include <initializer_list>
5-
#include <set>
65
#include <shared_mutex>
76
#include <unordered_map>
87
#include <unordered_set>
98
#include <utility>
9+
#include <vector>
1010

1111
#include "ArgConverter.h"
1212
#include "BuiltinLoader.h"
@@ -194,7 +194,10 @@ class SiblingGroup final : public std::enable_shared_from_this<SiblingGroup> {
194194
// group is one channel's two ends.
195195
const bool named_ = false;
196196
std::shared_mutex mutex_;
197-
std::set<PortData*> ports_;
197+
// In join order: HTML delivers a broadcast to its destinations in the order
198+
// the channels were created, and a burst that finds none of them already
199+
// scheduled posts their drains in that order.
200+
std::vector<PortData*> ports_;
198201
};
199202

200203
namespace {
@@ -233,7 +236,7 @@ DispatchResult SiblingGroup::Dispatch(PortData* source, std::shared_ptr<Message>
233236
std::string* error) {
234237
std::shared_lock<std::shared_mutex> lock(this->mutex_);
235238

236-
if (this->ports_.find(source) == this->ports_.end()) {
239+
if (std::find(this->ports_.begin(), this->ports_.end(), source) == this->ports_.end()) {
237240
if (error != nullptr) {
238241
*error = "Source MessagePort is not entangled with this group.";
239242
}
@@ -278,7 +281,7 @@ void SiblingGroup::Entangle(PortData* port) {
278281
void SiblingGroup::Entangle(std::initializer_list<PortData*> ports) {
279282
std::unique_lock<std::shared_mutex> lock(this->mutex_);
280283
for (PortData* data : ports) {
281-
this->ports_.insert(data);
284+
this->ports_.push_back(data);
282285
// group_ is written under the port's own mutex, which is what lets
283286
// PortData::Dispatch read it without racing a disentangle. Taken here in
284287
// the only legal order: this group's lock is already held.
@@ -292,7 +295,8 @@ void SiblingGroup::Disentangle(PortData* data) {
292295
// Keeps the group alive past the last member dropping its reference.
293296
std::shared_ptr<SiblingGroup> self = this->shared_from_this();
294297
std::unique_lock<std::shared_mutex> lock(this->mutex_);
295-
this->ports_.erase(data);
298+
this->ports_.erase(std::remove(this->ports_.begin(), this->ports_.end(), data),
299+
this->ports_.end());
296300
{
297301
std::lock_guard<std::mutex> dataLock(data->mutex_);
298302
data->group_.reset();
@@ -317,10 +321,11 @@ PortData::~PortData() {
317321

318322
void PortData::AddToIncomingQueue(std::shared_ptr<Message> message) {
319323
std::lock_guard<std::mutex> lock(this->mutex_);
324+
// Still holding the mutex: an owner read outside it could be detached by
325+
// the time the wake reaches it, and Start() checks the backlog under it.
326+
bool wake = this->owner_ != nullptr && this->owner_->WantsDrainFor(*message);
320327
this->incoming_.push_back(std::move(message));
321-
if (this->owner_ != nullptr) {
322-
// Still holding the mutex: an owner read outside it could be detached by
323-
// the time the wake reaches it.
328+
if (wake) {
324329
this->owner_->TriggerAsync();
325330
}
326331
}
@@ -397,6 +402,7 @@ std::shared_ptr<NativeMessagePort> NativeMessagePort::New(Local<Context> context
397402
} else {
398403
port->data_ = std::make_unique<PortData>(port.get());
399404
if (group != nullptr) {
405+
port->broadcast_ = true;
400406
group->Entangle(port->data_.get());
401407
}
402408
}
@@ -449,6 +455,19 @@ void NativeMessagePort::Stop() {
449455
this->receiving_ = false;
450456
}
451457

458+
bool NativeMessagePort::WantsDrainFor(const Message& message) const {
459+
// A port that was never started still learns that its sibling died.
460+
return this->receiving_ || message.IsCloseMessage();
461+
}
462+
463+
void NativeMessagePort::SetBroadcastTarget(Local<Object> target) {
464+
if (target.IsEmpty()) {
465+
this->broadcastTarget_.Reset();
466+
} else {
467+
this->broadcastTarget_.Reset(this->isolate_, target);
468+
}
469+
}
470+
452471
std::unique_ptr<PortData> NativeMessagePort::Detach() {
453472
// owner_ drops under the data mutex, so a producer either wakes this port
454473
// before the detach or never sees an owner at all. Node carries a separate
@@ -470,6 +489,7 @@ void NativeMessagePort::CloseHandle() {
470489
0, nullptr, v8::kEmbedderDataTypeTagDefault);
471490
this->wrapper_.Reset();
472491
}
492+
this->broadcastTarget_.Reset();
473493
MessagingState* state = State(isolate);
474494
if (state != nullptr) {
475495
state->livePorts.erase(this->shared_from_this());
@@ -499,8 +519,9 @@ void NativeMessagePort::Close() {
499519
}
500520
// Last, on the wrapper the port has just let go of, so a listener finds an
501521
// already-detached port and a close() from inside one is a no-op rather than
502-
// a recursion.
503-
if (!wrapper.IsEmpty()) {
522+
// a recursion. A broadcast port's wrapper never reaches app code, so there
523+
// is no one to tell.
524+
if (!wrapper.IsEmpty() && !this->broadcast_) {
504525
this->EmitClose(wrapper);
505526
}
506527
}
@@ -669,6 +690,17 @@ void NativeMessagePort::Drain() {
669690
return;
670691
}
671692

693+
Local<Object> receiver = wrapper;
694+
if (this->broadcast_) {
695+
// Per message: a listener can add or remove the channel's last listener.
696+
// With none, the event would reach no one and reading the value has no
697+
// other observable effect, so it is never read.
698+
if (this->broadcastTarget_.IsEmpty()) {
699+
continue;
700+
}
701+
receiver = this->broadcastTarget_.Get(isolate);
702+
}
703+
672704
Local<Value> payload;
673705
Local<Value> ports = v8::Undefined(isolate);
674706
bool read;
@@ -693,12 +725,12 @@ void NativeMessagePort::Drain() {
693725
}
694726
}
695727
if (!read) {
696-
this->Emit(context, wrapper, emitMessage, payload, v8::Undefined(isolate),
728+
this->Emit(context, receiver, emitMessage, payload, v8::Undefined(isolate),
697729
"messageerror");
698730
reschedule = true;
699731
break;
700732
}
701-
if (!this->Emit(context, wrapper, emitMessage, payload, ports, "message")) {
733+
if (!this->Emit(context, receiver, emitMessage, payload, ports, "message")) {
702734
reschedule = true;
703735
break;
704736
}
@@ -882,6 +914,19 @@ void StopCallback(const FunctionCallbackInfo<Value>& info) {
882914
}
883915
}
884916

917+
void SetBroadcastTargetCallback(const FunctionCallbackInfo<Value>& info) {
918+
Local<Object> wrapper;
919+
if (!PortArg(info, 0, &wrapper)) {
920+
return;
921+
}
922+
NativeMessagePort* port = PortFromWrapper(info.GetIsolate(), wrapper);
923+
if (port == nullptr) {
924+
return;
925+
}
926+
port->SetBroadcastTarget(info.Length() > 1 && info[1]->IsObject() ? info[1].As<Object>()
927+
: Local<Object>());
928+
}
929+
885930
void CloseCallback(const FunctionCallbackInfo<Value>& info) {
886931
Local<Object> wrapper;
887932
if (!PortArg(info, 0, &wrapper)) {
@@ -1038,6 +1083,7 @@ MaybeLocal<Object> CreateBinding(Local<Context> context) {
10381083
tns::SetMethod(context, binding, "postMessage", PostMessageCallback);
10391084
tns::SetMethod(context, binding, "start", StartCallback);
10401085
tns::SetMethod(context, binding, "stop", StopCallback);
1086+
tns::SetMethod(context, binding, "setBroadcastTarget", SetBroadcastTargetCallback);
10411087
tns::SetMethod(context, binding, "close", CloseCallback);
10421088
tns::SetMethod(context, binding, "drainOne", DrainOneCallback);
10431089
tns::SetMethodNoSideEffect(context, binding, "isDetached", IsDetachedCallback);

0 commit comments

Comments
0 (0)
You are visitor 01402520 — GeezerVison does not store the pages you look up.
Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required.