diff --git a/OSCRouter/MainWindow.cpp b/OSCRouter/MainWindow.cpp index 8fea6fc..ed1f81d 100644 --- a/OSCRouter/MainWindow.cpp +++ b/OSCRouter/MainWindow.cpp @@ -1714,6 +1714,10 @@ QString RoutingWidget::HeaderForCol(Col col) case Col::kOutMax: return tr("Max"); case Col::kOutScript: return tr("JS"); + + case Col::kMaxRateHz: return tr("Max Rate (Hz)"); + + case Col::kOnlyChanges: return tr("Only Changes"); } return QString(); @@ -1898,6 +1902,18 @@ void RoutingWidget::AddRow(size_t id, bool remove, const QString& label, const R row.outMax->setText(transformStr); AddCol(col++, row.outMax); + row.maxRateHz = new LineEdit(m_Cols->widget(col)); + row.maxRateHz->setToolTip(tr("Cap outgoing OSC rate to this destination (Hz)\n\nBlank or 0 = unlimited.\nApplies to OSC outputs only.")); + if (route.dst.maxRateHz > 0.0f) + row.maxRateHz->setText(QString::number(route.dst.maxRateHz)); + AddCol(col++, row.maxRateHz); + + row.onlyChanges = new RoutingCheckBox(id, m_Cols->widget(col)); + row.onlyChanges->setToolTip(tr("Suppress outgoing OSC messages that are byte-identical to the previous send to this destination.\n\nApplies to OSC outputs only.")); + row.onlyChanges->setFixedHeight(fh); + row.onlyChanges->setChecked(route.dst.onlyChanges); + AddCol(col++, row.onlyChanges, /*fixed*/ true); + row.addRemove = new RoutingButton(remove ? QLatin1String("-") : QLatin1String("+"), id, m_Cols->widget(col)); row.addRemove->setToolTip(remove ? tr("Remove this route") : tr("Add this route")); connect(row.addRemove, &RoutingButton::clickedWithId, this, &RoutingWidget::onAddRemoveClicked); @@ -2007,6 +2023,12 @@ void RoutingWidget::LoadLine(const QString& line, Router::ROUTES& routes, ItemSt if (items.size() > 17) route.dst.multicastInterfaceIP = items[17]; + if (items.size() > 18) + route.dst.maxRateHz = items[18].toFloat(); + + if (items.size() > 19) + route.dst.onlyChanges = (items[19].toInt() != 0); + routes.push_back(route); } else if (items.size() == 3 && items[0].compare(QLatin1String("Mute"), Qt::CaseInsensitive) == 0) @@ -2055,6 +2077,8 @@ void RoutingWidget::Save(QTextStream& stream) stream << QStringLiteral(",%1").arg(route.enable ? 1 : 0); stream << QStringLiteral(",%1").arg(route.mute ? 0 : 1); stream << QStringLiteral(",%1").arg(FileUtils::QuotedString(route.dst.multicastInterfaceIP)); + stream << QStringLiteral(",%1").arg(route.dst.maxRateHz, 0, 'f'); + stream << QStringLiteral(",%1").arg(route.dst.onlyChanges ? 1 : 0); stream << QLatin1Char('\n'); } } @@ -2118,6 +2142,14 @@ void RoutingWidget::SaveRoutes(Router::ROUTES& routes, ItemStateTable& itemState StringToTransform(row.outMin->text(), route.dst.outMin); StringToTransform(row.outMax->text(), route.dst.outMax); + { + const QString rateText = row.maxRateHz->text().trimmed(); + bool ok = false; + const float rate = rateText.toFloat(&ok); + route.dst.maxRateHz = (ok && rate > 0.0f) ? rate : 0.0f; + } + route.dst.onlyChanges = row.onlyChanges->isChecked(); + if (HasRoute(routes, route.src, route.dst)) continue; @@ -2330,6 +2362,8 @@ void RoutingWidget::UpdateEnableState() row.outScript->setEnabled(e); row.outMin->setEnabled(e); row.outMax->setEnabled(e); + row.maxRateHz->setEnabled(e); + row.onlyChanges->setEnabled(e && i != lastRow); } } @@ -2364,6 +2398,7 @@ void RoutingWidget::UpdateMuteState() SetMuted(row.outPath, mute); SetMuted(row.outMin, mute); SetMuted(row.outMax, mute); + SetMuted(row.maxRateHz, mute); } } diff --git a/OSCRouter/MainWindow.h b/OSCRouter/MainWindow.h index 1463ac6..a5c6c49 100644 --- a/OSCRouter/MainWindow.h +++ b/OSCRouter/MainWindow.h @@ -503,6 +503,8 @@ private slots: kOutScript, kOutMin, kOutMax, + kMaxRateHz, + kOnlyChanges, kButton, @@ -542,6 +544,8 @@ private slots: RoutingCheckBox* outScript = nullptr; LineEdit* outMin = nullptr; LineEdit* outMax = nullptr; + LineEdit* maxRateHz = nullptr; + RoutingCheckBox* onlyChanges = nullptr; RoutingButton* addRemove = nullptr; }; diff --git a/OSCRouter/NetworkUtils.cpp b/OSCRouter/NetworkUtils.cpp index 53b9afd..89c9097 100644 --- a/OSCRouter/NetworkUtils.cpp +++ b/OSCRouter/NetworkUtils.cpp @@ -293,7 +293,7 @@ bool EosRouteSrc::operator<(const EosRouteSrc &other) const bool EosRouteDst::operator==(const EosRouteDst &other) const { return (addr == other.addr && multicastInterfaceIP == other.multicastInterfaceIP && protocol == other.protocol && path == other.path && script == other.script && scriptText == other.scriptText && - inMin == other.inMin && inMax == other.inMax && outMin == other.outMin && outMax == other.outMax); + inMin == other.inMin && inMax == other.inMax && outMin == other.outMin && outMax == other.outMax && maxRateHz == other.maxRateHz && onlyChanges == other.onlyChanges); } //////////////////////////////////////////////////////////////////////////////// diff --git a/OSCRouter/NetworkUtils.h b/OSCRouter/NetworkUtils.h index cd9ff88..0890433 100644 --- a/OSCRouter/NetworkUtils.h +++ b/OSCRouter/NetworkUtils.h @@ -173,6 +173,16 @@ struct EosRouteDst sTransform inMax; sTransform outMin; sTransform outMax; + + // Rate limit (0 = unlimited; otherwise max sends per second). + // Coalesce semantics: at most N sends per second, emitting the most + // recently received message for this destination. OSC outputs only; + // ignored for sACN/ArtNet/MIDI/OTP/PSN. + float maxRateHz = 0.0f; + + // Send-on-change: when true, suppress sends that are byte-identical to + // the previous send to this destination. OSC outputs only. + bool onlyChanges = false; }; //////////////////////////////////////////////////////////////////////////////// diff --git a/OSCRouter/Router.cpp b/OSCRouter/Router.cpp index a7e860a..1bff2a6 100644 --- a/OSCRouter/Router.cpp +++ b/OSCRouter/Router.cpp @@ -36,12 +36,39 @@ #include #include +#include // must be last include #include "LeakWatcher.h" //////////////////////////////////////////////////////////////////////////////// +namespace +{ + +// True for protocols whose outbound dispatch sends OSC bytes on the wire. +// Used to scope the per-route rate limit to OSC outputs. +inline bool IsOSCOutput(Protocol p) +{ + return p != Protocol::ksACN && p != Protocol::kArtNet && p != Protocol::kMIDI && p != Protocol::kOTP && p != Protocol::kPSN; +} + +// Convert a max-rate Hz value to its inter-send interval in milliseconds. +inline qint64 RateLimitIntervalMs(float hz) +{ + return (hz > 0.0f) ? static_cast(1000.0f / hz + 0.5f) : 0; +} + +// Byte-for-byte equality check on built packets, used by the send-on-change gate. +inline bool PacketsEqual(const EosPacket &a, const EosPacket &b) +{ + return a.GetSize() == b.GetSize() && (a.GetSize() == 0 || std::memcmp(a.GetDataConst(), b.GetDataConst(), static_cast(a.GetSize())) == 0); +} + +} // namespace + +//////////////////////////////////////////////////////////////////////////////// + #define EPSILLON 0.00001f uint16_t Router::GetDefaultPSNPort() @@ -1816,7 +1843,7 @@ void RouterThread::ProcessRecvPacket(bool muteAllOutgoing, sACN &sacn, ArtNet &a const ROUTE_DESTINATIONS &destinations = **i; for (ROUTE_DESTINATIONS::const_iterator j = destinations.begin(); j != destinations.end(); j++) { - const sRouteDst &routeDst = *j; + sRouteDst &routeDst = const_cast(*j); SetItemActivity(routeDst.srcItemStateTableId); if (muteAllOutgoing || IsRouteMuted(routeDst.dstItemStateTableId)) @@ -1841,18 +1868,44 @@ void RouterThread::ProcessRecvPacket(bool muteAllOutgoing, sACN &sacn, ArtNet &a tcp = true; } + // Rate limit gate (OSC outputs, with src that yields a built OSC packet). + // Coalesce semantics: when the period hasn't elapsed, stash the latest built packet as + // pending and skip the immediate send. The deferred flush in MainLoop emits it once the + // period passes. + const bool srcCanBuildOSC = + (protocol == Protocol::kOSC || protocol == Protocol::ksACN || protocol == Protocol::kArtNet || protocol == Protocol::kMIDI || protocol == Protocol::kOTP); + const bool rateLimitApplies = (routeDst.dst.maxRateHz > 0.0f) && IsOSCOutput(routeDst.dst.protocol) && srcCanBuildOSC; + qint64 nowMs = 0; + if (rateLimitApplies) + { + const qint64 intervalMs = RateLimitIntervalMs(routeDst.dst.maxRateHz); + nowMs = QDateTime::currentMSecsSinceEpoch(); + if ((nowMs - routeDst.lastSendTimeMs) < intervalMs) + { + EosPacket coalesced; + if (MakeOSCPacket(artnet, addr, protocol, path, routeDst, args, argsCount, coalesced)) + { + routeDst.pendingPacket = coalesced; + routeDst.pendingSrcAddr = addr; + routeDst.pendingSrcProtocol = protocol; + routeDst.pendingSrcPath = path; + routeDst.pendingDstAddr = dstAddr; + routeDst.hasPending = true; + } + continue; + } + } + + bool didSend = false; if (tcp) { if (tcpClient) { if (protocol == Protocol::kOSC) { - EosPacket packet; - if (MakeOSCPacket(artnet, addr, protocol, path, routeDst, args, argsCount, packet) && tcpClient->SendFramed(packet)) - { - SetItemActivity(routeDst.dstItemStateTableId); - SetItemActivity(tcpClient->GetItemStateTableId()); - } + EosPacket builtPacket; + if (MakeOSCPacket(artnet, addr, protocol, path, routeDst, args, argsCount, builtPacket)) + didSend = DispatchBuiltPacket(sacn, artnet, midi, otpi, udpOutThreads, tcpClient, addr, protocol, routeDst, builtPacket, dstAddr); } else if (tcpClient->Send(recvPacket.packet)) { @@ -1861,45 +1914,11 @@ void RouterThread::ProcessRecvPacket(bool muteAllOutgoing, sACN &sacn, ArtNet &a } } } - else if (protocol == Protocol::kOSC || protocol == Protocol::ksACN || protocol == Protocol::kArtNet || protocol == Protocol::kMIDI || protocol == Protocol::kOTP) + else if (srcCanBuildOSC) { - EosPacket oscPacket; - MakeOSCPacket(artnet, addr, protocol, path, routeDst, args, argsCount, oscPacket); - - if (routeDst.dst.protocol == Protocol::kPSN) - { - EosPacket psnPacket; - if (MakePSNPacket(oscPacket, psnPacket)) - { - EosUdpOutThread *thread = CreateUdpOutThread(dstAddr, routeDst.dst.multicastInterfaceIP, routeDst.dstItemStateTableId, udpOutThreads); - if (thread && thread->Send(psnPacket)) - SetItemActivity(routeDst.dstItemStateTableId); - } - } - else if (routeDst.dst.protocol == Protocol::ksACN) - { - if (SendsACN(sacn, artnet, addr, protocol, routeDst, oscPacket)) - SetItemActivity(routeDst.dstItemStateTableId); - } - else if (routeDst.dst.protocol == Protocol::kArtNet) - { - if (SendArtNet(artnet, addr, protocol, routeDst.dst, oscPacket)) - SetItemActivity(routeDst.dstItemStateTableId); - } - else if (routeDst.dst.protocol == Protocol::kMIDI) - { - SendMIDI(midi, routeDst, oscPacket); - } - else if (routeDst.dst.protocol == Protocol::kOTP) - { - SendOTP(otpi, routeDst, oscPacket); - } - else if (oscPacket.GetDataConst() && oscPacket.GetSize() > 0) - { - EosUdpOutThread *thread = CreateUdpOutThread(dstAddr, routeDst.dst.multicastInterfaceIP, routeDst.dstItemStateTableId, udpOutThreads); - if (thread && thread->Send(oscPacket)) - SetItemActivity(routeDst.dstItemStateTableId); - } + EosPacket builtPacket; + MakeOSCPacket(artnet, addr, protocol, path, routeDst, args, argsCount, builtPacket); + didSend = DispatchBuiltPacket(sacn, artnet, midi, otpi, udpOutThreads, /*tcpClient*/ nullptr, addr, protocol, routeDst, builtPacket, dstAddr); } else { @@ -1907,6 +1926,9 @@ void RouterThread::ProcessRecvPacket(bool muteAllOutgoing, sACN &sacn, ArtNet &a if (thread && thread->Send(recvPacket.packet)) SetItemActivity(routeDst.dstItemStateTableId); } + + if (rateLimitApplies && didSend) + routeDst.lastSendTimeMs = nowMs; } } @@ -2036,6 +2058,139 @@ bool RouterThread::MakeOSCPacket(ArtNet &artnet, const EosAddr &addr, Protocol p //////////////////////////////////////////////////////////////////////////////// +bool RouterThread::DispatchBuiltPacket(sACN &sacn, ArtNet &artnet, MIDI &midi, OTPI &otpi, UDP_OUT_THREADS &udpOutThreads, EosTcpClientThread *tcpClient, const EosAddr &srcAddr, Protocol srcProtocol, + sRouteDst &routeDst, EosPacket &builtPacket, const EosAddr &dstAddr) +{ + // Send-on-change gate (OSC outputs). Suppress sends whose built bytes match the + // previous send to this destination. + const bool onlyChangesApplies = routeDst.dst.onlyChanges && IsOSCOutput(routeDst.dst.protocol); + if (onlyChangesApplies && routeDst.hasLastSent && PacketsEqual(builtPacket, routeDst.lastSentPacket)) + return false; + + bool sent = false; + + if (tcpClient) + { + // TCP framed dispatch: only for OSC dst when a tcp client thread is bound. + if (tcpClient->SendFramed(builtPacket)) + { + SetItemActivity(routeDst.dstItemStateTableId); + SetItemActivity(tcpClient->GetItemStateTableId()); + sent = true; + } + } + else if (routeDst.dst.protocol == Protocol::kPSN) + { + EosPacket psnPacket; + if (MakePSNPacket(builtPacket, psnPacket)) + { + EosUdpOutThread *thread = CreateUdpOutThread(dstAddr, routeDst.dst.multicastInterfaceIP, routeDst.dstItemStateTableId, udpOutThreads); + if (thread && thread->Send(psnPacket)) + { + SetItemActivity(routeDst.dstItemStateTableId); + sent = true; + } + } + } + else if (routeDst.dst.protocol == Protocol::ksACN) + { + if (SendsACN(sacn, artnet, srcAddr, srcProtocol, routeDst, builtPacket)) + { + SetItemActivity(routeDst.dstItemStateTableId); + sent = true; + } + } + else if (routeDst.dst.protocol == Protocol::kArtNet) + { + if (SendArtNet(artnet, srcAddr, srcProtocol, routeDst.dst, builtPacket)) + { + SetItemActivity(routeDst.dstItemStateTableId); + sent = true; + } + } + else if (routeDst.dst.protocol == Protocol::kMIDI) + { + SendMIDI(midi, routeDst, builtPacket); + sent = true; + } + else if (routeDst.dst.protocol == Protocol::kOTP) + { + SendOTP(otpi, routeDst, builtPacket); + sent = true; + } + else if (builtPacket.GetDataConst() && builtPacket.GetSize() > 0) + { + // Default: OSC over UDP + EosUdpOutThread *thread = CreateUdpOutThread(dstAddr, routeDst.dst.multicastInterfaceIP, routeDst.dstItemStateTableId, udpOutThreads); + if (thread && thread->Send(builtPacket)) + { + SetItemActivity(routeDst.dstItemStateTableId); + sent = true; + } + } + + if (sent && onlyChangesApplies) + { + routeDst.lastSentPacket = builtPacket; + routeDst.hasLastSent = true; + } + + return sent; +} + +//////////////////////////////////////////////////////////////////////////////// + +void RouterThread::FlushPendingRateLimited(ROUTES_BY_PORT &routes, qint64 nowMs, sACN &sacn, ArtNet &artnet, MIDI &midi, OTPI &otpi, UDP_OUT_THREADS &udpOutThreads, + TCP_CLIENT_THREADS &tcpClientThreads) +{ + for (ROUTES_BY_PORT::iterator portIter = routes.begin(); portIter != routes.end(); ++portIter) + { + ROUTES_BY_IP &routesByIp = portIter->second; + for (ROUTES_BY_IP::iterator ipIter = routesByIp.begin(); ipIter != routesByIp.end(); ++ipIter) + { + sRoutesByIp &group = ipIter->second; + ROUTES_BY_PATH *pathMaps[2] = {&group.routesByPath, &group.routesByWildcardPath}; + for (int m = 0; m < 2; ++m) + { + for (ROUTES_BY_PATH::iterator pathIter = pathMaps[m]->begin(); pathIter != pathMaps[m]->end(); ++pathIter) + { + ROUTE_DESTINATIONS &destinations = pathIter->second; + for (ROUTE_DESTINATIONS::iterator d = destinations.begin(); d != destinations.end(); ++d) + { + sRouteDst &routeDst = *d; + if (!routeDst.hasPending) + continue; + if (routeDst.dst.maxRateHz <= 0.0f) + { + routeDst.hasPending = false; + continue; + } + const qint64 intervalMs = RateLimitIntervalMs(routeDst.dst.maxRateHz); + if ((nowMs - routeDst.lastSendTimeMs) < intervalMs) + continue; + + EosTcpClientThread *tcpClient = nullptr; + if (routeDst.dst.protocol == Protocol::kOSC) + { + TCP_CLIENT_THREADS::const_iterator k = tcpClientThreads.find(routeDst.pendingDstAddr); + if (k != tcpClientThreads.end()) + tcpClient = k->second; + } + + const bool sent = + DispatchBuiltPacket(sacn, artnet, midi, otpi, udpOutThreads, tcpClient, routeDst.pendingSrcAddr, routeDst.pendingSrcProtocol, routeDst, routeDst.pendingPacket, routeDst.pendingDstAddr); + if (sent) + routeDst.lastSendTimeMs = nowMs; + routeDst.hasPending = false; + } + } + } + } + } +} + +//////////////////////////////////////////////////////////////////////////////// + bool GetFloat3(OSCArgument *args, size_t argCount, size_t index, psn::float3 &f3) { if (!args || (index + 2) >= argCount) @@ -3357,6 +3512,12 @@ void RouterThread::MainLoop() i++; } + // Flush any pending coalesced packets whose rate-limit period has elapsed. + { + const qint64 flushNowMs = QDateTime::currentMSecsSinceEpoch(); + FlushPendingRateLimited(routesByPort, flushNowMs, sacn, artnet, midi, otpi, udpOutThreads, tcpClientThreads); + } + // UDP output for (UDP_OUT_THREADS::iterator i = udpOutThreads.begin(); i != udpOutThreads.end();) { diff --git a/OSCRouter/Router.h b/OSCRouter/Router.h index d7e8336..03455da 100644 --- a/OSCRouter/Router.h +++ b/OSCRouter/Router.h @@ -448,6 +448,19 @@ class RouterThread : public QThread, private OSCParserClient, private IStreamACN EosRouteDst dst; ItemStateTable::ID srcItemStateTableId; ItemStateTable::ID dstItemStateTableId; + + // Rate limit runtime state (router thread only — no mutex). + qint64 lastSendTimeMs = 0; + bool hasPending = false; + EosPacket pendingPacket; + EosAddr pendingSrcAddr; + Protocol pendingSrcProtocol = Protocol::kDefault; + QString pendingSrcPath; + EosAddr pendingDstAddr; + + // Send-on-change runtime state. + bool hasLastSent = false; + EosPacket lastSentPacket; }; typedef std::vector ROUTE_DESTINATIONS; @@ -623,6 +636,10 @@ class RouterThread : public QThread, private OSCParserClient, private IStreamACN UDP_OUT_THREADS &udpOutThreads, TCP_SERVER_THREADS &tcpServerThreads, TCP_CLIENT_THREADS &tcpClientThreads, const EosAddr &addr, Protocol protocol, EosUdpInThread::sRecvPacket &recvPacket); virtual bool MakeOSCPacket(ArtNet &artnet, const EosAddr &addr, Protocol protocol, const QString &srcPath, const sRouteDst &route, OSCArgument *args, size_t argsCount, EosPacket &packet); + virtual bool DispatchBuiltPacket(sACN &sacn, ArtNet &artnet, MIDI &midi, OTPI &otpi, UDP_OUT_THREADS &udpOutThreads, EosTcpClientThread *tcpClient, const EosAddr &srcAddr, Protocol srcProtocol, + sRouteDst &routeDst, EosPacket &builtPacket, const EosAddr &dstAddr); + virtual void FlushPendingRateLimited(ROUTES_BY_PORT &routes, qint64 nowMs, sACN &sacn, ArtNet &artnet, MIDI &midi, OTPI &otpi, UDP_OUT_THREADS &udpOutThreads, + TCP_CLIENT_THREADS &tcpClientThreads); virtual bool MakePSNPacket(EosPacket &osc, EosPacket &psn); virtual bool SendsACN(sACN &sacn, ArtNet &artnet, const EosAddr &addr, Protocol protocol, const sRouteDst &routeDst, EosPacket &osc); virtual bool SendArtNet(ArtNet &artnet, const EosAddr &addr, Protocol protocol, const EosRouteDst &dst, EosPacket &osc);