updates to various PDU handling functionality;

configs/fne-config.example.yml

@@ -411,6 +411,28 @@ vtun:
     # Broadcast address of the tunnel network interface
     broadcast: 192.168.1.255
 
+    # VTUN ingress queue caps (drop oldest frame on overflow)
+    queue:
+        # Maximum number of queued VTUN frames waiting for air dispatch
+        maxFrames: 128
+        # Maximum total queued VTUN payload bytes
+        maxBytes: 262144
+
+    #
+    # Optional Linux qdisc setup for the VTUN interface.
+    #
+    qdisc:
+        # Enable automatic qdisc apply when VTUN is brought up.
+        enable: false
+        # qdisc kind (e.g. fq_codel, cake, tbf)
+        kind: fq_codel
+        # Optional qdisc args. Currently parsed for fq_codel.
+        # Supported keys: limit, flows, target, interval, ecn,
+        #   quantum, memory_limit, drop_batch_size, ce_threshold.
+        # Example:
+        # args: "limit=1024 flows=1024 target=5ms interval=100ms ecn=1"
+        args: ""
+
     #
     # P25 SNDCP Dynamic IP Allocation
     #

src/common/network/viface/VIFace.cpp

@@ -4,7 +4,7 @@
  * GPLv2 Open Source. Use is subject to license terms.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
- *  Copyright (C) 2024 Bryan Biedenkapp, N2PLL
+ *  Copyright (C) 2024,2026 Bryan Biedenkapp, N2PLL
  *
  */
 #if !defined(_WIN32)
@@ -21,10 +21,12 @@ using namespace network::viface;
 #include <fstream>
 #include <iomanip>
 #include <random>
+#include <unordered_map>
 
 #include <cassert>
 #include <cstring>
 #include <cerrno>
+#include <cctype>
 
 #include <unistd.h>
 #include <fcntl.h>
@@ -37,6 +39,8 @@ using namespace network::viface;
 #include <linux/if.h>
 #include <linux/if_tun.h>
 #include <linux/if_arp.h>
+#include <linux/rtnetlink.h>
+#include <linux/pkt_sched.h>
 #include <ifaddrs.h>
 
 // ---------------------------------------------------------------------------
@@ -221,6 +225,229 @@ void readVIFlags(int sockfd, std::string name, struct ifreq& ifr)
     }
 }
 
