stream: fix `pipeline(readable, Duplex.from(<async-iterator>), cb)` not triggering data or calling the cb #45534

rluvaton · 2022-11-20T01:33:53Z

Fix #45533

I think this should be ported to the rest of node versions...

Fix nodejs#45533

nodejs-github-bot · 2022-11-20T01:33:57Z

Review requested:

@nodejs/streams

rluvaton · 2022-11-20T01:35:18Z

lib/internal/streams/from.js

+  readable.on('pipe', readable.resume.bind(readable));
+  readable.on('unpipe', readable.pause.bind(readable));


Spent over 5 hours just for these 2 lines 😬

I don't think this is the correct fix for this issue. resume and pause should be automatically handled my the machinery during pipe and unpipe.

Why do you think so?

@ronag wdyt?

Yea, this is wrong.

rluvaton · 2022-11-20T01:40:12Z

lib/internal/streams/from.js

+  readable.on('pipe', readable.resume.bind(readable));
+  readable.on('unpipe', readable.pause.bind(readable));


on destroy are those listeners getting cleaned?

No, I don't think so.

ronag

This is incorrect. I'll have a deeper look at this issue next week.

rluvaton · 2022-11-20T08:10:59Z

This is incorrect. I'll have a deeper look at this issue next week.

@ronag Maybe I can try to fix it, I want to contribute.

Can you explain why it's incorrect? And maybe provide a hint to where I should focus

Because it's extending from Readable than it shouldn't have this?

The problem was that no one triggered the data to start for some reason

aduh95 · 2022-11-20T08:32:40Z

test/parallel/test-stream-duplex-from.js

+    assert.strictEqual(chunk, data[i++]);
+  }, 4);


Suggested change

assert.strictEqual(chunk, data[i++]);

}, 4);

assert.strictEqual(chunk, data.shift());

}, data.length);

ronag · 2022-11-20T08:54:50Z

This is incorrect. I'll have a deeper look at this issue next week.

@ronag Maybe I can try to fix it, I want to contribute.

Can you explain why it's incorrect? And maybe provide a hint to where I should focus

Because it's extending from Readable than it shouldn't have this?

The problem was that no one triggered the data to start for some reason

I'll have a look and try to explain it to you.

readable.pipe(Duplex.from(<async-iterator>)) should not trigger data

ronag · 2022-11-20T12:34:21Z

test/parallel/test-stream-duplex-from.js

+      yield chunk;
+    }
+  }));
+}


I don't understand this test?

to avoid making my mistake of starting the data flow before on data event listener is attached

The flow should start here, the callback should be called.

that's right, this is why cb = common.mustNotCall();

I don't understand. You are testing the opposite of what we want?

no, we did not attach on data listener so the stream should not start...

#45533 (comment)

the data listener should not be relevant here. You should not have to worry about that.

pipe should start the stream regardless.

that's what I thought at first, @mcollina WDYT?

test/parallel/test-stream-pipeline-duplex.js

ronag · 2022-11-20T12:51:32Z

Does any of these test fail without:

  readable.on('pipe', readable.resume.bind(readable));
  readable.on('unpipe', readable.pause.bind(readable));

I think the best way here would be if you could help to contribute with failing tests and then we can try to fix.

rluvaton · 2022-11-20T12:55:06Z

yes, they fail, otherwise I won't add them, I first write the tests and then fix the problem

for the pipeline duplex test:

Mismatched <anonymous> function calls. Expected exactly 1, actual 0.
    at mustCall (/Users/rluvaton/dev/open-source/node/node-fork/test/common/index.js:406:10)
    at Proxy.mustSucceed (/Users/rluvaton/dev/open-source/node/node-fork/test/common/index.js:410:10)
    at Object.<anonymous> (/Users/rluvaton/dev/open-source/node/node-fork/test/parallel/test-stream-pipeline-duplex.js:47:12)
    at Module._compile (node:internal/modules/cjs/loader:1213:14)
    at Module._extensions..js (node:internal/modules/cjs/loader:1267:10)
    at Module.load (node:internal/modules/cjs/loader:1076:32)
    at Module._load (node:internal/modules/cjs/loader:917:12)
    at Function.executeUserEntryPoint [as runMain] (node:internal/modules/run_main:82:12)
    at node:internal/main/run_main_module:23:47

For the pipe test:

