GCC Code Coverage Report

Directory:	./
File:	channeloutput/UDPOutput.h
Date:	2026-10-29 08:11:58

	Exec	Total	Coverage
Lines:	22	22	100.0%
Functions:	10	10	100.0%
Branches:	3	6	50.0%

  
      Line
      Branch
      Exec
      Source
    
      #pragma once
    
      /*
    
       * This file is part of the Falcon Player (FPP) and is Copyright (C)
    
       * 2013-2022 by the Falcon Player Developers.
    
       *
    
       * The Falcon Player (FPP) is free software, and is covered under
    
       * multiple Open Source licenses.  Please see the included 'LICENSES'
    
       * file for descriptions of what files are covered by each license.
    
       *
    
       * This source file is covered under the GPL v2 as described in the
    
       * included LICENSE.GPL file.
    
       */
    
      #include <set>
    
      #include <atomic>
    
      #include "fpp-json-fwd.h"
    
      #include <condition_variable>
    
      #include <list>
    
      #include <map>
    
      #include <memory>
    
      #include <mutex>
    
      #include <string>
    
      #include <thread>
    
      #include <vector>
    
      #include "../SysSocket.h"
    
      #include <netinet/in.h>
    
      #include "ChannelOutput.h"
    
      typedef void CURLM;
    
      #define MULTICAST_MESSAGES_KEY 0x00000001
    
      #define ANY_MESSAGES_KEY 0x00000002
    
      #define ARTNET_MESSAGES_KEY 0x00000003
    
      #define LATE_MESSAGES_START 0xFFFFFFF0
    
      #define E131_SYNC_KEY 0xFFFFFFFC
    
      #define ARTNET_SYNC_KEY 0xFFFFFFFD
    
      #define LATE_MULTICAST_MESSAGES_KEY 0xFFFFFFFE
    
      #define BROADCAST_MESSAGES_KEY 0xFFFFFFFF
    
      class SendSocketInfo;
    
      class UDPOutputMessages {
    
      public:
    
          UDPOutputMessages();
    
          ~UDPOutputMessages();
    
          void ForceSocket(unsigned int key, int socket, bool preventClose = false);
    
          int GetSocket(unsigned int key);
    
          std::vector<struct mmsghdr>& GetMessages(unsigned int key);
    
      929119
          std::vector<struct mmsghdr>& operator[](unsigned int key) { return GetMessages(key); }
    
      private:
    
          std::map<unsigned int, std::vector<struct mmsghdr>> messages;
    
          std::map<unsigned int, std::shared_ptr<SendSocketInfo>> sendSockets;
    
          // While set, every key an output asks for is noted here (once), so
    
          // UDPOutput knows which destinations each output's packets go to.
    
          std::vector<unsigned int>* recordKeys = nullptr;
    
          void clearMessages();
    
          void clearSockets();
    
          friend class UDPOutput;
    
      };
    
      class UDPOutputData {
    
      public:
    
          UDPOutputData(const Json::Value& config);
    
          virtual ~UDPOutputData();
    
          virtual bool IsPingable() = 0;
    
      941
          virtual bool Monitor() const { return monitor; }
    
          virtual void PrepareData(unsigned char* channelData, UDPOutputMessages& msgs) = 0;
    
      95410
          virtual void PostPrepareData(unsigned char* channelData, UDPOutputMessages& msgs) {}
    
          virtual void DumpConfig() = 0;
    
          // min is the first channel, 0-based; max is one past the last (every
    
          // protocol works it out as startChannel + count - 1 from the 1-based
    
          // startChannel).  UDPOutput converts it for ChannelOutput's inclusive max.
    
      1550
          virtual void GetRequiredChannelRange(int& min, int& max) {
    
      1550
              min = startChannel - 1;
    
      1550
              max = startChannel + channelCount - 1;
    
      1550
          }
    
      978
          virtual void StartingOutput() {}
    
      978
          virtual void StoppingOutput() {}
    
          virtual const std::string& GetOutputTypeString() const;
    
          // The messages last prepared are being sent.  A frame can be prepared
    
          // more than once before it goes out (its data changed in between), and
    
          // only the last preparation is sent, so a protocol sequence number
    
          // assigned in PrepareData() is committed here: advanced as it was
    
          // assigned, the discarded preparation skipped one on the wire.
    
      8452
          virtual void FrameSent() {}
    
          static in_addr_t toInetAddr(const std::string& ip, bool& valid);
    
      942
          static in_addr_t toInetAddr(const std::string& ip, std::atomic<bool>& valid) {
    
      942
              bool v = false;
    
        1/2✓ Branch 0 taken 942 times.
✗ Branch 1 not taken.

      942
              in_addr_t a = toInetAddr(ip, v);
    
      942
              valid = v;
    
      942
              return a;
    
          }
    
          std::string description;
    
          bool active;
    
          int startChannel;
    
          int channelCount;
    
          int type;
    
          std::string ipAddress;
    
          // Flipped by the ping thread (SetHostValid) while the output thread reads
    
          // it in PrepData().
    
          std::atomic<bool> valid;
    
          bool monitor;
    
          // per-destination pacing override in Mbps; -1 means "use the global rate".
    
          // 0 means this controller is explicitly unpaced (line rate).  A mix of a
    
          // gigabit FPP instance and a 30Mbps-class ESP32 controller can each get an
    
          // appropriate cap instead of one global setting for all of them.
    
          int pacingRateMbps = -1;
    
          // Set while StartingOutput() is owed its StoppingOutput().  valid can change
    
          // at runtime (a failed ping clears it), so it cannot say which outputs were
    
          // started; UDPOutput pairs the two calls through this instead.
    
          bool started = false;
    
          UDPOutputData(UDPOutputData const&) = delete;
    
          void operator=(UDPOutputData const& x) = delete;
    
      protected:
    
          void SaveFrame(unsigned char* channelData, int len);
    
          bool NeedToOutputFrame(unsigned char* channelData, int startChannel, int savedIdx, int count);
    
          bool deDuplicate = false;
    
          int skippedFrames;
    
          unsigned char* lastData;
    
      };
    
      class UDPOutput : public ChannelOutput {
    
      public:
    
          UDPOutput(unsigned int startChannel, unsigned int channelCount);
    
          virtual ~UDPOutput();
    
      1070
          virtual std::string GetOutputType() const override {
    
        1/2✓ Branch 0 taken 1070 times.
✗ Branch 1 not taken.

      2140
              return "UDPOutput";
    
          }
    
          virtual int Init(Json::Value config) override;
    
          virtual int Close(void) override;
    
          virtual void PrepData(unsigned char* channelData) override;
    
          virtual int SendData(unsigned char* channelData) override;
    
          virtual void DumpConfig(void) override;
    
          virtual void GetRequiredChannelRanges(const std::function<void(int, int)>& addRange) override;
    
          void addOutput(UDPOutputData*);
    
          // Outputs another channel output plugin (GenericUDP) sends through this
    
          // one.  Registered by owner with a factory rather than as data: this
    
          // instance is replaced whenever co-universes.json reloads, and the new one
    
          // has to rebuild them.  Unregistering only retires the data - a worker may
    
          // still be sending from its buffers - and it is freed with this instance.
    
          static void RegisterExternalOutput(const void* owner, std::function<UDPOutputData*()> create);
    
          static void UnregisterExternalOutput(const void* owner);
    
          int createSocket(int port = 0, bool broadCast = false, bool multiCast = false);
    
          static UDPOutput* INSTANCE;
    
          void BackgroundOutputWork();
    
          virtual void StartingOutput() override;
    
          virtual void StoppingOutput() override;
    
      private:
    
          int SendMessages(unsigned int key, SendSocketInfo* socketInfo, std::vector<struct mmsghdr>& sendmsgs);
    
          struct sockaddr_in localAddress;
    
          std::string outInterface;
    
          bool needsBroadcast = false;
    
          bool hasArtNet = false;
    
          // between StartingOutput() and StoppingOutput(); output thread only
    
          bool outputRunning = false;
    
          // tags this output's CurlManager requests so the destructor can cancel them
    
          const std::string curlOwner = "UDPOutput-" + std::to_string((uintptr_t)this);
    
          bool interfaceUp;
    
          // the "Invalid interface" banner InitNetwork() raised, while it is up;
    
          // InitNetwork() takes it down when the address arrives, Close() otherwise
    
          std::string interfaceWarning;
    
          // rebuildSockets: the address changed; drop the sockets made on the old one
    
          bool InitNetwork(bool rebuildSockets = false);
    
          std::shared_ptr<SendSocketInfo> findOrCreateSocket(unsigned int key, int sc = 1);
    
          // Reachability is a property of the controller, not of each universe sent
    
          // to it: one PingHost per address, shared by every output sending there,
    
          // so a controller that stops answering takes all its universes out of the
    
          // frame, and it is pinged once rather than once per universe.
    
          struct PingHost;
    
          std::map<std::string, std::shared_ptr<PingHost>> pingHosts;
    
          // Guards pingHosts, pingsStarted and pingsStopped.  Init() fills the map,
    
          // and outputs attached later (GenericUDP's, from the thread loading the
    
          // other outputs) add to and take from it while the output thread and the
    
          // network callback ping through it.  Taken after socketMutex when both are
    
          // held, and never with a PingHost's lock held.
    
          std::mutex pingHostsLock;
    
          // Init() has started the pings: outputs attached after that are pinged as
    
          // they attach.
    
          bool pingsStarted = false;
    
          // Adds a pingable output to its controller's PingHost, creating the host
    
          // if it is new (returned, so the caller can ping it).  pingHostsLock held.
    
          std::shared_ptr<PingHost> addPingTarget(UDPOutputData* o);
    
          // Takes an output back off its controller's host, and stops pinging a
    
          // controller left with no outputs.  pingHostsLock held.
    
          void removePingTarget(UDPOutputData* o);
    
          void OnPingResult(const std::shared_ptr<PingHost>& host, int ms);
    
          void SetHostValid(PingHost& host, bool valid);
    
          std::mutex socketMutex;
    
          UDPOutputMessages messages;
    
          bool enabled;
    
          std::list<UDPOutputData*> outputs;
    
          static std::mutex externalLock;
    
          static std::map<const void*, std::function<UDPOutputData*()>> externalFactories;
    
          std::map<const void*, UDPOutputData*> externalOutputs;
    
          void attachExternalOutput(const void* owner, UDPOutputData* d);
    
          static UDPOutputData* createExternalOutput(const std::function<UDPOutputData*()>& create);
    
          int networkCallbackId;
    
          void PingControllers(bool failedOnly);
    
          // Takes this output's periodic pings back out of the PingManager (their
    
          // callbacks capture this) and its "unreachable" warnings down. Close()
    
          // does it; the destructor does it for an output deleted without Close(),
    
          // as the loader deletes one whose Init() failed. Once only: after
    
          // Close(), a replacement output may hold pings for the same hosts.
    
          void StopPinging();
    
          bool pingsStopped = false;
    
          // controllers that have not answered yet; Init() waits on it
    
          std::atomic_int failedCount;
    
          std::string HexToIP(unsigned int hex);
    
          void CheckLocalDrops();
    
          // A WorkItem owns a copy of the mmsghdr list and a reference on the socket
    
          // info so a worker that outlives its frame (stuck send to a dead controller)
    
          // cannot hit freed memory when PrepData rebuilds the message vectors or
    
          // an address change drops the sockets.  Note the copy is shallow: the iovecs
    
          // still point at per-output buffers, so a late send may transmit the
    
          // *current* frame's bytes - acceptable, and no worse than it ever was.
    
          // generation identifies which frame queued it.
    
          class WorkItem {
    
          public:
    
      108214
              WorkItem(unsigned int i, std::shared_ptr<SendSocketInfo> si, const std::vector<struct mmsghdr>& m, unsigned int g) :
    
      108214
                  id(i),
    
      108214
                  socketInfo(si),
    
        1/2✓ Branch 0 taken 108214 times.
✗ Branch 1 not taken.

      108214
                  msgs(m),
    
      108214
                  generation(g) {}
    
              unsigned int id;
    
              std::shared_ptr<SendSocketInfo> socketInfo;
    
              std::vector<struct mmsghdr> msgs;
    
              unsigned int generation;
    
          };
    
          std::mutex workMutex;
    
          std::condition_variable workSignal;
    
          std::list<WorkItem> workQueue;
    
          // Keys a worker is sending to right now (under workMutex). With the keys
    
          // still queued, these are the destinations whose packet buffers a worker
    
          // may be reading, which PrepData() must not rewrite.
    
          std::multiset<unsigned int> sendingKeys;
    
          // The destination keys each output's packets went to when it was last
    
          // prepared, and the outputs PrepData() left alone this frame (both under
    
          // socketMutex).
    
          std::map<UDPOutputData*, std::vector<unsigned int>> outputKeys;
    
          std::set<UDPOutputData*> skippedOutputs;
    
          std::atomic_int doneWorkCount;
    
          std::atomic_int numWorkThreads;
    
          std::atomic_uint workGeneration;
    
          volatile bool runWorkThreads;
    
          bool useThreadedOutput;
    
          bool blockingOutput;
    
          // fq/SO_MAX_PACING_RATE based pacing; when active the kernel clocks packets
    
          // out at pacingRate per socket and the SIOCOUTQ drain waits are skipped
    
          bool pacingEnabled = false;
    
          unsigned int pacingRate = 0; // bytes per second (global default)
    
          // rate (bytes/sec) used only for the pre-sync drain-sleep estimate; the
    
          // smallest positive rate among the global rate and all per-destination
    
          // overrides, so a mixed set sleeps long enough for the slowest queue and we
    
          // never divide by a zero global rate when only overrides are paced.
    
          unsigned int pacingDrainRate = 0;
    
          // per-destination-IP pacing rate in bytes/sec, built in Init() from the
    
          // per-output pacingRate overrides.  A destination not present here uses the
    
          // global pacingRate; a value of ~0U means the controller was explicitly set
    
          // to unpaced.  The socket is shared per destination IP, so when several
    
          // outputs target the same controller with differing overrides the most
    
          // conservative (lowest) rate wins.
    
          std::map<in_addr_t, unsigned int> perDestPacingRate;
    
          // local drop diagnostics; the check runs on a short detached thread since
    
          // reading qdisc stats forks tc, which is too slow for the frame path
    
          int statCheckCounter = 0;
    
          std::atomic_bool statCheckRunning{ false };
    
          bool statBaselineValid = false;
    
          bool dropWarningActive = false;
    
          uint64_t lastSndbufErrors = 0;
    
          uint64_t lastTxDropped = 0;
    
          uint64_t lastQdiscDropped = 0;
    
      };

