GCC Code Coverage Report

Directory: ./
File: channeloutput/ThreadedChannelOutput.cpp
Date: 2026-10-29 08:11:58
Exec Total Coverage
Lines: 125 152 82.2%
Functions: 13 13 100.0%
Branches: 81 140 57.9%
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 LGPL v2.1 as described in the
10 * included LICENSE.LGPL file.
11 */
12
13 #include "fpp-pch.h"
14
15 #include "fpp-json.h"
16
17 #include <cstring>
18 #include <errno.h>
19 #include <unistd.h>
20
21 #include "../commands/Commands.h"
22 #include "../common.h"
23 #include "../log.h"
24
25 #include "ThreadedChannelOutput.h"
26
27 75 ThreadedChannelOutput::ThreadedChannelOutput(unsigned int startChannel,
28 75 unsigned int channelCount) :
29 ChannelOutput(startChannel, channelCount),
30 75 m_threadIsRunning(0),
31 75 m_runThread(0),
32 75 m_dataWaiting(0),
33 75 m_useDoubleBuffer(0),
34 75 m_maxWait(0),
35 75 m_inBuf(NULL),
36 75 m_outBuf(NULL) {
37 // std::thread/std::mutex/std::condition_variable are default
38 // constructed above (member declaration order in the header), no
39 // explicit init needed the way pthread_mutex_init()/pthread_cond_init()
40 // were required.
41 75 }
42
43 150 ThreadedChannelOutput::~ThreadedChannelOutput() {
44 // Unlike pthread_t, std::thread's destructor calls std::terminate()
45 // if the thread is still joinable (i.e. StopOutputThread() was never
46 // called, e.g. a subclass that skips Close()). The pthread version had
47 // no such requirement. This is a safety net only; normal shutdown
48 // already goes through Close() -> StopOutputThread() -> join().
49
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 75 times.
 150 if (m_thread.joinable()) {
50 ✗ StopOutputThread();
51 }
52 150 }
53
54 60 int ThreadedChannelOutput::Init(void) {
55 60 LogDebug(VB_CHANNELOUT, "ThreadedChannelOutput::Init()\n");
56
57
2/2
✓ Branch 0 taken 34 times.
✓ Branch 1 taken 26 times.
 60 if (m_useDoubleBuffer) {
58 34 m_inBuf = new unsigned char[m_channelCount];
59 34 m_outBuf = new unsigned char[m_channelCount];
60 }
61
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 60 times.
 60 if (!StartOutputThread()) {
62 // Nothing can consume the buffers without a worker and the caller
63 // discards an output whose Init() failed, so release them here.
64 // Close() may still run on the way out; nulling keeps its delete[]
65 // a no-op rather than a double free.
66 ✗ if (m_useDoubleBuffer) {
67 ✗ delete[] m_inBuf;
68 ✗ delete[] m_outBuf;
69 ✗ m_inBuf = nullptr;
70 ✗ m_outBuf = nullptr;
71 }
72 ✗ return 0;
73 }
74 60 DumpConfig();
75
76 60 return 1;
77 }
78
79 60 int ThreadedChannelOutput::Init(Json::Value config) {
80
6/14
✓ Branch 0 taken 60 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 60 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 60 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 60 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 60 times.
✗ Branch 9 not taken.
✓ Branch 10 taken 60 times.
✗ Branch 11 not taken.
✗ Branch 12 not taken.
✗ Branch 13 not taken.
 60 return ChannelOutput::Init(config) && Init();
81 }
82
83 60 int ThreadedChannelOutput::Close(void) {
84 60 LogDebug(VB_CHANNELOUT, "ThreadedChannelOutput::Close()\n");
85
86 60 StopOutputThread();
87
88
2/2
✓ Branch 0 taken 34 times.
✓ Branch 1 taken 26 times.
 60 if (m_useDoubleBuffer) {
89
1/2
✓ Branch 0 taken 34 times.
✗ Branch 1 not taken.
 34 delete[] m_inBuf;
90
1/2
✓ Branch 0 taken 34 times.
✗ Branch 1 not taken.
 34 delete[] m_outBuf;
91 }
92
93 60 return ChannelOutput::Close();
94 }
95
96 2398 int ThreadedChannelOutput::SendData(unsigned char* channelData) {
97 2398 LogExcess(VB_CHANNELOUT, "ThreadedChannelOutput::SendData(%p)\n", channelData);
98
99
2/2
✓ Branch 0 taken 1731 times.
✓ Branch 1 taken 667 times.
 2398 if (m_useDoubleBuffer) {
100
1/2
✓ Branch 0 taken 1731 times.
✗ Branch 1 not taken.
 1731 std::lock_guard<std::mutex> lock(m_bufLock);
101 1731 memcpy(m_inBuf, channelData, m_channelCount);
102 1731 }
103
104 // The flag has to be raised under m_sendLock: a store outside it can land
105 // between the worker's predicate check and its wait(), and the notify that
106 // follows is then delivered to nobody. The two locks are taken in
107 // sequence, never nested, so no ordering against the worker exists to
108 // violate. Without a double buffer the frame is handed over by pointer,
109 // and m_outBuf is written here and read by the worker for the frame it is
110 // still sending: it changes only under m_sendLock, and the worker copies
111 // it out under the same lock when it claims a frame.
112 {
113
1/2
✓ Branch 0 taken 2398 times.
✗ Branch 1 not taken.
 2398 std::lock_guard<std::mutex> lock(m_sendLock);
114
2/2
✓ Branch 0 taken 667 times.
✓ Branch 1 taken 1731 times.
 2398 if (!m_useDoubleBuffer) {
115 667 m_outBuf = channelData;
116 }
117 2398 m_dataWaiting = 1;
118 2398 }
119
120 2398 m_sendCond.notify_one();
121 2398 return 0;
122 }
123
124 2166 int ThreadedChannelOutput::SendOutputBuffer(void) {
125 2166 LogExcess(VB_CHANNELOUT, "ChannelOutput::SendOutputBuffer()\n");
126
127 // Claim the pending frame BEFORE copying it out. Clearing after the copy
128 // opens a window where a SendData landing between the copy and the clear
129 // has its flag wiped and its frame is never sent - fatal for a final
130 // blanking frame, which has no successor. Claim-first means a frame
131 // arriving mid-send leaves the flag raised and the newer data goes out on
132 // the worker's next pass (at worst the newest frame is sent twice).
133 unsigned char* outBuf;
134 {
135
1/2
✓ Branch 0 taken 2166 times.
✗ Branch 1 not taken.
 2166 std::lock_guard<std::mutex> lock(m_sendLock);
136 2166 m_dataWaiting = 0;
137 2166 outBuf = m_outBuf; // see SendData(): SendData() may replace it next
138 2166 }
139
140
2/2
✓ Branch 0 taken 1505 times.
✓ Branch 1 taken 661 times.
 2166 if (m_useDoubleBuffer) {
141
1/2
✓ Branch 0 taken 1505 times.
✗ Branch 1 not taken.
 1505 std::lock_guard<std::mutex> lock(m_bufLock);
142 1505 memcpy(outBuf, m_inBuf, m_channelCount);
143 1505 }
144
145 2166 RawSendData(outBuf);
146 2166 return m_channelCount;
147 }
148
149 114 void ThreadedChannelOutput::DumpConfig(void) {
150 114 ChannelOutput::DumpConfig();
151 114 LogDebug(VB_CHANNELOUT, " Thread Running : %u\n", m_threadIsRunning);
152 114 LogDebug(VB_CHANNELOUT, " Run Thread : %u\n", m_runThread);
153 114 LogDebug(VB_CHANNELOUT, " Data Waiting : %u\n", m_dataWaiting);
154 114 }
155
156 60 bool ThreadedChannelOutput::StartOutputThread(void) {
157 60 LogDebug(VB_CHANNELOUT, "ThreadedChannelOutput::StartOutputThread()\n");
158
159 60 m_runThread = 1;
160
161 // std::thread has no "not a member of the class" trampoline
162 // requirement like pthread_create() did (no void*(*)(void*) function
163 // pointer signature to satisfy), so OutputThread() can be invoked
164 // directly via a lambda. The lambda replicates the trampoline's
165 // original debug log line so log output is unchanged.
166 //
167 // Unlike pthread_create(), which reports failure via its int return
168 // value, std::thread's constructor reports failure by throwing
169 // std::system_error. Catch it here so the same error-code -> message
170 // mapping and the same "leave m_threadID/m_thread unset" behavior on
171 // failure are preserved.
172 try {
173 60 m_thread = std::thread([this]() {
174 60 LogDebug(VB_CHANNELOUT, "ThreadedChannelOutput::RunChannelOutputThread()\n");
175 60 OutputThread();
176
1/2
✓ Branch 0 taken 60 times.
✗ Branch 1 not taken.
 60 });
177 ✗ } catch (const std::system_error& ex) {
178 char msg[256];
179
180 ✗ m_runThread = 0;
181 ✗ switch (ex.code().value()) {
182 ✗ case EAGAIN:
183 ✗ strcpy(msg, "Insufficient Resources");
184 ✗ break;
185 ✗ case EINVAL:
186 ✗ strcpy(msg, "Invalid settings");
187 ✗ break;
188 ✗ case EPERM:
189 ✗ strcpy(msg, "Invalid Permissions");
190 ✗ break;
191 ✗ default:
192 // Any other code (std::thread isn't limited to the three
193 // pthread_create errno values) — report it rather than reading
194 // msg uninitialized, which the old pthread version did.
195 ✗ snprintf(msg, sizeof(msg), "%s (code %d)", ex.what(), ex.code().value());
196 ✗ break;
197 }
198 ✗ LogErr(VB_CHANNELOUT, "ERROR creating ChannelOutput thread: %s\n", msg);
199
200 // Only OutputThread() ever raises m_threadIsRunning, so falling into
201 // the wait below would spin forever.
202 ✗ return false;
203 ✗ }
204
205 // m_runThread can only be 0 here if construction failed, which returns
206 // above; the check keeps the wait bounded if that ever stops holding.
207 // Both are read under m_sendLock, which the worker holds to set them.
208 while (true) {
209 {
210
1/2
✓ Branch 0 taken 120 times.
✗ Branch 1 not taken.
 120 std::lock_guard<std::mutex> l(m_sendLock);
211
3/4
✗ Branch 0 not taken.
✓ Branch 1 taken 120 times.
✓ Branch 2 taken 60 times.
✓ Branch 3 taken 60 times.
 120 if (!m_runThread || m_threadIsRunning) {
212 break;
213 }
214
2/2
✓ Branch 0 taken 60 times.
✓ Branch 1 taken 60 times.
 120 }
215 60 usleep(10000);
216 60 }
217
218 60 return true;
219 }
220
221 124 int ThreadedChannelOutput::StopOutputThread(void) {
222 124 LogDebug(VB_CHANNELOUT, "ChannelOutput::StopOutputThread()\n");
223
224
2/2
✓ Branch 0 taken 64 times.
✓ Branch 1 taken 60 times.
 124 if (!m_thread.joinable())
225 64 return -1;
226
227 // Half of the worker's wait predicate, so it must change under m_sendLock;
228 // a store outside the lock can be missed by a worker that is between its
229 // predicate check and its wait(), leaving it parked until the next frame
230 // that will never come.
231 {
232
1/2
✓ Branch 0 taken 60 times.
✗ Branch 1 not taken.
 60 std::lock_guard<std::mutex> lock(m_sendLock);
233 60 m_runThread = 0;
234 60 }
235
236 60 m_sendCond.notify_one();
237
238 // Wait up to 110ms for a pending frame to be sent. The worker flushes
239 // m_dataWaiting before it honors the stop, so this is what lets the final
240 // blanking frame of a sequence reach the hardware.
241 60 int loops = 0;
242
2/2
✓ Branch 0 taken 100 times.
✓ Branch 1 taken 1 times.
 101 while (loops++ < 11) {
243 {
244
1/2
✓ Branch 0 taken 100 times.
✗ Branch 1 not taken.
 100 std::lock_guard<std::mutex> lock(m_sendLock);
245
5/6
✓ Branch 0 taken 41 times.
✓ Branch 1 taken 59 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 41 times.
✓ Branch 4 taken 59 times.
✓ Branch 5 taken 41 times.
 100 if (!m_dataWaiting || !m_threadIsRunning)
246 59 break;
247
2/2
✓ Branch 0 taken 41 times.
✓ Branch 1 taken 59 times.
 100 }
248 41 usleep(10000);
249 }
250
251
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 60 times.
 60 if (!m_thread.joinable())
252 ✗ return -1;
253
254 // m_bufLock must NOT be held across the join(): the worker takes it while
255 // flushing that pending frame on its way out. It guards only the contents
256 // of m_inBuf/m_outBuf, which join() does not touch, and Close() does not
257 // free them until after this returns.
258 60 m_thread.join();
259
260 60 return 0;
261 }
262
263 60 void ThreadedChannelOutput::OutputThread(void) {
264
1/2
✓ Branch 0 taken 60 times.
✗ Branch 1 not taken.
 60 LogDebug(VB_CHANNELOUT, "ThreadedChannelOutput::OutputThread()\n");
265
2/4
✓ Branch 0 taken 60 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 60 times.
✗ Branch 3 not taken.
 60 SetThreadName("FPP-" + m_outputType);
266
267
1/2
✓ Branch 0 taken 60 times.
✗ Branch 1 not taken.
 60 long long wakeTime = GetTime();
268
269 // Deferred: m_sendLock is held only while the predicate state is read or
270 // written, never across SendOutputBuffer()/WaitTimedOut(), which can block
271 // on hardware for as long as they like.
272 60 std::unique_lock<std::mutex> sendLock(m_sendLock, std::defer_lock);
273
274
1/2
✓ Branch 0 taken 60 times.
✗ Branch 1 not taken.
 60 sendLock.lock();
275 60 m_threadIsRunning = 1;
276
1/2
✓ Branch 0 taken 60 times.
✗ Branch 1 not taken.
 60 sendLock.unlock();
277
1/2
✓ Branch 0 taken 60 times.
✗ Branch 1 not taken.
 60 LogDebug(VB_CHANNELOUT, "ThreadedChannelOutput thread started\n");
278
279 // m_dataWaiting and m_runThread are both owned by m_sendLock, so this
280 // closes the window SendData's store-then-notify used to leave open.
281
4/4
✓ Branch 0 taken 2134 times.
✓ Branch 1 taken 2166 times.
✓ Branch 2 taken 52 times.
✓ Branch 3 taken 2082 times.
 4360 auto ready = [this]() { return m_dataWaiting || !m_runThread; };
282
283 while (true) {
284 // Wait for more data
285
1/2
✓ Branch 0 taken 2218 times.
✗ Branch 1 not taken.
 2218 sendLock.lock();
286
1/2
✓ Branch 0 taken 2218 times.
✗ Branch 1 not taken.
 2218 long long nowTime = GetTime();
287
1/2
✓ Branch 0 taken 2218 times.
✗ Branch 1 not taken.
 2218 LogExcess(VB_CHANNELOUT, "ThreadedChannelOutput thread: sent: %lld, elapsed: %lld\n",
288 nowTime, nowTime - wakeTime);
289
290
6/6
✓ Branch 0 taken 2083 times.
✓ Branch 1 taken 135 times.
✓ Branch 2 taken 48 times.
✓ Branch 3 taken 2035 times.
✓ Branch 4 taken 183 times.
✓ Branch 5 taken 2035 times.
 2218 if (m_dataWaiting || m_maxWait) {
291 // Old code computed an absolute CLOCK_REALTIME deadline
292 // (now + duration) via gettimeofday()/timespec math for
293 // pthread_cond_timedwait(). That duration was:
294 // - m_maxWait milliseconds, when m_maxWait is set
295 // - 200ms otherwise ((tv_usec + 200000us) rolled to nsec)
296 // wait_for() takes that same duration directly, so the
297 // timespec/overflow-carry arithmetic is no longer needed.
298
2/2
✓ Branch 0 taken 48 times.
✓ Branch 1 taken 135 times.
 183 std::chrono::milliseconds waitDuration(m_maxWait ? m_maxWait : 200);
299
300 // The predicate overload computes the deadline once, so spurious
301 // wakeups do not extend it and m_maxWait keeps its meaning: the
302 // interval at which a subclass wants WaitTimedOut() called.
303
1/2
✓ Branch 0 taken 183 times.
✗ Branch 1 not taken.
 183 m_sendCond.wait_for(sendLock, waitDuration, ready);
304 } else {
305
1/2
✓ Branch 0 taken 2035 times.
✗ Branch 1 not taken.
 2035 m_sendCond.wait(sendLock, ready);
306 }
307
308 // Sampled together under the lock so the two decisions below cannot
309 // see a half-updated state.
310 2218 bool haveData = m_dataWaiting;
311 2218 bool keepRunning = m_runThread;
312
313
1/2
✓ Branch 0 taken 2218 times.
✗ Branch 1 not taken.
 2218 sendLock.unlock();
314
315
1/2
✓ Branch 0 taken 2218 times.
✗ Branch 1 not taken.
 2218 wakeTime = GetTime();
316
1/2
✓ Branch 0 taken 2218 times.
✗ Branch 1 not taken.
 2218 LogExcess(VB_CHANNELOUT, "ThreadedChannelOutput thread: woke: %lld\n", wakeTime);
317
318 // A pending frame is flushed even once a stop has been requested: the
319 // last frame of a sequence is the blanking frame and StopOutputThread
320 // follows it immediately, so honoring the stop first leaves the lights
321 // lit. Reaching here with neither data nor a stop means the wait
322 // timed out, which is the only thing WaitTimedOut() is for.
323
2/2
✓ Branch 0 taken 2166 times.
✓ Branch 1 taken 52 times.
 2218 if (haveData) {
324
1/2
✓ Branch 0 taken 2166 times.
✗ Branch 1 not taken.
 2166 SendOutputBuffer();
325
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 52 times.
 52 } else if (keepRunning) {
326 ✗ WaitTimedOut();
327 }
328
329
2/2
✓ Branch 0 taken 60 times.
✓ Branch 1 taken 2158 times.
 2218 if (!keepRunning)
330 60 break;
331 2158 }
332
333
1/2
✓ Branch 0 taken 60 times.
✗ Branch 1 not taken.
 60 LogDebug(VB_CHANNELOUT, "ThreadedChannelOutput thread complete\n");
334
1/2
✓ Branch 0 taken 60 times.
✗ Branch 1 not taken.
 60 sendLock.lock();
335 60 m_threadIsRunning = 0;
336
1/2
✓ Branch 0 taken 60 times.
✗ Branch 1 not taken.
 60 sendLock.unlock();
337 60 }
338