feat(udp): optional per-endpoint ZSTD compression

.github/workflows/main.yml

@@ -43,10 +43,10 @@ jobs:
           if [ "${{ matrix.name }}" == "i686" ]; then
               sudo dpkg --add-architecture i386
               sudo apt-get update
-              sudo apt-get install --yes libgtest-dev:i386
+              sudo apt-get install --yes libgtest-dev:i386 libzstd-dev:i386
           else
               sudo apt-get update
-              sudo apt-get install --yes libgtest-dev
+              sudo apt-get install --yes libgtest-dev libzstd-dev
           fi
           sudo apt-get install --yes pkg-config gcc-multilib g++-multilib ninja-build python3-pip
           pip3 install meson pymavlink psutil
@@ -115,7 +115,7 @@ jobs:
     container: alpine:3.16
     steps:
       - name: install dependencies
-        run: apk update && apk add build-base git linux-headers pkgconf meson ninja
+        run: apk update && apk add build-base git linux-headers pkgconf meson ninja zstd-dev
 
       - uses: actions/checkout@v4
         with:

README.md

@@ -201,6 +201,61 @@ Defining endpoints:
     * (This means that UART, UDP and TCP client endpoints are never destroyed
       during runtime.)
 
+### ZSTD Compression
+A UDP endpoint can optionally ZSTD-compress its traffic, to save bandwidth on a
+constrained link (e.g. a telemetry radio). It is configured per endpoint:
+
+```ini
+[UdpEndpoint radio]
+Mode = Normal
+Address = 127.0.0.1
+Port = 13336
+ZstdCompression = true
+ZstdDictionary  = /usr/share/mavlink-router/mavlink_zstd.dict
+```
+
+  - `ZstdCompression` (bool, default `false`): enable compression on this
+    endpoint.
+  - `ZstdDictionary` (path, optional): a trained Zstandard dictionary to improve
+    the ratio on small MAVLink messages. Without it, compression still works but
+    is less effective.
+
+Behavior:
+
+  - Each outgoing datagram (one coalesced batch of framed MAVLink messages) is
+    compressed into a single standalone ZSTD frame (level 3).
+  - Time-critical messages (HEARTBEAT, SYSTEM_TIME, PING, MISSION_CURRENT,
+    COMMAND_LONG, COMMAND_ACK, TIMESYNC, CURRENT_EVENT_SEQUENCE, REQUEST_EVENT),
+    and any datagram that does not get smaller, are sent uncompressed instead.
+  - On receive, each datagram is auto-detected by its ZSTD magic and
+    decompressed; plain MAVLink is passed through unchanged. There is no
+    negotiation, so a mixed stream decodes correctly.
+
+Both ends of the link must enable `ZstdCompression`, and — if used — must be
+configured with the **same** `ZstdDictionary`.
+
+#### Generating the dictionary
+The dictionary is trained from a representative sample of the MAVLink traffic
+that will cross the link (ideally captured from real flights). Use
+[`tools/train-zstd-dict.sh`](tools/train-zstd-dict.sh), which needs `zstd` and
+`tshark` (and `tcpdump` for live capture):
+
+```sh
+# From a folder of pcaps (recommended; auto-detects the MAVLink port):
+tools/train-zstd-dict.sh --folder /path/to/pcaps 0 mavlink_zstd.dict
+
+# From a single pcap:
+tools/train-zstd-dict.sh --pcap capture.pcap 0 mavlink_zstd.dict
+
+# Or capture live on a port for N seconds, then train:
+tools/train-zstd-dict.sh 14550 120 mavlink_zstd.dict
+```
+
+It extracts each UDP payload as an individual sample file (zstd trains better on
+many small samples) and runs `zstd --train --maxdict=16384` to produce a 16 KiB
+dictionary (override with the `DICT_SIZE` env var). Install the result to the
+path referenced by `ZstdDictionary` and deploy the identical file to both ends.
+
 ### Message Routing
 In general, each message received on one endpoint is delivered to all endpoints
 in which that target system/component has been seen. If it's a broadcast

meson.build

@@ -27,6 +27,12 @@ dep_math = cxx.find_library('m')
 dep_rt = cxx.find_library('rt')
 dep_thread = dependency('threads')
 
+# Optional: ZSTD compression for UDP endpoints. Auto-enabled when libzstd is
+# available; compiled out cleanly otherwise (the ZstdCompression option then
+# fails at startup with a clear message).
+dep_zstd = dependency('libzstd', required: get_option('zstd'))
+have_zstd = dep_zstd.found()
+
 # Optional dependencies
 systemd_system_unit_dir = get_option('systemdsystemunitdir')
 if systemd_system_unit_dir == 'auto'
