stream: avoid drain for sync streams

[ronag]

Previously a sync writable receiving chunks
larger than highwatermark would unecessarily
ping pong needDrain.

300% improvement for sync streams when chunks are bigger than HWM.

 streams/writable-manywrites.js len=1024 callback='no' writev='no' sync='no' n=2000000                     1.84 %       ±3.27%  ±4.39%  ±5.79%
 streams/writable-manywrites.js len=1024 callback='no' writev='no' sync='yes' n=2000000                    2.00 %       ±2.02%  ±2.70%  ±3.54%
 streams/writable-manywrites.js len=1024 callback='no' writev='yes' sync='no' n=2000000                    1.68 %       ±2.12%  ±2.83%  ±3.68%
 streams/writable-manywrites.js len=1024 callback='no' writev='yes' sync='yes' n=2000000                   1.31 %       ±2.44%  ±3.25%  ±4.23%
 streams/writable-manywrites.js len=1024 callback='yes' writev='no' sync='no' n=2000000                    0.19 %       ±1.77%  ±2.36%  ±3.08%
 streams/writable-manywrites.js len=1024 callback='yes' writev='no' sync='yes' n=2000000            *      2.60 %       ±2.46%  ±3.28%  ±4.28%
 streams/writable-manywrites.js len=1024 callback='yes' writev='yes' sync='no' n=2000000                   1.32 %       ±2.37%  ±3.16%  ±4.14%
 streams/writable-manywrites.js len=1024 callback='yes' writev='yes' sync='yes' n=2000000                 -1.15 %       ±2.59%  ±3.46%  ±4.53%
 streams/writable-manywrites.js len=32768 callback='no' writev='no' sync='no' n=2000000                    0.42 %       ±2.27%  ±3.04%  ±4.00%
 streams/writable-manywrites.js len=32768 callback='no' writev='no' sync='yes' n=2000000          ***    308.72 %       ±7.84% ±10.53% ±13.91%
 streams/writable-manywrites.js len=32768 callback='no' writev='yes' sync='no' n=2000000                   0.22 %       ±1.78%  ±2.37%  ±3.10%
 streams/writable-manywrites.js len=32768 callback='no' writev='yes' sync='yes' n=2000000         ***    274.22 %       ±6.83%  ±9.20% ±12.20%
 streams/writable-manywrites.js len=32768 callback='yes' writev='no' sync='no' n=2000000                   1.56 %       ±2.40%  ±3.23%  ±4.27%
 streams/writable-manywrites.js len=32768 callback='yes' writev='no' sync='yes' n=2000000         ***    323.42 %       ±5.95%  ±8.00% ±10.60%
 streams/writable-manywrites.js len=32768 callback='yes' writev='yes' sync='no' n=2000000                 -0.42 %       ±1.90%  ±2.55%  ±3.35%
 streams/writable-manywrites.js len=32768 callback='yes' writev='yes' sync='yes' n=2000000        ***    287.75 %       ±7.60% ±10.21% ±13.50%

Checklist

• make -j4 test (UNIX), or vcbuild test (Windows) passes
• tests and/or benchmarks are included
• documentation is changed or added
• commit message follows commit guidelines

[ronag]

Not sure if this should be a semver-major.

[nodejs-github-bot]

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

[lpinca]

This looks great if it does not cause issues in user-land (it will almost certainly break ws tests).

[ronag]

This looks great if it does not cause issues in user-land (it will almost certainly break ws tests).

I'll run a CITGM.

[nodejs-github-bot]

CI: https://ci.nodejs.org/job/node-test-pull-request/30810/
CITGM: https://ci.nodejs.org/view/Node.js-citgm/job/citgm-smoker/2351/

[ronag]

CITGM looks good apart from ws. Sorry @lpinca. I'm fine with skipping this PR if it's a big effort from you part.

[lpinca]

It should be just a matter of fixing those two failing tests (not sure how), core functionality should not be broken.

