Skip to content

Commit 5ca0268

Browse files
committed
feat(event-loop): merge timers into the ordered lane; route __runOnMainThread through the internal lane
Timers merge (with tombstones): - Timers no longer owns a Java Handler: each scheduled timer posts one anonymous token through the EventLoop's ordered lane, and the token drain runs the earliest due item across timers and ordered macrotasks - one due-ordered domain, still strictly FIFO with Handler.post on the same looper. Token 'when' computation is unchanged, so the quiescent setTimeout-vs-Handler.post contract is preserved exactly. - clearTimeout/clearInterval tombstone the sorted entry instead of erasing it: the cleared timer's already-queued token consumes its own slot as a no-op, so no token gains surplus capacity to run a later-scheduled item (timer or macrotask) ahead of foreign Java messages queued between the two token positions. This also fixes the pre-existing congestion deviation where a leftover token could fire a later timer early. - FireTimer's internals (sub-ms sorted list, chromium-style interval catch-up, nesting clamp, TryCatch discipline) are untouched; the check-and-run happens in one OrderedTaskSource::RunIfEarliest call under a single Locker acquisition, because background threads mutate the timer bookkeeping through setTimeout under multithreaded JS. - TimerHandler.java is deleted. __runOnMainThread promotion: - The 2MB main-looper pipe and RunOnMainThreadFdCallback are replaced by bare internal-lane entries on the main runtime's EventLoop. Bare entries skip the loop's Locker/checkpoint: the closure locks the CALLER's isolate (a worker's, under multithreaded JS), and taking the main isolate's Locker first would nest Lockers across isolates and can deadlock against worker->main JNI entry paths. Delivery stays one-per-poll, matching the old fd callback. - The callback cache is now mutex-guarded: it was written from arbitrary threads under different isolates' Lockers, which provide no mutual exclusion; RemoveIsolateEntries also no longer erases while range-iterating. - Uncaught exceptions in the callbacks now surface as pending Java exceptions via the loop's guard instead of unwinding C++ through the ALooper callback frame. Tests: tombstone ordering specs (cleared timer's token vs java posts, for both a later timer and a queued macrotask), against the native __ns__ timers - the test app's global setTimeout is an old Handler-based polyfill with colliding ids, not the runtime timers.
1 parent 8da0d88 commit 5ca0268

10 files changed

Lines changed: 308 additions & 230 deletions

File tree

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

Lines changed: 40 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -86,10 +86,12 @@ describe("event loop ordered macrotasks", function () {
8686
Promise.resolve().then(() => order.push("microtask"));
8787
});
8888

89-
it("stays FIFO-ordered with setTimeout(0)", function (done) {
89+
// native timers (__ns__*): the app-level `setTimeout` global in this test
90+
// app is an old Handler-based polyfill, not the runtime timers
91+
it("stays FIFO-ordered with native setTimeout(0)", function (done) {
9092
const order = [];
9193
__ns__queueMacrotask(() => order.push("macro1"));
92-
setTimeout(() => order.push("timeout"), 0);
94+
__ns__setTimeout(() => order.push("timeout"), 0);
9395
__ns__queueMacrotask(() => {
9496
order.push("macro2");
9597
expect(order).toEqual(["macro1", "timeout", "macro2"]);
@@ -116,6 +118,42 @@ describe("event loop ordered macrotasks", function () {
116118
});
117119
});
118120

121+
// clearTimeout leaves a tombstone in the merged ordered domain, so the
122+
// cleared timer's already-queued token consumes its own slot as a no-op
123+
// instead of running a later-scheduled item ahead of Java messages queued
124+
// between the two tokens' positions.
125+
describe("event loop ordered tombstones", function () {
126+
it("cleared timeout's token does not run a later timer ahead of java posts", function (done) {
127+
const order = [];
128+
const handler = new android.os.Handler(android.os.Looper.myLooper());
129+
const t1 = __ns__setTimeout(() => order.push("cleared"), 0);
130+
__ns__clearTimeout(t1);
131+
handler.post(new java.lang.Runnable({
132+
run: () => order.push("java")
133+
}));
134+
__ns__setTimeout(() => {
135+
order.push("t2");
136+
expect(order).toEqual(["java", "t2"]);
137+
done();
138+
}, 0);
139+
});
140+
141+
it("cleared timeout's token does not run a queued macrotask ahead of java posts", function (done) {
142+
const order = [];
143+
const handler = new android.os.Handler(android.os.Looper.myLooper());
144+
const t1 = __ns__setTimeout(() => order.push("cleared"), 0);
145+
__ns__clearTimeout(t1);
146+
handler.post(new java.lang.Runnable({
147+
run: () => order.push("java")
148+
}));
149+
__ns__queueMacrotask(() => {
150+
order.push("macro");
151+
expect(order).toEqual(["java", "macro"]);
152+
done();
153+
});
154+
});
155+
});
156+
119157
describe("event loop internal lane", function () {
120158
// Regression for the eventfd unit-accounting bug: a worker reply's wakeup
121159
// arriving while an overdue waitAsync timeout is still unsignaled must not

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

Lines changed: 43 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -684,48 +684,60 @@ void CallbackHandlers::RunOnMainThreadCallback(const FunctionCallbackInfo<v8::Va
684684
uint64_t key = ++count_;
685685
Local<v8::Function> callback = args[0].As<v8::Function>();
686686

687-
bool inserted;
688-
std::tie(std::ignore, inserted) = cache_.try_emplace(key, isolate, callback);
689-
assert(inserted && "Main thread callback ID should not be duplicated");
687+
{
688+
std::lock_guard<std::mutex> lock(cacheMutex_);
689+
bool inserted;
690+
std::tie(std::ignore, inserted) = cache_.try_emplace(key, isolate, callback);
691+
assert(inserted && "Main thread callback ID should not be duplicated");
692+
}
690693

691-
auto value = Callback(key);
692-
auto size = sizeof(Callback);
693-
auto wrote = write(Runtime::GetWriter(),&value , size);
694+
auto mainLoop = Runtime::GetMainEventLoop();
695+
if (mainLoop == nullptr) {
696+
return;
697+
}
698+
// bare entry: the closure locks the CALLER's isolate (possibly a
699+
// worker's), so the loop must not take the main isolate's Locker first -
700+
// nesting the two can deadlock against multithreaded-JS entry paths
701+
mainLoop->PostInternalBare([key]() { RunMainThreadEntry(key); });
694702
}
695703

696-
int CallbackHandlers::RunOnMainThreadFdCallback(int fd, int events, void *data) {
697-
struct Callback value;
698-
auto size = sizeof(Callback);
699-
ssize_t nr = read(fd, &value, sizeof(value));
700-
701-
auto key = value.id_;
702-
703-
auto it = cache_.find(key);
704-
if (it == cache_.end()) {
705-
return 1;
704+
void CallbackHandlers::RunMainThreadEntry(uint64_t key) {
705+
Isolate *isolate;
706+
{
707+
std::lock_guard<std::mutex> lock(cacheMutex_);
708+
auto it = cache_.find(key);
709+
if (it == cache_.end()) {
710+
return;
711+
}
712+
isolate = it->second.isolate_;
706713
}
707714

708-
Isolate *isolate = it->second.isolate_;
709715
v8::Locker locker(isolate);
710716
Isolate::Scope isolate_scope(isolate);
711717
HandleScope handle_scope(isolate);
712-
Local<v8::Function> cb = it->second.callback_.Get(isolate);
718+
Local<v8::Function> cb;
719+
{
720+
std::lock_guard<std::mutex> lock(cacheMutex_);
721+
auto it = cache_.find(key);
722+
if (it == cache_.end()) {
723+
return;
724+
}
725+
cb = it->second.callback_.Get(isolate);
726+
cache_.erase(it);
727+
}
713728
Runtime* runtime = Runtime::GetRuntime(isolate);
714729
v8::Local<v8::Context> context = runtime->GetContext();
715730
Context::Scope context_scope(context);
716-
// erase the it here as we're already done with its values and the callback might invalidate the iterator
717-
cache_.erase(it);
718731

719732
v8::TryCatch tc(isolate);
720733

721734
cb->Call(context, context->Global(), 0, nullptr); // ignore JS return value
722735

723736
if (tc.HasCaught() &&
724737
!NativeScriptException::ContainUncaughtCallbackException(isolate, tc)) {
738+
// surfaces via the event loop's guard as a pending Java exception
725739
throw NativeScriptException(tc);
726740
}
727-
728-
return 1;
729741
}
730742

731743
void CallbackHandlers::LogMethodCallback(const v8::FunctionCallbackInfo<v8::Value> &args) {
@@ -1611,13 +1623,19 @@ void CallbackHandlers::CallWorkerScopeOnErrorHandle(Isolate *isolate, TryCatch &
16111623
}
16121624

16131625
void CallbackHandlers::RemoveIsolateEntries(v8::Isolate *isolate) {
1614-
for (auto &item: cache_) {
1615-
if (item.second.isolate_ == isolate) {
1616-
cache_.erase(item.first);
1626+
{
1627+
std::lock_guard<std::mutex> lock(cacheMutex_);
1628+
for (auto it = cache_.begin(); it != cache_.end();) {
1629+
if (it->second.isolate_ == isolate) {
1630+
it = cache_.erase(it);
1631+
} else {
1632+
++it;
1633+
}
16171634
}
16181635
}
16191636
}
16201637
robin_hood::unordered_map<uint64_t, CallbackHandlers::CacheEntry> CallbackHandlers::cache_;
1638+
std::mutex CallbackHandlers::cacheMutex_;
16211639

16221640

16231641
std::atomic_int64_t CallbackHandlers::count_ = {0};

‎test-app/runtime/src/main/cpp/CallbackHandlers.h‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33

44
#include <string>
55
#include <map>
6+
#include <mutex>
67
#include <vector>
78
#include "JEnv.h"
89
#include "ArgsWrapper.h"
@@ -73,7 +74,9 @@ namespace tns {
7374

7475
static void RunOnMainThreadCallback(const v8::FunctionCallbackInfo<v8::Value> &args);
7576

76-
static int RunOnMainThreadFdCallback(int fd, int events, void* data);
77+
// runs one cached __runOnMainThread callback on the main thread,
78+
// entering the CALLER's isolate (posted as a bare event-loop entry)
79+
static void RunMainThreadEntry(uint64_t key);
7780

7881
static void LogMethodCallback(const v8::FunctionCallbackInfo<v8::Value> &args);
7982

@@ -241,6 +244,10 @@ namespace tns {
241244
};
242245

243246
static robin_hood::unordered_map<uint64_t, CacheEntry> cache_;
247+
// guards cache_: __runOnMainThread is callable from any runtime's
248+
// thread (multithreaded JS, workers), each under a different
249+
// isolate's Locker, so the Lockers provide no mutual exclusion
250+
static std::mutex cacheMutex_;
244251

245252

246253
};

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

Lines changed: 75 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -116,6 +116,10 @@ void EventLoop::BindToCurrentThread() {
116116
for (auto& pair : ordered_.delayed) {
117117
env.CallVoidMethod(handler_, EVENT_LOOP_HANDLER_POST, (jlong) std::ceil(pair.first));
118118
}
119+
for (auto when : pendingTokens_) {
120+
env.CallVoidMethod(handler_, EVENT_LOOP_HANDLER_POST, when);
121+
}
122+
pendingTokens_.clear();
119123
}
120124

121125
void EventLoop::Shutdown() {
@@ -216,39 +220,65 @@ void EventLoop::PostInternal(std::function<void()> fn) {
216220
if (stopped_) {
217221
return;
218222
}
219-
PostInternalLocked(Entry{nullptr, std::move(fn), true, 0}, 0);
223+
PostInternalLocked(Entry{nullptr, std::move(fn), true, false, 0}, 0);
220224
}
221225

222226
void EventLoop::PostInternalDelayed(std::function<void()> fn, double delayMs) {
223227
std::lock_guard<std::mutex> lock(mutex_);
224228
if (stopped_) {
225229
return;
226230
}
227-
PostInternalLocked(Entry{nullptr, std::move(fn), true, 0}, delayMs);
231+
PostInternalLocked(Entry{nullptr, std::move(fn), true, false, 0}, delayMs);
232+
}
233+
234+
void EventLoop::PostInternalBare(std::function<void()> fn) {
235+
std::lock_guard<std::mutex> lock(mutex_);
236+
if (stopped_) {
237+
return;
238+
}
239+
PostInternalLocked(Entry{nullptr, std::move(fn), true, true, 0}, 0);
228240
}
229241

230242
void EventLoop::PostOrdered(std::function<void()> fn) {
231243
std::lock_guard<std::mutex> lock(mutex_);
232244
if (stopped_) {
233245
return;
234246
}
235-
PostOrderedLocked(Entry{nullptr, std::move(fn), true, 0}, 0);
247+
PostOrderedLocked(Entry{nullptr, std::move(fn), true, false, 0}, 0);
236248
}
237249

238250
void EventLoop::PostOrderedDelayed(std::function<void()> fn, double delayMs) {
239251
std::lock_guard<std::mutex> lock(mutex_);
240252
if (stopped_) {
241253
return;
242254
}
243-
PostOrderedLocked(Entry{nullptr, std::move(fn), true, 0}, delayMs);
255+
PostOrderedLocked(Entry{nullptr, std::move(fn), true, false, 0}, delayMs);
256+
}
257+
258+
void EventLoop::PostOrderedToken(jlong uptimeMillis) {
259+
std::lock_guard<std::mutex> lock(mutex_);
260+
if (stopped_) {
261+
return;
262+
}
263+
if (handler_ != nullptr) {
264+
JEnv env;
265+
env.CallVoidMethod(handler_, EVENT_LOOP_HANDLER_POST, uptimeMillis);
266+
} else {
267+
pendingTokens_.push_back(uptimeMillis);
268+
}
269+
}
270+
271+
void EventLoop::SetTimerSource(OrderedTaskSource* source) {
272+
// home thread only, like every consumer of timerSource_
273+
timerSource_ = source;
244274
}
245275

246276
void EventLoop::PostV8Task(std::unique_ptr<Task> task, bool nestable, double delaySeconds) {
247277
std::lock_guard<std::mutex> lock(mutex_);
248278
if (stopped_) {
249279
return;
250280
}
251-
PostInternalLocked(Entry{std::move(task), nullptr, nestable, 0},
281+
PostInternalLocked(Entry{std::move(task), nullptr, nestable, false, 0},
252282
delaySeconds * 1000.0);
253283
}
254284

@@ -288,6 +318,17 @@ std::unique_ptr<EventLoop::Entry> EventLoop::TakeDueLocked(Lane& lane, bool nest
288318
return nullptr;
289319
}
290320

321+
double EventLoop::PeekDueLocked(Lane& lane, double now) {
322+
// immediate entries are enqueued with monotonically increasing times, so
323+
// the front is the earliest
324+
double due = lane.immediate.empty() ? -1 : lane.immediate.front().time;
325+
if (!lane.delayed.empty() && lane.delayed.begin()->first <= now &&
326+
(due < 0 || lane.delayed.begin()->first < due)) {
327+
due = lane.delayed.begin()->first;
328+
}
329+
return due;
330+
}
331+
291332
void EventLoop::ArmTimerLocked(double now) {
292333
if (timerFd_ == -1) {
293334
return;
@@ -313,6 +354,12 @@ void EventLoop::ArmTimerLocked(double now) {
313354
}
314355

315356
void EventLoop::RunEntry(Entry& entry) {
357+
if (entry.bare) {
358+
// the fn locks its own (possibly different) isolate; taking this
359+
// loop's Locker here would nest Lockers across isolates
360+
entry.fn();
361+
return;
362+
}
316363
v8::Locker locker(isolate_);
317364
v8::Isolate::Scope isolate_scope(isolate_);
318365
v8::HandleScope handleScope(isolate_);
@@ -370,6 +417,27 @@ void EventLoop::RunNestableV8Tasks() {
370417
}
371418

372419
void EventLoop::RunOrderedTask() {
420+
// one anonymous token = one due slot across the whole ordered domain:
421+
// pick the earliest due item among the ordered entries and the timer
422+
// source, whichever it is. Timers and entries only ever run on this
423+
// thread, so the peeked winner can't be taken by anyone else before we
424+
// re-lock (a concurrent post can only add later work).
425+
auto now = now_ms();
426+
double entryDue;
427+
{
428+
std::lock_guard<std::mutex> lock(mutex_);
429+
if (stopped_) {
430+
return;
431+
}
432+
entryDue = PeekDueLocked(ordered_, now);
433+
}
434+
if (timerSource_ != nullptr && timerSource_->RunIfEarliest(now, entryDue)) {
435+
return;
436+
}
437+
if (entryDue < 0) {
438+
// leftover token: nothing in the domain is due yet
439+
return;
440+
}
373441
std::unique_ptr<Entry> entry;
374442
{
375443
std::lock_guard<std::mutex> lock(mutex_);
@@ -378,11 +446,9 @@ void EventLoop::RunOrderedTask() {
378446
}
379447
entry = TakeDueLocked(ordered_, false, false, false, now_ms());
380448
}
381-
if (entry == nullptr) {
382-
// leftover token: the earliest delayed entry isn't due yet
383-
return;
449+
if (entry != nullptr) {
450+
RunEntry(*entry);
384451
}
385-
RunEntry(*entry);
386452
}
387453

388454
int EventLoop::EventFdCallback(int fd, int events, void* data) {

0 commit comments

Comments
 (0)