Line	Branch	Exec	Source
1			#pragma once
2			/*
3			* This file is part of the Falcon Player (FPP) and is Copyright (C)
4			* 2013-2022 by the Falcon Player Developers.
5			*
6			* The Falcon Player (FPP) is free software, and is covered under
7			* multiple Open Source licenses. Please see the included 'LICENSES'
8			* file for descriptions of what files are covered by each license.
9			*
10			* This source file is covered under the GPL v2 as described in the
11			* included LICENSE.GPL file.
12			*/
13
14			#include <set>
15			#include <atomic>
16			#include "fpp-json-fwd.h"
17			#include <condition_variable>
18			#include <list>
19			#include <map>
20			#include <memory>
21			#include <mutex>
22			#include <string>
23			#include <thread>
24			#include <vector>
25
26			#include "../SysSocket.h"
27			#include <netinet/in.h>
28
29			#include "ChannelOutput.h"
30
31			typedef void CURLM;
32
33			#define MULTICAST_MESSAGES_KEY 0x00000001
34			#define ANY_MESSAGES_KEY 0x00000002
35			#define ARTNET_MESSAGES_KEY 0x00000003
36
37			#define LATE_MESSAGES_START 0xFFFFFFF0
38			#define E131_SYNC_KEY 0xFFFFFFFC
39			#define ARTNET_SYNC_KEY 0xFFFFFFFD
40			#define LATE_MULTICAST_MESSAGES_KEY 0xFFFFFFFE
41			#define BROADCAST_MESSAGES_KEY 0xFFFFFFFF
42
43			class SendSocketInfo;
44
45			class UDPOutputMessages {
46			public:
47			UDPOutputMessages();
48			~UDPOutputMessages();
49
50			void ForceSocket(unsigned int key, int socket, bool preventClose = false);
51			int GetSocket(unsigned int key);
52
53			std::vector<struct mmsghdr>& GetMessages(unsigned int key);
54		929119	std::vector<struct mmsghdr>& operator[](unsigned int key) { return GetMessages(key); }
55
56			private:
57			std::map<unsigned int, std::vector<struct mmsghdr>> messages;
58			std::map<unsigned int, std::shared_ptr<SendSocketInfo>> sendSockets;
59			// While set, every key an output asks for is noted here (once), so
60			// UDPOutput knows which destinations each output's packets go to.
61			std::vector<unsigned int>* recordKeys = nullptr;
62
63			void clearMessages();
64			void clearSockets();
65
66			friend class UDPOutput;
67			};
68
69			class UDPOutputData {
70			public:
71			UDPOutputData(const Json::Value& config);
72			virtual ~UDPOutputData();
73
74			virtual bool IsPingable() = 0;
75		941	virtual bool Monitor() const { return monitor; }
76			virtual void PrepareData(unsigned char* channelData, UDPOutputMessages& msgs) = 0;
77		95410	virtual void PostPrepareData(unsigned char* channelData, UDPOutputMessages& msgs) {}
78
79			virtual void DumpConfig() = 0;
80
81			// min is the first channel, 0-based; max is one past the last (every
82			// protocol works it out as startChannel + count - 1 from the 1-based
83			// startChannel). UDPOutput converts it for ChannelOutput's inclusive max.
84		1550	virtual void GetRequiredChannelRange(int& min, int& max) {
85		1550	min = startChannel - 1;
86		1550	max = startChannel + channelCount - 1;
87		1550	}
88
89		978	virtual void StartingOutput() {}
90		978	virtual void StoppingOutput() {}
91
92			virtual const std::string& GetOutputTypeString() const;
93
94			// The messages last prepared are being sent. A frame can be prepared
95			// more than once before it goes out (its data changed in between), and
96			// only the last preparation is sent, so a protocol sequence number
97			// assigned in PrepareData() is committed here: advanced as it was
98			// assigned, the discarded preparation skipped one on the wire.
99		8452	virtual void FrameSent() {}
100
101			static in_addr_t toInetAddr(const std::string& ip, bool& valid);
102		942	static in_addr_t toInetAddr(const std::string& ip, std::atomic<bool>& valid) {
103		942	bool v = false;
104	1/2 ✓ Branch 0 taken 942 times. ✗ Branch 1 not taken.	942	in_addr_t a = toInetAddr(ip, v);
105		942	valid = v;
106		942	return a;
107			}
108
109			std::string description;
110			bool active;
111			int startChannel;
112			int channelCount;
113			int type;
114			std::string ipAddress;
115			// Flipped by the ping thread (SetHostValid) while the output thread reads
116			// it in PrepData().
117			std::atomic<bool> valid;
118			bool monitor;
119
120			// per-destination pacing override in Mbps; -1 means "use the global rate".
121			// 0 means this controller is explicitly unpaced (line rate). A mix of a
122			// gigabit FPP instance and a 30Mbps-class ESP32 controller can each get an
123			// appropriate cap instead of one global setting for all of them.
124			int pacingRateMbps = -1;
125
126			// Set while StartingOutput() is owed its StoppingOutput(). valid can change
127			// at runtime (a failed ping clears it), so it cannot say which outputs were
128			// started; UDPOutput pairs the two calls through this instead.
129			bool started = false;
130
131			UDPOutputData(UDPOutputData const&) = delete;
132			void operator=(UDPOutputData const& x) = delete;
133
134			protected:
135			void SaveFrame(unsigned char* channelData, int len);
136			bool NeedToOutputFrame(unsigned char* channelData, int startChannel, int savedIdx, int count);
137			bool deDuplicate = false;
138			int skippedFrames;
139			unsigned char* lastData;
140			};
141
142			class UDPOutput : public ChannelOutput {
143			public:
144			UDPOutput(unsigned int startChannel, unsigned int channelCount);
145			virtual ~UDPOutput();
146
147		1070	virtual std::string GetOutputType() const override {
148	1/2 ✓ Branch 0 taken 1070 times. ✗ Branch 1 not taken.	2140	return "UDPOutput";
149			}
150			virtual int Init(Json::Value config) override;
151			virtual int Close(void) override;
152
153			virtual void PrepData(unsigned char* channelData) override;
154			virtual int SendData(unsigned char* channelData) override;
155
156			virtual void DumpConfig(void) override;
157
158			virtual void GetRequiredChannelRanges(const std::function<void(int, int)>& addRange) override;
159
160			void addOutput(UDPOutputData*);
161
162			// Outputs another channel output plugin (GenericUDP) sends through this
163			// one. Registered by owner with a factory rather than as data: this
164			// instance is replaced whenever co-universes.json reloads, and the new one
165			// has to rebuild them. Unregistering only retires the data - a worker may
166			// still be sending from its buffers - and it is freed with this instance.
167			static void RegisterExternalOutput(const void* owner, std::function<UDPOutputData*()> create);
168			static void UnregisterExternalOutput(const void* owner);
169
170			int createSocket(int port = 0, bool broadCast = false, bool multiCast = false);
171
172			static UDPOutput* INSTANCE;
173
174			void BackgroundOutputWork();
175
176			virtual void StartingOutput() override;
177			virtual void StoppingOutput() override;
178
179			private:
180			int SendMessages(unsigned int key, SendSocketInfo* socketInfo, std::vector<struct mmsghdr>& sendmsgs);
181			struct sockaddr_in localAddress;
182			std::string outInterface;
183			bool needsBroadcast = false;
184			bool hasArtNet = false;
185			// between StartingOutput() and StoppingOutput(); output thread only
186			bool outputRunning = false;
187			// tags this output's CurlManager requests so the destructor can cancel them
188			const std::string curlOwner = "UDPOutput-" + std::to_string((uintptr_t)this);
189			bool interfaceUp;
190			// the "Invalid interface" banner InitNetwork() raised, while it is up;
191			// InitNetwork() takes it down when the address arrives, Close() otherwise
192			std::string interfaceWarning;
193
194			// rebuildSockets: the address changed; drop the sockets made on the old one
195			bool InitNetwork(bool rebuildSockets = false);
196			std::shared_ptr<SendSocketInfo> findOrCreateSocket(unsigned int key, int sc = 1);
197
198			// Reachability is a property of the controller, not of each universe sent
199			// to it: one PingHost per address, shared by every output sending there,
200			// so a controller that stops answering takes all its universes out of the
201			// frame, and it is pinged once rather than once per universe.
202			struct PingHost;
203			std::map<std::string, std::shared_ptr<PingHost>> pingHosts;
204			// Guards pingHosts, pingsStarted and pingsStopped. Init() fills the map,
205			// and outputs attached later (GenericUDP's, from the thread loading the
206			// other outputs) add to and take from it while the output thread and the
207			// network callback ping through it. Taken after socketMutex when both are
208			// held, and never with a PingHost's lock held.
209			std::mutex pingHostsLock;
210			// Init() has started the pings: outputs attached after that are pinged as
211			// they attach.
212			bool pingsStarted = false;
213			// Adds a pingable output to its controller's PingHost, creating the host
214			// if it is new (returned, so the caller can ping it). pingHostsLock held.
215			std::shared_ptr<PingHost> addPingTarget(UDPOutputData* o);
216			// Takes an output back off its controller's host, and stops pinging a
217			// controller left with no outputs. pingHostsLock held.
218			void removePingTarget(UDPOutputData* o);
219			void OnPingResult(const std::shared_ptr<PingHost>& host, int ms);
220			void SetHostValid(PingHost& host, bool valid);
221
222			std::mutex socketMutex;
223			UDPOutputMessages messages;
224			bool enabled;
225
226			std::list<UDPOutputData*> outputs;
227
228			static std::mutex externalLock;
229			static std::map<const void, std::function<UDPOutputData()>> externalFactories;
230			std::map<const void, UDPOutputData> externalOutputs;
231			void attachExternalOutput(const void* owner, UDPOutputData* d);
232			static UDPOutputData* createExternalOutput(const std::function<UDPOutputData*()>& create);
233
234			int networkCallbackId;
235
236			void PingControllers(bool failedOnly);
237			// Takes this output's periodic pings back out of the PingManager (their
238			// callbacks capture this) and its "unreachable" warnings down. Close()
239			// does it; the destructor does it for an output deleted without Close(),
240			// as the loader deletes one whose Init() failed. Once only: after
241			// Close(), a replacement output may hold pings for the same hosts.
242			void StopPinging();
243			bool pingsStopped = false;
244			// controllers that have not answered yet; Init() waits on it
245			std::atomic_int failedCount;
246			std::string HexToIP(unsigned int hex);
247
248			void CheckLocalDrops();
249
250			// A WorkItem owns a copy of the mmsghdr list and a reference on the socket
251			// info so a worker that outlives its frame (stuck send to a dead controller)
252			// cannot hit freed memory when PrepData rebuilds the message vectors or
253			// an address change drops the sockets. Note the copy is shallow: the iovecs
254			// still point at per-output buffers, so a late send may transmit the
255			// current frame's bytes - acceptable, and no worse than it ever was.
256			// generation identifies which frame queued it.
257			class WorkItem {
258			public:
259		108214	WorkItem(unsigned int i, std::shared_ptr<SendSocketInfo> si, const std::vector<struct mmsghdr>& m, unsigned int g) :
260		108214	id(i),
261		108214	socketInfo(si),
262	1/2 ✓ Branch 0 taken 108214 times. ✗ Branch 1 not taken.	108214	msgs(m),
263		108214	generation(g) {}
264			unsigned int id;
265			std::shared_ptr<SendSocketInfo> socketInfo;
266			std::vector<struct mmsghdr> msgs;
267			unsigned int generation;
268			};
269
270			std::mutex workMutex;
271			std::condition_variable workSignal;
272			std::list<WorkItem> workQueue;
273			// Keys a worker is sending to right now (under workMutex). With the keys
274			// still queued, these are the destinations whose packet buffers a worker
275			// may be reading, which PrepData() must not rewrite.
276			std::multiset<unsigned int> sendingKeys;
277			// The destination keys each output's packets went to when it was last
278			// prepared, and the outputs PrepData() left alone this frame (both under
279			// socketMutex).
280			std::map<UDPOutputData*, std::vector<unsigned int>> outputKeys;
281			std::set<UDPOutputData*> skippedOutputs;
282			std::atomic_int doneWorkCount;
283			std::atomic_int numWorkThreads;
284			std::atomic_uint workGeneration;
285			volatile bool runWorkThreads;
286			bool useThreadedOutput;
287			bool blockingOutput;
288
289			// fq/SO_MAX_PACING_RATE based pacing; when active the kernel clocks packets
290			// out at pacingRate per socket and the SIOCOUTQ drain waits are skipped
291			bool pacingEnabled = false;
292			unsigned int pacingRate = 0; // bytes per second (global default)
293
294			// rate (bytes/sec) used only for the pre-sync drain-sleep estimate; the
295			// smallest positive rate among the global rate and all per-destination
296			// overrides, so a mixed set sleeps long enough for the slowest queue and we
297			// never divide by a zero global rate when only overrides are paced.
298			unsigned int pacingDrainRate = 0;
299
300			// per-destination-IP pacing rate in bytes/sec, built in Init() from the
301			// per-output pacingRate overrides. A destination not present here uses the
302			// global pacingRate; a value of ~0U means the controller was explicitly set
303			// to unpaced. The socket is shared per destination IP, so when several
304			// outputs target the same controller with differing overrides the most
305			// conservative (lowest) rate wins.
306			std::map<in_addr_t, unsigned int> perDestPacingRate;
307
308			// local drop diagnostics; the check runs on a short detached thread since
309			// reading qdisc stats forks tc, which is too slow for the frame path
310			int statCheckCounter = 0;
311			std::atomic_bool statCheckRunning{ false };
312			bool statBaselineValid = false;
313			bool dropWarningActive = false;
314			uint64_t lastSndbufErrors = 0;
315			uint64_t lastTxDropped = 0;
316			uint64_t lastQdiscDropped = 0;
317			};
318

