diff --git a/configs/fne-config.example.yml b/configs/fne-config.example.yml index 7435ffd3..82fcab48 100644 --- a/configs/fne-config.example.yml +++ b/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 # diff --git a/src/common/network/viface/VIFace.cpp b/src/common/network/viface/VIFace.cpp index 69034d93..3a331d54 100644 --- a/src/common/network/viface/VIFace.cpp +++ b/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 #include #include +#include #include #include #include +#include #include #include @@ -37,6 +39,8 @@ using namespace network::viface; #include #include #include +#include +#include #include // --------------------------------------------------------------------------- @@ -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 parseQDiscArgs(const std::string& args) +{ + std::unordered_map out; + if (args.empty()) { + return out; + } + + std::vector 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 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 // --------------------------------------------------------------------------- diff --git a/src/common/network/viface/VIFace.h b/src/common/network/viface/VIFace.h index bfbef0fb..4f2063a9 100644 --- a/src/common/network/viface/VIFace.h +++ b/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. diff --git a/src/fne/HostFNE.cpp b/src/fne/HostFNE.cpp index 8ccbd2ae..cd7cfa2b 100644 --- a/src/fne/HostFNE.cpp +++ b/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(false); + if (qdiscEnable) { + std::string qdiscKind = vtunQdisc["kind"].as("fq_codel"); + std::string qdiscArgs = vtunQdisc["args"].as(""); + + 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; diff --git a/src/fne/network/TrafficNetwork.cpp b/src/fne/network/TrafficNetwork.cpp index 791c268c..fd2d1a28 100644 --- a/src/fne/network/TrafficNetwork.cpp +++ b/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(128U); + m_vtunQueueMaxBytes = queue["maxBytes"].as(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("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"); diff --git a/src/fne/network/TrafficNetwork.h b/src/fne/network/TrafficNetwork.h index 96d512a1..57447d25 100644 --- a/src/fne/network/TrafficNetwork.h +++ b/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; diff --git a/src/fne/network/callhandler/packetdata/DMRPacketData.cpp b/src/fne/network/callhandler/packetdata/DMRPacketData.cpp index 9114020c..ad6caf93 100644 --- a/src/fne/network/callhandler/packetdata/DMRPacketData.cpp +++ b/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; diff --git a/src/fne/network/callhandler/packetdata/DMRPacketData.h b/src/fne/network/callhandler/packetdata/DMRPacketData.h index 7d15195d..82862a96 100644 --- a/src/fne/network/callhandler/packetdata/DMRPacketData.h +++ b/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 m_queuedFrames; + uint32_t m_queuedFrameBytes; /** * @brief Represents the receive status of a call. diff --git a/src/fne/network/callhandler/packetdata/P25PacketData.cpp b/src/fne/network/callhandler/packetdata/P25PacketData.cpp index 3f4258d2..bc3997dd 100644 --- a/src/fne/network/callhandler/packetdata/P25PacketData.cpp +++ b/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) } diff --git a/src/fne/network/callhandler/packetdata/P25PacketData.h b/src/fne/network/callhandler/packetdata/P25PacketData.h index 6aad025f..61429ea0 100644 --- a/src/fne/network/callhandler/packetdata/P25PacketData.h +++ b/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 m_queuedFrames; + uint32_t m_queuedFrameBytes; /** * @brief Represents the receive status of a call.