+/**
+ * @brief Helper to append a netlink attribute.
+ * @param n Netlink message header.
+ * @param maxLen Maximum message buffer length.
+ * @param type Attribute type.
+ * @param data Attribute payload.
+ * @param len Attribute payload length.
+ * @returns bool True, if attribute appended, otherwise false.
+ */
+bool nlAddAttr(struct nlmsghdr* n, size_t maxLen, int type, const void* data, size_t len)
+{
+    assert(n != nullptr);
+
+    const size_t attrLen = RTA_LENGTH(len);
+    const size_t newLen = NLMSG_ALIGN(n->nlmsg_len) + RTA_ALIGN(attrLen);
+    if (newLen > maxLen) {
+        return false;
+    }
+
+    struct rtattr* rta = (struct rtattr*)(((uint8_t*)n) + NLMSG_ALIGN(n->nlmsg_len));
+    rta->rta_type = type;
+    rta->rta_len = (uint16_t)attrLen;
+    if (len > 0U && data != nullptr) {
+        ::memcpy(RTA_DATA(rta), data, len);
+    }
+
+    n->nlmsg_len = (uint32_t)newLen;
+    return true;
+}
+
+/**
+ * @brief Helper to start a nested netlink attribute.
+ * @param n Netlink message header.
+ * @param maxLen Maximum message buffer length.
+ * @param type Attribute type.
+ * @returns rtattr* Nested attribute pointer or nullptr on failure.
+ */
+struct rtattr* nlNestStart(struct nlmsghdr* n, size_t maxLen, int type)
+{
+    struct rtattr* nest = (struct rtattr*)(((uint8_t*)n) + NLMSG_ALIGN(n->nlmsg_len));
+    if (!nlAddAttr(n, maxLen, type, nullptr, 0U)) {
+        return nullptr;
+    }
+    return nest;
+}
+
+/**
+ * @brief Helper to finalize a nested netlink attribute.
+ * @param n Netlink message header.
+ * @param nest Nested attribute pointer.
+ */
+void nlNestEnd(struct nlmsghdr* n, struct rtattr* nest)
+{
+    if (n == nullptr || nest == nullptr) {
+        return;
+    }
+
+    nest->rta_len = (uint16_t)(((uint8_t*)n + NLMSG_ALIGN(n->nlmsg_len)) - (uint8_t*)nest);
+}
+
+/**
+ * @brief Helper to lowercase a string.
+ * @param value Input string.
+ * @returns std::string Lowercased string.
+ */
+std::string toLowerStr(const std::string& value)
+{
+    std::string out = value;
+    for (char& c : out) {
+        c = (char)std::tolower((unsigned char)c);
+    }
+    return out;
+}
+
+/**
+ * @brief Helper to parse a size string with optional k/m/g suffix into bytes.
+ * @param value Input string.
+ * @param outBytes Parsed byte value.
+ * @returns bool True on success.
+ */
+bool parseSizeBytes(const std::string& value, uint32_t& outBytes)
+{
+    if (value.empty()) {
+        return false;
+    }
+
+    std::string lower = toLowerStr(value);
+    uint64_t mult = 1U;
+    if (lower.size() > 1U && lower.back() == 'b') {
+        lower.pop_back();
+    }
+
+    if (!lower.empty()) {
+        char sfx = lower.back();
+        if (sfx == 'k' || sfx == 'm' || sfx == 'g') {
+            lower.pop_back();
+            if (sfx == 'k') mult = 1024ULL;
+            if (sfx == 'm') mult = 1024ULL * 1024ULL;
+            if (sfx == 'g') mult = 1024ULL * 1024ULL * 1024ULL;
+        }
+    }
+
+    if (lower.empty()) {
+        return false;
+    }
+
+    char* end = nullptr;
+    errno = 0;
+    unsigned long long base = strtoull(lower.c_str(), &end, 10);
+    if (errno != 0 || end == nullptr || *end != '\0') {
+        return false;
+    }
+
+    uint64_t total = base * mult;
+    if (total > 0xFFFFFFFFULL) {
+        return false;
+    }
+
+    outBytes = (uint32_t)total;
+    return true;
+}
+
+/**
+ * @brief Helper to parse a duration string into microseconds.
+ * @param value Input value (supports us, ms, s).
+ * @param outUsec Parsed microseconds.
+ * @returns bool True on success.
+ */
+bool parseDurationUsec(const std::string& value, uint32_t& outUsec)
+{
+    if (value.empty()) {
+        return false;
+    }
+
+    std::string lower = toLowerStr(value);
+    uint64_t mult = 1U;
+
+    if (lower.size() >= 2U && lower.substr(lower.size() - 2U) == "us") {
+        mult = 1U;
+        lower = lower.substr(0U, lower.size() - 2U);
+    }
+    else if (lower.size() >= 2U && lower.substr(lower.size() - 2U) == "ms") {
+        mult = 1000U;
+        lower = lower.substr(0U, lower.size() - 2U);
+    }
+    else if (!lower.empty() && lower.back() == 's') {
+        mult = 1000000U;
+        lower.pop_back();
+    }
+
+    if (lower.empty()) {
+        return false;
+    }
+
+    char* end = nullptr;
+    errno = 0;
+    unsigned long long base = strtoull(lower.c_str(), &end, 10);
+    if (errno != 0 || end == nullptr || *end != '\0') {
+        return false;
+    }
+
+    uint64_t total = base * mult;
+    if (total > 0xFFFFFFFFULL) {
+        return false;
+    }
+
+    outUsec = (uint32_t)total;
+    return true;
+}
+
+/**
+ * @brief Parses qdisc args string into key/value pairs.
+ * Supports "key=value" and "key value" forms.
+ * @param args Input args string.
+ * @returns unordered_map Parsed key/value map.
+ */
+std::unordered_map<std::string, std::string> parseQDiscArgs(const std::string& args)
+{
+    std::unordered_map<std::string, std::string> out;
+    if (args.empty()) {
+        return out;
+    }
+
+    std::vector<std::string> tokens;
+    std::string token;
+    for (char c : args) {
+        if (std::isspace((unsigned char)c)) {
+            if (!token.empty()) {
+                tokens.push_back(token);
+                token.clear();
+            }
+        }
+        else {
+            token.push_back(c);
+        }
+    }
+    if (!token.empty()) {
+        tokens.push_back(token);
+    }
+
+    for (size_t i = 0U; i < tokens.size(); i++) {
+        std::string t = tokens[i];
+        size_t eq = t.find('=');
+        if (eq != std::string::npos) {
+            std::string k = toLowerStr(t.substr(0U, eq));
+            std::string v = t.substr(eq + 1U);
+            if (!k.empty() && !v.empty()) {
+                out[k] = v;
+            }
+            continue;
+        }
+
+        if ((i + 1U) < tokens.size()) {
+            std::string k = toLowerStr(t);
+            std::string v = tokens[i + 1U];
+            out[k] = v;
+            i++;
+        }
+    }
+
+    return out;
+}
+
 /**
  * @brief 
  * @param name 
@@ -697,6 +924,182 @@ uint32_t VIFace::getMTU() const
     return ifr.ifr_mtu;
 }
 
+/* Applies a root qdisc to the virtual interface using rtnetlink. */
+
+bool VIFace::setQDisc(std::string kind, std::string args)
+{
+    if (kind.empty()) {
+        LogError(LOG_NET, "Invalid qdisc kind for virtual interface %s", m_name.c_str());
+        return false;
+    }
+
+    kind = toLowerStr(kind);
+
+    struct ifreq ifr;
+    ::memset(&ifr, 0x00U, sizeof(struct ifreq));
+    ::strncpy(ifr.ifr_name, m_name.c_str(), IFNAMSIZ - 1);
+
+    if (::ioctl(m_ksFd, SIOCGIFINDEX, &ifr) != 0) {
+        LogError(LOG_NET, "Unable to get interface index for qdisc on %s, err: %d (%s)",
+            m_name.c_str(), errno, strerror(errno));
+        return false;
+    }
+
+    int nl = ::socket(AF_NETLINK, SOCK_RAW, NETLINK_ROUTE);
+    if (nl < 0) {
+        LogError(LOG_NET, "Unable to open rtnetlink socket for qdisc on %s, err: %d (%s)",
+            m_name.c_str(), errno, strerror(errno));
+        return false;
+    }
+
+    struct {
+        struct nlmsghdr n;
+        struct tcmsg t;
+        uint8_t buf[512U];
+    } req;
+    ::memset(&req, 0x00U, sizeof(req));
+
+    req.n.nlmsg_len = NLMSG_LENGTH(sizeof(struct tcmsg));
+    req.n.nlmsg_type = RTM_NEWQDISC;
+    req.n.nlmsg_flags = NLM_F_REQUEST | NLM_F_ACK | NLM_F_CREATE | NLM_F_REPLACE;
+    req.n.nlmsg_seq = 1U;
+
+    req.t.tcm_family = AF_UNSPEC;
+    req.t.tcm_ifindex = ifr.ifr_ifindex;
+    req.t.tcm_parent = TC_H_ROOT;
+    req.t.tcm_handle = 0U;
+
+    if (!nlAddAttr(&req.n, sizeof(req), TCA_KIND, kind.c_str(), kind.length() + 1U)) {
+        LogError(LOG_NET, "Unable to append qdisc kind attribute for %s", m_name.c_str());
+        ::close(nl);
+        return false;
+    }
+
+    if (!args.empty() && kind == "fq_codel") {
+        std::unordered_map<std::string, std::string> opts = parseQDiscArgs(args);
+        struct rtattr* optsNest = nlNestStart(&req.n, sizeof(req), TCA_OPTIONS);
+        if (optsNest == nullptr) {
+            LogError(LOG_NET, "Unable to append qdisc options for %s", m_name.c_str());
+            ::close(nl);
+            return false;
+        }
+
+        for (const auto& kv : opts) {
+            const std::string& key = kv.first;
+            const std::string& value = kv.second;
+
+            if (key == "limit") {
+                uint32_t v = 0U;
+                if (parseSizeBytes(value, v)) {
+                    (void)nlAddAttr(&req.n, sizeof(req), TCA_FQ_CODEL_LIMIT, &v, sizeof(v));
+                }
+            }
+            else if (key == "flows") {
+                uint32_t v = 0U;
+                if (parseSizeBytes(value, v)) {
+                    (void)nlAddAttr(&req.n, sizeof(req), TCA_FQ_CODEL_FLOWS, &v, sizeof(v));
+                }
+            }
+            else if (key == "quantum") {
+                uint32_t v = 0U;
+                if (parseSizeBytes(value, v)) {
+                    (void)nlAddAttr(&req.n, sizeof(req), TCA_FQ_CODEL_QUANTUM, &v, sizeof(v));
+                }
+            }
+            else if (key == "memory_limit") {
+                uint32_t v = 0U;
+                if (parseSizeBytes(value, v)) {
+#if defined(TCA_FQ_CODEL_MEMORY_LIMIT)
+                    (void)nlAddAttr(&req.n, sizeof(req), TCA_FQ_CODEL_MEMORY_LIMIT, &v, sizeof(v));
+#else
+                    LogWarning(LOG_NET, "fq_codel memory_limit not supported by kernel headers, interface %s", m_name.c_str());
+#endif
+                }
+            }
+            else if (key == "drop_batch_size") {
+                uint32_t v = 0U;
+                if (parseSizeBytes(value, v)) {
+#if defined(TCA_FQ_CODEL_DROP_BATCH_SIZE)
+                    (void)nlAddAttr(&req.n, sizeof(req), TCA_FQ_CODEL_DROP_BATCH_SIZE, &v, sizeof(v));
+#else
+                    LogWarning(LOG_NET, "fq_codel drop_batch_size not supported by kernel headers, interface %s", m_name.c_str());
+#endif
+                }
+            }
+            else if (key == "target") {
+                uint32_t v = 0U;
+                if (parseDurationUsec(value, v)) {
+                    (void)nlAddAttr(&req.n, sizeof(req), TCA_FQ_CODEL_TARGET, &v, sizeof(v));
+                }
+            }
+            else if (key == "interval") {
+                uint32_t v = 0U;
+                if (parseDurationUsec(value, v)) {
+                    (void)nlAddAttr(&req.n, sizeof(req), TCA_FQ_CODEL_INTERVAL, &v, sizeof(v));
+                }
+            }
+            else if (key == "ce_threshold") {
+                uint32_t v = 0U;
+                if (parseDurationUsec(value, v)) {
+#if defined(TCA_FQ_CODEL_CE_THRESHOLD)
+                    (void)nlAddAttr(&req.n, sizeof(req), TCA_FQ_CODEL_CE_THRESHOLD, &v, sizeof(v));
+#else
+                    LogWarning(LOG_NET, "fq_codel ce_threshold not supported by kernel headers, interface %s", m_name.c_str());
+#endif
+                }
+            }
+            else if (key == "ecn") {
+                uint32_t v = 0U;
+                if (parseSizeBytes(value, v)) {
+                    (void)nlAddAttr(&req.n, sizeof(req), TCA_FQ_CODEL_ECN, &v, sizeof(v));
+                }
+            }
+        }
+
+        nlNestEnd(&req.n, optsNest);
+    }
+    else if (!args.empty()) {
+        LogWarning(LOG_NET, "qdisc args are currently supported only for fq_codel, interface %s", m_name.c_str());
+    }
+
+    ssize_t sent = ::send(nl, &req, req.n.nlmsg_len, 0);
+    if (sent < 0) {
+        LogError(LOG_NET, "Unable to send rtnetlink qdisc request for %s, err: %d (%s)",
+            m_name.c_str(), errno, strerror(errno));
+        ::close(nl);
+        return false;
+    }
+
+    uint8_t replyBuf[512U];
+    ssize_t recvd = ::recv(nl, replyBuf, sizeof(replyBuf), 0);
+    if (recvd < 0) {
+        LogError(LOG_NET, "Unable to receive rtnetlink qdisc response for %s, err: %d (%s)",
+            m_name.c_str(), errno, strerror(errno));
+        ::close(nl);
+        return false;
+    }
+
+    struct nlmsghdr* nh = (struct nlmsghdr*)replyBuf;
+    for (; NLMSG_OK(nh, (uint32_t)recvd); nh = NLMSG_NEXT(nh, recvd)) {
+        if (nh->nlmsg_type == NLMSG_ERROR) {
+            struct nlmsgerr* err = (struct nlmsgerr*)NLMSG_DATA(nh);
+            if (err->error == 0) {
+                ::close(nl);
+                return true;
+            }
+
+            LogError(LOG_NET, "Kernel rejected qdisc %s on %s, err: %d (%s)",
+                kind.c_str(), m_name.c_str(), -err->error, strerror(-err->error));
+            ::close(nl);
+            return false;
+        }
+    }
+
+    ::close(nl);
+    LogError(LOG_NET, "Unexpected rtnetlink response while setting qdisc on %s", m_name.c_str());
+    return false;
+}
+
 // ---------------------------------------------------------------------------
 //  Private Class Members
 // ---------------------------------------------------------------------------