Mismatched <anonymous> function calls. Expected exactly 4, actual 0.
    at Proxy.mustCall (/Users/rluvaton/dev/open-source/node/node-fork/test/common/index.js:406:10)
    at Object.<anonymous> (/Users/rluvaton/dev/open-source/node/node-fork/test/parallel/test-stream-duplex-from.js:308:21)
    at Module._compile (node:internal/modules/cjs/loader:1213:14)
    at Module._extensions..js (node:internal/modules/cjs/loader:1267:10)
    at Module.load (node:internal/modules/cjs/loader:1076:32)
    at Module._load (node:internal/modules/cjs/loader:917:12)
    at Function.executeUserEntryPoint [as runMain] (node:internal/modules/run_main:82:12)
    at node:internal/main/run_main_module:23:47

the problem is that nothing triggers the data to start flowing

rluvaton · 2022-11-20T13:39:43Z

test/parallel/test-stream-duplex-from.js

+  const cb = common.mustCall((chunk) => {
+    assert.strictEqual(chunk, data.shift());
+  }, data.length);


Actually, this should be called only once, right?

Because the highWaterMark for the duplex stream that created from async generator function is 1:

node/lib/internal/streams/from.js

Line 41 in 61805fb

highWaterMark: 1,

And according to the docs:

The readable.pipe() method attaches a Writable stream to the readable, causing it to switch automatically into flowing mode and push all of its data to the attached Writable. The flow of data will be automatically managed so that the destination Writable stream is not overwhelmed by a faster Readable stream.

readable.pipe(destination[, options])

and because no data callback exists here we reached the max buffer size.

WDYT @ronag?

ronag · 2022-11-20T16:36:59Z

So the first case:

 pipeline(
  Readable.from(['a', 'b', 'c', 'd']),
  Duplex.from(async function*(stream) {
    console.error("###start")
    for await (const chunk of stream) {
      yield chunk;
    }
  }),
  common.mustCall(),
).on('error', common.mustNotCall());

Where the last stream is still readable will return a readable stream. So the correct test looks like this:

const readable = pipeline(
  Readable.from(['a', 'b', 'c', 'd']),
  Duplex.from(async function*(stream) {
    console.error("###start")
    for await (const chunk of stream) {
      yield chunk;
    }
  }),
  common.mustCall(),
).on('error', common.mustNotCall());
readable.resume();

Which works as expected. Maybe improved docs would be appropriate?

ronag · 2022-11-20T16:38:59Z

The other test has the same problem. I don't see an issue here.

rluvaton · 2022-11-20T17:43:25Z

Shouldn't:

 pipeline(
  Readable.from(['a', 'b', 'c', 'd']),
  Duplex.from(async function*(stream) {
    console.error("###start")
    for await (const chunk of stream) {
      yield chunk;
    }
  }),
  common.mustCall(),
).on('error', common.mustNotCall());

should behave the same as:

 pipeline(
  Readable.from(['a', 'b', 'c', 'd']),
  async function*(stream) {
    console.error("###start")
    for await (const chunk of stream) {
      yield chunk;
    }
  },
  common.mustCall(),
).on('error', common.mustNotCall());

ronag · 2022-11-20T18:27:46Z

Not necessarily. In the case of Duplex.from some buffering is introduced.

ronag · 2022-11-20T18:38:42Z

If you make Readable.from(['a', 'b', 'c', 'd']), contain more elements (i.e. more than watermark) they will behave the same.

rluvaton · 2022-11-20T23:35:17Z

Ok, before I close this PR (😢 ) just wanna make sure

You said that:

pipe should start the stream regardless.
#45534 (comment)

so why would I need to call readable.resume();?

If you make Readable.from(['a', 'b', 'c', 'd']), contain more elements (i.e. more than watermark) they will behave the same.

I assume you mean that you will need to call resume but from what I remembered when I worked with streams and pipeline to manipulate files with 1-7 GB in size I never had to do that at all (at the time it was node 14 I think)

ronag · 2022-11-21T07:11:52Z

pipe will start the stream if/when the destination wants to receive data (i.e. !writableNeedDrain). Also Readable.pipe and stream.pipeline are different things.

rluvaton · 2022-11-21T07:33:16Z

Also Readable.pipe and stream.pipeline are different things.

I know but pipeline use pipe under the hood

thanks anyway!

stream: fix readable.pipe(Duplex.from(<async-iterator>)) not triggering

61805fb

Fix nodejs#45533

nodejs-github-bot added the needs-ci PRs that need a full CI run. label Nov 20, 2022

rluvaton changed the title ~~stream: fix readable.pipe(Duplex.from(<async-iterator>)) not triggering~~ stream: fix readable.pipe(Duplex.from(<async-iterator>)) not triggering data Nov 20, 2022

