custom nodejs duplex stream with single reader and multiple writers

[pps83]

My script makes a GET http request and saves received data into a Buffer.
Then this script listens for incoming connections and when they are established sends out what's stored in the Buffer.

If it was synchronous pseudo-code it would be like this:

// single reader, all data loaded into `buf`
var buf = http.get('http://example.com/movie.mp4');
app.use('/movie.mp4', function(req, res, next){
  res.end(buf);  // multiple writers
  next();
));

Here's what I want to do now: I want to be able to serve 'multiple writers' while 'single reader' is still reading data, basically when I get 'data' event for 'single reader' I'd store it to a buffer and pipe it to all connected 'multiple writers'. While all this happens new writers could be accepted, for them I'd need to first send whatever had been buffered so far into buf and when it's finished all new 'data' events on the 'single reader' would also be piped to these new writers. All the data to be piped to 'multiple writers' is always buffered inside buf, that is I'd prefer to implement it so that there is no needless copying or buffering going on, it's always fully buffered inside buf

How can I write that in nodejs's streams?

[addaleax]

Hi @pps83, that’s a question I’ve been asking myself quite a few times, too. I have not found anything that already exists in the wild, although caching-stream comes kind of close.

So, I’ve decided to try and code something together myself: https://github.com/addaleax/playback-stream … feel free to take a look and let me know if that’s the kind of thing you’re looking for.

[pps83]