src/common/network/viface/VIFace.h

@@ -4,7 +4,7 @@
  * GPLv2 Open Source. Use is subject to license terms.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
- *  Copyright (C) 2024 Bryan Biedenkapp, N2PLL
+ *  Copyright (C) 2024,2026 Bryan Biedenkapp, N2PLL
  *
  */
 /**
@@ -217,6 +217,14 @@ namespace network
              */
             uint32_t getMTU() const;
 
+            /**
+             * @brief Applies a root qdisc to the virtual interface using rtnetlink.
+             * @param kind qdisc kind (e.g. fq_codel, cake, tbf).
+             * @param args qdisc argument string (e.g. "limit=1024 target=5ms").
+             * @returns bool True, if qdisc was applied successfully, otherwise false.
+             */
+            bool setQDisc(std::string kind, std::string args = "");
+
         public:
             /** 
              * @brief Virtual Interface associated name.

src/fne/HostFNE.cpp

@@ -966,6 +966,20 @@ bool HostFNE::createVirtualNetworking()
         m_tun->setMTU(DEFAULT_MTU_SIZE);
 
         m_tun->up();
+
+        yaml::Node vtunQdisc = vtunConf["qdisc"];
+        bool qdiscEnable = vtunQdisc["enable"].as<bool>(false);
+        if (qdiscEnable) {
+            std::string qdiscKind = vtunQdisc["kind"].as<std::string>("fq_codel");
+            std::string qdiscArgs = vtunQdisc["args"].as<std::string>("");
+
+            if (m_tun->setQDisc(qdiscKind, qdiscArgs)) {
+                LogInfo("    VTUN qdisc applied (programmatic): %s", qdiscKind.c_str());
+            }
+            else {
+                LogWarning(LOG_HOST, "Failed to apply VTUN qdisc (programmatic): %s", qdiscKind.c_str());
+            }
+        }
     }
 #endif // !defined(_WIN32)
     return true;

src/fne/network/TrafficNetwork.cpp

@@ -151,6 +151,8 @@ TrafficNetwork::TrafficNetwork(HostFNE* host, const std::string& address, uint16
     m_disablePacketData(false),
     m_dumpPacketData(false),
     m_verbosePacketData(false),
+    m_vtunQueueMaxFrames(128U),
+    m_vtunQueueMaxBytes(262144U),
     m_sndcpStartAddr(__IP_FROM_STR("10.10.1.10")),
     m_sndcpEndAddr(__IP_FROM_STR("10.10.1.254")),
     m_totalActiveCalls(0U),
@@ -305,8 +307,26 @@ void TrafficNetwork::setOptions(yaml::Node& conf, bool printOptions)
     // SNDCP IP allocation configuration
     m_sndcpStartAddr = __IP_FROM_STR("10.10.1.10");
     m_sndcpEndAddr = __IP_FROM_STR("10.10.1.254");
+    m_vtunQueueMaxFrames = 128U;
+    m_vtunQueueMaxBytes = 262144U;
     yaml::Node& vtun = conf["vtun"];
     if (vtun.size() > 0U) {
+        yaml::Node& queue = vtun["queue"];
+        if (queue.size() > 0U) {
+            m_vtunQueueMaxFrames = queue["maxFrames"].as<uint32_t>(128U);
+            m_vtunQueueMaxBytes = queue["maxBytes"].as<uint32_t>(262144U);
+
+            if (m_vtunQueueMaxFrames == 0U) {
+                LogWarning(LOG_MASTER, "VTUN queue maxFrames is 0, clamping to 1");
+                m_vtunQueueMaxFrames = 1U;
+            }
+
+            if (m_vtunQueueMaxBytes == 0U) {
+                LogWarning(LOG_MASTER, "VTUN queue maxBytes is 0, clamping to 512");
+                m_vtunQueueMaxBytes = 512U;
+            }
+        }
+
         yaml::Node& sndcp = vtun["sndcp"];
         if (sndcp.size() > 0U) {
             std::string startAddrStr = sndcp["startAddress"].as<std::string>("10.10.1.10");
@@ -364,6 +384,8 @@ void TrafficNetwork::setOptions(yaml::Node& conf, bool printOptions)
         }
         LogInfo("    Disable Packet Data: %s", m_disablePacketData ? "yes" : "no");
         LogInfo("    Dump Packet Data: %s", m_dumpPacketData ? "yes" : "no");
+        LogInfo("    VTUN Queue Max Frames: %u", m_vtunQueueMaxFrames);
+        LogInfo("    VTUN Queue Max Bytes: %u", m_vtunQueueMaxBytes);
         LogInfo("    Disable P25 ADJ_STS_BCAST to neighbor peers: %s", m_disallowExtAdjStsBcast ? "yes" : "no");
         LogInfo("    Disable P25 Radio Monitor to any peers: %s", m_disallowRadioMonitor ? "yes" : "no");
         LogInfo("    Disable P25 TDULC call termination broadcasts to any peers: %s", m_disallowCallTerm ? "yes" : "no");

src/fne/network/TrafficNetwork.h

@@ -432,6 +432,9 @@ namespace network
         bool m_dumpPacketData;
         bool m_verbosePacketData;
 
+        uint32_t m_vtunQueueMaxFrames;
+        uint32_t m_vtunQueueMaxBytes;
+
         uint32_t m_sndcpStartAddr;
         uint32_t m_sndcpEndAddr;
 

src/fne/network/callhandler/packetdata/DMRPacketData.cpp

@@ -55,6 +55,7 @@ DMRPacketData::DMRPacketData(TrafficNetwork* network, TagDMRData* tag, bool debu
     m_network(network),
     m_tag(tag),
     m_queuedFrames(),
+    m_queuedFrameBytes(0U),
     m_status(),
     m_arpTable(),
     m_readyForNextPkt(),
@@ -67,7 +68,20 @@ DMRPacketData::DMRPacketData(TrafficNetwork* network, TagDMRData* tag, bool debu
 
 /* Finalizes a instance of the DMRPacketData class. */
 
