Skip to content
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

Jump to bottom

Multiple fixes for async iterators #23901

Closed
wants to merge 2 commits into from
Closed

Multiple fixes for async iterators #23901

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6f5c14f
stream: ended streams should resolve the async iteration
mcollina Oct 26, 2018
3d41222
stream: do not error async iterators on destroy(null)
mcollina Oct 26, 2018
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
7 changes: 5 additions & 2 deletions lib/internal/streams/async_iterator.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,7 +127,10 @@ const createReadableStreamAsyncIterator = (stream) => {
[kLastResolve]: { value: null, writable: true },
[kLastReject]: { value: null, writable: true },
[kError]: { value: null, writable: true },
[kEnded]: { value: false, writable: true },
[kEnded]: {
value: stream._readableState.endEmitted,
TimothyGu marked this conversation as resolved.
Show resolved Hide resolved
writable: true
},
[kLastPromise]: { value: null, writable: true },
// the function passed to new Promise
// is cached so we avoid allocating a new
Expand All @@ -150,7 +153,7 @@ const createReadableStreamAsyncIterator = (stream) => {
});

finished(stream, (err) => {
if (err) {
if (err && err.code !== 'ERR_STREAM_PREMATURE_CLOSE') {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this the only way to fix it? I would’ve expected us to perhaps do something with end-of-stream rather than selectively ignoring errors.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

end-of-stream treats non-ending (destroy) streams that close early as an error condition. I think it is right in doing so.

However this does not play well with async iterators as you pointed out. We had a different logic before that did the same but was less explicit about it: I like this one better.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

end-of-stream treats non-ending (destroy) streams that close early as an error condition. I think it is right in doing so.

Can you elaborate? It sounds to me that destroyed streams (with null as error value) should be treated no differently as any other stream that ended. I.e., it's reasonable to believe a stream can be destroyed whenever.

In fact, I would say that the following test case added in #23785 should in fact finish gracefully with an { done: true, value: undefined } rather than throwing an error:

node/test/parallel/test-stream-readable-async-iterators.js

Lines 328 to 343 in 482b56a

await (async function() {
console.log('.next() on destroyed stream');
const readable = new Readable({
read() {
// no-op
}
});
readable.destroy();
try {
await readable[Symbol.asyncIterator]().next();
} catch (e) {
assert.strictEqual(e.code, 'ERR_STREAM_PREMATURE_CLOSE');
}
})();

We see precedent for invalidated iterator still operating gracefully in ECMA-262 iterators:

const array = [0, 1, 2, 3];
const iterated = [];
for (const a of array) {
  iterated.push(a);
  if (a === 1) {
    array.splice(0);
  }
  // loop ends immediately.
}
// iterated is [0, 1]

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

First, end-of-stream is one of the most downloaded utilities on npm, and keeping the API and behavior similar has been a goal. Second, from a consumer perspective a stream that was ended or destroyed is significantly different. That could have easily been a flag as an argument. I think it errors because in the context of pipeline/pump, in fact it is an error: the pipeline did not complete correctly but was interrupted (as an example, gzipping a file).

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I’ll check again that test asap, it does not sound correct with this change.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Okay, as long as async iteration isn't affected it works for me.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@TimothyGu the test was botched anyway, and the catch clause was never executed in this case. I've fixed it.

const reject = iterator[kLastReject];
// reject if we are waiting for data in the Promise
// returned by next() and store the error
Expand Down
41 changes: 36 additions & 5 deletions test/parallel/test-stream-readable-async-iterators.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -335,11 +335,8 @@ async function tests() {

readable.destroy();

try {
await readable[Symbol.asyncIterator]().next();
} catch (e) {
assert.strictEqual(e.code, 'ERR_STREAM_PREMATURE_CLOSE');
}
const { done } = await readable[Symbol.asyncIterator]().next();
assert.strictEqual(done, true);
})();

await (async function() {
Expand All @@ -362,6 +359,40 @@ async function tests() {
assert.strictEqual(e, err);
}
})();

await (async () => {
console.log('iterating on an ended stream completes');
const r = new Readable({
objectMode: true,
read() {
this.push('asdf');
this.push('hehe');
this.push(null);
}
});
// eslint-disable-next-line no-unused-vars
for await (const a of r) {
}
// eslint-disable-next-line no-unused-vars
for await (const b of r) {
}
})();

await (async () => {
console.log('destroy mid-stream does not error');
const r = new Readable({
objectMode: true,
read() {
this.push('asdf');
this.push('hehe');
}
});

// eslint-disable-next-line no-unused-vars
for await (const a of r) {
r.destroy(null);
}
})();
}

// to avoid missing some tests if a promise does not resolve
Expand Down