gui: compress ws messages

Author: jherrera-jump

book/api/websocket.md

@@ -38,6 +38,36 @@ file.
 The API is split into various topics which will be streamed to any and
 all connected clients.
 
+## Compression
+If configured properly, the server can optionally compress messages
+larger than 200 bytes. In order to enable this feature, the client must
+specify the `compress-zstd` subprotocol in the opening websocket
+handshake.
+
+::: code-group
+
+```js [client.js]
+client = new WebSocket("ws://localhost:80/websocket", protocols=['compress-zstd']);
+client.binaryType = "arraybuffer";
+```
+
+```
+ws.onmessage = function onmessage(ev: MessageEvent<unknown>) {
+  if (typeof ev.data === 'string') {
+    ... parse string
+  } else if (ev.data instanceof ArrayBuffer) {
+    ... decompress then parse
+  }
+};
+```
+
+:::
+
+In order to distinguish between compressed and non-compressed messages,
+the server will send compressed messages as a binary websocket frame
+(i.e. opcode=0x2) and regular messages as a text websocket frame (i.e.
+opcode=0x1).
+
 ## Keeping Up
 The server does not drop information, slow down, or stop publishing the
 stream of information if the client cannot keep up. A client that is

src/app/fdctl/config/default.toml

@@ -1747,3 +1747,13 @@ dynamic_port_range = "8900-9000"
     # production networking stack.
     [development.udpecho]
         affinity = "auto"
+
+    # Experimental options for the gui tile.  These should not be
+    # changed on a production validator.
+    [development.gui]
+        # Enables backend support for websocket compression.  Websocket
+        # messages typically consist of a single websocket text frame
+        # with a header and a json payload.  Enabling this option will
+        # make the backend compress large json payloads with ZSTD stream
+        # compression, sending the compressed payload in a binary frame.
+        websocket_compression = false

src/app/fdctl/topology.c

@@ -510,6 +510,7 @@ fd_topo_initialize( config_t * config ) {
       tile->gui.max_http_request_length   = config->tiles.gui.max_http_request_length;
       tile->gui.send_buffer_size_mb       = config->tiles.gui.send_buffer_size_mb;
       tile->gui.schedule_strategy         = config->tiles.pack.schedule_strategy_enum;
+      tile->gui.websocket_compression     = config->development.gui.websocket_compression;
     } else if( FD_UNLIKELY( !strcmp( tile->name, "plugin" ) ) ) {
 
     } else {

src/app/firedancer/config/default.toml

@@ -1573,3 +1573,13 @@ user = ""
     # production networking stack.
     [development.udpecho]
         affinity = "auto"
+
+    # Experimental options for the gui tile.  These should not be
+    # changed on a production validator.
+    [development.gui]
+        # Enables backend support for websocket compression.  Websocket
+        # messages typically consist of a single websocket text frame
+        # with a header and a json payload.  Enabling this option will
+        # make the backend compress large json payloads with ZSTD stream
+        # compression, sending the compressed payload in a binary frame.
+        websocket_compression = false

src/app/shared/fd_config.h

@@ -329,6 +329,10 @@ struct fd_config {
     struct {
       char affinity[ AFFINITY_SZ ];
     } snapshot_load;
+
+    struct {
+      int websocket_compression;
+    } gui;
   } development;
 
   struct {

src/app/shared/fd_config_parse.c

@@ -316,6 +316,8 @@ fd_config_extract_pod( uchar *       pod,
 
   CFG_POP      ( cstr,   development.udpecho.affinity                     );
 
+  CFG_POP      ( bool,   development.gui.websocket_compression            );
+
   if( FD_UNLIKELY( config->is_firedancer ) ) {
     if( FD_UNLIKELY( !fd_config_extract_podf( pod, &config->firedancer ) ) ) return NULL;
     fd_config_check_configf( config, &config->firedancer );

src/disco/gui/bandwidth.py

@@ -3,21 +3,28 @@
 import time
 import json
 from collections import defaultdict
+import zstandard as zstd
 
 async def measure_bandwidth(uri):
-    async with websockets.connect(uri, max_size=1_000_000_000) as websocket:
+    dcmp = zstd.ZstdDecompressor()
+    async with websockets.connect(uri, max_size=1_000_000_000, subprotocols=['compress-zstd']) as websocket:
         start_time = time.time()
         total_bytes_by_group = defaultdict(int)
         overall_total_bytes = 0
 
         while True:
             frame = await websocket.recv()
-            data = json.loads(frame)
+            if(isinstance(frame, bytes)):
+                frame_sz = len(frame)
+                data = json.loads(dcmp.stream_reader(frame).readall())
+            else:
+                frame_sz = len(frame.encode('utf-8'))
+                data = json.loads(frame)
             topic = data.get("topic")
             key = data.get("key")
             group = (topic, key)
-            total_bytes_by_group[group] += len(frame)
-            overall_total_bytes += len(frame)
+            total_bytes_by_group[group] += frame_sz
+            overall_total_bytes += frame_sz
             elapsed_time = time.time() - start_time
 
             if elapsed_time >= 1.0:
@@ -26,17 +33,17 @@ async def measure_bandwidth(uri):
                     bandwidth_mbps = (total_bytes * 8) / (elapsed_time * 1_000_000)
                     if bandwidth_mbps >= 0.001:
                         bandwidths.append((group, bandwidth_mbps))
-                
+
                 # Sort by bandwidth in descending order
                 bandwidths.sort(key=lambda x: x[1], reverse=True)
-                
+
                 for group, bandwidth_mbps in bandwidths:
-                    print(f"Incoming bandwidth for {group}: {bandwidth_mbps:.2f} Mbps")
-                
+                    print(f"Incoming bandwidth for {str(group):50}: {bandwidth_mbps:6.2f} Mbps")
+
                 # Calculate and print the overall total bandwidth
                 overall_bandwidth_mbps = (overall_total_bytes * 8) / (elapsed_time * 1_000_000)
                 print(f"Total incoming bandwidth: {overall_bandwidth_mbps:.2f} Mbps")
-                
+
                 start_time = time.time()
                 total_bytes_by_group.clear()
                 overall_total_bytes = 0

src/disco/gui/dist/LICENSE_DEPENDENCIES

@@ -2690,6 +2690,45 @@ THE SOFTWARE.
 
 ---
 
+Name: @oneidentity/zstd-js
+Version: 1.0.3
+License: SEE LICENSE IN LICENSE
+Private: false
+Description: Browser side compression library from the official Zstandard library.
+Repository: git+https://github.com/OneIdentity/zstd-js.git
+Homepage: https://github.com/OneIdentity/zstd-js
+Author: Zsombor Szende - @szendezsombor
+Contributors:
+  Csaba Tamás - @tamascsaba
+  Gergely Szabó - @szaboge
+  László Zana - @zanalaci
+License Copyright:
+===
+
+MIT License
+
+Copyright (c) 2022 One Identity
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+
+---
+
 Name: @ag-grid-community/core
 Version: 32.3.4
 License: MIT