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

@@ -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);
     }
   }
 }

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');
   }
 }