At what stage is Transferable Streams implementation for worker threads at?

[crozzfire]

Hey folks!
We have a usecase where Transferable Streams will help us streaming data from worker threads back to main thread. I did find some discussions around this for browser implementation (Here and Here), but none for NodeJS workers.

If none exist, what would be the process to file a official feature request ticket in this community? Thank you!

[avivkeller]

I did find some discussions around this for browser implementation (Here and Here), but none for NodeJS workers.

@nodejs/workers

---

If none exists, what would be the process to file a official feature request ticket in this community? Thank you!

I've CC'd the team associated with worker threads. As for submitting a feature request, when choosing to open an issue in this repository, one of the options is to open a feature request.

[benjamingr]

Initially chunks enqueued in such a stream will always be cloned, ie. all data will be copied. Future work will extend the Streams APIs to support transferring objects (ie. zero copy).

I'm not sure I see the point in this, It's pretty easy to add by just having one stream postMessage the data to another stream in the worker. Though Node does support a variation of this for its own streams in workers (stdout/stderr/stdin) so that's an indication usages exist.

[crozzfire]

It's pretty easy to add by just having one stream postMessage the data to another stream in the worker.

@benjamingr Can you please help me understand this better? Here's our exact usecase:
Our workers perform certain action and one of them being downloading large files. What we want to achieve is to stream the fetch Response data from our worker to the main thread (where the action handlers exist). While doing so, we do not want to read the entire stream in the worker, derive its UInt8 array and then postMessage it in the main thread.
What we want to achieve is virtually piping the stream over postMessage, such that the memory footprint isn't duplicated across V8 isolates and the main thread, from where it is going to be pushed to another service over the wire. From what I have experimented with, streams can't be cloned with structuredClone() and passing it over postMessage always gives me a undefined at the receiver end.

That said, surely I'm missing something here and looking forward to learning from you. At the end of the day, we want to optimize on memory usages since we deal with large files.

[benjamingr]

What we want to achieve is to stream the fetch Response data from our worker to the main thread (where the action handlers exist). While doing so, we do not want to read the entire stream in the worker, derive its UInt8 array and then postMessage it in the main thread.

The scheme is like this:

• Main thread has a Readable stream, worker thread has a readable stream
• Both streams are in non-flowing mode (pull and not push, "readable" and not "data")
• When the worker thread needs data, it .reads on the stream (.e.g with a for await loop, or literal .read inside "readable"), it acts like a regular readable stream
• That calls read on the stream implementation, you "postMessage" a "read" message to the parent worker
• The parent worker calls read on the parent stream (the one with all your data), you postMessages the read data back to the worker.
• You postMessage stream events (e.g. error) you receive on the parent stream on the child stream.

That lets you read all the data while maintaining backpressure support and reading as much as needed. Reading my message back, I kind of see your point and why @mcollina thinks it may be worth implementing in core, this isn't as trivial as it first seemed in my head.

(This stream is for Node streams but it's identical in principle in web streams)

[crozzfire]

@benjamingr Ah yes I see where you were coming from initially 👍 If I understood you correctly, every "read" event in parent (as an ex) will postMessage the corresponding read event with the chunk data in the thread. My question in that approach is sequencing, since messaging over postMessage aren't guaranteed to be in order at the receiver end. And that way, the read events might be read out-of-order at the consumer side. The node implementation might differ here and perhaps Node does it over a single queue that also guarantees ordering? Otherwise, we didn't want to put a sequencer logic in place and that bloats up the complexity on our side.

I would love to know more about this internal detail / or if you can link the entry point to the code where this sort of event dispatching happens in Node :)

[benjamingr]

If none exist, what would be the process to file a official feature request ticket in this community? Thank you!

Also, you can open a feature request in https://github.com/nodejs/node/issues/new?assignees=&labels=feature+request&projects=&template=2-feature-request.yml

[mcollina]

I plan to start working on a "pass through" pair that can transfer data through quickly.

[crozzfire]

Yeah what we are essentially looking is for a way to do something like the following over worker thread / main thread boundary:

  // Request to the sourceEndpoint
  const response = await fetch(sourceUrl, sourceUrlOpts);
  
   // Pipe the body to the destinationEndpoint and don't hold a copy
  const destinationOpts = {
    body: response.body, // Directly relay/pass the ReadableStream to the destination
    method: "POST",
    duplex: "half"
  };
  return fetch(destinationUrl, destinationOpts);

Taking the above code in perspective, because of how our action executors are setup, the first (source) fetch call happens in a worker thread and the intention is to pipe response.body over to the main thread, where it will be pushed to the sink/destination.

This might be a little off topic, but whatwg has an open issue whatwg/fetch#1254 where fetch won't resolve until the the incoming stream is closed, essentially voiding the full duplex capability.
This might not be the problem in Node, depending on it's implementation. But I at least wanted to clarify the usecase.

[mcollina]

That already works. Web streams are transferable.