GCC Code Coverage Report

Directory: ./
File: channeloutput/UDPOutput.cpp
Date: 2026-10-29 08:11:58
Exec Total Coverage
Lines: 1087 1168 93.1%
Functions: 73 75 97.3%
Branches: 1282 2050 62.5%
Line Branch Exec Source
1 /*
2 * This file is part of the Falcon Player (FPP) and is Copyright (C)
3 * 2013-2022 by the Falcon Player Developers.
4 *
5 * The Falcon Player (FPP) is free software, and is covered under
6 * multiple Open Source licenses. Please see the included 'LICENSES'
7 * file for descriptions of what files are covered by each license.
8 *
9 * This source file is covered under the GPL v2 as described in the
10 * included LICENSE.GPL file.
11 */
12
13 #include "fpp-pch.h"
14
15 #include "fpp-json.h"
16
17 #ifndef PLATFORM_OSX
18 #include <linux/sockios.h>
19 #include <sys/ioctl.h>
20 #endif
21
22 #include <arpa/inet.h>
23 #include <netinet/in.h>
24 #include <sys/socket.h>
25 #include <ifaddrs.h>
26 #include <netdb.h>
27 #include <unistd.h>
28
29 #include <curl/curl.h>
30 #include <cmath> // ceil -- pacing-budget warning rate (needed directly for NOPCH builds)
31 #include <algorithm> // find_if -- one pending work item per destination (needed directly for NOPCH builds)
32 #include <filesystem>
33 #include <set>
34 #include <sstream>
35
36 #include "../CurlManager.h"
37 #include "../Sequence.h"
38 #include "../Warnings.h"
39 #include "../common.h"
40 #include "../log.h"
41 #include "../settings.h"
42
43 #include "UDPOutput.h"
44 #include "ChannelOutputSetup.h"
45 #include "ping.h"
46
47 #include "NetworkMonitor.h"
48
49 #include "ArtNet.h"
50 #include "DDP.h"
51 #include "E131.h"
52 #include "KiNet.h"
53 #include "Twinkly.h"
54 #include "../e131bridge.h"
55
56 #include "Plugin.h"
57
58 constexpr int UDP_PING_TIMEOUT = 250;
59
60 class UDPPlugin : public FPPPlugins::Plugin, public FPPPlugins::ChannelOutputPlugin {
61 public:
62 793 UDPPlugin() :
63
2/4
✓ Branch 0 taken 793 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 793 times.
✗ Branch 3 not taken.
 2379 FPPPlugins::Plugin("UDP") {
64 793 }
65 793 virtual ChannelOutput* createChannelOutput(unsigned int startChannel, unsigned int channelCount) override {
66
2/6
✓ Branch 0 taken 793 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 793 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
 1586 return new UDPOutput(startChannel, channelCount);
67 }
68 };
69
70 extern "C" {
71 793 FPPPlugins::Plugin* createPlugin() {
72
2/6
✓ Branch 0 taken 793 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 793 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
 1586 return new UDPPlugin();
73 }
74 }
75
76 67 static inline std::string createWarning(const std::string& host, const std::string& type, const std::string& description) {
77
5/10
✓ Branch 0 taken 67 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 67 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 67 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 67 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 67 times.
✗ Branch 9 not taken.
 67 return "Cannot Ping " + type + " Channel Data Target " + host + " " + description;
78 }
79
80 struct UDPOutput::PingHost {
81 std::string address;
82 std::vector<UDPOutputData*> outputs;
83 std::string warning;
84 // the ping thread and the HEAD probe's callback (main loop) both update it
85 std::mutex lock;
86 int failCount = 0;
87 bool answered = false;
88 bool valid = true;
89 bool headPending = false;
90 };
91
92 // Interface names that reach shells (interfaceHasFQ's tc popen below) and
93 // sysfs paths must be plausible kernel names: IFNAMSIZ-1 chars, no '/' and no
94 // leading '.'/'-'. The configured value is operator-editable JSON. Deliberately
95 // platform-independent (used on all targets); manual char checks avoid any new
96 // header dependency.
97 1559 static bool isValidInterfaceName(const std::string& n) {
98
5/6
✓ Branch 0 taken 1559 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1 times.
✓ Branch 3 taken 1558 times.
✓ Branch 4 taken 1 times.
✓ Branch 5 taken 1558 times.
 1559 if (n.empty() || n.size() > 15) {
99 1 return false;
100 }
101 6525 auto ok = [](char c) {
102
8/18
✓ Branch 0 taken 5669 times.
✓ Branch 1 taken 856 times.
✓ Branch 2 taken 5669 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 5669 times.
✓ Branch 5 taken 856 times.
✗ Branch 6 not taken.
✓ Branch 7 taken 5669 times.
✓ Branch 8 taken 856 times.
✗ Branch 9 not taken.
✗ Branch 10 not taken.
✓ Branch 11 taken 856 times.
✗ Branch 12 not taken.
✗ Branch 13 not taken.
✗ Branch 14 not taken.
✗ Branch 15 not taken.
✗ Branch 16 not taken.
✗ Branch 17 not taken.
 6525 return (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') || c == '_' || c == '.' || c == '-';
103 };
104
4/8
✓ Branch 0 taken 1558 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1558 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 1558 times.
✗ Branch 6 not taken.
✓ Branch 7 taken 1558 times.
 1558 if (!ok(n[0]) || n[0] == '.' || n[0] == '-') {
105 ✗ return false;
106 }
107
2/2
✓ Branch 0 taken 4967 times.
✓ Branch 1 taken 1558 times.
 6525 for (char c : n) {
108
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 4967 times.
 4967 if (!ok(c)) {
109 ✗ return false;
110 }
111 }
112 1558 return true;
113 }
114
115 #ifndef PLATFORM_OSX
116 // SO_MAX_PACING_RATE only works if the interface's root qdisc is fq.
117 // Some kernels expose the qdisc name in sysfs; fall back to asking tc.
118 1537 static bool queryInterfaceHasFQ(const std::string& iface) {
119
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1537 times.
 1537 if (!isValidInterfaceName(iface)) {
120 ✗ return false;
121 }
122
3/6
✓ Branch 0 taken 1537 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1537 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 1537 times.
✗ Branch 5 not taken.
 1537 std::string qdisc = GetFileContents("/sys/class/net/" + iface + "/qdisc");
123
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1537 times.
 1537 if (!qdisc.empty()) {
124 ✗ TrimWhiteSpace(qdisc);
125 ✗ return qdisc == "fq";
126 }
127 // some kernels don't expose the qdisc name in sysfs; ask tc
128
4/8
✓ Branch 0 taken 1537 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1537 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 1537 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 1537 times.
✗ Branch 7 not taken.
 1537 std::string cmd = TcPath() + " qdisc show dev " + iface + " 2>/dev/null";
129
1/2
✓ Branch 0 taken 1537 times.
✗ Branch 1 not taken.
 1537 FILE* pipe = popen(cmd.c_str(), "r");
130
1/2
✓ Branch 0 taken 1537 times.
✗ Branch 1 not taken.
 1537 if (pipe) {
131 char buf[256];
132
3/4
✓ Branch 0 taken 3796 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2259 times.
✓ Branch 3 taken 1537 times.
 3796 while (fgets(buf, sizeof(buf), pipe)) {
133
4/4
✓ Branch 0 taken 1496 times.
✓ Branch 1 taken 763 times.
✓ Branch 2 taken 770 times.
✓ Branch 3 taken 726 times.
 2259 if (strstr(buf, " root ") && strncmp(buf, "qdisc fq ", 9) == 0) {
134
1/2
✓ Branch 0 taken 770 times.
✗ Branch 1 not taken.
 770 qdisc = "fq";
135 }
136 }
137
1/2
✓ Branch 0 taken 1537 times.
✗ Branch 1 not taken.
 1537 pclose(pipe);
138 }
139
1/2
✓ Branch 0 taken 1537 times.
✗ Branch 1 not taken.
 1537 return qdisc == "fq";
140 1537 }
141 // The answer per interface, because the sysfs file does not exist on the
142 // kernels FPP ships (6.x), so every query forks tc - and findOrCreateSocket()
143 // asks once per destination, on the output thread, at startup, on every
144 // reload and after every address loss. fppinit sets the qdisc at boot, so
145 // it does not change under a running output; UDPOutput::Init() clears the
146 // cache and warms it from the main thread.
147 static std::mutex fqCacheLock;
148 static std::map<std::string, bool> fqCache;
149 816 static void clearFQCache() {
150
1/2
✓ Branch 0 taken 816 times.
✗ Branch 1 not taken.
 816 std::lock_guard<std::mutex> lk(fqCacheLock);
151 816 fqCache.clear();
152 816 }
153 2427 static bool interfaceHasFQ(const std::string& iface) {
154 {
155
1/2
✓ Branch 0 taken 2427 times.
✗ Branch 1 not taken.
 2427 std::lock_guard<std::mutex> lk(fqCacheLock);
156
1/2
✓ Branch 0 taken 2427 times.
✗ Branch 1 not taken.
 2427 auto it = fqCache.find(iface);
157
2/2
✓ Branch 0 taken 890 times.
✓ Branch 1 taken 1537 times.
 2427 if (it != fqCache.end()) {
158 890 return it->second;
159 }
160
2/2
✓ Branch 0 taken 1537 times.
✓ Branch 1 taken 890 times.
 2427 }
161 // not under the lock: tc can take a while, and a duplicate query by
162 // another thread is harmless
163
1/2
✓ Branch 0 taken 1537 times.
✗ Branch 1 not taken.
 1537 bool fq = queryInterfaceHasFQ(iface);
164
1/2
✓ Branch 0 taken 1537 times.
✗ Branch 1 not taken.
 1537 std::lock_guard<std::mutex> lk(fqCacheLock);
165
1/2
✓ Branch 0 taken 1537 times.
✗ Branch 1 not taken.
 1537 fqCache[iface] = fq;
166 1537 return fq;
167 1537 }
168 // Unicast sockets follow the routing table, not the configured output
169 // interface, and the interface scan at startup can race the network coming
170 // up. fppinit installs fq on every ethernet interface at boot, so consider
171 // pacing available if any of them has it. Whether a specific destination's
172 // socket is actually paced is decided per destination by looking up the
173 // egress interface of its route (see findOrCreateSocket).
174 1576 static bool anyEthernetHasFQ(const std::string& primary) {
175
5/6
✓ Branch 0 taken 1576 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1511 times.
✓ Branch 3 taken 65 times.
✓ Branch 4 taken 1511 times.
✓ Branch 5 taken 65 times.
 1576 if (!primary.empty() && interfaceHasFQ(primary)) {
176 1511 return true;
177 }
178
5/8
✓ Branch 0 taken 65 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 65 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 70 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 90 times.
✓ Branch 7 taken 45 times.
 135 for (const auto& entry : std::filesystem::directory_iterator("/sys/class/net/")) {
179
2/4
✓ Branch 0 taken 90 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 90 times.
✗ Branch 3 not taken.
 90 std::string dev = entry.path().filename();
180
9/12
✓ Branch 0 taken 90 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 25 times.
✓ Branch 3 taken 65 times.
✓ Branch 4 taken 20 times.
✓ Branch 5 taken 5 times.
✓ Branch 6 taken 20 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 20 times.
✗ Branch 9 not taken.
✓ Branch 10 taken 20 times.
✓ Branch 11 taken 70 times.
 90 if (IsShowEthernetInterface(dev) && dev != primary && interfaceHasFQ(dev)) {
181 20 return true;
182 }
183
8/8
✓ Branch 0 taken 70 times.
✓ Branch 1 taken 20 times.
✓ Branch 2 taken 45 times.
✓ Branch 3 taken 20 times.
✓ Branch 4 taken 45 times.
✓ Branch 5 taken 20 times.
✓ Branch 6 taken 45 times.
✓ Branch 7 taken 20 times.
 195 }
184 45 return false;
185 }
186 // Send buffer for paced sockets: the kernel silently caps SO_SNDBUF at
187 // wmem_max (set by fppinit at boot), so ask for exactly what it will grant.
188 102 static int pacedSndBufSize() {
189 116 static int size = []() {
190
2/4
✓ Branch 0 taken 58 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 58 times.
✗ Branch 3 not taken.
 58 int v = atoi(GetFileContents("/proc/sys/net/core/wmem_max").c_str());
191
1/2
✓ Branch 0 taken 58 times.
✗ Branch 1 not taken.
 58 return v > 0 ? v : 393216;
192
4/8
✓ Branch 0 taken 58 times.
✓ Branch 1 taken 44 times.
✓ Branch 2 taken 58 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 58 times.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
✗ Branch 7 not taken.
 102 }();
193 102 return size;
194 }
195 // Which interface will packets to this destination actually leave through?
196 // A connect() on a UDP socket does the route lookup without sending anything.
197 // (The address->interface tail is intentionally not FindInterfaceForIP(): that
198 // helper returns a static buffer and round-trips through strings; comparing
199 // s_addr directly is thread-safe and exact.)
200 836 static std::string egressInterfaceFor(in_addr_t dest) {
201 836 int s = OpenSocket(AF_INET, SOCK_DGRAM, 0);
202
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 836 times.
 836 if (s < 0) {
203 ✗ return "";
204 }
205 struct sockaddr_in sa;
206 836 memset(&sa, 0, sizeof(sa));
207 836 sa.sin_family = AF_INET;
208 836 sa.sin_port = htons(53);
209 836 sa.sin_addr.s_addr = dest;
210 836 std::string ret;
211
3/4
✓ Branch 0 taken 836 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 832 times.
✓ Branch 3 taken 4 times.
 836 if (connect(s, (struct sockaddr*)&sa, sizeof(sa)) == 0) {
212 struct sockaddr_in local;
213 832 socklen_t len = sizeof(local);
214
1/2
✓ Branch 0 taken 832 times.
✗ Branch 1 not taken.
 832 if (getsockname(s, (struct sockaddr*)&local, &len) == 0) {
215 struct ifaddrs *interfaces, *tmp;
216
1/2
✓ Branch 0 taken 832 times.
✗ Branch 1 not taken.
 832 if (getifaddrs(&interfaces) == 0) {
217
1/2
✓ Branch 0 taken 2788 times.
✗ Branch 1 not taken.
 2788 for (tmp = interfaces; tmp; tmp = tmp->ifa_next) {
218
3/4
✓ Branch 0 taken 2788 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1036 times.
✓ Branch 3 taken 1752 times.
 2788 if (tmp->ifa_addr && tmp->ifa_addr->sa_family == AF_INET &&
219
2/2
✓ Branch 0 taken 832 times.
✓ Branch 1 taken 204 times.
 1036 ((struct sockaddr_in*)tmp->ifa_addr)->sin_addr.s_addr == local.sin_addr.s_addr) {
220
1/2
✓ Branch 0 taken 832 times.
✗ Branch 1 not taken.
 832 ret = tmp->ifa_name;
221 832 break;
222 }
223 }
224 832 freeifaddrs(interfaces);
225 }
226 }
227 }
228
1/2
✓ Branch 0 taken 836 times.
✗ Branch 1 not taken.
 836 close(s);
229 836 return ret;
230 836 }
231 #endif
232
233 // Sites that log a failed send. A destination that is persistently behind
234 // fails every frame, and logging each one wrote 1.4 million lines (360MB of
235 // fppd.log) on one player, so each site logs a destination at most once per
236 // SEND_ERR_LOG_INTERVAL_MS. CheckPacingBudget() reports the totals.
237 enum SendErrLogSite { SEND_ERR_BUDGET,
238 SEND_ERR_EAGAIN,
239 SEND_ERR_FRAME,
240 SEND_ERR_SITES };
241 static constexpr long long SEND_ERR_LOG_INTERVAL_MS = 10000;
242
243 class SendSocketInfo {
244 public:
245 880 SendSocketInfo() {
246 880 errCount = 0;
247 880 curSocket = -1;
248 880 }
249 ~SendSocketInfo();
250
251 // Never holds -1: a socket that couldn't be opened (out of descriptors)
252 // is left out, so the list stays empty and findOrCreateSocket() tries
253 // again on the next frame. A -1 kept here made the list look complete,
254 // and every later send failed on it with EBADF until the config reloaded.
255 std::vector<int> sockets;
256 // read/written by workers and the frame thread without a common lock;
257 // atomic so the errCount >= 3 re-ping trigger can't act on a torn value.
258 // A counter only, so relaxed throughout.
259 std::atomic_int errCount;
260 // Runs on every successful send, per destination per frame: skip the
261 // store (two barriers on an AM335x) when there is nothing to clear.
262 107704 void clearErrors() {
263
2/2
✓ Branch 0 taken 466 times.
✓ Branch 1 taken 107238 times.
 215408 if (errCount.load(std::memory_order_relaxed) != 0)
264 466 errCount.store(0, std::memory_order_relaxed);
265 107704 }
266 int curSocket;
267 bool preventClose = false;
268 // kernel (fq) pacing is active on these sockets; drain waits are the
269 // qdisc's job and EAGAIN means "wait", not "get another socket".
270 // pacedChecked latches once the destination's route has resolved; until
271 // then (e.g. interface still waiting on DHCP at startup) the pacing
272 // decision is re-attempted periodically (pacedCheckCountdown frames).
273 bool paced = false;
274 bool pacedChecked = false;
275 int pacedCheckCountdown = 0;
276 // Serializes all use/mutation of sockets/curSocket (errCount is atomic
277 // and accessed outside the lock). A worker that outlives its frame
278 // (stuck send) can still be inside SendMessages when the next frame's
279 // worker gets the same destination, and SendMessages rotates curSocket
280 // and grows the sockets vector on EAGAIN.
281 std::mutex sendLock;
282
283 // Pacing-budget accounting, read and reset by CheckPacingBudget().
284 // pacedRate is the rate (bytes/sec) the kernel clocks this destination at,
285 // 0 while it isn't paced. framesBehind counts frames that were folded
286 // into a still-queued item or left out because the destination was still
287 // busy (late), or whose send came up short. Counted per destination per
288 // frame, so the increments are relaxed: statistics, nothing published
289 // through them.
290 std::atomic_uint pacedRate{ 0 };
291 std::atomic<uint64_t> bytesQueued{ 0 };
292 std::atomic_uint framesQueued{ 0 };
293 std::atomic_uint framesBehind{ 0 };
294 // the size of the last frame queued for this destination, which a frame
295 // it misses would have been; frame thread only, under socketMutex
296 uint64_t lastFrameBytes = 0;
297 // only the stats thread touches these, and only one runs at a time.
298 // budgetWarning is the pacing-budget banner raised for this destination;
299 // it comes down with this record (see ~SendSocketInfo).
300 std::chrono::steady_clock::time_point budgetCheckTime = std::chrono::steady_clock::now();
301 std::string budgetWarning;
302
303 std::atomic<long long> lastSendErrLog[SEND_ERR_SITES] = {};
304 };
305
306 3667 static bool ShouldLogSendError(SendSocketInfo* si, SendErrLogSite site) {
307
1/2
✓ Branch 0 taken 3667 times.
✗ Branch 1 not taken.
 3667 long long now = GetMonotonicTimeMS(); // the wall clock stepping back silenced it
308 3667 long long last = si->lastSendErrLog[site].load();
309
5/6
✓ Branch 0 taken 3632 times.
✓ Branch 1 taken 35 times.
✓ Branch 2 taken 44 times.
✓ Branch 3 taken 3588 times.
✓ Branch 4 taken 79 times.
✗ Branch 5 not taken.
 7413 return (last == 0 || now - last >= SEND_ERR_LOG_INTERVAL_MS) && si->lastSendErrLog[site].compare_exchange_strong(last, now);
310 }
311
312 108916 static uint64_t MessageBytes(const std::vector<struct mmsghdr>& msgs) {
313 108916 uint64_t bytes = 0;
314
2/2
✓ Branch 0 taken 889973 times.
✓ Branch 1 taken 108916 times.
 998889 for (auto& m : msgs) {
315
2/2
✓ Branch 0 taken 1773683 times.
✓ Branch 1 taken 889973 times.
 2663656 for (size_t v = 0; v < m.msg_hdr.msg_iovlen; v++)
316 1773683 bytes += m.msg_hdr.msg_iov[v].iov_len;
317 }
318 108916 return bytes;
319 }
320
321 // A destination that needs more than its pacing rate can't be fixed from here:
322 // the kernel clocks its packets out at that rate, so each frame finishes late
323 // or is cut off at the send budget, and the lights on that controller stutter.
324 // Nothing else tells the user. The packets wait in our own socket, not in a
325 // queue that drops them, so the local-drop check never sees them.
326 static constexpr int PACING_BUDGET_WARNING_ID = 68;
327 // Re-raised on every check (every 1500 frames, 40-75 s at typical rates) and
328 // removed by the first clean one, so this only matters once output stops: the
329 // banner then ages out instead of outliving the show.
330 static constexpr int PACING_BUDGET_WARNING_TIMEOUT_S = 300;
331
332 3 static void RaisePacingBudgetWarning(SendSocketInfo* si, const std::string& msg) {
333
2/6
✗ Branch 0 not taken.
✓ Branch 1 taken 3 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 3 times.
 3 if (!si->budgetWarning.empty() && si->budgetWarning != msg)
334 ✗ WarningHolder::RemoveWarning(PACING_BUDGET_WARNING_ID, si->budgetWarning);
335 3 si->budgetWarning = msg;
336
5/10
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 3 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 3 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 6 times.
✓ Branch 7 taken 3 times.
✗ Branch 8 not taken.
✗ Branch 9 not taken.
 15 WarningHolder::AddWarningTimeout(PACING_BUDGET_WARNING_TIMEOUT_S, PACING_BUDGET_WARNING_ID, msg,
337 { { "fixUrl", "channeloutputs.php#tab-e131" }, { "fixText", "Channel Outputs" } });
338
2/8
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 3 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
✗ Branch 7 not taken.
 6 }
339
340 903 static void ClearPacingBudgetWarning(SendSocketInfo* si) {
341
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 900 times.
 903 if (!si->budgetWarning.empty()) {
342
2/4
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 3 times.
✗ Branch 3 not taken.
 3 WarningHolder::RemoveWarning(PACING_BUDGET_WARNING_ID, si->budgetWarning);
343 3 si->budgetWarning.clear();
344 }
345 903 }
346
347 // A destination's record is dropped whenever its sockets are: by Close()
348 // when the outputs reload, and by an address change. Its banner goes with
349 // it; a destination still over budget raises it again from its new record.
350 // Clearing it from ~UDPOutput instead found the map already emptied by
351 // Close(), and the banner stayed for its whole timeout. The last reference
352 // can be a worker's or the stats thread's, so this is the one place that
353 // sees every record go.
354 880 SendSocketInfo::~SendSocketInfo() {
355 880 ClearPacingBudgetWarning(this);
356
2/2
✓ Branch 0 taken 872 times.
✓ Branch 1 taken 8 times.
 880 if (!preventClose) {
357
2/2
✓ Branch 0 taken 872 times.
✓ Branch 1 taken 872 times.
 1744 for (int x : sockets) {
358 872 close(x);
359 }
360 }
361 880 }
362
363 // A unicast destination as the frame thread saw it; the address is formatted
364 // there because HexToIP() is a UDPOutput member.
365 struct PacingBudgetDest {
366 unsigned int key;
367 std::string ip;
368 std::shared_ptr<SendSocketInfo> info;
369 };
370 static void CheckPacingBudget(const std::vector<PacingBudgetDest>& dests,
371 const std::list<UDPOutputData*>& outputs);
372
373 793 UDPOutputMessages::UDPOutputMessages() {
374 793 }
375 793 UDPOutputMessages::~UDPOutputMessages() {
376 793 sendSockets.clear();
377 793 }
378 13096 int UDPOutputMessages::GetSocket(unsigned int key) {
379 13096 std::shared_ptr<SendSocketInfo>& info = sendSockets[key];
380
2/2
✓ Branch 0 taken 11615 times.
✓ Branch 1 taken 1481 times.
 13096 if (info) {
381
1/2
✓ Branch 0 taken 11615 times.
✗ Branch 1 not taken.
 11615 std::lock_guard<std::mutex> lg(info->sendLock);
382
1/2
✓ Branch 0 taken 11615 times.
✗ Branch 1 not taken.
 11615 if (!info->sockets.empty()) {
383 11615 return info->sockets[0];
384 }
385
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 11615 times.
 11615 }
386 1481 return -1;
387 }
388 14 void UDPOutputMessages::ForceSocket(unsigned int key, int socket, bool preventClose) {
389
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 14 times.
 14 if (socket < 0) {
390 // left for findOrCreateSocket() to make on a later frame
391 ✗ return;
392 }
393
1/2
✓ Branch 0 taken 14 times.
✗ Branch 1 not taken.
 14 std::shared_ptr<SendSocketInfo>& info = sendSockets[key];
394
1/2
✓ Branch 0 taken 14 times.
✗ Branch 1 not taken.
 14 if (!info) {
395
1/2
✓ Branch 0 taken 14 times.
✗ Branch 1 not taken.
 14 info = std::make_shared<SendSocketInfo>();
396 }
397
1/2
✓ Branch 0 taken 14 times.
✗ Branch 1 not taken.
 14 std::lock_guard<std::mutex> lg(info->sendLock);
398
1/2
✓ Branch 0 taken 14 times.
✗ Branch 1 not taken.
 14 if (!info->preventClose) {
399
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 14 times.
 14 for (int x = 0; x < info->sockets.size(); x++) {
400 ✗ close(info->sockets[x]);
401 }
402 }
403 14 info->preventClose = preventClose;
404 14 info->sockets.clear();
405
1/2
✓ Branch 0 taken 14 times.
✗ Branch 1 not taken.
 14 info->sockets.push_back(socket);
406 14 }
407 929119 std::vector<struct mmsghdr>& UDPOutputMessages::GetMessages(unsigned int key) {
408
7/8
✓ Branch 0 taken 924659 times.
✓ Branch 1 taken 4460 times.
✓ Branch 2 taken 924659 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 119561 times.
✓ Branch 5 taken 805098 times.
✓ Branch 6 taken 119561 times.
✓ Branch 7 taken 809558 times.
 929119 if (recordKeys && std::find(recordKeys->begin(), recordKeys->end(), key) == recordKeys->end()) {
409 119561 recordKeys->push_back(key);
410 }
411 929119 return messages[key];
412 }
413 109885 void UDPOutputMessages::clearMessages() {
414
2/2
✓ Branch 0 taken 118711 times.
✓ Branch 1 taken 109885 times.
 228596 for (auto& m : messages) {
415 118711 m.second.clear();
416 }
417 109885 }
418 798 void UDPOutputMessages::clearSockets() {
419 // sockets are shared_ptrs; any worker thread still mid-send keeps its
420 // SendSocketInfo alive until it finishes
421 798 sendSockets.clear();
422 798 }
423
424 // Keys whose packets go to multicast groups. Their sockets need
425 // IP_MULTICAST_IF, or the packets leave by the default route instead of the
426 // output interface. E1.31 sync is always sent to 239.255.<universe>.
427 1045 static bool isMulticastKey(unsigned int key) {
428
5/6
✓ Branch 0 taken 1032 times.
✓ Branch 1 taken 13 times.
✓ Branch 2 taken 1032 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 3 times.
✓ Branch 5 taken 1029 times.
 1045 return key == MULTICAST_MESSAGES_KEY || key == LATE_MULTICAST_MESSAGES_KEY || key == E131_SYNC_KEY;
429 }
430
431 static constexpr int LOCAL_DROP_WARNING_ID = 69;
432 static const std::string LOCAL_DROP_WARNING = "The network interface is dropping outgoing packets before they reach the wire. The link may be down, saturated, or misconfigured. See logs for details.";
433
434 UDPOutput* UDPOutput::INSTANCE = nullptr;
435 std::mutex UDPOutput::externalLock;
436 std::map<const void*, std::function<UDPOutputData*()>> UDPOutput::externalFactories;
437
438 949 UDPOutputData::UDPOutputData(const Json::Value& config) :
439 949 valid(true),
440 949 type(0),
441 949 monitor(true),
442 949 lastData(nullptr),
443 949 skippedFrames(0) {
444
3/4
✓ Branch 0 taken 949 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 947 times.
✓ Branch 3 taken 2 times.
 949 if (config.isMember("description")) {
445
2/4
✓ Branch 0 taken 947 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 947 times.
✗ Branch 3 not taken.
 947 description = config["description"].asString();
446 }
447
2/4
✓ Branch 0 taken 949 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 949 times.
✗ Branch 3 not taken.
 949 startChannel = config["startChannel"].asInt();
448
2/4
✓ Branch 0 taken 949 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 949 times.
✗ Branch 3 not taken.
 949 channelCount = config["channelCount"].asInt();
449
3/4
✓ Branch 0 taken 949 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 897 times.
✓ Branch 3 taken 52 times.
 949 if (config.isMember("active")) {
450
2/4
✓ Branch 0 taken 897 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 897 times.
✗ Branch 3 not taken.
 897 active = config["active"].asInt();
451 } else {
452 52 active = 1;
453 }
454
3/4
✓ Branch 0 taken 949 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 897 times.
✓ Branch 3 taken 52 times.
 949 if (config.isMember("type")) {
455
2/4
✓ Branch 0 taken 897 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 897 times.
✗ Branch 3 not taken.
 897 type = config["type"].asInt();
456 }
457
2/4
✓ Branch 0 taken 949 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 949 times.
✗ Branch 3 not taken.
 949 if (config.isMember("address")) {
458
2/4
✓ Branch 0 taken 949 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 949 times.
✗ Branch 3 not taken.
 949 ipAddress = config["address"].asString();
459 }
460
3/4
✓ Branch 0 taken 949 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 897 times.
✓ Branch 3 taken 52 times.
 949 if (config.isMember("monitor")) {
461
2/4
✓ Branch 0 taken 897 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 897 times.
✗ Branch 3 not taken.
 897 monitor = config["monitor"].asInt() ? true : false;
462 }
463
3/4
✓ Branch 0 taken 949 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 897 times.
✓ Branch 3 taken 52 times.
 949 if (config.isMember("deDuplicate")) {
464
2/4
✓ Branch 0 taken 897 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 897 times.
✗ Branch 3 not taken.
 897 deDuplicate = config["deDuplicate"].asInt() ? true : false;
465 }
466 // per-destination pacing override; absent (or -1) keeps the global rate
467
3/4
✓ Branch 0 taken 949 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✓ Branch 3 taken 947 times.
 949 if (config.isMember("pacingRate")) {
468
2/4
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
 2 pacingRateMbps = config["pacingRate"].asInt();
469 }
470 949 }
471 1898 UDPOutputData::~UDPOutputData() {
472
2/2
✓ Branch 0 taken 5 times.
✓ Branch 1 taken 944 times.
 1898 if (lastData) {
473 10 free(lastData);
474 }
475 1898 }
476
477 static const std::string UNKNOWN_TYPE = "UDP";
478 ✗ const std::string& UDPOutputData::GetOutputTypeString() const {
479 ✗ return UNKNOWN_TYPE;
480 }
481
482 947 in_addr_t UDPOutputData::toInetAddr(const std::string& ipAddress, bool& valid) {
483 947 valid = true;
484 947 bool isAlpha = false;
485
2/2
✓ Branch 0 taken 8722 times.
✓ Branch 1 taken 947 times.
 9669 for (int x = 0; x < ipAddress.length(); x++) {
486 8722 isAlpha |= isalpha(ipAddress[x]);
487 }
488
489
2/2
✓ Branch 0 taken 13 times.
✓ Branch 1 taken 934 times.
 947 if (isAlpha) {
490 // ResolveHostToIPv4() is thread-safe; gethostbyname() returns a shared
491 // static hostent and can be corrupted by concurrent resolves on other
492 // threads. It also caches the answer, so a destination configured by a
493 // name that no longer resolves costs its multi-second timeout once at
494 // startup rather than once per caller.
495 13 uint32_t addr = 0;
496
3/4
✓ Branch 0 taken 13 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 11 times.
✓ Branch 3 taken 2 times.
 13 if (!ResolveHostToIPv4(ipAddress, addr)) {
497
1/2
✓ Branch 0 taken 11 times.
✗ Branch 1 not taken.
 11 LogErr(VB_CHANNELOUT,
498 "Error looking up hostname: %s\n",
499 ipAddress.c_str());
500 11 valid = false;
501 11 return 0;
502 }
503 2 return addr;
504 }
505 934 return inet_addr(ipAddress.c_str());
506 }
507
508 119461 void UDPOutputData::SaveFrame(unsigned char* channelData, int len) {
509
2/2
✓ Branch 0 taken 85 times.
✓ Branch 1 taken 119376 times.
 119461 if (deDuplicate) {
510
2/2
✓ Branch 0 taken 5 times.
✓ Branch 1 taken 80 times.
 85 if (lastData == nullptr) {
511 5 lastData = (unsigned char*)calloc(1, len);
512 }
513 /*
514 printf("Saving: %d %2X%2X%2X %2X%2X%2X %2X%2X%2X %2X%2X%2X\n", len,
515 channelData[0], channelData[1], channelData[2], channelData[3], channelData[4], channelData[5],
516 channelData[6], channelData[7], channelData[8], channelData[9], channelData[10], channelData[11]);
517 */
518 85 memcpy(lastData, channelData, len);
519 }
520 119461 }
521
522 928795 bool UDPOutputData::NeedToOutputFrame(unsigned char* channelData, int startChannel, int savedIdx, int count) {
523
4/4
✓ Branch 0 taken 7249 times.
✓ Branch 1 taken 921546 times.
✓ Branch 2 taken 7235 times.
✓ Branch 3 taken 14 times.
 928795 if (deDuplicate && skippedFrames < 10) {
524
2/2
✓ Branch 0 taken 2003 times.
✓ Branch 1 taken 5232 times.
 7235 if (lastData == nullptr) {
525 2003 return true;
526 }
527
2/2
✓ Branch 0 taken 5784591 times.
✓ Branch 1 taken 4167 times.
 5788758 for (int x = 0; x < count; x++) {
528
2/2
✓ Branch 0 taken 1065 times.
✓ Branch 1 taken 5783526 times.
 5784591 if (channelData[x + savedIdx + startChannel] != lastData[x + savedIdx]) {
529 /*
530 printf("ND: %d %2X%2X%2X %2X%2X%2X %2X%2X%2X %2X%2X%2X\n", x,
531 lastData[0], lastData[1], lastData[2], lastData[3], lastData[4], lastData[5],
532 lastData[6], lastData[7], lastData[8], lastData[9], lastData[10], lastData[11]);
533 printf("%d %d %2X%2X%2X %2X%2X%2X %2X%2X%2X %2X%2X%2X\n", start, startChannel, x,
534 channelData[start + startChannel + 0], channelData[start + startChannel + 1], channelData[start + startChannel + 2],
535 channelData[start + startChannel + 3], channelData[start + startChannel + 4], channelData[start + startChannel + 5],
536 channelData[start + startChannel + 6], channelData[start + startChannel + 7], channelData[start + startChannel + 8],
537 channelData[start + startChannel + 9], channelData[start + startChannel + 10], channelData[start + startChannel + 11]);
538 printf("New data %d %d %d %X %X\n", x, startChannel, savedIdx, channelData[x + savedIdx + startChannel], lastData[x + savedIdx]);
539 */
540 1065 return true;
541 }
542 }
543 4167 return false;
544 }
545 921560 return true;
546 }
547
548 793 UDPOutput::UDPOutput(unsigned int startChannel, unsigned int channelCount) :
549 793 networkCallbackId(-1),
550 793 doneWorkCount(0),
551 793 numWorkThreads(0),
552 793 workGeneration(0),
553 793 runWorkThreads(true),
554 793 useThreadedOutput(true),
555
3/6
✓ Branch 0 taken 793 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 793 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 793 times.
✗ Branch 5 not taken.
 1586 blockingOutput(false) {
556 793 INSTANCE = this;
557 793 }
558 4758 UDPOutput::~UDPOutput() {
559 1586 StopPinging();
560 {
561 // under workMutex: a worker between its runWorkThreads check and its
562 // wait() would otherwise miss this notify and sleep forever, and the
563 // loop below would then spin with outputThreadLock held
564 1586 std::lock_guard<std::mutex> lk(workMutex);
565 1586 runWorkThreads = false;
566 1586 }
567 1586 workSignal.notify_all();
568
569 {
570 1586 std::lock_guard<std::mutex> lk(externalLock);
571 1586 INSTANCE = nullptr;
572 1586 }
573 1586 NetworkMonitor::INSTANCE.removeCallback(networkCallbackId);
574
2/2
✓ Branch 0 taken 435 times.
✓ Branch 1 taken 793 times.
 2456 while (numWorkThreads) {
575 870 std::this_thread::sleep_for(std::chrono::milliseconds(1));
576 }
577
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 793 times.
 1586 while (statCheckRunning) {
578 ✗ std::this_thread::sleep_for(std::chrono::milliseconds(1));
579 }
580 // Requests in flight capture this (the HEAD probe) or an output deleted
581 // below (Twinkly, which cancels its own). Cancel ours rather than
582 // draining every request in the process, which waited out unrelated
583 // ones - up to the 10s curl timeout each - with outputThreadLock held.
584 // The destructor runs on the main loop, where cancelling is allowed.
585 1586 CurlManager::INSTANCE.cancelRequests(curlOwner);
586
2/2
✓ Branch 0 taken 947 times.
✓ Branch 1 taken 793 times.
 3480 for (auto a : outputs) {
587
1/2
✓ Branch 0 taken 947 times.
✗ Branch 1 not taken.
 1894 delete a;
588 }
589 3172 }
590
591 // Each output sends straight out of the sequence buffer from the range its
592 // startChannel/channelCount (times universeCount) describe. A startChannel
593 // of 0 made that range start at -1, and a large universeCount overflowed the
594 // int it is computed in, wrapping it to the "nothing to send" form the range
595 // check lets through; both then read outside the buffer every frame. Worked
596 // out in 64 bits from the config, before the output allocates per universe.
597 899 static bool UniverseRangeFits(const Json::Value& s, int type) {
598
2/4
✓ Branch 0 taken 899 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 899 times.
✗ Branch 3 not taken.
 899 int64_t start = s["startChannel"].asInt64();
599
2/4
✓ Branch 0 taken 899 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 899 times.
✗ Branch 3 not taken.
 899 int64_t count = s["channelCount"].asInt64();
600 899 int64_t universes = 1;
601
8/8
✓ Branch 0 taken 716 times.
✓ Branch 1 taken 183 times.
✓ Branch 2 taken 715 times.
✓ Branch 3 taken 1 times.
✓ Branch 4 taken 714 times.
✓ Branch 5 taken 1 times.
✓ Branch 6 taken 3 times.
✓ Branch 7 taken 711 times.
 899 if (type <= 3 || type == 6 || type == 7 || type == 9) {
602 // E1.31, ArtNet and KiNet repeat channelCount per universe/port
603
3/4
✓ Branch 0 taken 188 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 46 times.
✓ Branch 3 taken 142 times.
 188 if (s.isMember("universeCount")) {
604
2/4
✓ Branch 0 taken 46 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 46 times.
✗ Branch 3 not taken.
 46 universes = std::max<int64_t>(1, s["universeCount"].asInt64());
605 }
606 }
607
6/10
✓ Branch 0 taken 899 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 899 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 899 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 899 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 898 times.
✓ Branch 9 taken 1 times.
 899 bool active = !s.isMember("active") || s["active"].asInt();
608
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 898 times.
 899 if (!active) {
609 1 return true;
610 }
611
4/6
✓ Branch 0 taken 897 times.
✓ Branch 1 taken 1 times.
✓ Branch 2 taken 897 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 897 times.
✗ Branch 5 not taken.
 898 if (start >= 1 && count >= 0 && universes <= 64000 &&
612
2/2
✓ Branch 0 taken 896 times.
✓ Branch 1 taken 1 times.
 897 start - 1 + count * universes <= (int64_t)FPPD_MAX_CHANNELS) {
613 896 return true;
614 }
615
4/8
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 2 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 2 times.
✗ Branch 7 not taken.
 2 std::string desc = s.isMember("description") ? s["description"].asString() : std::string();
616
3/10
✗ Branch 0 not taken.
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✓ Branch 6 taken 2 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 2 times.
✗ Branch 9 not taken.
 4 std::string warning = "Network output " + (desc.empty() ? s["address"].asString() : desc) +
617
2/4
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
 8 " needs channels outside 1-" + std::to_string(FPPD_MAX_CHANNELS) +
618
6/12
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 2 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 2 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 2 times.
✗ Branch 9 not taken.
✓ Branch 10 taken 2 times.
✗ Branch 11 not taken.
 8 " (start " + std::to_string(start) + ", " + std::to_string(count) +
619
4/8
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 2 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 2 times.
✗ Branch 7 not taken.
 6 " channels x " + std::to_string(universes) + "). Output disabled.";
620
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 AddOutputConfigWarning(26, warning);
621
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 LogErr(VB_CHANNELOUT, "%s\n", warning.c_str());
622 2 return false;
623 2 }
624
625 793 int UDPOutput::Init(Json::Value config) {
626 // a reload may have dropped the ArtNet output that last wanted the socket
627
1/2
✓ Branch 0 taken 793 times.
✗ Branch 1 not taken.
 793 ResetArtNetSocketFailure();
628
2/4
✓ Branch 0 taken 793 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 793 times.
✗ Branch 3 not taken.
 793 enabled = config["enabled"].asInt();
629 793 hasArtNet = false;
630
4/6
✓ Branch 0 taken 1692 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1692 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 901 times.
✓ Branch 5 taken 791 times.
 1692 for (int i = 0; i < config["universes"].size(); i++) {
631
3/6
✓ Branch 0 taken 901 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 901 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 901 times.
✗ Branch 5 not taken.
 901 Json::Value s = config["universes"][i];
632
4/4
✓ Branch 0 taken 900 times.
✓ Branch 1 taken 1 times.
✓ Branch 2 taken 899 times.
✓ Branch 3 taken 1 times.
 901 int type = s["type"].asInt();
633
3/4
✓ Branch 0 taken 899 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✓ Branch 3 taken 897 times.
 899 if (!UniverseRangeFits(s, type)) {
634 2 continue;
635 }
636
5/6
✓ Branch 0 taken 158 times.
✓ Branch 1 taken 26 times.
✓ Branch 2 taken 696 times.
✓ Branch 3 taken 2 times.
✓ Branch 4 taken 15 times.
✗ Branch 5 not taken.
 897 switch (type) {
637 158 case 0:
638 case 1:
639 // E1.31 types
640
4/10
✓ Branch 0 taken 158 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 158 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 158 times.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
✓ Branch 7 taken 158 times.
✗ Branch 8 not taken.
✗ Branch 9 not taken.
 158 outputs.push_back(new E131OutputData(s));
641 158 break;
642 26 case 2:
643 case 3:
644 case 9:
645 // ArtNet types
646
4/10
✓ Branch 0 taken 26 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 26 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 26 times.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
✓ Branch 7 taken 26 times.
✗ Branch 8 not taken.
✗ Branch 9 not taken.
 26 outputs.push_back(new ArtNetOutputData(s));
647 26 hasArtNet = true;
648 26 break;
649 696 case 4:
650 case 5:
651 // DDP types
652
4/10
✓ Branch 0 taken 696 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 696 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 696 times.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
✓ Branch 7 taken 696 times.
✗ Branch 8 not taken.
✗ Branch 9 not taken.
 696 outputs.push_back(new DDPOutputData(s));
653 696 break;
654 2 case 6:
655 case 7:
656 // KiNet types
657
4/10
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 2 times.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
✓ Branch 7 taken 2 times.
✗ Branch 8 not taken.
✗ Branch 9 not taken.
 2 outputs.push_back(new KiNetOutputData(s));
658 2 break;
659 15 case 8:
660 // Twinkly types
661
4/10
✓ Branch 0 taken 15 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 15 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 15 times.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
✓ Branch 7 taken 15 times.
✗ Branch 8 not taken.
✗ Branch 9 not taken.
 15 outputs.push_back(new TwinklyOutputData(s));
662 15 break;
663 ✗ default:
664 ✗ LogErr(VB_CHANNELOUT, "Unknown IP output type %d\n", type);
665 ✗ break;
666 }
667
668
3/4
✓ Branch 0 taken 893 times.
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 893 times.
 897 needsBroadcast |= (type == 0 || type == 2);
669
2/2
✓ Branch 0 taken 897 times.
✓ Branch 1 taken 2 times.
 901 }
670
3/4
✓ Branch 0 taken 791 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 29 times.
✓ Branch 3 taken 762 times.
 791 if (config.isMember("threaded")) {
671
2/4
✓ Branch 0 taken 29 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 29 times.
✗ Branch 3 not taken.
 29 int style = config["threaded"].asInt();
672
2/4
✓ Branch 0 taken 29 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 29 times.
 29 useThreadedOutput = style == 1 || style == 3;
673
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 29 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 29 blockingOutput = style == 0 || style == 1;
674 }
675
3/4
✓ Branch 0 taken 791 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 22 times.
✓ Branch 3 taken 769 times.
 791 if (config.isMember("interface")) {
676
2/4
✓ Branch 0 taken 22 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 22 times.
✗ Branch 3 not taken.
 22 std::string configured = config["interface"].asString();
677
5/6
✓ Branch 0 taken 22 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 21 times.
✓ Branch 3 taken 1 times.
✓ Branch 4 taken 21 times.
✓ Branch 5 taken 1 times.
 22 if (configured.empty() || isValidInterfaceName(configured)) {
678
1/2
✓ Branch 0 taken 21 times.
✗ Branch 1 not taken.
 21 outInterface = configured;
679 } else {
680 // Garbage never produced working output (address/sysfs lookups fail
681 // on it) and reached a shell below -- fall through to auto-detect.
682
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 LogErr(VB_CHANNELOUT, "UDPOutput: ignoring invalid interface '%s'\n", configured.c_str());
683 }
684 22 }
685
686 1582 std::set<std::string> myIps;
687 // get all the addresses
688 struct ifaddrs *interfaces, *tmp;
689 791 getifaddrs(&interfaces);
690 791 tmp = interfaces;
691 // loop through all the interfaces and get the addresses
692 char address[16];
693 791 memset(address, 0, sizeof(address));
694
1/2
✓ Branch 0 taken 791 times.
✗ Branch 1 not taken.
 1582 std::string firstInterface = "";
695
2/2
✓ Branch 0 taken 4066 times.
✓ Branch 1 taken 791 times.
 4857 while (tmp) {
696
3/4
✓ Branch 0 taken 4066 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1607 times.
✓ Branch 3 taken 2459 times.
 4066 if (tmp->ifa_addr && tmp->ifa_addr->sa_family == AF_INET) {
697
1/2
✓ Branch 0 taken 1607 times.
✗ Branch 1 not taken.
 1607 GetInterfaceAddress(tmp->ifa_name, address, NULL, NULL);
698
1/2
✓ Branch 0 taken 1607 times.
✗ Branch 1 not taken.
 1607 myIps.emplace(address);
699
7/8
✓ Branch 0 taken 1607 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1521 times.
✓ Branch 3 taken 86 times.
✓ Branch 4 taken 750 times.
✓ Branch 5 taken 771 times.
✓ Branch 6 taken 750 times.
✓ Branch 7 taken 857 times.
 1607 if ((outInterface == "") && (strcmp(tmp->ifa_name, "lo"))) {
700
2/4
✓ Branch 0 taken 750 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 750 times.
✗ Branch 3 not taken.
 750 if (firstInterface == "") {
701
1/2
✓ Branch 0 taken 750 times.
✗ Branch 1 not taken.
 750 firstInterface = tmp->ifa_name;
702 }
703
1/2
✓ Branch 0 taken 750 times.
✗ Branch 1 not taken.
 750 if (tmp->ifa_name[0] == 'e') {
704
1/2
✓ Branch 0 taken 750 times.
✗ Branch 1 not taken.
 750 outInterface = tmp->ifa_name;
705 }
706 }
707 }
708 4066 tmp = tmp->ifa_next;
709 }
710 791 freeifaddrs(interfaces);
711
3/4
✓ Branch 0 taken 791 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 20 times.
✓ Branch 3 taken 771 times.
 791 if (outInterface == "") {
712
1/2
✓ Branch 0 taken 20 times.
✗ Branch 1 not taken.
 20 outInterface = firstInterface;
713 }
714
3/4
✓ Branch 0 taken 791 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 20 times.
✓ Branch 3 taken 771 times.
 791 if (outInterface == "") {
715
1/2
✓ Branch 0 taken 20 times.
✗ Branch 1 not taken.
 20 outInterface = "eth0";
716 }
717
718 // Build the per-destination pacing overrides keyed by controller IP. The
719 // send socket is shared per destination IP, so when multiple outputs target
720 // the same controller with different overrides we keep the most conservative
721 // (lowest) cap; an explicit "unpaced" (0 Mbps) is stored as ~0U so any real
722 // rate on the same IP wins over it.
723
2/2
✓ Branch 0 taken 897 times.
✓ Branch 1 taken 791 times.
 1688 for (auto o : outputs) {
724
5/6
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 895 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 2 times.
✓ Branch 4 taken 895 times.
✓ Branch 5 taken 2 times.
 897 if (o->pacingRateMbps < 0 || o->ipAddress.empty()) {
725 895 continue;
726 }
727 2 bool valid = false;
728
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 in_addr_t addr = UDPOutputData::toInetAddr(o->ipAddress, valid);
729
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2 times.
 2 if (!valid) {
730 ✗ continue;
731 }
732
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 1 times.
 2 unsigned int rate = o->pacingRateMbps == 0 ? ~0U : (unsigned int)(o->pacingRateMbps * 1000000ULL / 8);
733
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 auto it = perDestPacingRate.find(addr);
734
2/6
✗ Branch 0 not taken.
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 2 times.
✗ Branch 5 not taken.
 2 if (it == perDestPacingRate.end() || rate < it->second) {
735
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 perDestPacingRate[addr] = rate;
736 }
737 }
738
739 #ifndef PLATFORM_OSX
740 // Kernel-level packet pacing. Bursting a full frame at line rate overruns
741 // switch buffers on the 1G->100M speed mismatch to most pixel controllers
742 // and the drops are silent. With the fq qdisc installed (fppinit does this
743 // at boot), SO_MAX_PACING_RATE lets the kernel clock each destination's
744 // packets out at a rate the far end can actually absorb.
745 // pacing rate lives in the outputs config (hot-reloadable with the rest
746 // of it); fall back to the old global setting for configs saved before
747 // the UI moved, then to the 90Mbps default
748
6/10
✓ Branch 0 taken 791 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 10 times.
✓ Branch 3 taken 781 times.
✓ Branch 4 taken 10 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 10 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 781 times.
✗ Branch 9 not taken.
 791 int pacingMbps = config.isMember("pacingRate") ? config["pacingRate"].asInt() : getSettingInt("UDPPacingRate", 90);
749 // per-output overrides can request pacing even when the global rate is
750 // Disabled (e.g. only a single ESP32 needs a cap); install fq and arm the
751 // machinery if either the global rate or any override wants it
752 791 bool anyPerDestPacing = false;
753
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 790 times.
 791 for (const auto& e : perDestPacingRate) {
754
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 if (e.second != ~0U) {
755 1 anyPerDestPacing = true;
756 1 break;
757 }
758 }
759 // a reload re-reads the qdiscs, and warms the cache here on the main
760 // thread rather than in the first frame
761
1/2
✓ Branch 0 taken 791 times.
✗ Branch 1 not taken.
 791 clearFQCache();
762
3/4
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 788 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 3 times.
 791 if (pacingMbps > 0 || anyPerDestPacing) {
763
3/4
✓ Branch 0 taken 788 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 25 times.
✓ Branch 3 taken 763 times.
 788 if (!anyEthernetHasFQ(outInterface)) {
764 // fppinit only installs fq at boot; if the user enabled pacing
765 // since then, install it now so the setting takes effect on the
766 // fppd restart it advertises rather than silently needing a
767 // reboot. Only interfaces with an IPv4 address can carry the
768 // routed UDP we pace - an address-less interface dedicated to a
769 // raw layer-2 output (ColorLight) keeps its default qdisc.
770
5/8
✓ Branch 0 taken 25 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 25 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 30 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 30 times.
✓ Branch 7 taken 25 times.
 55 for (const auto& entry : std::filesystem::directory_iterator("/sys/class/net/")) {
771
2/4
✓ Branch 0 taken 30 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 30 times.
✗ Branch 3 not taken.
 30 std::string dev = entry.path().filename();
772
7/10
✓ Branch 0 taken 30 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 5 times.
✓ Branch 3 taken 25 times.
✓ Branch 4 taken 5 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 5 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 5 times.
✓ Branch 9 taken 25 times.
 30 if (IsShowEthernetInterface(dev) && InterfaceHasRoutableIPv4(dev)) {
773
1/2
✓ Branch 0 taken 5 times.
✗ Branch 1 not taken.
 5 InstallShowTrafficQdisc(dev);
774 }
775 55 }
776
1/2
✓ Branch 0 taken 25 times.
✗ Branch 1 not taken.
 25 clearFQCache();
777 }
778
3/4
✓ Branch 0 taken 788 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 768 times.
✓ Branch 3 taken 20 times.
 788 if (anyEthernetHasFQ(outInterface)) {
779 768 pacingEnabled = true;
780 // 0 when the global rate is Disabled: destinations without an
781 // override then stay unpaced while overridden ones still get capped
782
1/2
✓ Branch 0 taken 768 times.
✗ Branch 1 not taken.
 768 pacingRate = pacingMbps > 0 ? (unsigned int)(pacingMbps * 1000000ULL / 8) : 0;
783 // slowest positive rate in play, for the drain-sleep math below
784 768 pacingDrainRate = pacingRate;
785
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 768 times.
 770 for (const auto& e : perDestPacingRate) {
786
4/6
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 1 times.
✓ Branch 2 taken 1 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 1 times.
✗ Branch 5 not taken.
 2 if (e.second != ~0U && (pacingDrainRate == 0 || e.second < pacingDrainRate)) {
787 1 pacingDrainRate = e.second;
788 }
789 }
790
1/2
✓ Branch 0 taken 768 times.
✗ Branch 1 not taken.
 768 if (pacingMbps > 0) {
791
1/2
✓ Branch 0 taken 768 times.
✗ Branch 1 not taken.
 768 LogInfo(VB_CHANNELOUT, "UDP output pacing enabled at %d Mbps per destination (%d per-destination override(s))\n",
792 pacingMbps, (int)perDestPacingRate.size());
793 } else {
794 ✗ LogInfo(VB_CHANNELOUT, "UDP output pacing enabled for %d per-destination override(s), global rate disabled\n",
795 (int)perDestPacingRate.size());
796 }
797 } else {
798
1/2
✓ Branch 0 taken 20 times.
✗ Branch 1 not taken.
 20 LogInfo(VB_CHANNELOUT, "UDP output pacing requested but %s does not have the fq qdisc, using buffer drain pacing\n",
799 outInterface.c_str());
800 }
801 788 } else {
802
1/2
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
 3 LogInfo(VB_CHANNELOUT, "UDP output pacing disabled, using buffer drain pacing\n");
803 }
804 #endif
805
806
1/2
✓ Branch 0 taken 791 times.
✗ Branch 1 not taken.
 791 bool disableFakeBridges = getSettingInt("DisableFakeNetworkBridges");
807
808
2/2
✓ Branch 0 taken 897 times.
✓ Branch 1 taken 791 times.
 1688 for (auto o : outputs) {
809
7/8
✓ Branch 0 taken 897 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 893 times.
✓ Branch 3 taken 4 times.
✓ Branch 4 taken 881 times.
✓ Branch 5 taken 12 times.
✓ Branch 6 taken 881 times.
✓ Branch 7 taken 16 times.
 897 if (o->IsPingable() && o->active) {
810
2/2
✓ Branch 0 taken 61 times.
✓ Branch 1 taken 820 times.
 881 if (!disableFakeBridges) {
811
1/2
✓ Branch 0 taken 61 times.
✗ Branch 1 not taken.
 61 std::string host = o->ipAddress;
812
3/4
✓ Branch 0 taken 61 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 8 times.
✓ Branch 3 taken 53 times.
 61 if (myIps.find(host) != myIps.end()) {
813 // trying to send UDP data to myself, that's bad. Disable
814
1/2
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
 8 std::string msg = "UDP Output set to send data to myself. Disabling ";
815
1/2
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
 8 msg += host.c_str();
816
1/2
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
 8 LogWarn(VB_CHANNELOUT, msg);
817
1/2
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
 8 AddOutputConfigWarning(6, msg);
818 8 o->active = false;
819 8 }
820 61 }
821 }
822 }
823
824 1732 std::function<void(NetworkMonitor::NetEventType, int, const std::string&)> f = [this](NetworkMonitor::NetEventType i, int up, const std::string& s) {
825
1/2
✓ Branch 0 taken 150 times.
✗ Branch 1 not taken.
 150 std::string interface = outInterface;
826
5/6
✓ Branch 0 taken 15 times.
✓ Branch 1 taken 135 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 15 times.
✓ Branch 4 taken 46 times.
✓ Branch 5 taken 89 times.
 150 if ((i == NetworkMonitor::NetEventType::NEW_ADDR && up) || i == NetworkMonitor::NetEventType::DEL_ADDR) {
827 // any address change can move routes; re-evaluate the pacing
828 // decision for every unicast destination. This covers both
829 // directions: a destination that couldn't be paced yet (route
830 // resolved via wifi while ethernet waited on DHCP) and a paced
831 // destination whose route failed over to a non-fq interface,
832 // which must be un-paced or it sends unpaced line-rate bursts
833 // with the large buffer and no drain waits. Stagger the
834 // re-checks across frames so one network event doesn't trigger
835 // a route-lookup storm in a single frame.
836
1/2
✓ Branch 0 taken 61 times.
✗ Branch 1 not taken.
 61 std::unique_lock<std::mutex> lk(socketMutex);
837 61 int stagger = 0;
838
2/2
✓ Branch 0 taken 84 times.
✓ Branch 1 taken 61 times.
 145 for (auto& si : messages.sendSockets) {
839
1/2
✓ Branch 0 taken 84 times.
✗ Branch 1 not taken.
 84 if (si.second) {
840 // try_lock: blocking here (while holding socketMutex) on a
841 // straggler worker would transitively stall the frame thread
842 84 std::unique_lock<std::mutex> lg(si.second->sendLock, std::try_to_lock);
843
1/2
✓ Branch 0 taken 84 times.
✗ Branch 1 not taken.
 84 if (lg.owns_lock()) {
844 84 si.second->pacedChecked = false;
845 84 si.second->pacedCheckCountdown = stagger++ % 16;
846 }
847 84 }
848 }
849 61 }
850
7/8
✓ Branch 0 taken 22 times.
✓ Branch 1 taken 128 times.
✓ Branch 2 taken 5 times.
✓ Branch 3 taken 17 times.
✓ Branch 4 taken 5 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 5 times.
✓ Branch 7 taken 145 times.
 150 if (s == interface && i == NetworkMonitor::NetEventType::NEW_ADDR && up) {
851
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 1 times.
 5 if (!interfaceUp) {
852
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4 LogInfo(VB_CHANNELOUT, "UDP Interface %s now up\n", s.c_str());
853
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4 PingControllers(false);
854
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4 std::this_thread::sleep_for(std::chrono::milliseconds(200));
855 // the frame thread kept creating sockets while the address was
856 // missing, bound to (and multicasting out of) loopback or the
857 // old address; drop them so they are rebuilt on the new one
858
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4 interfaceUp = InitNetwork(true);
859
2/4
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 4 times.
 4 if (interfaceUp && hasArtNet) {
860 ✗ CreateArtNetSocket(localAddress.sin_addr.s_addr, true);
861 }
862 }
863
6/6
✓ Branch 0 taken 17 times.
✓ Branch 1 taken 128 times.
✓ Branch 2 taken 3 times.
✓ Branch 3 taken 14 times.
✓ Branch 4 taken 3 times.
✓ Branch 5 taken 142 times.
 145 } else if (s == interface && i == NetworkMonitor::NetEventType::DEL_ADDR) {
864 // The interface can still have an address: a DHCP lease that
865 // replaces a fallback address arrives before the old one goes, and
866 // its NEW_ADDR was ignored because the interface was up. Re-read
867 // the address, as NEW_ADDR does, or the sockets are rebuilt bound
868 // to (and multicasting out of) the address just deleted.
869
1/2
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
 3 interfaceUp = InitNetwork(true);
870
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 2 times.
 3 if (interfaceUp) {
871
1/2
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
 1 LogInfo(VB_CHANNELOUT, "UDP Interface %s address changed\n", s.c_str());
872
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 times.
 1 if (hasArtNet) {
873 ✗ CreateArtNetSocket(localAddress.sin_addr.s_addr, true);
874 }
875 } else {
876
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 LogInfo(VB_CHANNELOUT, "UDP Interface %s now down\n", s.c_str());
877 }
878
1/2
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
 3 PingControllers(false);
879 }
880 941 };
881
1/2
✓ Branch 0 taken 791 times.
✗ Branch 1 not taken.
 791 networkCallbackId = NetworkMonitor::INSTANCE.registerCallback(f);
882
883 // An interface with no address yet (DHCP still waiting, cable out) is
884 // down: the NEW_ADDR event for it then rebuilds the sockets on the real
885 // address instead of being ignored.
886
1/2
✓ Branch 0 taken 791 times.
✗ Branch 1 not taken.
 791 interfaceUp = InitNetwork();
887 {
888
1/2
✓ Branch 0 taken 791 times.
✗ Branch 1 not taken.
 791 std::lock_guard<std::mutex> l(pingHostsLock);
889
2/2
✓ Branch 0 taken 897 times.
✓ Branch 1 taken 791 times.
 1688 for (auto o : outputs) {
890
1/2
✓ Branch 0 taken 897 times.
✗ Branch 1 not taken.
 897 addPingTarget(o);
891 }
892 791 failedCount = (int)pingHosts.size();
893 791 }
894 // need to do three pings to detect down hosts
895
1/2
✓ Branch 0 taken 791 times.
✗ Branch 1 not taken.
 791 PingControllers(false);
896 791 int sleepCount = 0;
897
6/6
✓ Branch 0 taken 3395 times.
✓ Branch 1 taken 778 times.
✓ Branch 2 taken 3382 times.
✓ Branch 3 taken 13 times.
✓ Branch 4 taken 3382 times.
✓ Branch 5 taken 791 times.
 4173 while (failedCount > 0 && sleepCount < (UDP_PING_TIMEOUT + 10)) {
898
1/2
✓ Branch 0 taken 3382 times.
✗ Branch 1 not taken.
 3382 std::this_thread::sleep_for(std::chrono::milliseconds(1));
899 3382 ++sleepCount;
900 }
901
1/2
✓ Branch 0 taken 791 times.
✗ Branch 1 not taken.
 791 PingControllers(true);
902 791 sleepCount = 0;
903
6/6
✓ Branch 0 taken 6130 times.
✓ Branch 1 taken 779 times.
✓ Branch 2 taken 6118 times.
✓ Branch 3 taken 12 times.
✓ Branch 4 taken 6118 times.
✓ Branch 5 taken 791 times.
 6909 while (failedCount > 0 && sleepCount < (UDP_PING_TIMEOUT * 2)) {
904
1/2
✓ Branch 0 taken 6118 times.
✗ Branch 1 not taken.
 6118 std::this_thread::sleep_for(std::chrono::milliseconds(1));
905
1/2
✓ Branch 0 taken 6118 times.
✗ Branch 1 not taken.
 6118 CurlManager::INSTANCE.processCurls();
906 6118 ++sleepCount;
907 }
908
1/2
✓ Branch 0 taken 791 times.
✗ Branch 1 not taken.
 791 PingControllers(true);
909 791 sleepCount = 0;
910
6/6
✓ Branch 0 taken 5695 times.
✓ Branch 1 taken 780 times.
✓ Branch 2 taken 5684 times.
✓ Branch 3 taken 11 times.
✓ Branch 4 taken 5684 times.
✓ Branch 5 taken 791 times.
 6475 while (failedCount > 0 && sleepCount < (UDP_PING_TIMEOUT * 2)) {
911
1/2
✓ Branch 0 taken 5684 times.
✗ Branch 1 not taken.
 5684 std::this_thread::sleep_for(std::chrono::milliseconds(1));
912
1/2
✓ Branch 0 taken 5684 times.
✗ Branch 1 not taken.
 5684 CurlManager::INSTANCE.processCurls();
913 5684 ++sleepCount;
914 }
915
4/4
✓ Branch 0 taken 6 times.
✓ Branch 1 taken 785 times.
✓ Branch 2 taken 4 times.
✓ Branch 3 taken 2 times.
 791 if (hasArtNet && interfaceUp) {
916 // we have artnet packets so we'll need to get the special artnet socket created,
917 // but we want to make sure it's on the given interface. Without an
918 // address yet this waits for NEW_ADDR; binding it now would pin the
919 // shared ArtNet socket to loopback for good, and if ArtNet sends
920 // first it creates its own socket on INADDR_ANY.
921
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4 CreateArtNetSocket(localAddress.sin_addr.s_addr, true);
922 }
923 {
924 // From here an attached output's controller is pinged as it attaches;
925 // left in the list above, the ones attached below were never pinged.
926
1/2
✓ Branch 0 taken 791 times.
✗ Branch 1 not taken.
 791 std::lock_guard<std::mutex> l(pingHostsLock);
927 791 pingsStarted = true;
928 791 }
929 {
930 // outputs registered by other plugins before this instance existed,
931 // or kept across a co-universes reload that replaced the previous one
932
1/2
✓ Branch 0 taken 791 times.
✗ Branch 1 not taken.
 791 std::lock_guard<std::mutex> lk(externalLock);
933
2/2
✓ Branch 0 taken 7 times.
✓ Branch 1 taken 791 times.
 798 for (auto& [owner, create] : externalFactories) {
934
2/4
✓ Branch 0 taken 7 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 7 times.
✗ Branch 3 not taken.
 7 attachExternalOutput(owner, createExternalOutput(create));
935 }
936 791 }
937
2/4
✓ Branch 0 taken 791 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 791 times.
✗ Branch 3 not taken.
 1582 return ChannelOutput::Init(config);
938 }
939 791 int UDPOutput::Close() {
940 791 NetworkMonitor::INSTANCE.removeCallback(networkCallbackId);
941 791 messages.clearMessages();
942 791 messages.clearSockets();
943 791 StopPinging();
944
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 789 times.
 791 if (dropWarningActive) {
945 // the replacement instance starts with dropWarningActive false and
946 // would never remove this one
947
2/4
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
 2 WarningHolder::RemoveWarning(LOCAL_DROP_WARNING_ID, LOCAL_DROP_WARNING);
948 2 dropWarningActive = false;
949 }
950
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 791 times.
 791 if (!interfaceWarning.empty()) {
951 // the replacement raises its own if the interface still has no address
952 ✗ WarningHolder::RemoveWarning(22, interfaceWarning);
953 ✗ interfaceWarning.clear();
954 }
955 791 return ChannelOutput::Close();
956 }
957 109094 void UDPOutput::PrepData(unsigned char* channelData) {
958
1/2
✓ Branch 0 taken 109094 times.
✗ Branch 1 not taken.
 109094 if (enabled) {
959
1/2
✓ Branch 0 taken 109094 times.
✗ Branch 1 not taken.
 109094 std::unique_lock<std::mutex> lk(socketMutex);
960
1/2
✓ Branch 0 taken 109094 times.
✗ Branch 1 not taken.
 109094 messages.clearMessages();
961 // A destination that couldn't keep up still has a work item from an
962 // earlier frame, queued or being sent, whose iovecs point at its
963 // outputs' packet buffers (headers, sequence numbers, the dedup copy).
964 // Preparing those outputs again rewrote the buffers while a worker
965 // read them, so packets went out half one frame, half the next. Leave
966 // them as they are this frame: the destination is behind anyway, and
967 // its pending item still goes out.
968 109094 std::set<unsigned int> busy;
969
2/2
✓ Branch 0 taken 108378 times.
✓ Branch 1 taken 716 times.
 109094 if (useThreadedOutput) {
970
1/2
✓ Branch 0 taken 108378 times.
✗ Branch 1 not taken.
 108378 std::lock_guard<std::mutex> wl(workMutex);
971
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 108378 times.
 108378 for (auto& w : workQueue) {
972 ✗ busy.insert(w.id);
973 }
974
1/2
✓ Branch 0 taken 108378 times.
✗ Branch 1 not taken.
 108378 busy.insert(sendingKeys.begin(), sendingKeys.end());
975 108378 }
976 109094 skippedOutputs.clear();
977
2/2
✓ Branch 0 taken 126046 times.
✓ Branch 1 taken 109094 times.
 235140 for (auto a : outputs) {
978
6/6
✓ Branch 0 taken 124807 times.
✓ Branch 1 taken 1239 times.
✓ Branch 2 taken 123716 times.
✓ Branch 3 taken 1091 times.
✓ Branch 4 taken 123716 times.
✓ Branch 5 taken 2330 times.
 126046 if (a->valid.load(std::memory_order_relaxed) && a->active) {
979
2/2
✓ Branch 0 taken 3729 times.
✓ Branch 1 taken 119987 times.
 123716 if (!busy.empty()) {
980
1/2
✓ Branch 0 taken 3729 times.
✗ Branch 1 not taken.
 3729 auto k = outputKeys.find(a);
981
2/4
✓ Branch 0 taken 3729 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 3729 times.
✗ Branch 3 not taken.
 7458 if (k != outputKeys.end() &&
982
2/4
✓ Branch 0 taken 3729 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 3729 times.
✗ Branch 3 not taken.
 3729 std::any_of(k->second.begin(), k->second.end(),
983 3729 [&busy](unsigned int key) { return busy.count(key) != 0; })) {
984
1/2
✓ Branch 0 taken 3729 times.
✗ Branch 1 not taken.
 3729 skippedOutputs.insert(a);
985 3729 continue;
986 }
987 }
988
4/4
✓ Branch 0 taken 119610 times.
✓ Branch 1 taken 377 times.
✓ Branch 2 taken 3 times.
✓ Branch 3 taken 119607 times.
 119987 if (outputRunning && !a->started) {
989 // StartingOutput() only starts the outputs that were
990 // reachable then; one whose controller answers later (a
991 // Twinkly powered on after the show started) is started
992 // here, or it would get frames without ever having
993 // authenticated
994 3 a->started = true;
995
1/2
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
 3 a->StartingOutput();
996 }
997
1/2
✓ Branch 0 taken 119987 times.
✗ Branch 1 not taken.
 119987 std::vector<unsigned int>& keys = outputKeys[a];
998 119987 keys.clear();
999 119987 messages.recordKeys = &keys;
1000
1/2
✓ Branch 0 taken 119987 times.
✗ Branch 1 not taken.
 119987 a->PrepareData(channelData, messages);
1001 119987 messages.recordKeys = nullptr;
1002 }
1003 }
1004 // add any sync packets or whatever that are needed
1005
2/2
✓ Branch 0 taken 126046 times.
✓ Branch 1 taken 109094 times.
 235140 for (auto a : outputs) {
1006
9/10
✓ Branch 0 taken 124807 times.
✓ Branch 1 taken 1239 times.
✓ Branch 2 taken 123716 times.
✓ Branch 3 taken 1091 times.
✓ Branch 4 taken 123716 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 119987 times.
✓ Branch 7 taken 3729 times.
✓ Branch 8 taken 119987 times.
✓ Branch 9 taken 6059 times.
 126046 if (a->valid.load(std::memory_order_relaxed) && a->active && !skippedOutputs.count(a)) {
1007
1/2
✓ Branch 0 taken 119987 times.
✗ Branch 1 not taken.
 119987 a->PostPrepareData(channelData, messages);
1008 }
1009 }
1010 109094 }
1011 109094 }
1012 1691 void UDPOutput::GetRequiredChannelRanges(const std::function<void(int, int)>& addRange) {
1013
1/2
✓ Branch 0 taken 1691 times.
✗ Branch 1 not taken.
 1691 if (enabled) {
1014
2/2
✓ Branch 0 taken 1965 times.
✓ Branch 1 taken 1691 times.
 3656 for (auto a : outputs) {
1015
2/2
✓ Branch 0 taken 1917 times.
✓ Branch 1 taken 48 times.
 1965 if (a->active) {
1016 int mi, mx;
1017
1/2
✓ Branch 0 taken 1917 times.
✗ Branch 1 not taken.
 1917 a->GetRequiredChannelRange(mi, mx);
1018 // max is one past the last channel here and inclusive in
1019 // addRange(): handed on as it was, an output ending on the
1020 // last channel of the buffer failed the loader's range check
1021 // and took every UDP output down with it
1022
1/2
✓ Branch 0 taken 1917 times.
✗ Branch 1 not taken.
 1917 addRange(mi, mx - 1);
1023 }
1024 }
1025 }
1026 1691 }
1027
1028 ✗ void UDPOutput::addOutput(UDPOutputData* out) {
1029 // the output thread walks the list in PrepData under socketMutex
1030 ✗ std::lock_guard<std::mutex> lk(socketMutex);
1031 ✗ outputs.push_back(out);
1032 ✗ }
1033
1034 1584 void UDPOutput::StopPinging() {
1035
1/2
✓ Branch 0 taken 1584 times.
✗ Branch 1 not taken.
 1584 std::lock_guard<std::mutex> lk(pingHostsLock);
1036
2/2
✓ Branch 0 taken 791 times.
✓ Branch 1 taken 793 times.
 1584 if (pingsStopped) {
1037 791 return;
1038 }
1039 793 pingsStopped = true;
1040
2/2
✓ Branch 0 taken 60 times.
✓ Branch 1 taken 793 times.
 853 for (auto& h : pingHosts) {
1041
1/2
✓ Branch 0 taken 60 times.
✗ Branch 1 not taken.
 60 PingManager::INSTANCE.removePeriodicPing(h.first);
1042 // the replacement output raises it again if the controller is still
1043 // down; one removed from the config would otherwise keep its
1044 // "unreachable" banner until fppd restarted
1045
1/2
✓ Branch 0 taken 60 times.
✗ Branch 1 not taken.
 60 std::unique_lock<std::mutex> l(h.second->lock);
1046
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 56 times.
 60 if (!h.second->valid) {
1047
2/4
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 4 times.
✗ Branch 3 not taken.
 8 WarningHolder::RemoveWarning(27, h.second->warning);
1048 }
1049 60 }
1050
2/2
✓ Branch 0 taken 793 times.
✓ Branch 1 taken 791 times.
 1584 }
1051
1052 // An external output whose config can't be read (a GenericUDP output with a
1053 // list for its port) throws from its constructor. That failed this whole
1054 // output, every E1.31/DDP universe with it, at each rebuild; skip just that one.
1055 52 UDPOutputData* UDPOutput::createExternalOutput(const std::function<UDPOutputData*()>& create) {
1056 try {
1057
2/2
✓ Branch 0 taken 50 times.
✓ Branch 1 taken 2 times.
 52 return create();
1058
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2 times.
 2 } catch (const std::exception& e) {
1059
3/6
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 2 times.
✗ Branch 5 not taken.
 4 std::string warning = std::string("Could not create a network output from its config (") + e.what() + "). Output disabled.";
1060
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 LogErr(VB_CHANNELOUT, "%s\n", warning.c_str());
1061
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 AddOutputConfigWarning(26, warning);
1062 2 return nullptr;
1063 2 }
1064 }
1065
1066 52 void UDPOutput::attachExternalOutput(const void* owner, UDPOutputData* d) {
1067
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 50 times.
 52 if (d == nullptr) {
1068 2 return;
1069 }
1070 {
1071
1/2
✓ Branch 0 taken 50 times.
✗ Branch 1 not taken.
 50 std::lock_guard<std::mutex> lk(socketMutex);
1072
1/2
✓ Branch 0 taken 50 times.
✗ Branch 1 not taken.
 50 outputs.push_back(d);
1073
1/2
✓ Branch 0 taken 50 times.
✗ Branch 1 not taken.
 50 externalOutputs[owner] = d;
1074 50 }
1075 // Under pingHostsLock to the end, as in PingControllers().
1076
1/2
✓ Branch 0 taken 50 times.
✗ Branch 1 not taken.
 50 std::lock_guard<std::mutex> lk(pingHostsLock);
1077
2/4
✓ Branch 0 taken 50 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 50 times.
 50 if (!pingsStarted || pingsStopped) {
1078 // Init() pings what is attached before it starts the pings
1079 ✗ return;
1080 }
1081
3/4
✓ Branch 0 taken 50 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 48 times.
✓ Branch 3 taken 2 times.
 50 if (std::shared_ptr<PingHost> host = addPingTarget(d)) {
1082 48 ++failedCount;
1083
2/4
✓ Branch 0 taken 48 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 48 times.
✗ Branch 3 not taken.
 48 PingManager::INSTANCE.addPeriodicPing(host->address, UDP_PING_TIMEOUT, 15000, [this, host](int ms) {
1084 51 OnPingResult(host, ms);
1085 51 });
1086 50 }
1087
1/2
✓ Branch 0 taken 50 times.
✗ Branch 1 not taken.
 50 }
1088
1089 947 std::shared_ptr<UDPOutput::PingHost> UDPOutput::addPingTarget(UDPOutputData* o) {
1090
9/12
✓ Branch 0 taken 947 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 941 times.
✓ Branch 3 taken 6 times.
✓ Branch 4 taken 941 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 78 times.
✓ Branch 7 taken 863 times.
✗ Branch 8 not taken.
✓ Branch 9 taken 78 times.
✓ Branch 10 taken 869 times.
✓ Branch 11 taken 78 times.
 947 if (!o->IsPingable() || !o->Monitor() || !o->active) {
1091 869 return nullptr;
1092 }
1093
1/2
✓ Branch 0 taken 78 times.
✗ Branch 1 not taken.
 78 std::shared_ptr<PingHost>& host = pingHosts[o->ipAddress];
1094 78 bool created = !host;
1095
2/2
✓ Branch 0 taken 67 times.
✓ Branch 1 taken 11 times.
 78 if (created) {
1096
1/2
✓ Branch 0 taken 67 times.
✗ Branch 1 not taken.
 67 host = std::make_shared<PingHost>();
1097
1/2
✓ Branch 0 taken 67 times.
✗ Branch 1 not taken.
 67 host->address = o->ipAddress;
1098
2/4
✓ Branch 0 taken 67 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 67 times.
✗ Branch 3 not taken.
 67 host->warning = createWarning(o->ipAddress, o->GetOutputTypeString(), o->description);
1099 }
1100
1/2
✓ Branch 0 taken 78 times.
✗ Branch 1 not taken.
 78 std::unique_lock<std::mutex> l(host->lock);
1101
1/2
✓ Branch 0 taken 78 times.
✗ Branch 1 not taken.
 78 host->outputs.push_back(o);
1102
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 78 times.
 78 if (!host->valid) {
1103 // one joining a controller already found down starts out of the frame too
1104 ✗ o->valid = false;
1105 }
1106
2/2
✓ Branch 0 taken 67 times.
✓ Branch 1 taken 11 times.
 78 return created ? host : nullptr;
1107 78 }
1108
1109 48 void UDPOutput::removePingTarget(UDPOutputData* o) {
1110
1/2
✓ Branch 0 taken 48 times.
✗ Branch 1 not taken.
 48 auto it = pingHosts.find(o->ipAddress);
1111
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 46 times.
 48 if (it == pingHosts.end()) {
1112 2 return;
1113 }
1114 46 std::shared_ptr<PingHost> host = it->second;
1115 {
1116
1/2
✓ Branch 0 taken 46 times.
✗ Branch 1 not taken.
 46 std::unique_lock<std::mutex> l(host->lock);
1117 46 auto& outs = host->outputs;
1118
2/4
✓ Branch 0 taken 46 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 46 times.
✗ Branch 3 not taken.
 46 outs.erase(std::remove(outs.begin(), outs.end(), o), outs.end());
1119
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 46 times.
 46 if (!outs.empty()) {
1120 ✗ return;
1121 }
1122
1/2
✓ Branch 0 taken 46 times.
✗ Branch 1 not taken.
 46 }
1123 // Nothing sends there any more. Not under host->lock: this waits out a
1124 // running ping callback, which takes it.
1125
1/2
✓ Branch 0 taken 46 times.
✗ Branch 1 not taken.
 46 pingHosts.erase(it);
1126
2/2
✓ Branch 0 taken 7 times.
✓ Branch 1 taken 39 times.
 46 if (!pingsStopped) {
1127
1/2
✓ Branch 0 taken 7 times.
✗ Branch 1 not taken.
 7 PingManager::INSTANCE.removePeriodicPing(host->address);
1128 }
1129
1/2
✓ Branch 0 taken 46 times.
✗ Branch 1 not taken.
 46 std::unique_lock<std::mutex> l(host->lock);
1130
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 45 times.
 46 if (!host->valid) {
1131
2/4
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1 times.
✗ Branch 3 not taken.
 2 WarningHolder::RemoveWarning(27, host->warning);
1132 }
1133
1/2
✓ Branch 0 taken 46 times.
✗ Branch 1 not taken.
 46 }
1134
1135 56 void UDPOutput::RegisterExternalOutput(const void* owner, std::function<UDPOutputData*()> create) {
1136
1/2
✓ Branch 0 taken 56 times.
✗ Branch 1 not taken.
 56 std::lock_guard<std::mutex> lk(externalLock);
1137
2/4
✓ Branch 0 taken 56 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 56 times.
✗ Branch 3 not taken.
 56 externalFactories[owner] = create;
1138
2/2
✓ Branch 0 taken 45 times.
✓ Branch 1 taken 11 times.
 56 if (INSTANCE) {
1139
2/4
✓ Branch 0 taken 45 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 45 times.
✗ Branch 3 not taken.
 45 INSTANCE->attachExternalOutput(owner, createExternalOutput(create));
1140 }
1141 56 }
1142
1143 112 void UDPOutput::UnregisterExternalOutput(const void* owner) {
1144
1/2
✓ Branch 0 taken 112 times.
✗ Branch 1 not taken.
 112 std::lock_guard<std::mutex> lk(externalLock);
1145
1/2
✓ Branch 0 taken 112 times.
✗ Branch 1 not taken.
 112 externalFactories.erase(owner);
1146
2/2
✓ Branch 0 taken 59 times.
✓ Branch 1 taken 53 times.
 112 if (INSTANCE) {
1147
1/2
✓ Branch 0 taken 59 times.
✗ Branch 1 not taken.
 59 std::lock_guard<std::mutex> slk(INSTANCE->socketMutex);
1148
1/2
✓ Branch 0 taken 59 times.
✗ Branch 1 not taken.
 59 auto it = INSTANCE->externalOutputs.find(owner);
1149
2/2
✓ Branch 0 taken 48 times.
✓ Branch 1 taken 11 times.
 59 if (it != INSTANCE->externalOutputs.end()) {
1150 // Retired, not deleted: a queued or running send may still point
1151 // into its packet buffers. It stays in outputs, inactive, so
1152 // PrepData skips it and the destructor frees it once the workers
1153 // have exited.
1154 48 it->second->active = false;
1155 {
1156
1/2
✓ Branch 0 taken 48 times.
✗ Branch 1 not taken.
 48 std::lock_guard<std::mutex> plk(INSTANCE->pingHostsLock);
1157
1/2
✓ Branch 0 taken 48 times.
✗ Branch 1 not taken.
 48 INSTANCE->removePingTarget(it->second);
1158 48 }
1159
1/2
✓ Branch 0 taken 48 times.
✗ Branch 1 not taken.
 48 INSTANCE->externalOutputs.erase(it);
1160 }
1161 59 }
1162 112 }
1163
1164 69307 static void flushBuffers(int socket, int msgs, int total) {
1165 #ifndef PLATFORM_OSX
1166 69307 int bytes_in_buffer = 0;
1167
1/2
✓ Branch 0 taken 69307 times.
✗ Branch 1 not taken.
 69307 if (ioctl(socket, SIOCOUTQ, &bytes_in_buffer) == 0) {
1168 // Check bytes_in_buffer, if its too high, wait for it to drain
1169 69307 int cnt = 0;
1170 69307 int start = bytes_in_buffer;
1171
4/4
✓ Branch 0 taken 943194 times.
✓ Branch 1 taken 50813 times.
✓ Branch 2 taken 924700 times.
✓ Branch 3 taken 18494 times.
 994007 while (bytes_in_buffer > 1024 && cnt < 50) {
1172 924700 cnt++;
1173
1/2
✓ Branch 0 taken 924700 times.
✗ Branch 1 not taken.
 924700 std::this_thread::sleep_for(std::chrono::microseconds(1));
1174 924700 ioctl(socket, SIOCOUTQ, &bytes_in_buffer);
1175 }
1176 // if (cnt > 0) {
1177 // printf("Flush: Socket %d had to wait %d (%d->%d) (%d/%d)...\n", socket, cnt, start, bytes_in_buffer, msgs, total);
1178 // }
1179 }
1180 #endif
1181 69307 }
1182
1183 constexpr int MSGS_PER_SENDMMSG = 8;
1184 // kernel-paced sockets take the whole frame in one call (kernel caps a single
1185 // sendmmsg at UIO_MAXIOV=1024 anyway)
1186 constexpr int PACED_MSGS_PER_SENDMMSG = 1024;
1187 109421 int UDPOutput::SendMessages(unsigned int socketKey, SendSocketInfo* socketInfo, std::vector<struct mmsghdr>& sendmsgs) {
1188 109421 errno = 0;
1189 109421 struct mmsghdr* msgs = &sendmsgs[0];
1190 109421 int msgCount = sendmsgs.size();
1191
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 109421 times.
 109421 if (msgCount == 0) {
1192 ✗ return 0;
1193 }
1194
1195 109421 int newSockKey = socketKey;
1196 // Serialize against a straggler worker from a previous frame still sending
1197 // to this destination; SendMessages mutates curSocket/sockets. Blocking
1198 // (rather than try_lock) is intentional and safe for the frame thread:
1199 // workers exclusively own the early-message keys, and the frame thread
1200 // only calls SendMessages for late/sync keys (threaded mode) or when no
1201 // workers exist at all (non-threaded mode), so it can never block on a
1202 // straggler here.
1203
1/2
✓ Branch 0 taken 109421 times.
✗ Branch 1 not taken.
 109421 std::lock_guard<std::mutex> sendGuard(socketInfo->sendLock);
1204
2/2
✓ Branch 0 taken 90 times.
✓ Branch 1 taken 109331 times.
 109421 if (socketInfo->sockets.empty()) {
1205 90 return 0;
1206 }
1207
3/6
✓ Branch 0 taken 109331 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 109331 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 109331 times.
 109331 if (socketInfo->curSocket < 0 || socketInfo->curSocket >= socketInfo->sockets.size()) {
1208 ✗ socketInfo->curSocket = 0;
1209 }
1210 109331 int sendSocket = socketInfo->sockets[socketInfo->curSocket];
1211 109331 errno = 0;
1212 // uint64_t st = GetTimeMicros();
1213
1214 // when the kernel is pacing this socket (fq + SO_MAX_PACING_RATE) the
1215 // socket buffer is sized for the whole frame; hand everything over at once
1216 // and skip the drain waits, the qdisc clocks packets out at the target rate
1217 109331 const bool paced = socketInfo->paced;
1218
2/2
✓ Branch 0 taken 9049 times.
✓ Branch 1 taken 100282 times.
 109331 const int maxBatch = paced ? PACED_MSGS_PER_SENDMMSG : MSGS_PER_SENDMMSG;
1219
1220 109331 int outputCount = 0;
1221
3/4
✓ Branch 0 taken 702 times.
✓ Branch 1 taken 108629 times.
✓ Branch 2 taken 702 times.
✗ Branch 3 not taken.
 109331 if (blockingOutput && !paced) {
1222 // Blocking mode only applies to unpaced sockets. For paced sockets
1223 // the kernel throttling that blocking mode approximates is done
1224 // properly by the fq qdisc, and the per-packet 1ms SO_SNDTIMEO would
1225 // abort a frame that legitimately sits in the paced buffer; they
1226 // always use the batched path below.
1227 702 int errorCount = 0;
1228
2/2
✓ Branch 0 taken 702 times.
✓ Branch 1 taken 702 times.
 1404 for (int x = 0; x < msgCount; x++) {
1229
1/2
✓ Branch 0 taken 702 times.
✗ Branch 1 not taken.
 702 flushBuffers(sendSocket, x, msgCount);
1230
1/2
✓ Branch 0 taken 702 times.
✗ Branch 1 not taken.
 702 ssize_t s = sendmsg(sendSocket, &msgs[x].msg_hdr, 0);
1231
1/2
✓ Branch 0 taken 702 times.
✗ Branch 1 not taken.
 702 if (s != -1) {
1232 702 errorCount = 0;
1233 702 ++outputCount;
1234 ✗ } else if (errorCount) {
1235 ✗ return outputCount;
1236 } else {
1237 // didn't send, we'll yield once and re-send
1238 ✗ --x;
1239 ✗ ++errorCount;
1240 ✗ std::this_thread::yield();
1241 }
1242 }
1243 702 } else {
1244
1/2
✓ Branch 0 taken 108629 times.
✗ Branch 1 not taken.
 108629 int oc = sendmmsg(sendSocket, msgs, std::min(msgCount, maxBatch), MSG_DONTWAIT);
1245
2/2
✓ Branch 0 taken 107935 times.
✓ Branch 1 taken 694 times.
 108629 if (oc > 0) {
1246 107935 outputCount += oc;
1247 }
1248
2/2
✓ Branch 0 taken 8806 times.
✓ Branch 1 taken 99823 times.
 108629 if (outputCount != msgCount) {
1249 #ifndef PLATFORM_OSX
1250
2/2
✓ Branch 0 taken 7822 times.
✓ Branch 1 taken 984 times.
 8806 if (!paced) {
1251
1/2
✓ Branch 0 taken 7822 times.
✗ Branch 1 not taken.
 7822 flushBuffers(sendSocket, outputCount, msgCount);
1252 }
1253 #else
1254 // On OSX we have no good way to check the socket buffer, so just
1255 // sleep a bit to allow the stack to flush
1256 std::this_thread::sleep_for(std::chrono::microseconds(100));
1257 #endif
1258 8806 int outMsgCnt = msgCount - outputCount;
1259
1/2
✓ Branch 0 taken 8806 times.
✗ Branch 1 not taken.
 8806 oc = sendmmsg(sendSocket, &msgs[outputCount], std::min(outMsgCnt, maxBatch), MSG_DONTWAIT);
1260
2/2
✓ Branch 0 taken 61010 times.
✓ Branch 1 taken 8806 times.
 69816 while (oc > 0) {
1261 61010 outputCount += oc;
1262 #ifndef PLATFORM_OSX
1263
2/2
✓ Branch 0 taken 60783 times.
✓ Branch 1 taken 227 times.
 61010 if (!paced) {
1264
1/2
✓ Branch 0 taken 60783 times.
✗ Branch 1 not taken.
 60783 flushBuffers(sendSocket, outputCount, msgCount);
1265 }
1266 #else
1267 std::this_thread::sleep_for(std::chrono::microseconds(100));
1268 #endif
1269 61010 outMsgCnt = msgCount - outputCount;
1270
1/2
✓ Branch 0 taken 61010 times.
✗ Branch 1 not taken.
 61010 oc = sendmmsg(sendSocket, &msgs[outputCount], std::min(outMsgCnt, maxBatch), MSG_DONTWAIT);
1271 }
1272 }
1273 }
1274 // uint64_t ed = GetTimeMicros();
1275 // uint64_t tm = ed - st;
1276 // printf("MSG: %d/%d %d \n", outputCount, msgCount, (int)tm);
1277
1278 109331 int errCount = 0;
1279 109331 auto retryStart = std::chrono::steady_clock::now();
1280
2/2
✓ Branch 0 taken 48880 times.
✓ Branch 1 taken 107734 times.
 156614 while (outputCount != msgCount) {
1281 // Absolute wall-clock bound in addition to the consecutive-failure
1282 // count: a destination that needs more data than the pacing rate
1283 // allows keeps making slow progress (resetting errCount), and without
1284 // this cap the worker would hold sendLock across whole frames as a
1285 // permanent straggler.
1286
4/6
✓ Branch 0 taken 48880 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 48880 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 839 times.
✓ Branch 5 taken 48041 times.
 48880 if (std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - retryStart).count() > 100) {
1287
3/4
✓ Branch 0 taken 839 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 18 times.
✓ Branch 3 taken 821 times.
 839 if (ShouldLogSendError(socketInfo, SEND_ERR_BUDGET))
1288
2/4
✓ Branch 0 taken 18 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 18 times.
✗ Branch 3 not taken.
 18 LogErr(VB_CHANNELOUT, "sendmmsg() send to %s exceeded time budget, dropping remainder of frame (output count: %d/%d)\n",
1289 HexToIP(socketKey).c_str(), outputCount, msgCount);
1290 839 return outputCount;
1291 }
1292
3/4
✓ Branch 0 taken 5235 times.
✓ Branch 1 taken 42806 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 5235 times.
 48041 if (errno == EAGAIN || errno == EWOULDBLOCK) {
1293
2/2
✓ Branch 0 taken 40313 times.
✓ Branch 1 taken 2493 times.
 42806 if (paced) {
1294 // the qdisc is intentionally holding packets; wait for the
1295 // buffer to drain at the pacing rate rather than spreading the
1296 // overflow across additional sockets, which would multiply the
1297 // per-controller rate cap and defeat the pacing
1298
1/2
✓ Branch 0 taken 40313 times.
✗ Branch 1 not taken.
 40313 std::this_thread::sleep_for(std::chrono::milliseconds(2));
1299
1/2
✓ Branch 0 taken 2493 times.
✗ Branch 1 not taken.
 2493 } else if (socketInfo->preventClose) {
1300 // a shared socket pinned with ForceSocket (the ArtNet socket,
1301 // bound to port 6454 with SO_BROADCAST): an extra socket made
1302 // here would have neither, so broadcast ArtNet and ArtNet sync
1303 // would fail on it for good, and it would never be closed.
1304 // Wait for room on the one socket instead.
1305
1/2
✓ Branch 0 taken 2493 times.
✗ Branch 1 not taken.
 2493 std::this_thread::sleep_for(std::chrono::milliseconds(1));
1306 ✗ } else if (socketKey != BROADCAST_MESSAGES_KEY) {
1307 ✗ ++socketInfo->curSocket;
1308 ✗ if (socketInfo->curSocket == socketInfo->sockets.size()) {
1309 // the extra socket must keep the multicast setup or the
1310 // packets will egress the default route instead of the
1311 // show network
1312 ✗ int s = socketInfo->sockets.size() < 5 ? createSocket(0, false, isMulticastKey(socketKey)) : -1;
1313 ✗ if (s != -1) {
1314 ✗ socketInfo->sockets.push_back(s);
1315 } else {
1316 ✗ socketInfo->curSocket = 0;
1317 }
1318 }
1319 ✗ sendSocket = socketInfo->sockets[socketInfo->curSocket];
1320 } else {
1321 ✗ return outputCount;
1322 }
1323 }
1324 48041 ++errCount;
1325
2/2
✓ Branch 0 taken 758 times.
✓ Branch 1 taken 47283 times.
 48041 if (errCount >= 10) {
1326
3/4
✓ Branch 0 taken 758 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 8 times.
✓ Branch 3 taken 750 times.
 758 if (ShouldLogSendError(socketInfo, SEND_ERR_EAGAIN))
1327
3/6
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 8 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 8 times.
✗ Branch 5 not taken.
 8 LogErr(VB_CHANNELOUT, "sendmmsg() failed for UDP output (IP: %s Socket: %d output count: %d/%d) with error: %d %s\n",
1328 HexToIP(socketKey).c_str(), sendSocket,
1329 outputCount, msgCount,
1330 errno,
1331 FPPstrerror(errno));
1332 758 return outputCount;
1333 }
1334 47283 errno = 0;
1335
1/2
✓ Branch 0 taken 47283 times.
✗ Branch 1 not taken.
 47283 int oc = sendmmsg(sendSocket, &msgs[outputCount], msgCount - outputCount, MSG_DONTWAIT);
1336
2/2
✓ Branch 0 taken 42798 times.
✓ Branch 1 taken 47283 times.
 90081 while (oc > 0) {
1337
2/2
✓ Branch 0 taken 42742 times.
✓ Branch 1 taken 56 times.
 42798 if (paced) {
1338 // making progress; for a paced socket EAGAIN is the expected
1339 // steady state when a frame exceeds the buffer, so only give
1340 // up after consecutive no-progress attempts. Unpaced sockets
1341 // keep the historical ~10-attempt bound so a congested
1342 // destination cannot pin a worker for a whole frame.
1343 42742 errCount = 0;
1344 }
1345 42798 outputCount += oc;
1346
1/2
✓ Branch 0 taken 42798 times.
✗ Branch 1 not taken.
 42798 std::this_thread::sleep_for(std::chrono::microseconds(100));
1347
1/2
✓ Branch 0 taken 42798 times.
✗ Branch 1 not taken.
 42798 oc = sendmmsg(sendSocket, &msgs[outputCount], msgCount - outputCount, MSG_DONTWAIT);
1348 }
1349 }
1350 107734 return outputCount;
1351 109421 }
1352
1353 829 static void DoWorkThread(UDPOutput* output) {
1354
2/4
✓ Branch 0 taken 829 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 829 times.
✗ Branch 3 not taken.
 829 SetThreadName("FPP-UDPWork");
1355 // Keep the send threads from being delayed by other load (video decode,
1356 // php, etc.). A delayed frame bunches with the next one and doubles the
1357 // burst hitting the switch. Priority sits below the output thread (10)
1358 // so the frame loop always preempts stragglers; worker count can grow to
1359 // the destination fan-out and these all park on workSignal when idle.
1360 829 SetThreadRealtimePriority(5);
1361 829 output->BackgroundOutputWork();
1362 829 }
1363
1364 829 void UDPOutput::BackgroundOutputWork() {
1365 std::chrono::high_resolution_clock clock;
1366
2/2
✓ Branch 0 taken 109043 times.
✓ Branch 1 taken 829 times.
 109872 while (runWorkThreads) {
1367
1/2
✓ Branch 0 taken 109043 times.
✗ Branch 1 not taken.
 109043 std::unique_lock<std::mutex> lock(workMutex);
1368
5/6
✓ Branch 0 taken 109043 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 216684 times.
✓ Branch 3 taken 829 times.
✓ Branch 4 taken 108214 times.
✓ Branch 5 taken 108470 times.
 326556 workSignal.wait(lock, [this]() { return !runWorkThreads || !workQueue.empty(); });
1369
2/2
✓ Branch 0 taken 108214 times.
✓ Branch 1 taken 829 times.
 109043 if (!workQueue.empty()) {
1370 108214 WorkItem i = std::move(workQueue.front());
1371 108214 workQueue.pop_front();
1372
1/2
✓ Branch 0 taken 108214 times.
✗ Branch 1 not taken.
 108214 sendingKeys.insert(i.id);
1373
1/2
✓ Branch 0 taken 108214 times.
✗ Branch 1 not taken.
 108214 lock.unlock();
1374
1375 108214 auto t1 = clock.now();
1376
1/2
✓ Branch 0 taken 108214 times.
✗ Branch 1 not taken.
 108214 int outputCount = SendMessages(i.id, i.socketInfo.get(), i.msgs);
1377 108214 auto t2 = clock.now();
1378
1379
2/4
✓ Branch 0 taken 108214 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 108214 times.
✗ Branch 3 not taken.
 108214 long diff = std::chrono::duration_cast<std::chrono::milliseconds>(t2 - t1).count();
1380
2/2
✓ Branch 0 taken 1595 times.
✓ Branch 1 taken 106619 times.
 108214 if (outputCount != i.msgs.size())
1381 1595 i.socketInfo->framesBehind.fetch_add(1, std::memory_order_relaxed);
1382
6/6
✓ Branch 0 taken 106619 times.
✓ Branch 1 taken 1595 times.
✓ Branch 2 taken 140 times.
✓ Branch 3 taken 106479 times.
✓ Branch 4 taken 1735 times.
✓ Branch 5 taken 106479 times.
 108214 if ((outputCount != i.msgs.size()) || (diff > 100)) {
1383 1735 i.socketInfo->errCount++;
1384
1385 // failed to send all messages or it took more than 100ms to send them
1386
3/4
✓ Branch 0 taken 1735 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 34 times.
✓ Branch 3 taken 1701 times.
 1735 if (ShouldLogSendError(i.socketInfo.get(), SEND_ERR_FRAME))
1387
6/8
✓ Branch 0 taken 32 times.
✓ Branch 1 taken 2 times.
✓ Branch 2 taken 34 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 2 times.
✓ Branch 5 taken 32 times.
✓ Branch 6 taken 34 times.
✗ Branch 7 not taken.
 68 LogErr(VB_CHANNELOUT, "%s() failed for UDP output (IP: %s output count: %d/%d time: %u ms errCount: %d) with error: %d %s\n",
1388 blockingOutput ? "sendmsg" : "sendmmsg", HexToIP(i.id).c_str(),
1389 outputCount, i.msgs.size(), diff, i.socketInfo->errCount.load(),
1390 errno,
1391 FPPstrerror(errno));
1392 } else {
1393 106479 i.socketInfo->clearErrors();
1394 }
1395
1396 // only count work for the current frame; if SendData timed out and
1397 // moved on, this completion must not skew the next frame's count.
1398 // The check and increment must be one unit under workMutex or a
1399 // generation bump between them counts old work toward the new frame.
1400 {
1401
1/2
✓ Branch 0 taken 108214 times.
✗ Branch 1 not taken.
 108214 std::unique_lock<std::mutex> countLock(workMutex);
1402
2/4
✓ Branch 0 taken 108214 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 108214 times.
✗ Branch 3 not taken.
 108214 sendingKeys.erase(sendingKeys.find(i.id));
1403
2/2
✓ Branch 0 taken 107062 times.
✓ Branch 1 taken 1152 times.
 108214 if (i.generation == workGeneration) {
1404 107062 doneWorkCount++;
1405 }
1406 108214 }
1407 108214 }
1408 109043 }
1409
1410 829 --numWorkThreads;
1411 829 }
1412
1413 104867 int UDPOutput::SendData(unsigned char* channelData) {
1414
1/2
✓ Branch 0 taken 104867 times.
✗ Branch 1 not taken.
 104867 std::unique_lock<std::mutex> lk(socketMutex);
1415
5/6
✓ Branch 0 taken 104867 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 206 times.
✓ Branch 3 taken 104661 times.
✓ Branch 4 taken 206 times.
✓ Branch 5 taken 104661 times.
 104867 if (!enabled || messages.messages.empty()) {
1416 206 return 0;
1417 }
1418 104661 std::set<unsigned int> skippedKeys;
1419
2/2
✓ Branch 0 taken 121323 times.
✓ Branch 1 taken 104661 times.
 225984 for (auto a : outputs) {
1420 // A skipped output prepared nothing, so it has nothing to commit.
1421
3/4
✓ Branch 0 taken 121323 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 117595 times.
✓ Branch 3 taken 3728 times.
 121323 if (!skippedOutputs.count(a)) {
1422
1/2
✓ Branch 0 taken 117595 times.
✗ Branch 1 not taken.
 117595 a->FrameSent();
1423 } else {
1424
1/2
✓ Branch 0 taken 3728 times.
✗ Branch 1 not taken.
 3728 auto k = outputKeys.find(a);
1425
1/2
✓ Branch 0 taken 3728 times.
✗ Branch 1 not taken.
 3728 if (k != outputKeys.end()) {
1426
1/2
✓ Branch 0 taken 3728 times.
✗ Branch 1 not taken.
 3728 skippedKeys.insert(k->second.begin(), k->second.end());
1427 }
1428 }
1429 }
1430 // A destination PrepData() left alone because it was still busy misses
1431 // this frame. It has no messages, so the loop below counts nothing for
1432 // it: count the frame as late here, or a controller that keeps up with
1433 // only every other frame (each send slower than a frame, but inside the
1434 // send budget, so never short) never raises the pacing-budget banner.
1435
2/2
✓ Branch 0 taken 3728 times.
✓ Branch 1 taken 104661 times.
 108389 for (unsigned int key : skippedKeys) {
1436
1/2
✓ Branch 0 taken 3728 times.
✗ Branch 1 not taken.
 3728 auto m = messages.messages.find(key);
1437
1/2
✓ Branch 0 taken 3728 times.
✗ Branch 1 not taken.
 3728 auto si = messages.sendSockets.find(key);
1438
5/10
✓ Branch 0 taken 3728 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 3728 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 3728 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 3728 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 3728 times.
✗ Branch 9 not taken.
 3728 if ((m == messages.messages.end() || m->second.empty()) && si != messages.sendSockets.end() && si->second) {
1439 3728 si->second->bytesQueued.fetch_add(si->second->lastFrameBytes, std::memory_order_relaxed);
1440 3728 si->second->framesQueued.fetch_add(1, std::memory_order_relaxed);
1441 3728 si->second->framesBehind.fetch_add(1, std::memory_order_relaxed);
1442 }
1443 }
1444
2/2
✓ Branch 0 taken 25 times.
✓ Branch 1 taken 104636 times.
 104661 if (++statCheckCounter >= 1500) {
1445 // roughly once a minute at typical frame rates; run off-thread so the
1446 // tc fork in the qdisc check can't hiccup the frame
1447 25 statCheckCounter = 0;
1448
1/2
✓ Branch 0 taken 25 times.
✗ Branch 1 not taken.
 25 if (!statCheckRunning.exchange(true)) {
1449 // the socket map is only safe to walk under socketMutex, held here
1450 25 std::vector<PacingBudgetDest> dests;
1451
2/2
✓ Branch 0 taken 26 times.
✓ Branch 1 taken 25 times.
 51 for (auto& si : messages.sendSockets) {
1452
4/8
✓ Branch 0 taken 26 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 26 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 26 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 26 times.
✗ Branch 7 not taken.
 26 if (si.second && si.first > ARTNET_MESSAGES_KEY && si.first < LATE_MESSAGES_START)
1453 26 dests.push_back({ si.first, HexToIP(si.first), si.second });
1454 }
1455 // a copy: the list itself can grow under socketMutex while the
1456 // stats thread runs; the entries live as long as this instance
1457
1/4
✓ Branch 0 taken 25 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 75 std::thread([this, dests = std::move(dests), outs = outputs]() {
1458
2/4
✓ Branch 0 taken 25 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 25 times.
✗ Branch 3 not taken.
 25 SetThreadName("FPP-UDPStats");
1459 try {
1460
1/2
✓ Branch 0 taken 25 times.
✗ Branch 1 not taken.
 25 CheckLocalDrops();
1461
1/2
✓ Branch 0 taken 25 times.
✗ Branch 1 not taken.
 25 CheckPacingBudget(dests, outs);
1462 ✗ } catch (...) {
1463 // an escaped exception on a detached thread would call
1464 // std::terminate and take down the show over a diagnostic
1465 ✗ }
1466 25 statCheckRunning = false;
1467
2/4
✓ Branch 0 taken 25 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 25 times.
✗ Branch 3 not taken.
 50 }).detach();
1468 25 }
1469 }
1470 std::chrono::high_resolution_clock clock;
1471
2/2
✓ Branch 0 taken 103959 times.
✓ Branch 1 taken 702 times.
 104661 if (useThreadedOutput) {
1472 unsigned int generation;
1473 {
1474 // start a new generation so late completions of the previous
1475 // frame's work don't count toward this frame.
1476
1/2
✓ Branch 0 taken 103959 times.
✗ Branch 1 not taken.
 103959 std::unique_lock<std::mutex> lock(workMutex);
1477 103959 generation = ++workGeneration;
1478 103959 doneWorkCount = 0;
1479 103959 }
1480 103959 int total = 0;
1481 103959 auto t1 = clock.now();
1482
2/2
✓ Branch 0 taken 113565 times.
✓ Branch 1 taken 103959 times.
 217524 for (auto& msgs : messages.messages) {
1483
6/6
✓ Branch 0 taken 108719 times.
✓ Branch 1 taken 4846 times.
✓ Branch 2 taken 108214 times.
✓ Branch 3 taken 505 times.
✓ Branch 4 taken 108214 times.
✓ Branch 5 taken 5351 times.
 113565 if (!msgs.second.empty() && msgs.first < LATE_MESSAGES_START) {
1484
1/2
✓ Branch 0 taken 108214 times.
✗ Branch 1 not taken.
 108214 std::shared_ptr<SendSocketInfo> socketInfo = findOrCreateSocket(msgs.first);
1485
1/2
✓ Branch 0 taken 108214 times.
✗ Branch 1 not taken.
 108214 std::unique_lock<std::mutex> lock(workMutex);
1486 // A destination that still has an unstarted item from an
1487 // earlier frame is behind: SendMessages() serializes on its
1488 // sendLock, so another item would only queue up behind it, and
1489 // every queued item gets a worker thread below. A destination
1490 // that can't keep up (more data than its pacing rate allows)
1491 // grew one item and one blocked thread per frame until thread
1492 // creation failed and std::terminate took fppd down. Fold this
1493 // frame into the pending item instead. Nothing is lost: the
1494 // iovecs point at the live packet buffers, so the pending item
1495 // sends this frame's data when it runs, and the union keeps any
1496 // message the dedup logic left out of this frame but not that one.
1497
1/2
✓ Branch 0 taken 108214 times.
✗ Branch 1 not taken.
 108214 auto pending = std::find_if(workQueue.begin(), workQueue.end(),
1498 13060 [&](const WorkItem& w) { return w.id == msgs.first; });
1499 108214 socketInfo->lastFrameBytes = MessageBytes(msgs.second);
1500 108214 socketInfo->bytesQueued.fetch_add(socketInfo->lastFrameBytes, std::memory_order_relaxed);
1501 108214 socketInfo->framesQueued.fetch_add(1, std::memory_order_relaxed);
1502
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 108214 times.
 108214 if (pending != workQueue.end()) {
1503 ✗ socketInfo->framesBehind.fetch_add(1, std::memory_order_relaxed);
1504 ✗ std::set<std::pair<const void*, const void*>> have;
1505 ✗ for (auto& m : pending->msgs)
1506 ✗ have.emplace(m.msg_hdr.msg_iov, m.msg_hdr.msg_name);
1507 ✗ for (auto& m : msgs.second) {
1508 ✗ if (!have.count({ m.msg_hdr.msg_iov, m.msg_hdr.msg_name }))
1509 ✗ pending->msgs.push_back(m);
1510 }
1511 ✗ pending->generation = generation;
1512 ✗ } else {
1513
1/2
✓ Branch 0 taken 108214 times.
✗ Branch 1 not taken.
 108214 workQueue.emplace_back(msgs.first, socketInfo, msgs.second, generation);
1514 }
1515
1/2
✓ Branch 0 taken 108214 times.
✗ Branch 1 not taken.
 108214 lock.unlock();
1516 108214 workSignal.notify_one();
1517 108214 ++total;
1518 108214 }
1519 }
1520
1/2
✓ Branch 0 taken 103959 times.
✗ Branch 1 not taken.
 103959 std::unique_lock<std::mutex> lock(workMutex);
1521
2/2
✓ Branch 0 taken 829 times.
✓ Branch 1 taken 103959 times.
 104788 while (numWorkThreads < workQueue.size()) {
1522
2/4
✓ Branch 0 taken 829 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 829 times.
✗ Branch 3 not taken.
 829 std::thread(DoWorkThread, this).detach();
1523 829 numWorkThreads++;
1524 }
1525
2/2
✓ Branch 0 taken 98043 times.
✓ Branch 1 taken 5916 times.
 103959 if (workQueue.size()) {
1526 98043 workSignal.notify_all();
1527 }
1528
1/2
✓ Branch 0 taken 103959 times.
✗ Branch 1 not taken.
 103959 lock.unlock();
1529 103959 auto t2 = clock.now();
1530
8/10
✓ Branch 0 taken 370113 times.
✓ Branch 1 taken 102807 times.
✓ Branch 2 taken 370113 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 370113 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 368961 times.
✓ Branch 7 taken 1152 times.
✓ Branch 8 taken 368961 times.
✓ Branch 9 taken 103959 times.
 472920 while (doneWorkCount != total && std::chrono::duration_cast<std::chrono::milliseconds>(t2 - t1).count() < 50) {
1531
1/2
✓ Branch 0 taken 368961 times.
✗ Branch 1 not taken.
 368961 std::this_thread::sleep_for(std::chrono::microseconds(250));
1532 368961 t2 = clock.now();
1533 }
1534
2/2
✓ Branch 0 taken 102807 times.
✓ Branch 1 taken 1152 times.
 103959 if (doneWorkCount == total) {
1535 #ifndef PLATFORM_OSX
1536 // now make sure the buffers are drained for the early packets so that they are
1537 // fully received before we send the late packets (sync or broadcast).
1538 // Only worth the wait if there actually are late packets this frame.
1539 102807 bool haveLate = false;
1540
2/2
✓ Branch 0 taken 112413 times.
✓ Branch 1 taken 102302 times.
 214715 for (auto& msgs : messages.messages) {
1541
6/6
✓ Branch 0 taken 107567 times.
✓ Branch 1 taken 4846 times.
✓ Branch 2 taken 505 times.
✓ Branch 3 taken 107062 times.
✓ Branch 4 taken 505 times.
✓ Branch 5 taken 111908 times.
 112413 if (!msgs.second.empty() && msgs.first >= LATE_MESSAGES_START) {
1542 505 haveLate = true;
1543 505 break;
1544 }
1545 }
1546
2/2
✓ Branch 0 taken 505 times.
✓ Branch 1 taken 102302 times.
 102807 if (haveLate) {
1547 // Collect every early-message socket that may still have data
1548 // queued, then wait on them together: the drains overlap in
1549 // wall-clock, so the bound is one drain, not one per destination.
1550 505 std::vector<int> drainFds;
1551 505 bool anyPaced = false;
1552
2/2
✓ Branch 0 taken 1014 times.
✓ Branch 1 taken 505 times.
 1519 for (auto& msgs : messages.messages) {
1553
6/6
✓ Branch 0 taken 917 times.
✓ Branch 1 taken 97 times.
✓ Branch 2 taken 412 times.
✓ Branch 3 taken 505 times.
✓ Branch 4 taken 412 times.
✓ Branch 5 taken 602 times.
 1014 if (!msgs.second.empty() && msgs.first < LATE_MESSAGES_START) {
1554
1/2
✓ Branch 0 taken 412 times.
✗ Branch 1 not taken.
 412 std::shared_ptr<SendSocketInfo> socketInfo = findOrCreateSocket(msgs.first);
1555 // try_lock: a straggler worker still sending here means
1556 // sync ordering is already lost for this destination;
1557 // don't stall the frame thread waiting for it
1558 412 std::unique_lock<std::mutex> lg(socketInfo->sendLock, std::try_to_lock);
1559
3/6
✓ Branch 0 taken 412 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 412 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 412 times.
✗ Branch 5 not taken.
 412 if (lg.owns_lock() && !socketInfo->sockets.empty()) {
1560
1/2
✓ Branch 0 taken 412 times.
✗ Branch 1 not taken.
 412 drainFds.push_back(socketInfo->sockets[socketInfo->curSocket]);
1561 412 anyPaced |= socketInfo->paced;
1562 }
1563 412 }
1564 }
1565 // Paced packets sit in the fq qdisc for several ms; sleep
1566 // proportionally to the deepest queue instead of busy-polling.
1567 // The bound must cover a full send buffer draining at the
1568 // pacing rate (~34ms for 384KB at 90Mbps) or the sync packet
1569 // can beat the tail of the data it synchronizes.
1570
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 505 times.
 505 int totalBudget = anyPaced ? 35000 : 5000;
1571 505 int waited = 0;
1572
2/2
✓ Branch 0 taken 9521 times.
✓ Branch 1 taken 184 times.
 9705 while (waited < totalBudget) {
1573 9521 int maxBytes = 0;
1574
2/2
✓ Branch 0 taken 9428 times.
✓ Branch 1 taken 9521 times.
 18949 for (int fd : drainFds) {
1575 9428 int b = 0;
1576
5/6
✓ Branch 0 taken 9428 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 9200 times.
✓ Branch 3 taken 228 times.
✓ Branch 4 taken 9200 times.
✓ Branch 5 taken 228 times.
 9428 if (ioctl(fd, SIOCOUTQ, &b) == 0 && b > maxBytes) {
1577 9200 maxBytes = b;
1578 }
1579 }
1580
2/2
✓ Branch 0 taken 321 times.
✓ Branch 1 taken 9200 times.
 9521 if (maxBytes <= 1024) {
1581 321 break;
1582 }
1583 9200 int us = 100;
1584
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 9200 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 9200 if (anyPaced && pacingDrainRate > 0) {
1585 ✗ us = std::clamp((int)(((uint64_t)maxBytes * 8000000ULL) / pacingDrainRate), 100, 2000);
1586 }
1587 9200 us = std::min(us, totalBudget - waited);
1588
1/2
✓ Branch 0 taken 9200 times.
✗ Branch 1 not taken.
 9200 std::this_thread::sleep_for(std::chrono::microseconds(us));
1589 9200 waited += us;
1590 }
1591 505 }
1592 #endif
1593 // now output the LATE/Broadcast packets (likely sync packets)
1594
2/2
✓ Branch 0 taken 112413 times.
✓ Branch 1 taken 102807 times.
 215220 for (auto& msgs : messages.messages) {
1595
2/2
✓ Branch 0 taken 107567 times.
✓ Branch 1 taken 4846 times.
 112413 if (!msgs.second.empty()) {
1596
1/2
✓ Branch 0 taken 107567 times.
✗ Branch 1 not taken.
 107567 std::shared_ptr<SendSocketInfo> socketInfo = findOrCreateSocket(msgs.first);
1597
2/2
✓ Branch 0 taken 505 times.
✓ Branch 1 taken 107062 times.
 107567 if (msgs.first >= LATE_MESSAGES_START) {
1598 505 t1 = clock.now();
1599
1/2
✓ Branch 0 taken 505 times.
✗ Branch 1 not taken.
 505 int outputCount = SendMessages(msgs.first, socketInfo.get(), msgs.second);
1600 505 t2 = clock.now();
1601
2/4
✓ Branch 0 taken 505 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 505 times.
✗ Branch 3 not taken.
 505 long diff = std::chrono::duration_cast<std::chrono::milliseconds>(t2 - t1).count();
1602
5/6
✓ Branch 0 taken 413 times.
✓ Branch 1 taken 92 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 413 times.
✓ Branch 4 taken 92 times.
✓ Branch 5 taken 413 times.
 505 if ((outputCount != msgs.second.size()) || (diff > 100)) {
1603 92 socketInfo->errCount++;
1604
1605 // failed to send all messages or it took more than 100ms to send them
1606
3/4
✓ Branch 0 taken 92 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1 times.
✓ Branch 3 taken 91 times.
 92 if (ShouldLogSendError(socketInfo.get(), SEND_ERR_FRAME))
1607
3/6
✓ Branch 0 taken 1 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 1 times.
✗ Branch 5 not taken.
 2 LogErr(VB_CHANNELOUT, "sendmmsg() failed for UDP output (IP: %s output count: %d/%d time: %u ms errCount: %d) with error: %d %s\n",
1608 HexToIP(msgs.first).c_str(),
1609 outputCount, msgs.second.size(), diff, socketInfo->errCount.load(),
1610 errno,
1611 FPPstrerror(errno));
1612 } else {
1613 413 socketInfo->clearErrors();
1614 }
1615 }
1616
2/2
✓ Branch 0 taken 276 times.
✓ Branch 1 taken 107291 times.
 215134 if (socketInfo->errCount.load(std::memory_order_relaxed) >= 3) {
1617 // we'll ping the controllers and rebuild the valid message list, this could take time
1618
1/2
✓ Branch 0 taken 276 times.
✗ Branch 1 not taken.
 276 PingControllers(false);
1619 276 socketInfo->clearErrors();
1620 }
1621 107567 }
1622 }
1623 }
1624 103959 return 1;
1625 103959 }
1626
2/2
✓ Branch 0 taken 702 times.
✓ Branch 1 taken 702 times.
 1404 for (auto& msgs : messages.messages) {
1627
1/2
✓ Branch 0 taken 702 times.
✗ Branch 1 not taken.
 702 if (!msgs.second.empty()) {
1628
1/2
✓ Branch 0 taken 702 times.
✗ Branch 1 not taken.
 702 std::shared_ptr<SendSocketInfo> socketInfo = findOrCreateSocket(msgs.first);
1629 702 auto t1 = clock.now();
1630
1/2
✓ Branch 0 taken 702 times.
✗ Branch 1 not taken.
 702 int outputCount = SendMessages(msgs.first, socketInfo.get(), msgs.second);
1631 702 auto t2 = clock.now();
1632
2/4
✓ Branch 0 taken 702 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 702 times.
✗ Branch 3 not taken.
 702 long diff = std::chrono::duration_cast<std::chrono::milliseconds>(t2 - t1).count();
1633 702 socketInfo->bytesQueued.fetch_add(MessageBytes(msgs.second), std::memory_order_relaxed);
1634 702 socketInfo->framesQueued.fetch_add(1, std::memory_order_relaxed);
1635
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 702 times.
 702 if (outputCount != msgs.second.size())
1636 ✗ socketInfo->framesBehind.fetch_add(1, std::memory_order_relaxed);
1637
5/6
✓ Branch 0 taken 702 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 243 times.
✓ Branch 3 taken 459 times.
✓ Branch 4 taken 243 times.
✓ Branch 5 taken 459 times.
 702 if ((outputCount != msgs.second.size()) || (diff > 100)) {
1638 243 socketInfo->errCount++;
1639
1640 // failed to send all messages or it took more than 100ms to send them
1641
3/4
✓ Branch 0 taken 243 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 18 times.
✓ Branch 3 taken 225 times.
 243 if (ShouldLogSendError(socketInfo.get(), SEND_ERR_FRAME))
1642
3/6
✓ Branch 0 taken 18 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 18 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 18 times.
✗ Branch 5 not taken.
 36 LogErr(VB_CHANNELOUT, "sendmmsg() failed for UDP output (IP: %s output count: %d/%d time: %u ms errCount: %d) with error: %d %s\n",
1643 HexToIP(msgs.first).c_str(),
1644 outputCount, msgs.second.size(), diff, socketInfo->errCount.load(),
1645 errno,
1646 FPPstrerror(errno));
1647
1648
2/2
✓ Branch 0 taken 77 times.
✓ Branch 1 taken 166 times.
 486 if (socketInfo->errCount.load(std::memory_order_relaxed) >= 3) {
1649 // we'll ping the controllers and rebuild the valid message list, this could take time
1650
1/2
✓ Branch 0 taken 77 times.
✗ Branch 1 not taken.
 77 PingControllers(false);
1651 77 socketInfo->clearErrors();
1652 }
1653 } else {
1654 459 socketInfo->clearErrors();
1655 }
1656 702 }
1657 }
1658 702 return 1;
1659
2/6
✓ Branch 0 taken 26 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 26 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
 104893 }
