streams: pipeline with signal

[ronag]

Generators in pipeline must be able to be aborted or pipeline
can deadlock.

[mcollina]

lgtm

[targos]

Can you please improve the commit message title? It should start with an imperative verb.

[nodejs-github-bot]

CI: https://ci.nodejs.org/job/node-test-pull-request/38677/

[ronag]

@nodejs/streams

[benjamingr]

I'm not sure this makes sense and I'll need a few days to think about this API (Sorry!).

The thing is - generators already have a way to signal cancellation (through calling .return on them which is runs finally blocks).

This is kind of scary since it behaves differently for example from doing an addAbortSignal on a Readable.from which will call .return on the generator instead of passing a signal argument.

I think Readable.from(generator) and generator should behave relatively similarly when passed to pipeline?

Is the right thing to pass an abort signal to generators passed to from and other places that accept async iterables and abort the related controller when the stream is destroyed?

[benjamingr]

In any case I feel strongly that async generators should behave similarly in terms of AbortSignal in all of our APIs.

[ronag]

I’m fine with that but we need a way to abort calls inside a async generator.

[benjamingr]

And would that replace .return being called on the generator so we'll no longer be calling .return on generators when doing Readable.from (for instance) but instead passing in and aborting a signal?

[benjamingr]

(You can already abort it by calling .return (or .destroy after Readable.from which will run finally blocks - it will also abort sub-actions if they are generators as well and using yield* but that's quirky and users would find it surprising so I lean towards passing AbortSignal and avoiding .return altogether)

[ronag]

@benjamingr I'm unsure how .return() would help with the following case...

{
  const ac = new AbortController();
  const signal = ac.signal;
  pipelinep(
    async function * (signal) {
      await tsp.setTimeout(1e6, signal); // How to abort without passing signal?
    },
    async function(source) {

    },
    { signal }
  ).catch(common.mustCall((err) => {
    assert.strictEqual(err.name, 'AbortError');
  }));
  ac.abort();
}

[ronag]

@benjamingr have you given this PR any more thought?

[benjamingr]

Yes I think this should land if and only if we make Readable.from (and other APIs that take an async iterator) take a signal - otherwise we’re left with an inconsistent API - if you’re willing to do that (or wait a month for me I’ll happily contribute the code once I’m back from paternity leave) this sounds good to me.

[ronag]

if we make Readable.from (and other APIs that take an async iterator) take a signal - otherwise we’re left with an inconsistent API

I'm not sure what all those API's are and how you imagine it should look so I guess I'll wait and leave it to you.

[ronag]

I'm a little unsure if the signature should be (source, signal) or (source, { signal }).

[benjamingr]

The latter if we follow the DOM's API design guidelines

[ronag]

@nodejs/streams

[nodejs-github-bot]

CI: https://ci.nodejs.org/job/node-test-pull-request/39649/

[nodejs-github-bot]

CI: https://ci.nodejs.org/job/node-test-pull-request/39651/

[nodejs-github-bot]

CI: https://ci.nodejs.org/job/node-test-pull-request/39653/

[nodejs-github-bot]

CI: https://ci.nodejs.org/job/node-test-pull-request/39657/ ✅

[mcollina]

lgtm

[ronag]

Landed in c04d621