Skip to content

Commit 47333f4

Browse files
committed
feat(runtime): one pump primitive that drains due JS timers
EventLoop::PumpUntil replaces the three hand-rolled module pump loops: settled-check first (a disk graph pays nothing), execution-termination and shutdown exits, deadline, nestable-task drain, microtask checkpoint, then a bounded drain of DUE ordered-lane entries - JS timers included - before idling on the loop's own fds. A pump can finally settle an entry that awaits setTimeout: the ordered lane's work lives in native bookkeeping and its Java Handler messages are only wakeups, so the drain runs due items through the same earliest-due-across-domain selection the token path uses, and the orphaned tokens retire against the claim gate exactly as a cancelled timer's do. An 8ms slice bound keeps a setTimeout(0) chain from pinning the pump past its deadline checks. WaitForInternalWork blocks properly instead of spinning: each posted entry records whether it issued an eventfd unit, direct drains count the units they orphan, and the wait swallows exactly those before polling. Its due-work early-return applies the same nestable/v8 filter the pump's drain does - a due entry the drain cannot take (a non-nestable task or plain post) must not turn the wait into a spin, and when such an entry pins the fds readable the wait falls back to a plain sleep until the looper can service it. A drained delayed entry rearms the timerfd. Timers::RunIfEarliest saves and restores the ambient nesting level instead of resetting it: nested dispatch under a pump would have handed the outer callback's remaining setTimeout calls a nesting level of zero.
1 parent c167db4 commit 47333f4

4 files changed

Lines changed: 226 additions & 36 deletions

File tree

‎docs/ns-builtin-modules.md‎