[lpinca]

Perhaps wait until #32780 lands so I don't have to fix them twice.

[ronag]

@nodejs/streams

[mcollina]

LGTM

[mcollina]

I would add a few comments about this change here.

[ronag]

I'm not sure what kind of comment?

[mcollina]

I tagged this as "dont-land" on everything but v14.x.

[lpinca]

I've fixed the failing tests: websockets/ws@18d773d.

[nodejs-github-bot]

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

[ronag]

Landed in 003fb53

[mscdex]

@ronag This PR (found after bisecting on v14.x) seems to be causing issues for ssh2/ssh2-streams when large amounts of data are being passed through a stream. The same regression still exists on master as of this writing. Can we at least revert this on v14.x, especially if there is no guidance on how to deal with the breakage in a backwards-compatible manner?

[mcollina]

@mscdex +1 in reverting this. I would love to have a regression test added (even in ssh2 and we can add it to citgm) so that we do not regress again.

[ronag]

I'm ok with revert + a comment in the code. Just a note here, reverting this might have performance impact e.g. when piping from a file stream (which has a larger highwater mark) to a sync transform stream (which has a smaller highwatermark) e.g. for hashing.

It's a bit strange to me that this would cause breakage. Is ssh2 doing something funky with streams?

[p-j]

Hi,
If it can help, as part of troubleshooting a related issue @theophilusx made this script that will hang forever if the passed data is large enough as discussed here.

[mcollina]

None of those examples are self-contained :/, they need an active ssh server to run.

[mscdex]

