ComFreek
commented
9 years ago Streams do support backpressure (with highWaterMark) — although I didn't get it to work (= the data just flows) and I currently don't have much time to investigate further. For this to work we have to introduce some kind of function whose call blocks the component. Using Promise.coroutine() and yield would look as follows:
// a simple copier
var ip;
while ((ip = yield inport.receive()) !== null) {
yield outport.send(ip);
}My implementation of OutputPort#blockingSend looks as follows:
OutputPort.prototype.sendBlocking = function(ip) {
var self = this;
function write(_ip, resolve) {
var ok = self.conn.write(_ip);
if (ok) {
resolve();
}
else {
self.conn.once('drain', function () {
write(_ip, resolve);
});
}
}
return new Promise(function (resolve, reject) {
write(ip, resolve);
});
};The call to self.conn.write() tries to write data to the ProcessConnection. Ideally, the ProcessConnection returns false in case the limit of unread IPs (given by the so-called 'capacity') is reached. The ProcessConnection should also send a 'drain' event in case an IP has been read, so that an IP can be pushed.
The problem is that we introduce yet another ugly 'yield' statement. 'yield' makes perfect sense when reading, but I don't like its appearance when sending.
tlrobinson
kenhkan
jpaulm
@tlrobinson @ComFreek
I've been trying to tackle back pressure in my fork, and I must say I'm hitting a dead-end and it'd be nice to hear your thoughts on this.
Take a look at this gist. Replace the content of
recvr.json @ComFreek 's fork and runfbptest01.js. You should see something similar to this. The timing doesn't matter, but (correct me if I'm wrong please @jpaulm ) the copier should stop after the 5th copy is sent, because of a capacity of 5 set on the connetion between copier and recvr.I know that @ComFreek might not have gotten to that point in his fork. Me neither. But the issue is that it seems to be rather difficult in JS (at least with ES5 as that was my thesis).
With generators, the control flow is inverted here so there doesn't seem to be an obvious way to have the callee of yield to suspend, or continue to suspend, the caller (in this case, the FBP process). In this particular case,
yieldis returned a bluebird promise butPromise.coroutine()would continue the generator as soon asreceivesettles, which it must do. I haven't put too much thought into this problem in a streams-based solution, but I suspect that it's a similar situation.Any idea on this problem?