Skip to content

quic: remove unused fin flag from blob reader wakeup #65315

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
trivenay wants to merge 1 commit into
base: main
Choose a base branch
from
+11 −24
Open

quic: remove unused fin flag from blob reader wakeup #65315

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 1 addition & 9 deletions lib/internal/blob.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -618,9 +618,7 @@ async function* createBlobReaderIterable(reader, options = kEmptyObject) {
const { getReadError } = options;
let wakeup = PromiseWithResolvers();
let immediate;
let fin = false;
reader.setWakeup((setfin) => {
fin ||= setfin;
reader.setWakeup(() => {
immediate ??= setImmediate(() => {
immediate = undefined;
wakeup.resolve?.();
Expand Down Expand Up @@ -672,12 +670,6 @@ async function* createBlobReaderIterable(reader, options = kEmptyObject) {
if (blocked) {
await wakeup.promise;
wakeup = PromiseWithResolvers();
// If the wakeup was triggered by FIN (EndReadable), the DataQueue
// is capped. Continue the loop to pull again -- the next pull will
// return EOS. Without this, a race between the data notification
// and the FIN notification can leave the iterator waiting for a
// wakeup that will never come.
if (fin) continue;
}
}
} finally {
Expand Down
16 changes: 6 additions & 10 deletions src/node_blob.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -420,20 +420,16 @@ void Blob::Reader::SetWakeup(const FunctionCallbackInfo<Value>& args) {
reader->wakeup_.Reset(args.GetIsolate(), args[0].As<Function>());
}

void Blob::Reader::NotifyPull(bool fin) {
void Blob::Reader::NotifyPull() {
if (wakeup_.IsEmpty() || !env()->can_call_into_js()) return;
// FIN notifications always fire — they must not be suppressed by
// pull_pending_ because there will be no further notifications to
// wake the iterator. Regular data notifications respect pull_pending_
// to coalesce multiple deliveries within a single packet.
if (!fin && pull_pending_) return;
// Coalesce notifications: if a wakeup is already pending and the reader
// has not yet pulled, skip re-notifying to avoid redundant wakeups
// within a single packet.
if (pull_pending_) return;
pull_pending_ = true;
HandleScope handle_scope(env()->isolate());
Local<Function> fn = wakeup_.Get(env()->isolate());
// Pass fin as the first argument so the JS iterator knows EOS is
// imminent and should pull again without waiting for another wakeup.
Local<Value> argv[] = {v8::Boolean::New(env()->isolate(), fin)};
MakeCallback(fn, 1, argv);
MakeCallback(fn, 0, nullptr);
}

BaseObjectPtr<BaseObject> Blob::BlobTransferData::Deserialize(
Expand Down
2 changes: 1 addition & 1 deletion src/node_blob.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,7 @@ class Blob : public BaseObject {
BaseObjectPtr<Blob> blob);
static void Pull(const v8::FunctionCallbackInfo<v8::Value>& args);
static void SetWakeup(const v8::FunctionCallbackInfo<v8::Value>& args);
void NotifyPull(bool fin = false);
void NotifyPull();

explicit Reader(Environment* env,
v8::Local<v8::Object> obj,
Expand Down
7 changes: 3 additions & 4 deletions src/quic/streams.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1614,10 +1614,9 @@ void Stream::EndReadable(std::optional<uint64_t> maybe_final_size) {
FlushAccumulation();
set_final_size(maybe_final_size.value_or(STAT_GET(Stats, bytes_received)));
inbound_->cap(STAT_GET(Stats, final_size));
// Notify the JS reader so it can see EOS. Pass fin=true so the
// wakeup promise resolves with a value the iterator can check to
// avoid waiting for another wakeup that will never come.
if (reader_) reader_->NotifyPull(true);
// Notify the JS reader so it can see EOS. The subsequent pull observes
// the now-capped DataQueue and returns EOS.
if (reader_) reader_->NotifyPull();
}

void Stream::Destroy(QuicError error) {
Expand Down