@@ -111,6 +117,8 @@ conf.set10('HAVE_WADDRESS_OF_PACKED_MEMBER', has_waddress_of_packed_member)
 has_aio_init_symbols = cxx.has_header_symbol('aio.h', 'aio_init')
 conf.set10('HAVE_DECL_AIO_INIT', has_aio_init_symbols)
 
+conf.set10('HAVE_ZSTD', have_zstd)
+
 # Always include config.h
 config_h = configure_file(output : 'config.h', configuration : conf)
 add_project_arguments('-include', 'config.h', language : 'cpp')

meson_options.txt

@@ -2,3 +2,7 @@ option('systemdsystemunitdir',
        description: 'systemd unit directory',
        type: 'string',
        value: 'auto')
+option('zstd',
+       description: 'ZSTD compression support for UDP endpoints (needs libzstd)',
+       type: 'feature',
+       value: 'auto')

src/endpoint.cpp

@@ -48,7 +48,9 @@
 
 #include "mainloop.h"
 
-#define RX_BUF_MAX_SIZE (MAVLINK_MAX_PACKET_LEN * 4)
+/* Large enough that one decompressed UDP datagram (a whole coalesced batch, up
+ * to TX_BUF_MAX_SIZE) fits when a ZstdCompression endpoint inflates it. */
+#define RX_BUF_MAX_SIZE (8U * 1024U)
 #define TX_BUF_MAX_SIZE (8U * 1024U)
 
 #define UART_BAUD_RETRY_SEC 5
@@ -101,6 +103,8 @@ const ConfFile::OptionsTable UdpEndpoint::option_table[] = {
     {"CoalesceMs",      false,  ConfFile::parse_ul,             OPTIONS_TABLE_STRUCT_FIELD(UdpEndpointConfig, coalesce_ms)},
     {"CoalesceNoDelay", false,  ConfFile::parse_uint32_vector,  OPTIONS_TABLE_STRUCT_FIELD(UdpEndpointConfig, coalesce_nodelay)},
     {"MsgThrottling",   false,  ConfFile::parse_pair_vector,    OPTIONS_TABLE_STRUCT_FIELD(UdpEndpointConfig, message_throttling)},
+    {"ZstdCompression", false,  ConfFile::parse_bool,           OPTIONS_TABLE_STRUCT_FIELD(UdpEndpointConfig, zstd_compression)},
+    {"ZstdDictionary",  false,  ConfFile::parse_stdstring,      OPTIONS_TABLE_STRUCT_FIELD(UdpEndpointConfig, zstd_dictionary)},
     {}
 };
 
@@ -1237,6 +1241,21 @@ bool UdpEndpoint::setup(UdpEndpointConfig conf)
         this->set_message_throttling(msg_id, throttle_cfg.second);
     }
 
+    // Per-datagram ZSTD compression. When enabled, every outgoing datagram is
+    // compressed into a standalone frame and every incoming one is decompressed
+    // if it carries the ZSTD magic.
+    if (conf.zstd_compression) {
+        _zstd_codec = std::make_unique<ZstdCodec>();
+        if (!_zstd_codec->init(conf.zstd_dictionary)) {
+            log_error("UDP %s: failed to initialise ZSTD codec", conf.name.c_str());
+            _zstd_codec.reset();
+            return false;
+        }
+        log_info("UDP %s: ZSTD compression enabled (dictionary: %s)",
+                 conf.name.c_str(),
+                 conf.zstd_dictionary.empty() ? "none" : conf.zstd_dictionary.c_str());
+    }
+
     return true;
 }
 
@@ -1429,6 +1448,17 @@ ssize_t UdpEndpoint::_read_msg(uint8_t *buf, size_t len)
         return -errno;
     }
 
+    // Decompress the datagram in place if this endpoint is compressed and the
+    // payload carries the ZSTD magic (plain MAVLink is passed through). One UDP
+    // datagram is exactly one ZSTD frame, so no cross-datagram reassembly.
+    if (_zstd_codec && r > 0) {
+        ssize_t d = _zstd_codec->inflate(buf, (size_t)r, len);
+        if (d < 0) {
+            return 0; // corrupt / oversized frame -> drop this datagram
+        }
+        r = d;
+    }
+
     // Update timeout
     if (nomessage_timeout) {
         Mainloop::get_instance().mod_timeout(nomessage_timeout, 5 * MSEC_PER_SEC);
@@ -1523,7 +1553,21 @@ int UdpEndpoint::flush_pending_msgs()
         return 0;
     }
 