-DMRPacketData::~DMRPacketData() = default;
+DMRPacketData::~DMRPacketData()
+{
+    while (m_queuedFrames.size() > 0U) {
+        QueuedDataFrame* frame = m_queuedFrames[0U];
+        m_queuedFrames.pop_front();
+        if (frame != nullptr) {
+            if (frame->header != nullptr)
+                delete frame->header;
+            if (frame->userData != nullptr)
+                delete[] frame->userData;
+            delete frame;
+        }
+    }
+}
 
 /* Process a data frame from the network. */
 
@@ -357,6 +371,44 @@ void DMRPacketData::processPacketFrame(const uint8_t* data, uint32_t len, bool a
     DECLARE_UINT8_ARRAY(pduUserData, pduLength);
     ::memcpy(pduUserData, data, pktLen);
 
+    if (pduLength > m_network->m_vtunQueueMaxBytes) {
+        LogWarning(LOG_DMR, "VTUN queue drop, frame too large for queue cap, frameBytes = %u, capBytes = %u",
+            pduLength, m_network->m_vtunQueueMaxBytes);
+        return;
+    }
+
+    uint32_t droppedFrames = 0U;
+    while (m_queuedFrames.size() >= m_network->m_vtunQueueMaxFrames ||
+           (m_queuedFrameBytes + pduLength) > m_network->m_vtunQueueMaxBytes) {
+        if (m_queuedFrames.size() == 0U) {
+            break;
+        }
+
+        QueuedDataFrame* oldFrame = m_queuedFrames[0U];
+        m_queuedFrames.pop_front();
+        if (oldFrame != nullptr) {
+            if (oldFrame->userDataLen <= m_queuedFrameBytes) {
+                m_queuedFrameBytes -= oldFrame->userDataLen;
+            }
+            else {
+                m_queuedFrameBytes = 0U;
+            }
+
+            if (oldFrame->header != nullptr)
+                delete oldFrame->header;
+            if (oldFrame->userData != nullptr)
+                delete[] oldFrame->userData;
+            delete oldFrame;
+        }
+
+        droppedFrames++;
+    }
+
+    if (droppedFrames > 0U) {
+        LogWarning(LOG_DMR, "VTUN queue cap reached, dropped %u oldest frame(s), queuedFrames = %u, queuedBytes = %u",
+            droppedFrames, (uint32_t)m_queuedFrames.size(), m_queuedFrameBytes);
+    }
+
     // queue frame for dispatch
     QueuedDataFrame* qf = new QueuedDataFrame();
     qf->retryCnt = 0U;
@@ -372,6 +424,7 @@ void DMRPacketData::processPacketFrame(const uint8_t* data, uint32_t len, bool a
     qf->userDataLen = pduLength;
 
     m_queuedFrames.push_back(qf);
+    m_queuedFrameBytes += pduLength;
 #endif // !defined(_WIN32)
 }
 
@@ -419,6 +472,12 @@ void DMRPacketData::clock(uint32_t ms)
                     frame->retryCnt++;
                 } else {
                     LogWarning(LOG_DMR, "DMR, failed to resolve ARP for dstId %u", frame->dstId);
+                    if (frame->userDataLen <= m_queuedFrameBytes) {
+                        m_queuedFrameBytes -= frame->userDataLen;
+                    }
+                    else {
+                        m_queuedFrameBytes = 0U;
+                    }
                     delete frame->header;
                     delete[] frame->userData;
                     delete frame;
@@ -428,6 +487,13 @@ void DMRPacketData::clock(uint32_t ms)
                 // transmit the PDU frame
                 dispatchUserFrameToFNE(*frame->header, frame->userData);
 
+                if (frame->userDataLen <= m_queuedFrameBytes) {
+                    m_queuedFrameBytes -= frame->userDataLen;
+                }
+                else {
+                    m_queuedFrameBytes = 0U;
+                }
+
                 delete frame->header;
                 delete[] frame->userData;
                 delete frame;

src/fne/network/callhandler/packetdata/DMRPacketData.h

@@ -118,6 +118,7 @@ namespace network
                     bool extendRetry;               //!< Flag indicating whether or not to extend the retry count for this packet.
                 };
                 concurrent::deque<QueuedDataFrame*> m_queuedFrames;