1660
1661 // Not 57: that id belongs to the plugin-reinstall warning (Plugins.cpp), and
1662 // sharing it gave this banner that warning's title and help.
1663
1664 #ifndef PLATFORM_OSX
1665 // Total packets dropped by the qdiscs on the ethernet interfaces. This is
1666 // the only place fq enqueue drops (e.g. flow_limit) are visible: they return
1667 // NET_XMIT_CN, which UDP reports to the sender as success, so neither
1668 // SndbufErrors nor tx_dropped ever increments for them.
1669 25 static uint64_t qdiscDroppedCount() {
1670 25 uint64_t total = 0;
1671
5/8
✓ Branch 0 taken 25 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 25 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 47 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 47 times.
✓ Branch 7 taken 25 times.
 72 for (const auto& entry : std::filesystem::directory_iterator("/sys/class/net/")) {
1672
2/4
✓ Branch 0 taken 47 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 47 times.
✗ Branch 3 not taken.
 47 std::string dev = entry.path().filename();
1673
3/4
✓ Branch 0 taken 47 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 27 times.
✓ Branch 3 taken 20 times.
 47 if (!IsShowEthernetInterface(dev)) {
1674 27 continue;
1675 }
1676
4/8
✓ Branch 0 taken 20 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 20 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 20 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 20 times.
✗ Branch 7 not taken.
 20 std::string cmd = TcPath() + " -s qdisc show dev " + dev + " 2>/dev/null";
1677
1/2
✓ Branch 0 taken 20 times.
✗ Branch 1 not taken.
 20 FILE* pipe = popen(cmd.c_str(), "r");
1678
1/2
✓ Branch 0 taken 20 times.
✗ Branch 1 not taken.
 20 if (pipe) {
1679 char buf[512];
1680
3/4
✓ Branch 0 taken 100 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 80 times.
✓ Branch 3 taken 20 times.
 100 while (fgets(buf, sizeof(buf), pipe)) {
1681 80 const char* d = strstr(buf, "(dropped ");
1682
2/2
✓ Branch 0 taken 17 times.
✓ Branch 1 taken 63 times.
 80 if (d) {
1683 17 total += strtoull(d + 9, nullptr, 10);
1684 }
1685 }
1686
1/2
✓ Branch 0 taken 20 times.
✗ Branch 1 not taken.
 20 pclose(pipe);
1687 }
1688
2/2
✓ Branch 0 taken 20 times.
✓ Branch 1 taken 27 times.
 72 }
1689 25 return total;
1690 }
1691 #endif
1692
1693 // Called with the unicast destinations from the frame thread's snapshot, on
1694 // the stats thread. Raises the over-budget banner for a paced destination that
1695 // fell behind on more than 1% of its frames since the last check, and clears it
1696 // on the first check that doesn't.
1697 25 static void CheckPacingBudget(const std::vector<PacingBudgetDest>& dests,
1698 const std::list<UDPOutputData*>& outputs) {
1699 25 auto now = std::chrono::steady_clock::now();
1700
2/2
✓ Branch 0 taken 26 times.
✓ Branch 1 taken 25 times.
 51 for (auto& [key, ip, si] : dests) {
1701
2/4
✓ Branch 0 taken 26 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 26 times.
✗ Branch 3 not taken.
 26 double secs = std::chrono::duration<double>(now - si->budgetCheckTime).count();
1702 26 si->budgetCheckTime = now;
1703 26 uint64_t bytes = si->bytesQueued.exchange(0);
1704 26 unsigned int frames = si->framesQueued.exchange(0);
1705 26 unsigned int behind = si->framesBehind.exchange(0);
1706 26 unsigned int rate = si->pacedRate;
1707 // a handful of late frames is a hiccup, not something to change config for
1708
7/10
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 22 times.
✓ Branch 2 taken 4 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 4 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 3 times.
✓ Branch 7 taken 1 times.
✗ Branch 8 not taken.
✓ Branch 9 taken 3 times.
 26 if (rate == 0 || frames == 0 || secs <= 0 || behind < 10 || behind * 100ULL < frames) {
1709
1/2
✓ Branch 0 taken 23 times.
✗ Branch 1 not taken.
 23 ClearPacingBudgetWarning(si.get());
1710 23 continue;
1711 }
1712 // Rounded up to 10 Mbps so the text, which is what identifies the
1713 // banner, stays the same from one check to the next.
1714 3 unsigned int needMbps = (unsigned int)std::ceil(bytes * 8 / secs / 1000000.0 / 10.0) * 10;
1715 3 unsigned int pacedMbps = (unsigned int)(rate * 8ULL / 1000000ULL);
1716
1/2
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
 3 std::string name = ip;
1717
1/2
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
 3 for (auto o : outputs) {
1718 3 bool valid = false;
1719
5/10
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 3 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 3 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 3 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 3 times.
✗ Branch 9 not taken.
 3 if (!o->description.empty() && UDPOutputData::toInetAddr(o->ipAddress, valid) == (in_addr_t)key && valid) {
1720
3/6
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 3 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 3 times.
✗ Branch 5 not taken.
 3 name += " (" + o->description + ")";
1721 3 break;
1722 }
1723 }
1724
1/2
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
 3 LogWarn(VB_CHANNELOUT, "UDP output to %s: %u of %u frames late or incomplete in the last %.0f s; sending ~%u Mbps, paced at %u Mbps\n",
1725 name.c_str(), behind, frames, secs, needMbps, pacedMbps);
1726
3/8
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✓ Branch 3 taken 1 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
✗ Branch 7 not taken.
 5 std::string need = needMbps > pacedMbps ? "needs about " + std::to_string(needMbps) + " Mbps but is paced at "
1727
8/12
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 1 times.
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 1 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 1 times.
✓ Branch 7 taken 2 times.
✓ Branch 8 taken 2 times.
✓ Branch 9 taken 1 times.
✗ Branch 10 not taken.
✗ Branch 11 not taken.
 7 : "is sending close to its pacing rate of ";
1728
5/10
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 3 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 3 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 3 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 3 times.
✗ Branch 9 not taken.
 6 RaisePacingBudgetWarning(si.get(), "Controller " + name + " " + need + std::to_string(pacedMbps) +
1729
1/2
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
 3 " Mbps, so its frames arrive late or incomplete and its lights will stutter.");
1730 3 }
1731 25 }
1732
1733 25 void UDPOutput::CheckLocalDrops() {
1734 #ifndef PLATFORM_OSX
1735 // The protocols have no acks, so the only drops we can actually see are the
1736 // local ones. Surface them so users can tell "player is dropping" apart
1737 // from "switch/controller is dropping".
1738 25 uint64_t sndbufErrors = 0;
1739
3/6
✓ Branch 0 taken 25 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 25 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 25 times.
✗ Branch 5 not taken.
 25 std::istringstream snmp(GetFileContents("/proc/net/snmp"));
1740 25 std::string line;
1741 25 std::vector<std::string> headers;
1742
3/6
✓ Branch 0 taken 250 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 250 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 250 times.
✗ Branch 5 not taken.
 250 while (std::getline(snmp, line)) {
1743
3/4
✓ Branch 0 taken 250 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 50 times.
✓ Branch 3 taken 200 times.
 250 if (line.rfind("Udp:", 0) == 0) {
1744
1/2
✓ Branch 0 taken 50 times.
✗ Branch 1 not taken.
 50 std::istringstream ss(line);
1745 50 std::vector<std::string> tokens;
1746 50 std::string token;
1747
4/6
✓ Branch 0 taken 550 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 550 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 500 times.
✓ Branch 5 taken 50 times.
 550 while (ss >> token) {
1748
1/2
✓ Branch 0 taken 500 times.
✗ Branch 1 not taken.
 500 tokens.push_back(token);
1749 }
1750
2/2
✓ Branch 0 taken 25 times.
✓ Branch 1 taken 25 times.
 50 if (headers.empty()) {
1751
1/2
✓ Branch 0 taken 25 times.
✗ Branch 1 not taken.
 25 headers = tokens;
1752 } else {
1753
5/6
✓ Branch 0 taken 225 times.
✓ Branch 1 taken 25 times.
✓ Branch 2 taken 225 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 225 times.
✓ Branch 5 taken 25 times.
 250 for (size_t x = 1; x < headers.size() && x < tokens.size(); x++) {
1754
3/4
✓ Branch 0 taken 225 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 25 times.
✓ Branch 3 taken 200 times.
 225 if (headers[x] == "SndbufErrors") {
1755 // strtoull rather than stoull: this runs on a detached
1756 // thread where an unexpected exception is fatal
1757 25 sndbufErrors = strtoull(tokens[x].c_str(), nullptr, 10);
1758 }
1759 }
1760 25 break;
1761 }
1762
6/6
✓ Branch 0 taken 25 times.
✓ Branch 1 taken 25 times.
✓ Branch 2 taken 25 times.
✓ Branch 3 taken 25 times.
✓ Branch 4 taken 25 times.
✓ Branch 5 taken 25 times.
 100 }
1763 }
1764
3/6
✓ Branch 0 taken 25 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 25 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 25 times.
✗ Branch 5 not taken.
 25 uint64_t txDropped = strtoull(GetFileContents("/sys/class/net/" + outInterface + "/statistics/tx_dropped").c_str(), nullptr, 10);
1765
1/2
✓ Branch 0 taken 25 times.
✗ Branch 1 not taken.
 25 uint64_t qdiscDropped = qdiscDroppedCount();
1766
1767 // counters can go backwards if the interface bounces or a driver reloads;
1768 // rebase silently rather than computing a huge unsigned delta
1769
5/8
✓ Branch 0 taken 9 times.
✓ Branch 1 taken 16 times.
✓ Branch 2 taken 9 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 9 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 9 times.
✗ Branch 7 not taken.
 25 if (statBaselineValid && sndbufErrors >= lastSndbufErrors && txDropped >= lastTxDropped && qdiscDropped >= lastQdiscDropped) {
1770 9 uint64_t sbDelta = sndbufErrors - lastSndbufErrors;
1771 9 uint64_t txDelta = txDropped - lastTxDropped;
1772 9 uint64_t qdDelta = qdiscDropped - lastQdiscDropped;
1773
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 7 times.
 9 if (sbDelta) {
1774 // host-wide counter (covers every UDP socket of every process),
1775 // so log it as a diagnostic breadcrumb but don't raise a warning
1776
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 LogWarn(VB_CHANNELOUT, "System-wide UDP send buffer errors since last check: %llu (may include non-FPP traffic)\n",
1777 (unsigned long long)sbDelta);
1778 }
1779
3/4
✓ Branch 0 taken 9 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✓ Branch 3 taken 7 times.
 9 if (txDelta || qdDelta) {
1780
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 LogWarn(VB_CHANNELOUT, "Local packet drops since last check: %llu on interface %s, %llu in ethernet qdiscs\n",
1781 (unsigned long long)txDelta, outInterface.c_str(), (unsigned long long)qdDelta);
1782
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 if (!dropWarningActive) {
1783
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 WarningHolder::AddWarning(LOCAL_DROP_WARNING_ID, LOCAL_DROP_WARNING);
1784 2 dropWarningActive = true;
1785 }
1786
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 7 times.
 7 } else if (dropWarningActive) {
1787 ✗ WarningHolder::RemoveWarning(LOCAL_DROP_WARNING_ID, LOCAL_DROP_WARNING);
1788 ✗ dropWarningActive = false;
1789 }
1790 }
1791 25 lastSndbufErrors = sndbufErrors;
1792 25 lastTxDropped = txDropped;
1793 25 lastQdiscDropped = qdiscDropped;
1794 25 statBaselineValid = true;
1795 #endif
1796 25 }
1797
1798 2733 void UDPOutput::PingControllers(bool failedOnly) {
1799
1/2
✓ Branch 0 taken 2733 times.
✗ Branch 1 not taken.
 2733 LogExcess(VB_CHANNELOUT, "Pinging controllers to see what is online\n");
1800 // Held across the pings: a host removed meanwhile (its output deleted)
1801 // would be pinged again after removePingTarget() stopped it, and that
1802 // ping's callback would outlive this output.
1803
1/2
✓ Branch 0 taken 2733 times.
✗ Branch 1 not taken.
 2733 std::lock_guard<std::mutex> lk(pingHostsLock);
1804
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2733 times.
 2733 if (pingsStopped) {
1805 ✗ return;
1806 }
1807
2/2
✓ Branch 0 taken 147 times.
✓ Branch 1 taken 2733 times.
 2880 for (auto& h : pingHosts) {
1808 147 std::shared_ptr<PingHost> host = h.second;
1809
2/2
✓ Branch 0 taken 38 times.
✓ Branch 1 taken 109 times.
 147 if (failedOnly) {
1810
1/2
✓ Branch 0 taken 38 times.
✗ Branch 1 not taken.
 38 std::unique_lock<std::mutex> l(host->lock);
1811
5/6
✓ Branch 0 taken 10 times.
✓ Branch 1 taken 28 times.
✓ Branch 2 taken 10 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 10 times.
✓ Branch 5 taken 28 times.
 38 if (host->answered && host->failCount == 0) {
1812 10 continue;
1813 }
1814
2/2
✓ Branch 0 taken 28 times.
✓ Branch 1 taken 10 times.
 38 }
1815
2/4
✓ Branch 0 taken 137 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 137 times.
✗ Branch 3 not taken.
 137 PingManager::INSTANCE.addPeriodicPing(host->address, UDP_PING_TIMEOUT, 15000, [this, host](int ms) {
1816 59 OnPingResult(host, ms);
1817 59 });
1818
2/2
✓ Branch 0 taken 137 times.
✓ Branch 1 taken 10 times.
 147 }
1819
1/2
✓ Branch 0 taken 2733 times.
✗ Branch 1 not taken.
 2733 }
