From ce1b2ebe6a7607c634a17b4197e1799a00bf3cb5 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Sat, 17 Jan 2026 23:21:16 -0500 Subject: [PATCH 01/11] WIP synchronous outlet 1. User creates outlet with transp_sync_blocking flag: lsl::stream_outlet outlet(info, 0, 360, transp_sync_blocking); 2. When a consumer connects, the socket is handed off from client_session to sync_write_handler after the feed header handshake (no transfer thread is spawned). 3. When push_sample() is called: - Timestamp is encoded and stored in sync_timestamps_ - User's data buffer pointer is wrapped in asio::const_buffer (zero copy) - If pushthrough=true, all buffers are written to all consumers via blocking gather-write --- examples/CMakeLists.txt | 1 + examples/SendDataSyncBlocking.cpp | 84 +++++++++++++++++++++++ include/lsl/common.h | 6 ++ src/sample.h | 5 ++ src/stream_outlet_impl.cpp | 81 ++++++++++++++++++++--- src/stream_outlet_impl.h | 25 +++++++ src/tcp_server.cpp | 106 +++++++++++++++++++++++++++++- src/tcp_server.h | 24 ++++++- 8 files changed, 320 insertions(+), 12 deletions(-) create mode 100644 examples/SendDataSyncBlocking.cpp diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index d4caf88d4..9b2e70d28 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -91,6 +91,7 @@ addlslexample(SendDataSimple cpp) addlslexample(SendMultipleStreams cpp) addlslexample(SendStringMarkers cpp) addlslexample(SendStringMarkersC c) +addlslexample(SendDataSyncBlocking cpp) addlslexample(TestSyncWithoutData cpp) target_link_libraries(TestSyncWithoutData PRIVATE Threads::Threads) diff --git a/examples/SendDataSyncBlocking.cpp b/examples/SendDataSyncBlocking.cpp new file mode 100644 index 000000000..c379dbbce --- /dev/null +++ b/examples/SendDataSyncBlocking.cpp @@ -0,0 +1,84 @@ +#include +#include +#include +#include +#include +#include + +/** + * This example demonstrates the synchronous (zero-copy) outlet mode. + * + * When using transp_sync_blocking, push_sample() blocks until data is written + * to all connected consumers. This eliminates data copies (user buffer is sent + * directly to the network socket) which reduces CPU usage for high-bandwidth streams. + * + * Trade-off: Call latency depends on network speed and number of consumers. + * + * Usage: SendDataSyncBlocking [stream_name] [num_channels] [sample_rate] + * Default: SyncStream, 64 channels, 1000 Hz + */ + +int main(int argc, char *argv[]) { + std::string name = argc > 1 ? argv[1] : "SyncStream"; + int nchannels = argc > 2 ? std::atoi(argv[2]) : 64; + double srate = argc > 3 ? std::atof(argv[3]) : 1000.0; + + std::cout << "Creating sync outlet: " << name << " with " << nchannels << " channels @ " + << srate << " Hz\n"; + + // Create stream info + lsl::stream_info info(name, "EEG", nchannels, srate, lsl::cf_float32); + + // Create outlet with transp_sync_blocking flag for zero-copy transfer + // Note: The third parameter (max_buffered) is less important in sync mode + // since data goes directly to the socket without intermediate buffering. + lsl::stream_outlet outlet(info, 0, 360, lsl::transp_sync_blocking); + + std::cout << "Waiting for consumers...\n"; + while (!outlet.wait_for_consumers(5)) { + std::cout << " (still waiting)\n"; + } + std::cout << "Consumer connected! Starting data transmission.\n"; + + // Allocate sample buffer + std::vector sample(nchannels); + + // Calculate sleep duration between samples + auto sample_interval = std::chrono::duration(1.0 / srate); + auto next_sample_time = std::chrono::steady_clock::now(); + + // Statistics + uint64_t samples_sent = 0; + auto start_time = std::chrono::steady_clock::now(); + + // Send data while consumers are connected + while (outlet.have_consumers()) { + // Generate sample data (in real applications, this would be acquired from hardware) + for (int c = 0; c < nchannels; c++) { + sample[c] = static_cast((std::rand() % 1000) / 500.0 - 1.0); + } + + // Push sample - this BLOCKS until data is written to all consumers + // The sample buffer is used directly (zero-copy) so it must remain valid + // until push_sample returns. + outlet.push_sample(sample); + samples_sent++; + + // Print statistics every second + auto now = std::chrono::steady_clock::now(); + auto elapsed = std::chrono::duration(now - start_time).count(); + if (samples_sent % static_cast(srate) == 0) { + std::cout << "Sent " << samples_sent << " samples, effective rate: " + << (samples_sent / elapsed) << " Hz\n"; + } + + // Pace the data transmission + next_sample_time += std::chrono::duration_cast( + sample_interval); + std::this_thread::sleep_until(next_sample_time); + } + + std::cout << "Consumer disconnected. Total samples sent: " << samples_sent << "\n"; + + return 0; +} diff --git a/include/lsl/common.h b/include/lsl/common.h index 855ca78b7..89ce83b97 100644 --- a/include/lsl/common.h +++ b/include/lsl/common.h @@ -161,6 +161,12 @@ typedef enum { /// The supplied max_buf should be scaled by 0.001. transp_bufsize_thousandths = 2, + /// Use synchronous (blocking) socket writes for zero-copy data transfer. + /// When enabled, push_sample/push_chunk will block until data is written to all consumers. + /// Reduces CPU usage for high-bandwidth streams at the cost of increased call latency. + /// Not compatible with string-format streams. + transp_sync_blocking = 4, + // prevent compilers from assuming an instance fits in a single byte _lsl_transport_options_maxval = 0x7f000000 } lsl_transport_options_t; diff --git a/src/sample.h b/src/sample.h index a60b4f2f0..86553c056 100644 --- a/src/sample.h +++ b/src/sample.h @@ -52,6 +52,11 @@ class factory { /// Reclaim a sample that's no longer used. void reclaim_sample(sample *s); + /// Return the size of the sample data payload in bytes. + std::size_t datasize() const { + return format_sizes[fmt_] * static_cast(num_chans_); + } + private: /// Pop a sample from the freelist (multi-producer/single-consumer queue by Dmitry Vjukov) sample *pop_freelist(); diff --git a/src/stream_outlet_impl.cpp b/src/stream_outlet_impl.cpp index 43cca9d5b..5068cfc56 100644 --- a/src/stream_outlet_impl.cpp +++ b/src/stream_outlet_impl.cpp @@ -7,7 +7,9 @@ #include "udp_server.h" #include #include +#include #include +#include namespace lsl { @@ -23,8 +25,20 @@ stream_outlet_impl::stream_outlet_impl(const stream_info_impl &info, int32_t chu info_(std::make_shared(info)), send_buffer_(std::make_shared(chunk_size_)), io_ctx_data_(std::make_shared(1)), - io_ctx_service_(std::make_shared(1)) { + io_ctx_service_(std::make_shared(1)), + sync_mode_((flags & transp_sync_blocking) != 0) { ensure_lsl_initialized(); + + // Validate sync mode constraints + if (sync_mode_) { + if (info.channel_format() == cft_string) { + throw std::invalid_argument( + "Synchronous (zero-copy) mode is not supported for string-format streams"); + } + LOG_F(INFO, "Creating outlet in synchronous (zero-copy) mode for stream '%s'", + info.name().c_str()); + } + const api_config *cfg = api_config::get_instance(); // instantiate IPv4 and/or IPv6 stacks (depending on settings) @@ -41,7 +55,7 @@ stream_outlet_impl::stream_outlet_impl(const stream_info_impl &info, int32_t chu // create TCP data server tcp_server_ = std::make_shared(info_, io_ctx_data_, send_buffer_, sample_factory_, - chunk_size_, cfg->allow_ipv4(), cfg->allow_ipv6()); + chunk_size_, cfg->allow_ipv4(), cfg->allow_ipv6(), sync_mode_); // fail if both stacks failed to instantiate if (udp_servers_.empty()) @@ -146,10 +160,17 @@ stream_outlet_impl::~stream_outlet_impl() { void stream_outlet_impl::push_numeric_raw(const void *data, double timestamp, bool pushthrough) { if (lsl::api_config::get_instance()->force_default_timestamps()) timestamp = 0.0; - sample_p smp( - sample_factory_->new_sample(timestamp == 0.0 ? lsl_clock() : timestamp, pushthrough)); - smp->assign_untyped(data); - send_buffer_->push_sample(smp); + if (sync_mode_) { + // Sync path: directly use user's buffer (zero-copy) + enqueue_sync(asio::const_buffer(data, sample_factory_->datasize()), + timestamp == 0.0 ? lsl_clock() : timestamp, pushthrough); + } else { + // Async path: copy into sample + sample_p smp( + sample_factory_->new_sample(timestamp == 0.0 ? lsl_clock() : timestamp, pushthrough)); + smp->assign_untyped(data); + send_buffer_->push_sample(smp); + } } bool stream_outlet_impl::have_consumers() { return send_buffer_->have_consumers(); } @@ -161,10 +182,18 @@ bool stream_outlet_impl::wait_for_consumers(double timeout) { template void stream_outlet_impl::enqueue(const T *data, double timestamp, bool pushthrough) { if (lsl::api_config::get_instance()->force_default_timestamps()) timestamp = 0.0; - sample_p smp( - sample_factory_->new_sample(timestamp == 0.0 ? lsl_clock() : timestamp, pushthrough)); - smp->assign_typed(data); - send_buffer_->push_sample(smp); + // Sync mode only for non-string types (strings have variable size, can't do zero-copy) + if (sync_mode_ && !std::is_same::value) { + // Sync path: directly use user's buffer (zero-copy) + enqueue_sync(asio::const_buffer(data, sample_factory_->datasize()), + timestamp == 0.0 ? lsl_clock() : timestamp, pushthrough); + } else { + // Async path: copy into sample + sample_p smp( + sample_factory_->new_sample(timestamp == 0.0 ? lsl_clock() : timestamp, pushthrough)); + smp->assign_typed(data); + send_buffer_->push_sample(smp); + } } template void stream_outlet_impl::enqueue(const char *data, double, bool); @@ -175,4 +204,36 @@ template void stream_outlet_impl::enqueue(const float *data, double, bool template void stream_outlet_impl::enqueue(const double *data, double, bool); template void stream_outlet_impl::enqueue(const std::string *data, double, bool); +// === Sync mode implementation === + +void stream_outlet_impl::push_timestamp_sync(double timestamp) { + // Allocate storage for timestamp tag + value in sync_timestamps_ + // Use TAG_TRANSMITTED_TIMESTAMP (2) for explicit timestamps + sync_timestamps_.emplace_back(TAG_TRANSMITTED_TIMESTAMP, timestamp); + auto &ts_entry = sync_timestamps_.back(); + // Add buffers for tag and timestamp + sync_buffers_.push_back(asio::const_buffer(&ts_entry.first, 1)); // tag byte + sync_buffers_.push_back(asio::const_buffer(&ts_entry.second, sizeof(double))); +} + +void stream_outlet_impl::flush_sync() { + if (sync_buffers_.empty()) return; + // Write all buffers to connected consumers + tcp_server_->write_all_blocking(sync_buffers_); + // Clear buffers and timestamp storage for next batch + sync_buffers_.clear(); + sync_timestamps_.clear(); +} + +void stream_outlet_impl::enqueue_sync(asio::const_buffer buf, double timestamp, bool pushthrough) { + // Add timestamp + push_timestamp_sync(timestamp); + // Add sample data buffer (points to user's buffer - zero copy!) + sync_buffers_.push_back(buf); + // Flush if pushthrough is requested + if (pushthrough) { + flush_sync(); + } +} + } // namespace lsl diff --git a/src/stream_outlet_impl.h b/src/stream_outlet_impl.h index 1d6d3b084..00d9bcff8 100644 --- a/src/stream_outlet_impl.h +++ b/src/stream_outlet_impl.h @@ -4,11 +4,13 @@ #include "common.h" #include "forward.h" #include "stream_info_impl.h" +#include #include #include #include #include #include +#include #include using asio::ip::tcp; @@ -294,6 +296,9 @@ class stream_outlet_impl { /// Wait until some consumer shows up. bool wait_for_consumers(double timeout = FOREVER); + /// Check if this outlet is in synchronous (zero-copy) mode + bool is_sync_mode() const { return sync_mode_; } + private: /// Instantiate a new server stack. void instantiate_stack(udp udp_protocol); @@ -301,6 +306,17 @@ class stream_outlet_impl { /// Allocate and enqueue a new sample into the send buffer. template void enqueue(const T *data, double timestamp, bool pushthrough); + // === Sync mode helpers === + + /// Append timestamp encoding to sync_buffers_ + void push_timestamp_sync(double timestamp); + + /// Flush sync_buffers_ to all connected consumers (blocking) + void flush_sync(); + + /// Enqueue a buffer for sync transfer (single sample) + void enqueue_sync(asio::const_buffer buf, double timestamp, bool pushthrough); + /** * Check whether some given number of channels matches the stream's channel_count. * Throws an error if not. @@ -331,6 +347,15 @@ class stream_outlet_impl { std::vector responders_; /// threads that handle the I/O operations (two per stack: one for UDP and one for TCP) std::vector io_threads_; + + // === Sync mode members === + + /// Flag indicating sync (zero-copy blocking) mode is enabled + bool sync_mode_{false}; + /// Buffers accumulated for sync gather-write + std::vector sync_buffers_; + /// Storage for timestamps in sync mode (tag + timestamp pairs) + std::vector> sync_timestamps_; }; } // namespace lsl diff --git a/src/tcp_server.cpp b/src/tcp_server.cpp index fc9747929..c4b1ba690 100644 --- a/src/tcp_server.cpp +++ b/src/tcp_server.cpp @@ -60,6 +60,84 @@ namespace lsl { * - So memory is generally owned by the code (functors and stack frames) that needs to refer to * it for the duration of the execution. */ +/** + * Handler for synchronous (blocking) writes to all connected clients. + * This class manages sockets that have been handed off from client_session + * for zero-copy synchronous data transfer. + */ +class sync_write_handler { +public: + sync_write_handler() : io_ctx_(1) {} + + ~sync_write_handler() { + // Close all sockets + for (auto &sock : sockets_) { + if (sock && sock->is_open()) { + asio::error_code ec; + sock->close(ec); + } + } + } + + /// Add a socket for sync writes (called from client_session after handshake) + void add_socket(tcp_socket::native_handle_type handle, tcp_socket::protocol_type protocol) { + std::lock_guard lock(mutex_); + sockets_.push_back(std::make_unique(io_ctx_, protocol, handle)); + LOG_F(INFO, "Added sync socket, now have %zu consumers", sockets_.size()); + } + + /// Write buffers to all connected sockets (blocking gather-write) + void write_all_blocking(const std::vector &bufs) { + std::lock_guard lock(mutex_); + if (sockets_.empty()) return; + + bool any_broken = false; + + for (auto &sock : sockets_) { + if (!sock || !sock->is_open()) continue; + + asio::error_code ec; + asio::write(*sock, bufs, ec); + + if (ec) { + switch (ec.value()) { + case asio::error::broken_pipe: + case asio::error::connection_reset: + case asio::error::not_connected: + LOG_F(WARNING, "Sync socket disconnected: %s", ec.message().c_str()); + sock->close(ec); + any_broken = true; + break; + default: + LOG_F(ERROR, "Sync write error: %s", ec.message().c_str()); + sock->close(ec); + any_broken = true; + } + } + } + + // Remove closed sockets + if (any_broken) { + sockets_.erase( + std::remove_if(sockets_.begin(), sockets_.end(), + [](const tcp_socket_p &s) { return !s || !s->is_open(); }), + sockets_.end()); + LOG_F(INFO, "After cleanup, have %zu sync consumers", sockets_.size()); + } + } + + /// Check if there are any connected consumers + bool have_consumers() const { + std::lock_guard lock(mutex_); + return !sockets_.empty(); + } + +private: + asio::io_context io_ctx_; + std::vector sockets_; + mutable std::mutex mutex_; +}; + class client_session : public std::enable_shared_from_this { public: @@ -143,9 +221,14 @@ class client_session : public std::enable_shared_from_this { }; tcp_server::tcp_server(stream_info_impl_p info, io_context_p io, send_buffer_p sendbuf, - factory_p factory, int chunk_size, bool allow_v4, bool allow_v6) + factory_p factory, int chunk_size, bool allow_v4, bool allow_v6, bool do_sync) : chunk_size_(chunk_size), info_(std::move(info)), io_(std::move(io)), factory_(std::move(factory)), send_buffer_(std::move(sendbuf)) { + // Create sync handler if sync mode is requested + if (do_sync) { + sync_handler_ = std::make_unique(); + LOG_F(INFO, "TCP server initialized in synchronous (zero-copy) mode"); + } // assign connection-dependent fields info_->session_id(api_config::get_instance()->session_id()); info_->reset_uid(); @@ -178,6 +261,15 @@ tcp_server::tcp_server(stream_info_impl_p info, io_context_p io, send_buffer_p s throw std::runtime_error("Failed to instantiate socket acceptors for the TCP server"); } +tcp_server::~tcp_server() { + // sync_handler_ destructor will close all sync sockets +} + +void tcp_server::write_all_blocking(const std::vector &bufs) { + if (sync_handler_) { + sync_handler_->write_all_blocking(bufs); + } +} // === externally issued asynchronous commands === @@ -538,6 +630,18 @@ void client_session::handle_send_feedheader_outcome(err_t err, std::size_t n) { // convenient for unit tests if (max_buffered_ <= 0) return; + // If server is in sync mode, hand off the socket to the sync handler + if (serv->is_sync_mode()) { + LOG_F(INFO, "Handing off socket to sync handler for zero-copy transfer"); + auto protocol = sock_.local_endpoint().protocol(); + // Release the socket from this io_context and add to sync handler + // See https://stackoverflow.com/q/52671836/73299 + serv->sync_handler_->add_socket(sock_.release(), protocol); + // Unregister this session since we're handing off the socket + serv->unregister_inflight_session(this); + return; + } + // determine transfer parameters auto queue = serv->send_buffer_->new_consumer(max_buffered_); diff --git a/src/tcp_server.h b/src/tcp_server.h index 37314a15f..6fa63ccc9 100644 --- a/src/tcp_server.h +++ b/src/tcp_server.h @@ -3,11 +3,13 @@ #include "forward.h" #include "socket_utils.h" +#include #include #include #include #include #include +#include using asio::ip::tcp; using err_t = const asio::error_code &; @@ -19,6 +21,9 @@ using tcp_socket_p = std::shared_ptr; /// shared pointer to an acceptor socket using tcp_acceptor_p = std::unique_ptr; +/// Forward declaration for synchronous write handler +class sync_write_handler; + /** * The TCP data server. * @@ -47,9 +52,13 @@ class tcp_server : public std::enable_shared_from_this { * @param protocol The protocol (IPv4 or IPv6) that shall be serviced by this server. * @param chunk_size The preferred chunk size, in samples. If 0, the pushthrough flag determines * the effective chunking. + * @param do_sync If true, use synchronous (blocking) socket writes for zero-copy transfer. */ tcp_server(stream_info_impl_p info, io_context_p io, send_buffer_p sendbuf, factory_p factory, - int chunk_size, bool allow_v4, bool allow_v6); + int chunk_size, bool allow_v4, bool allow_v6, bool do_sync = false); + + /// Destructor (must be defined in .cpp due to unique_ptr to incomplete type) + ~tcp_server(); /** * Begin serving TCP connections. @@ -67,6 +76,16 @@ class tcp_server : public std::enable_shared_from_this { */ void end_serving(); + /** + * Write buffers to all connected sync sockets (blocking). + * Only valid when the server was created with do_sync=true. + * @param bufs Vector of const_buffers to write (gather-write). + */ + void write_all_blocking(const std::vector &bufs); + + /// Check if this server is in sync mode + bool is_sync_mode() const { return sync_handler_ != nullptr; } + private: friend class client_session; @@ -102,6 +121,9 @@ class tcp_server : public std::enable_shared_from_this { // some cached data std::string shortinfo_msg_; // pre-computed short-info server response std::string fullinfo_msg_; // pre-computed full-info server response + + // synchronous write handler (only set when do_sync=true) + std::unique_ptr sync_handler_; }; } // namespace lsl From b0af5bc379aa6d0fda7ea5830e5c1a6928c5150f Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Sun, 18 Jan 2026 13:20:34 -0500 Subject: [PATCH 02/11] Add endianness support (native + byte-swapped socket groups) --- src/tcp_server.cpp | 167 +++++++++++++++++++++++++++++++++++---------- 1 file changed, 132 insertions(+), 35 deletions(-) diff --git a/src/tcp_server.cpp b/src/tcp_server.cpp index c4b1ba690..4057d2681 100644 --- a/src/tcp_server.cpp +++ b/src/tcp_server.cpp @@ -16,6 +16,7 @@ #include #include #include +#include #include #include #include @@ -64,36 +65,99 @@ namespace lsl { * Handler for synchronous (blocking) writes to all connected clients. * This class manages sockets that have been handed off from client_session * for zero-copy synchronous data transfer. + * + * Supports clients with different endianness by maintaining two socket groups: + * - sockets_native_: clients with same endianness (zero-copy writes) + * - sockets_swapped_: clients needing byte-swapped data (one copy for all) */ class sync_write_handler { public: - sync_write_handler() : io_ctx_(1) {} + /** + * @param sample_bytes Total bytes per sample (channel_count * value_size) + * @param value_size Bytes per channel value (for byte-swapping) + * @param num_channels Number of channels in the stream + */ + sync_write_handler(std::size_t sample_bytes, std::size_t value_size, uint32_t num_channels) + : io_ctx_(1), sample_bytes_(sample_bytes), value_size_(value_size), + num_channels_(num_channels) { + // Pre-allocate scratch buffer for byte-swapping (sample + timestamp) + scratch_.resize(sample_bytes + sizeof(double)); + } ~sync_write_handler() { // Close all sockets - for (auto &sock : sockets_) { - if (sock && sock->is_open()) { - asio::error_code ec; - sock->close(ec); + auto close_sockets = [](std::vector &sockets) { + for (auto &sock : sockets) { + if (sock && sock->is_open()) { + asio::error_code ec; + sock->close(ec); + } } - } + }; + close_sockets(sockets_native_); + close_sockets(sockets_swapped_); } /// Add a socket for sync writes (called from client_session after handshake) - void add_socket(tcp_socket::native_handle_type handle, tcp_socket::protocol_type protocol) { + void add_socket(tcp_socket::native_handle_type handle, tcp_socket::protocol_type protocol, + bool reverse_byte_order) { std::lock_guard lock(mutex_); - sockets_.push_back(std::make_unique(io_ctx_, protocol, handle)); - LOG_F(INFO, "Added sync socket, now have %zu consumers", sockets_.size()); + auto sock = std::make_unique(io_ctx_, protocol, handle); + if (reverse_byte_order) { + sockets_swapped_.push_back(std::move(sock)); + LOG_F(INFO, "Added sync socket (swapped endian), now have %zu native + %zu swapped", + sockets_native_.size(), sockets_swapped_.size()); + } else { + sockets_native_.push_back(std::move(sock)); + LOG_F(INFO, "Added sync socket (native endian), now have %zu native + %zu swapped", + sockets_native_.size(), sockets_swapped_.size()); + } } /// Write buffers to all connected sockets (blocking gather-write) void write_all_blocking(const std::vector &bufs) { std::lock_guard lock(mutex_); - if (sockets_.empty()) return; + if (sockets_native_.empty() && sockets_swapped_.empty()) return; bool any_broken = false; - for (auto &sock : sockets_) { + // Write to same-endianness clients (zero-copy) + if (!sockets_native_.empty()) { + any_broken |= write_to_group(sockets_native_, bufs); + } + + // Write to reverse-endianness clients (one copy, byte-swapped) + if (!sockets_swapped_.empty()) { + auto swapped_bufs = swap_buffers(bufs); + any_broken |= write_to_group(sockets_swapped_, swapped_bufs); + } + + // Remove closed sockets from both groups + if (any_broken) { + auto remove_closed = [](std::vector &sockets) { + sockets.erase(std::remove_if(sockets.begin(), sockets.end(), + [](const tcp_socket_p &s) { return !s || !s->is_open(); }), + sockets.end()); + }; + remove_closed(sockets_native_); + remove_closed(sockets_swapped_); + LOG_F(INFO, "After cleanup, have %zu native + %zu swapped sync consumers", + sockets_native_.size(), sockets_swapped_.size()); + } + } + + /// Check if there are any connected consumers + bool have_consumers() const { + std::lock_guard lock(mutex_); + return !sockets_native_.empty() || !sockets_swapped_.empty(); + } + +private: + /// Write buffers to a group of sockets, returns true if any socket was broken + bool write_to_group(std::vector &sockets, + const std::vector &bufs) { + bool any_broken = false; + for (auto &sock : sockets) { if (!sock || !sock->is_open()) continue; asio::error_code ec; @@ -105,37 +169,66 @@ class sync_write_handler { case asio::error::connection_reset: case asio::error::not_connected: LOG_F(WARNING, "Sync socket disconnected: %s", ec.message().c_str()); - sock->close(ec); - any_broken = true; break; - default: - LOG_F(ERROR, "Sync write error: %s", ec.message().c_str()); - sock->close(ec); - any_broken = true; + default: LOG_F(ERROR, "Sync write error: %s", ec.message().c_str()); } + sock->close(ec); + any_broken = true; } } - - // Remove closed sockets - if (any_broken) { - sockets_.erase( - std::remove_if(sockets_.begin(), sockets_.end(), - [](const tcp_socket_p &s) { return !s || !s->is_open(); }), - sockets_.end()); - LOG_F(INFO, "After cleanup, have %zu sync consumers", sockets_.size()); - } + return any_broken; } - /// Check if there are any connected consumers - bool have_consumers() const { - std::lock_guard lock(mutex_); - return !sockets_.empty(); + /// Byte-swap buffers for reverse-endianness clients + /// Buffer structure: [tag:1][timestamp:8][sample:N] repeated + std::vector swap_buffers(const std::vector &bufs) { + swapped_data_.clear(); + std::vector result; + result.reserve(bufs.size()); + + for (size_t i = 0; i < bufs.size(); ++i) { + const auto &buf = bufs[i]; + size_t size = buf.size(); + const char *data = static_cast(buf.data()); + + if (size == 1) { + // Tag byte - no swap needed, pass through directly + result.push_back(buf); + } else if (size == sizeof(double)) { + // Timestamp - swap as double + size_t offset = swapped_data_.size(); + swapped_data_.resize(offset + sizeof(double)); + std::memcpy(swapped_data_.data() + offset, data, sizeof(double)); + endian_reverse_inplace(*reinterpret_cast(swapped_data_.data() + offset)); + result.push_back(asio::const_buffer(swapped_data_.data() + offset, sizeof(double))); + } else if (size == sample_bytes_) { + // Sample data - swap each value + size_t offset = swapped_data_.size(); + swapped_data_.resize(offset + sample_bytes_); + std::memcpy(swapped_data_.data() + offset, data, sample_bytes_); + sample::convert_endian( + swapped_data_.data() + offset, num_channels_, static_cast(value_size_)); + result.push_back(asio::const_buffer(swapped_data_.data() + offset, sample_bytes_)); + } else { + // Unknown buffer size - pass through (shouldn't happen) + LOG_F(WARNING, "Unexpected buffer size %zu in sync write", size); + result.push_back(buf); + } + } + return result; } -private: asio::io_context io_ctx_; - std::vector sockets_; + std::vector sockets_native_; // same endianness as server + std::vector sockets_swapped_; // need byte-swapped data mutable std::mutex mutex_; + + // For byte-swapping + std::size_t sample_bytes_; // total bytes per sample + std::size_t value_size_; // bytes per channel value + uint32_t num_channels_; // number of channels + std::vector scratch_; // pre-allocated scratch buffer (unused, kept for potential future use) + std::vector swapped_data_; // storage for byte-swapped data }; class client_session : public std::enable_shared_from_this { @@ -226,7 +319,10 @@ tcp_server::tcp_server(stream_info_impl_p info, io_context_p io, send_buffer_p s factory_(std::move(factory)), send_buffer_(std::move(sendbuf)) { // Create sync handler if sync mode is requested if (do_sync) { - sync_handler_ = std::make_unique(); + std::size_t sample_bytes = factory_->datasize(); + std::size_t value_size = format_sizes[info_->channel_format()]; + uint32_t num_channels = info_->channel_count(); + sync_handler_ = std::make_unique(sample_bytes, value_size, num_channels); LOG_F(INFO, "TCP server initialized in synchronous (zero-copy) mode"); } // assign connection-dependent fields @@ -632,11 +728,12 @@ void client_session::handle_send_feedheader_outcome(err_t err, std::size_t n) { // If server is in sync mode, hand off the socket to the sync handler if (serv->is_sync_mode()) { - LOG_F(INFO, "Handing off socket to sync handler for zero-copy transfer"); + LOG_F(INFO, "Handing off socket to sync handler for zero-copy transfer (reverse_byte_order=%d)", + reverse_byte_order_); auto protocol = sock_.local_endpoint().protocol(); // Release the socket from this io_context and add to sync handler // See https://stackoverflow.com/q/52671836/73299 - serv->sync_handler_->add_socket(sock_.release(), protocol); + serv->sync_handler_->add_socket(sock_.release(), protocol, reverse_byte_order_); // Unregister this session since we're handing off the socket serv->unregister_inflight_session(this); return; From bbf3b5a0d7c4078703af323a1cb52ac9f71ceed9 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Sun, 18 Jan 2026 13:39:06 -0500 Subject: [PATCH 03/11] Added nullptr check to stream_outlet constructor. This makes the wrapper consistent with stream_info and properly throws on construction failure. --- include/lsl_cpp.h | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/include/lsl_cpp.h b/include/lsl_cpp.h index 4b2fde345..f249d4596 100644 --- a/include/lsl_cpp.h +++ b/include/lsl_cpp.h @@ -409,7 +409,9 @@ class stream_outlet { lsl_transport_options_t flags = transp_default) : channel_count(info.channel_count()), sample_rate(info.nominal_srate()), obj(lsl_create_outlet_ex(info.handle().get(), chunk_size, max_buffered, flags), - &lsl_destroy_outlet) {} + &lsl_destroy_outlet) { + if (obj == nullptr) throw std::invalid_argument(lsl_last_error()); + } // ======================================== // === Pushing a sample into the outlet === From 1860d9004637bbcac29e8f43b092d5a549c75441 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Sun, 18 Jan 2026 22:37:29 -0500 Subject: [PATCH 04/11] Improve sync outlet: fix consumer detection, optimize chunk push - Fix have_consumers()/wait_for_consumers() to detect sync consumers - Handle DEDUCED_TIMESTAMP in sync mode for proper chunk timing - Change sync_timestamps_ to deque to prevent pointer invalidation - Add optimized enqueue_chunk_sync() for batched chunk transfers - Add have_sync_consumers() to tcp_server - Add sync outlet tests and benchmark tool --- examples/BenchmarkSyncVsAsync.cpp | 262 ++++++++++++++++++++++++++++++ examples/CMakeLists.txt | 2 + src/stream_outlet_impl.cpp | 76 ++++++++- src/stream_outlet_impl.h | 23 ++- src/tcp_server.cpp | 4 + src/tcp_server.h | 3 + testing/CMakeLists.txt | 2 + testing/ext/sync_outlet.cpp | 205 +++++++++++++++++++++++ testing/int/sync_endian.cpp | 198 ++++++++++++++++++++++ 9 files changed, 763 insertions(+), 12 deletions(-) create mode 100644 examples/BenchmarkSyncVsAsync.cpp create mode 100644 testing/ext/sync_outlet.cpp create mode 100644 testing/int/sync_endian.cpp diff --git a/examples/BenchmarkSyncVsAsync.cpp b/examples/BenchmarkSyncVsAsync.cpp new file mode 100644 index 000000000..a820b4fc1 --- /dev/null +++ b/examples/BenchmarkSyncVsAsync.cpp @@ -0,0 +1,262 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +/** + * Benchmark comparing synchronous (zero-copy) vs asynchronous outlet performance. + * + * Measures: + * - Push latency: Time for push_sample() to return + * - CPU time: User and system CPU time consumed + * - Throughput: Samples pushed per second + * + * Usage: BenchmarkSyncVsAsync [num_channels] [num_samples] [num_consumers] [sample_rate] [chunk_size] + * Default: 64 channels, 10000 samples, 1 consumer, 0 (unlimited), 1 (push_sample) + * Use sample_rate > 0 to pace the benchmark (e.g., 1000 for 1 kHz) + * Use chunk_size > 1 to test push_chunk instead of push_sample + */ + +struct Stats { + double min_us, max_us, mean_us, median_us, stddev_us; + double total_ms; + double throughput; // samples/sec + double cpu_user_ms; // user CPU time in ms + double cpu_system_ms; // system CPU time in ms +}; + +// Get current CPU time (user, system) for this process in ms +std::pair get_cpu_time_ms() { + struct rusage usage; + getrusage(RUSAGE_SELF, &usage); + double user_ms = usage.ru_utime.tv_sec * 1000.0 + usage.ru_utime.tv_usec / 1000.0; + double sys_ms = usage.ru_stime.tv_sec * 1000.0 + usage.ru_stime.tv_usec / 1000.0; + return {user_ms, sys_ms}; +} + +Stats compute_stats(std::vector &latencies_us, double total_time_ms, int num_samples, + double cpu_user_ms, double cpu_system_ms) { + Stats s{}; + if (latencies_us.empty()) return s; + + std::sort(latencies_us.begin(), latencies_us.end()); + s.min_us = latencies_us.front(); + s.max_us = latencies_us.back(); + s.median_us = latencies_us[latencies_us.size() / 2]; + + double sum = std::accumulate(latencies_us.begin(), latencies_us.end(), 0.0); + s.mean_us = sum / latencies_us.size(); + + double sq_sum = 0; + for (double v : latencies_us) { sq_sum += (v - s.mean_us) * (v - s.mean_us); } + s.stddev_us = std::sqrt(sq_sum / latencies_us.size()); + + s.total_ms = total_time_ms; + s.throughput = num_samples / (total_time_ms / 1000.0); + s.cpu_user_ms = cpu_user_ms; + s.cpu_system_ms = cpu_system_ms; + + return s; +} + +void print_stats(const char *label, const Stats &s, int nsamples) { + std::cout << std::fixed << std::setprecision(2); + std::cout << label << ":\n"; + std::cout << " Latency (us): min=" << s.min_us << ", max=" << s.max_us << ", mean=" << s.mean_us + << ", median=" << s.median_us << ", stddev=" << s.stddev_us << "\n"; + std::cout << " Wall time: " << s.total_ms << " ms, Throughput: " << std::setprecision(0) + << s.throughput << " samples/sec\n"; + double total_cpu = s.cpu_user_ms + s.cpu_system_ms; + double cpu_per_sample_us = (total_cpu * 1000.0) / nsamples; + std::cout << std::setprecision(2); + std::cout << " CPU time: " << total_cpu << " ms (user: " << s.cpu_user_ms + << ", sys: " << s.cpu_system_ms << "), " << cpu_per_sample_us << " us/sample\n"; +} + +// Consumer thread: pulls samples until signaled to stop +void consumer_thread(const std::string &stream_name, std::atomic &running, + std::atomic &samples_received) { + try { + auto found = lsl::resolve_stream("name", stream_name, 1, 10.0); + if (found.empty()) { + std::cout << " [Consumer] ERROR: Could not find stream " << stream_name << "\n" << std::flush; + return; + } + std::cout << " [Consumer] Found stream, opening..." << std::flush; + lsl::stream_inlet inlet(found[0]); + inlet.open_stream(5.0); + std::cout << " opened.\n" << std::flush; + + int nchannels = inlet.info().channel_count(); + std::vector sample(nchannels); + + while (running) { + double ts = inlet.pull_sample(sample, 0.1); + if (ts != 0.0) { samples_received++; } + } + + // Drain remaining samples + while (inlet.pull_sample(sample, 0.01) != 0.0) { samples_received++; } + } catch (std::exception &e) { std::cerr << "Consumer error: " << e.what() << "\n"; } +} + +Stats run_benchmark(const std::string &name, int nchannels, int nsamples, int nconsumers, + lsl_transport_options_t flags, double sample_rate = 0, int chunk_size = 1) { + // Create outlet + double nominal_rate = sample_rate > 0 ? sample_rate : lsl::IRREGULAR_RATE; + lsl::stream_info info(name, "Benchmark", nchannels, nominal_rate, lsl::cf_float32, name); + lsl::stream_outlet outlet(info, 0, 360, flags); + + // Start consumer threads + std::atomic running{true}; + std::vector> samples_received(nconsumers); + std::vector consumers; + + for (int i = 0; i < nconsumers; i++) { + samples_received[i] = 0; + consumers.emplace_back(consumer_thread, name, std::ref(running), std::ref(samples_received[i])); + } + + // Wait for consumers to connect + std::cout << " Waiting for " << nconsumers << " consumer(s)..." << std::flush; + while (!outlet.wait_for_consumers(1.0)) { std::cout << "." << std::flush; } + std::cout << " connected!\n" << std::flush; + // Give sockets time to be handed off (for sync mode) + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + + // Prepare sample/chunk buffer + std::vector chunk_buf(nchannels * chunk_size); + for (int c = 0; c < nchannels * chunk_size; c++) { chunk_buf[c] = static_cast(c % nchannels); } + + // Run benchmark + std::vector latencies_us; + int num_pushes = (nsamples + chunk_size - 1) / chunk_size; // ceiling division + latencies_us.reserve(num_pushes); + + // Calculate pacing interval if sample_rate is specified + std::chrono::nanoseconds chunk_interval_ns{0}; + if (sample_rate > 0) { + chunk_interval_ns = std::chrono::nanoseconds(static_cast(1e9 * chunk_size / sample_rate)); + std::cout << " Pushing " << nsamples << " samples"; + if (chunk_size > 1) std::cout << " (chunks of " << chunk_size << ")"; + std::cout << " @ " << sample_rate << " Hz..." << std::flush; + } else { + std::cout << " Pushing " << nsamples << " samples"; + if (chunk_size > 1) std::cout << " (chunks of " << chunk_size << ")"; + std::cout << " (max speed)..." << std::flush; + } + + // Measure CPU time before and after + auto [cpu_user_start, cpu_sys_start] = get_cpu_time_ms(); + auto start = std::chrono::high_resolution_clock::now(); + auto next_chunk_time = start; + + int samples_pushed = 0; + while (samples_pushed < nsamples) { + // Pace if sample_rate is set + if (sample_rate > 0) { + std::this_thread::sleep_until(next_chunk_time); + next_chunk_time += chunk_interval_ns; + } + + // Determine actual chunk size for this push (may be smaller for last chunk) + int this_chunk = std::min(chunk_size, nsamples - samples_pushed); + + auto t0 = std::chrono::high_resolution_clock::now(); + if (this_chunk == 1) { + outlet.push_sample(chunk_buf.data()); + } else { + outlet.push_chunk_multiplexed(chunk_buf.data(), this_chunk * nchannels); + } + auto t1 = std::chrono::high_resolution_clock::now(); + + double latency_us = std::chrono::duration(t1 - t0).count(); + latencies_us.push_back(latency_us); + samples_pushed += this_chunk; + } + + auto end = std::chrono::high_resolution_clock::now(); + auto [cpu_user_end, cpu_sys_end] = get_cpu_time_ms(); + + double total_ms = std::chrono::duration(end - start).count(); + double cpu_user_ms = cpu_user_end - cpu_user_start; + double cpu_sys_ms = cpu_sys_end - cpu_sys_start; + std::cout << " done.\n" << std::flush; + + // Stop consumers + running = false; + for (auto &t : consumers) { t.join(); } + + // Report received samples + int total_received = 0; + for (int i = 0; i < nconsumers; i++) { total_received += samples_received[i].load(); } + std::cout << " Consumers received: " << total_received << "/" << (nsamples * nconsumers) + << " samples\n" << std::flush; + + return compute_stats(latencies_us, total_ms, nsamples, cpu_user_ms, cpu_sys_ms); +} + +int main(int argc, char *argv[]) { + int nchannels = argc > 1 ? std::atoi(argv[1]) : 64; + int nsamples = argc > 2 ? std::atoi(argv[2]) : 10000; + int nconsumers = argc > 3 ? std::atoi(argv[3]) : 1; + double sample_rate = argc > 4 ? std::atof(argv[4]) : 0; // 0 = unlimited + int chunk_size = argc > 5 ? std::atoi(argv[5]) : 1; // 1 = push_sample + + std::cout << "=== LSL Sync vs Async Outlet Benchmark ===\n"; + std::cout << "Channels: " << nchannels << ", Samples: " << nsamples + << ", Consumers: " << nconsumers; + if (sample_rate > 0) { + std::cout << ", Rate: " << sample_rate << " Hz"; + } + if (chunk_size > 1) { + std::cout << ", Chunk: " << chunk_size; + } + std::cout << "\n"; + std::cout << "Sample size: " << (nchannels * sizeof(float)) << " bytes\n\n" << std::flush; + + // Run async benchmark + std::cout << "Running ASYNC benchmark...\n"; + Stats async_stats = run_benchmark("BenchAsync", nchannels, nsamples, nconsumers, transp_default, sample_rate, chunk_size); + print_stats("ASYNC", async_stats, nsamples); + std::cout << "\n"; + + // Delay between tests for cleanup (outlets need time to fully shut down) + std::cout << "Waiting for cleanup..." << std::flush; + std::this_thread::sleep_for(std::chrono::seconds(2)); + std::cout << " done.\n" << std::flush; + + // Run sync benchmark + std::cout << "Running SYNC benchmark...\n"; + Stats sync_stats = + run_benchmark("BenchSync", nchannels, nsamples, nconsumers, transp_sync_blocking, sample_rate, chunk_size); + print_stats("SYNC", sync_stats, nsamples); + std::cout << "\n"; + + // Summary comparison + std::cout << "=== Summary ===\n"; + std::cout << std::fixed << std::setprecision(2); + + double async_cpu_total = async_stats.cpu_user_ms + async_stats.cpu_system_ms; + double sync_cpu_total = sync_stats.cpu_user_ms + sync_stats.cpu_system_ms; + double async_cpu_per_sample = (async_cpu_total * 1000.0) / nsamples; + double sync_cpu_per_sample = (sync_cpu_total * 1000.0) / nsamples; + + std::cout << "CPU per sample: ASYNC=" << async_cpu_per_sample << " us, SYNC=" << sync_cpu_per_sample + << " us (ratio: " << (sync_cpu_per_sample / async_cpu_per_sample) << "x)\n"; + std::cout << "Latency: ASYNC=" << async_stats.mean_us << " us, SYNC=" << sync_stats.mean_us + << " us (ratio: " << (sync_stats.mean_us / async_stats.mean_us) << "x)\n"; + std::cout << "Throughput: ASYNC=" << std::setprecision(0) << async_stats.throughput + << ", SYNC=" << sync_stats.throughput + << " samples/sec (ratio: " << std::setprecision(2) + << (sync_stats.throughput / async_stats.throughput) << "x)\n"; + + return 0; +} diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index 9b2e70d28..13a68621d 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -93,8 +93,10 @@ addlslexample(SendStringMarkers cpp) addlslexample(SendStringMarkersC c) addlslexample(SendDataSyncBlocking cpp) addlslexample(TestSyncWithoutData cpp) +addlslexample(BenchmarkSyncVsAsync cpp) target_link_libraries(TestSyncWithoutData PRIVATE Threads::Threads) +target_link_libraries(BenchmarkSyncVsAsync PRIVATE Threads::Threads) # Windows doesn't have RPATH so we put the dll into the same directory as the executable. if(WIN32) diff --git a/src/stream_outlet_impl.cpp b/src/stream_outlet_impl.cpp index 5068cfc56..b95c20177 100644 --- a/src/stream_outlet_impl.cpp +++ b/src/stream_outlet_impl.cpp @@ -173,9 +173,23 @@ void stream_outlet_impl::push_numeric_raw(const void *data, double timestamp, bo } } -bool stream_outlet_impl::have_consumers() { return send_buffer_->have_consumers(); } +bool stream_outlet_impl::have_consumers() { + // Check both async consumers (via send_buffer) and sync consumers (via tcp_server) + return send_buffer_->have_consumers() || tcp_server_->have_sync_consumers(); +} bool stream_outlet_impl::wait_for_consumers(double timeout) { + // For sync mode, we need to poll since sync consumers aren't registered with send_buffer + if (sync_mode_) { + auto start = std::chrono::steady_clock::now(); + double elapsed = 0.0; + while (elapsed < timeout) { + if (tcp_server_->have_sync_consumers()) return true; + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + elapsed = std::chrono::duration(std::chrono::steady_clock::now() - start).count(); + } + return tcp_server_->have_sync_consumers(); + } return send_buffer_->wait_for_consumers(timeout); } @@ -208,12 +222,18 @@ template void stream_outlet_impl::enqueue(const std::string *data, void stream_outlet_impl::push_timestamp_sync(double timestamp) { // Allocate storage for timestamp tag + value in sync_timestamps_ - // Use TAG_TRANSMITTED_TIMESTAMP (2) for explicit timestamps - sync_timestamps_.emplace_back(TAG_TRANSMITTED_TIMESTAMP, timestamp); - auto &ts_entry = sync_timestamps_.back(); - // Add buffers for tag and timestamp - sync_buffers_.push_back(asio::const_buffer(&ts_entry.first, 1)); // tag byte - sync_buffers_.push_back(asio::const_buffer(&ts_entry.second, sizeof(double))); + if (timestamp == DEDUCED_TIMESTAMP) { + // Deduced timestamp: just send the 1-byte tag, no timestamp value + sync_timestamps_.emplace_back(TAG_DEDUCED_TIMESTAMP, 0.0); + auto &ts_entry = sync_timestamps_.back(); + sync_buffers_.push_back(asio::const_buffer(&ts_entry.first, 1)); // tag byte only + } else { + // Explicit timestamp: send tag + 8-byte timestamp value + sync_timestamps_.emplace_back(TAG_TRANSMITTED_TIMESTAMP, timestamp); + auto &ts_entry = sync_timestamps_.back(); + sync_buffers_.push_back(asio::const_buffer(&ts_entry.first, 1)); // tag byte + sync_buffers_.push_back(asio::const_buffer(&ts_entry.second, sizeof(double))); + } } void stream_outlet_impl::flush_sync() { @@ -236,4 +256,46 @@ void stream_outlet_impl::enqueue_sync(asio::const_buffer buf, double timestamp, } } +template +void stream_outlet_impl::enqueue_chunk_sync( + const T *data, std::size_t num_samples, double timestamp, bool pushthrough) { + if (lsl::api_config::get_instance()->force_default_timestamps()) timestamp = 0.0; + if (timestamp == 0.0) timestamp = lsl_clock(); + + std::size_t sample_bytes = sample_factory_->datasize(); + std::size_t num_chans = info_->channel_count(); + + // Reserve space upfront to avoid reallocations in sync_buffers_ + // Each sample needs: 1-2 buffers for timestamp + 1 buffer for data + // Note: sync_timestamps_ is a deque (no reserve) to ensure pointer stability + sync_buffers_.reserve(sync_buffers_.size() + num_samples * 3); + + // First sample: explicit timestamp + push_timestamp_sync(timestamp); + sync_buffers_.push_back(asio::const_buffer(data, sample_bytes)); + + // Remaining samples: deduced timestamps, contiguous data + for (std::size_t k = 1; k < num_samples; k++) { + push_timestamp_sync(DEDUCED_TIMESTAMP); + sync_buffers_.push_back( + asio::const_buffer(data + k * num_chans, sample_bytes)); + } + + if (pushthrough) flush_sync(); +} + +// Explicit template instantiations for enqueue_chunk_sync +template void stream_outlet_impl::enqueue_chunk_sync( + const char *, std::size_t, double, bool); +template void stream_outlet_impl::enqueue_chunk_sync( + const int16_t *, std::size_t, double, bool); +template void stream_outlet_impl::enqueue_chunk_sync( + const int32_t *, std::size_t, double, bool); +template void stream_outlet_impl::enqueue_chunk_sync( + const int64_t *, std::size_t, double, bool); +template void stream_outlet_impl::enqueue_chunk_sync( + const float *, std::size_t, double, bool); +template void stream_outlet_impl::enqueue_chunk_sync( + const double *, std::size_t, double, bool); + } // namespace lsl diff --git a/src/stream_outlet_impl.h b/src/stream_outlet_impl.h index 00d9bcff8..a704bfd87 100644 --- a/src/stream_outlet_impl.h +++ b/src/stream_outlet_impl.h @@ -6,10 +6,12 @@ #include "stream_info_impl.h" #include #include +#include #include #include #include #include +#include #include #include @@ -256,10 +258,15 @@ class stream_outlet_impl { if (timestamp == 0.0) timestamp = lsl_clock(); if (info().nominal_srate() != IRREGULAR_RATE) timestamp = timestamp - (num_samples - 1) / info().nominal_srate(); - push_sample(buffer, timestamp, pushthrough && (num_samples == 1)); - for (std::size_t k = 1; k < num_samples; k++) - push_sample(&buffer[k * num_chans], DEDUCED_TIMESTAMP, - pushthrough && (k == num_samples - 1)); + // Use optimized sync path for non-string types in sync mode + if (sync_mode_ && !std::is_same::value) { + enqueue_chunk_sync(buffer, num_samples, timestamp, pushthrough); + } else { + push_sample(buffer, timestamp, pushthrough && (num_samples == 1)); + for (std::size_t k = 1; k < num_samples; k++) + push_sample(&buffer[k * num_chans], DEDUCED_TIMESTAMP, + pushthrough && (k == num_samples - 1)); + } } } @@ -317,6 +324,11 @@ class stream_outlet_impl { /// Enqueue a buffer for sync transfer (single sample) void enqueue_sync(asio::const_buffer buf, double timestamp, bool pushthrough); + /// Enqueue a chunk for sync transfer (optimized for multiple samples) + template + void enqueue_chunk_sync( + const T *data, std::size_t num_samples, double timestamp, bool pushthrough); + /** * Check whether some given number of channels matches the stream's channel_count. * Throws an error if not. @@ -355,7 +367,8 @@ class stream_outlet_impl { /// Buffers accumulated for sync gather-write std::vector sync_buffers_; /// Storage for timestamps in sync mode (tag + timestamp pairs) - std::vector> sync_timestamps_; + /// Using deque instead of vector to ensure pointers remain valid when adding elements + std::deque> sync_timestamps_; }; } // namespace lsl diff --git a/src/tcp_server.cpp b/src/tcp_server.cpp index 4057d2681..3dd841350 100644 --- a/src/tcp_server.cpp +++ b/src/tcp_server.cpp @@ -367,6 +367,10 @@ void tcp_server::write_all_blocking(const std::vector &bufs) } } +bool tcp_server::have_sync_consumers() const { + return sync_handler_ && sync_handler_->have_consumers(); +} + // === externally issued asynchronous commands === void tcp_server::begin_serving() { diff --git a/src/tcp_server.h b/src/tcp_server.h index 6fa63ccc9..94bf52bdb 100644 --- a/src/tcp_server.h +++ b/src/tcp_server.h @@ -86,6 +86,9 @@ class tcp_server : public std::enable_shared_from_this { /// Check if this server is in sync mode bool is_sync_mode() const { return sync_handler_ != nullptr; } + /// Check if there are any sync consumers connected (only valid if is_sync_mode()) + bool have_sync_consumers() const; + private: friend class client_session; diff --git a/testing/CMakeLists.txt b/testing/CMakeLists.txt index 90928601f..9f1df4f74 100644 --- a/testing/CMakeLists.txt +++ b/testing/CMakeLists.txt @@ -71,6 +71,7 @@ add_executable(lsl_test_exported ext/discovery.cpp ext/move.cpp ext/streaminfo.cpp + ext/sync_outlet.cpp ext/time.cpp ) target_link_libraries(lsl_test_exported PRIVATE lsl common catch_main) @@ -84,6 +85,7 @@ set(LSL_TEST_INTERNAL_SRCS int/samples.cpp int/postproc.cpp int/serialization_v100.cpp + int/sync_endian.cpp int/tcpserver.cpp int/sendbuffer.cpp ) diff --git a/testing/ext/sync_outlet.cpp b/testing/ext/sync_outlet.cpp new file mode 100644 index 000000000..adc78a95e --- /dev/null +++ b/testing/ext/sync_outlet.cpp @@ -0,0 +1,205 @@ +#include "../common/create_streampair.hpp" +#include "../common/lsltypes.hpp" +#include +#include +#include +#include +#include +#include + +// clazy:excludeall=non-pod-global-static + +/// Helper to create a sync outlet + inlet pair +inline Streampair create_sync_streampair(const lsl::stream_info &info) { + lsl::stream_outlet out(info, 0, 360, transp_sync_blocking); + auto found_stream_info = lsl::resolve_stream("name", info.name(), 1, 5.0); + if (found_stream_info.empty()) throw std::runtime_error("sync outlet not found"); + lsl::stream_inlet in(found_stream_info[0]); + + in.open_stream(2); + out.wait_for_consumers(2); + return Streampair(std::move(out), std::move(in)); +} + +TEST_CASE("sync_outlet_basic", "[sync][basic]") { + const int nchannels = 4; + const int nsamples = 100; + + auto sp = create_sync_streampair( + lsl::stream_info("SyncBasic", "Test", nchannels, 100, lsl::cf_float32, "sync_basic")); + + std::vector send_buf(nchannels); + std::vector recv_buf(nchannels); + + for (int i = 0; i < nsamples; ++i) { + // Fill with test pattern + for (int c = 0; c < nchannels; ++c) { send_buf[c] = static_cast(i * nchannels + c); } + + double send_ts = lsl::local_clock(); + sp.out_.push_sample(send_buf, send_ts, true); + + double recv_ts = sp.in_.pull_sample(recv_buf, 2.0); + REQUIRE(recv_ts != 0.0); + + for (int c = 0; c < nchannels; ++c) { + CHECK(recv_buf[c] == Catch::Approx(send_buf[c])); + } + } +} + +TEMPLATE_TEST_CASE( + "sync_outlet_datatypes", "[sync][datatransfer]", char, int16_t, int32_t, int64_t, float, double) { + const int nchannels = 2; + const int nsamples = 32; + const char *name = SampleType::fmt_string(); + auto cf = static_cast(SampleType::chan_fmt); + + lsl::stream_info info(std::string("Sync_") + name, "Test", nchannels, 100, cf, "sync_dtype"); + auto sp = create_sync_streampair(info); + + TestType send_buf[nchannels]; + TestType recv_buf[nchannels]; + + // Test with shifting bit pattern to exercise all bits + send_buf[0] = 1; + for (int i = 0; i < nsamples; ++i) { + send_buf[1] = static_cast(-send_buf[0] + 1); + + sp.out_.push_sample(send_buf, lsl::local_clock(), true); + double ts = sp.in_.pull_sample(recv_buf, 2.0); + REQUIRE(ts != 0.0); + + CHECK(recv_buf[0] == Catch::Approx(send_buf[0])); + CHECK(recv_buf[1] == Catch::Approx(send_buf[1])); + + send_buf[0] = static_cast(static_cast(send_buf[0]) << 1); + if (send_buf[0] == 0) send_buf[0] = 1; // Reset if we shifted out all bits + } +} + +TEST_CASE("sync_outlet_string_rejected", "[sync][validation]") { + // Sync mode should reject string-format streams + lsl::stream_info info("SyncString", "Test", 1, 100, lsl::cf_string, "sync_string"); + + CHECK_THROWS_AS( + lsl::stream_outlet(info, 0, 360, transp_sync_blocking), std::invalid_argument); +} + +TEST_CASE("sync_outlet_push_chunk", "[sync][chunk]") { + const int nchannels = 4; + const int chunk_size = 10; + + auto sp = create_sync_streampair( + lsl::stream_info("SyncChunk", "Test", nchannels, 100, lsl::cf_float32, "sync_chunk")); + + // Create chunk data (multiplexed: ch0_s0, ch1_s0, ch2_s0, ch3_s0, ch0_s1, ...) + std::vector chunk(nchannels * chunk_size); + for (int s = 0; s < chunk_size; ++s) { + for (int c = 0; c < nchannels; ++c) { + chunk[s * nchannels + c] = static_cast(s * 100 + c); + } + } + + // Push chunk with single timestamp - this uses DEDUCED_TIMESTAMP for samples 2-N + sp.out_.push_chunk_multiplexed(chunk, lsl::local_clock(), true); + + // Pull and verify all samples + std::vector recv_buf(nchannels); + for (int s = 0; s < chunk_size; ++s) { + double ts = sp.in_.pull_sample(recv_buf, 2.0); + INFO("Sample " << s << ": ts=" << ts); + REQUIRE(ts != 0.0); + for (int c = 0; c < nchannels; ++c) { + INFO(" ch" << c << ": got=" << recv_buf[c] << " expected=" << chunk[s * nchannels + c]); + CHECK(recv_buf[c] == Catch::Approx(chunk[s * nchannels + c])); + } + } +} + +TEST_CASE("sync_outlet_multi_consumer", "[sync][multi]") { + const int nchannels = 2; + const int nsamples = 50; + + lsl::stream_info info("SyncMulti", "Test", nchannels, 100, lsl::cf_float32, "sync_multi"); + lsl::stream_outlet out(info, 0, 360, transp_sync_blocking); + + // Resolve and connect two inlets + auto found = lsl::resolve_stream("name", "SyncMulti", 1, 5.0); + REQUIRE(!found.empty()); + + lsl::stream_inlet in1(found[0]); + lsl::stream_inlet in2(found[0]); + + in1.open_stream(2); + in2.open_stream(2); + out.wait_for_consumers(2); + + // Give sockets time to be handed off to sync handler + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + std::vector send_buf(nchannels); + std::vector recv_buf1(nchannels); + std::vector recv_buf2(nchannels); + + for (int i = 0; i < nsamples; ++i) { + send_buf[0] = static_cast(i); + send_buf[1] = static_cast(-i); + + out.push_sample(send_buf, lsl::local_clock(), true); + + double ts1 = in1.pull_sample(recv_buf1, 2.0); + double ts2 = in2.pull_sample(recv_buf2, 2.0); + + REQUIRE(ts1 != 0.0); + REQUIRE(ts2 != 0.0); + + CHECK(recv_buf1[0] == Catch::Approx(send_buf[0])); + CHECK(recv_buf1[1] == Catch::Approx(send_buf[1])); + CHECK(recv_buf2[0] == Catch::Approx(send_buf[0])); + CHECK(recv_buf2[1] == Catch::Approx(send_buf[1])); + } +} + +TEST_CASE("sync_outlet_consumer_disconnect", "[sync][disconnect]") { + const int nchannels = 2; + + lsl::stream_info info( + "SyncDisconnect", "Test", nchannels, 100, lsl::cf_float32, "sync_disconnect"); + lsl::stream_outlet out(info, 0, 360, transp_sync_blocking); + + { + // Create inlet in a scope so it gets destroyed + auto found = lsl::resolve_stream("name", "SyncDisconnect", 1, 5.0); + REQUIRE(!found.empty()); + + lsl::stream_inlet in(found[0]); + in.open_stream(2); + out.wait_for_consumers(2); + + // Give socket time to be handed off + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + // Send some samples successfully + std::vector buf = {1.0f, 2.0f}; + out.push_sample(buf, lsl::local_clock(), true); + + std::vector recv(nchannels); + CHECK(in.pull_sample(recv, 2.0) != 0.0); + } + // Inlet is now destroyed + + // Give time for disconnect to propagate + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + // Outlet should handle the disconnected consumer gracefully + // (no crash, no hang - just returns quickly with no consumers) + // Note: First write(s) might succeed (OS buffers), so we do multiple writes + // with delays to ensure the broken pipe error is eventually detected. + std::vector buf = {3.0f, 4.0f}; + for (int i = 0; i < 5 && out.have_consumers(); ++i) { + out.push_sample(buf, lsl::local_clock(), true); + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + } + + CHECK(!out.have_consumers()); +} diff --git a/testing/int/sync_endian.cpp b/testing/int/sync_endian.cpp new file mode 100644 index 000000000..47efc0797 --- /dev/null +++ b/testing/int/sync_endian.cpp @@ -0,0 +1,198 @@ +#include "sample.h" +#include "util/endian.hpp" +#include +#include +#include + +// clazy:excludeall=non-pod-global-static + +// Test the byte-swapping logic used by sync_write_handler + +TEST_CASE("endian_reverse_inplace", "[sync][endian]") { + SECTION("16-bit") { + int16_t val = 0x0102; + lsl::endian_reverse_inplace(val); + CHECK(val == 0x0201); + } + + SECTION("32-bit") { + int32_t val = 0x01020304; + lsl::endian_reverse_inplace(val); + CHECK(val == 0x04030201); + } + + SECTION("64-bit") { + int64_t val = 0x0102030405060708LL; + lsl::endian_reverse_inplace(val); + CHECK(val == 0x0807060504030201LL); + } + + SECTION("float") { + // Test that float byte-swap is reversible + float original = 3.14159f; + float val = original; + lsl::endian_reverse_inplace(val); + CHECK(val != original); // Should be different after swap + lsl::endian_reverse_inplace(val); + CHECK(val == original); // Should be same after double-swap + } + + SECTION("double") { + double original = 3.141592653589793; + double val = original; + lsl::endian_reverse_inplace(val); + CHECK(val != original); + lsl::endian_reverse_inplace(val); + CHECK(val == original); + } +} + +TEST_CASE("sample::convert_endian", "[sync][endian]") { + SECTION("int16 array") { + int16_t data[] = {0x0102, 0x0304, 0x0506, 0x0708}; + lsl::sample::convert_endian(data, 4, sizeof(int16_t)); + CHECK(data[0] == 0x0201); + CHECK(data[1] == 0x0403); + CHECK(data[2] == 0x0605); + CHECK(data[3] == 0x0807); + } + + SECTION("int32 array") { + int32_t data[] = {0x01020304, 0x05060708}; + lsl::sample::convert_endian(data, 2, sizeof(int32_t)); + CHECK(data[0] == 0x04030201); + CHECK(data[1] == 0x08070605); + } + + SECTION("float array") { + float original[] = {1.0f, 2.0f, 3.0f, 4.0f}; + float data[4]; + std::memcpy(data, original, sizeof(data)); + + lsl::sample::convert_endian(data, 4, sizeof(float)); + // After swap, values should be different (garbage floats) + for (int i = 0; i < 4; ++i) { CHECK(data[i] != original[i]); } + + // Swap back, values should match original + lsl::sample::convert_endian(data, 4, sizeof(float)); + for (int i = 0; i < 4; ++i) { CHECK(data[i] == original[i]); } + } + + SECTION("double array") { + double original[] = {1.0, 2.0, 3.0, 4.0}; + double data[4]; + std::memcpy(data, original, sizeof(data)); + + lsl::sample::convert_endian(data, 4, sizeof(double)); + for (int i = 0; i < 4; ++i) { CHECK(data[i] != original[i]); } + + lsl::sample::convert_endian(data, 4, sizeof(double)); + for (int i = 0; i < 4; ++i) { CHECK(data[i] == original[i]); } + } + + SECTION("1-byte no-op") { + // 1-byte values should not be modified + char data[] = {0x01, 0x02, 0x03, 0x04}; + char original[] = {0x01, 0x02, 0x03, 0x04}; + lsl::sample::convert_endian(data, 4, 1); + CHECK(std::memcmp(data, original, 4) == 0); + } +} + +TEST_CASE("sync buffer swap simulation", "[sync][endian]") { + // Simulate the buffer structure that sync_write_handler processes: + // [tag:1][timestamp:8][sample:N] repeated + // This tests the logic without needing actual sockets + + const uint32_t num_channels = 4; + const uint32_t value_size = sizeof(float); + const std::size_t sample_bytes = num_channels * value_size; + + // Create test buffers mimicking the sync path structure + struct TestBuffer { + const char *data; + std::size_t size; + }; + + // Tag byte + char tag = 0x02; // TAG_TRANSMITTED + + // Timestamp + double timestamp = 1234567890.123456; + double ts_copy = timestamp; + + // Sample data (4 floats) + float sample_data[] = {1.0f, 2.0f, 3.0f, 4.0f}; + float sample_copy[4]; + std::memcpy(sample_copy, sample_data, sizeof(sample_copy)); + + std::vector bufs = {{&tag, 1}, + {reinterpret_cast(&ts_copy), sizeof(double)}, + {reinterpret_cast(sample_copy), sample_bytes}}; + + // Simulate swap_buffers logic + std::vector swapped_data; + for (const auto &buf : bufs) { + if (buf.size == 1) { + // Tag byte - no swap, would just pass through + CHECK(buf.data[0] == tag); + } else if (buf.size == sizeof(double)) { + // Timestamp - swap as double + size_t offset = swapped_data.size(); + swapped_data.resize(offset + sizeof(double)); + std::memcpy(swapped_data.data() + offset, buf.data, sizeof(double)); + lsl::endian_reverse_inplace(*reinterpret_cast(swapped_data.data() + offset)); + + // Verify it's different from original + double swapped_ts; + std::memcpy(&swapped_ts, swapped_data.data() + offset, sizeof(double)); + CHECK(swapped_ts != timestamp); + } else if (buf.size == sample_bytes) { + // Sample data - swap each value + size_t offset = swapped_data.size(); + swapped_data.resize(offset + sample_bytes); + std::memcpy(swapped_data.data() + offset, buf.data, sample_bytes); + lsl::sample::convert_endian(swapped_data.data() + offset, num_channels, value_size); + + // Verify values are different from original + float *swapped_samples = reinterpret_cast(swapped_data.data() + offset); + for (uint32_t i = 0; i < num_channels; ++i) { + CHECK(swapped_samples[i] != sample_data[i]); + } + } + } + + // Now verify that swapping back recovers original values + // Swap timestamp back + lsl::endian_reverse_inplace(*reinterpret_cast(swapped_data.data())); + double recovered_ts; + std::memcpy(&recovered_ts, swapped_data.data(), sizeof(double)); + CHECK(recovered_ts == timestamp); + + // Swap samples back + lsl::sample::convert_endian(swapped_data.data() + sizeof(double), num_channels, value_size); + float *recovered_samples = reinterpret_cast(swapped_data.data() + sizeof(double)); + for (uint32_t i = 0; i < num_channels; ++i) { + CHECK(recovered_samples[i] == sample_data[i]); + } +} + +TEST_CASE("can_convert_endian", "[sync][endian]") { + // 1-byte values are always convertible + CHECK(lsl::can_convert_endian(lsl::LSL_LITTLE_ENDIAN, 1)); + CHECK(lsl::can_convert_endian(lsl::LSL_BIG_ENDIAN, 1)); + + // Standard endianness should be convertible for multi-byte values + CHECK(lsl::can_convert_endian(lsl::LSL_LITTLE_ENDIAN, 2)); + CHECK(lsl::can_convert_endian(lsl::LSL_LITTLE_ENDIAN, 4)); + CHECK(lsl::can_convert_endian(lsl::LSL_LITTLE_ENDIAN, 8)); + CHECK(lsl::can_convert_endian(lsl::LSL_BIG_ENDIAN, 2)); + CHECK(lsl::can_convert_endian(lsl::LSL_BIG_ENDIAN, 4)); + CHECK(lsl::can_convert_endian(lsl::LSL_BIG_ENDIAN, 8)); + + // Exotic endianness should not be convertible for multi-byte values + CHECK_FALSE(lsl::can_convert_endian(lsl::LSL_PORTABLE_ENDIAN, 2)); + CHECK_FALSE(lsl::can_convert_endian(lsl::LSL_LITTLE_ENDIAN_BUT_BIG_FLOAT, 4)); + CHECK_FALSE(lsl::can_convert_endian(lsl::LSL_BIG_ENDIAN_BUT_LITTLE_FLOAT, 4)); + CHECK_FALSE(lsl::can_convert_endian(lsl::LSL_PDP11, 2)); +} From 30616231acb3f8593cb73aa9689911bb3447a8b8 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Sun, 18 Jan 2026 23:59:20 -0500 Subject: [PATCH 05/11] 1. Removed lsl:: prefix from transp_sync_blocking (it's a C enum, not in the namespace) 2. Added #include for std::sort --- examples/BenchmarkSyncVsAsync.cpp | 1 + examples/SendDataSyncBlocking.cpp | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/examples/BenchmarkSyncVsAsync.cpp b/examples/BenchmarkSyncVsAsync.cpp index a820b4fc1..c74ebb2a3 100644 --- a/examples/BenchmarkSyncVsAsync.cpp +++ b/examples/BenchmarkSyncVsAsync.cpp @@ -1,3 +1,4 @@ +#include #include #include #include diff --git a/examples/SendDataSyncBlocking.cpp b/examples/SendDataSyncBlocking.cpp index c379dbbce..3689aecc9 100644 --- a/examples/SendDataSyncBlocking.cpp +++ b/examples/SendDataSyncBlocking.cpp @@ -32,7 +32,7 @@ int main(int argc, char *argv[]) { // Create outlet with transp_sync_blocking flag for zero-copy transfer // Note: The third parameter (max_buffered) is less important in sync mode // since data goes directly to the socket without intermediate buffering. - lsl::stream_outlet outlet(info, 0, 360, lsl::transp_sync_blocking); + lsl::stream_outlet outlet(info, 0, 360, transp_sync_blocking); std::cout << "Waiting for consumers...\n"; while (!outlet.wait_for_consumers(5)) { From e94ea6cd9bc33cf0b06d826a2b856061ec4dae61 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Mon, 19 Jan 2026 00:24:24 -0500 Subject: [PATCH 06/11] if constexpr prevents the compiler from instantiating enqueue_chunk_sync, which resolves the Windows linker error --- src/stream_outlet_impl.h | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/src/stream_outlet_impl.h b/src/stream_outlet_impl.h index a704bfd87..6d004c383 100644 --- a/src/stream_outlet_impl.h +++ b/src/stream_outlet_impl.h @@ -259,14 +259,18 @@ class stream_outlet_impl { if (info().nominal_srate() != IRREGULAR_RATE) timestamp = timestamp - (num_samples - 1) / info().nominal_srate(); // Use optimized sync path for non-string types in sync mode - if (sync_mode_ && !std::is_same::value) { - enqueue_chunk_sync(buffer, num_samples, timestamp, pushthrough); - } else { - push_sample(buffer, timestamp, pushthrough && (num_samples == 1)); - for (std::size_t k = 1; k < num_samples; k++) - push_sample(&buffer[k * num_chans], DEDUCED_TIMESTAMP, - pushthrough && (k == num_samples - 1)); + // if constexpr prevents instantiation of enqueue_chunk_sync + if constexpr (!std::is_same::value) { + if (sync_mode_) { + enqueue_chunk_sync(buffer, num_samples, timestamp, pushthrough); + return; + } } + // Async path (or string types which don't support sync mode) + push_sample(buffer, timestamp, pushthrough && (num_samples == 1)); + for (std::size_t k = 1; k < num_samples; k++) + push_sample(&buffer[k * num_chans], DEDUCED_TIMESTAMP, + pushthrough && (k == num_samples - 1)); } } From d0cf7e9c3f54030bc6ac755b905395341c475c27 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Mon, 19 Jan 2026 00:30:10 -0500 Subject: [PATCH 07/11] Added Windows support using GetProcessTimes() with proper #ifdef _WIN32 guards. --- examples/BenchmarkSyncVsAsync.cpp | 23 ++++++++++++++++++++++- 1 file changed, 22 insertions(+), 1 deletion(-) diff --git a/examples/BenchmarkSyncVsAsync.cpp b/examples/BenchmarkSyncVsAsync.cpp index c74ebb2a3..dc32570df 100644 --- a/examples/BenchmarkSyncVsAsync.cpp +++ b/examples/BenchmarkSyncVsAsync.cpp @@ -7,10 +7,16 @@ #include #include #include -#include #include #include +#ifdef _WIN32 +#define WIN32_LEAN_AND_MEAN +#include +#else +#include +#endif + /** * Benchmark comparing synchronous (zero-copy) vs asynchronous outlet performance. * @@ -35,11 +41,26 @@ struct Stats { // Get current CPU time (user, system) for this process in ms std::pair get_cpu_time_ms() { +#ifdef _WIN32 + FILETIME creation, exit, kernel, user; + if (GetProcessTimes(GetCurrentProcess(), &creation, &exit, &kernel, &user)) { + // FILETIME is in 100-nanosecond intervals + auto to_ms = [](const FILETIME &ft) { + ULARGE_INTEGER li; + li.LowPart = ft.dwLowDateTime; + li.HighPart = ft.dwHighDateTime; + return static_cast(li.QuadPart) / 10000.0; // 100ns -> ms + }; + return {to_ms(user), to_ms(kernel)}; + } + return {0.0, 0.0}; +#else struct rusage usage; getrusage(RUSAGE_SELF, &usage); double user_ms = usage.ru_utime.tv_sec * 1000.0 + usage.ru_utime.tv_usec / 1000.0; double sys_ms = usage.ru_stime.tv_sec * 1000.0 + usage.ru_stime.tv_usec / 1000.0; return {user_ms, sys_ms}; +#endif } Stats compute_stats(std::vector &latencies_us, double total_time_ms, int num_samples, From 7f8b97cfcc6c94bcdef8a68e1ed09d7ff528c0f1 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Mon, 19 Jan 2026 00:34:58 -0500 Subject: [PATCH 08/11] 1. Added NOMINMAX to prevent Windows min/max macro conflicts 2. Replaced C++17 structured bindings with .first/.second pair access for C++11/14 compatibility --- examples/BenchmarkSyncVsAsync.cpp | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/examples/BenchmarkSyncVsAsync.cpp b/examples/BenchmarkSyncVsAsync.cpp index dc32570df..524da29b2 100644 --- a/examples/BenchmarkSyncVsAsync.cpp +++ b/examples/BenchmarkSyncVsAsync.cpp @@ -12,6 +12,7 @@ #ifdef _WIN32 #define WIN32_LEAN_AND_MEAN +#define NOMINMAX #include #else #include @@ -176,7 +177,7 @@ Stats run_benchmark(const std::string &name, int nchannels, int nsamples, int nc } // Measure CPU time before and after - auto [cpu_user_start, cpu_sys_start] = get_cpu_time_ms(); + auto cpu_start = get_cpu_time_ms(); auto start = std::chrono::high_resolution_clock::now(); auto next_chunk_time = start; @@ -205,11 +206,11 @@ Stats run_benchmark(const std::string &name, int nchannels, int nsamples, int nc } auto end = std::chrono::high_resolution_clock::now(); - auto [cpu_user_end, cpu_sys_end] = get_cpu_time_ms(); + auto cpu_end = get_cpu_time_ms(); double total_ms = std::chrono::duration(end - start).count(); - double cpu_user_ms = cpu_user_end - cpu_user_start; - double cpu_sys_ms = cpu_sys_end - cpu_sys_start; + double cpu_user_ms = cpu_end.first - cpu_start.first; + double cpu_sys_ms = cpu_end.second - cpu_start.second; std::cout << " done.\n" << std::flush; // Stop consumers From 04f74322f718d3e3d76136221e609a886a8a0d73 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 17 Jun 2026 00:47:51 -0400 Subject: [PATCH 09/11] sync outlet: fix cross-endian byte-swap, force-flush, document semantics Addresses review findings on the synchronous (zero-copy) outlet: #1 Memory safety: the byte-swap path grew its scratch buffer with resize() inside the loop while holding pointers into it, so a reallocation dangled earlier const_buffers and sent freed memory to byte-swapped clients. The new sync_swap_buffers() reserves the exact size up front. #2 Correctness: buffers were classified by size, so an 8-byte sample (e.g. 2x int32 or 4x int16) was mistaken for a timestamp and reversed as one double instead of per channel value. Swapping is now driven by the sample tag, not buffer size, so it is correct for any sample_bytes. #3 Portability: the per-sample tag was stored in a uint64_t and sent via its first byte, which is 0x00 on a big-endian host. It is now a uint8_t (sync_ts_entry), so the correct tag byte goes on the wire everywhere. #4 Lifetime: sync mode now flushes on every push and ignores pushthrough. The gather buffers alias the caller's memory, so deferring the write (pushthrough == false) would retain dangling pointers once the caller reuses the buffer. #5/#7 Docs: document sync mode as single-producer (the push path is unsynchronized) and note that disconnected consumers are detected lazily. Also removes the unused scratch_ member and extracts the swap into an inline, unit-testable helper (src/sync_serialization.h). New internal tests drive that helper across the 8-byte-collision and many-sample (reallocation) cases that no little-endian integration test can reach. Co-Authored-By: Claude Opus 4.8 (1M context) --- include/lsl/common.h | 8 +++- src/stream_outlet_impl.cpp | 38 +++++++++------- src/stream_outlet_impl.h | 32 ++++++++++---- src/sync_serialization.h | 88 +++++++++++++++++++++++++++++++++++++ src/tcp_server.cpp | 47 +++----------------- src/tcp_server.h | 4 +- testing/int/sync_endian.cpp | 86 ++++++++++++++++++++++++++++++++++++ 7 files changed, 233 insertions(+), 70 deletions(-) create mode 100644 src/sync_serialization.h diff --git a/include/lsl/common.h b/include/lsl/common.h index 89ce83b97..74cc6bddb 100644 --- a/include/lsl/common.h +++ b/include/lsl/common.h @@ -162,9 +162,13 @@ typedef enum { transp_bufsize_thousandths = 2, /// Use synchronous (blocking) socket writes for zero-copy data transfer. - /// When enabled, push_sample/push_chunk will block until data is written to all consumers. + /// When enabled, push_sample/push_chunk write the caller's buffer directly to every + /// connected consumer and block until the data has been handed to the OS for all of them. /// Reduces CPU usage for high-bandwidth streams at the cost of increased call latency. - /// Not compatible with string-format streams. + /// Notes: + /// - Not compatible with string-format streams (variable-size samples). + /// - Single-producer: push from only one thread at a time (the sync path is unsynchronized). + /// - The pushthrough flag is ignored; every push sends immediately (no internal buffering). transp_sync_blocking = 4, // prevent compilers from assuming an instance fits in a single byte diff --git a/src/stream_outlet_impl.cpp b/src/stream_outlet_impl.cpp index b95c20177..62f9a38fc 100644 --- a/src/stream_outlet_impl.cpp +++ b/src/stream_outlet_impl.cpp @@ -221,18 +221,20 @@ template void stream_outlet_impl::enqueue(const std::string *data, // === Sync mode implementation === void stream_outlet_impl::push_timestamp_sync(double timestamp) { - // Allocate storage for timestamp tag + value in sync_timestamps_ + // Encode one sample's timestamp into the wire format and append gather buffers that + // reference the stable deque storage. The tag is a single byte (sync_ts_entry::tag), + // so its address transmits the correct byte regardless of host endianness. if (timestamp == DEDUCED_TIMESTAMP) { - // Deduced timestamp: just send the 1-byte tag, no timestamp value - sync_timestamps_.emplace_back(TAG_DEDUCED_TIMESTAMP, 0.0); - auto &ts_entry = sync_timestamps_.back(); - sync_buffers_.push_back(asio::const_buffer(&ts_entry.first, 1)); // tag byte only + // Deduced timestamp: just the 1-byte tag, no timestamp value + sync_timestamps_.push_back({TAG_DEDUCED_TIMESTAMP, 0.0}); + auto &e = sync_timestamps_.back(); + sync_buffers_.push_back(asio::const_buffer(&e.tag, 1)); } else { - // Explicit timestamp: send tag + 8-byte timestamp value - sync_timestamps_.emplace_back(TAG_TRANSMITTED_TIMESTAMP, timestamp); - auto &ts_entry = sync_timestamps_.back(); - sync_buffers_.push_back(asio::const_buffer(&ts_entry.first, 1)); // tag byte - sync_buffers_.push_back(asio::const_buffer(&ts_entry.second, sizeof(double))); + // Transmitted timestamp: 1-byte tag + 8-byte timestamp value + sync_timestamps_.push_back({TAG_TRANSMITTED_TIMESTAMP, timestamp}); + auto &e = sync_timestamps_.back(); + sync_buffers_.push_back(asio::const_buffer(&e.tag, 1)); + sync_buffers_.push_back(asio::const_buffer(&e.timestamp, sizeof(double))); } } @@ -245,20 +247,20 @@ void stream_outlet_impl::flush_sync() { sync_timestamps_.clear(); } -void stream_outlet_impl::enqueue_sync(asio::const_buffer buf, double timestamp, bool pushthrough) { +void stream_outlet_impl::enqueue_sync(asio::const_buffer buf, double timestamp, bool /*pushthrough*/) { // Add timestamp push_timestamp_sync(timestamp); // Add sample data buffer (points to user's buffer - zero copy!) sync_buffers_.push_back(buf); - // Flush if pushthrough is requested - if (pushthrough) { - flush_sync(); - } + // Sync mode always sends before returning, ignoring pushthrough: the gather buffers alias + // the caller's buffer, so deferring the write would leave dangling pointers once the caller + // reuses or frees that memory. + flush_sync(); } template void stream_outlet_impl::enqueue_chunk_sync( - const T *data, std::size_t num_samples, double timestamp, bool pushthrough) { + const T *data, std::size_t num_samples, double timestamp, bool /*pushthrough*/) { if (lsl::api_config::get_instance()->force_default_timestamps()) timestamp = 0.0; if (timestamp == 0.0) timestamp = lsl_clock(); @@ -281,7 +283,9 @@ void stream_outlet_impl::enqueue_chunk_sync( asio::const_buffer(data + k * num_chans, sample_bytes)); } - if (pushthrough) flush_sync(); + // Always send the whole chunk before returning (see enqueue_sync): the gather buffers + // alias the caller's contiguous data, so the write must complete while it is still valid. + flush_sync(); } // Explicit template instantiations for enqueue_chunk_sync diff --git a/src/stream_outlet_impl.h b/src/stream_outlet_impl.h index 6d004c383..e8fb3116a 100644 --- a/src/stream_outlet_impl.h +++ b/src/stream_outlet_impl.h @@ -258,8 +258,9 @@ class stream_outlet_impl { if (timestamp == 0.0) timestamp = lsl_clock(); if (info().nominal_srate() != IRREGULAR_RATE) timestamp = timestamp - (num_samples - 1) / info().nominal_srate(); - // Use optimized sync path for non-string types in sync mode - // if constexpr prevents instantiation of enqueue_chunk_sync + // Use optimized sync path for non-string types in sync mode. Sync mode always + // sends immediately, so pushthrough is ignored (the whole chunk is one blocking + // gather-write). if constexpr prevents instantiation of enqueue_chunk_sync. if constexpr (!std::is_same::value) { if (sync_mode_) { enqueue_chunk_sync(buffer, num_samples, timestamp, pushthrough); @@ -325,10 +326,12 @@ class stream_outlet_impl { /// Flush sync_buffers_ to all connected consumers (blocking) void flush_sync(); - /// Enqueue a buffer for sync transfer (single sample) + /// Enqueue a buffer for sync transfer (single sample). Always flushes before returning; + /// the pushthrough argument is ignored (see enqueue_sync in the .cpp for why). void enqueue_sync(asio::const_buffer buf, double timestamp, bool pushthrough); - /// Enqueue a chunk for sync transfer (optimized for multiple samples) + /// Enqueue a chunk for sync transfer (optimized for multiple samples). Always flushes the + /// whole chunk before returning; the pushthrough argument is ignored. template void enqueue_chunk_sync( const T *data, std::size_t num_samples, double timestamp, bool pushthrough); @@ -365,14 +368,27 @@ class stream_outlet_impl { std::vector io_threads_; // === Sync mode members === + // + // Sync mode is single-producer: push_sample/push_chunk must be called from one thread + // at a time. Unlike the async path (which feeds a lock-free send_buffer), the sync path + // mutates sync_buffers_/sync_timestamps_ directly without locking, so concurrent pushes + // from multiple threads are a data race. + + /// One encoded timestamp for the sync wire format: a 1-byte tag optionally followed by + /// the 8-byte timestamp value. The tag is stored as a uint8_t (not packed into a wider + /// integer) so that sending its address transmits the correct byte on any endianness. + struct sync_ts_entry { + uint8_t tag; ///< TAG_DEDUCED_TIMESTAMP or TAG_TRANSMITTED_TIMESTAMP + double timestamp; ///< only sent when tag == TAG_TRANSMITTED_TIMESTAMP + }; /// Flag indicating sync (zero-copy blocking) mode is enabled bool sync_mode_{false}; - /// Buffers accumulated for sync gather-write + /// Buffers accumulated for sync gather-write (alias the caller's data and sync_timestamps_) std::vector sync_buffers_; - /// Storage for timestamps in sync mode (tag + timestamp pairs) - /// Using deque instead of vector to ensure pointers remain valid when adding elements - std::deque> sync_timestamps_; + /// Backing storage for the timestamp tags/values referenced by sync_buffers_. + /// A deque (not a vector) keeps element addresses stable as entries are appended. + std::deque sync_timestamps_; }; } // namespace lsl diff --git a/src/sync_serialization.h b/src/sync_serialization.h new file mode 100644 index 000000000..9097e5719 --- /dev/null +++ b/src/sync_serialization.h @@ -0,0 +1,88 @@ +#ifndef SYNC_SERIALIZATION_H +#define SYNC_SERIALIZATION_H + +#include "sample.h" +#include +#include +#include +#include +#include + +namespace lsl { + +/** + * Byte-swap a synchronous gather-write buffer sequence for a reverse-endianness client. + * + * The input is the on-the-wire layout produced by the sync outlet, one group per sample: + * [tag:1] ( [timestamp:8] iff tag == TAG_TRANSMITTED_TIMESTAMP ) [sample:sample_bytes] + * + * The sequence is walked by tag rather than by guessing from buffer sizes, so it stays + * correct even when a sample happens to be 8 bytes (e.g. 2x int32 or 4x int16), which a + * size-based classifier would mistake for a timestamp. Tag bytes pass through unchanged, + * timestamps are reversed as a single 8-byte value, and sample payloads are reversed per + * channel value (@p value_size bytes each; @p value_size == 1 is a no-op). + * + * Swapped bytes are written into @p storage, which is cleared and reserved to the exact + * required size up front so that the returned const_buffers referencing it remain valid + * for the lifetime of the returned vector (a growing vector would otherwise reallocate and + * dangle earlier pointers). + * + * @param bufs Gather buffers for one or more samples, in the layout described above. + * @param storage Scratch storage that backs the swapped buffers; must outlive the result. + * @param sample_bytes Total bytes per sample (channel_count * value_size). + * @param value_size Bytes per channel value (1, 2, 4 or 8). + * @param num_channels Number of channels per sample. + * @return const_buffers describing the byte-swapped stream; tag buffers alias @p bufs, the + * rest point into @p storage. + */ +inline std::vector sync_swap_buffers( + const std::vector &bufs, std::vector &storage, + std::size_t sample_bytes, std::size_t value_size, uint32_t num_channels) { + (void)sample_bytes; // implied by num_channels * value_size; kept for call-site clarity + + // Reserve the exact swapped byte count (everything except the 1-byte tags) before + // appending so that storage never reallocates and the returned pointers stay valid. + std::size_t needed = 0; + for (const auto &b : bufs) + if (b.size() != 1) needed += b.size(); + storage.clear(); + storage.reserve(needed); + + std::vector result; + result.reserve(bufs.size()); + + const auto swap_into = [&](const asio::const_buffer &buf, std::size_t width) { + const std::size_t offset = storage.size(); + const std::size_t n = buf.size(); + storage.resize(offset + n); // within reserved capacity -> no reallocation + std::memcpy(storage.data() + offset, buf.data(), n); + sample::convert_endian(storage.data() + offset, static_cast(n / width), + static_cast(width)); + result.push_back(asio::const_buffer(storage.data() + offset, n)); + }; + + std::size_t i = 0; + while (i < bufs.size()) { + // Each sample group starts with a 1-byte tag, which is never byte-swapped. + const asio::const_buffer &tagbuf = bufs[i]; + const uint8_t tag = *static_cast(tagbuf.data()); + result.push_back(tagbuf); + ++i; + // A transmitted-timestamp tag is followed by an 8-byte timestamp value. + if (tag == TAG_TRANSMITTED_TIMESTAMP && i < bufs.size()) { + swap_into(bufs[i], sizeof(double)); + ++i; + } + // Then the fixed-size sample payload, swapped per channel value. + if (i < bufs.size()) { + swap_into(bufs[i], value_size); + ++i; + } + } + (void)num_channels; + return result; +} + +} // namespace lsl + +#endif // SYNC_SERIALIZATION_H diff --git a/src/tcp_server.cpp b/src/tcp_server.cpp index 3dd841350..092b93dcb 100644 --- a/src/tcp_server.cpp +++ b/src/tcp_server.cpp @@ -5,6 +5,7 @@ #include "send_buffer.h" #include "socket_utils.h" #include "stream_info_impl.h" +#include "sync_serialization.h" #include "util/cast.hpp" #include "util/endian.hpp" #include "util/strfuns.hpp" @@ -79,10 +80,7 @@ class sync_write_handler { */ sync_write_handler(std::size_t sample_bytes, std::size_t value_size, uint32_t num_channels) : io_ctx_(1), sample_bytes_(sample_bytes), value_size_(value_size), - num_channels_(num_channels) { - // Pre-allocate scratch buffer for byte-swapping (sample + timestamp) - scratch_.resize(sample_bytes + sizeof(double)); - } + num_channels_(num_channels) {} ~sync_write_handler() { // Close all sockets @@ -179,43 +177,9 @@ class sync_write_handler { return any_broken; } - /// Byte-swap buffers for reverse-endianness clients - /// Buffer structure: [tag:1][timestamp:8][sample:N] repeated + /// Byte-swap buffers for reverse-endianness clients (see sync_swap_buffers). std::vector swap_buffers(const std::vector &bufs) { - swapped_data_.clear(); - std::vector result; - result.reserve(bufs.size()); - - for (size_t i = 0; i < bufs.size(); ++i) { - const auto &buf = bufs[i]; - size_t size = buf.size(); - const char *data = static_cast(buf.data()); - - if (size == 1) { - // Tag byte - no swap needed, pass through directly - result.push_back(buf); - } else if (size == sizeof(double)) { - // Timestamp - swap as double - size_t offset = swapped_data_.size(); - swapped_data_.resize(offset + sizeof(double)); - std::memcpy(swapped_data_.data() + offset, data, sizeof(double)); - endian_reverse_inplace(*reinterpret_cast(swapped_data_.data() + offset)); - result.push_back(asio::const_buffer(swapped_data_.data() + offset, sizeof(double))); - } else if (size == sample_bytes_) { - // Sample data - swap each value - size_t offset = swapped_data_.size(); - swapped_data_.resize(offset + sample_bytes_); - std::memcpy(swapped_data_.data() + offset, data, sample_bytes_); - sample::convert_endian( - swapped_data_.data() + offset, num_channels_, static_cast(value_size_)); - result.push_back(asio::const_buffer(swapped_data_.data() + offset, sample_bytes_)); - } else { - // Unknown buffer size - pass through (shouldn't happen) - LOG_F(WARNING, "Unexpected buffer size %zu in sync write", size); - result.push_back(buf); - } - } - return result; + return sync_swap_buffers(bufs, swapped_data_, sample_bytes_, value_size_, num_channels_); } asio::io_context io_ctx_; @@ -227,8 +191,7 @@ class sync_write_handler { std::size_t sample_bytes_; // total bytes per sample std::size_t value_size_; // bytes per channel value uint32_t num_channels_; // number of channels - std::vector scratch_; // pre-allocated scratch buffer (unused, kept for potential future use) - std::vector swapped_data_; // storage for byte-swapped data + std::vector swapped_data_; // backing storage for byte-swapped buffers }; class client_session : public std::enable_shared_from_this { diff --git a/src/tcp_server.h b/src/tcp_server.h index 94bf52bdb..efe13762e 100644 --- a/src/tcp_server.h +++ b/src/tcp_server.h @@ -86,7 +86,9 @@ class tcp_server : public std::enable_shared_from_this { /// Check if this server is in sync mode bool is_sync_mode() const { return sync_handler_ != nullptr; } - /// Check if there are any sync consumers connected (only valid if is_sync_mode()) + /// Check if there are any sync consumers connected (only valid if is_sync_mode()). + /// Note: disconnected consumers are detected lazily, on the next write_all_blocking that + /// fails to reach them, so this may briefly keep reporting a consumer that has gone away. bool have_sync_consumers() const; private: diff --git a/testing/int/sync_endian.cpp b/testing/int/sync_endian.cpp index 47efc0797..34fa9dc4f 100644 --- a/testing/int/sync_endian.cpp +++ b/testing/int/sync_endian.cpp @@ -1,6 +1,9 @@ #include "sample.h" +#include "sync_serialization.h" #include "util/endian.hpp" +#include #include +#include #include #include @@ -196,3 +199,86 @@ TEST_CASE("can_convert_endian", "[sync][endian]") { CHECK_FALSE(lsl::can_convert_endian(lsl::LSL_BIG_ENDIAN_BUT_LITTLE_FLOAT, 4)); CHECK_FALSE(lsl::can_convert_endian(lsl::LSL_PDP11, 2)); } + +// Exercise the real lsl::sync_swap_buffers on the wire layout the sync outlet produces: +// [tag:1] ([ts:8] iff transmitted) [sample:sample_bytes] repeated, first sample transmitted. +// This is the cross-endian path that no integration test reaches on a little-endian CI host. +template static void check_sync_swap_roundtrip(uint32_t nchan, int nsamples) { + const std::size_t value_size = sizeof(T); + const std::size_t sample_bytes = nchan * value_size; + const double ts0 = 1234.5678; + + // Build the host-endian wire bytes into one stable backing buffer, remembering the + // (offset, size) of each gather buffer so we can build const_buffers once it is final. + std::vector backing; + struct Seg { + std::size_t off, size; + }; + std::vector segs; + auto append = [&](const void *p, std::size_t n) { + const std::size_t off = backing.size(); + const char *cp = static_cast(p); + backing.insert(backing.end(), cp, cp + n); + segs.push_back({off, n}); + }; + + std::vector> values(nsamples, std::vector(nchan)); + for (int s = 0; s < nsamples; ++s) { + const uint8_t tag = (s == 0) ? lsl::TAG_TRANSMITTED_TIMESTAMP : lsl::TAG_DEDUCED_TIMESTAMP; + append(&tag, 1); + if (s == 0) append(&ts0, sizeof(double)); + for (uint32_t c = 0; c < nchan; ++c) + values[s][c] = static_cast((s + 1) * 1000 + c * 7 + 1); + append(values[s].data(), sample_bytes); + } + + std::vector bufs; + for (const auto &sg : segs) bufs.push_back(asio::const_buffer(backing.data() + sg.off, sg.size)); + + std::vector storage; + auto out = lsl::sync_swap_buffers(bufs, storage, sample_bytes, value_size, nchan); + REQUIRE(out.size() == bufs.size()); + + std::size_t bi = 0; + for (int s = 0; s < nsamples; ++s) { + // tag byte passes through unchanged + REQUIRE(out[bi].size() == 1); + CHECK(*static_cast(out[bi].data()) == + (s == 0 ? lsl::TAG_TRANSMITTED_TIMESTAMP : lsl::TAG_DEDUCED_TIMESTAMP)); + ++bi; + // transmitted timestamp is reversed as a single 8-byte value + if (s == 0) { + REQUIRE(out[bi].size() == sizeof(double)); + double exp = ts0; + lsl::endian_reverse_inplace(exp); + CHECK(std::memcmp(out[bi].data(), &exp, sizeof(double)) == 0); + ++bi; + } + // sample payload is reversed PER CHANNEL VALUE, not as one block — this is what the + // old size-based classifier got wrong when sample_bytes happened to equal 8. + REQUIRE(out[bi].size() == sample_bytes); + for (uint32_t c = 0; c < nchan; ++c) { + T exp = values[s][c]; + lsl::endian_reverse_inplace(exp); + const char *got = static_cast(out[bi].data()) + c * value_size; + INFO("sample " << s << " channel " << c); + CHECK(std::memcmp(got, &exp, value_size) == 0); + } + ++bi; + } +} + +TEST_CASE("sync_swap_buffers per-value swap and pointer stability", "[sync][endian]") { + // 2x int32 and 4x int16 both make sample_bytes == 8 (== sizeof(double)); the swap must + // still reverse each channel value independently rather than as one 8-byte timestamp. + SECTION("2x int32 (8-byte sample collides with timestamp size)") { + check_sync_swap_roundtrip(2, 16); + } + SECTION("4x int16 (8-byte sample)") { check_sync_swap_roundtrip(4, 16); } + // Genuine 8-byte values must keep working (reversed as one unit). + SECTION("1x double (8-byte value)") { check_sync_swap_roundtrip(1, 16); } + SECTION("1x int64 (8-byte value)") { check_sync_swap_roundtrip(1, 16); } + // Non-colliding sizes. + SECTION("3x float") { check_sync_swap_roundtrip(3, 16); } + SECTION("8x int16") { check_sync_swap_roundtrip(8, 16); } +} From e61ba87525ed15e9616b344ea39cae4d71618dbd Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 17 Jun 2026 01:22:32 -0400 Subject: [PATCH 10/11] sync outlet: bound blocking sends with a configurable send timeout (#6, #221) A synchronous (transp_sync_blocking) push blocks until every consumer accepts the sample, which provides the self-regulating backpressure requested in #221 (push runs only as fast as the slowest receiver drains) but could otherwise stall the producer forever on a dead/stalled consumer. Add a per-consumer send timeout, configurable via [tuning] SyncSendTimeout (seconds). It defaults to the reconnect watchdog threshold (WatchdogTimeThreshold, 15s) so a consumer that can't keep up for as long as a stalled connection is treated like one; 0 means block forever (previous behavior). No new exported symbols -- the knob is config-only. Implementation uses SO_SNDTIMEO on each handed-off sync socket (forcing blocking mode first), so the hot path stays a plain blocking asio::write with no per-push overhead. On expiry the write returns would_block/timed_out, which reuses the existing close-and-drop-consumer path; TCP framing means we cannot resume a half-written sample, so a timed-out consumer is disconnected. runtime_config test verifies SyncSendTimeout defaults to WatchdogTimeThreshold. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/api_config.cpp | 3 +++ src/api_config.h | 5 ++++ src/tcp_server.cpp | 43 ++++++++++++++++++++++++++++++---- testing/int/runtime_config.cpp | 6 ++++- 4 files changed, 52 insertions(+), 5 deletions(-) diff --git a/src/api_config.cpp b/src/api_config.cpp index 4aeb43950..63c6feeb9 100644 --- a/src/api_config.cpp +++ b/src/api_config.cpp @@ -301,6 +301,9 @@ void api_config::load(INI &pt) { LSL_PROTOCOL_VERSION, pt.get("tuning.UseProtocolVersion", LSL_PROTOCOL_VERSION)); watchdog_check_interval_ = pt.get("tuning.WatchdogCheckInterval", 15.0); watchdog_time_threshold_ = pt.get("tuning.WatchdogTimeThreshold", 15.0); + // Default the synchronous-outlet send timeout to the reconnect watchdog threshold: a sync + // consumer that can't keep up for that long is treated like a stalled connection. + sync_send_timeout_ = pt.get("tuning.SyncSendTimeout", watchdog_time_threshold_); multicast_min_rtt_ = pt.get("tuning.MulticastMinRTT", 0.5); multicast_max_rtt_ = pt.get("tuning.MulticastMaxRTT", 3.0); unicast_min_rtt_ = pt.get("tuning.UnicastMinRTT", 0.75); diff --git a/src/api_config.h b/src/api_config.h index fe1646725..06f722a48 100644 --- a/src/api_config.h +++ b/src/api_config.h @@ -185,6 +185,10 @@ class api_config { /// The watchdog takes no action if not at least this much time has passed since the last /// receipt of data. In seconds. double watchdog_time_threshold() const { return watchdog_time_threshold_; } + /// Send timeout for synchronous (transp_sync_blocking) outlets, in seconds. A blocking push + /// to a consumer that cannot accept the sample within this time disconnects that consumer + /// (it would otherwise stall the producer indefinitely). 0 means block forever. + double sync_send_timeout() const { return sync_send_timeout_; } /// The minimum assumed round-trip-time for a multicast query. Any subsequent packet wave would /// be started no earlier than this. double multicast_min_rtt() const { return multicast_min_rtt_; } @@ -285,6 +289,7 @@ class api_config { int use_protocol_version_; double watchdog_time_threshold_; double watchdog_check_interval_; + double sync_send_timeout_; double multicast_min_rtt_; double multicast_max_rtt_; double unicast_min_rtt_; diff --git a/src/tcp_server.cpp b/src/tcp_server.cpp index 092b93dcb..b3e4d39d9 100644 --- a/src/tcp_server.cpp +++ b/src/tcp_server.cpp @@ -77,10 +77,13 @@ class sync_write_handler { * @param sample_bytes Total bytes per sample (channel_count * value_size) * @param value_size Bytes per channel value (for byte-swapping) * @param num_channels Number of channels in the stream + * @param send_timeout Per-consumer blocking-send timeout in seconds; 0 means block forever. + * A consumer that can't accept a sample within this time is disconnected. */ - sync_write_handler(std::size_t sample_bytes, std::size_t value_size, uint32_t num_channels) + sync_write_handler(std::size_t sample_bytes, std::size_t value_size, uint32_t num_channels, + double send_timeout) : io_ctx_(1), sample_bytes_(sample_bytes), value_size_(value_size), - num_channels_(num_channels) {} + num_channels_(num_channels), send_timeout_(send_timeout) {} ~sync_write_handler() { // Close all sockets @@ -101,6 +104,7 @@ class sync_write_handler { bool reverse_byte_order) { std::lock_guard lock(mutex_); auto sock = std::make_unique(io_ctx_, protocol, handle); + apply_send_timeout(*sock); if (reverse_byte_order) { sockets_swapped_.push_back(std::move(sock)); LOG_F(INFO, "Added sync socket (swapped endian), now have %zu native + %zu swapped", @@ -151,6 +155,26 @@ class sync_write_handler { } private: + /// Apply the configured blocking-send timeout to a socket (no-op if disabled). SO_SNDTIMEO + /// only takes effect on a blocking socket, so we force blocking mode first. On expiry a + /// synchronous write returns try_again/would_block/timed_out, which write_to_group treats + /// as a disconnect. + void apply_send_timeout(tcp_socket &sock) const { + if (send_timeout_ <= 0) return; + asio::error_code ec; + sock.native_non_blocking(false, ec); +#if defined(_WIN32) + DWORD ms = static_cast(send_timeout_ * 1000.0); + ::setsockopt(sock.native_handle(), SOL_SOCKET, SO_SNDTIMEO, + reinterpret_cast(&ms), sizeof(ms)); +#else + struct timeval tv; + tv.tv_sec = static_cast(send_timeout_); + tv.tv_usec = static_cast((send_timeout_ - static_cast(tv.tv_sec)) * 1e6); + ::setsockopt(sock.native_handle(), SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)); +#endif + } + /// Write buffers to a group of sockets, returns true if any socket was broken bool write_to_group(std::vector &sockets, const std::vector &bufs) { @@ -168,6 +192,13 @@ class sync_write_handler { case asio::error::not_connected: LOG_F(WARNING, "Sync socket disconnected: %s", ec.message().c_str()); break; + // SO_SNDTIMEO expiry surfaces as would_block (== try_again on POSIX) or, on + // Windows, timed_out. + case asio::error::would_block: + case asio::error::timed_out: + LOG_F(WARNING, "Sync consumer too slow (send timed out after %.1fs), dropping it", + send_timeout_); + break; default: LOG_F(ERROR, "Sync write error: %s", ec.message().c_str()); } sock->close(ec); @@ -192,6 +223,7 @@ class sync_write_handler { std::size_t value_size_; // bytes per channel value uint32_t num_channels_; // number of channels std::vector swapped_data_; // backing storage for byte-swapped buffers + double send_timeout_; // per-consumer blocking-send timeout in seconds (0 = forever) }; class client_session : public std::enable_shared_from_this { @@ -285,8 +317,11 @@ tcp_server::tcp_server(stream_info_impl_p info, io_context_p io, send_buffer_p s std::size_t sample_bytes = factory_->datasize(); std::size_t value_size = format_sizes[info_->channel_format()]; uint32_t num_channels = info_->channel_count(); - sync_handler_ = std::make_unique(sample_bytes, value_size, num_channels); - LOG_F(INFO, "TCP server initialized in synchronous (zero-copy) mode"); + double send_timeout = api_config::get_instance()->sync_send_timeout(); + sync_handler_ = std::make_unique( + sample_bytes, value_size, num_channels, send_timeout); + LOG_F(INFO, "TCP server initialized in synchronous (zero-copy) mode (send timeout %.1fs)", + send_timeout); } // assign connection-dependent fields info_->session_id(api_config::get_instance()->session_id()); diff --git a/testing/int/runtime_config.cpp b/testing/int/runtime_config.cpp index 86209a3bf..83830ae6a 100644 --- a/testing/int/runtime_config.cpp +++ b/testing/int/runtime_config.cpp @@ -13,11 +13,15 @@ TEST_CASE("runtime config content overrides defaults", "[api_config][runtime_con "[ports]\n" "BasePort = 30000\n" "[tuning]\n" - "UseProtocolVersion = 100\n"); + "UseProtocolVersion = 100\n" + "WatchdogTimeThreshold = 7.5\n"); const auto *cfg = lsl::api_config::get_instance(); REQUIRE(cfg != nullptr); CHECK(cfg->session_id() == "runtime_config_test"); CHECK(cfg->base_port() == 30000); CHECK(cfg->use_protocol_version() == 100); + CHECK(cfg->watchdog_time_threshold() == 7.5); + // SyncSendTimeout was not set, so it defaults to the reconnect watchdog threshold. + CHECK(cfg->sync_send_timeout() == 7.5); } From 454b4e97aff99fda725096cf61be1f48309926ef Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 17 Jun 2026 13:57:06 -0400 Subject: [PATCH 11/11] sync outlet: alignment-safe byte-swap; add missing Address Copilot review on PR #256: - sync_swap_buffers(): byte-swap the gather-write buffers in place with byte operations instead of sample::convert_endian(). The scratch store is a std::vector with values packed at arbitrary offsets, so the typed dereferences in convert_endian() can be unaligned -- UB on strict-alignment targets. Swapping bytes directly avoids that; same wire output, async path untouched. - tcp_server.cpp: include for std::remove_if (was relying on a transitive include). - sync_endian.cpp: make the swap simulation test alignment-safe the same way and read values back via memcpy into aligned locals. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/sync_serialization.h | 10 ++++++++-- src/tcp_server.cpp | 1 + testing/int/sync_endian.cpp | 31 ++++++++++++++++++++++--------- 3 files changed, 31 insertions(+), 11 deletions(-) diff --git a/src/sync_serialization.h b/src/sync_serialization.h index 9097e5719..a4ac9a705 100644 --- a/src/sync_serialization.h +++ b/src/sync_serialization.h @@ -2,6 +2,7 @@ #define SYNC_SERIALIZATION_H #include "sample.h" +#include #include #include #include @@ -56,8 +57,13 @@ inline std::vector sync_swap_buffers( const std::size_t n = buf.size(); storage.resize(offset + n); // within reserved capacity -> no reallocation std::memcpy(storage.data() + offset, buf.data(), n); - sample::convert_endian(storage.data() + offset, static_cast(n / width), - static_cast(width)); + // Reverse each width-sized value in place using byte operations only. storage is a + // vector and values are packed at arbitrary offsets, so it may be unaligned for + // the typed dereferences in sample::convert_endian() -- undefined behavior on + // strict-alignment targets. Swapping bytes directly avoids that entirely. + if (width > 1) + for (char *p = storage.data() + offset, *end = p + n; p < end; p += width) + std::reverse(p, p + width); result.push_back(asio::const_buffer(storage.data() + offset, n)); }; diff --git a/src/tcp_server.cpp b/src/tcp_server.cpp index b3e4d39d9..c4173b55b 100644 --- a/src/tcp_server.cpp +++ b/src/tcp_server.cpp @@ -15,6 +15,7 @@ #include #include #include +#include #include #include #include diff --git a/testing/int/sync_endian.cpp b/testing/int/sync_endian.cpp index 34fa9dc4f..8b3c84ca5 100644 --- a/testing/int/sync_endian.cpp +++ b/testing/int/sync_endian.cpp @@ -1,6 +1,7 @@ #include "sample.h" #include "sync_serialization.h" #include "util/endian.hpp" +#include #include #include #include @@ -135,16 +136,30 @@ TEST_CASE("sync buffer swap simulation", "[sync][endian]") { // Simulate swap_buffers logic std::vector swapped_data; + + // Reverse each width-sized value in a byte range in place. Matches the production + // sync_swap_buffers() approach: pure byte operations, so it stays valid even though the + // std::vector backing store may be unaligned for the value type (typed dereferences + // would be undefined behavior on strict-alignment targets). + const auto swap_bytes = [](char *p, std::size_t n, std::size_t width) { + for (char *end = p + n; p < end; p += width) std::reverse(p, p + width); + }; + // Read a value out of the byte store via an aligned local (no typed dereference of store). + const auto read_float = [&](std::size_t offset) { + float v; + std::memcpy(&v, swapped_data.data() + offset, sizeof(float)); + return v; + }; for (const auto &buf : bufs) { if (buf.size == 1) { // Tag byte - no swap, would just pass through CHECK(buf.data[0] == tag); } else if (buf.size == sizeof(double)) { - // Timestamp - swap as double + // Timestamp - swap as a single 8-byte value size_t offset = swapped_data.size(); swapped_data.resize(offset + sizeof(double)); std::memcpy(swapped_data.data() + offset, buf.data, sizeof(double)); - lsl::endian_reverse_inplace(*reinterpret_cast(swapped_data.data() + offset)); + swap_bytes(swapped_data.data() + offset, sizeof(double), sizeof(double)); // Verify it's different from original double swapped_ts; @@ -155,28 +170,26 @@ TEST_CASE("sync buffer swap simulation", "[sync][endian]") { size_t offset = swapped_data.size(); swapped_data.resize(offset + sample_bytes); std::memcpy(swapped_data.data() + offset, buf.data, sample_bytes); - lsl::sample::convert_endian(swapped_data.data() + offset, num_channels, value_size); + swap_bytes(swapped_data.data() + offset, sample_bytes, value_size); // Verify values are different from original - float *swapped_samples = reinterpret_cast(swapped_data.data() + offset); for (uint32_t i = 0; i < num_channels; ++i) { - CHECK(swapped_samples[i] != sample_data[i]); + CHECK(read_float(offset + i * value_size) != sample_data[i]); } } } // Now verify that swapping back recovers original values // Swap timestamp back - lsl::endian_reverse_inplace(*reinterpret_cast(swapped_data.data())); + swap_bytes(swapped_data.data(), sizeof(double), sizeof(double)); double recovered_ts; std::memcpy(&recovered_ts, swapped_data.data(), sizeof(double)); CHECK(recovered_ts == timestamp); // Swap samples back - lsl::sample::convert_endian(swapped_data.data() + sizeof(double), num_channels, value_size); - float *recovered_samples = reinterpret_cast(swapped_data.data() + sizeof(double)); + swap_bytes(swapped_data.data() + sizeof(double), sample_bytes, value_size); for (uint32_t i = 0; i < num_channels; ++i) { - CHECK(recovered_samples[i] == sample_data[i]); + CHECK(read_float(sizeof(double) + i * value_size) == sample_data[i]); } }