From e44806d39aebfa0f500f9abddec5990ddc700b89 Mon Sep 17 00:00:00 2001 From: Stefan Appelhoff Date: Thu, 27 Aug 2026 10:24:13 +0200 Subject: [PATCH 1/2] fix: make thread shutdown interruptible and abort sockets on teardown --- src/recording.cpp | 78 ++++++++++++++++++++++++++++++++++++++++------- src/recording.h | 8 +++-- 2 files changed, 73 insertions(+), 13 deletions(-) diff --git a/src/recording.cpp b/src/recording.cpp index 0b8b43e..74dcc4b 100644 --- a/src/recording.cpp +++ b/src/recording.cpp @@ -1,6 +1,7 @@ #include "recording.h" //#include "conversions.h" +#include #include #include #ifdef XDFZ_SUPPORT @@ -36,7 +37,7 @@ inline bool timed_join(thread_p &thread, std::chrono::milliseconds duration = ma const auto start = Clock::now(); while (Clock::now() - start < duration) { if (try_join_once(thread)) return true; - std::this_thread::sleep_for(std::chrono::milliseconds(500)); + std::this_thread::sleep_for(std::chrono::milliseconds(20)); } return false; } @@ -72,7 +73,7 @@ inline void timed_join_or_detach( else ++it; } - std::this_thread::sleep_for(std::chrono::milliseconds(500)); + std::this_thread::sleep_for(std::chrono::milliseconds(20)); } if (!threads.empty()) { std::cout << threads.size() << " stream threads still running!" << std::endl; @@ -103,10 +104,21 @@ recording::~recording() { try { // set the shutdown flag (from now on no more new streams) shutdown_ = true; + shutdown_cv_.notify_all(); + + // close all inlets to unblock any pending network I/O immediately + { + std::lock_guard lock(inlets_mut_); + for (auto &in : active_inlets_) { + if (in) { + try { in->close_stream(); } catch (...) {} + } + } + } // stop the threads timed_join_or_detach(stream_threads_, max_join_wait); - if (!timed_join(boundary_thread_, max_join_wait + boundary_interval)) { + if (!timed_join(boundary_thread_, max_join_wait)) { std::cout << "boundary_thread didn't finish in time!" << std::endl; boundary_thread_->detach(); } @@ -119,6 +131,15 @@ recording::~recording() { void recording::requestStop() noexcept { shutdown_ = true; + shutdown_cv_.notify_all(); + { + std::lock_guard lock(inlets_mut_); + for (auto &in : active_inlets_) { + if (in) { + try { in->close_stream(); } catch (...) {} + } + } + } } void recording::record_from_query_results(const std::string &query) { @@ -173,6 +194,10 @@ void recording::record_from_streaminfo(const lsl::stream_info &src, bool phase_l // open an inlet to read from (and subscribe to data immediately) in.reset(new lsl::stream_inlet(src)); + { + std::lock_guard lock(inlets_mut_); + active_inlets_.push_back(in); + } auto it = sync_options_by_stream_.find(src.name() + " (" + src.hostname() + ")"); if (it != sync_options_by_stream_.end()) in->set_postprocessing(it->second); @@ -276,6 +301,12 @@ void recording::record_from_streaminfo(const lsl::stream_info &src, bool phase_l leave_footers_phase(phase_locked); throw; } + if (in) { + std::lock_guard lock(inlets_mut_); + active_inlets_.erase( + std::remove(active_inlets_.begin(), active_inlets_.end(), in), + active_inlets_.end()); + } } catch (std::exception &e) { std::cout << "Error in the record_from_streaminfo thread: " << e.what() << std::endl; } @@ -285,7 +316,15 @@ void recording::record_boundaries() { try { auto next_boundary = Clock::now() + boundary_interval; while (!shutdown_) { - std::this_thread::sleep_for(std::chrono::milliseconds(500)); + { + std::unique_lock cv_lock(shutdown_mut_); + if (shutdown_cv_.wait_for(cv_lock, std::chrono::milliseconds(500), [this] { + return shutdown_.load(); + })) { + break; + } + } + if (Clock::now() > next_boundary) { file_.write_boundary_chunk(); next_boundary = Clock::now() + boundary_interval; @@ -301,7 +340,15 @@ void recording::record_offsets( try { while (!shutdown_ && !offset_shutdown) { // sleep for the interval - std::this_thread::sleep_for(offset_interval); + { + std::unique_lock cv_lock(shutdown_mut_); + if (shutdown_cv_.wait_for(cv_lock, offset_interval, [this, &offset_shutdown] { + return shutdown_.load() || offset_shutdown.load(); + })) { + break; + } + } + // query the time offset double offset, now; try { @@ -311,9 +358,10 @@ void recording::record_offsets( std::cerr << "Timeout in time correction query for stream " << streamid << std::endl; } + if (shutdown_ || offset_shutdown) break; file_.write_stream_offset(streamid, now, offset); // also append to the offset lists - std::lock_guard lock(offset_mut_); + std::lock_guard offset_lock(offset_mut_); offset_lists_[streamid].emplace_back(now - offset, offset); } } catch (std::exception &e) { @@ -382,8 +430,8 @@ void recording::typed_transfer_loop(streamid_t streamid, double srate, const inl // Pull the first sample first_timestamp = 0.0; while(!shutdown_ && first_timestamp == 0.0) - first_timestamp = last_timestamp = in->pull_sample(chunk, 4.0); - if (!shutdown_) { + first_timestamp = last_timestamp = in->pull_sample(chunk, 0.1); + if (!shutdown_ && first_timestamp != 0.0) { timestamps.push_back(first_timestamp); file_.write_data_chunk(streamid, timestamps, chunk, (uint32_t)in->get_channel_count()); sample_count += timestamps.size(); @@ -403,17 +451,25 @@ void recording::typed_transfer_loop(streamid_t streamid, double srate, const inl last_timestamp = ts; } // write the actual chunk - file_.write_data_chunk(streamid, timestamps, chunk, in->get_channel_count()); - sample_count += timestamps.size(); + if (!timestamps.empty()) { + file_.write_data_chunk(streamid, timestamps, chunk, in->get_channel_count()); + sample_count += timestamps.size(); + } next_pull += chunk_interval; - std::this_thread::sleep_until(next_pull); + std::unique_lock cv_lock(shutdown_mut_); + if (shutdown_cv_.wait_until(cv_lock, next_pull, [this] { return shutdown_.load(); })) { + break; + } } } catch (std::exception &e) { std::cerr << "Error in transfer thread: " << e.what() << std::endl; offset_shutdown = true; + shutdown_cv_.notify_all(); timed_join_or_detach(offset_thread); throw; } + offset_shutdown = true; + shutdown_cv_.notify_all(); timed_join_or_detach(offset_thread); } diff --git a/src/recording.h b/src/recording.h index 0b198ba..56fb4fc 100644 --- a/src/recording.h +++ b/src/recording.h @@ -29,8 +29,8 @@ const auto max_footers_wait = std::chrono::seconds(2); // maximum waiting time for subscribing to a stream, in seconds (if exceeded, stream subscription // will take place later) const double max_open_wait = 5; -// maximum time that we wait to join a thread, in seconds -const std::chrono::seconds max_join_wait(5); +// maximum time that we wait to join a thread +const auto max_join_wait = std::chrono::seconds(2); using streamid_t = uint32_t; @@ -87,6 +87,10 @@ class recording { // phase-of-recording state (headers, streaming data, or footers) std::atomic shutdown_; // whether we are trying to shut down + std::condition_variable shutdown_cv_; // condition variable to wake threads immediately on shutdown + std::mutex shutdown_mut_; // mutex for shutdown condition variable + std::vector active_inlets_; // active inlets to abort on teardown + std::mutex inlets_mut_; // mutex to protect active inlets list uint32_t headers_to_finish_; // the number of streams that still need to write their header // (i.e., are not yet ready to write streaming content) uint32_t streaming_to_finish_; // the number of streams that still need to finish the streaming From c5d3038a6b57f4af7952d06cf93959842fcd3ba3 Mon Sep 17 00:00:00 2001 From: Stefan Appelhoff Date: Thu, 27 Aug 2026 10:24:14 +0200 Subject: [PATCH 2/2] test: add automated integration test for instant shutdown and XDF validation --- scripts/test_recording_teardown.py | 121 +++++++++++++++++++++++++++++ 1 file changed, 121 insertions(+) create mode 100644 scripts/test_recording_teardown.py diff --git a/scripts/test_recording_teardown.py b/scripts/test_recording_teardown.py new file mode 100644 index 0000000..209d112 --- /dev/null +++ b/scripts/test_recording_teardown.py @@ -0,0 +1,121 @@ +#!/usr/bin/env python +""" +Automated integration test for LabRecorder teardown and XDF integrity. +Tests that LabRecorder stops cleanly and instantly (< 500 ms) and produces valid XDF footers. +""" + +import argparse +import os +import subprocess +import sys +import time +import pylsl +import pyxdf + + +def run_test(cli_path, output_xdf="test_recording.xdf"): + if not os.path.exists(cli_path): + print(f"Error: LabRecorderCLI binary not found at '{cli_path}'") + return False + + if os.path.exists(output_xdf): + os.remove(output_xdf) + + print(f"--- Starting LSL test streams ---") + info_eeg = pylsl.StreamInfo("TestEEG", "EEG", 8, 100, "float32", "test_eeg_source_123") + outlet_eeg = pylsl.StreamOutlet(info_eeg) + + info_marker = pylsl.StreamInfo("TestMarker", "Markers", 1, 0, "string", "test_marker_source_123") + outlet_marker = pylsl.StreamOutlet(info_marker) + + time.sleep(0.5) + + print(f"--- Launching LabRecorderCLI ({cli_path}) ---") + proc = subprocess.Popen( + [cli_path, output_xdf, "name='TestEEG'", "name='TestMarker'"], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + + print(f"--- Streaming samples for 2 seconds ---") + start_time = time.time() + sample_val = 0.0 + while time.time() - start_time < 2.0: + outlet_eeg.push_sample([sample_val] * 8) + sample_val += 1.0 + time.sleep(0.01) + + print(f"--- Triggering shutdown (Enter key to stdin) ---") + t0 = time.perf_counter() + try: + stdout, stderr = proc.communicate(input="\n", timeout=4.0) + except subprocess.TimeoutExpired: + proc.kill() + print("FAIL: LabRecorderCLI hung during shutdown (> 4.0s)!") + return False + + stop_duration = time.perf_counter() - t0 + print(f"--- Teardown completed in {stop_duration:.3f} seconds ---") + + if stop_duration > 1.5: + print(f"FAIL: Shutdown took too long ({stop_duration:.3f}s > 1.5s)") + return False + else: + print(f"PASS: Instant shutdown verified (< 1.5s)") + + if not os.path.exists(output_xdf): + print(f"FAIL: Output file '{output_xdf}' was not created!") + return False + + file_size_kb = os.path.getsize(output_xdf) / 1024.0 + print(f"--- Output XDF file size: {file_size_kb:.2f} KB ---") + + print(f"--- Validating XDF file with pyxdf ---") + try: + streams, header = pyxdf.load_xdf(output_xdf) + except Exception as e: + print(f"FAIL: pyxdf failed to load XDF: {e}") + return False + + if len(streams) != 2: + print(f"FAIL: Expected 2 streams in XDF, got {len(streams)}") + return False + + eeg_stream = next((s for s in streams if s["info"]["name"][0] == "TestEEG"), None) + if not eeg_stream: + print("FAIL: TestEEG stream not found in XDF") + return False + + if len(eeg_stream["time_series"]) == 0: + print("FAIL: TestEEG has 0 recorded samples!") + return False + + print(f"PASS: TestEEG has {len(eeg_stream['time_series'])} samples recorded.") + + # Check footer + if "footer" not in eeg_stream or eeg_stream["footer"]["info"] is None: + print("FAIL: TestEEG is missing footer info!") + return False + + print("PASS: Stream footers are present and valid.") + print("=== ALL INTEGRATION TESTS PASSED ===") + + # Cleanup + if os.path.exists(output_xdf): + os.remove(output_xdf) + + return True + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Test LabRecorder teardown and XDF validity") + parser.add_argument( + "--bin", + default="./build/install/bin/LabRecorderCLI", + help="Path to LabRecorderCLI binary", + ) + args = parser.parse_args() + success = run_test(args.bin) + sys.exit(0 if success else 1)