1820
1821 // caller holds host.lock
1822 13 void UDPOutput::SetHostValid(PingHost& host, bool valid) {
1823 13 host.valid = valid;
1824
2/2
✓ Branch 0 taken 15 times.
✓ Branch 1 taken 13 times.
 28 for (auto o : host.outputs) {
1825 15 o->valid = valid;
1826 }
1827
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 9 times.
 13 if (valid) {
1828
2/4
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 4 times.
✗ Branch 3 not taken.
 4 WarningHolder::RemoveWarning(27, host.warning);
1829 4 LogWarn(VB_CHANNELOUT, "Could ping host %s, re-adding to outputs\n", host.address.c_str());
1830 } else {
1831
1/2
✓ Branch 0 taken 9 times.
✗ Branch 1 not taken.
 9 WarningHolder::AddWarning(27, host.warning);
1832 9 LogWarn(VB_CHANNELOUT, "Could not ping host %s, removing from output\n", host.address.c_str());
1833 }
1834 13 }
1835
1836 110 void UDPOutput::OnPingResult(const std::shared_ptr<PingHost>& host, int ms) {
1837
1/2
✓ Branch 0 taken 110 times.
✗ Branch 1 not taken.
 110 std::unique_lock<std::mutex> l(host->lock);
1838
2/2
✓ Branch 0 taken 77 times.
✓ Branch 1 taken 33 times.
 110 if (ms > 0) {
1839 77 host->failCount = 0;
1840
2/2
✓ Branch 0 taken 60 times.
✓ Branch 1 taken 17 times.
 77 if (!host->answered) {
1841 60 host->answered = true;
1842 60 --failedCount;
1843 }
1844
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 73 times.
 77 if (!host->valid) {
1845
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4 SetHostValid(*host, true);
1846 }
1847 77 return;
1848 }
1849
1/2
✓ Branch 0 taken 33 times.
✗ Branch 1 not taken.
 33 LogDebug(VB_CHANNELOUT, "Could not ping host %s Fail count: %d Currently Valid: %d\n", host->address.c_str(), host->failCount, host->valid);
1850 33 host->failCount++;
1851
2/2
✓ Branch 0 taken 20 times.
✓ Branch 1 taken 13 times.
 33 if (host->failCount == 1) {
1852 // ignore a single ping failure, could be transient
1853
2/2
✓ Branch 0 taken 11 times.
✓ Branch 1 taken 9 times.
 20 } else if (host->failCount == 2) {
1854 // if two pings fail, we'll try a HEAD request via HTTP. One at a
1855 // time: a probe can take the whole curl timeout, and the pings keep
1856 // failing meanwhile.
1857
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 11 times.
 11 if (host->headPending) {
1858 ✗ return;
1859 }
1860 11 host->headPending = true;
1861
1/2
✓ Branch 0 taken 11 times.
✗ Branch 1 not taken.
 11 l.unlock();
1862 11 std::shared_ptr<PingHost> h = host;
1863
7/14
✓ Branch 0 taken 11 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 11 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 11 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 11 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 11 times.
✗ Branch 9 not taken.
✓ Branch 10 taken 11 times.
✗ Branch 11 not taken.
✓ Branch 12 taken 11 times.
✗ Branch 13 not taken.
 66 CurlManager::INSTANCE.add("http://" + h->address + "/", "HEAD", "", {}, [this, h](int rc, const std::string& resp) {
1864
1/2
✓ Branch 0 taken 9 times.
✗ Branch 1 not taken.
 9 std::unique_lock<std::mutex> l(h->lock);
1865 9 h->headPending = false;
1866
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 9 times.
 9 if (rc) {
1867 ✗ h->failCount = 0;
1868 ✗ if (!h->answered) {
1869 ✗ h->answered = true;
1870 ✗ --failedCount;
1871 }
1872 ✗ if (!h->valid) {
1873 ✗ SetHostValid(*h, true);
1874 }
1875 }
1876 9 },
1877 11 curlOwner);
1878
1/2
✓ Branch 0 taken 9 times.
✗ Branch 1 not taken.
 20 } else if (host->failCount >= 3) {
1879 // three pings and a HEAD request failed, mark invalid
1880
1/2
✓ Branch 0 taken 9 times.
✗ Branch 1 not taken.
 9 if (host->valid) {
1881
1/2
✓ Branch 0 taken 9 times.
✗ Branch 1 not taken.
 9 SetHostValid(*host, false);
1882 }
1883
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 9 times.
 9 if (host->failCount > 5) {
1884 // make sure we wrap around so another HEAD request later may pick it up
1885 ✗ host->failCount = 0;
1886 }
1887 }
1888
2/2
✓ Branch 0 taken 33 times.
✓ Branch 1 taken 77 times.
 110 }
