From e44806d39aebfa0f500f9abddec5990ddc700b89 Mon Sep 17 00:00:00 2001 From: Stefan Appelhoff Date: Thu, 27 Aug 2026 10:24:13 +0200 Subject: [PATCH 1/4] 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/4] 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) From 54ffd28dc2068a677ec71c43a21a203848706424 Mon Sep 17 00:00:00 2001 From: Stefan Appelhoff Date: Sun, 20 Sep 2026 12:50:35 +0200 Subject: [PATCH 3/4] Bound every wait in recording teardown and stop dropping buffered samples Follow-up to the interruptible-teardown work, addressing review feedback. Shutdown flags are now published under the mutex that the condition variable predicates read them under. Setting an atomic outside that mutex and then notifying leaves a window in which a waiter that has just evaluated its predicate as false enters the wait and misses the notification, so the offset thread could still sleep out its full five-second interval. The blocking calls that a stop could not interrupt are now issued in short slices that observe the shutdown flag: - stream_inlet::info() was called twice with the default infinite timeout. close_stream() only stops the data receiver, so an unreachable metadata endpoint blocked a stop indefinitely. The info is now fetched once and reused for the header and the nominal rate. - open_stream() could hold a stop for up to max_open_wait. - time_correction() could hold it for the full query timeout. - The watchlist resolver blocked for a whole resolve_interval; it now resolves briefly and waits out the rest interruptibly. - The phase gates could park a stream for max_headers_wait with no way out, so a stream could lose its footer waiting for one that had hung. Joining is bounded for the first time: try_join_once() called std::thread::join(), which has no timeout, so polling it could never enforce max_join_wait. Threads are now paired with a future that becomes ready when the body returns, which can be waited on with a deadline. Closing the inlets the moment stop is pressed discards everything still buffered in them; a recording of 40 markers came back with 2. Inlets are now closed only after the stream threads have been given a grace period to drain and write their footers, and the transfer loop does a final non-blocking pull on the way out, so a stop no longer costs samples that had already arrived. Also fixed along the way: record_offsets() wrote uninitialised offset and timestamp values into the file when a time correction query timed out; the inlet bookkeeping leaked a registration on every exception path; and a stream that failed mid-recording was left without a footer although its header was already on disk. scripts/test_recording_teardown.py now covers a plain stop, a stop before the first sample, a stop while subscribing to a source that has gone away, repeated start/stop cycles, and that no buffered sample is lost. It checks the exit status and the footers of every stream, holds one stated teardown budget instead of documenting one and asserting another, and runs on all three platforms in CI. --- .github/workflows/build.yml | 36 ++- .gitignore | 1 + scripts/test_recording_teardown.py | 408 +++++++++++++++++++++----- src/recording.cpp | 454 +++++++++++++++++------------ src/recording.h | 118 +++++++- 5 files changed, 717 insertions(+), 300 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 8b29136..6f5048e 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -38,10 +38,10 @@ jobs: fail-fast: false matrix: config: - - { name: "Ubuntu 22.04", os: ubuntu-22.04 } - - { name: "Ubuntu 24.04", os: ubuntu-24.04 } - - { name: "macOS", os: macos-14, cmake_extra: '-DCMAKE_OSX_ARCHITECTURES="x86_64;arm64"' } - - { name: "Windows", os: windows-latest } + - { name: "Ubuntu 22.04", os: ubuntu-22.04, cli: "install/bin/LabRecorderCLI" } + - { name: "Ubuntu 24.04", os: ubuntu-24.04, cli: "install/bin/LabRecorderCLI" } + - { name: "macOS", os: macos-14, cmake_extra: '-DCMAKE_OSX_ARCHITECTURES="x86_64;arm64"', cli: "install/LabRecorderCLI" } + - { name: "Windows", os: windows-latest, cli: "install/LabRecorderCLI.exe" } steps: - name: Checkout @@ -97,17 +97,27 @@ jobs: # ----------------------------------------------------------------------- # Test CLI # ----------------------------------------------------------------------- - - name: Test CLI (Linux) - if: runner.os == 'Linux' - run: ./install/bin/LabRecorderCLI --help || true + - name: Test CLI + shell: bash + run: ./${{ matrix.config.cli }} --help || true + + # ----------------------------------------------------------------------- + # Integration test: teardown latency and XDF integrity + # ----------------------------------------------------------------------- + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.12' - - name: Test CLI (macOS) - if: runner.os == 'macOS' - run: ./install/LabRecorderCLI --help || true + - name: Install integration test dependencies + run: python -m pip install --upgrade pip pylsl pyxdf - - name: Test CLI (Windows) - if: runner.os == 'Windows' - run: ./install/LabRecorderCLI.exe --help || true + # pylsl brings its own liblsl; it only has to speak the same wire protocol as the liblsl + # bundled with the recorder, not be the same build. Set PYLSL_LIB here if that ever stops + # holding. + - name: Test recording teardown + shell: bash + run: python scripts/test_recording_teardown.py --bin "${{ matrix.config.cli }}" # ----------------------------------------------------------------------- # Package diff --git a/.gitignore b/.gitignore index 2333b58..deafc5b 100644 --- a/.gitignore +++ b/.gitignore @@ -18,3 +18,4 @@ liblsl.deb install-qt.sh .DS_Store .codegraph/ +__pycache__/ diff --git a/scripts/test_recording_teardown.py b/scripts/test_recording_teardown.py index 209d112..43a62f2 100644 --- a/scripts/test_recording_teardown.py +++ b/scripts/test_recording_teardown.py @@ -1,121 +1,367 @@ #!/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. +"""Integration test for LabRecorderCLI teardown and XDF integrity. + +Starts LSL outlets, records them with LabRecorderCLI, stops the recording and checks that + +* the recorder exits within ``--max-stop`` seconds (1.0 s by default) and with status 0, +* every recorded stream has a stream footer whose sample count matches its data, +* pyxdf does not report the file as damaged, and +* nothing that was sent before the stop is missing from the file. + +The cases cover a plain stop, a stop before any sample arrives, a stop while the recorder is +still subscribing to a source that has gone away, and repeated start/stop cycles. + +Requires ``pylsl`` and ``pyxdf``. """ import argparse +import contextlib +import logging import os import subprocess import sys +import tempfile +import threading import time + import pylsl import pyxdf +EEG_NAME = "TeardownTestEEG" +MARKER_NAME = "TeardownTestMarkers" +NAMES = (EEG_NAME, MARKER_NAME) +EEG_CHANNELS = 8 +EEG_RATE = 100.0 -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 +# time given to LSL to make a new outlet discoverable, and to flush the last samples over TCP +SETTLE = 0.5 - 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) +class TestFailure(AssertionError): + """Raised when a case does not hold up.""" - info_marker = pylsl.StreamInfo("TestMarker", "Markers", 1, 0, "string", "test_marker_source_123") - outlet_marker = pylsl.StreamOutlet(info_marker) - time.sleep(0.5) +def check(condition, message): + if not condition: + raise TestFailure(message) - 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, + +def make_outlets(): + """Create the EEG and marker outlets used by every case.""" + eeg_info = pylsl.StreamInfo( + EEG_NAME, "EEG", EEG_CHANNELS, EEG_RATE, "float32", "teardown_test_eeg" + ) + marker_info = pylsl.StreamInfo( + MARKER_NAME, "Markers", 1, pylsl.IRREGULAR_RATE, "string", "teardown_test_markers" ) + return pylsl.StreamOutlet(eeg_info), pylsl.StreamOutlet(marker_info) + + +class Recorder: + """A running LabRecorderCLI, with its output read as it appears. + + Reading the output as it appears is what lets a case wait for the recorder to actually be + collecting before it sends anything: an outlet does not replay what it pushed before the + recorder subscribed, so pushing too early silently loses samples. + """ + + def __init__(self, cli_path, xdf_path): + self._proc = subprocess.Popen( + [cli_path, xdf_path, f"name='{EEG_NAME}'", f"name='{MARKER_NAME}'"], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=1, + ) + self.lines = [] + self._reader = threading.Thread(target=self._read_output, daemon=True) + self._reader.start() - 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) + def _read_output(self): + for line in self._proc.stdout: + self.lines.append(line.rstrip()) - print(f"--- Triggering shutdown (Enter key to stdin) ---") - t0 = time.perf_counter() + def wait_for(self, needles, timeout=20.0): + """Block until every needle has shown up in the output.""" + deadline = time.time() + timeout + while time.time() < deadline: + joined = "\n".join(self.lines) + if all(needle in joined for needle in needles): + return + if self._proc.poll() is not None: + raise TestFailure( + f"LabRecorderCLI exited (status {self._proc.returncode}) before it was ready" + ) + time.sleep(0.02) + raise TestFailure(f"LabRecorderCLI did not report {needles!r} within {timeout} s") + + def wait_until_collecting(self): + self.wait_for([f"Started data collection for stream {name}." for name in NAMES]) + + def stop(self): + """Send the quit key and return how long the recorder took to exit.""" + started = time.perf_counter() + self._proc.stdin.write("\n") + self._proc.stdin.flush() + self._proc.stdin.close() + try: + # generously above any bound asserted on, so a hang is reported as a hang rather + # than as a timeout of this harness + self._proc.wait(timeout=30.0) + except subprocess.TimeoutExpired: + self._proc.kill() + self._proc.wait() + raise TestFailure("LabRecorderCLI did not exit within 30 s of the stop request") + duration = time.perf_counter() - started + self._reader.join(timeout=5.0) + check( + self._proc.returncode == 0, + f"LabRecorderCLI exited with status {self._proc.returncode}", + ) + return duration + + def terminate(self): + if self._proc.poll() is None: + self._proc.kill() + self._proc.wait() + + +@contextlib.contextmanager +def recorder(cli_path, xdf_path): + """Run LabRecorderCLI over both test streams, making sure it is gone afterwards.""" + rec = Recorder(cli_path, xdf_path) 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 + yield rec + finally: + rec.terminate() + for line in rec.lines: + print(f" | {line}") + - stop_duration = time.perf_counter() - t0 - print(f"--- Teardown completed in {stop_duration:.3f} seconds ---") +# pyxdf reports a damaged file through its logger rather than by raising, so these are the +# substrings that mark a load as failed. Other warnings (about jitter or clock offsets, say) say +# something about the data, not about the file being intact, and are only printed. +INTEGRITY_WARNINGS = ("footer", "truncat", "corrupt", "incomplete", "unexpected", "not parse") - 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 +def load_xdf_strict(xdf_path): + """Load an XDF file and fail if pyxdf reports it as damaged.""" + records = [] - file_size_kb = os.path.getsize(output_xdf) / 1024.0 - print(f"--- Output XDF file size: {file_size_kb:.2f} KB ---") + class Collector(logging.Handler): + def emit(self, record): + records.append(record) - print(f"--- Validating XDF file with pyxdf ---") + handler = Collector(level=logging.WARNING) + logger = logging.getLogger("pyxdf") + logger.addHandler(handler) try: - streams, header = pyxdf.load_xdf(output_xdf) - except Exception as e: - print(f"FAIL: pyxdf failed to load XDF: {e}") - return False + streams, header = pyxdf.load_xdf(xdf_path) + finally: + logger.removeHandler(handler) - if len(streams) != 2: - print(f"FAIL: Expected 2 streams in XDF, got {len(streams)}") - return False + problems = [] + for record in records: + message = record.getMessage() + if record.levelno >= logging.ERROR or any( + marker in message.lower() for marker in INTEGRITY_WARNINGS + ): + problems.append(message) + else: + print(f" (pyxdf) {message}") + if problems: + raise TestFailure(f"pyxdf reported a damaged file: {'; '.join(problems)}") + return streams, header - 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 +def stream_by_name(streams, name): + for stream in streams: + if stream["info"]["name"][0] == name: + return stream + raise TestFailure(f"stream {name!r} is missing from the recording") - 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 +def check_footer(stream): + """Check that a stream carries a footer consistent with its data.""" + name = stream["info"]["name"][0] + footer = stream.get("footer") + check(footer and footer.get("info"), f"stream {name!r} has no footer") - print("PASS: Stream footers are present and valid.") - print("=== ALL INTEGRATION TESTS PASSED ===") + info = footer["info"] + recorded = len(stream["time_series"]) + reported = int(info["sample_count"][0]) + check( + reported == recorded, + f"stream {name!r} footer claims {reported} samples but holds {recorded}", + ) + # these are written from the same footer and must be parseable for dejittering to work + float(info["first_timestamp"][0]) + float(info["last_timestamp"][0]) - # Cleanup - if os.path.exists(output_xdf): - os.remove(output_xdf) - return True +def check_stop(duration, max_stop): + print(f" teardown took {duration:.3f} s (budget {max_stop:.3f} s)") + check( + duration <= max_stop, + f"teardown took {duration:.3f} s, which is above the {max_stop:.3f} s budget", + ) -if __name__ == "__main__": - parser = argparse.ArgumentParser(description="Test LabRecorder teardown and XDF validity") +def push_eeg_for(outlet, seconds): + """Push EEG samples at the nominal rate for the given duration.""" + deadline = time.time() + seconds + value = 0.0 + while time.time() < deadline: + outlet.push_sample([value] * EEG_CHANNELS) + value += 1.0 + time.sleep(1.0 / EEG_RATE) + + +def case_normal_stop(cli_path, xdf_path, max_stop): + """Record both streams for two seconds, then stop.""" + eeg, markers = make_outlets() + time.sleep(SETTLE) + with recorder(cli_path, xdf_path) as rec: + rec.wait_until_collecting() + push_eeg_for(eeg, 2.0) + time.sleep(SETTLE) + duration = rec.stop() + + check_stop(duration, max_stop) + streams, _ = load_xdf_strict(xdf_path) + check(len(streams) == 2, f"expected 2 streams in the recording, got {len(streams)}") + + eeg_stream = stream_by_name(streams, EEG_NAME) + check(len(eeg_stream["time_series"]) > 0, "no EEG samples were recorded") + # the marker stream stays silent on purpose: a stream that never sends must still be + # closed out properly + check_footer(eeg_stream) + check_footer(stream_by_name(streams, MARKER_NAME)) + del eeg, markers + + +def case_stop_before_first_sample(cli_path, xdf_path, max_stop): + """Stop once the recorder is collecting but before either stream has sent anything.""" + eeg, markers = make_outlets() + time.sleep(SETTLE) + with recorder(cli_path, xdf_path) as rec: + rec.wait_until_collecting() + duration = rec.stop() + + check_stop(duration, max_stop) + streams, _ = load_xdf_strict(xdf_path) + check(len(streams) == 2, f"expected 2 streams in the recording, got {len(streams)}") + for name in NAMES: + stream = stream_by_name(streams, name) + check( + len(stream["time_series"]) == 0, + f"stream {name!r} recorded samples although none were sent", + ) + check_footer(stream) + del eeg, markers + + +def case_stop_while_subscribing(cli_path, xdf_path, max_stop): + """Stop while the recorder is still subscribing, so it is blocked on the network. + + The sources are dropped as soon as the recorder has found them, which leaves it waiting on + an endpoint that will never answer -- the case the data receiver alone cannot unblock. + """ + eeg, markers = make_outlets() + time.sleep(SETTLE) + with recorder(cli_path, xdf_path) as rec: + rec.wait_for([f"Found {name}" for name in NAMES]) + del eeg, markers + duration = rec.stop() + + check_stop(duration, max_stop) + # the file is checked for integrity, but not for content: how far the recorder got before + # the sources went away is timing dependent + load_xdf_strict(xdf_path) + + +def case_repeated_shutdown(cli_path, xdf_path, max_stop): + """Run three start/stop cycles, so leaked threads or stale state show up.""" + for cycle in range(3): + print(f" cycle {cycle + 1}/3") + case_normal_stop(cli_path, xdf_path, max_stop) + + +def case_no_buffered_samples_lost(cli_path, xdf_path, max_stop): + """Every marker pushed before the stop must end up in the file.""" + marker_count = 40 + eeg, markers = make_outlets() + time.sleep(SETTLE) + with recorder(cli_path, xdf_path) as rec: + rec.wait_until_collecting() + push_eeg_for(eeg, 1.0) + for i in range(marker_count): + markers.push_sample([f"marker-{i}"]) + # let the markers reach the recorder; whatever is still sitting in its inlet at the + # stop has to be drained rather than dropped + time.sleep(SETTLE) + duration = rec.stop() + + check_stop(duration, max_stop) + streams, _ = load_xdf_strict(xdf_path) + marker_stream = stream_by_name(streams, MARKER_NAME) + recorded = len(marker_stream["time_series"]) + check( + recorded == marker_count, + f"{marker_count} markers were sent but {recorded} were recorded", + ) + check_footer(marker_stream) + del eeg, markers + + +CASES = [ + ("normal stop", case_normal_stop), + ("stop before first sample", case_stop_before_first_sample), + ("stop while subscribing", case_stop_while_subscribing), + ("repeated shutdown", case_repeated_shutdown), + ("no buffered samples lost", case_no_buffered_samples_lost), +] + + +def main(): + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--bin", required=True, help="path to the LabRecorderCLI binary") parser.add_argument( - "--bin", - default="./build/install/bin/LabRecorderCLI", - help="Path to LabRecorderCLI binary", + "--max-stop", + type=float, + default=1.0, + help="upper bound in seconds for how long teardown may take (default: %(default)s)", ) args = parser.parse_args() - success = run_test(args.bin) - sys.exit(0 if success else 1) + + if not os.path.exists(args.bin): + print(f"LabRecorderCLI binary not found at {args.bin!r}") + return 1 + # absolute, and with native separators: CreateProcess does not accept a relative path + # spelled with forward slashes + cli_path = os.path.abspath(args.bin) + + failures = [] + with tempfile.TemporaryDirectory() as workdir: + for name, case in CASES: + xdf_path = os.path.join(workdir, f"{name.replace(' ', '_')}.xdf") + print(f"--- {name} ---") + try: + case(cli_path, xdf_path, args.max_stop) + except TestFailure as exc: + print(f"FAIL: {name}: {exc}") + failures.append(name) + else: + print(f"PASS: {name}") + + print() + if failures: + print(f"{len(failures)}/{len(CASES)} cases failed: {', '.join(failures)}") + return 1 + print(f"all {len(CASES)} cases passed") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/recording.cpp b/src/recording.cpp index 74dcc4b..6548274 100644 --- a/src/recording.cpp +++ b/src/recording.cpp @@ -10,75 +10,81 @@ #include #endif +namespace { + +// time spent waiting between two resolves of a watchlist query; the resolve itself already takes +// resolve_timeout, so together they keep the resolve_interval cadence +const auto resolve_pause = std::chrono::duration_cast( + std::chrono::duration(resolve_interval - resolve_timeout)); + +/// convert a timeout given in seconds into a Clock duration +inline Clock::duration seconds_to_duration(double seconds) { + return std::chrono::duration_cast(std::chrono::duration(seconds)); +} + +} // namespace + // Thread utilities -using Clock = std::chrono::high_resolution_clock; /** - * @brief try_join_once joins and deconstructs the thread if possible - * @param thread unique_ptr to a std::tread. Will be reset on success - * @return true if the thread was successfully joined, false otherwise + * @brief timed_join Waits up to duration for the worker to finish, then joins it + * @param w unique_ptr to a worker. Will be reset on success + * @param duration max duration to wait + * @return true if the worker finished and was joined, false if it is still running */ -inline bool try_join_once(std::unique_ptr &thread) { - if (thread && thread->joinable()) { - thread->join(); - thread.reset(); - return true; - } - return false; +inline bool timed_join(worker_p &w, std::chrono::milliseconds duration = max_join_wait) { + if (!w) return true; + // wait on the future rather than calling join() directly: join() has no timeout, so a single + // call against a hung thread would never return and no deadline could be enforced + if (w->done.wait_for(duration) != std::future_status::ready) return false; + w->thread.join(); + w.reset(); + return true; } /** - * @brief timed_join Tries to join the passed thread until it succeeds or duration passes - * @param thread unique_ptr to a std::tread. Will be reset on success - * @param duration max duration to try joining - * @return true on success, false otherwise + * @brief timed_join_or_detach Join the worker or detach it if not possible within specified + * duration + * @param w unique_ptr to a worker. Will be reset either way + * @param duration max duration to wait */ -inline bool timed_join(thread_p &thread, std::chrono::milliseconds duration = max_join_wait) { - 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(20)); +inline void timed_join_or_detach(worker_p &w, std::chrono::milliseconds duration = max_join_wait) { + if (!timed_join(w, duration)) { + w->thread.detach(); + w.reset(); + std::cerr << "Thread didn't join in time!" << std::endl; } - return false; } /** - * @brief timed_join_or_detach Join the thread or detach it if not possible within specified - * duration - * @param thread unique_ptr to a std::tread. Will be reset on success - * @param duration max duration to try joining + * @brief timed_join_some Join whichever workers finish within duration, leave the rest in place + * @param workers list of workers. Joined ones are erased from it + * @param duration duration to wait, shared across all workers */ -inline void timed_join_or_detach( - thread_p &thread, std::chrono::milliseconds duration = max_join_wait) { - if (!timed_join(thread, duration)) { - thread->detach(); - std::cerr << "Thread didn't join in time!" << std::endl; +inline void timed_join_some(std::list &workers, std::chrono::milliseconds duration) { + const auto deadline = Clock::now() + duration; + for (auto it = workers.begin(); it != workers.end();) { + const auto remaining = + std::chrono::duration_cast(deadline - Clock::now()); + if (timed_join(*it, std::max(remaining, std::chrono::milliseconds(0)))) + it = workers.erase(it); + else + ++it; } } /** - * @brief timed_join_or_detach Join the thread or detach it if not possible within specified - * duration - * @param threads list of unique_ptrs to std::threads. Guaranteed to be empty - * afterwards. - * @param duration duration to try joining + * @brief timed_join_or_detach Join the workers or detach those that don't finish in time + * @param workers list of workers. Guaranteed to be empty afterwards. + * @param duration duration to wait, shared across all workers */ inline void timed_join_or_detach( - std::list &threads, std::chrono::milliseconds duration = max_join_wait) { - const auto start = Clock::now(); - while (Clock::now() - start < duration && !threads.empty()) { - for (auto it = threads.begin(); it != threads.end();) { - if (try_join_once(*it)) - it = threads.erase(it); - else - ++it; - } - std::this_thread::sleep_for(std::chrono::milliseconds(20)); - } - if (!threads.empty()) { - std::cout << threads.size() << " stream threads still running!" << std::endl; - for (auto &t : threads) t->detach(); - threads.clear(); + std::list &workers, std::chrono::milliseconds duration = max_join_wait) { + timed_join_some(workers, duration); + if (!workers.empty()) { + std::cout << workers.size() << " stream threads still running!" << std::endl; + for (auto &w : workers) w->thread.detach(); + workers.clear(); } } @@ -91,86 +97,147 @@ recording::recording(const std::string &filename, const std::vector(&recording::record_boundaries, this); + boundary_thread_ = spawn_worker([this] { record_boundaries(); }); } 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)) { - std::cout << "boundary_thread didn't finish in time!" << std::endl; - boundary_thread_->detach(); + // set the shutdown flag (from now on no more new streams) and wake every waiting thread + requestStop(); + + // give the stream threads a moment to drain their inlets and write their footers by + // themselves; closing an inlet discards what it still holds, so that is a last resort + timed_join_some(stream_threads_, teardown_grace); + if (!stream_threads_.empty()) { + // a thread is stuck in a blocking socket call; closing its inlet aborts that call + close_active_inlets(); + timed_join_or_detach(stream_threads_, max_join_wait); } + timed_join_or_detach(boundary_thread_, max_join_wait); std::cout << "Closing the file." << std::endl; } catch (std::exception &e) { std::cout << "Error while closing the recording: " << e.what() << std::endl; } } -void recording::requestStop() noexcept -{ - shutdown_ = true; +void recording::requestStop() noexcept { + { + // publish the flag under the mutex that the shutdown_cv_ predicates read it under: a + // waiter that has just evaluated its predicate as false would otherwise miss the + // notification below and sleep out its full interval + std::lock_guard lock(shutdown_mut_); + shutdown_ = true; + } shutdown_cv_.notify_all(); + + // the phase gates test shutdown_ under phase_mut_, so take and release it for the same reason + { std::lock_guard lock(phase_mut_); } + ready_for_streaming_.notify_all(); + ready_for_footers_.notify_all(); +} + +bool recording::wait_until_shutdown(Clock::time_point deadline, const std::atomic *extra) { + std::unique_lock lock(shutdown_mut_); + return shutdown_cv_.wait_until( + lock, deadline, [this, extra] { return shutdown_.load() || (extra && extra->load()); }); +} + +void recording::stop_offsets(const offset_flag_p &offset_shutdown) noexcept { { - std::lock_guard lock(inlets_mut_); - for (auto &in : active_inlets_) { - if (in) { - try { in->close_stream(); } catch (...) {} - } + std::lock_guard lock(shutdown_mut_); + *offset_shutdown = true; + } + shutdown_cv_.notify_all(); +} + +void recording::register_inlet(const inlet_p &in) { + std::lock_guard lock(inlets_mut_); + active_inlets_.push_back(in); +} + +void recording::unregister_inlet(const inlet_p &in) noexcept { + if (!in) return; + std::lock_guard lock(inlets_mut_); + active_inlets_.erase( + std::remove(active_inlets_.begin(), active_inlets_.end(), in), active_inlets_.end()); +} + +void recording::close_active_inlets() noexcept { + std::lock_guard lock(inlets_mut_); + for (auto &in : active_inlets_) { + try { + in->close_stream(); + } catch (std::exception &e) { + std::cerr << "Error while closing an inlet: " << e.what() << std::endl; } } } +bool recording::open_inlet(const inlet_p &in) { + // subscribe in short slices: a single open_stream(max_open_wait) would keep us from noticing a + // stop for up to max_open_wait seconds + const auto deadline = Clock::now() + seconds_to_duration(max_open_wait); + while (Clock::now() < deadline && !shutdown_) { + try { + in->open_stream(network_poll_interval); + return true; + } catch (lsl::timeout_error &) {} + } + return false; +} + +lsl::stream_info recording::fetch_info(const inlet_p &in) { + // the metadata receiver is separate from the data receiver, so close_stream() does not abort + // this call; poll in short slices instead, or an unreachable source blocks us indefinitely. + // A stop does not cut this off immediately: a source that is still reachable gets a short + // grace period, so its header (and with it its footer) still makes it into the file. + auto deadline = Clock::time_point::max(); + while (Clock::now() < deadline) { + if (shutdown_ && deadline == Clock::time_point::max()) + deadline = Clock::now() + teardown_grace; + try { + return in->info(network_poll_interval); + } catch (lsl::timeout_error &) {} + } + throw shutdown_requested("stopped while retrieving the stream metadata"); +} + void recording::record_from_query_results(const std::string &query) { try { std::set known_uids; // set of previously seen stream uid's std::set known_source_ids; // set of previously seen source id's - std::list threads; // our spawned threads + std::list threads; // our spawned threads std::cout << "Watching for a stream with properties " << query << std::endl; while (!shutdown_) { - // periodically re-resolve the query - const std::vector results = lsl::resolve_stream(query, 0, resolve_interval); + // periodically re-resolve the query. The resolve itself is kept short and the rest of + // the interval is spent in an interruptible wait, so a stop is noticed quickly. + const std::vector results = + lsl::resolve_stream(query, 0, resolve_timeout); // for each result... for (const auto &result : results) { // if it is a new stream... - std::string _uid = result.uid(); - std::string _src_id = result.source_id(); if (!known_uids.count(result.uid())) // and doesn't have a previously seen source id... if (!result.source_id().empty() && - (!known_source_ids.count(result.source_id()))) { + (!known_source_ids.count(result.source_id()))) { std::cout << "Found a new stream named " << result.name() << ", adding it to the recording." << std::endl; // start a new recording thread - threads.emplace_back(new std::thread( - &recording::record_from_streaminfo, this, result, false)); + threads.emplace_back(spawn_worker( + [this, result] { record_from_streaminfo(result, false); })); // ... and add it to the lists of known id's known_uids.insert(result.uid()); if (!result.source_id().empty()) known_source_ids.insert(result.source_id()); } } + if (wait_for_shutdown(resolve_pause)) break; } // wait for all our threads to join timed_join_or_detach(threads, max_join_wait); @@ -180,39 +247,38 @@ void recording::record_from_query_results(const std::string &query) { } void recording::record_from_streaminfo(const lsl::stream_info &src, bool phase_locked) { + inlet_p in; try { - double first_timestamp, last_timestamp; + // initialised here because a stream that fails mid-recording still writes a footer + double first_timestamp = 0.0, last_timestamp = 0.0; uint64_t sample_count = 0; + double nominal_srate = 0; // obtain a fresh streamid streamid_t streamid = fresh_streamid(); - inlet_p in; - // --- headers phase try { enter_headers_phase(phase_locked); // 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); - } + in = std::make_shared(src); + register_inlet(in); auto it = sync_options_by_stream_.find(src.name() + " (" + src.hostname() + ")"); if (it != sync_options_by_stream_.end()) in->set_postprocessing(it->second); - try { - in->open_stream(max_open_wait); + if (open_inlet(in)) std::cout << "Opened the stream " << src.name() << "." << std::endl; - } catch (lsl::timeout_error &) { + else if (!shutdown_) std::cout << "Subscribing to the stream " << src.name() << " is taking relatively long; collection from this stream will be delayed." << std::endl; - } - // retrieve the stream header & get its XML version - file_.write_stream_header(streamid, in->info().as_xml()); + // retrieve the stream header & get its XML version. The nominal rate is taken from + // the same info, saving a second round trip to the source. + const lsl::stream_info info = fetch_info(in); + nominal_srate = info.nominal_srate(); + file_.write_stream_header(streamid, info.as_xml()); std::cout << "Received header for stream " << src.name() << "." << std::endl; leave_headers_phase(phase_locked); @@ -232,33 +298,31 @@ void recording::record_from_streaminfo(const lsl::stream_info &src, bool phase_l enter_streaming_phase(phase_locked); std::cout << "Started data collection for stream " << src.name() << "." << std::endl; - const double nominal_srate = in->info().nominal_srate(); - // now write the actual sample chunks... switch (src.channel_format()) { case lsl::cf_int8: - typed_transfer_loop(streamid, nominal_srate, in, first_timestamp, - last_timestamp, sample_count); + typed_transfer_loop( + streamid, nominal_srate, in, first_timestamp, last_timestamp, sample_count); break; case lsl::cf_int16: - typed_transfer_loop(streamid, nominal_srate, in, first_timestamp, - last_timestamp, sample_count); + typed_transfer_loop( + streamid, nominal_srate, in, first_timestamp, last_timestamp, sample_count); break; case lsl::cf_int32: - typed_transfer_loop(streamid, nominal_srate, in, first_timestamp, - last_timestamp, sample_count); + typed_transfer_loop( + streamid, nominal_srate, in, first_timestamp, last_timestamp, sample_count); break; case lsl::cf_float32: - typed_transfer_loop(streamid, nominal_srate, in, first_timestamp, - last_timestamp, sample_count); + typed_transfer_loop( + streamid, nominal_srate, in, first_timestamp, last_timestamp, sample_count); break; case lsl::cf_double64: - typed_transfer_loop(streamid, nominal_srate, in, first_timestamp, - last_timestamp, sample_count); + typed_transfer_loop( + streamid, nominal_srate, in, first_timestamp, last_timestamp, sample_count); break; case lsl::cf_string: - typed_transfer_loop(streamid, nominal_srate, in, - first_timestamp, last_timestamp, sample_count); + typed_transfer_loop( + streamid, nominal_srate, in, first_timestamp, last_timestamp, sample_count); break; default: // unsupported channel format @@ -267,9 +331,12 @@ void recording::record_from_streaminfo(const lsl::stream_info &src, bool phase_l } leave_streaming_phase(phase_locked); - } catch (std::exception &) { + } catch (std::exception &e) { leave_streaming_phase(phase_locked); - throw; + // the header is already on disk, so fall through to the footer instead of leaving the + // stream without one + std::cerr << "Error while recording from " << src.name() << ": " << e.what() + << std::endl; } // --- footers phase @@ -301,34 +368,19 @@ 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 (shutdown_requested &e) { + std::cout << "Recording from " << src.name() << " ended: " << e.what() << std::endl; } catch (std::exception &e) { std::cout << "Error in the record_from_streaminfo thread: " << e.what() << std::endl; } + unregister_inlet(in); } void recording::record_boundaries() { try { - auto next_boundary = Clock::now() + boundary_interval; while (!shutdown_) { - { - 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; - } + if (wait_for_shutdown(boundary_interval)) break; + file_.write_boundary_chunk(); } } catch (std::exception &e) { std::cout << "Error in the record_boundaries thread: " << e.what() << std::endl; @@ -336,32 +388,34 @@ void recording::record_boundaries() { } void recording::record_offsets( - streamid_t streamid, const inlet_p &in, std::atomic &offset_shutdown) noexcept { + streamid_t streamid, inlet_p in, offset_flag_p offset_shutdown) noexcept { try { - while (!shutdown_ && !offset_shutdown) { + while (!shutdown_ && !*offset_shutdown) { // sleep for the 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(); - })) { + if (wait_for_shutdown(offset_interval, offset_shutdown.get())) break; + + // query the time offset, again in short slices so that a stop is noticed promptly + double offset = 0, now = 0; + bool have_offset = false; + const auto deadline = Clock::now() + max_time_correction_wait; + while (!shutdown_ && !*offset_shutdown && Clock::now() < deadline) { + try { + offset = in->time_correction(network_poll_interval); + now = lsl::local_clock(); + have_offset = true; break; - } + } catch (lsl::timeout_error &) {} } - - // query the time offset - double offset, now; - try { - offset = in->time_correction(2); - now = lsl::local_clock(); - } catch (lsl::timeout_error &) { + if (!have_offset) { + if (shutdown_ || *offset_shutdown) break; std::cerr << "Timeout in time correction query for stream " << streamid << std::endl; + continue; } - if (shutdown_ || offset_shutdown) break; + file_.write_stream_offset(streamid, now, offset); // also append to the offset lists - std::lock_guard offset_lock(offset_mut_); + std::lock_guard lock(offset_mut_); offset_lists_[streamid].emplace_back(now - offset, offset); } } catch (std::exception &e) { @@ -389,8 +443,11 @@ void recording::leave_headers_phase(bool phase_locked) { void recording::enter_streaming_phase(bool phase_locked) { if (phase_locked) { std::unique_lock lock(phase_mut_); - ready_for_streaming_.wait_for( - lock, max_headers_wait, [this]() { return this->ready_for_streaming(); }); + // on shutdown the gate is dropped: the transfer loop exits immediately anyway, and waiting + // out max_headers_wait for a stream that is never going to report in only delays the + // footer of this one + ready_for_streaming_.wait_for(lock, max_headers_wait, + [this]() { return this->ready_for_streaming() || shutdown_.load(); }); streaming_to_finish_++; } } @@ -407,8 +464,9 @@ void recording::leave_streaming_phase(bool phase_locked) { void recording::enter_footers_phase(bool phase_locked) { if (phase_locked) { std::unique_lock lock(phase_mut_); - ready_for_footers_.wait_for( - lock, max_footers_wait, [this]() { return this->ready_for_footers(); }); + // see enter_streaming_phase: a footer written slightly out of order beats no footer at all + ready_for_footers_.wait_for(lock, max_footers_wait, + [this]() { return this->ready_for_footers() || shutdown_.load(); }); } } @@ -416,10 +474,12 @@ template void recording::typed_transfer_loop(streamid_t streamid, double srate, const inlet_p &in, double &first_timestamp, double &last_timestamp, uint64_t &sample_count) { // optionally start an offset collection thread for this stream - std::atomic offset_shutdown{false}; - thread_p offset_thread(offsets_enabled_ ? new std::thread(&recording::record_offsets, this, - streamid, in, std::ref(offset_shutdown)) - : nullptr); + auto offset_shutdown = std::make_shared>(false); + worker_p offset_thread(offsets_enabled_ + ? spawn_worker([this, streamid, in, offset_shutdown] { + record_offsets(streamid, in, offset_shutdown); + }) + : nullptr); try { double sample_interval = srate ? 1.0 / srate : 0; @@ -427,21 +487,9 @@ void recording::typed_transfer_loop(streamid_t streamid, double srate, const inl std::vector chunk; std::vector timestamps; - // Pull the first sample - first_timestamp = 0.0; - while(!shutdown_ && first_timestamp == 0.0) - 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(); - } - - auto next_pull = Clock::now(); - while (!shutdown_) { - // get a chunk from the stream - in->pull_chunk_multiplexed(chunk, ×tamps, 1e-6); - // for each sample... + // deduce the timestamps that can be deduced and write the chunk out + auto write_chunk = [&] { + if (timestamps.empty()) return; for (double &ts : timestamps) { // if the time stamp can be deduced from the previous one... if (last_timestamp + sample_interval == ts) { @@ -450,26 +498,48 @@ void recording::typed_transfer_loop(streamid_t streamid, double srate, const inl } else last_timestamp = ts; } - // write the actual chunk - if (!timestamps.empty()) { - file_.write_data_chunk(streamid, timestamps, chunk, in->get_channel_count()); - sample_count += timestamps.size(); - } + file_.write_data_chunk(streamid, timestamps, chunk, in->get_channel_count()); + sample_count += timestamps.size(); + }; + // Pull the first sample + first_timestamp = 0.0; + while (!shutdown_ && first_timestamp == 0.0) + first_timestamp = last_timestamp = in->pull_sample(chunk, network_poll_interval); + if (first_timestamp != 0.0) { + // written directly: the very first sample anchors the stream and must keep its + // timestamp even when the nominal interval is zero + timestamps.assign(1, first_timestamp); + file_.write_data_chunk(streamid, timestamps, chunk, (uint32_t)in->get_channel_count()); + sample_count += timestamps.size(); + } + + auto next_pull = Clock::now() + chunk_interval; + while (!shutdown_) { + // get a chunk from the stream + in->pull_chunk_multiplexed(chunk, ×tamps, 1e-6); + write_chunk(); + if (wait_until_shutdown(next_pull)) break; next_pull += chunk_interval; - std::unique_lock cv_lock(shutdown_mut_); - if (shutdown_cv_.wait_until(cv_lock, next_pull, [this] { return shutdown_.load(); })) { - break; + } + + if (first_timestamp != 0.0) { + // one final non-blocking pull, so that samples already buffered in the inlet when the + // stop arrived end up in the file rather than being dropped + try { + in->pull_chunk_multiplexed(chunk, ×tamps, 0.0); + write_chunk(); + } catch (std::exception &e) { + // the inlet was closed under us during teardown; the footer matters more + std::cerr << "Could not drain stream " << streamid << " on stop: " << e.what() + << std::endl; } } - } catch (std::exception &e) { - std::cerr << "Error in transfer thread: " << e.what() << std::endl; - offset_shutdown = true; - shutdown_cv_.notify_all(); + } catch (std::exception &) { + stop_offsets(offset_shutdown); timed_join_or_detach(offset_thread); throw; } - offset_shutdown = true; - shutdown_cv_.notify_all(); + stop_offsets(offset_shutdown); timed_join_or_detach(offset_thread); } diff --git a/src/recording.h b/src/recording.h index 56fb4fc..9b63da2 100644 --- a/src/recording.h +++ b/src/recording.h @@ -5,13 +5,18 @@ #include #include #include +#include #include #include #include #include +#include #include +#include +#include #include #include +#include // timings in the recording process (e.g., rate of boundary chunks and for cases where a stream // hangs) approx. interval between boundary chunks @@ -20,6 +25,9 @@ const auto boundary_interval = std::chrono::seconds(10); const auto offset_interval = std::chrono::seconds(5); // approx. interval between resolves for outstanding streams on the watchlist, in seconds const double resolve_interval = 5; +// timeout of a single resolve attempt, in seconds; the rest of resolve_interval is spent in an +// interruptible wait so that a shutdown request need not wait out a resolve +const double resolve_timeout = 1; // approx. interval between pulling chunks from outlets const auto chunk_interval = std::chrono::milliseconds(500); // maximum waiting time for moving past the headers phase while recording @@ -29,15 +37,59 @@ 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 waiting time for a single time correction query +const auto max_time_correction_wait = std::chrono::seconds(2); +// blocking network calls are issued in slices of this length (in seconds) so that a shutdown +// request is noticed promptly instead of after the full timeout +const double network_poll_interval = 0.2; +// time granted to the stream threads to drain their inlets and write their footers before the +// inlets are forcibly closed +const auto teardown_grace = std::chrono::milliseconds(300); // maximum time that we wait to join a thread const auto max_join_wait = std::chrono::seconds(2); +// steady_clock (not high_resolution_clock, which is an alias for the wall clock in some standard +// libraries) so that waits are unaffected by clock adjustments +using Clock = std::chrono::steady_clock; + using streamid_t = uint32_t; -// pointer to a thread -using thread_p = std::unique_ptr; +/// thrown by the interruptible helpers when the recording is being torn down +class shutdown_requested : public std::runtime_error { +public: + explicit shutdown_requested(const std::string &what) : std::runtime_error(what) {} +}; + +/** + * A thread paired with a future that becomes ready once the thread body has returned. + * + * std::thread::join() blocks indefinitely, so polling it cannot enforce a deadline: a single call + * against a hung thread never comes back. The future can be waited on with a timeout, and only + * once it is ready do we join (which then returns promptly). A std::packaged_task future is used + * rather than std::async because the latter blocks in its future destructor. + */ +struct worker { + std::thread thread; + std::future done; +}; +// pointer to a worker thread +using worker_p = std::unique_ptr; + +/// start a worker thread running fn +template worker_p spawn_worker(F &&fn) { + auto task = std::make_shared>(std::forward(fn)); + auto w = std::make_unique(); + w->done = task->get_future(); + // the task is kept alive by the lambda, so the worker may be detached safely + w->thread = std::thread([task] { (*task)(); }); + return w; +} + // pointer to a stream inlet using inlet_p = std::shared_ptr; +// pointer to a per-stream flag asking that stream's offset thread to finish. Shared rather than +// referenced so that an offset thread which had to be detached cannot outlive its flag. +using offset_flag_p = std::shared_ptr>; // a list of clock offset estimates (time,value) using offset_list = std::list>; // a map from streamid to offset_list @@ -55,11 +107,11 @@ class recording { /** * Construct a new background recording process. * @param filename The file name to record to (should end in .xdf). - * @param streams An array of LSL streaminfo's that identify the set of streams to record into + * @param streams An array of LSL streaminfos that identify the set of streams to record into *the file. * @param watchfor An optional "watchlist" of LSL query predicates (see lsl::resolve_bypred) to *resolve streams to record from. This can be a specific stream that you know should be recorded - *but is not yet online, or a more generic query (e.g., "record from everything that's out + *but is not yet online, or a more generic query (e.g., "record from everything that is out *there"). * @param collect_offsets Whether to collect time offset measurements periodically. */ @@ -72,6 +124,8 @@ class recording { */ ~recording(); + /// Ask all recording threads to wrap up. Returns immediately; the threads are joined by the + /// destructor. void requestStop() noexcept; private: @@ -86,11 +140,11 @@ class recording { std::atomic streamid_; // the highest streamid allocated so far // 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 + std::atomic shutdown_; // whether we are trying to shut down + std::condition_variable + shutdown_cv_; // signals shutdown so that every interruptible wait returns at once + std::mutex shutdown_mut_; // protects publication of shutdown_ and of the per-stream offset + // shutdown flags, which the shutdown_cv_ predicates read under it 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 @@ -103,14 +157,19 @@ class recording { // recording jobs and are now ready to write a footer std::mutex phase_mut_; // a mutex to protect the phase state + // inlets with potentially pending network I/O, to be aborted if their thread does not stop in + // time + std::vector active_inlets_; + std::mutex inlets_mut_; // a mutex to protect the active inlet list + // data structure to collect the time offsets for every stream offset_lists offset_lists_; // the clock offset lists for each stream (to be written into the footer) std::mutex offset_mut_; // a mutex to protect the offset lists // data for shutdown / final joining - std::list stream_threads_; // the spawned stream handling threads - thread_p boundary_thread_; // the spawned boundary-recording thread + std::list stream_threads_; // the spawned stream handling threads + worker_p boundary_thread_; // the spawned boundary-recording thread // for enabling online sync options std::map sync_options_by_stream_; @@ -135,7 +194,7 @@ class recording { // record ClockOffset chunks from a given stream void record_offsets( - streamid_t streamid, const inlet_p &in, std::atomic &offset_shutdown) noexcept; + streamid_t streamid, inlet_p in, offset_flag_p offset_shutdown) noexcept; // sample collection loop for a numeric stream @@ -143,6 +202,37 @@ class recording { void typed_transfer_loop(streamid_t streamid, double srate, const inlet_p &in, double &first_timestamp, double &last_timestamp, uint64_t &sample_count); + // === interruptible waiting & bounded network calls === + + /// wait until deadline, returning true if the wait was cut short by a shutdown request + /// @param extra an optional additional flag (e.g. a per-stream offset shutdown) that also ends + /// the wait + bool wait_until_shutdown(Clock::time_point deadline, const std::atomic *extra = nullptr); + + /// wait for timeout, returning true if the wait was cut short by a shutdown request + bool wait_for_shutdown(Clock::duration timeout, const std::atomic *extra = nullptr) { + return wait_until_shutdown(Clock::now() + timeout, extra); + } + + /// publish a per-stream offset shutdown flag and wake the corresponding offset thread + void stop_offsets(const offset_flag_p &offset_shutdown) noexcept; + + /// subscribe to a stream, giving up after max_open_wait + /// @return whether the subscription completed (if not, it will take place later) + /// @throws shutdown_requested if the recording was stopped while subscribing + bool open_inlet(const inlet_p &in); + + /// retrieve the full stream info, including the extended description + /// @throws shutdown_requested if the recording was stopped while retrieving the metadata + lsl::stream_info fetch_info(const inlet_p &in); + + // === inlet bookkeeping === + + void register_inlet(const inlet_p &in); + void unregister_inlet(const inlet_p &in) noexcept; + /// close every registered inlet, aborting any blocking socket call in progress + void close_active_inlets() noexcept; + // === phase registration & condition checks === // writing is coordinated across threads in three phases to keep the file chunks sorted @@ -159,9 +249,9 @@ class recording { void leave_footers_phase(bool) { /* Nothing to do. Ignore warning. */ } - /// a condition that indicates that we're ready to write streaming content into the file + /// a condition that indicates that we are ready to write streaming content into the file bool ready_for_streaming() const { return headers_to_finish_ <= 0; } - /// a condition that indicates that we're ready to write footers into the file + /// a condition that indicates that we are ready to write footers into the file bool ready_for_footers() const { return streaming_to_finish_ <= 0 && headers_to_finish_ <= 0; } /// allocate a fresh stream id From c81229299bc689d5404f7d2604133e6d9f467b3f Mon Sep 17 00:00:00 2001 From: Stefan Appelhoff Date: Sun, 20 Sep 2026 21:25:36 +0200 Subject: [PATCH 4/4] Keep recording state alive for threads that outlive teardown, and drain a gated stream Addresses the second round of review on the teardown work. The future-based join added in the previous commit is the first version that can actually reach the detach path, because the blocking join() it replaced never returned for a hung thread. Detached threads ran on a raw `this` and wrote into the file, the mutexes and the offset lists, all of which the destructor had already destroyed. Recording state and the thread bodies now live in a shared implementation object that every thread holds a reference to, so the state outlives a teardown that had to leave a thread running, and the last one to finish closes the file. A harness that stalls a stream thread past the join deadline faults with an access violation and writes no footer before this change, and exits cleanly with both footers intact after it. The gate that lets a stream wait for another stream's header is released on shutdown, which meant a stream could reach its transfer loop with the shutdown already set, pull no first sample, and then skip the final drain because it had no first timestamp to compare against -- writing a zero-sample footer although its inlet had been subscribed and buffering the whole time. The first sample now anchors the stream wherever it arrives from, including from the drain, and the drain is unconditional. Two things found while testing this: Splitting the time correction query into network_poll_interval slices was unsound: the query needs a round trip to complete, so restarting it every 200 ms means it need never finish. It goes back to a single call with the full budget, as before the teardown work. Teardown stays bounded because the transfer thread now stops waiting for the offset thread after the teardown grace period and leaves it running, which is safe now that a thread left running keeps its state alive. Recording threads all logged through unsynchronised << chains, so their output interleaved mid-line. Lines are now composed and written under a mutex. scripts/test_recording_teardown.py gains a case for the gated stream: it holds one stream at the headers gate behind another whose source has gone away, buffers 40 samples into it and stops. Before the drain fix that case records 0 of 40. --- scripts/test_recording_teardown.py | 72 ++++- src/clirecorder.cpp | 2 + src/recording.cpp | 460 +++++++++++++++++++++++------ src/recording.h | 226 +------------- 4 files changed, 443 insertions(+), 317 deletions(-) diff --git a/scripts/test_recording_teardown.py b/scripts/test_recording_teardown.py index 43a62f2..aa0ceee 100644 --- a/scripts/test_recording_teardown.py +++ b/scripts/test_recording_teardown.py @@ -65,9 +65,12 @@ class Recorder: recorder subscribed, so pushing too early silently loses samples. """ - def __init__(self, cli_path, xdf_path): + def __init__(self, cli_path, xdf_path, stream_order=NAMES): + # the recorder spawns one thread per stream in the order given here, and each thread + # registers with the headers phase as it starts. A case that needs one stream to be held + # at the headers-to-streaming gate by another therefore has to control this order. self._proc = subprocess.Popen( - [cli_path, xdf_path, f"name='{EEG_NAME}'", f"name='{MARKER_NAME}'"], + [cli_path, xdf_path] + [f"name='{name}'" for name in stream_order], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, @@ -99,6 +102,9 @@ def wait_for(self, needles, timeout=20.0): def wait_until_collecting(self): self.wait_for([f"Started data collection for stream {name}." for name in NAMES]) + def saw(self, needle): + return needle in "\n".join(self.lines) + def stop(self): """Send the quit key and return how long the recorder took to exit.""" started = time.perf_counter() @@ -128,9 +134,9 @@ def terminate(self): @contextlib.contextmanager -def recorder(cli_path, xdf_path): +def recorder(cli_path, xdf_path, stream_order=NAMES): """Run LabRecorderCLI over both test streams, making sure it is gone afterwards.""" - rec = Recorder(cli_path, xdf_path) + rec = Recorder(cli_path, xdf_path, stream_order) try: yield rec finally: @@ -218,6 +224,13 @@ def push_eeg_for(outlet, seconds): time.sleep(1.0 / EEG_RATE) +def push_eeg_samples(outlet, count): + """Push exactly count EEG samples at the nominal rate.""" + for value in range(count): + outlet.push_sample([float(value)] * EEG_CHANNELS) + time.sleep(1.0 / EEG_RATE) + + def case_normal_stop(cli_path, xdf_path, max_stop): """Record both streams for two seconds, then stop.""" eeg, markers = make_outlets() @@ -315,12 +328,63 @@ def case_no_buffered_samples_lost(cli_path, xdf_path, max_stop): del eeg, markers +def case_gated_stream_is_drained(cli_path, xdf_path, max_stop): + """A stream held at the headers gate must still write what its inlet buffered. + + A stream that is through its own header waits for every other stream's header before it may + write data. If a stop arrives while it waits there, it reaches its transfer loop with the + shutdown already set and never pulls a first sample -- but its inlet has been subscribed and + buffering the whole time, so that data has to be drained on the way out. + + The marker stream is listed first so its thread registers with the headers phase before the + EEG thread can leave it, then it is taken away so its header never arrives and the EEG thread + stays at the gate. + """ + sample_count = 40 + eeg, markers = make_outlets() + time.sleep(SETTLE) + with recorder(cli_path, xdf_path, stream_order=(MARKER_NAME, EEG_NAME)) as rec: + # The recorder resolves for a second before it opens anything, so the marker outlet has + # to survive long enough to be resolved and be gone before its metadata is fetched. + # Waiting for the "Found" line instead would be too late: it is printed when the resolve + # returns, microseconds before the inlets are opened. + time.sleep(0.6) + del markers + rec.wait_for([f"Found {name}" for name in NAMES]) + rec.wait_for([f"Received header for stream {EEG_NAME}."]) + check( + not rec.saw(f"Started data collection for stream {EEG_NAME}."), + "precondition not met: the EEG stream was not held at the headers gate" + + ( + " (the marker header arrived before its outlet was removed)" + if rec.saw(f"Received header for stream {MARKER_NAME}.") + else "" + ), + ) + + push_eeg_samples(eeg, sample_count) + time.sleep(SETTLE) + duration = rec.stop() + + check_stop(duration, max_stop) + streams, _ = load_xdf_strict(xdf_path) + eeg_stream = stream_by_name(streams, EEG_NAME) + recorded = len(eeg_stream["time_series"]) + check( + recorded == sample_count, + f"{sample_count} samples were buffered at the gate but {recorded} were recorded", + ) + check_footer(eeg_stream) + del eeg + + CASES = [ ("normal stop", case_normal_stop), ("stop before first sample", case_stop_before_first_sample), ("stop while subscribing", case_stop_while_subscribing), ("repeated shutdown", case_repeated_shutdown), ("no buffered samples lost", case_no_buffered_samples_lost), + ("gated stream is drained", case_gated_stream_is_drained), ] diff --git a/src/clirecorder.cpp b/src/clirecorder.cpp index 59e3736..55772fa 100644 --- a/src/clirecorder.cpp +++ b/src/clirecorder.cpp @@ -1,6 +1,8 @@ #include "recording.h" #include "xdfwriter.h" +#include + int main(int argc, char **argv) { if (argc < 3 || (argc == 2 && std::string(argv[1]) == "-h")) { std::cout << "Usage: " << argv[0] << " outputfile.xdf 'searchstr' ['searchstr2' ...]\n\n" diff --git a/src/recording.cpp b/src/recording.cpp index 6548274..ea84fba 100644 --- a/src/recording.cpp +++ b/src/recording.cpp @@ -1,17 +1,127 @@ #include "recording.h" //#include "conversions.h" +#include "xdfwriter.h" #include +#include +#include +#include +#include +#include +#include +#include +#include #include #include +#include +#include #ifdef XDFZ_SUPPORT #include #include #include #endif +// timings in the recording process (e.g., rate of boundary chunks and for cases where a stream +// hangs) approx. interval between boundary chunks +const auto boundary_interval = std::chrono::seconds(10); +// approx. interval between offset measurements +const auto offset_interval = std::chrono::seconds(5); +// approx. interval between resolves for outstanding streams on the watchlist, in seconds +const double resolve_interval = 5; +// timeout of a single resolve attempt, in seconds; the rest of resolve_interval is spent in an +// interruptible wait so that a shutdown request need not wait out a resolve +const double resolve_timeout = 1; +// approx. interval between pulling chunks from outlets +const auto chunk_interval = std::chrono::milliseconds(500); +// maximum waiting time for moving past the headers phase while recording +const auto max_headers_wait = std::chrono::seconds(10); +// maximum waiting time for moving into the footers phase while recording +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 waiting time for a single time correction query, in seconds +const double max_time_correction_wait = 2; +// blocking network calls are issued in slices of this length (in seconds) so that a shutdown +// request is noticed promptly instead of after the full timeout +const double network_poll_interval = 0.2; +// time granted to the stream threads to drain their inlets and write their footers before the +// inlets are forcibly closed +const auto teardown_grace = std::chrono::milliseconds(300); +// maximum time that we wait to join a thread +const auto max_join_wait = std::chrono::seconds(2); + +// steady_clock (not high_resolution_clock, which is an alias for the wall clock in some standard +// libraries) so that waits are unaffected by clock adjustments +using Clock = std::chrono::steady_clock; + +using streamid_t = uint32_t; + +/// thrown by the interruptible helpers when the recording is being torn down +class shutdown_requested : public std::runtime_error { +public: + explicit shutdown_requested(const std::string &what) : std::runtime_error(what) {} +}; + +/** + * A thread paired with a future that becomes ready once the thread body has returned. + * + * std::thread::join() blocks indefinitely, so polling it cannot enforce a deadline: a single call + * against a hung thread never comes back. The future can be waited on with a timeout, and only + * once it is ready do we join (which then returns promptly). A std::packaged_task future is used + * rather than std::async because the latter blocks in its future destructor. + */ +struct worker { + std::thread thread; + std::future done; +}; +// pointer to a worker thread +using worker_p = std::unique_ptr; + +/// start a worker thread running fn +template worker_p spawn_worker(F &&fn) { + auto task = std::make_shared>(std::forward(fn)); + auto w = std::make_unique(); + w->done = task->get_future(); + // the task is kept alive by the lambda, so the worker may be detached safely + w->thread = std::thread([task] { (*task)(); }); + return w; +} + +// pointer to a stream inlet +using inlet_p = std::shared_ptr; +// pointer to a per-stream flag asking that stream's offset thread to finish. Shared rather than +// referenced so that an offset thread which had to be detached cannot outlive its flag. +using offset_flag_p = std::shared_ptr>; +// a list of clock offset estimates (time,value) +using offset_list = std::list>; +// a map from streamid to offset_list +using offset_lists = std::map; + namespace { +/// Write one line, atomically with respect to the other recording threads. +/// +/// Every stream has its own thread and they all report progress; an unsynchronised chain of << +/// lets two of them interleave in the middle of a line, which garbles the log and defeats anything +/// that reads it. +template void log_line(std::ostream &out, Args &&...args) { + std::ostringstream line; + (line << ... << std::forward(args)); + line << '\n'; + static std::mutex log_mut; + std::lock_guard lock(log_mut); + out << line.str() << std::flush; +} + +template void log_out(Args &&...args) { + log_line(std::cout, std::forward(args)...); +} + +template void log_err(Args &&...args) { + log_line(std::cerr, std::forward(args)...); +} + // time spent waiting between two resolves of a watchlist query; the resolve itself already takes // resolve_timeout, so together they keep the resolve_interval cadence const auto resolve_pause = std::chrono::duration_cast( @@ -52,7 +162,7 @@ inline void timed_join_or_detach(worker_p &w, std::chrono::milliseconds duration if (!timed_join(w, duration)) { w->thread.detach(); w.reset(); - std::cerr << "Thread didn't join in time!" << std::endl; + log_err("Thread didn't join in time!"); } } @@ -82,31 +192,181 @@ inline void timed_join_or_detach( std::list &workers, std::chrono::milliseconds duration = max_join_wait) { timed_join_some(workers, duration); if (!workers.empty()) { - std::cout << workers.size() << " stream threads still running!" << std::endl; + log_out(workers.size(), " stream threads still running!"); for (auto &w : workers) w->thread.detach(); workers.clear(); } } -recording::recording(const std::string &filename, const std::vector &streams, - const std::vector &watchfor, std::map syncOptions, - bool collect_offsets) - : file_(filename), offsets_enabled_(collect_offsets), unsorted_(false), streamid_(0), - shutdown_(false), headers_to_finish_(0), streaming_to_finish_(0), - sync_options_by_stream_(std::move(syncOptions)) { +/** + * The recording state, and the thread bodies that operate on it. + * + * Every recording thread holds a shared_ptr to this, as does the recording object. A thread that + * could not be joined within the teardown deadline is left running rather than blocking the + * caller, which is typically the UI thread, so the state it writes into has to be able to outlive + * the recording object. Whoever drops the last reference destroys it, and that is what closes the + * file. + */ +struct recording::impl : std::enable_shared_from_this { + impl(const std::string &filename, std::map syncOptions, bool collect_offsets) + : file_(filename), offsets_enabled_(collect_offsets), unsorted_(false), streamid_(0), + shutdown_(false), headers_to_finish_(0), streaming_to_finish_(0), + sync_options_by_stream_(std::move(syncOptions)) {} + + /// Deliberately joins nothing: the last recording thread to finish drops the final reference, + /// so this runs on that very thread and joining here would be a self-join. stop_and_join() + /// leaves the worker containers empty, so there is nothing left to clean up. + ~impl() = default; + + /// Spawn the recording threads. Separate from the constructor because the threads need a + /// shared_ptr to this, which shared_from_this() cannot hand out during construction. + void start( + const std::vector &streams, const std::vector &watchfor); + + /// Ask the threads to finish and wait a bounded amount of time for them, leaving any that are + /// still stuck running. Called from the recording object, never from a recording thread. + void stop_and_join() noexcept; + + void requestStop() noexcept; + + // the file stream + XDFWriter file_; // the file output stream + // static information + bool offsets_enabled_; // whether to collect time offset information alongside with the stream + // contents + bool unsorted_; // whether this file may contain unsorted chunks (e.g., of late streams) + + // streamid allocation + std::atomic streamid_; // the highest streamid allocated so far + + // phase-of-recording state (headers, streaming data, or footers) + std::atomic shutdown_; // whether we are trying to shut down + std::condition_variable + shutdown_cv_; // signals shutdown so that every interruptible wait returns at once + std::mutex shutdown_mut_; // protects publication of shutdown_ and of the per-stream offset + // shutdown flags, which the shutdown_cv_ predicates read under it + 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 + // phase (i.e., are not yet ready for writing their footer) + std::condition_variable + ready_for_streaming_; // condition variable signaling that all streams have finished writing + // their headers and are now ready to write streaming content + std::condition_variable + ready_for_footers_; // condition variable signaling that all streams have finished their + // recording jobs and are now ready to write a footer + std::mutex phase_mut_; // a mutex to protect the phase state + + // inlets with potentially pending network I/O, to be aborted if their thread does not stop in + // time + std::vector active_inlets_; + std::mutex inlets_mut_; // a mutex to protect the active inlet list + + // data structure to collect the time offsets for every stream + offset_lists + offset_lists_; // the clock offset lists for each stream (to be written into the footer) + std::mutex offset_mut_; // a mutex to protect the offset lists + + // data for shutdown / final joining + std::list stream_threads_; // the spawned stream handling threads + worker_p boundary_thread_; // the spawned boundary-recording thread + + // for enabling online sync options + std::map sync_options_by_stream_; + + // === recording thread functions === + + /// record from results of a query (spawn a recording thread for every result produced by the + /// query) + /// @param query The query string + void record_from_query_results(const std::string &query); + + /// record from a given stream (identified by its streaminfo) + /// @param src the stream_info from which to record + /// @param phase_locked whether this is a stream that is locked to the phases (1. Headers, 2. + /// Streaming Content, 3. Footers) + /// Late-added streams (e.g. forgotten devices) are not phase-locked. + void record_from_streaminfo(const lsl::stream_info &src, bool phase_locked); + + /// record boundary markers every few seconds + void record_boundaries(); + + // record ClockOffset chunks from a given stream + void record_offsets(streamid_t streamid, inlet_p in, offset_flag_p offset_shutdown) noexcept; + + // sample collection loop for a numeric stream + template + void typed_transfer_loop(streamid_t streamid, double srate, const inlet_p &in, + double &first_timestamp, double &last_timestamp, uint64_t &sample_count); + + // === interruptible waiting & bounded network calls === + + /// wait until deadline, returning true if the wait was cut short by a shutdown request + /// @param extra an optional additional flag (e.g. a per-stream offset shutdown) that also ends + /// the wait + bool wait_until_shutdown(Clock::time_point deadline, const std::atomic *extra = nullptr); + + /// wait for timeout, returning true if the wait was cut short by a shutdown request + bool wait_for_shutdown(Clock::duration timeout, const std::atomic *extra = nullptr) { + return wait_until_shutdown(Clock::now() + timeout, extra); + } + + /// publish a per-stream offset shutdown flag and wake the corresponding offset thread + void stop_offsets(const offset_flag_p &offset_shutdown) noexcept; + + /// subscribe to a stream, giving up after max_open_wait + /// @return whether the subscription completed (if not, it will take place later) + bool open_inlet(const inlet_p &in); + + /// retrieve the full stream info, including the extended description + /// @throws shutdown_requested if the recording was stopped while retrieving the metadata + lsl::stream_info fetch_info(const inlet_p &in); + + // === inlet bookkeeping === + + void register_inlet(const inlet_p &in); + void unregister_inlet(const inlet_p &in) noexcept; + /// close every registered inlet, aborting any blocking socket call in progress + void close_active_inlets() noexcept; + + // === phase registration & condition checks === + // writing is coordinated across threads in three phases to keep the file chunks sorted + + void enter_headers_phase(bool phase_locked); + void leave_headers_phase(bool phase_locked); + void enter_streaming_phase(bool phase_locked); + void leave_streaming_phase(bool phase_locked); + void enter_footers_phase(bool phase_locked); + void leave_footers_phase(bool) { /* Nothing to do. Ignore warning. */ + } + + /// a condition that indicates that we are ready to write streaming content into the file + bool ready_for_streaming() const { return headers_to_finish_ <= 0; } + /// a condition that indicates that we are ready to write footers into the file + bool ready_for_footers() const { return streaming_to_finish_ <= 0 && headers_to_finish_ <= 0; } + + /// allocate a fresh stream id + streamid_t fresh_streamid() { return ++streamid_; } +}; + +void recording::impl::start( + const std::vector &streams, const std::vector &watchfor) { + // the threads hold a reference to us, so the state they write into outlives a teardown that + // had to leave one of them running + auto self = shared_from_this(); // create a recording thread for each stream for (const auto &stream : streams) stream_threads_.emplace_back( - spawn_worker([this, stream] { record_from_streaminfo(stream, true); })); + spawn_worker([self, stream] { self->record_from_streaminfo(stream, true); })); // create a resolve-and-record thread for each item in the watchlist for (const auto &query : watchfor) stream_threads_.emplace_back( - spawn_worker([this, query] { record_from_query_results(query); })); + spawn_worker([self, query] { self->record_from_query_results(query); })); // create a boundary chunk writer thread - boundary_thread_ = spawn_worker([this] { record_boundaries(); }); + boundary_thread_ = spawn_worker([self] { self->record_boundaries(); }); } -recording::~recording() { +void recording::impl::stop_and_join() noexcept { try { // set the shutdown flag (from now on no more new streams) and wake every waiting thread requestStop(); @@ -120,13 +380,30 @@ recording::~recording() { timed_join_or_detach(stream_threads_, max_join_wait); } timed_join_or_detach(boundary_thread_, max_join_wait); - std::cout << "Closing the file." << std::endl; + log_out("Closing the file."); } catch (std::exception &e) { - std::cout << "Error while closing the recording: " << e.what() << std::endl; + log_out("Error while closing the recording: ", e.what()); + } +} + +recording::recording(const std::string &filename, const std::vector &streams, + const std::vector &watchfor, std::map syncOptions, + bool collect_offsets) + : impl_(std::make_shared(filename, std::move(syncOptions), collect_offsets)) { + try { + impl_->start(streams, watchfor); + } catch (...) { + // some threads may already be running, and our destructor will not run if we throw + impl_->stop_and_join(); + throw; } } -void recording::requestStop() noexcept { +recording::~recording() { impl_->stop_and_join(); } + +void recording::requestStop() noexcept { impl_->requestStop(); } + +void recording::impl::requestStop() noexcept { { // publish the flag under the mutex that the shutdown_cv_ predicates read it under: a // waiter that has just evaluated its predicate as false would otherwise miss the @@ -142,13 +419,13 @@ void recording::requestStop() noexcept { ready_for_footers_.notify_all(); } -bool recording::wait_until_shutdown(Clock::time_point deadline, const std::atomic *extra) { +bool recording::impl::wait_until_shutdown(Clock::time_point deadline, const std::atomic *extra) { std::unique_lock lock(shutdown_mut_); return shutdown_cv_.wait_until( lock, deadline, [this, extra] { return shutdown_.load() || (extra && extra->load()); }); } -void recording::stop_offsets(const offset_flag_p &offset_shutdown) noexcept { +void recording::impl::stop_offsets(const offset_flag_p &offset_shutdown) noexcept { { std::lock_guard lock(shutdown_mut_); *offset_shutdown = true; @@ -156,30 +433,30 @@ void recording::stop_offsets(const offset_flag_p &offset_shutdown) noexcept { shutdown_cv_.notify_all(); } -void recording::register_inlet(const inlet_p &in) { +void recording::impl::register_inlet(const inlet_p &in) { std::lock_guard lock(inlets_mut_); active_inlets_.push_back(in); } -void recording::unregister_inlet(const inlet_p &in) noexcept { +void recording::impl::unregister_inlet(const inlet_p &in) noexcept { if (!in) return; std::lock_guard lock(inlets_mut_); active_inlets_.erase( std::remove(active_inlets_.begin(), active_inlets_.end(), in), active_inlets_.end()); } -void recording::close_active_inlets() noexcept { +void recording::impl::close_active_inlets() noexcept { std::lock_guard lock(inlets_mut_); for (auto &in : active_inlets_) { try { in->close_stream(); } catch (std::exception &e) { - std::cerr << "Error while closing an inlet: " << e.what() << std::endl; + log_err("Error while closing an inlet: ", e.what()); } } } -bool recording::open_inlet(const inlet_p &in) { +bool recording::impl::open_inlet(const inlet_p &in) { // subscribe in short slices: a single open_stream(max_open_wait) would keep us from noticing a // stop for up to max_open_wait seconds const auto deadline = Clock::now() + seconds_to_duration(max_open_wait); @@ -192,7 +469,7 @@ bool recording::open_inlet(const inlet_p &in) { return false; } -lsl::stream_info recording::fetch_info(const inlet_p &in) { +lsl::stream_info recording::impl::fetch_info(const inlet_p &in) { // the metadata receiver is separate from the data receiver, so close_stream() does not abort // this call; poll in short slices instead, or an unreachable source blocks us indefinitely. // A stop does not cut this off immediately: a source that is still reachable gets a short @@ -208,12 +485,12 @@ lsl::stream_info recording::fetch_info(const inlet_p &in) { throw shutdown_requested("stopped while retrieving the stream metadata"); } -void recording::record_from_query_results(const std::string &query) { +void recording::impl::record_from_query_results(const std::string &query) { try { std::set known_uids; // set of previously seen stream uid's std::set known_source_ids; // set of previously seen source id's std::list threads; // our spawned threads - std::cout << "Watching for a stream with properties " << query << std::endl; + log_out("Watching for a stream with properties ", query); while (!shutdown_) { // periodically re-resolve the query. The resolve itself is kept short and the rest of // the interval is spent in an interruptible wait, so a stop is noticed quickly. @@ -226,11 +503,11 @@ void recording::record_from_query_results(const std::string &query) { // and doesn't have a previously seen source id... if (!result.source_id().empty() && (!known_source_ids.count(result.source_id()))) { - std::cout << "Found a new stream named " << result.name() - << ", adding it to the recording." << std::endl; + log_out("Found a new stream named ", result.name(), ", adding it to the recording."); // start a new recording thread - threads.emplace_back(spawn_worker( - [this, result] { record_from_streaminfo(result, false); })); + threads.emplace_back(spawn_worker([self = shared_from_this(), result] { + self->record_from_streaminfo(result, false); + })); // ... and add it to the lists of known id's known_uids.insert(result.uid()); if (!result.source_id().empty()) @@ -242,11 +519,11 @@ void recording::record_from_query_results(const std::string &query) { // wait for all our threads to join timed_join_or_detach(threads, max_join_wait); } catch (std::exception &e) { - std::cout << "Error in the record_from_query_results thread: " << e.what() << std::endl; + log_out("Error in the record_from_query_results thread: ", e.what()); } } -void recording::record_from_streaminfo(const lsl::stream_info &src, bool phase_locked) { +void recording::impl::record_from_streaminfo(const lsl::stream_info &src, bool phase_locked) { inlet_p in; try { // initialised here because a stream that fails mid-recording still writes a footer @@ -267,19 +544,17 @@ void recording::record_from_streaminfo(const lsl::stream_info &src, bool phase_l if (it != sync_options_by_stream_.end()) in->set_postprocessing(it->second); if (open_inlet(in)) - std::cout << "Opened the stream " << src.name() << "." << std::endl; + log_out("Opened the stream ", src.name(), "."); else if (!shutdown_) - std::cout - << "Subscribing to the stream " << src.name() - << " is taking relatively long; collection from this stream will be delayed." - << std::endl; + log_out("Subscribing to the stream ", src.name(), + " is taking relatively long; collection from this stream will be delayed."); // retrieve the stream header & get its XML version. The nominal rate is taken from // the same info, saving a second round trip to the source. const lsl::stream_info info = fetch_info(in); nominal_srate = info.nominal_srate(); file_.write_stream_header(streamid, info.as_xml()); - std::cout << "Received header for stream " << src.name() << "." << std::endl; + log_out("Received header for stream ", src.name(), "."); leave_headers_phase(phase_locked); } catch (std::exception &) { @@ -296,7 +571,7 @@ void recording::record_from_streaminfo(const lsl::stream_info &src, bool phase_l // "forgot to turn on" before the recording started; in that case the file would have to // be post-processed to be in properly sorted (seekable) format enter_streaming_phase(phase_locked); - std::cout << "Started data collection for stream " << src.name() << "." << std::endl; + log_out("Started data collection for stream ", src.name(), "."); // now write the actual sample chunks... switch (src.channel_format()) { @@ -335,8 +610,7 @@ void recording::record_from_streaminfo(const lsl::stream_info &src, bool phase_l leave_streaming_phase(phase_locked); // the header is already on disk, so fall through to the footer instead of leaving the // stream without one - std::cerr << "Error while recording from " << src.name() << ": " << e.what() - << std::endl; + log_err("Error while recording from ", src.name(), ": ", e.what()); } // --- footers phase @@ -362,54 +636,49 @@ void recording::record_from_streaminfo(const lsl::stream_info &src, bool phase_l } file_.write_stream_footer(streamid, footer.str()); - std::cout << "Wrote footer for stream " << src.name() << "." << std::endl; + log_out("Wrote footer for stream ", src.name(), "."); leave_footers_phase(phase_locked); } catch (std::exception &) { leave_footers_phase(phase_locked); throw; } } catch (shutdown_requested &e) { - std::cout << "Recording from " << src.name() << " ended: " << e.what() << std::endl; + log_out("Recording from ", src.name(), " ended: ", e.what()); } catch (std::exception &e) { - std::cout << "Error in the record_from_streaminfo thread: " << e.what() << std::endl; + log_out("Error in the record_from_streaminfo thread: ", e.what()); } unregister_inlet(in); } -void recording::record_boundaries() { +void recording::impl::record_boundaries() { try { while (!shutdown_) { if (wait_for_shutdown(boundary_interval)) break; file_.write_boundary_chunk(); } } catch (std::exception &e) { - std::cout << "Error in the record_boundaries thread: " << e.what() << std::endl; + log_out("Error in the record_boundaries thread: ", e.what()); } } -void recording::record_offsets( +void recording::impl::record_offsets( streamid_t streamid, inlet_p in, offset_flag_p offset_shutdown) noexcept { try { while (!shutdown_ && !*offset_shutdown) { // sleep for the interval if (wait_for_shutdown(offset_interval, offset_shutdown.get())) break; - // query the time offset, again in short slices so that a stop is noticed promptly - double offset = 0, now = 0; - bool have_offset = false; - const auto deadline = Clock::now() + max_time_correction_wait; - while (!shutdown_ && !*offset_shutdown && Clock::now() < deadline) { - try { - offset = in->time_correction(network_poll_interval); - now = lsl::local_clock(); - have_offset = true; - break; - } catch (lsl::timeout_error &) {} - } - if (!have_offset) { - if (shutdown_ || *offset_shutdown) break; - std::cerr << "Timeout in time correction query for stream " << streamid - << std::endl; + // Query the time offset in one call with the whole budget, not in short slices: the + // query needs a round trip to complete, and restarting it every network_poll_interval + // means it never finishes, so no offset is ever recorded. Teardown does not depend on + // this returning quickly -- the transfer thread stops waiting for us after + // teardown_grace and leaves us running, and we keep the file alive while we do. + double offset, now; + try { + offset = in->time_correction(max_time_correction_wait); + now = lsl::local_clock(); + } catch (lsl::timeout_error &) { + log_err("Timeout in time correction query for stream ", streamid); continue; } @@ -419,19 +688,19 @@ void recording::record_offsets( offset_lists_[streamid].emplace_back(now - offset, offset); } } catch (std::exception &e) { - std::cout << "Error in the record_offsets thread: " << e.what() << std::endl; + log_out("Error in the record_offsets thread: ", e.what()); } - std::cout << "Offsets thread is finished" << std::endl; + log_out("Offsets thread is finished"); } -void recording::enter_headers_phase(bool phase_locked) { +void recording::impl::enter_headers_phase(bool phase_locked) { if (phase_locked) { std::lock_guard lock(phase_mut_); headers_to_finish_++; } } -void recording::leave_headers_phase(bool phase_locked) { +void recording::impl::leave_headers_phase(bool phase_locked) { if (phase_locked) { std::unique_lock lock(phase_mut_); headers_to_finish_--; @@ -440,7 +709,7 @@ void recording::leave_headers_phase(bool phase_locked) { } } -void recording::enter_streaming_phase(bool phase_locked) { +void recording::impl::enter_streaming_phase(bool phase_locked) { if (phase_locked) { std::unique_lock lock(phase_mut_); // on shutdown the gate is dropped: the transfer loop exits immediately anyway, and waiting @@ -452,7 +721,7 @@ void recording::enter_streaming_phase(bool phase_locked) { } } -void recording::leave_streaming_phase(bool phase_locked) { +void recording::impl::leave_streaming_phase(bool phase_locked) { if (phase_locked) { std::unique_lock lock(phase_mut_); streaming_to_finish_--; @@ -461,7 +730,7 @@ void recording::leave_streaming_phase(bool phase_locked) { } } -void recording::enter_footers_phase(bool phase_locked) { +void recording::impl::enter_footers_phase(bool phase_locked) { if (phase_locked) { std::unique_lock lock(phase_mut_); // see enter_streaming_phase: a footer written slightly out of order beats no footer at all @@ -471,13 +740,14 @@ void recording::enter_footers_phase(bool phase_locked) { } template -void recording::typed_transfer_loop(streamid_t streamid, double srate, const inlet_p &in, +void recording::impl::typed_transfer_loop(streamid_t streamid, double srate, const inlet_p &in, double &first_timestamp, double &last_timestamp, uint64_t &sample_count) { // optionally start an offset collection thread for this stream auto offset_shutdown = std::make_shared>(false); + auto self = shared_from_this(); worker_p offset_thread(offsets_enabled_ - ? spawn_worker([this, streamid, in, offset_shutdown] { - record_offsets(streamid, in, offset_shutdown); + ? spawn_worker([self, streamid, in, offset_shutdown] { + self->record_offsets(streamid, in, offset_shutdown); }) : nullptr); try { @@ -491,6 +761,12 @@ void recording::typed_transfer_loop(streamid_t streamid, double srate, const inl auto write_chunk = [&] { if (timestamps.empty()) return; for (double &ts : timestamps) { + if (first_timestamp == 0.0) { + // the first sample anchors the stream and is written verbatim: at a nominal + // interval of zero the deduction below would otherwise zero out its timestamp + first_timestamp = last_timestamp = ts; + continue; + } // if the time stamp can be deduced from the previous one... if (last_timestamp + sample_interval == ts) { last_timestamp = ts + sample_interval; @@ -502,16 +778,15 @@ void recording::typed_transfer_loop(streamid_t streamid, double srate, const inl sample_count += timestamps.size(); }; - // Pull the first sample + // Wait for the first sample, unless the stop got here first. A stream held at the headers + // gate reaches this point with the shutdown already set, having pulled nothing, while its + // inlet has been subscribed and buffering the whole time -- the drain below picks that up. first_timestamp = 0.0; - while (!shutdown_ && first_timestamp == 0.0) - first_timestamp = last_timestamp = in->pull_sample(chunk, network_poll_interval); - if (first_timestamp != 0.0) { - // written directly: the very first sample anchors the stream and must keep its - // timestamp even when the nominal interval is zero - timestamps.assign(1, first_timestamp); - file_.write_data_chunk(streamid, timestamps, chunk, (uint32_t)in->get_channel_count()); - sample_count += timestamps.size(); + while (!shutdown_ && first_timestamp == 0.0) { + const double ts = in->pull_sample(chunk, network_poll_interval); + if (ts == 0.0) continue; + timestamps.assign(1, ts); + write_chunk(); } auto next_pull = Clock::now() + chunk_interval; @@ -523,23 +798,20 @@ void recording::typed_transfer_loop(streamid_t streamid, double srate, const inl next_pull += chunk_interval; } - if (first_timestamp != 0.0) { - // one final non-blocking pull, so that samples already buffered in the inlet when the - // stop arrived end up in the file rather than being dropped - try { - in->pull_chunk_multiplexed(chunk, ×tamps, 0.0); - write_chunk(); - } catch (std::exception &e) { - // the inlet was closed under us during teardown; the footer matters more - std::cerr << "Could not drain stream " << streamid << " on stop: " << e.what() - << std::endl; - } + // one final non-blocking pull, so that samples already buffered in the inlet when the stop + // arrived end up in the file rather than being dropped + try { + in->pull_chunk_multiplexed(chunk, ×tamps, 0.0); + write_chunk(); + } catch (std::exception &e) { + // the inlet was closed under us during teardown; the footer matters more + log_err("Could not drain stream ", streamid, " on stop: ", e.what()); } } catch (std::exception &) { stop_offsets(offset_shutdown); - timed_join_or_detach(offset_thread); + timed_join_or_detach(offset_thread, teardown_grace); throw; } stop_offsets(offset_shutdown); - timed_join_or_detach(offset_thread); + timed_join_or_detach(offset_thread, teardown_grace); } diff --git a/src/recording.h b/src/recording.h index 9b63da2..5eee784 100644 --- a/src/recording.h +++ b/src/recording.h @@ -1,101 +1,12 @@ #ifndef RECORDING_H #define RECORDING_H -#include "xdfwriter.h" -#include -#include -#include -#include -#include -#include #include #include #include -#include -#include #include -#include -#include #include -// timings in the recording process (e.g., rate of boundary chunks and for cases where a stream -// hangs) approx. interval between boundary chunks -const auto boundary_interval = std::chrono::seconds(10); -// approx. interval between offset measurements -const auto offset_interval = std::chrono::seconds(5); -// approx. interval between resolves for outstanding streams on the watchlist, in seconds -const double resolve_interval = 5; -// timeout of a single resolve attempt, in seconds; the rest of resolve_interval is spent in an -// interruptible wait so that a shutdown request need not wait out a resolve -const double resolve_timeout = 1; -// approx. interval between pulling chunks from outlets -const auto chunk_interval = std::chrono::milliseconds(500); -// maximum waiting time for moving past the headers phase while recording -const auto max_headers_wait = std::chrono::seconds(10); -// maximum waiting time for moving into the footers phase while recording -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 waiting time for a single time correction query -const auto max_time_correction_wait = std::chrono::seconds(2); -// blocking network calls are issued in slices of this length (in seconds) so that a shutdown -// request is noticed promptly instead of after the full timeout -const double network_poll_interval = 0.2; -// time granted to the stream threads to drain their inlets and write their footers before the -// inlets are forcibly closed -const auto teardown_grace = std::chrono::milliseconds(300); -// maximum time that we wait to join a thread -const auto max_join_wait = std::chrono::seconds(2); - -// steady_clock (not high_resolution_clock, which is an alias for the wall clock in some standard -// libraries) so that waits are unaffected by clock adjustments -using Clock = std::chrono::steady_clock; - -using streamid_t = uint32_t; - -/// thrown by the interruptible helpers when the recording is being torn down -class shutdown_requested : public std::runtime_error { -public: - explicit shutdown_requested(const std::string &what) : std::runtime_error(what) {} -}; - -/** - * A thread paired with a future that becomes ready once the thread body has returned. - * - * std::thread::join() blocks indefinitely, so polling it cannot enforce a deadline: a single call - * against a hung thread never comes back. The future can be waited on with a timeout, and only - * once it is ready do we join (which then returns promptly). A std::packaged_task future is used - * rather than std::async because the latter blocks in its future destructor. - */ -struct worker { - std::thread thread; - std::future done; -}; -// pointer to a worker thread -using worker_p = std::unique_ptr; - -/// start a worker thread running fn -template worker_p spawn_worker(F &&fn) { - auto task = std::make_shared>(std::forward(fn)); - auto w = std::make_unique(); - w->done = task->get_future(); - // the task is kept alive by the lambda, so the worker may be detached safely - w->thread = std::thread([task] { (*task)(); }); - return w; -} - -// pointer to a stream inlet -using inlet_p = std::shared_ptr; -// pointer to a per-stream flag asking that stream's offset thread to finish. Shared rather than -// referenced so that an offset thread which had to be detached cannot outlive its flag. -using offset_flag_p = std::shared_ptr>; -// a list of clock offset estimates (time,value) -using offset_list = std::list>; -// a map from streamid to offset_list -using offset_lists = std::map; - - /** * A recording process using the lab streaming layer. * An instance of this class is created with a list of stream references to record from. @@ -120,142 +31,19 @@ class recording { bool collect_offsets = true); /** Destructor. - * Stops the recording and closes the file. + * Asks the recording threads to finish and waits a bounded amount of time for them. A thread + * that is still stuck after that is left running; the file is closed once it finishes. */ ~recording(); - /// Ask all recording threads to wrap up. Returns immediately; the threads are joined by the - /// destructor. + /// Ask all recording threads to wrap up. Returns immediately. void requestStop() noexcept; private: - // the file stream - XDFWriter file_; // the file output stream - // static information - bool offsets_enabled_; // whether to collect time offset information alongside with the stream - // contents - bool unsorted_; // whether this file may contain unsorted chunks (e.g., of late streams) - - // streamid allocation - std::atomic streamid_; // the highest streamid allocated so far - - // phase-of-recording state (headers, streaming data, or footers) - std::atomic shutdown_; // whether we are trying to shut down - std::condition_variable - shutdown_cv_; // signals shutdown so that every interruptible wait returns at once - std::mutex shutdown_mut_; // protects publication of shutdown_ and of the per-stream offset - // shutdown flags, which the shutdown_cv_ predicates read under it - 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 - // phase (i.e., are not yet ready for writing their footer) - std::condition_variable - ready_for_streaming_; // condition variable signaling that all streams have finished writing - // their headers and are now ready to write streaming content - std::condition_variable - ready_for_footers_; // condition variable signaling that all streams have finished their - // recording jobs and are now ready to write a footer - std::mutex phase_mut_; // a mutex to protect the phase state - - // inlets with potentially pending network I/O, to be aborted if their thread does not stop in - // time - std::vector active_inlets_; - std::mutex inlets_mut_; // a mutex to protect the active inlet list - - // data structure to collect the time offsets for every stream - offset_lists - offset_lists_; // the clock offset lists for each stream (to be written into the footer) - std::mutex offset_mut_; // a mutex to protect the offset lists - - // data for shutdown / final joining - std::list stream_threads_; // the spawned stream handling threads - worker_p boundary_thread_; // the spawned boundary-recording thread - - // for enabling online sync options - std::map sync_options_by_stream_; - - // === recording thread functions === - - /// record from results of a query (spawn a recording thread for every result produced by the - /// query) - /// @param query The query string - void record_from_query_results(const std::string &query); - - /// record from a given stream (identified by its streaminfo) - /// @param src the stream_info from which to record - /// @param phase_locked whether this is a stream that is locked to the phases (1. Headers, 2. - /// Streaming Content, 3. Footers) - /// Late-added streams (e.g. forgotten devices) are not phase-locked. - void record_from_streaminfo(const lsl::stream_info &src, bool phase_locked); - - - /// record boundary markers every few seconds - void record_boundaries(); - - // record ClockOffset chunks from a given stream - void record_offsets( - streamid_t streamid, inlet_p in, offset_flag_p offset_shutdown) noexcept; - - - // sample collection loop for a numeric stream - template - void typed_transfer_loop(streamid_t streamid, double srate, const inlet_p &in, - double &first_timestamp, double &last_timestamp, uint64_t &sample_count); - - // === interruptible waiting & bounded network calls === - - /// wait until deadline, returning true if the wait was cut short by a shutdown request - /// @param extra an optional additional flag (e.g. a per-stream offset shutdown) that also ends - /// the wait - bool wait_until_shutdown(Clock::time_point deadline, const std::atomic *extra = nullptr); - - /// wait for timeout, returning true if the wait was cut short by a shutdown request - bool wait_for_shutdown(Clock::duration timeout, const std::atomic *extra = nullptr) { - return wait_until_shutdown(Clock::now() + timeout, extra); - } - - /// publish a per-stream offset shutdown flag and wake the corresponding offset thread - void stop_offsets(const offset_flag_p &offset_shutdown) noexcept; - - /// subscribe to a stream, giving up after max_open_wait - /// @return whether the subscription completed (if not, it will take place later) - /// @throws shutdown_requested if the recording was stopped while subscribing - bool open_inlet(const inlet_p &in); - - /// retrieve the full stream info, including the extended description - /// @throws shutdown_requested if the recording was stopped while retrieving the metadata - lsl::stream_info fetch_info(const inlet_p &in); - - // === inlet bookkeeping === - - void register_inlet(const inlet_p &in); - void unregister_inlet(const inlet_p &in) noexcept; - /// close every registered inlet, aborting any blocking socket call in progress - void close_active_inlets() noexcept; - - // === phase registration & condition checks === - // writing is coordinated across threads in three phases to keep the file chunks sorted - - void enter_headers_phase(bool phase_locked); - - void leave_headers_phase(bool phase_locked); - - void enter_streaming_phase(bool phase_locked); - - void leave_streaming_phase(bool phase_locked); - - void enter_footers_phase(bool phase_locked); - - void leave_footers_phase(bool) { /* Nothing to do. Ignore warning. */ - } - - /// a condition that indicates that we are ready to write streaming content into the file - bool ready_for_streaming() const { return headers_to_finish_ <= 0; } - /// a condition that indicates that we are ready to write footers into the file - bool ready_for_footers() const { return streaming_to_finish_ <= 0 && headers_to_finish_ <= 0; } - - /// allocate a fresh stream id - streamid_t fresh_streamid() { return ++streamid_; } + struct impl; + /// Shared rather than unique: a recording thread that had to be left running keeps the state + /// it writes into -- the file, the mutexes, the offset lists -- alive until it is done. + std::shared_ptr impl_; }; #endif