Conform to a Stream Interface

[reconbot]

We currently emit data events, and are sort of pausable (not sure if we're implemented correctly) but don't conform to a streams interface. Since we're compatible across node 0.10 through 6 we need to make sure we stay compatible if possible. If not, what is different?

[reconbot]

Oh and I'd love to have a test suite that proves we confirm. How do we test streams?

[chrisdickinson]

A bit of an info dump follows — apologies if this repeats stuff you already know!

Here's the streams3 flow from the perspective of the events emitted (relations are temporal, not indicative of causation):

And a (slightly out of date, but not horribly so) interaction chart of methods for streams3 (relations are causative, not temporal!):

A couple of differences jump out at me when comparing and contrasting the stream implementation in serialport vs. the charts above (& implementation code):

1️⃣ .pause() and .resume() are overridden. Even though they're overridden, from a user's perspective they still do what they say on the tin, so it's okay — the biggest user-facing difference is that the "resume" and "pause" events aren't emitted, and isPaused() won't be updated.

2️⃣ Is probably the trickier difference: _read appears to be called from different parts of the implementation. streams2+ makes the assumption that the streams machinery will be the only thing calling ._read.

The idea is that streams2+ readable streams are backed by a lower level resource — in this case, our fd — and that streams2+ will communicate the need to start or stop reading to implementing code via _read(n) and .push(data) === false. streams2+ will call ._read when more data is needed, and will respond to .push(data) with false when no more data is needed. In practice, Node stream subclasses (like net.Socket) will start reading from an underlying handle (readStart() here) on _read if not already reading, and stop reading on push(data) === false (readStop() here). The goal of letting streams2+ manage the begin/end cycle of reads separately from .read() calls is to insulate the underlying resources from transitioning between reading and paused states too frequently due to backpressure.

Luckily, it appears that the existing implementation is only missing the .push call (& the corresponding check to see if the stream should stop reading), but it has captured that with _emitData.

3️⃣ There will be some differences in .pipe() when subclassing from stream.Duplex (which uses stream.Readable#pipe) vs stream.Stream#pipe. I haven't looked too closely at what that would mean for this project yet though — it definitely bears some attention.

4️⃣ .write(buffer, cb) would be replaced by ._write(chunk, encoding, cb) (& should pass the encoding to new Buffer() if chunk is a string), but otherwise shouldn't have to change much.

Apologies again for the info dump — hopefully this helps! I'm happy to answer any questions that arise from this as well!

[reconbot]

This is beautiful @chrisdickinson this is exactly what I needed. Do you know which node version streams 2 appeared in? We have a very wide range of support.

[chrisdickinson]

Streams2 first appeared (in early form) in v0.9, and officially for the first time in v0.10. If you want to conform to a streams3 interface on earlier platforms (which is recommended since the interface is backwards compatible), you can use the readable-stream package @ ^2.0.0.

[mcollina]

In order to conform with Streams3, serial-port should inherits from Duplex from readable-stream. It's basically a rewrite, in the sense that most of the code in https://github.com/EmergingTechnologyAdvisors/node-serialport/blob/master/lib/serialport.js#L328-L347 is to deal with streams. All of that should be handled by Duplex, and the only bits should be dealing with SerialPortBinding.

Currently we are targeting node v0.10 with readable-stream, so it will not be ok to use on v0.8 (not sure if you still support 0.8 or not).

[reconbot]

We only go back to 10, so that's OK. Having less code off our own is OK too.

What's the best way to test?

On Mon, May 30, 2016, 6:26 AM Matteo Collina [email protected]
wrote:

In order to conform with Streams3, serial-port should inherits from Duplex
from readable-stream. It's basically a rewrite, in the sense that most of
the code in
https://github.com/EmergingTechnologyAdvisors/node-serialport/blob/master/lib/serialport.js#L328-L347
is to deal with streams. All of that should be handled by Duplex, and the
only bits should be dealing with SerialPortBinding.

Currently we are targeting node v0.10 with readable-stream, so it will not
be ok to use on v0.8 (not sure if you still support 0.8 or not).

—
You are receiving this because you authored the thread.
Reply to this email directly, view it on GitHub
#832 (comment),
or mute the thread
https://github.com/notifications/unsubscribe/AABlbns7LGRIIv9byfAo43BaCrulYqv4ks5qGrtLgaJpZM4IpNsy
.

[mcollina]

just pipe an instance of Readable to it and pipe back to a Writable, and have a little device doing "echo", should be enough. I will be happy to review, just cc me.

[reconbot]

Hi @chrisdickinson and @mcollina I did a spike on implementing streams today and have a few questions.

1. Currently we have no control over our windows reading. It can't be paused. If we keep calling push() what negative effects does that have? This is something we'll fix eventually but probably not before moving to readable-stream.
2. Opening and closing and pausing and resuming are not a perfect match.

We have an open/not open states (Including opening, closing and closed as not open states.) How do we communicate that with stream semantics? Currently we throw errors and make people wait for our open event before writing and setting options. We also let people create a port, open it, close it, then open it again. This doesn't seem possible with the streams interface. Unless.. are we just paused when we're not open?

I checked out net.Socket and they have a destroy method, no open or reconnect method. And it's not clear to me what happens if we can't open the socket. (error?)

What do you think?

Thanks!

[mcollina]

1. Currently we have no control over our windows reading. It can't be paused. If we keep calling push() what negative effects does that have? This is something we'll fix eventually but probably not before moving to readable-stream.

I will fill up the internal buffer, and in turn the heap of your node process. It's basically an externally-driven memory leak. However, buffering unlimited amount of data makes very little sense from a program point of view, so you can probably code some bits to mitigate.

I highly recommend controlling the buffering window from Node.

2. Opening and closing and pausing and resuming are not a perfect match.

Why not? Can you please articulate?

We have an open/not open states (Including opening, closing and closed as not open states.) How do we communicate that with stream semantics? Currently we throw errors and make people wait for our open event before writing and setting options. We also let people create a port, open it, close it, then open it again. This doesn't seem possible with the streams interface. Unless.. are we just paused when we're not open?

The stream interface has a very specific state machine. node-serialport was born before that state machine was finalized. Can you please articulate on what those states mean?

[reconbot]

I'd like to make some flow charts to help show how SP currently works, what did you use to make yours?

I was looking at fs.createReadStream which is the closest in form to a serialport. It can optionally open or close the related file descriptor on stream creation and stream end.

We have a port object that starts closed, we have async open and close operations with an open and close events. A port object can be opened or closed as often as a user likes. The port also has a few internal transitions closing and opening where the port is not open. (.isOpen() determine that)

write operations currently error if the port is closed. We have a slew of async configuration and control methods (set, flush, and drain which have nothing to do with the nodejs stream) that also error when called on not open ports.

[mcollina]

as far as I remember, serialport is polling for reads at c++ level, right? I think the best starting point would be to avoid that in the first place.

[reconbot]

only on windows, which is why I brought it up

[reconbot]

I'm continuing to work on this in #906 and have come to what feels like the most complex interface _read().

I've moved our c++ code into an abstraction (currently being called the bindings layer) and I'm defining and testing that api. This abstraction needs to support our code but also needs to be reasonable for chrome serialport, the web serial api, and alternative bindings. (eg, one written in Rust, ones that are really bluetooth devices, ones that change the backend to work over tcp/udp - These are current examples in the wild.)

I'm stuck on what I want reading to look like. Should we ask the binding to provide an implementation of _read() or ask them to provide a read() function that calls back when data is available and managing pushing, pausing and resuming in the SerialPort class?

I know push is also used to close the port push(null) but that's not a facility we need to expose here as they can make use of a disconnected callback.

var binding = new Binding({ push: push, disconnect: disconnectCB});
binding.startReading(bytes);

describe('#startReading', function() {
  it('keeps calling push() until it returns false');
  it("doesn't mind being called multiple times");
  it('always calls push at least once');
});

This would allow the binding authors to do whatever they want wherever they want. C++ bindings can keep as much processing in c++ as possible, including leveraging data available events (epoll, the mess we do with windows, or just waiting for the next data event). However it's complex.

A more simple api might be.

var binding = new Binding({ disconnect: disconnectCB});
binding.read(function(data) {
  // data is a buffer that is guaranteed to have at least 1 byte of data
  // binding is required to call the disconnect callback if the port is unexpectedly closed
});

describe('#read', function() {
  it('returns with available data when it becomes available');
});

In the back in my head I think this limits bindings and might possibly cause data bottlenecks down the line but I haven't gamed it out. I'd love some feedback.

[mcollina]

I think you either have two options, a push interface or a pull interface. Both approach are sound.

push interface

function push (data) {
  console.log(data)
}

// emits 'error' if something goes wrong
var binding = new Binding({ push });
binding.start()
setInterval(function () {
  if (binding.running) {
    binding.pause()
  } else {
    binding.resume()
  }
}, 1000)

setTimeout(function () {
  binding.stop()
}, 10000)

This is essentially the same that you have right now, maybe in a different form. The serial port pushes data to the JS world. With this API, the actual high-level work would be to offer this through a Readable, which works in pull mode.

pull interface

var binding = new Binding();
binding.read(function(err, data) {
  // data is a buffer that is guaranteed to have at least 1 byte of data
  // err is present if the port is unexpectedly closed
});

This is extremely simpler from serial-port perspective, but it put more work on the binding developers. I think that's a good thing, because it also keep a very simple API interface for bindings. Some bindings might end up implementing some form of buffering themselves, if they cannot pause internally.

[reconbot]

I'd even argue the pull interface maps to fs.read, read() and ReadFile() so it might not be harder to impliment.

Though I do notice that chrome serialport and the web serialport spec on the other hand are push interfaces, so they'd have to jump through some hoops, but at least that would all be in JS land.

[reconbot]

I've merged #906, and serialport is now a stream.

I'm wondering if we're doing close correctly as we have our own close method. Seems to work pretty good in general.

[johnelliott]

🔥🔥🔥