1889 791 void UDPOutput::DumpConfig() {
1890 791 ChannelOutput::DumpConfig();
1891 791 LogDebug(VB_CHANNELOUT, " Interface : %s\n", outInterface.c_str());
1892 791 LogDebug(VB_CHANNELOUT, " Threaded : %d\n", useThreadedOutput);
1893 791 LogDebug(VB_CHANNELOUT, " Blocking : %d\n", blockingOutput);
1894 791 LogDebug(VB_CHANNELOUT, " Needs Broadcast : %d\n", needsBroadcast);
1895
2/2
✓ Branch 0 taken 902 times.
✓ Branch 1 taken 791 times.
 1693 for (auto u : outputs) {
1896
1/2
✓ Branch 0 taken 902 times.
✗ Branch 1 not taken.
 902 u->DumpConfig();
1897 }
1898 791 }
1899
1900 216895 std::shared_ptr<SendSocketInfo> UDPOutput::findOrCreateSocket(unsigned int socketKey, int sc) {
1901
1/2
✓ Branch 0 taken 216895 times.
✗ Branch 1 not taken.
 216895 auto it = messages.sendSockets.find(socketKey);
1902 216895 std::shared_ptr<SendSocketInfo> info;
1903
2/2
✓ Branch 0 taken 866 times.
✓ Branch 1 taken 216029 times.
 216895 if (it == messages.sendSockets.end()) {
1904
1/2
✓ Branch 0 taken 866 times.
✗ Branch 1 not taken.
 866 info = std::make_shared<SendSocketInfo>();
1905
1/2
✓ Branch 0 taken 866 times.
✗ Branch 1 not taken.
 866 messages.sendSockets[socketKey] = info;
1906 } else {
1907 216029 info = it->second;
1908 }
1909
1910 // try_lock: if a straggler worker from a previous frame is still sending
1911 // to this destination, its sockets already exist and the pacing check can
1912 // wait a frame - nothing here is worth stalling the frame thread on
1913 216895 std::unique_lock<std::mutex> lg(info->sendLock, std::try_to_lock);
1914
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 216895 times.
 216895 if (!lg.owns_lock()) {
1915 ✗ return info;
1916 }
1917
2/2
✓ Branch 0 taken 1045 times.
✓ Branch 1 taken 215850 times.
 216895 if (info->sockets.empty()) {
1918 1045 bool broadCast = socketKey == BROADCAST_MESSAGES_KEY;
1919 1045 bool multiCast = isMulticastKey(socketKey);
1920
2/2
✓ Branch 0 taken 1045 times.
✓ Branch 1 taken 866 times.
 1911 for (int x = info->sockets.size(); x < sc; x++) {
1921
1/2
✓ Branch 0 taken 1045 times.
✗ Branch 1 not taken.
 1045 int s = createSocket(0, broadCast, multiCast);
1922
2/2
✓ Branch 0 taken 179 times.
✓ Branch 1 taken 866 times.
 1045 if (s == -1) {
1923 179 break;
1924 }
1925
1/2
✓ Branch 0 taken 866 times.
✗ Branch 1 not taken.
 866 info->sockets.push_back(s);
1926 }
1927 }
1928 #if !defined(PLATFORM_OSX) && defined(SO_MAX_PACING_RATE)
1929 // Pacing is only applied to unicast destination sockets: the key IS
1930 // the destination IP, so the rate cap is truly per controller, and we
1931 // can check that the route's egress interface actually has fq (wifi
1932 // and other non-fq routes keep the legacy drain-based pacing). The
1933 // shared multicast socket is left unpaced - it carries the aggregate
1934 // of all multicast controllers so a per-controller rate would be wrong.
1935 // Not while the socket couldn't be opened: the decision latches, and the
1936 // socket made later would never get the rate and buffer it applies.
1937
4/4
✓ Branch 0 taken 5012 times.
✓ Branch 1 taken 151789 times.
✓ Branch 2 taken 4833 times.
✓ Branch 3 taken 179 times.
 156801 if (pacingEnabled && !info->pacedChecked && !info->sockets.empty() &&
1938
8/8
✓ Branch 0 taken 156801 times.
✓ Branch 1 taken 60094 times.
✓ Branch 2 taken 1838 times.
✓ Branch 3 taken 2995 times.
✓ Branch 4 taken 1333 times.
✓ Branch 5 taken 505 times.
✓ Branch 6 taken 1333 times.
✓ Branch 7 taken 215562 times.
 373696 socketKey > ARTNET_MESSAGES_KEY && socketKey < LATE_MESSAGES_START) {
1939
2/2
✓ Branch 0 taken 497 times.
✓ Branch 1 taken 836 times.
 1333 if (info->pacedCheckCountdown > 0) {
1940 497 --info->pacedCheckCountdown;
1941 } else {
1942
1/2
✓ Branch 0 taken 836 times.
✗ Branch 1 not taken.
 836 std::string iface = egressInterfaceFor((in_addr_t)socketKey);
1943
2/2
✓ Branch 0 taken 832 times.
✓ Branch 1 taken 4 times.
 836 if (!iface.empty()) {
1944 // only latch the decision once the route has resolved; during
1945 // startup the route may not exist yet (interface waiting on
1946 // DHCP) and we want to try again later or on an address change
1947 832 info->pacedChecked = true;
1948 // per-destination rate override (e.g. a slower cap for an
1949 // ESP32-class controller, or a faster one for a gigabit FPP
1950 // instance) falls back to the global pacingRate. ~0U in the
1951 // map means the controller was explicitly set to unpaced.
1952 832 unsigned int destRate = pacingRate;
1953
1/2
✓ Branch 0 taken 832 times.
✗ Branch 1 not taken.
 832 auto rit = perDestPacingRate.find((in_addr_t)socketKey);
1954
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 830 times.
 832 if (rit != perDestPacingRate.end()) {
1955 2 destRate = rit->second;
1956 }
1957 // pace only if this destination actually has a rate cap (a
1958 // global/override rate that isn't 0 or explicitly-unpaced) AND
1959 // the route egresses an fq interface
1960
6/8
✓ Branch 0 taken 832 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 831 times.
✓ Branch 3 taken 1 times.
✓ Branch 4 taken 831 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 104 times.
✓ Branch 7 taken 727 times.
 832 bool shouldPace = destRate != 0 && destRate != ~0U && interfaceHasFQ(iface);
1961
2/2
✓ Branch 0 taken 105 times.
✓ Branch 1 taken 727 times.
 832 if (shouldPace != info->paced) {
1962
4/6
✓ Branch 0 taken 102 times.
✓ Branch 1 taken 3 times.
✓ Branch 2 taken 105 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 105 times.
✗ Branch 5 not taken.
 105 LogInfo(VB_CHANNELOUT, "UDP output to %s via %s is %s\n",
1963 HexToIP(socketKey).c_str(), iface.c_str(),
1964 shouldPace ? "paced" : "not paced (no fq qdisc on egress)");
1965 105 info->paced = shouldPace;
1966
2/2
✓ Branch 0 taken 102 times.
✓ Branch 1 taken 3 times.
 105 info->pacedRate = shouldPace ? destRate : 0;
1967 // ~0 = unlimited; used when un-pacing after the route
1968 // moved to a non-fq interface (e.g. eth -> wifi failover)
1969
2/2
✓ Branch 0 taken 102 times.
✓ Branch 1 taken 3 times.
 105 unsigned int rate = shouldPace ? destRate : ~0U;
1970 // paced sockets need the buffer to hold a full frame for
1971 // the qdisc to clock out; unpaced revert to the small
1972 // buffer the drain-based pacing depends on
1973
4/6
✓ Branch 0 taken 102 times.
✓ Branch 1 taken 3 times.
✓ Branch 2 taken 102 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 3 times.
 105 int bufSize = shouldPace ? pacedSndBufSize() : ((blockingOutput ? 4096 : (MSGS_PER_SENDMMSG * 1511)) - 1);
1974
2/2
✓ Branch 0 taken 105 times.
✓ Branch 1 taken 105 times.
 210 for (int s : info->sockets) {
1975 105 setsockopt(s, SOL_SOCKET, SO_MAX_PACING_RATE, &rate, sizeof(rate));
1976 105 setsockopt(s, SOL_SOCKET, SO_SNDBUF, &bufSize, sizeof(bufSize));
1977 }
1978 }
1979 } else {
1980 // no route: don't re-run the lookup syscalls on the hot path
1981 // every frame, retry after a few seconds worth of frames
1982 4 info->pacedCheckCountdown = 150;
1983 }
1984 836 }
1985 }
1986 #endif
1987
2/2
✓ Branch 0 taken 874 times.
✓ Branch 1 taken 216021 times.
 216895 if (info->curSocket == -1) {
1988 874 info->curSocket = 0;
1989 }
1990 216895 return info;
1991 216895 }
1992
1993 1051 int UDPOutput::createSocket(int port, bool broadCast, bool multiCast) {
1994 1051 int sendSocket = OpenSocket(AF_INET, SOCK_DGRAM, 0);
1995
2/2
✓ Branch 0 taken 179 times.
✓ Branch 1 taken 872 times.
 1051 if (sendSocket < 0) {
1996 179 int err = errno;
1997 // Out of file descriptors, typically. This used to exit(1) and take
1998 // the show down; the destination is left without a socket, its sends
1999 // fail and are counted, and the next frame tries again, so output
2000 // comes back once descriptors free up (as e131bridge does since
2001 // 42183b6d4). That is a try per destination per frame, so the log
2002 // line is limited to one every few seconds.
2003 static std::atomic<long long> lastLog{ 0 };
2004
1/2
✓ Branch 0 taken 179 times.
✗ Branch 1 not taken.
 179 long long now = GetMonotonicTimeMS();
2005 179 long long last = lastLog.load();
2006
6/8
✓ Branch 0 taken 177 times.
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 177 times.
✓ Branch 4 taken 2 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 2 times.
✓ Branch 7 taken 177 times.
 181 if ((last == 0 || now - last >= SEND_ERR_LOG_INTERVAL_MS) && lastLog.compare_exchange_strong(last, now)) {
2007
2/4
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
 2 LogErr(VB_CHANNELOUT, "Error opening datagram socket: %s\n", FPPstrerror(err));
2008 }
2009 179 return -1;
2010 }
2011
2012 872 errno = 0;
2013 // Disable loopback so I do not receive my own datagrams.
2014 872 char loopch = 0;
2015
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 872 times.
 872 if (setsockopt(sendSocket, IPPROTO_IP, IP_MULTICAST_LOOP, (char*)&loopch, sizeof(loopch)) < 0) {
2016 ✗ LogErr(VB_CHANNELOUT, "Error setting IP_MULTICAST_LOOP error\n");
2017 ✗ close(sendSocket);
2018 ✗ return -1;
2019 }
2020 // Mark show traffic as Expedited Forwarding (DSCP EF). Managed switches
2021 // prioritize it out of the box and on wifi it maps to the WMM voice queue.
2022 872 int tos = 0xB8;
2023 872 setsockopt(sendSocket, IPPROTO_IP, IP_TOS, &tos, sizeof(tos));
2024 #ifndef PLATFORM_OSX
2025 // ahead of fppd's own HTTP/MQTT/curl traffic in the local qdisc
2026 872 int prio = 6;
2027 872 setsockopt(sendSocket, SOL_SOCKET, SO_PRIORITY, &prio, sizeof(prio));
2028 #endif
2029 // make sure the send buffer is actually set to a reasonable size for non-blocking mode
2030 // (paced sockets get a larger buffer in findOrCreateSocket)
2031
2/2
✓ Branch 0 taken 28 times.
✓ Branch 1 taken 844 times.
 872 int bufSize = (blockingOutput ? 4096 : (MSGS_PER_SENDMMSG * 1511)) - 1;
2032 872 setsockopt(sendSocket, SOL_SOCKET, SO_SNDBUF, &bufSize, sizeof(bufSize));
2033 // these sockets are for sending only, don't need a large receive buffer so
2034 // free some memory by setting to just a single page
2035 872 bufSize = 4096;
2036 872 setsockopt(sendSocket, SOL_SOCKET, SO_RCVBUF, &bufSize, sizeof(bufSize));
2037
2038
2/2
✓ Branch 0 taken 28 times.
✓ Branch 1 taken 844 times.
 872 if (blockingOutput) {
2039 struct timeval timeout;
2040 28 timeout.tv_sec = 0;
2041 28 timeout.tv_usec = 1000; // 1ms timeout
2042 28 setsockopt(sendSocket, SOL_SOCKET, SO_SNDTIMEO, &timeout, sizeof timeout);
2043 }
2044
2045
2/2
✓ Branch 0 taken 6 times.
✓ Branch 1 taken 866 times.
 872 if (broadCast) {
2046 6 int broadcast = 1;
2047
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6 times.
 6 if (setsockopt(sendSocket, SOL_SOCKET, SO_BROADCAST, &broadcast, sizeof(broadcast)) < 0) {
2048 ✗ LogErr(VB_CHANNELOUT, "Error setting SO_BROADCAST: %s\n", FPPstrerror(errno));
2049 ✗ close(sendSocket);
2050 ✗ return -1;
2051 }
2052 #ifndef PLATFORM_OSX
2053
2/2
✓ Branch 0 taken 16 times.
✓ Branch 1 taken 850 times.
 866 } else if (multiCast) {
2054 static struct sockaddr_in address;
2055 16 memcpy(&address, &localAddress, sizeof(localAddress));
2056 16 address.sin_port = ntohs(port);
2057
2058
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 16 times.
 16 if (setsockopt(sendSocket, IPPROTO_IP, IP_MULTICAST_IF, (char*)&address, sizeof(address)) < 0) {
2059 ✗ LogErr(VB_CHANNELOUT, "Error setting IP_MULTICAST_IF error\n");
2060 }
2061 #endif
2062 }
2063
4/4
✓ Branch 0 taken 866 times.
✓ Branch 1 taken 6 times.
✓ Branch 2 taken 16 times.
✓ Branch 3 taken 850 times.
 872 if (broadCast || multiCast) {
2064 static struct sockaddr_in address;
2065 22 memcpy(&address, &localAddress, sizeof(localAddress));
2066 22 address.sin_port = ntohs(port);
2067
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 22 times.
 22 if (bind(sendSocket, (struct sockaddr*)&address, sizeof(struct sockaddr_in)) == -1) {
2068 ✗ LogErr(VB_CHANNELOUT, "Error in bind:errno=%d, %s\n", errno, FPPstrerror(errno));
2069 }
2070 #ifndef PLATFORM_OSX
2071
2/4
✓ Branch 0 taken 22 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 22 times.
 22 if (connect(sendSocket, (struct sockaddr*)&address, sizeof(address)) < 0) {
2072 ✗ LogErr(VB_CHANNELOUT, "Error connecting IP_MULTICAST_LOOP socket\n");
2073 }
2074 #endif
2075 }
2076 872 return sendSocket;
2077 }
2078
2079 798 bool UDPOutput::InitNetwork(bool rebuildSockets) {
2080 // Returns true when the output interface has a usable address.
2081 char E131LocalAddress[16];
2082
1/2
✓ Branch 0 taken 798 times.
✗ Branch 1 not taken.
 798 int rv = GetInterfaceAddress(outInterface.c_str(), E131LocalAddress, NULL, NULL);
2083
1/2
✓ Branch 0 taken 798 times.
✗ Branch 1 not taken.
 798 LogDebug(VB_CHANNELOUT, "UDPLocalAddress = %s\n", E131LocalAddress);
2084 // GetInterfaceAddress() reports 127.0.0.1 for an interface without an
2085 // address; the entire 127.* subnet is localhost
2086
4/4
✓ Branch 0 taken 774 times.
✓ Branch 1 taken 24 times.
✓ Branch 2 taken 772 times.
✓ Branch 3 taken 2 times.
 798 bool haveAddress = rv == 0 && strncmp(E131LocalAddress, "127.", 4) != 0;
2087
2088 // createSocket() binds broadcast/multicast sockets to this address and
2089 // sends multicast out of it, and it runs on the frame thread under
2090 // socketMutex, so the address changes under that lock too. It is
2091 // refreshed every time: resetting it to loopback and then returning early
2092 // because sockets already existed is what left multicast going out of lo
2093 // after an address change. After an address change the sockets are
2094 // dropped in the same critical section, so a frame can't rebuild them on
2095 // the old address between the two.
2096
1/2
✓ Branch 0 taken 798 times.
✗ Branch 1 not taken.
 798 std::unique_lock<std::mutex> lk(socketMutex);
2097 798 memset(&localAddress, 0, sizeof(struct sockaddr_in));
2098 798 localAddress.sin_family = AF_INET;
2099 798 localAddress.sin_addr.s_addr = inet_addr(E131LocalAddress);
2100
2/2
✓ Branch 0 taken 7 times.
✓ Branch 1 taken 791 times.
 798 if (rebuildSockets) {
2101
1/2
✓ Branch 0 taken 7 times.
✗ Branch 1 not taken.
 7 messages.clearSockets();
2102 }
2103
6/6
✓ Branch 0 taken 774 times.
✓ Branch 1 taken 24 times.
✓ Branch 2 taken 3 times.
✓ Branch 3 taken 771 times.
✓ Branch 4 taken 3 times.
✓ Branch 5 taken 795 times.
 798 if (rv == 0 && !interfaceWarning.empty()) {
2104 // An interface that had no address yet (DHCP still waiting, cable
2105 // out) is usable once NEW_ADDR brings one; the banner was otherwise
2106 // only retracted at the next config load.
2107
2/4
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 3 times.
✗ Branch 3 not taken.
 3 WarningHolder::RemoveWarning(22, interfaceWarning);
2108 3 interfaceWarning.clear();
2109 }
2110
2111
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 798 times.
 798 if (!messages.sendSockets.empty()) {
2112 ✗ return haveAddress;
2113 }
2114
2/2
✓ Branch 0 taken 9 times.
✓ Branch 1 taken 789 times.
 798 if (needsBroadcast) {
2115
2/2
✓ Branch 0 taken 3 times.
✓ Branch 1 taken 6 times.
 9 if (rv) {
2116
1/2
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
 3 LogErr(VB_CHANNELOUT, "Invalid interface %s\n", outInterface.c_str());
2117
1/2
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
 3 interfaceWarning = "Invalid interface for UDP broadcast/multicast: " + outInterface;
2118
1/2
✓ Branch 0 taken 3 times.
✗ Branch 1 not taken.
 3 AddOutputConfigWarning(22, interfaceWarning);
2119 3 return false;
2120 }
2121
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6 times.
 6 if (!haveAddress) {
2122 ✗ return false;
2123 }
2124
1/2
✓ Branch 0 taken 6 times.
✗ Branch 1 not taken.
 6 int broadcastSocket = createSocket(0, true);
2125
1/2
✓ Branch 0 taken 6 times.
✗ Branch 1 not taken.
 6 messages.ForceSocket(BROADCAST_MESSAGES_KEY, broadcastSocket);
2126 }
2127 795 return haveAddress;
2128 798 }
2129
2130 826 void UDPOutput::StartingOutput() {
2131 826 outputRunning = true;
2132
2/2
✓ Branch 0 taken 1005 times.
✓ Branch 1 taken 826 times.
 1831 for (auto a : outputs) {
2133
7/8
✓ Branch 0 taken 995 times.
✓ Branch 1 taken 10 times.
✓ Branch 2 taken 992 times.
✓ Branch 3 taken 3 times.
✓ Branch 4 taken 992 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 992 times.
✓ Branch 7 taken 13 times.
 1005 if (a->valid && a->active && !a->started) {
2134 992 a->started = true;
2135
1/2
✓ Branch 0 taken 992 times.
✗ Branch 1 not taken.
 992 a->StartingOutput();
2136 }
2137 }
2138 826 }
2139 826 void UDPOutput::StoppingOutput() {
2140 // Every output that was started is stopped, whether or not it is still
2141 // valid: one whose ping failed mid-show would otherwise keep whatever
2142 // StartingOutput() set up (Twinkly's token timer) past its own deletion.
2143 826 outputRunning = false;
2144
2/2
✓ Branch 0 taken 1007 times.
✓ Branch 1 taken 826 times.
 1833 for (auto a : outputs) {
2145
2/2
✓ Branch 0 taken 995 times.
✓ Branch 1 taken 12 times.
 1007 if (a->started) {
2146 995 a->started = false;
2147
1/2
✓ Branch 0 taken 995 times.
✗ Branch 1 not taken.
 995 a->StoppingOutput();
2148 }
2149 }
2150 826 }
2151
2152 210 std::string UDPOutput::HexToIP(unsigned int hex) {
2153 // Extract each byte in LSB order
2154 210 uint8_t octet1 = hex & 0xFF; // Least significant byte
2155 210 uint8_t octet2 = (hex >> 8) & 0xFF;
2156 210 uint8_t octet3 = (hex >> 16) & 0xFF;
2157 210 uint8_t octet4 = (hex >> 24) & 0xFF; // Most significant byte
2158
2159
1/2
✓ Branch 0 taken 210 times.
✗ Branch 1 not taken.
 436 return std::to_string(octet1) + "." +
2160
2/4
✓ Branch 0 taken 210 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 210 times.
✗ Branch 3 not taken.
 824 std::to_string(octet2) + "." +
2161
2/4
✓ Branch 0 taken 210 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 210 times.
✗ Branch 3 not taken.
 824 std::to_string(octet3) + "." +
2162
1/2
✓ Branch 0 taken 210 times.
✗ Branch 1 not taken.
 614 std::to_string(octet4);
2163 }
2164