-    ssize_t r = ::sendto(fd, tx_buf.data, tx_buf.len, 0, sock, addrlen);
+    // Compress the whole coalesced batch into one standalone ZSTD frame. The
+    // codec returns nullptr (send plain) for skip-listed messages or when
+    // compression would not shrink the datagram.
+    const uint8_t *out_data = tx_buf.data;
+    size_t out_len = tx_buf.len;
+    if (_zstd_codec) {
+        size_t clen = 0;
+        const uint8_t *c = _zstd_codec->compress(tx_buf.data, tx_buf.len, &clen);
+        if (c != nullptr) {
+            out_data = c;
+            out_len = clen;
+        }
+    }
+
+    ssize_t r = ::sendto(fd, out_data, out_len, 0, sock, addrlen);
     if (r == -1) {
         if (errno != EAGAIN && errno != ECONNREFUSED && errno != ENETUNREACH) {
             log_error("UDP %s: Error sending udp packet (%m)", _name.c_str());
@@ -1534,9 +1578,9 @@ int UdpEndpoint::flush_pending_msgs()
     _stat.write.total++;
     _stat.write.bytes += r;
 
-    tx_buf.len = std::max(ssize_t(0), (ssize_t)tx_buf.len - r);
-    // memcpy isn't safe for overlapping regions
-    memmove(tx_buf.data, &tx_buf.data[r], tx_buf.len);
+    // UDP sendto is atomic: on success the whole datagram (compressed or not)
+    // left in one packet, so the tx buffer is now fully consumed.
+    tx_buf.len = 0;
 
     log_trace("UDP [%d]%s: Wrote %zd bytes", fd, _name.c_str(), r);
 

src/endpoint.h

@@ -32,6 +32,7 @@
 #include "comm.h"
 #include "pollable.h"
 #include "timeout.h"
+#include "zstd_codec.h"
 
 #define DEFAULT_BAUDRATE 115200U
 
@@ -85,6 +86,8 @@ struct UdpEndpointConfig {
     unsigned long coalesce_ms;
     std::vector<uint32_t> coalesce_nodelay;
     std::vector<std::pair<float, float>> message_throttling;
+    bool zstd_compression{false};
+    std::string zstd_dictionary;
 };
 
 struct TcpEndpointConfig {
@@ -409,6 +412,10 @@ class UdpEndpoint : public Endpoint {
     struct sockaddr_in sockaddr;
     struct sockaddr_in6 sockaddr6;
     std::set<uint32_t> _coalesce_nodelay{}; // immediately send if a mavlink msg_id is in this set
+
+    // Per-datagram ZSTD compression on this endpoint's link (null when
+    // ZstdCompression is off for this endpoint).
+    std::unique_ptr<ZstdCodec> _zstd_codec;
 };
 
 class TcpEndpoint : public Endpoint {

src/meson.build

@@ -6,6 +6,7 @@ exe_mavlink_router = executable('mavlink-routerd',
          'binlog.cpp',
          'dedup.cpp',
          'endpoint.cpp',
+         'zstd_codec.cpp',
          'git_version.cpp',
          'logendpoint.cpp',
          'main.cpp',
@@ -20,6 +21,7 @@ exe_mavlink_router = executable('mavlink-routerd',
         dependencies: [
                 dep_math,
                 dep_rt,
+            dep_zstd,
                 dep_thread,
         ],
         link_with: libcommon_private,
@@ -33,6 +35,7 @@ if dep_gtest.found()
             'binlog.cpp',
             'dedup.cpp',
             'endpoint.cpp',
+            'zstd_codec.cpp',
             'logendpoint.cpp',
             'mainloop.cpp',
             'mainloop_test.cpp',
@@ -46,6 +49,7 @@ if dep_gtest.found()
           dependencies : [
             dep_gtest,
             dep_rt,
+            dep_zstd,
           ],
           link_with: libcommon_private,
           install: false,
@@ -59,6 +63,7 @@ if dep_gtest.found()
             'binlog.cpp',
             'dedup.cpp',
             'endpoint.cpp',
+            'zstd_codec.cpp',
             'endpoints_test.cpp',
             'logendpoint.cpp',
             'mainloop.cpp',
@@ -72,6 +77,7 @@ if dep_gtest.found()
           dependencies : [
             dep_gtest,
             dep_rt,
+            dep_zstd,
           ],
           link_with: libcommon_private,
           install: false,