Lines changed: 16 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -779,13 +779,22 @@ graph walk, and — doubled, at 120 seconds — the boot backstop, so the waits
779779
stay ordered: transport timeouts < module deadline < boot backstop. The local
780780
entry's short yield is deliberately *not* derived from that constant: it is an
781781
independent one-second literal in `BootEntryEvaluationOptions`, with
782-
`return-pending` behavior and no looper slicing. An HTTP entry instead gets the
783-
full deadline, throws on expiry, and does slice the looper, because the tooling
784-
driving it needs the rejection reason synchronously. The backstop itself is
785-
`HoldBootBackstop` in
786-
`Runtime.cpp`, called from both `Runtime::RunModule` overloads; it pumps
787-
nestable V8 tasks, microtask checkpoints and `ALooper_pollOnce` until the entry
788-
and all async graph work settle.
782+
`return-pending` behavior. An HTTP entry instead gets the full deadline and
783+
throws on expiry, because the tooling driving it needs the rejection reason
784+
synchronously. The backstop itself is `HoldBootBackstop` in `Runtime.cpp`,
785+
called from both `Runtime::RunModule` overloads; it pumps the isolate's event
786+
loop in place (`EventLoop::PumpUntil`) until the entry and all async graph
787+
work settle.
788+
789+
Every pump on Android — a pumping require, the graph walk, the boot backstop —
790+
runs the same `EventLoop::PumpUntil` slice: nestable V8 tasks, a microtask
791+
checkpoint, and the loop's own **ordered lane drained directly** (JS timers
792+
ride Java `Handler` messages, which cannot dispatch while the pump's JS frames
793+
hold the thread — the drain is what lets an entry or a pumped graph parked on
794+
`setTimeout` settle in-pump). The pump never re-enters the platform looper, so
795+
on Android `pumpRunLoop` is validated and carried but adds nothing beyond that
796+
baseline; the option's cross-platform meaning and its warnings above are
797+
unchanged.
789798
790799
Workers copy the loader vocabulary from the parent at spawn
791800
(`CaptureLoaderVocabulary` on the parent's thread, `InstallLoaderVocabulary`

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

Lines changed: 159 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88

99
#include <algorithm>
1010
#include <cerrno>
11+
#include <chrono>
1112
#include <cmath>
1213
#include <cstring>
1314
#include <ctime>
@@ -133,9 +134,10 @@ void EventLoop::BindToCurrentThread() {
133134

134135
// flush work buffered before the home thread was known
135136
auto now = now_ms();
136-
for (size_t i = 0; i < internal_.immediate.size(); i++) {
137+
for (auto& entry : internal_.immediate) {
137138
uint64_t value = 1;
138139
write(eventFd_, &value, sizeof(value));
140+
entry.unitIssued = true;
139141
}
140142
ArmTimerLocked(now);
141143
for (auto& entry : ordered_.immediate) {
@@ -204,6 +206,7 @@ void EventLoop::PostInternalLocked(Entry entry, double delayMs) {
204206
auto now = now_ms();
205207
if (delayMs <= 0) {
206208
entry.time = now;
209+
entry.unitIssued = eventFd_ != -1;
207210
internal_.immediate.push_back(std::move(entry));
208211
if (eventFd_ != -1) {
209212
uint64_t value = 1;
@@ -451,6 +454,28 @@ double EventLoop::PeekDueLocked(Lane& lane, double now) {
451454
return due;
452455
}
453456

457+
double EventLoop::PeekDueFilteredLocked(Lane& lane, bool nestableOnly, bool v8Only, double now) {
458+
auto matches = [&](const Entry& e) {
459+
return (!nestableOnly || e.nestable) && (!v8Only || e.task != nullptr);
460+
};
461+
double due = -1;
462+
for (const auto& e : lane.immediate) {
463+
if (matches(e)) {
464+
due = e.time;
465+
break;
466+
}
467+
}
468+
for (const auto& pair : lane.delayed) {
469+
if (pair.first > now) {
470+
break;
471+
}
472+
if (matches(pair.second) && (due < 0 || pair.first < due)) {
473+
due = pair.first;
474+
}
475+
}
476+
return due;
477+
}
478+
454479
void EventLoop::ArmTimerLocked(double now) {
455480
if (timerFd_ == -1) {
456481
return;
@@ -503,11 +528,15 @@ void EventLoop::RunOneInternal() {
503528
return;
504529
}
505530
entry = TakeDueLocked(internal_, false, false, true, now_ms());
506-
}
507-
if (entry == nullptr) {
508-
// leftover unit: the work it represented ran early from a nested loop
509-
// drain
510-
return;
531+
if (entry == nullptr) {
532+
// leftover unit: the work it represented ran early from a direct
533+
// drain - this dispatch just consumed it, so it is no longer
534+
// WaitForInternalWork's to swallow
535+
if (leftoverUnits_ > 0) {
536+
leftoverUnits_--;
537+
}
538+
return;
539+
}
511540
}
512541
RunEntry(*entry);
513542
}
@@ -527,49 +556,121 @@ void EventLoop::RunNestableV8Tasks() {
527556
if (stopped_) {
528557
return;
529558
}
559+
const size_t delayedBefore = internal_.delayed.size();
530560
entry = TakeDueLocked(internal_, true, true, false, now_ms());
531-
}
532-
if (entry == nullptr) {
533-
return;
561+
if (entry == nullptr) {
562+
return;
563+
}
564+
if (entry->unitIssued) {
565+
leftoverUnits_++;
566+
}
567+
if (internal_.delayed.size() != delayedBefore) {
568+
// a drained delayed entry may leave the timerfd armed (or
569+
// expired unread) for it; rearming to the queue's new
570+
// earliest also discards the stale expiration
571+
ArmTimerLocked(now_ms());
572+
}
534573
}
535574
// the pause loops call this from inside v8 inspector frames - a C++
536575
// exception must not unwind through them
537576
RunGuarded([&] { RunEntry(*entry); });
538577
}
539578
}
540579

541-
void EventLoop::RunOrderedTask() {
542-
// one anonymous token = one due slot across the whole ordered domain:
543-
// pick the earliest due item among the ordered entries and the timer
544-
// source, whichever it is. Timers and entries only ever run on this
545-
// thread, so the peeked winner can't be taken by anyone else before we
546-
// re-lock (a concurrent post can only add later work).
580+
bool EventLoop::RunOneOrderedDue() {
581+
// one due slot across the whole ordered domain: pick the earliest due
582+
// item among the ordered entries and the timer source, whichever it is.
583+
// Timers and entries only ever run on this thread, so the peeked winner
584+
// can't be taken by anyone else before we re-lock (a concurrent post can
585+
// only add later work).
547586
auto now = now_ms();
548587
double entryDue;
549588
{
550589
std::lock_guard<std::mutex> lock(mutex_);
551590
if (stopped_) {
552-
return;
591+
return false;
553592
}
554593
entryDue = PeekDueLocked(ordered_, now);
555594
}
556595
if (timerSource_ != nullptr && timerSource_->RunIfEarliest(now, entryDue)) {
557-
return;
596+
return true;
558597
}
559598
if (entryDue < 0) {
560-
// leftover token: nothing in the domain is due yet
561-
return;
599+
// leftover token, or an idle drain: nothing in the domain is due yet
600+
return false;
562601
}
563602
std::unique_ptr<Entry> entry;
564603
{
565604
std::lock_guard<std::mutex> lock(mutex_);
566605
if (stopped_) {
567-
return;
606+
return false;
568607
}
569608
entry = TakeDueLocked(ordered_, false, false, false, now_ms());
570609
}
571-
if (entry != nullptr) {
572-
RunEntry(*entry);
610+
if (entry == nullptr) {
611+
return false;
612+
}
613+
RunEntry(*entry);
614+
return true;
615+
}
616+
617+
void EventLoop::RunOrderedTask() {
618+
// one anonymous token = one due slot; a token whose item a pump drained
619+
// early finds nothing due and dies here
620+
RunOneOrderedDue();
621+
}
622+
623+
int EventLoop::RunDueOrderedEntries() {
624+
// Bounded slice: a callback that keeps minting due-now work (a
625+
// setTimeout(0) chain) must not pin the calling pump past its own
626+
// deadline checks, so the drain yields after a few milliseconds and the
627+
// pump comes back for the rest on its next iteration.
628+
constexpr double kSliceMs = 8.0;
629+
const double start = now_ms();
630+
int ran = 0;
631+
while (RunOneOrderedDue()) {
632+
ran++;
633+
if (now_ms() - start >= kSliceMs) {
634+
break;
635+
}
636+
}
637+
return ran;
638+
}
639+
640+
EventLoop::PumpResult EventLoop::PumpUntil(double deadlineSeconds,
641+
const std::function<bool()>& settled) {
642+
const auto deadline = std::chrono::steady_clock::now() +
643+
std::chrono::duration<double>(deadlineSeconds);
644+
for (;;) {
645+
if (settled()) {
646+
return PumpResult::kSettled;
647+
}
648+
if (isolate_ != nullptr && isolate_->IsExecutionTerminating()) {
649+
return PumpResult::kTerminated;
650+
}
651+
if (IsStopped()) {
652+
// a stopped loop drops every post, so nothing can settle anymore
653+
return PumpResult::kTerminated;
654+
}
655+
if (std::chrono::steady_clock::now() >= deadline) {
656+
return PumpResult::kDeadline;
657+
}
658+
RunNestableV8Tasks();
659+
{
660+
// work may enqueue microtasks without entering JS; scopes are
661+
// re-entrant, so callers already holding them pay nothing
662+
v8::Locker locker(isolate_);
663+
v8::Isolate::Scope isolateScope(isolate_);
664+
v8::HandleScope handleScope(isolate_);
665+
isolate_->PerformMicrotaskCheckpoint();
666+
}
667+
const int ranOrdered = RunDueOrderedEntries();
668+
if (settled()) {
669+
return PumpResult::kSettled;
670+
}
671+
if (ranOrdered == 0) {
672+
WaitForInternalWork(10);
673+
}
573674
}
574675
}
575676

@@ -589,12 +690,44 @@ bool EventLoop::IsInLooperCallback() { return t_looperCallbackDepth > 0; }
589690
void EventLoop::WaitForInternalWork(int timeoutMs) {
590691
struct pollfd fds[2];
591692
nfds_t count = 0;
693+
bool sleepOnly = false;
592694
{
593695
std::lock_guard<std::mutex> lock(mutex_);
594-
if (eventFd_ != -1) fds[count++] = {eventFd_, POLLIN, 0};
595-
if (timerFd_ != -1) fds[count++] = {timerFd_, POLLIN, 0};
696+
if (stopped_) {
697+
sleepOnly = true;
698+
} else {
699+
const double now = now_ms();
700+
// Drainable work already due: the caller's drain runs it, waiting
701+
// would only add latency. The filter must match RunNestableV8Tasks
702+
// (nestable v8 tasks only) — a due entry the pump cannot take must
703+
// not turn the wait into a no-op.
704+
if (PeekDueFilteredLocked(internal_, /*nestableOnly=*/true, /*v8Only=*/true, now) >=
705+
0) {
706+
return;
707+
}
708+
// units whose entries a direct drain already consumed keep the
709+
// eventfd readable; swallow them or the poll below returns
710+
// immediately on every call
711+
while (leftoverUnits_ > 0 && eventFd_ != -1) {
712+
uint64_t value;
713+
if (read(eventFd_, &value, sizeof(value)) != sizeof(value)) {
714+
break;
715+
}
716+
leftoverUnits_--;
717+
}
718+
// A due entry the drain cannot take (non-nestable task, plain fn
719+
// post) pins its unread unit in the eventfd, so the fds cannot go
720+
// quiet — polling them would spin. Plain sleep is the only honest
721+
// wait until the looper resumes and runs it.
722+
if (PeekDueLocked(internal_, now) >= 0) {
723+
sleepOnly = true;
724+
} else {
725+
if (eventFd_ != -1) fds[count++] = {eventFd_, POLLIN, 0};
726+
if (timerFd_ != -1) fds[count++] = {timerFd_, POLLIN, 0};
727+
}
728+
}
596729
}
597-
if (count == 0) {
730+
if (sleepOnly || count == 0) {
598731
usleep(static_cast<useconds_t>(timeoutMs) * 1000);
599732
return;
600733
}
@@ -635,6 +768,7 @@ int EventLoop::TimerFdCallback(int fd, int events, void* data) {
635768
}
636769
if (!pair.second.signaled) {
637770
pair.second.signaled = true;
771+
pair.second.unitIssued = true;
638772
due++;
639773
}
640774
}

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

Lines changed: 46 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -197,12 +197,41 @@ class EventLoop {
197197
/**
198198
* Blocks the calling thread until this loop's internal lane has work (the
199199
* eventfd or timerfd is readable) or `timeoutMs` elapses, whichever comes
200-
* first, without consuming either fd and without entering the looper - so
201-
* it is safe where IsInLooperCallback forbids polling. Pumps pair it with
202-
* RunNestableV8Tasks, which drains the queue directly.
200+
* first, without entering the looper - so it is safe where
201+
* IsInLooperCallback forbids polling. Returns immediately when internal
202+
* work is already due. Eventfd units left over from entries a direct
203+
* drain (RunNestableV8Tasks) consumed are swallowed first, so the wait
204+
* only wakes for new work instead of spinning on stale readability.
203205
*/
204206
void WaitForInternalWork(int timeoutMs);
205207

208+
/**
209+
* Runs every ordered-lane item that is due NOW - Java-token entries and
210+
* timer-source items alike, in due order - directly from the calling
211+
* (home) thread, without waiting for their Handler messages. The messages
212+
* still arrive later and die as leftover tokens: a token whose item was
213+
* drained early finds nothing due and no-ops, and its claim cell is
214+
* retired by the dispatch gate as usual. Bounded to a short slice so a
215+
* callback minting due-now work (a setTimeout(0) chain) cannot pin the
216+
* caller past its own deadline checks. Returns the number of items run.
217+
*/
218+
int RunDueOrderedEntries();
219+
220+
enum class PumpResult { kSettled, kDeadline, kTerminated };
221+
222+
/**
223+
* Drives this loop in place on the home thread until `settled()` returns
224+
* true or `deadlineSeconds` elapses: nestable v8 tasks, a microtask
225+
* checkpoint, and due ordered-lane work per iteration, idling in
226+
* WaitForInternalWork between slices. The one pump primitive behind
227+
* module evaluation, the graph walk, and the boot backstop - JS frames
228+
* are on the stack throughout, so non-nestable v8 tasks and plain
229+
* internal posts stay queued, exactly as in the inspector pause loops.
230+
* Returns kTerminated when the isolate is terminating or the loop has
231+
* been shut down; `settled` may throw and the exception propagates.
232+
*/
233+
PumpResult PumpUntil(double deadlineSeconds, const std::function<bool()>& settled);
234+
206235
/**
207236
* Runs at most one due ordered-lane entry, then performs a microtask
208237
* checkpoint. Invoked by Java EventLoopHandler.handleMessage once per
@@ -228,6 +257,10 @@ class EventLoop {
228257
// this entry (written when its timerfd deadline fired), so a later
229258
// timer fire must not issue a second one
230259
bool signaled = false;
260+
// internal entries only: an eventfd unit backs this entry, so a
261+
// direct (unit-free) drain that consumes it must count the unit as
262+
// leftover for WaitForInternalWork to swallow
263+
bool unitIssued = false;
231264
};
232265
struct Lane {
233266
std::deque<Entry> immediate;
@@ -248,9 +281,15 @@ class EventLoop {
248281
bool requireSignaledDelayed, double now);
249282
// earliest due entry time in the lane, or a negative value if none is due
250283
static double PeekDueLocked(Lane& lane, double now);
284+
// same, but only over entries matching TakeDueLocked's nestable/v8 filter
285+
static double PeekDueFilteredLocked(Lane& lane, bool nestableOnly, bool v8Only, double now);
251286
void ArmTimerLocked(double now);
252287
void RunEntry(Entry& entry);
253288
void RunOneInternal();
289+
// one due slot across the ordered domain (entries + timer source); true
290+
// when a slot was consumed. The body behind both RunOrderedTask (one call
291+
// per Java token) and RunDueOrderedEntries (looped by the pumps).
292+
bool RunOneOrderedDue();
254293

255294
static int EventFdCallback(int fd, int events, void* data);
256295
static int TimerFdCallback(int fd, int events, void* data);
@@ -307,6 +346,10 @@ class EventLoop {
307346
int eventFd_ = -1;
308347
int timerFd_ = -1;
309348
bool stopped_ = false;
349+
// units written to eventFd_ whose entries a direct drain already ran;
350+
// consumed by WaitForInternalWork (or by an EventFdCallback that finds
351+
// nothing due). Guarded by mutex_.
352+
uint64_t leftoverUnits_ = 0;
310353

311354
// process-wide JNI cache, written once under the first bind's lock (the
312355
// main runtime binds before any worker thread exists)

0 commit comments

Comments
 (0)