Hi @addaleax,
that looks very close what I'm trying to do. I'm curious how many concurrent connections nodejs would be able to serve and at what total bitrate. The other point: if the file is already stored in memory, all these pipes/streams: do have internal buffering (when it's actually not needed since the file is fully buffered in memory). If I do same in C++ I'm pretty sure I can easily max out available network channel or cpu. I'm thinking to try to compare this scenario with nodejs.

[addaleax]

The other point: if the file is already stored in memory, all these pipes/streams: do have internal buffering (when it's actually not needed since the file is fully buffered in memory).

If it’s any consolation, the streams here have their own control structures, but the underlying memory for the file content’s won’t be duplicated. But tbh, I’m not sure I understand… if the file is already buffered in memory, why don’t you write it directly to all receivers?

[pps83]

I'm not a nodejs expert so I'm not sure how exactly it works internally. I think I've read somewhere that writing a chunk transfers ownership or something like that, which effectively means that it should duplicate it. That's why I though it copies data internally.

In your code you store list of chunks and then use them. What about pre-allocating entire buffer based on content-length (provided you trust the upstream) and then writing incoming chunks into that whole buffer.
When I perf tested ngins serving files from ssd takes 3-4 times less cpu that node serving with your readable stream. When I use one chunk buffer, cpu goes down to approximately the same level as nginx.

I also noticed that if I serve at max speed up to 20-30 simultaneous downloads then I achieve close to max limit of my hw interface. However, if I serve up to 500 simultaneous downloads then combined total transfer rate is about 65% of max.

[addaleax]

I think I've read somewhere that writing a chunk transfers ownership or something like that, which effectively means that it should duplicate it.

Had to test it myself, but it doesn’t look like duplication takes place to me:

const a = Buffer.allocUnsafe(1024);
const p = new stream.PassThrough();

p.write(a);
console.log(a === p.read()); // true

What about pre-allocating entire buffer based on content-length (provided you trust the upstream) and then writing incoming chunks into that whole buffer.

I guess I’ll try to add that as a feature to my module? 😄

Two settings you can always try fine-tuning to your system, especially if you have a lot of memory available, are Buffer.poolSize and the highWaterMark setting of streams. Raising their values will likely reduce CPU load but increase memory requirements.

[pps83]

Regarding using single buffer: if nodejs doesn't clone buffers, then there is another point: does it copy memory anywhere at all or it passes raw memory from written chunks to OS api without copying it (which should be the case). Basically, when you stream.write(chunk) IMO it assumes that chunk is now owned by the stream, it also doesn't notify that this chunk is now free (because it assumes that this chunk isn't used by anything else at that point). Also, if there is no copying then there is no point modifying highWaterMark, perhaps it matters when writing smaller chunks?.. I cannot find all these details in nodejs docs, perhaps I just need to clone nodejs and see the code to understand what's going on

[addaleax]

does it copy memory anywhere at all or it passes raw memory from written chunks to OS api without copying it

I don’t think there’s any case where it needs to be copied, so that would be interesting to look up.

Also, if there is no copying then there is no point modifying highWaterMark, perhaps it matters when writing smaller chunks?..

I think that’s not entirely true. The highWaterMark does affect the overhead of management operations that take place.

Basically, when you stream.write(chunk) IMO it assumes that chunk is now owned by the stream, it also doesn't notify that this chunk is now free

Buffer instances are typed arrays whose lifetime is managed by the JS engine – the is no concept of ownership involved here, garbage collection is what cleans up after them (you could say they are owned by the engine).

[pps83]

I don’t think there’s any case where it needs to be copied, so that would be interesting to look up.

Yes, should would be no point, it should pass Buffer's raw mem pointer to socket api.

I think that’s not entirely true. The highWaterMark does affect the overhead of management operations that take place.

I think I misunderstood meaning of this highWaterMark. I don't know how exactly it works, but I have two ideas about this highWaterMark: 1) if you write byte by byte to a socket (or a stream) then the stream won't make many small writes down the chain and buffer them until highWaterMark bytes accumulated and make only one bigger write. In this case, it would need to copy bytes to a bigger buffer. 2) when a chunk is written to a stream it's passed as-is to the underlying stream mechanism, stream will queue multiple chunks and will send them once underlying mechanism can accept them; so in this case highWaterMark is an indication to the stream how much data it should accept into its queue before it stops pulling more data.
I guess, 1) is wrong, then in case if it's 2) then there is no need for copying.

Buffer instances are typed arrays whose lifetime is managed by the JS engine – the is no concept of ownership involved here, garbage collection is what cleans up after them (you could say they are owned by the engine).

By ownership I mean that when a chunk is written to a stream the stream assumes that nobody uses that chunk anymore (e.g. doesn't modify bytes in it). Otherwise, if you create a buffer, write some data to it and then write that buffer to a stream and right after that if you try to reuse that buffer for the next operation you may affect previous write to the stream.
Maybe I'm doing something wrong, I'll explain why I bother with this stuff. I'm a C++ guy, in C++ for high traffic code I could preallocate certain memory pool (let say in 64KB chunks) and try to reuse it. I was thinking to do something like that for this streaming code to avoid taxing GC. If there is 1gbps and if _chunks is a list of 1KB sized buffers you can calculate how many buffers are constantly created and re-created. Not sure if that's an issue in V8/node, in c++ for example even for 1gbps it's most likely OK to to allocate/release buffers and perhaps there is no need to use a pool there.
Also, I'm not too much worried about mem copying, since any decent CPU nowadays can easily do multiple GB/s. in playback-stream when chunks are received from upstream they are allocated by V8/node and memory is copied from OS's socket api code (most likely it works like that I assume), in C++ I could use my pool for reading as well. That way, for a given load I could know that I need lets say 20GB of ram and that's it to ensure that it doesn't fall apart if buffers aren't cleaned fast enough by GC and lead to quick heap growth.

[addaleax]

Some things:
https://nodejs.org/api/stream.html#stream_buffering – for the docs on the HWM
https://github.com/nodejs/node/blob/master/lib/_stream_readable.js – Readable stream implementation

if you write byte by byte to a socket (or a stream) then the stream won't make many small writes down the chain and buffer them until highWaterMark bytes accumulated and make only one bigger write. In this case, it would need to copy bytes to a bigger buffer.

No, for that you’d probably have to call .cork() and .uncork(). Your 2)nd assumption seems to be the right one.

By ownership I mean that when a chunk is written to a stream the stream assumes that nobody uses that chunk anymore (e.g. doesn't modify bytes in it). Otherwise, if you create a buffer, write some data to it and then write that buffer to a stream and right after that if you try to reuse that buffer for the next operation you may affect previous write to the stream.

Yeah, that buffer is assumed to be unmodified. The actual details are a bit more complicated; buffers may get concatenated into a new, independent buffer, if that makes sense – that should not be a problem here, normally streams try to read their input in the size of one HWM, keep that in the internal buffer, and pass it along as one unit.

I could preallocate certain memory pool (let say in 64KB chunks) and try to reuse it. I was thinking to do something like that for this streaming code to avoid taxing GC.

There are memory pools both for Buffers, controlled by Buffer.poolSize, and for fs.ReadStreams in the size of the HWM… those should match your idea of a C++ memory pool.

in playback-stream when chunks are received from upstream they are allocated by V8/node and memory is copied from OS's socket api code (most likely it works like that I assume),

Not exactly – for fs streams, the Buffer is allocated in JS and then filled with data from the underlying source. For net sockets, the Buffer is allocated once data is available via a normal malloc(), then filled using the standard C socket API, and then a Buffer object is created that points to that memory region (V8 does support external memory sources).

if buffers aren't cleaned fast enough by GC and lead to quick heap growth.

I’m not a 100 % sure on the details, but the garbage collector in V8 does keep an eye on the current memory situation and makes it decisions for when to run based on that. I think --max_old_space_size is the runtime option that lets you turn that screw.

And on a more general note: Thanks for asking these questions! It’s pretty helpful to go through all of this and make sure that everything is implemented the way it should be, or getting input on where to look for improvements. If you want to scan through the source code yourself, you’ll always find someone here or on freenode in #node-dev to help you find your way through the source.

One more things that may or may not interest you: nodejs/node-eps#27 – A proposal to add a .readInto method for streams that gives the user full control over the memory allocation mechanisms.

[pps83]

I wanted to ask if there was a possibility to read into predefined buffer, but I'd already written a lot in previous message so I didn't ask. But you still answered :) 👍

I'm kind of new to nodejs, I've done some easy stuff, nothing complex though, so I'm a bit lost in details. By the way, around 2003-2004 I downloaded spidermonkey (first original js implementation) and used it to control some stuff my C++ project (a telephony server). I really liked it back then and wanted to expand functionality in my free time but got stuck implementing #include "..." (aka require(...) in node) moved to another project and never came back to it. Now I'm back trying to use js for server code :)

[addaleax]

I wanted to ask if there was a possibility to read into predefined buffer, but I'd already written a lot in previous message so I didn't ask. But you still answered :) 👍

I don’t think that is generally supported for any kind of stream right now – the lowlevel fs.read() function does that, though (docs). You could probably also call that with a socket fd if you want to, but if the read call has to wait for data it might block one thread in the thread pool (I think).

Truth is, most JS developers don’t really care about the way the Node internals work, and that’s a good sign, because it means that they are working most of the time, and for people new to Node.js, the huge diversity of modules in the ecosystem is probably a lot more interesting than any of that. :)

Also, counter-fun-fact: https://github.com/mozilla/spidernode ;)