custom flow control and discard limit

demo/client.ts

@@ -215,7 +215,6 @@ function initOptions(term: TerminalType): void {
     'handler',
     'screenKeys',
     'termName',
-    'useFlowControl',
     // Complex option
     'theme'
   ];

demo/server.js

@@ -9,13 +9,16 @@ var pty = require('node-pty');
  */
 const USE_BINARY_UTF8 = false;
 
+// pty --> websocket buffering
+const MAX_SEND_INTERVAL = 5;
+const MAX_CHUNK_SIZE = 16384;
+
 
 function startServer() {
   var app = express();
   expressWs(app);
 
-  var terminals = {},
-      logs = {};
+  var terminals = {};
 
   app.use('/xterm.css', express.static(__dirname + '/../css/xterm.css'));
   app.get('/logo.png', (req, res) => res.sendFile(__dirname + '/logo.png'));
@@ -45,15 +48,14 @@ function startServer() {
           rows: rows || 24,
           cwd: process.env.PWD,
           env: process.env,
-          encoding: USE_BINARY_UTF8 ? null : 'utf8'
+          encoding: USE_BINARY_UTF8 ? null : 'utf8',
+          handleFlowControl: true,
+          flowControlPause: '\x1b^pause\x1b\\',
+          flowControlResume: '\x1b^resume\x1b\\'
         });
 
     console.log('Created terminal with PID: ' + term.pid);
     terminals[term.pid] = term;
-    logs[term.pid] = '';
-    term.on('data', function(data) {
-      logs[term.pid] += data;
-    });
     res.send(term.pid.toString());
     res.end();
   });
@@ -72,59 +74,94 @@ function startServer() {
   app.ws('/terminals/:pid', function (ws, req) {
     var term = terminals[parseInt(req.params.pid)];
     console.log('Connected to terminal ' + term.pid);
-    ws.send(logs[term.pid]);
+
+    const _send = data => {
+      // handle only 'open' websocket state
+      if (ws.readyState === 1) {
+        setTimeout(() => ws.send(data), 200);
+      }
+    }
+
+    const ACK_WATERMARK = 131072;//524288;
+    const FLOW_CONTROL_ACK = '\x1b^ack\x1b\\'; // PM ack ST
+    const MAX_ACK_DIFF = 7;
+    let ack_expected = 0;
+    let sent = 0;
 
     // string message buffering
-    function buffer(socket, timeout) {
+    function buffer(timeout, limit) {
       let s = '';
       let sender = null;
       return (data) => {
         s += data;
-        if (!sender) {
+        if (s.length > limit) {
+          clearTimeout(sender);
+          _send(s);
+          s = '';
+          sender = null;
+        } else if (!sender) {
           sender = setTimeout(() => {
-            socket.send(s);
+            _send(s);
             s = '';
             sender = null;
           }, timeout);
         }
       };
     }
     // binary message buffering
-    function bufferUtf8(socket, timeout) {
+    function bufferUtf8(timeout, limit) {
       let buffer = [];
       let sender = null;
       let length = 0;
       return (data) => {
         buffer.push(data);
         length += data.length;
-        if (!sender) {
+        if (length > limit) {
+          clearTimeout(sender);
+          _send(Buffer.concat(buffer, length));
+          buffer = [];
+          sender = null;
+          length = 0;
+        } else if (!sender) {
           sender = setTimeout(() => {
-            socket.send(Buffer.concat(buffer, length));
+            _send(Buffer.concat(buffer, length));
             buffer = [];
             sender = null;
             length = 0;
           }, timeout);
         }
       };
     }
-    const send = USE_BINARY_UTF8 ? bufferUtf8(ws, 5) : buffer(ws, 5);
-
-    term.on('data', function(data) {
-      try {
-        send(data);
-      } catch (ex) {
-        // The WebSocket is not open, ignore
+    const send = (USE_BINARY_UTF8 ? bufferUtf8 : buffer)(MAX_SEND_INTERVAL, MAX_CHUNK_SIZE);
+
+    term.on('data', data => {
+      send(data);
+      sent += data.length;
+      if (sent > ACK_WATERMARK) {
+        ack_expected++;
+        sent -= ACK_WATERMARK;
+        if (ack_expected > MAX_ACK_DIFF) {
+          term.pause();
+        }
       }
     });
+
     ws.on('message', function(msg) {
+      //console.log([msg, sent]);
+      if (msg === FLOW_CONTROL_ACK) {
+        ack_expected--;
+        if (ack_expected <= MAX_ACK_DIFF) {
+          term.resume();
+        }
+        return;
+      }
       term.write(msg);
     });
     ws.on('close', function () {
       term.kill();
       console.log('Closed terminal ' + term.pid);
       // Clean things up
       delete terminals[term.pid];
-      delete logs[term.pid];
     });
   });
 

fast_producer.c

@@ -0,0 +1,29 @@
+#include <stdio.h>
+#include <unistd.h>
+#include <string.h>
+
+static char MSG[10][10] = {
+  { [0 ... 8] = '0', '\n' },
+  { [0 ... 8] = '1', '\n' },
+  { [0 ... 8] = '2', '\n' },
+  { [0 ... 8] = '3', '\n' },
+  { [0 ... 8] = '4', '\n' },
+  { [0 ... 8] = '5', '\n' },
+  { [0 ... 8] = '6', '\n' },
+  { [0 ... 8] = '7', '\n' },
+  { [0 ... 8] = '8', '\n' },
+  { [0 ... 8] = '9', '\n' },
+};
+static char ALL[60000];
+
+int main(int argc, char **argv) {
+  int i;
+  // fill 60kB buffer
+  for (i=0; i<600; ++i) {
+    memcpy(ALL+i*100, MSG, 100);
+  }
+  // copy buffer data as fast as possible (~6GB/s on my machine)
+  while (1) {
+    write(1, ALL, 60000);
+  }
+}