[feature]: zlib deflate concurrency limit

doc/ws.md

@@ -76,6 +76,10 @@ provided then that is extension parameters:
 - `memLevel` {Number} The value of zlib's `memLevel` param (1-9, default 8).
 - `threshold` {Number} Payloads smaller than this will not be compressed.
   Defaults to 1024 bytes.
+- `concurrencyLimit` {Number} The number of concurrent calls to zlib.
+  Calls above this limit will be queued. Default 10. You usually won't
+  need to touch this option. See [concurrency-limit][this issue] for more
+  details.
 
 If a property is empty then either an offered configuration or a default value
 is used.
@@ -425,4 +429,5 @@ Forcibly close the connection.
 
 The URL of the WebSocket server. Server clients don't have this attribute.
 
+[concurrency-limit]: https://github.com/websockets/ws/issues/1202
 [permessage-deflate]: https://tools.ietf.org/html/draft-ietf-hybi-permessage-compression-19

lib/PerMessageDeflate.js

@@ -2,6 +2,7 @@
 
 const safeBuffer = require('safe-buffer');
 const zlib = require('zlib');
+const Limiter = require('async-limiter');
 
 const bufferUtil = require('./BufferUtil');
 
@@ -10,6 +11,14 @@ const Buffer = safeBuffer.Buffer;
 const TRAILER = Buffer.from([0x00, 0x00, 0xff, 0xff]);
 const EMPTY_BLOCK = Buffer.from([0x00]);
 
+// We limit zlib concurrency, which prevents severe memory fragmentation
+// as documented in https://github.com/nodejs/node/issues/8871#issuecomment-250915913
+// and https://github.com/websockets/ws/issues/1202
+//
+// Intentionally global; it's the global thread pool that's
+// an issue.
+let zlibLimiter;
+
 /**
  * Per-message Deflate implementation.
  */
@@ -25,6 +34,13 @@ class PerMessageDeflate {
     this._inflate = null;
 
     this.params = null;
+
+    if (!zlibLimiter) {
+      const concurrency = this._options.concurrencyLimit !== undefined
+        ? this._options.concurrencyLimit
+        : 10;
+      zlibLimiter = new Limiter({ concurrency });
+    }
   }
 
   static get extensionName () {
@@ -249,14 +265,48 @@ class PerMessageDeflate {
   }
 
   /**
-   * Decompress data.
+   * Decompress data. Concurrency limited by async-limiter.
    *
    * @param {Buffer} data Compressed data
    * @param {Boolean} fin Specifies whether or not this is the last fragment
    * @param {Function} callback Callback
    * @public
    */
   decompress (data, fin, callback) {
+    zlibLimiter.push((done) => {
+      this._decompress(data, fin, (err, result) => {
+        done();
+        callback(err, result);
+      });
+    });
+  }
+
+  /**
+   * Compress data. Concurrency limited by async-limiter.
+   *
+   * @param {Buffer} data Data to compress
+   * @param {Boolean} fin Specifies whether or not this is the last fragment
+   * @param {Function} callback Callback
+   * @public
+   */
+  compress (data, fin, callback) {
+    zlibLimiter.push((done) => {
+      this._compress(data, fin, (err, result) => {
+        done();
+        callback(err, result);
+      });
+    });
+  }
+
+  /**
+   * Decompress data.
+   *
+   * @param {Buffer} data Compressed data
+   * @param {Boolean} fin Specifies whether or not this is the last fragment
+   * @param {Function} callback Callback
+   * @private
+   */
+  _decompress (data, fin, callback) {
     const endpoint = this._isServer ? 'client' : 'server';
 
     if (!this._inflate) {
@@ -322,9 +372,9 @@ class PerMessageDeflate {
    * @param {Buffer} data Data to compress
    * @param {Boolean} fin Specifies whether or not this is the last fragment
    * @param {Function} callback Callback
-   * @public
+   * @private
    */
-  compress (data, fin, callback) {
+  _compress (data, fin, callback) {
     if (!data || data.length === 0) {
       process.nextTick(callback, null, EMPTY_BLOCK);
       return;

package.json

@@ -26,6 +26,7 @@
     "lint": "eslint ."
   },
   "dependencies": {
+    "async-limiter": "~1.0.0",
     "safe-buffer": "~5.1.0",
     "ultron": "~1.1.0"
   },