rluvaton commented Nov 20, 2022

View reviewed changes

ronag requested changes Nov 20, 2022

View reviewed changes

aduh95 reviewed Nov 20, 2022

View reviewed changes

rluvaton changed the title ~~stream: fix readable.pipe(Duplex.from(<async-iterator>)) not triggering data~~ stream: fix pipeline(readable, Duplex.from(<async-iterator>), cb) not triggering data or calling the cb Nov 20, 2022

rluvaton mentioned this pull request Nov 20, 2022

pipeline and duplex having multiple error as well as not calling the pipeline function on some cases #45522

Open

rluvaton added 2 commits November 20, 2022 14:20

stream: update test

69ea7a9

readable.pipe(Duplex.from(<async-iterator>)) should not trigger data

stream: update test

ca7853a

ronag reviewed Nov 20, 2022

View reviewed changes

test/parallel/test-stream-pipeline-duplex.js Show resolved Hide resolved

rluvaton added 2 commits November 20, 2022 14:37

stream: wip

1eba009

stream: wip

5988cb9

rluvaton commented Nov 20, 2022

View reviewed changes

rluvaton closed this Nov 21, 2022

rluvaton deleted the fix-45533 branch November 21, 2022 07:33

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

stream: fix `pipeline(readable, Duplex.from(<async-iterator>), cb)` not triggering data or calling the cb #45534

stream: fix `pipeline(readable, Duplex.from(<async-iterator>), cb)` not triggering data or calling the cb #45534

rluvaton commented Nov 20, 2022 •

edited

Loading

nodejs-github-bot commented Nov 20, 2022

rluvaton Nov 20, 2022

mcollina Nov 20, 2022

ronag Nov 20, 2022

rluvaton Nov 20, 2022

mcollina Nov 20, 2022

ronag left a comment

rluvaton commented Nov 20, 2022 •

edited

Loading

aduh95 Nov 20, 2022

ronag commented Nov 20, 2022

ronag Nov 20, 2022

rluvaton Nov 20, 2022

ronag Nov 20, 2022

rluvaton Nov 20, 2022

ronag Nov 20, 2022

rluvaton Nov 20, 2022

ronag Nov 20, 2022

ronag Nov 20, 2022 •

edited

Loading

rluvaton Nov 20, 2022

ronag commented Nov 20, 2022

rluvaton commented Nov 20, 2022 •

edited

Loading

rluvaton Nov 20, 2022

ronag commented Nov 20, 2022 •

edited

Loading

ronag commented Nov 20, 2022

rluvaton commented Nov 20, 2022 •

edited

Loading

ronag commented Nov 20, 2022

ronag commented Nov 20, 2022

rluvaton commented Nov 20, 2022

ronag commented Nov 21, 2022

rluvaton commented Nov 21, 2022

		readable.on('pipe', readable.resume.bind(readable));
		readable.on('unpipe', readable.pause.bind(readable));

stream: fix pipeline(readable, Duplex.from(<async-iterator>), cb) not triggering data or calling the cb #45534

stream: fix pipeline(readable, Duplex.from(<async-iterator>), cb) not triggering data or calling the cb #45534

Conversation

rluvaton commented Nov 20, 2022 • edited Loading

nodejs-github-bot commented Nov 20, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ronag left a comment

Choose a reason for hiding this comment

rluvaton commented Nov 20, 2022 • edited Loading

Choose a reason for hiding this comment

ronag commented Nov 20, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ronag Nov 20, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ronag commented Nov 20, 2022

rluvaton commented Nov 20, 2022 • edited Loading

Choose a reason for hiding this comment

ronag commented Nov 20, 2022 • edited Loading

ronag commented Nov 20, 2022

rluvaton commented Nov 20, 2022 • edited Loading

ronag commented Nov 20, 2022

ronag commented Nov 20, 2022

rluvaton commented Nov 20, 2022

ronag commented Nov 21, 2022

rluvaton commented Nov 21, 2022

stream: fix `pipeline(readable, Duplex.from(<async-iterator>), cb)` not triggering data or calling the cb #45534

stream: fix `pipeline(readable, Duplex.from(<async-iterator>), cb)` not triggering data or calling the cb #45534

rluvaton commented Nov 20, 2022 •

edited

Loading

rluvaton commented Nov 20, 2022 •

edited

Loading

ronag Nov 20, 2022 •

edited

Loading

rluvaton commented Nov 20, 2022 •

edited

Loading

ronag commented Nov 20, 2022 •

edited

Loading

rluvaton commented Nov 20, 2022 •

edited

Loading