+                uint32_t m_queuedFrameBytes;
 
                 /**
                  * @brief Represents the receive status of a call.

src/fne/network/callhandler/packetdata/P25PacketData.cpp

@@ -60,6 +60,7 @@ P25PacketData::P25PacketData(TrafficNetwork* network, TagP25Data* tag, bool debu
     m_tag(tag),
     m_assembler(nullptr),
     m_queuedFrames(),
+    m_queuedFrameBytes(0U),
     m_status(),
     m_arpTable(),
     m_readyForNextPkt(),
@@ -94,6 +95,18 @@ P25PacketData::P25PacketData(TrafficNetwork* network, TagP25Data* tag, bool debu
 
 P25PacketData::~P25PacketData()
 {
+    while (m_queuedFrames.size() > 0U) {
+        QueuedDataFrame* frame = m_queuedFrames[0U];
+        m_queuedFrames.pop_front();
+        if (frame != nullptr) {
+            if (frame->userData != nullptr)
+                delete[] frame->userData;
+            if (frame->header != nullptr)
+                delete frame->header;
+            delete frame;
+        }
+    }
+
     if (m_assembler != nullptr)
         delete m_assembler;
 }
@@ -390,6 +403,44 @@ void P25PacketData::processPacketFrame(const uint8_t* data, uint32_t len, bool a
     Utils::dump(1U, "P25, P25PacketData::processPacketFrame(), pduUserData", pduUserData, pduLength);
 //#endif
 
+    if (pduLength > m_network->m_vtunQueueMaxBytes) {
+        LogWarning(LOG_P25, "VTUN queue drop, frame too large for queue cap, frameBytes = %u, capBytes = %u",
+            pduLength, m_network->m_vtunQueueMaxBytes);
+        return;
+    }
+
+    uint32_t droppedFrames = 0U;
+    while (m_queuedFrames.size() >= m_network->m_vtunQueueMaxFrames ||
+           (m_queuedFrameBytes + pduLength) > m_network->m_vtunQueueMaxBytes) {
+        if (m_queuedFrames.size() == 0U) {
+            break;
+        }
+
+        QueuedDataFrame* oldFrame = m_queuedFrames[0U];
+        m_queuedFrames.pop_front();
+        if (oldFrame != nullptr) {
+            if (oldFrame->userDataLen <= m_queuedFrameBytes) {
+                m_queuedFrameBytes -= oldFrame->userDataLen;
+            }
+            else {
+                m_queuedFrameBytes = 0U;
+            }
+
+            if (oldFrame->userData != nullptr)
+                delete[] oldFrame->userData;
+            if (oldFrame->header != nullptr)
+                delete oldFrame->header;
+            delete oldFrame;
+        }
+
+        droppedFrames++;
+    }
+
+    if (droppedFrames > 0U) {
+        LogWarning(LOG_P25, "VTUN queue cap reached, dropped %u oldest frame(s), queuedFrames = %u, queuedBytes = %u",
+            droppedFrames, (uint32_t)m_queuedFrames.size(), m_queuedFrameBytes);
+    }
+
     // queue frame for dispatch
     QueuedDataFrame* qf = new QueuedDataFrame();
     qf->retryCnt = 0U;
@@ -405,6 +456,7 @@ void P25PacketData::processPacketFrame(const uint8_t* data, uint32_t len, bool a
     qf->userDataLen = pduLength;
 
     m_queuedFrames.push_back(qf);
+    m_queuedFrameBytes += pduLength;
 #endif // !defined(_WIN32)
 }
 
@@ -555,6 +607,15 @@ void P25PacketData::clock(uint32_t ms)
 
 pkt_clock_abort:
     m_queuedFrames.pop_front();
+    if (frame != nullptr) {
+        if (frame->userDataLen <= m_queuedFrameBytes) {
+            m_queuedFrameBytes -= frame->userDataLen;
+        }
+        else {
+            m_queuedFrameBytes = 0U;
+        }
+    }
+
     if (processed) {
         if (frame->userData != nullptr)
             delete[] frame->userData;
@@ -564,6 +625,7 @@ pkt_clock_abort:
     } else {
         // requeue packet
         m_queuedFrames.push_back(frame);
+        m_queuedFrameBytes += frame->userDataLen;
     }
 #endif // !defined(_WIN32)
 }

src/fne/network/callhandler/packetdata/P25PacketData.h

@@ -151,6 +151,7 @@ namespace network
                     bool extendRetry;               //!< Flag indicating whether or not to extend the retry count for this packet.
                 };
                 concurrent::deque<QueuedDataFrame*> m_queuedFrames;
+                uint32_t m_queuedFrameBytes;
 
                 /**
                  * @brief Represents the receive status of a call.