From 0df89c01877d99d1520e203b22cbbce3563913be Mon Sep 17 00:00:00 2001 From: Stephan Hesse Date: Mon, 3 Aug 2020 09:58:35 +0200 Subject: [PATCH] ReadableStream performance improvements (#7) * various fixes to readable-stream implementation using blocking srt calls. - refactor reading/pushing routine in order to allow higher througput: * call srt read() in the same tick/stack as the class _read impl call * schedule a timer only as needed if no connection yet or socket buffer empty * blocking loop to push all bytes requested, or until push returns false or until no more data is in the socket buffer * generally, handle case properly where no data is yet available from srt-socket read() * in case no more data, retains the currently requested value remainder from last _read call -> this will allow for getting as much data as requested (if socket delivers it) within the tick where it was requested or any next one in a best effort manner. it will block the main-thread with "too long" execution frames/ticks eventually as the data-demand from the _read calls becomes high in amout/frequency. but that is simply a consequence of the fact the SRT calls here are blocking I/O. the previous implementation was also resulting in this congestion of the main loop queue as read calls would block it, but at best there was a 100ms lag between a request for read and its actual fulfillment. one further improvement of this fix could be to split up the requested bytes amount onto several ticks (executing one partial read directly, and further ones in further timer scheduled calls, in order to avoid blocking ticks). - use 50ms timer default for read continuation retry timer - use 50ms interval for initial connection epoll'ing - use consts for all timeout values - implement base class _destroy method & refactor close() in consequence - trigger Stream base class "readable" event on connection and first packet * server: add an event for when a client was accepted to pass the fd --- src/server.js | 3 +- src/stream.js | 116 ++++++++++++++++++++++++++++++++++++++------------ 2 files changed, 90 insertions(+), 29 deletions(-) diff --git a/src/server.js b/src/server.js index 506fd2d..115fb7b 100644 --- a/src/server.js +++ b/src/server.js @@ -21,7 +21,8 @@ class SRTServer extends EventEmitter { this.emit("listening", iface, port); while (true) { const fhandle = libSRT.accept(this.socket); - debug("New client connected"); + debug("Client connection accepted"); + this.emit("accepted", fhandle); } } } diff --git a/src/stream.js b/src/stream.js index 4e0aa78..f9fec6e 100644 --- a/src/stream.js +++ b/src/stream.js @@ -2,7 +2,9 @@ const { Readable, Writable } = require('stream'); const LIB = require('../build/Release/node_srt.node'); const debug = require('debug')('srt-stream'); -const EPOLLUWAIT_CALL_PERIOD_MS = 500; +const CONNECTION_ACCEPT_POLLING_INTERVAL_MS = 50; +const READ_WAIT_INTERVAL_MS = 50; + const EPOLLUWAIT_TIMEOUT_MS = 1000; const SOCKET_LISTEN_BACKLOG = 10; @@ -27,37 +29,45 @@ class SRTReadStream extends Readable { this.address = address; this.port = port; this.fd = null; - this.readTimer = null; + + this._eventPollInterval = null; + this._readTimer = null; } - listen(cb) { + /** + * + * @param {Function} onData Passes this stream instance as first arg to callback + */ + listen(onData) { this.srt.bind(this.socket, this.address, this.port); this.srt.listen(this.socket, SOCKET_LISTEN_BACKLOG); const epid = this.srt.epollCreate(); this.srt.epollAddUsock(epid, this.socket, LIB.SRT.EPOLL_IN | LIB.SRT.EPOLL_ERR); - let t = setInterval(() => { + const interval = this._eventPollInterval = setInterval(() => { const events = this.srt.epollUWait(epid, EPOLLUWAIT_TIMEOUT_MS); events.forEach(event => { const status = this.srt.getSockState(event.socket); if (status === LIB.SRT.SRTS_BROKEN || status === LIB.SRT.SRTS_NONEXIST || status === LIB.SRT.SRTS_CLOSED) { - debug("Client disconnected"); + debug("Client disconnected with socket:", event.socket); this.srt.close(event.socket); this.push(null); this.emit('end'); } else if (event.socket === this.socket) { const fhandle = this.srt.accept(this.socket); - debug("New connection"); + debug("New client connected with socket:", this.socket, "and fhandle:", fhandle); this.srt.epollAddUsock(epid, fhandle, LIB.SRT.EPOLL_IN | LIB.SRT.EPOLL_ERR); + this.emit('readable'); } else { - debug("Data from client"); + debug("Data from client on fd:", event.socket); this.fd = event.socket; - clearInterval(t); - cb(this); + clearInterval(interval); + onData(this); + this.emit('readable'); } }); - }, EPOLLUWAIT_CALL_PERIOD_MS); + }, CONNECTION_ACCEPT_POLLING_INTERVAL_MS); } connect(cb) { @@ -69,26 +79,76 @@ class SRTReadStream extends Readable { } close() { + this.destroy(); + } + + _scheduleNextRead(requestedBytes, timeoutMs) { + this._readTimer = setTimeout(this._readSocketAndPush.bind(this, requestedBytes), timeoutMs); + } + + _clearScheduledRead() { + clearTimeout(this._readTimer); + this._readTimer = null; + } + + _readSocketAndPush(bytes) { + this._clearScheduledRead(); + if (this.fd === null) { + this._scheduleNextRead(bytes, READ_WAIT_INTERVAL_MS); + return; + } + let remainingBytes = bytes; + while(true) { + const buffer = this.srt.read(this.fd, bytes); + if (buffer === null) { // connection likely died + debug("Socket read call returned 'null'"); + this.close(); + break; + } + // we expect a Buffer object here, but + // -1 is the SRT_ERROR value that would get returned + // if there is no data to read yet/anymore + if (buffer === -1) { + this._scheduleNextRead(remainingBytes, READ_WAIT_INTERVAL_MS); + break; + } + //debug(`Read ${buffer.length} bytes from fd`); + + // @see https://nodejs.org/api/stream.html#stream_readable_push_chunk_encoding + if (this.push(buffer)) { + remainingBytes -= buffer.length; + if (remainingBytes <= 0) { + // 0 as a timer value acts as setImmediate task (next tick ideally) + break; + } + } else { + debug("Readable.push returned 'false' at remaining bytes:", remainingBytes); + break; + } + } + } + + /** + * @see https://nodejs.org/api/stream.html#stream_readable_read_size_1 + * @param {number} bytes + */ + _read(bytes) { + debug('Readable._read(): requested bytes:', bytes); + this._readSocketAndPush(bytes); + } + + /** + * @see https://nodejs.org/api/stream.html#stream_readable_destroy_err_callback + * @param {Error} err + * @param {Function} cb + */ + _destroy(err, cb) { + // guard from closing multiple times + if (this.fd === null) return; this.srt.close(this.socket); this.fd = null; - } - - _readStart(fd, size) { - this.readTimer = setInterval(() => { - let chunk = this.srt.read(fd, size); - debug(`Read chunk ${chunk.length}`); - if (!this.push(chunk)) { - this._readStop(); - } - }, 100); - } - - _readStop() { - clearInterval(this.readTimer); - } - - _read(size) { - this._readStart(this.fd, size); + this._clearScheduledRead(); + this.emit('close'); } }