From ef1829b179c8779c11b7c67e353355feddd4cedf Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 5 Jul 2026 07:53:52 +0000 Subject: [PATCH 1/3] Add NEON GF(256) kernels for FEC encode/decode on ARM targets The FEC codec's only optimized path was 32-bit x86 assembly, so all real targets (Pi, Radxa, OpenIPC) ran the scalar byte-at-a-time table lookup loops. Add NEON implementations of addmul1/mul1 using nibble-split vtbl table lookups, processing 16 bytes per iteration, for both aarch64 and armv7 NEON. The NEON path is enabled only after a one-time self test in fec_init verifies it matches the scalar reference exactly; platforms without NEON (e.g. Pi Zero W, ARMv6) keep the scalar path. Add test_fec (make test_fec) that encodes, erases and decodes blocks and verifies recovery; verified on x86 (scalar), aarch64 (NEON), armv7+NEON and armv6 (scalar) via qemu-user. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01TWFTzQ3aWucyVgKpyf6HcT --- Makefile | 7 +- code/r_tests/test_fec.c | 79 +++++++++++++++++++++ code/radio/fec.c | 153 +++++++++++++++++++++++++++++++++++++++- 3 files changed, 235 insertions(+), 4 deletions(-) create mode 100644 code/r_tests/test_fec.c diff --git a/Makefile b/Makefile index 0f3b1191..2ccdc28f 100644 --- a/Makefile +++ b/Makefile @@ -303,11 +303,14 @@ ruby_player_radxa:code/r_player/ruby_player_radxa.o code/r_player/mpp_core.o $(F $(CXX) $(_CPPFLAGS) $(CFLAGS_RENDERER) -o $@ $^ $(_LDFLAGS) $(LDFLAGS_RENDERER) $(LDFLAGS_CENTRAL) $(LDFLAGS_CENTRAL2) -ldl -lc -lrockchip_mpp ifeq ($(RUBY_BUILD_ENV),radxa) -tests: test_port_rx test_port_tx test_link +tests: test_port_rx test_port_tx test_link test_fec else -tests: test_gpio test_port_rx test_port_tx test_link +tests: test_gpio test_port_rx test_port_tx test_link test_fec endif +test_fec:$(FOLDER_TESTS)/test_fec.o $(FOLDER_RADIO)/fec.o + $(CC) $(_CFLAGS) -o $@ $^ + test_cairo:$(FOLDER_TESTS)/test_cairo.o $(MODULE_BASE) $(MODULE_BASE2) $(MODULE_COMMON) $(MODULE_RADIO) $(MODULE_MODELS) $(CXX) $(_CPPFLAGS) -o $@ $^ $(_LDFLAGS) -ldl -lc diff --git a/code/r_tests/test_fec.c b/code/r_tests/test_fec.c new file mode 100644 index 00000000..46ace9b8 --- /dev/null +++ b/code/r_tests/test_fec.c @@ -0,0 +1,79 @@ +// Functional test for code/radio/fec.c: encode, erase, decode, verify. +// On ARM builds this also exercises the NEON GF kernels (enabled by the +// self test in fec_init) against the scalar reference implementation. +#include +#include +#include +#include "../radio/fec.h" + +#define BLOCK_SIZE 1109 // odd size to exercise vector tails +#define K 8 +#define M 4 + +static unsigned int s_uSeed = 0xC0FFEE42; +static unsigned char rnd_byte(void) +{ + s_uSeed = s_uSeed * 1103515245 + 12345; + return (unsigned char)(s_uSeed >> 16); +} + +int main(void) +{ + unsigned char* pData[K]; + unsigned char* pOrig[K]; + unsigned char* pFec[M]; + int i, j; + + fec_init(); + + for( i=0; i %s\n", + iErasures, iRes, iFailures, (iFailures==0 && iRes==0) ? "PASS" : "FAIL"); + if ( iFailures || iRes ) + iTotalFailures++; + for( i=0; i + +static int s_iFecNeonEnabled = 0; + +static inline uint8x16_t gf_neon_tbl16(uint8x16_t vTable, uint8x16_t vIndex) +{ +#if defined(__aarch64__) + return vqtbl1q_u8(vTable, vIndex); +#else + /* armv7 NEON has no single-instruction 16-byte table lookup */ + uint8x8x2_t t; + t.val[0] = vget_low_u8(vTable); + t.val[1] = vget_high_u8(vTable); + return vcombine_u8(vtbl2_u8(t, vget_low_u8(vIndex)), vtbl2_u8(t, vget_high_u8(vIndex))); +#endif +} + +static inline void gf_neon_build_nibble_tables(gf c, uint8x16_t* pvTableLow, uint8x16_t* pvTableHigh) +{ + gf* pMulRow = &gf_mul_table[((int)c)<<8]; + unsigned char uTableLow[16], uTableHigh[16]; + int i; + for( i=0; i<16; i++ ) + { + uTableLow[i] = pMulRow[i]; + uTableHigh[i] = pMulRow[i<<4]; + } + *pvTableLow = vld1q_u8(uTableLow); + *pvTableHigh = vld1q_u8(uTableHigh); +} + +static void +neon_addmul1(gf *dst1, gf *src1, gf c, int sz) +{ + uint8x16_t vTableLow, vTableHigh; + uint8x16_t vNibbleMask = vdupq_n_u8(0x0F); + gf *dst = dst1, *src = src1; + gf *pMulRow = &gf_mul_table[((int)c)<<8]; + int i; + int iBlocks = sz >> 4; + + gf_neon_build_nibble_tables(c, &vTableLow, &vTableHigh); + + for( i=0; i> 4; + + gf_neon_build_nibble_tables(c, &vTableLow, &vTableHigh); + + for( i=0; i> 16); + uSeed = uSeed * 1103515245 + 12345; + bufScalar[i] = bufNeon[i] = (gf)(uSeed >> 16); + } + slow_addmul1(bufScalar, bufSrc, c, sizeof(bufSrc)); + neon_addmul1(bufNeon, bufSrc, c, sizeof(bufSrc)); + if ( 0 != memcmp(bufScalar, bufNeon, sizeof(bufScalar)) ) + s_iFecNeonEnabled = 0; + + slow_mul1(bufScalar, bufSrc, c, sizeof(bufSrc)); + neon_mul1(bufNeon, bufSrc, c, sizeof(bufSrc)); + if ( 0 != memcmp(bufScalar, bufNeon, sizeof(bufScalar)) ) + s_iFecNeonEnabled = 0; + } + if ( 0 == s_iFecNeonEnabled ) + fprintf(stderr, "fec: NEON GF kernels failed self test, using scalar path\n"); +} +#endif /* FEC_HAVE_NEON */ + void fec_init(void) { TICK(ticks[0]); @@ -706,6 +852,9 @@ void fec_init(void) init_mul_table(); TOCK(ticks[0]); DDB(fprintf(stderr, "init_mul_table took %ldus\n", ticks[0]);) +#ifdef FEC_HAVE_NEON + fec_neon_self_test(); +#endif fec_initialized = 1 ; } From a1c4f62a201a83d2b80b0a1f7c70534c43d2a1c1 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 5 Jul 2026 07:54:04 +0000 Subject: [PATCH 2/3] Reduce rx packet handoff latency and CPU in radio rx queues - Consumer wait: replace the sem_trywait + fixed 200us sleep + retry poll with a blocking sem_clockwait on CLOCK_MONOTONIC honoring the caller's timeout (glibc >= 2.30; other libcs keep the previous behavior). The consumer now wakes immediately on sem_post instead of finishing a fixed sleep, and does not burn CPU polling while idle. The previously commented-out sem_timedwait attempt had a unit bug (timeout scaled by 1e6 instead of 1e3) and used CLOCK_REALTIME, which jumps on NTP sync. - Enqueue ordering: fill the queue slot (payload + metadata) before advancing the published write index, so a consumer woken between the index update and the memcpy can no longer read a partially written slot. - Rx thread: map poll fd indexes to radio interface indexes when building the pollfd array instead of rescanning all interfaces for every ready fd. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01TWFTzQ3aWucyVgKpyf6HcT --- code/radio/radio_rx.c | 84 +++++++++++++++++++++---------------------- 1 file changed, 40 insertions(+), 44 deletions(-) diff --git a/code/radio/radio_rx.c b/code/radio/radio_rx.c index 5df98fa1..ce6d595a 100644 --- a/code/radio/radio_rx.c +++ b/code/radio/radio_rx.c @@ -30,6 +30,11 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. */ +// For sem_clockwait (glibc >= 2.30) +#ifndef _GNU_SOURCE +#define _GNU_SOURCE +#endif + #include "../base/base.h" #include "../base/encr.h" #include "../base/config_hw.h" @@ -41,6 +46,12 @@ #include "radio_duplicate_det.h" #include +// sem_clockwait lets the consumer block on the rx queues against CLOCK_MONOTONIC +// (immune to wall clock jumps) instead of polling with sem_trywait + sleep +#if defined(__GLIBC__) && ((__GLIBC__ > 2) || ((__GLIBC__ == 2) && (__GLIBC_MINOR__ >= 30))) +#define RADIO_RX_USE_SEM_CLOCKWAIT 1 +#endif + int s_iRadioRxInitialized = 0; int s_iRadioRxThreadRunning = 0; @@ -266,33 +277,30 @@ void _radio_rx_update_fd_sets() u8* _radio_rx_wait_get_queue_packet(t_radio_rx_state_packets_queue* pQueue, int iHighPriorityQueue, u32 uTimeoutMicroSec, int* pLength, int* pIsShortPacket, int* pRadioInterfaceIndex) { int iRes = -1; - /* - if ( 0 == uTimeoutMicroSec ) - iRes = sem_trywait(pQueue->pSemaphoreRead); - else + iRes = sem_trywait(pQueue->pSemaphoreRead); + if ( (0 != iRes) && (0 != uTimeoutMicroSec) ) { +#ifdef RADIO_RX_USE_SEM_CLOCKWAIT + // Block until a packet is posted or the timeout expires. Waking up on + // sem_post directly (instead of finishing a fixed sleep first) reduces + // the packet handoff latency from rx thread to consumer. struct timespec ts; - clock_gettime(CLOCK_REALTIME, &ts); - ts.tv_nsec += 1000LL*(long long)uTimeoutMicroSec*1000LL; - if ( ts.tv_nsec >= 1000000000LL ) + clock_gettime(CLOCK_MONOTONIC, &ts); + ts.tv_nsec += (long)uTimeoutMicroSec * 1000L; + while ( ts.tv_nsec >= 1000000000L ) { ts.tv_sec++; ts.tv_nsec -= 1000000000L; } - iRes = sem_timedwait(pQueue->pSemaphoreRead, &ts); - } - if ( 0 != iRes ) - { - if ( errno != ETIMEDOUT ) - log_softerror_and_alarm("[RadioRx] Failed to timewait on %s semaphore for %u micros. Error: %d, %d, %s", iHighPriorityQueue?"high prio":"reg prio", uTimeoutMicroSec, iRes, errno, strerror(errno)); - return NULL; - } - */ - iRes = sem_trywait(pQueue->pSemaphoreRead); - if ( (0 != iRes) && (0 != uTimeoutMicroSec) ) - { + do + { + iRes = sem_clockwait(pQueue->pSemaphoreRead, CLOCK_MONOTONIC, &ts); + } + while ( (0 != iRes) && (EINTR == errno) ); +#else hardware_sleep_micros(200); iRes = sem_trywait(pQueue->pSemaphoreRead); +#endif } if ( 0 != iRes ) return NULL; @@ -460,9 +468,16 @@ void _radio_rx_add_packet_to_rx_queue(u8* pPacket, int iLength, int iRadioInterf if ( uPacketFlags & PACKET_FLAGS_BIT_HIGH_PRIORITY ) pQueue = &s_RadioRxState.queue_high_priority; - int iIndexToWriteTo = -1; + // Fill the slot before publishing it to the consumer: this thread is the + // only writer of iCurrentPacketIndexToWrite and the consumer never reads + // past it, so the slot data must be complete before the index advances + int iIndexToWriteTo = pQueue->iCurrentPacketIndexToWrite; + pQueue->uPacketsRxInterface[iIndexToWriteTo] = iRadioInterface; + pQueue->uPacketsAreShort[iIndexToWriteTo] = 0; + pQueue->iPacketsLengths[iIndexToWriteTo] = iLength; + memcpy(pQueue->pPacketsBuffers[iIndexToWriteTo], pPacket, iLength); + pthread_mutex_lock(&pQueue->mutexLock); - iIndexToWriteTo = pQueue->iCurrentPacketIndexToWrite; // No more room? Discard oldest packet if ( ((pQueue->iCurrentPacketIndexToWrite+1) % pQueue->iQueueSize) == pQueue->iCurrentPacketIndexToConsume ) pQueue->iCurrentPacketIndexToConsume = (pQueue->iCurrentPacketIndexToConsume+1) % pQueue->iQueueSize; @@ -470,12 +485,6 @@ void _radio_rx_add_packet_to_rx_queue(u8* pPacket, int iLength, int iRadioInterf pQueue->iCurrentPacketIndexToWrite = (pQueue->iCurrentPacketIndexToWrite + 1) % pQueue->iQueueSize; pthread_mutex_unlock(&pQueue->mutexLock); - // Add the packet to the queue - pQueue->uPacketsRxInterface[iIndexToWriteTo] = iRadioInterface; - pQueue->uPacketsAreShort[iIndexToWriteTo] = 0; - pQueue->iPacketsLengths[iIndexToWriteTo] = iLength; - memcpy(pQueue->pPacketsBuffers[iIndexToWriteTo], pPacket, iLength); - if ( (NULL != pQueue->pSemaphoreWrite) && (0 != sem_post(pQueue->pSemaphoreWrite)) ) log_softerror_and_alarm("Failed to set semaphore for packet ready."); @@ -1090,6 +1099,7 @@ void * _thread_radio_rx(void *argument) iLoopParsedPackets = 0; struct pollfd fds[MAX_RADIO_INTERFACES]; + int iPollIndexToInterfaceIndex[MAX_RADIO_INTERFACES]; int iRadioInterfacesWherePaused[MAX_RADIO_INTERFACES]; s_iRadioRxCountFDs = 0; for( int i=0; iruntimeInterfaceInfoRx.selectable_fd; fds[s_iRadioRxCountFDs].revents = 0; fds[s_iRadioRxCountFDs].events = POLLIN; + iPollIndexToInterfaceIndex[s_iRadioRxCountFDs] = i; s_iRadioRxCountFDs++; } @@ -1177,24 +1188,9 @@ void * _thread_radio_rx(void *argument) if ( 0 == (fds[iPollIndex].revents & POLLIN) ) continue; - int iInterfaceIndex = -1; - for( int i=0; iopenedForRead) ) - continue; - if ( s_RadioRxState.iRadioInterfacesBroken[i] ) - continue; - if ( iRadioInterfacesWherePaused[i] ) - continue; - if ( fds[iPollIndex].fd == pRadioHWInfo->runtimeInterfaceInfoRx.selectable_fd ) - { - iInterfaceIndex = i; - break; - } - } + int iInterfaceIndex = iPollIndexToInterfaceIndex[iPollIndex]; - if ( (iInterfaceIndex == -1) || (iInterfaceIndex >= MAX_RADIO_INTERFACES) ) + if ( (iInterfaceIndex < 0) || (iInterfaceIndex >= MAX_RADIO_INTERFACES) ) continue; radio_hw_info_t* pRadioHWInfo = hardware_get_radio_info(iInterfaceIndex); From 0730c7ebe42f6607615284076f73745cd965f1d3 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 5 Jul 2026 07:54:17 +0000 Subject: [PATCH 3/3] Move streamer pipe writes off the router loop to a writer thread The video output FIFO to the streamer is opened blocking, so a stalled video player made write() block the router main loop, delaying retransmission requests and all radio/IPC processing (the 'main loop took too long' overloads). Route the writes through a 512KB ring buffer drained by a dedicated writer thread: - The router loop now only memcpys into the ring (never blocks on the player). If the player stalls long enough to fill the ring, new data is dropped and reported as a truncated write, feeding the existing IO error alarm logic. Writer IO errors (EPIPE etc) are reported back on the next output call, feeding the same restart logic as before. - The writer thread polls POLLOUT with a short timeout and writes at most PIPE_BUF bytes per wakeup, so it never blocks indefinitely and stops promptly when the pipe is closed or reopened. Also fix the pipe open retry loop, which did not stop on success: it opened the FIFO up to 200 times, leaking the first 199 descriptors, and pass a mode to open(O_CREAT) as required. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01TWFTzQ3aWucyVgKpyf6HcT --- code/r_station/rx_video_output.cpp | 204 +++++++++++++++++++++++++++-- 1 file changed, 192 insertions(+), 12 deletions(-) diff --git a/code/r_station/rx_video_output.cpp b/code/r_station/rx_video_output.cpp index 87e6d373..3d587e7a 100644 --- a/code/r_station/rx_video_output.cpp +++ b/code/r_station/rx_video_output.cpp @@ -36,6 +36,7 @@ #include #include #include +#include #include #include #include @@ -152,6 +153,177 @@ u32 s_uOutputBitrateToLocalVideoPlayerUDP = 0; char s_szOutputVideoStreamerFilename[MAX_FILE_PATH_SIZE]; int s_iPIDVideoStreamer = -1; +// A write() on the streamer FIFO blocks if the video player stalls, and doing it +// on the router loop delays retransmissions and radio packets processing. +// So the writes are done on a dedicated thread, fed through a ring buffer. +// If the player stalls long enough to fill the ring, incoming data is dropped +// (reported to the caller as a truncated write, same as before). +#define VIDEO_OUT_PIPE_RING_SIZE 512000 +// POLLOUT on a pipe guarantees PIPE_BUF (4096) bytes can be written without blocking +#define VIDEO_OUT_PIPE_MAX_WRITE_SIZE 4096 + +pthread_t s_ThreadVideoOutPipeWriter; +bool s_bVideoOutPipeWriterThreadRunning = false; +volatile bool s_bVideoOutPipeWriterMustStop = false; +pthread_mutex_t s_MutexVideoOutPipeWriter = PTHREAD_MUTEX_INITIALIZER; +pthread_cond_t s_CondVideoOutPipeWriter = PTHREAD_COND_INITIALIZER; +u8* s_pVideoOutPipeRing = NULL; +int s_iVideoOutPipeRingReadPos = 0; +int s_iVideoOutPipeRingWritePos = 0; +int s_iVideoOutPipeRingBytes = 0; +volatile int s_iVideoOutPipeWriterLastError = 0; + +static void * _thread_rx_video_output_pipe_writer(void *argument) +{ + log_line("[VideoOutput] Started pipe writer thread."); + while ( ! s_bVideoOutPipeWriterMustStop ) + { + pthread_mutex_lock(&s_MutexVideoOutPipeWriter); + while ( (0 == s_iVideoOutPipeRingBytes) && (! s_bVideoOutPipeWriterMustStop) ) + pthread_cond_wait(&s_CondVideoOutPipeWriter, &s_MutexVideoOutPipeWriter); + int iReadPos = s_iVideoOutPipeRingReadPos; + int iChunk = s_iVideoOutPipeRingBytes; + pthread_mutex_unlock(&s_MutexVideoOutPipeWriter); + + if ( s_bVideoOutPipeWriterMustStop ) + break; + + if ( iChunk > VIDEO_OUT_PIPE_RING_SIZE - iReadPos ) + iChunk = VIDEO_OUT_PIPE_RING_SIZE - iReadPos; + if ( iChunk > VIDEO_OUT_PIPE_MAX_WRITE_SIZE ) + iChunk = VIDEO_OUT_PIPE_MAX_WRITE_SIZE; + + int fd = s_fPipeVideoOutToStreamer; + if ( fd < 0 ) + { + hardware_sleep_ms(5); + continue; + } + + // Poll with a short timeout (instead of a plain blocking write) so the + // thread can be stopped promptly even if the player never reads + struct pollfd pfd; + pfd.fd = fd; + pfd.events = POLLOUT; + pfd.revents = 0; + int iPollRes = poll(&pfd, 1, 50); + if ( s_bVideoOutPipeWriterMustStop ) + break; + if ( 0 == iPollRes ) + continue; + int iRes = -1; + if ( (iPollRes > 0) && (pfd.revents & POLLOUT) ) + iRes = write(fd, &s_pVideoOutPipeRing[iReadPos], iChunk); + else if ( iPollRes < 0 ) + { + if ( EINTR == errno ) + continue; + } + else + errno = EPIPE; + + if ( iRes > 0 ) + { + pthread_mutex_lock(&s_MutexVideoOutPipeWriter); + s_iVideoOutPipeRingReadPos = (s_iVideoOutPipeRingReadPos + iRes) % VIDEO_OUT_PIPE_RING_SIZE; + s_iVideoOutPipeRingBytes -= iRes; + pthread_mutex_unlock(&s_MutexVideoOutPipeWriter); + continue; + } + if ( (iRes < 0) && ((EINTR == errno) || (EAGAIN == errno)) ) + continue; + + // Pipe is broken (player stopped or crashed). Report the error to the + // main thread (picked up on the next output call), drop the pending + // data and go easy until the pipe is reopened. + s_iVideoOutPipeWriterLastError = (errno != 0) ? errno : EPIPE; + pthread_mutex_lock(&s_MutexVideoOutPipeWriter); + s_iVideoOutPipeRingReadPos = s_iVideoOutPipeRingWritePos; + s_iVideoOutPipeRingBytes = 0; + pthread_mutex_unlock(&s_MutexVideoOutPipeWriter); + hardware_sleep_ms(10); + } + log_line("[VideoOutput] Stopped pipe writer thread."); + return NULL; +} + +static void _rx_video_output_start_pipe_writer_thread() +{ + if ( s_bVideoOutPipeWriterThreadRunning ) + return; + + if ( NULL == s_pVideoOutPipeRing ) + { + s_pVideoOutPipeRing = (u8*)malloc(VIDEO_OUT_PIPE_RING_SIZE); + if ( NULL == s_pVideoOutPipeRing ) + { + log_softerror_and_alarm("[VideoOutput] Failed to allocate pipe writer ring buffer. Pipe writes will be done in place."); + return; + } + } + s_iVideoOutPipeRingReadPos = 0; + s_iVideoOutPipeRingWritePos = 0; + s_iVideoOutPipeRingBytes = 0; + s_iVideoOutPipeWriterLastError = 0; + s_bVideoOutPipeWriterMustStop = false; + + if ( 0 != pthread_create(&s_ThreadVideoOutPipeWriter, NULL, &_thread_rx_video_output_pipe_writer, NULL) ) + { + log_softerror_and_alarm("[VideoOutput] Failed to create pipe writer thread. Pipe writes will be done in place."); + return; + } + s_bVideoOutPipeWriterThreadRunning = true; +} + +static void _rx_video_output_stop_pipe_writer_thread() +{ + if ( ! s_bVideoOutPipeWriterThreadRunning ) + return; + pthread_mutex_lock(&s_MutexVideoOutPipeWriter); + s_bVideoOutPipeWriterMustStop = true; + pthread_cond_signal(&s_CondVideoOutPipeWriter); + pthread_mutex_unlock(&s_MutexVideoOutPipeWriter); + pthread_join(s_ThreadVideoOutPipeWriter, NULL); + s_bVideoOutPipeWriterThreadRunning = false; +} + +// Queues data for the pipe writer thread. +// Returns the count of bytes accepted (less than iLength if the ring is full, +// which the caller reports as a truncated write) or -1 if the writer thread +// hit an IO error since the last call (with errno set to that error). +static int _rx_video_output_pipe_write_async(u8* pData, int iLength) +{ + if ( (! s_bVideoOutPipeWriterThreadRunning) || (NULL == s_pVideoOutPipeRing) ) + return write(s_fPipeVideoOutToStreamer, pData, iLength); + + int iErr = s_iVideoOutPipeWriterLastError; + if ( 0 != iErr ) + { + s_iVideoOutPipeWriterLastError = 0; + errno = iErr; + return -1; + } + + pthread_mutex_lock(&s_MutexVideoOutPipeWriter); + int iToCopy = VIDEO_OUT_PIPE_RING_SIZE - s_iVideoOutPipeRingBytes; + if ( iToCopy > iLength ) + iToCopy = iLength; + int iCopied = 0; + while ( iCopied < iToCopy ) + { + int iChunk = iToCopy - iCopied; + if ( iChunk > VIDEO_OUT_PIPE_RING_SIZE - s_iVideoOutPipeRingWritePos ) + iChunk = VIDEO_OUT_PIPE_RING_SIZE - s_iVideoOutPipeRingWritePos; + memcpy(&s_pVideoOutPipeRing[s_iVideoOutPipeRingWritePos], pData + iCopied, iChunk); + s_iVideoOutPipeRingWritePos = (s_iVideoOutPipeRingWritePos + iChunk) % VIDEO_OUT_PIPE_RING_SIZE; + iCopied += iChunk; + } + s_iVideoOutPipeRingBytes += iCopied; + pthread_cond_signal(&s_CondVideoOutPipeWriter); + pthread_mutex_unlock(&s_MutexVideoOutPipeWriter); + return iCopied; +} + void rx_video_output_start_video_streamer() { log_line("[VideoOutput] Starting video streamer [%s]", s_szOutputVideoStreamerFilename); @@ -312,6 +484,7 @@ void rx_video_output_start_video_streamer() void rx_video_output_stop_video_streamer() { log_line("[VideoOutput] Stopping video streamer..."); + _rx_video_output_stop_pipe_writer_thread(); if ( -1 != s_fPipeVideoOutToStreamer ) { log_line("[VideoOutput] Closed video output pipe to streamer."); @@ -535,16 +708,15 @@ void _rx_video_output_open_pipe_to_streamer() { iRetries--; //s_fPipeVideoOutToStreamer = open(FIFO_RUBY_STATION_VIDEO_STREAM_DISPLAY, O_CREAT | O_WRONLY | O_NONBLOCK); - s_fPipeVideoOutToStreamer = open(FIFO_RUBY_STATION_VIDEO_STREAM_DISPLAY, O_CREAT | O_WRONLY); - if ( s_fPipeVideoOutToStreamer < 0 ) - { - log_error_and_alarm("[VideoOutput] Failed to open video output pipe to streamer write endpoint: %s, error code (%d): [%s]", - FIFO_RUBY_STATION_VIDEO_STREAM_DISPLAY, errno, strerror(errno)); - if ( iRetries == 0 ) - return; - else - hardware_sleep_ms(10); - } + s_fPipeVideoOutToStreamer = open(FIFO_RUBY_STATION_VIDEO_STREAM_DISPLAY, O_CREAT | O_WRONLY, 0644); + if ( s_fPipeVideoOutToStreamer >= 0 ) + break; + log_error_and_alarm("[VideoOutput] Failed to open video output pipe to streamer write endpoint: %s, error code (%d): [%s]", + FIFO_RUBY_STATION_VIDEO_STREAM_DISPLAY, errno, strerror(errno)); + if ( iRetries == 0 ) + return; + else + hardware_sleep_ms(10); } log_line("[VideoOutput] Opened video output pipe to streamer write endpoint: %s", FIFO_RUBY_STATION_VIDEO_STREAM_DISPLAY); log_line("[VideoOutput] Video output pipe to streamer flags: %s", str_get_pipe_flags(fcntl(s_fPipeVideoOutToStreamer, F_GETFL))); @@ -558,6 +730,8 @@ void _rx_video_output_open_pipe_to_streamer() fcntl(s_fPipeVideoOutToStreamer, F_SETPIPE_SZ, 250000); log_line("[VideoOutput] Video streamer FIFO new size: %d bytes", fcntl(s_fPipeVideoOutToStreamer, F_GETPIPE_SZ)); s_bDidSentAnyDataToVideoStreamerPipe = false; + + _rx_video_output_start_pipe_writer_thread(); } void rx_video_output_init() @@ -756,6 +930,7 @@ void rx_video_output_uninit() s_VideoETHOutputInfo.s_bForwardIsETHForwardEnabled = false; s_VideoETHOutputInfo.s_bForwardETHPipeEnabled = false; + _rx_video_output_stop_pipe_writer_thread(); if ( -1 != s_fPipeVideoOutToStreamer ) { log_line("[VideoOutput] Closed video output pipe to streamer."); @@ -770,6 +945,10 @@ void rx_video_output_uninit() s_pPipeVideoOutputBuffer = NULL; s_iPipeVideoOutputPos = 0; + if ( NULL != s_pVideoOutPipeRing ) + free(s_pVideoOutPipeRing); + s_pVideoOutPipeRing = NULL; + if ( -1 != s_VideoUSBOutputInfo.socketUSBOutput ) close(s_VideoUSBOutputInfo.socketUSBOutput); s_VideoUSBOutputInfo.socketUSBOutput = -1; @@ -828,6 +1007,7 @@ void rx_video_output_disable_streamer_output() log_line("[VideoOutput] Disable video output to streamer."); s_bEnableVideoStreamerOutput = false; + _rx_video_output_stop_pipe_writer_thread(); if ( -1 != s_fPipeVideoOutToStreamer ) { close( s_fPipeVideoOutToStreamer ); @@ -988,7 +1168,7 @@ void _rx_video_output_to_video_streamer_pipe(u8* pBuffer, int iLength, bool bWai if ( (NULL == s_pPipeVideoOutputBuffer) || (! bWaitFullFrame) ) { g_pProcessStats->uInBlockingOperation = 1; - iRes = write(s_fPipeVideoOutToStreamer, pBuffer, iLength); + iRes = _rx_video_output_pipe_write_async(pBuffer, iLength); g_pProcessStats->uInBlockingOperation = 0; } else @@ -1003,7 +1183,7 @@ void _rx_video_output_to_video_streamer_pipe(u8* pBuffer, int iLength, bool bWai { iExpectedWriteResult = s_iPipeVideoOutputPos; g_pProcessStats->uInBlockingOperation = 1; - iRes = write(s_fPipeVideoOutToStreamer, s_pPipeVideoOutputBuffer, s_iPipeVideoOutputPos); + iRes = _rx_video_output_pipe_write_async(s_pPipeVideoOutputBuffer, s_iPipeVideoOutputPos); g_pProcessStats->uInBlockingOperation = 0; s_iPipeVideoOutputPos = 0; }