Function (Line)	Call count	Block coverage
UDPOutput::GetOutputType[abi:cxx11]() const (line 147)	called 1070 times, returned 1070 times	468.0%
UDPOutput::WorkItem::WorkItem(unsigned int, std::shared_ptr<SendSocketInfo>, std::vector<mmsghdr, std::allocator<mmsghdr> > const&, unsigned int) (line 259)	called 108214 times, returned 108214 times	480.0%
UDPOutputData::FrameSent() (line 99)	called 8452 times, returned 8452 times	600.0%
UDPOutputData::GetRequiredChannelRange(int&, int&) (line 84)	called 1550 times, returned 1550 times	600.0%
UDPOutputData::Monitor() const (line 75)	called 941 times, returned 941 times	600.0%
UDPOutputData::PostPrepareData(unsigned char*, UDPOutputMessages&) (line 77)	called 95410 times, returned 95410 times	600.0%
UDPOutputData::StartingOutput() (line 89)	called 978 times, returned 978 times	600.0%
UDPOutputData::StoppingOutput() (line 90)	called 978 times, returned 978 times	600.0%
UDPOutputData::toInetAddr(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&, std::atomic<bool>&) (line 102)	called 942 times, returned 942 times	600.0%
UDPOutputMessages::operator[](unsigned int) (line 54)	called 929119 times, returned 929119 times	600.0%