I've stripped down the relevant ssh2 internals so that it is a completely standalone test that exhibits the problem. The code here doesn't exactly represent what is happening internally in most cases (mostly referring to the write happening before piping, that's done here to simulate backpressure and buffered data), but it should sufficiently demonstrate the same issue. Also it's possible the test code could be simplified further if someone wanted to have a go at it.

Source code

'use strict';

const { Duplex, Transform } = require('stream');
const { connect, createServer } = require('net');
const { inherits } = require('util');

function Channel(protoStream, socket) {
  const streamOpts = {
    highWaterMark: 2 * 1024 * 1024,
    allowHalfOpen: false
  };

  Duplex.call(this, streamOpts);

  socket.on('drain', () => {
    console.log('ondrain()', 'waitClientDrain', this._waitSocketDrain);
    if (this._waitSocketDrain) {
      this._waitSocketDrain = false;
      if (this._chunk)
        this._write(this._chunk, null, this._chunkcb);
      else if (this._chunkcb)
        this._chunkcb();
    }
  });

  this._protoStream = protoStream;

  // outgoing data
  this._waitSocketDrain = false;

  this._chunk = undefined;
  this._chunkcb = undefined;
}
inherits(Channel, Duplex);

Channel.prototype._read = function(n) {};

Channel.prototype._write = function(data, encoding, cb) {
  const protoStream = this._protoStream;
  const packetSize = 64 * 1024;
  const len = data.length;
  let p = 0;

  while (len - p > 0) {
    let sliceLen = len - p;
    if (sliceLen > packetSize)
      sliceLen = packetSize;

    const slice = data.slice(p, p + sliceLen);
    const ret = protoStream.sendData(slice);
    console.log(`Channel._write() ret = ${ret} after writing ${slice.length} byte(s)`);

    p += sliceLen;

    if (!ret) {
      this._waitSocketDrain = true;
      this._chunk = undefined;
      this._chunkcb = cb;
      break;
    }
  }

  console.log(`Channel._write() outside of loop; p=${p} len=${len} waitSocketDrain=${this._waitSocketDrain}`);

  if (len - p > 0) {
    if (p > 0) {
      // partial
      let buf = Buffer.allocUnsafe(len - p);
      data.copy(buf, 0, p);
      this._chunk = buf;
    } else {
      this._chunk = data;
    }
    this._chunkcb = cb;
    return;
  }

  if (!this._waitSocketDrain)
    cb();
};

function ProtocolStream() {
  Transform.call(this, { highWaterMark: 32 * 1024 });
}
inherits(ProtocolStream, Transform);

ProtocolStream.prototype.sendData = function(data) {
  return this.push(data);
};

createServer(function(socket) {
  this.close();
}).listen(0, '127.0.0.1', function() {
  const { address, port } = this.address();
  connect(port, address, function() {
    console.log('Client connected');
    const protoStream = new ProtocolStream();
    const channel = new Channel(protoStream, this);
    channel.on('finish', () => {
      console.log('============ Channel finish');
      this.destroy();
    });
    channel.write(Buffer.alloc(128 * 1024));
    protoStream.pipe(this).pipe(protoStream);
    channel.end(Buffer.alloc(128 * 1024));
  });
});

Executing the test on node v10.19.0 results in the following output:

$ node-v10.19.0 stream-issue.js
Client connected
Channel._write() ret = false after writing 65536 byte(s)
Channel._write() outside of loop; p=65536 len=131072 waitSocketDrain=true
ondrain() waitClientDrain true
Channel._write() ret = false after writing 65536 byte(s)
Channel._write() outside of loop; p=65536 len=65536 waitSocketDrain=true
ondrain() waitClientDrain true
Channel._write() ret = false after writing 65536 byte(s)
Channel._write() outside of loop; p=65536 len=131072 waitSocketDrain=true
ondrain() waitClientDrain true
Channel._write() ret = false after writing 65536 byte(s)
Channel._write() outside of loop; p=65536 len=65536 waitSocketDrain=true
ondrain() waitClientDrain true
============ Channel finish
$

Executing the test on node v14.1.0 or later results in the following output:

$ node-v14.7.0 stream-issue.js
Client connected
Channel._write() ret = false after writing 65536 byte(s)
Channel._write() outside of loop; p=65536 len=131072 waitSocketDrain=true

(... and the process never exits)

[ronag]

Looks like it ends up making an incorrect assumption regarding whether or not 'drain' will be emitted. Maybe something with _write being called in the drain handler which can set _waitSocketDrain to true without the stream actually needing a drain.

I think this is weird/incorrect/bug usage and node streams are behaving correctly. However, I guess it is is breaking change so if no one feels strongly about the performance loss I think we just revert it.

[mcollina]

@mscdex @ronag how hard would it be to fix on the ssh2 side? What puzzles me is that all of ssh2 tests were passing, so it would have been extremely hard to spot this.

Given that we were on the fence on this being semver-major we can revert on v14 and keep it on master/v15.

[theophilusx]

wrt to ssh2 and being hard to spot. My testing found this error to be (or appear to be) intermittent. i.e. I could run the same test 5 times and it would not fail until the 5th time (with same test input), but sometimes it would fail on the first test, or second etc. When I tried testing with input data sizes which increased by 10k per run, if I started with a relatively small size e.g. 90k, it would not fail (I wold kill the test when it reached 50Mb). If on the other hand, I started with 200k or 500k, it would usually fail immediately. I also found that if I was connecting with a destination of localhost, rather than a remote sftp server or one running on a vbox image, I did not see the failure, though I didn't test as extensively with this target. It is quite likely the existing tests just didn't trigger the issue. So fixing in ssh2 will be a little challenging given how difficult it is to know if the issue is fixed or just hasn't been triggered. Reverting and putting into v15 would at least buy some time to try and find the right fix for ssh2. Just my 2c - ssh2/ssh2-streams author may have a different view. Happy to assist with testing if that is at all helpful.

[schmod]

Not to do the "+1" thing here, but we've seen this regression in the wild with the (quite popular) ssh2 library.

Is there any chance of getting this reverted?

[wa-Nadoo]

@schmod, the changes reverted in #35941. @ronag, maybe this PR should be reopened?