René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"pipeline deadlock with readable tail?","articleBody":"https://github.com/nodejs/node/pull/40653#discussion_r739873780\r\n\r\n@lpinca feel free to edit issue and fill out with your concern. Otherwise I will dig into this at some later point.\r\n\r\n---\r\n\r\nThe `pipeline()` callback might never be called if the destination is a `Transform` stream that is not read. Here is an example.\r\n\r\n```js\r\nconst stream = require('stream');\r\n\r\nconst chunk = Buffer.alloc(1024);\r\nlet bytesRead = 0;\r\n\r\nconst readable = new stream.Readable({\r\n  read() {\r\n    if (bytesRead === 100 * 1024) {\r\n      readable.push(null);\r\n    } else {\r\n      bytesRead += chunk.length;\r\n      readable.push(chunk);\r\n    }\r\n  }\r\n});\r\n\r\nconst passThrough = new stream.PassThrough();\r\n\r\nstream.pipeline(readable, passThrough, function (err) {\r\n  if (err) {\r\n    throw err;\r\n  }\r\n\r\n  console.log('done');\r\n});\r\n```\r\n\r\nIn the callback version this is not a big issue because `pipeline()` returns the last stream, so the user could resume the returned stream.\r\n\r\n```js\r\npipeline(readable, passThrough, fn).resume();\r\n```\r\n\r\nIn the promisified variant, `pipeline()` is just a wrapper that returns a promise that is fulfilled when the callback of the callback variant is called.\r\n\r\nHowever the user could not be able to resume the returned stream because they might not have direct control on it. For example if the last entry is a generator.\r\n\r\n```js\r\nconst stream = require('stream');\r\nconst streamPromises = require('stream/promises');\r\n\r\nconst chunk = Buffer.alloc(1024);\r\nlet bytesRead = 0;\r\n\r\nconst readable = new stream.Readable({\r\n  read() {\r\n    if (bytesRead === 100 * 1024) {\r\n      readable.push(null);\r\n    } else {\r\n      bytesRead += chunk.length;\r\n      readable.push(chunk);\r\n    }\r\n  }\r\n});\r\n\r\nasync function* passThrough(source) {\r\n  for await (const chunk of source) {\r\n    yield chunk;\r\n  }\r\n}\r\n\r\nasync function run() {\r\n  await streamPromises.pipeline(readable, passThrough);\r\n}\r\n\r\nrun()\r\n  .then(function () {\r\n    console.log('done');\r\n  })\r\n  .catch(console.error);\r\n```\r\n\r\nIn this case `stream.pipeline()` returns an internal `PassThrough` proxy that the user cannot resume so the promise returned by `streamPromises.pipeline()` might never be fulfilled.\r\n\r\nA possible workaround is to resume the returned stream in the promisified variant.\r\n\r\n```diff\r\ndiff --git a/lib/stream/promises.js b/lib/stream/promises.js\r\nindex 0db01a8b20..5bbadd43c5 100644\r\n--- a/lib/stream/promises.js\r\n+++ b/lib/stream/promises.js\r\n@@ -23,13 +23,17 @@ function pipeline(...streams) {\r\n       signal = options.signal;\r\n     }\r\n \r\n-    pl(streams, (err, value) => {\r\n+    const stream = pl(streams, (err, value) => {\r\n       if (err) {\r\n         reject(err);\r\n       } else {\r\n         resolve(value);\r\n       }\r\n     }, { signal });\r\n+\r\n+    if (stream.readable) {\r\n+      stream.resume();\r\n+    }\r\n   });\r\n }\r\n```","author":{"url":"https://github.com/ronag","@type":"Person","name":"ronag"},"datePublished":"2021-11-01T08:43:39Z","url":"https://github.com/nodejs/node/issues/40685"}