Add SHARED and DEDICATED compressor options

misc/main.cpp

@@ -22,7 +22,7 @@ int main(int argc, char **argv) {
         res->end("Hello HTTP!");
     }).ws<PerSocketData>("/*", {
         /* Settings */
-        .compression = true,
+        .compression = uWS::DEDICATED_COMPRESSOR,
         .maxPayloadLength = 16 * 1024,
         /* Handlers */
         .open = [](auto *ws, auto *req) {

src/App.h

@@ -30,6 +30,16 @@
 
 namespace uWS {
 
+/* Compress options (really more like PerMessageDeflateOptions) */
+enum CompressOptions {
+    /* Compression disabled */
+    DISABLED = 0,
+    /* We compress using a shared non-sliding window. No added memory usage, worse compression. */
+    SHARED_COMPRESSOR = 1,
+    /* We compress using a dedicated sliding window. Major memory usage added, better compression of similarly repeated messages. */
+    DEDICATED_COMPRESSOR = 2
+};
+
 template <bool SSL>
 struct TemplatedApp : StaticDispatch<SSL> {
 private:
@@ -53,7 +63,7 @@ public:
     }
 
     struct WebSocketBehavior {
-        bool compression = false;
+        CompressOptions compression = DISABLED;
         int maxPayloadLength = 16 * 1024;
         std::function<void(uWS::WebSocket<SSL, true> *, HttpRequest *)> open = nullptr;
         std::function<void(uWS::WebSocket<SSL, true> *, std::string_view, uWS::OpCode)> message = nullptr;
@@ -70,14 +80,12 @@ public:
 
         /* If we are the first one to use compression, initialize it */
         if (behavior.compression) {
-
             LoopData *loopData = (LoopData *) us_loop_ext(static_dispatch(us_ssl_socket_context_loop, us_socket_context_loop)(webSocketContext->getSocketContext()));
 
-            if (!loopData->inflationStream) {
+            /* Initialize loop's deflate inflate streams */
+            if (!loopData->zlibContext) {
+                loopData->zlibContext = new ZlibContext;
                 loopData->inflationStream = new InflationStream;
-            }
-
-            if (!loopData->deflationStream) {
                 loopData->deflationStream = new DeflationStream;
             }
         }
@@ -102,14 +110,24 @@ public:
 
                 /* Negotiate compression */
                 bool perMessageDeflate = false;
-                if (behavior.compression) {
+                bool slidingDeflateWindow = false;
+                if (behavior.compression != DISABLED) {
                     std::string_view extensions = req->getHeader("sec-websocket-extensions");
                     if (extensions.length()) {
-                        // basically: parse<isServer>(options, extensions)
+                        /* We never support client context takeover (the client cannot compress with a sliding window). */
-                        ExtensionsNegotiator<true> extensionsNegotiator(PERMESSAGE_DEFLATE | CLIENT_NO_CONTEXT_TAKEOVER); // take options
+                        int wantedOptions = PERMESSAGE_DEFLATE | CLIENT_NO_CONTEXT_TAKEOVER;
+
+                        /* Shared compressor is the default */
+                        if (behavior.compression == SHARED_COMPRESSOR) {
+                            /* Disable per-socket compressor */
+                            wantedOptions |= SERVER_NO_CONTEXT_TAKEOVER;
+                        }
+
+                        /* isServer = true */
+                        ExtensionsNegotiator<true> extensionsNegotiator(wantedOptions);
                         extensionsNegotiator.readOffer(extensions);
 
-                        //std::cout << extensions << " => " << extensionsNegotiator.generateOffer() << std::endl;
+                        std::cout << extensions << " => " << extensionsNegotiator.generateOffer() << std::endl;
 
                         /* Todo: remove these mid string copies */
                         res->writeHeader("Sec-WebSocket-Extensions", extensionsNegotiator.generateOffer());
@@ -118,6 +136,11 @@ public:
                         if (extensionsNegotiator.getNegotiatedOptions() & PERMESSAGE_DEFLATE) {
                             perMessageDeflate = true;
                         }
+
+                        /* Is the server allowed to compress with a sliding window? */
+                        if (!(extensionsNegotiator.getNegotiatedOptions() & SERVER_NO_CONTEXT_TAKEOVER)) {
+                            slidingDeflateWindow = true;
+                        }
                     }
                 }
 
@@ -132,7 +155,7 @@ public:
                 webSocket->cork();
 
                 httpContext->upgradeToWebSocket(
-                            webSocket->init(perMessageDeflate)
+                            webSocket->init(perMessageDeflate, slidingDeflateWindow)
                             );
 
                 /* Emit open event */

src/LoopData.h

@@ -46,7 +46,8 @@ public:
     int corkOffset = 0;
     void *corkedSocket = nullptr;
 
-    /* Compression data */
+    /* Per message deflate data */
+    ZlibContext *zlibContext = nullptr;
     InflationStream *inflationStream = nullptr;
     DeflationStream *deflationStream = nullptr;
 };

src/PerMessageDeflate.h

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-// inflationStream? Ciompression
+/* This standalone module implements deflate / inflate streams */
 
 #ifndef PERMESSAGEDEFLATE_H
 #define PERMESSAGEDEFLATE_H
@@ -28,73 +28,73 @@
 
 #define LARGE_BUFFER_SIZE 16000 // fix this
 
-// we also need DeflationStream
+struct ZlibContext {
+    /* Any returned data is valid until next same-class call.
+     * We need to have two classes to allow inflation followed
+     * by many deflations without modifying the inflation */
+    std::string dynamicDeflationBuffer;
+    std::string dynamicInflationBuffer;
+    char *deflationBuffer;
+    char *inflationBuffer;
+
+    ZlibContext() {
+        deflationBuffer = (char *) malloc(LARGE_BUFFER_SIZE);
+        inflationBuffer = (char *) malloc(LARGE_BUFFER_SIZE);
+    }
+
+    ~ZlibContext() {
+        free(deflationBuffer);
+        free(inflationBuffer);
+    }
+};
 
 struct DeflationStream {
-
-    // share this under the Loop
-    std::string dynamicZlibBuffer;
     z_stream deflationStream = {};
-    char *zlibBuffer;
 
     DeflationStream() {
-        std::cout << "Constructing DeflationStream" << std::endl;
-        zlibBuffer = (char *) malloc(LARGE_BUFFER_SIZE);
-
         deflateInit2(&deflationStream, 1, Z_DEFLATED, -15, 8, Z_DEFAULT_STRATEGY);
     }
 
-    std::string_view deflate(std::string_view raw) {
+    /* Deflate and optionally reset */
+    std::string_view deflate(ZlibContext *zlibContext, std::string_view raw, bool reset) {
+        /* Odd place to clear this one, fix */
+        zlibContext->dynamicDeflationBuffer.clear();
 
-        // slidingDeflateWindow är input, length är in/ut
+        deflationStream.next_in = (Bytef *) raw.data();
+        deflationStream.avail_in = (unsigned int) raw.length();
 
-        z_stream *slidingDeflateWindow = nullptr;
+        /* This buffer size has to be at least 6 bytes for Z_SYNC_FLUSH to work */
-
-        dynamicZlibBuffer.clear();
-
-        z_stream *compressor = slidingDeflateWindow ? slidingDeflateWindow : &deflationStream;
-
-        compressor->next_in = (Bytef *) raw.data();
-        compressor->avail_in = (unsigned int) raw.length();
-
-        // note: zlib requires more than 6 bytes with Z_SYNC_FLUSH
         const int DEFLATE_OUTPUT_CHUNK = LARGE_BUFFER_SIZE;
 
         int err;
         do {
-            compressor->next_out = (Bytef *) zlibBuffer;
+            deflationStream.next_out = (Bytef *) zlibContext->deflationBuffer;
-            compressor->avail_out = DEFLATE_OUTPUT_CHUNK;
+            deflationStream.avail_out = DEFLATE_OUTPUT_CHUNK;
 
-            err = ::deflate(compressor, Z_SYNC_FLUSH);
+            err = ::deflate(&deflationStream, Z_SYNC_FLUSH);
-            if (Z_OK == err && compressor->avail_out == 0) {
+            if (Z_OK == err && deflationStream.avail_out == 0) {
-                dynamicZlibBuffer.append(zlibBuffer, DEFLATE_OUTPUT_CHUNK - compressor->avail_out);
+                zlibContext->dynamicDeflationBuffer.append(zlibContext->deflationBuffer, DEFLATE_OUTPUT_CHUNK - deflationStream.avail_out);
                 continue;
             } else {
                 break;
             }
         } while (true);
 
-        // note: should not change avail_out
+        /* This must not change avail_out */
-        if (!slidingDeflateWindow) {
+        if (reset) {
-            deflateReset(compressor);
+            deflateReset(&deflationStream);
         }
 
-        if (dynamicZlibBuffer.length()) {
+        if (zlibContext->dynamicDeflationBuffer.length()) {
-            dynamicZlibBuffer.append(zlibBuffer, DEFLATE_OUTPUT_CHUNK - compressor->avail_out);
+            zlibContext->dynamicDeflationBuffer.append(zlibContext->deflationBuffer, DEFLATE_OUTPUT_CHUNK - deflationStream.avail_out);
 
-            return {(char *) dynamicZlibBuffer.data(), dynamicZlibBuffer.length() - 4};
+            return {(char *) zlibContext->dynamicDeflationBuffer.data(), zlibContext->dynamicDeflationBuffer.length() - 4};
-
-            //length = dynamicZlibBuffer.length() - 4;
-            //return (char *) dynamicZlibBuffer.data();
         }
 
         return {
-            zlibBuffer,
+            zlibContext->deflationBuffer,
-            DEFLATE_OUTPUT_CHUNK - compressor->avail_out - 4
+            DEFLATE_OUTPUT_CHUNK - deflationStream.avail_out - 4
         };
-
-        //length = DEFLATE_OUTPUT_CHUNK - compressor->avail_out - 4;
-        //return zlibBuffer;
     }
 
     ~DeflationStream() {
@@ -102,54 +102,46 @@ struct DeflationStream {
     }
 };
 
-// the loop holds one of these
 struct InflationStream {
-
-    // share this under the Loop
-    std::string dynamicZlibBuffer;
     z_stream inflationStream = {};
-    char *zlibBuffer;
 
     InflationStream() {
-        std::cout << "Initliazing shared compression" << std::endl;
-        zlibBuffer = (char *) malloc(LARGE_BUFFER_SIZE);
-
         inflateInit2(&inflationStream, -15);
     }
 
-    std::string_view inflate(std::string_view compressed) {
+    std::string_view inflate(ZlibContext *zlibContext, std::string_view compressed) {
 
         int maxPayload = 160000; // todo: fix this
 
-        dynamicZlibBuffer.clear();
+        zlibContext->dynamicInflationBuffer.clear();
 
         inflationStream.next_in = (Bytef *) compressed.data();
         inflationStream.avail_in = (unsigned int) compressed.length();
 
         int err;
         do {
-            inflationStream.next_out = (Bytef *) zlibBuffer;
+            inflationStream.next_out = (Bytef *) zlibContext->inflationBuffer;
             inflationStream.avail_out = LARGE_BUFFER_SIZE;
             err = ::inflate(&inflationStream, Z_FINISH);
             if (!inflationStream.avail_in) {
                 break;
             }
 
-            dynamicZlibBuffer.append(zlibBuffer, LARGE_BUFFER_SIZE - inflationStream.avail_out);
+            zlibContext->dynamicInflationBuffer.append(zlibContext->inflationBuffer, LARGE_BUFFER_SIZE - inflationStream.avail_out);
-        } while (err == Z_BUF_ERROR && dynamicZlibBuffer.length() <= maxPayload);
+        } while (err == Z_BUF_ERROR && zlibContext->dynamicInflationBuffer.length() <= maxPayload);
 
         inflateReset(&inflationStream);
 
-        if ((err != Z_BUF_ERROR && err != Z_OK) || dynamicZlibBuffer.length() > maxPayload) {
+        if ((err != Z_BUF_ERROR && err != Z_OK) || zlibContext->dynamicInflationBuffer.length() > maxPayload) {
             return {nullptr, 0};
         }
 
-        if (dynamicZlibBuffer.length()) {
+        if (zlibContext->dynamicInflationBuffer.length()) {
-            dynamicZlibBuffer.append(zlibBuffer, LARGE_BUFFER_SIZE - inflationStream.avail_out);
+            zlibContext->dynamicInflationBuffer.append(zlibContext->inflationBuffer, LARGE_BUFFER_SIZE - inflationStream.avail_out);
-            return {dynamicZlibBuffer.data(), dynamicZlibBuffer.length()};
+            return {zlibContext->dynamicInflationBuffer.data(), zlibContext->dynamicInflationBuffer.length()};
         }
 
-        return {zlibBuffer, LARGE_BUFFER_SIZE - inflationStream.avail_out};
+        return {zlibContext->inflationBuffer, LARGE_BUFFER_SIZE - inflationStream.avail_out};
     }
 
 };

src/WebSocket.h

@@ -33,8 +33,8 @@ private:
     using SOCKET_TYPE = typename StaticDispatch<SSL>::SOCKET_TYPE;
     using StaticDispatch<SSL>::static_dispatch;
 
-    void *init(bool perMessageDeflate) {
+    void *init(bool perMessageDeflate, bool slidingCompression) {
-        new (static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) this)) WebSocketData(perMessageDeflate);
+        new (static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) this)) WebSocketData(perMessageDeflate, slidingCompression);
         return this;
     }
 public:
@@ -58,9 +58,13 @@ public:
 
             /* Check and correct the compress hint */
             if (opCode < 3 && webSocketData->compressionStatus == WebSocketData::ENABLED) {
-                // todo: shared deflate window
                 LoopData *loopData = Super::getLoopData();
-                message = loopData->deflationStream->deflate(message);
+                /* Compress using either shared or dedicated deflationStream */
+                if (webSocketData->deflationStream) {
+                    message = webSocketData->deflationStream->deflate(loopData->zlibContext, message, false);
+                } else {
+                    message = loopData->deflationStream->deflate(loopData->zlibContext, message, true);
+                }
             } else {
                 compress = false;
             }

src/WebSocketContext.h

@@ -101,7 +101,7 @@ private:
 
                                     );
 
-                        std::string_view inflatedFrame = loopData->inflationStream->inflate({data, length});
+                        std::string_view inflatedFrame = loopData->inflationStream->inflate(loopData->zlibContext, {data, length});
                         if (!inflatedFrame.length()) {
                             forceClose(webSocketState, s);
                             return true;
@@ -156,7 +156,7 @@ private:
                                         );
 
 
-                            std::string_view inflatedFrame = loopData->inflationStream->inflate({webSocketData->fragmentBuffer.data(), webSocketData->fragmentBuffer.length() - 4});
+                            std::string_view inflatedFrame = loopData->inflationStream->inflate(loopData->zlibContext, {webSocketData->fragmentBuffer.data(), webSocketData->fragmentBuffer.length() - 4});
                             if (!inflatedFrame.length()) {
                                 forceClose(webSocketState, s);
                                 return true;

src/WebSocketData.h

@@ -19,6 +19,7 @@
 
 #include "WebSocketProtocol.h"
 #include "AsyncSocketData.h"
+#include "PerMessageDeflate.h"
 
 #include <string>
 
@@ -37,10 +38,17 @@ private:
         ENABLED,
         COMPRESSED_FRAME
     } compressionStatus;
+
+    /* We might have a dedicated compressor */
+    DeflationStream *deflationStream = nullptr;
 public:
-    WebSocketData(bool perMessageDeflate) : WebSocketState<true>() {
+    WebSocketData(bool perMessageDeflate, bool slidingCompression) : WebSocketState<true>() {
-        //std::cout << "perMessageDeflate: " << perMessageDeflate << std::endl;
         compressionStatus = perMessageDeflate ? ENABLED : DISABLED;
+
+        /* Initialize the dedicated sliding window */
+        if (perMessageDeflate && slidingCompression) {
+            deflationStream = new DeflationStream;
+        }
     }
 };
 

src/WebSocketExtensions.h

@@ -129,16 +129,10 @@ std::string ExtensionsNegotiator<isServer>::generateOffer() {
             extensionsOffer += "; client_no_context_takeover";
         }
 
-        // we do not support accepting this yet
+        /* It is questionable sending this improves anything */
-        // todo: if we agree on this, do not allocate a compressor
+        /*if (options & Options::SERVER_NO_CONTEXT_TAKEOVER) {
-        // per socket!
+            extensionsOffer += "; server_no_context_takeover";
-
+        }*/
-        // It is RECOMMENDED that a server supports the
-        // "server_no_context_takeover" extension parameter in an extension
-        // negotiation offer.
-        if (options & Options::SERVER_NO_CONTEXT_TAKEOVER) {
-            //extensionsOffer += "; server_no_context_takeover";
-        }
     }
 
     return extensionsOffer;
@@ -153,11 +147,12 @@ void ExtensionsNegotiator<isServer>::readOffer(std::string_view offer) {
                 options |= CLIENT_NO_CONTEXT_TAKEOVER;
             }
 
+            /* We leave this option for us to read even if the client did not send it */
             if (extensionsParser.serverNoContextTakeover) {
                 options |= SERVER_NO_CONTEXT_TAKEOVER;
-            } else {
+            }/* else {
                 options &= ~SERVER_NO_CONTEXT_TAKEOVER;
-            }
+            }*/
         } else {
             options &= ~PERMESSAGE_DEFLATE;
         }