diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 8b29136..b9b691a 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 @@ -80,6 +80,7 @@ jobs: -DCMAKE_BUILD_TYPE=${{ env.BUILD_TYPE }} -DCMAKE_INSTALL_PREFIX=${{ github.workspace }}/install -DLSL_FETCH_IF_MISSING=ON + -DLABRECORDER_BUILD_TESTING=ON ${{ matrix.config.cmake_extra }} # ----------------------------------------------------------------------- @@ -88,6 +89,9 @@ jobs: - name: Build run: cmake --build build --config ${{ env.BUILD_TYPE }} --parallel + - name: Test recording finalization + run: ctest --test-dir build -C ${{ env.BUILD_TYPE }} --output-on-failure + # ----------------------------------------------------------------------- # Install # ----------------------------------------------------------------------- @@ -97,17 +101,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/BUILD.md b/BUILD.md index c3973b9..03aa553 100644 --- a/BUILD.md +++ b/BUILD.md @@ -92,6 +92,29 @@ The command line install feature does not put build products in the sample place If any significant changes are made to the project (such as changing Qt or Visual Stuido version) it is recommended that you delete or rename the build folder and start over. Various partial cleaning processes do not work well. +## Recording tests + +Enable the deterministic shutdown tests when configuring, then build and run CTest: + +```sh +cmake -S . -B build -DLABRECORDER_BUILD_TESTING=ON +cmake --build build --config Release +ctest --test-dir build -C Release --output-on-failure +``` + +These tests compile the recording implementation and XDF writer against a controlled inlet. +They cover delayed clock measurements, stopping during an unavailable measurement, and a +worker that exceeds the join warning deadline. Completion is checked only after the file is +closed; stop requests remain nonblocking. A subprocess test verifies bounded CLI exit with a +worker that never returns, and GUI builds test event-loop responsiveness, stalled-state controls, +remote status, restart rejection, and deferred window close. The Qt test uses the offscreen platform. + +The real-stream integration suite additionally requires `pylsl` and `pyxdf`: + +```sh +python scripts/test_recording_teardown.py --bin /path/to/LabRecorderCLI +``` + ## Linux * Ubuntu (/Debian) @@ -122,4 +145,3 @@ If any significant changes are made to the project (such as changing Qt or Visua 1. You may need to specify additional cmake options. . Build everything and copy the files to the `install` folder: * `cmake --build . --target install` - diff --git a/CMakeLists.txt b/CMakeLists.txt index a7d904e..2919219 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -30,6 +30,7 @@ set(CMAKE_EXPORT_COMPILE_COMMANDS ON) # Build Options # ============================================================================= option(LABRECORDER_BUILD_GUI "Build the GUI application (requires Qt6)" ON) +option(LABRECORDER_BUILD_TESTING "Build deterministic recording tests" OFF) # ============================================================================= # LSL Discovery Options @@ -218,6 +219,49 @@ target_link_libraries(${PROJECT_NAME}CLI PRIVATE LSL::lsl ) +if(LABRECORDER_BUILD_TESTING) + enable_testing() + add_executable(test_recording_finalization + tests/recording_finalization.cpp + src/recording.cpp + ) + # Compile the real recording implementation against a controlled inlet. Do not link + # liblsl into this target: its timeout behavior is supplied by the test double. + target_include_directories(test_recording_finalization PRIVATE tests/fake_lsl src) + target_link_libraries(test_recording_finalization PRIVATE xdfwriter Threads::Threads) + foreach(scenario IN ITEMS delayed unavailable stalled failed abandoned catchup cutoff clocksync unknown_clock finish_now independent_progress nonadvancing) + add_test(NAME recording_finalization_${scenario} + COMMAND test_recording_finalization ${scenario}) + set_tests_properties(recording_finalization_${scenario} PROPERTIES TIMEOUT 20) + endforeach() + find_package(Python3 REQUIRED COMPONENTS Interpreter) + add_executable(test_recording_cli src/clirecorder.cpp src/recording.cpp) + target_include_directories(test_recording_cli PRIVATE tests/fake_lsl src) + target_link_libraries(test_recording_cli PRIVATE xdfwriter Threads::Threads) + add_test(NAME recording_cli_timeout + COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/tests/cli_finalization.py + $) + set_tests_properties(recording_cli_timeout PROPERTIES TIMEOUT 25) + if(LABRECORDER_BUILD_GUI) + add_executable(test_gui_finalization tests/gui_finalization.cpp tests/gui_recording_stub.cpp + src/mainwindow.cpp src/mainwindow.h src/mainwindow.ui src/tcpinterface.cpp src/tcpinterface.h) + target_include_directories(test_gui_finalization PRIVATE src) + target_link_libraries(test_gui_finalization PRIVATE Qt6::Widgets Qt6::Network LSL::lsl Threads::Threads) + if(WIN32) + add_custom_command(TARGET test_gui_finalization POST_BUILD + COMMAND ${CMAKE_COMMAND} -E copy_if_different + $ $) + endif() + add_test(NAME recording_gui_finalization COMMAND test_gui_finalization) + add_test(NAME recording_gui_finish_now COMMAND test_gui_finalization --finish-now) + add_test(NAME recording_gui_force_exit + COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/tests/gui_force_exit.py + $) + set_tests_properties(recording_gui_finalization recording_gui_force_exit recording_gui_finish_now PROPERTIES TIMEOUT 25 + ENVIRONMENT "QT_QPA_PLATFORM=offscreen") + endif() +endif() + # ============================================================================= # Copy config file to build directory for testing # ============================================================================= diff --git a/README.md b/README.md index 8664d72..4bd3ec2 100644 --- a/README.md +++ b/README.md @@ -63,9 +63,17 @@ The Block/Task field can be overwriten or selected among a list of items found i -Click "Start" to start a recording. If everything goes well, the status bar will now display the time since you started the recording, and more importantly, the current file size (the number before the kb) will grow slowly. This is a way to check whether you are still in fact recording data. The recording program cannot be closed while you are recording (as a safety measure). +Click "Start" to start a recording. If everything goes well, the status bar will now display the time since you started the recording, and more importantly, the current file size (the number before the kb) will grow slowly. This is a way to check whether you are still in fact recording data. Closing the window requests Stop and waits for finalization while keeping the interface responsive. -When you are done recording, click the "Stop" button. You can now close the program. See [the xdf repository](https://github.com/sccn/xdf) for tools and information on how to use the XDF files. +When you are done recording, click **Stop**. This fixes a cutoff in the recorder's clock. Inlets stay subscribed to catch up on delayed data timestamped at or before that cutoff. The interface distinguishes **Catching up**, **Waiting for pre-stop data**, and **Closing recording file**. A new recording cannot start until the file is flushed and closed. + +Catch-up continues while pre-stop timestamps advance, even if it takes longer than five seconds. Each stream gets a **two-second inactivity grace**, measured since its last advancing pre-stop timestamp (or the start of catch-up). Samples arriving out of order are accepted during that grace; receiving a post-stop sample does not close the inlet immediately. Silent marker streams, disconnected sources, and non-advancing timestamps therefore cannot hold collection open indefinitely. Data arriving after the grace expires may be missed: neither silence nor crossing the cutoff proves that an outlet's buffers are empty. + +Comparison uses an LSL clock-correction estimate, without modifying the timestamps written to XDF or correcting an already clock-synchronized inlet twice. Clock estimates and source timestamps can be imperfect. If no correction is available, arrivals are preserved during a finite grace, potentially including post-stop samples, and the footer records this limitation. Each stream footer's `collection_end` records the recorder-clock `stop_time`, the grace interval, and a reason: `cutoff_observed`, `inactivity_timeout`, `clock_unavailable`, `unreliable_timestamps`, `user_requested`, or `transfer_error`. `cutoff_observed` means a later timestamp was seen and the grace elapsed; it is not a guarantee against arbitrarily delayed or reordered samples. + +**Finish now…** lets you end catch-up early and finalize the samples already saved; its confirmation explains that additional data may be missed. If a worker or file operation makes no progress for five seconds, **Force quit…** becomes available. Progress on one stream cannot hide a stalled worker on another. Force quitting may leave an incomplete file or lose buffered samples; it never reports successful finalization. Completed chunks are flushed periodically and footers immediately to improve recovery. **Stopped — file finalized** means the workers finished and the file was flushed and closed; the status also notes streams that ended without observing the cutoff. + +`LabRecorderCLI` uses a five-second **inactivity** timeout after Enter, extended by progress rather than elapsed time alone. Use `--stop-timeout SECONDS` before the output filename to change it (greater than zero, at most 3600; values below the two-second grace can interrupt normal catch-up). Exit status **0** means the file finalized, subject to its footer's collection-end reasons; **3** means a worker exceeded the inactivity timeout and the file may be incomplete; **4** means recording or finalization failed. A continuously advancing backlog has no fixed overall deadline. See [the xdf repository](https://github.com/sccn/xdf) for tools and information on how to use the XDF files. ## Preparing a Full Study @@ -89,9 +97,12 @@ Currently supported commands include: * `select none` * `start` * `stop` +* `status` * `update` * `filename ...` +`stop` acknowledges the request with `OK`; this is not a completion notification. Poll `status` for a newline-terminated state: `recording`, `finishing` (initial stop request), `catching_up`, `waiting`, `closing`, `stalled`, `stopped`, or `error`. Wait for `stopped` before restarting or using the file. `start` is rejected with `ERROR ` while a recording is active or finalizing. + `filename` is followed by a series of space-delimited options enclosed in curly braces. e.g. {root:C:\root_data_dir} * `root` - Sets the root data directory. * `template` - sets the File Name / Template. Will unselect BIDS option. May contain wildcards. diff --git a/scripts/test_recording_teardown.py b/scripts/test_recording_teardown.py new file mode 100644 index 0000000..f59b37f --- /dev/null +++ b/scripts/test_recording_teardown.py @@ -0,0 +1,504 @@ +#!/usr/bin/env python +"""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 (4.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 + +# time given to LSL to make a new outlet discoverable, and to flush the last samples over TCP +SETTLE = 0.5 + + +class TestFailure(AssertionError): + """Raised when a case does not hold up.""" + + +def check(condition, message): + if not condition: + raise TestFailure(message) + + +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, 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='{name}'" for name in stream_order], + 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() + + def _read_output(self): + for line in self._proc.stdout: + self.lines.append(line.rstrip()) + + 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 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() + 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, stream_order=NAMES): + """Run LabRecorderCLI over both test streams, making sure it is gone afterwards.""" + rec = Recorder(cli_path, xdf_path, stream_order) + try: + yield rec + finally: + rec.terminate() + for line in rec.lines: + print(f" | {line}") + + +# 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") + + +def load_xdf_strict(xdf_path): + """Load an XDF file and fail if pyxdf reports it as damaged.""" + records = [] + + class Collector(logging.Handler): + def emit(self, record): + records.append(record) + + handler = Collector(level=logging.WARNING) + logger = logging.getLogger("pyxdf") + logger.addHandler(handler) + try: + streams, header = pyxdf.load_xdf(xdf_path) + finally: + logger.removeHandler(handler) + + 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 + + +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") + + +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") + + 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]) + + +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", + ) + + +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 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() + 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 + + +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 + + +def case_clock_offsets_collected(cli_path, xdf_path, max_stop): + """Run past the first offset query and verify short waits still obtain its result.""" + eeg, markers = make_outlets() + time.sleep(SETTLE) + with recorder(cli_path, xdf_path) as rec: + rec.wait_until_collecting() + # Offset queries begin after five seconds. Allow the local LSL probe exchange to + # complete across several 200 ms waits before stopping. + push_eeg_for(eeg, 6.5) + duration = rec.stop() + + check_stop(duration, max_stop) + streams, _ = load_xdf_strict(xdf_path) + for name in NAMES: + stream = stream_by_name(streams, name) + check_footer(stream) + offsets = stream["footer"]["info"]["clock_offsets"][0] + check(offsets and offsets.get("offset"), f"stream {name!r} has no clock offsets") + del eeg, markers + + +def case_delayed_upstream(cli_path, xdf_path, max_stop): + """Keep receiving old data for longer than the CLI's five-second stop timeout.""" + eeg, markers = make_outlets() + time.sleep(SETTLE) + with recorder(cli_path, xdf_path, stream_order=(EEG_NAME,)) as rec: + rec.wait_for([f"Started data collection for stream {EEG_NAME}"]) + # Establish clock mapping while simulating an inlet that is already behind. + # Later backlog timestamps must advance beyond any locally buffered samples. + warmup_until = time.monotonic() + 6.5 + while time.monotonic() < warmup_until: + eeg.push_sample([0.0] * EEG_CHANNELS, timestamp=pylsl.local_clock() - 30) + time.sleep(1 / EEG_RATE) + old = pylsl.local_clock() - 20 + # Include a deducible interval followed by a gap: timestamp compression + # must not move the gap's sample backwards by one nominal interval. + second = old + 1 / EEG_RATE + expected = [old, second, second + 1 / EEG_RATE + 1 / EEG_RATE] + expected += [old + i for i in range(3, 7)] + def backlog(): + for i, timestamp in enumerate(expected): + time.sleep(0.8) + eeg.push_sample([10000.0 + i] * EEG_CHANNELS, timestamp=timestamp) + # A future timestamp followed by another pre-stop timestamp must not + # cause an immediate close that drops the reordered sample. + eeg.push_sample([99999.0] * EEG_CHANNELS, timestamp=pylsl.local_clock() + 100) + time.sleep(0.3) + eeg.push_sample([10007.0] * EEG_CHANNELS, timestamp=old + 7) + publisher = threading.Thread(target=backlog) + publisher.start() + try: + duration = rec.stop() + finally: + publisher.join() + check(5 < duration < 10, f"catch-up did not extend the stop deadline: {duration}") + streams, _ = load_xdf_strict(xdf_path) + stream = stream_by_name(streams, EEG_NAME) + check_footer(stream) + values = stream["time_series"][:, 0].tolist() + check(all(10000.0 + i in values for i in range(8)), "late pre-stop samples were lost") + check(99999.0 not in values, "post-stop sample was retained") + raw, _ = pyxdf.load_xdf(xdf_path, synchronize_clocks=False, dejitter_timestamps=False) + raw_stream = stream_by_name(raw, EEG_NAME) + for i, timestamp in enumerate(expected + [old + 7]): + index = raw_stream["time_series"][:, 0].tolist().index(10000.0 + i) + check(raw_stream["time_stamps"][index] == timestamp, "stored timestamp was rewritten") + reason = stream["footer"]["info"]["collection_end"][0]["reason"][0] + check(reason == "cutoff_observed", f"wrong collection end reason: {reason}") + 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), + ("gated stream is drained", case_gated_stream_is_drained), + ("clock offsets collected", case_clock_offsets_collected), + ("delayed upstream catch-up", case_delayed_upstream), +] + + +def main(): + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--bin", required=True, help="path to the LabRecorderCLI binary") + parser.add_argument( + "--max-stop", + type=float, + default=4.0, + help="upper bound in seconds for how long teardown may take (default: %(default)s)", + ) + args = parser.parse_args() + + 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/clirecorder.cpp b/src/clirecorder.cpp index 59e3736..838ed13 100644 --- a/src/clirecorder.cpp +++ b/src/clirecorder.cpp @@ -1,20 +1,41 @@ #include "recording.h" #include "xdfwriter.h" +#include +#include +#include +#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" - << "searchstr can be anything accepted by lsl_resolve_bypred\n"; - std::cout << "Keep in mind that your shell might remove quotes\n"; - std::cout << "Examples:\n\t" << argv[0] << " foo.xdf 'type=\"EEG\"' "; - std::cout << " 'host=\"LabPC1\" or host=\"LabPC2\"'\n\t"; - std::cout << argv[0] << " foo.xdf'name=\"Tobii and type=\"Eyetracker\"'\n"; + int output_arg = 1; + double stop_timeout = 5.0; + if (argc > 1 && std::string(argv[1]) == "--stop-timeout") { + try { + if (argc < 3) throw std::invalid_argument("missing value"); + size_t used = 0; + stop_timeout = std::stod(argv[2], &used); + if (used != std::string(argv[2]).size() || !std::isfinite(stop_timeout) || + stop_timeout <= 0 || stop_timeout > 3600) + throw std::invalid_argument("must be greater than zero and at most 3600 seconds"); + output_arg = 3; + } catch (const std::exception &e) { + std::cerr << "Invalid --stop-timeout: " << e.what() << std::endl; + return 1; + } + } + if (argc < output_arg + 2 || std::string(argv[output_arg]) == "--help" || + std::string(argv[output_arg]) == "-h") { + std::cout + << "Usage: " << argv[0] + << " [--stop-timeout SECONDS] outputfile.xdf 'searchstr' ['searchstr2' ...]\n" + << "Search strings use lsl_resolve_bypred syntax.\n" + << "Stop inactivity timeout defaults to 5 seconds; an unfinished file exits with status 3.\n"; return 1; } std::vector infos = lsl::resolve_streams(), recordstreams; - for (int i = 2; i < argc; ++i) { + for (int i = output_arg + 1; i < argc; ++i) { bool matched = false; for (const auto &info : infos) { if (info.matches_query(argv[i])) { @@ -33,7 +54,33 @@ int main(int argc, char **argv) { std::vector watchfor; std::map sync_options; std::cout << "Starting the recording, press Enter to quit" << std::endl; - recording r(argv[1], recordstreams, watchfor, sync_options, true); - std::cin.get(); - return 0; + try { + recording r(argv[output_arg], recordstreams, watchfor, sync_options, true); + std::cin.get(); + r.requestStop(); + const auto timeout = std::chrono::duration_cast( + std::chrono::duration(stop_timeout)); + while (!r.waitForFinished(std::chrono::milliseconds(50))) { + if (r.finalizationProgress().idle < timeout) continue; + // Even reporting the timeout must not hang if a stalled worker holds an iostream + // lock or stderr is backed by a blocked pipe. Allow a brief best-effort diagnostic. + std::thread([] { + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + std::_Exit(3); + }).detach(); + std::cerr << "Finalization timed out. The recording may be incomplete: " + << argv[output_arg] << std::endl; + // No destructors/atexit handlers: a stalled worker could hold their locks too. + std::_Exit(3); + } + const auto error = r.finalizationError(); + if (!error.empty()) { + std::cerr << "Finalization failed: " << error << std::endl; + return 4; + } + return 0; + } catch (const std::exception &e) { + std::cerr << "Recording failed: " << e.what() << std::endl; + return 4; + } } diff --git a/src/mainwindow.cpp b/src/mainwindow.cpp index be82724..3ba5c96 100644 --- a/src/mainwindow.cpp +++ b/src/mainwindow.cpp @@ -15,6 +15,7 @@ using QRegExp = QRegularExpression; #endif #include +#include #include // recording class @@ -88,29 +89,91 @@ MainWindow::MainWindow(QWidget *parent, const char *config_file) timer = std::make_unique(this); connect(&*timer, &QTimer::timeout, this, &MainWindow::statusUpdate); - timer->start(1000); + timer->start(100); QString cfgfilepath = find_config_file(config_file); load_config(cfgfilepath); } -void MainWindow::statusUpdate() const { - if (currentRecording) { - auto elapsed = static_cast(lsl::local_clock() - startTime); - QString recFilename = replaceFilename(QDir::cleanPath(ui->lineEdit_template->text())); - auto fileinfo = QFileInfo(QDir::cleanPath(ui->rootEdit->text()) + '/' + recFilename); - fileinfo.refresh(); - auto size = fileinfo.size(); - QString timeString = QStringLiteral("Recording to %1 (%2; %3kb)") - .arg(QDir::toNativeSeparators(recFilename), - QTime(0,0).addSecs(elapsed).toString("hh:mm:ss"), - QString::number(size / 1000)); - statusBar()->showMessage(timeString); +void MainWindow::setRemoteState(const QString &state) { + if (rcs) rcs->recordingState = state; +} + +void MainWindow::statusUpdate() { + if (!currentRecording) return; + if (finishing) { + if (currentRecording->isFinished()) { + const auto error = currentRecording->finalizationError(); + const auto fallback = currentRecording->finalizationProgress().fallback_streams; + currentRecording.reset(); + finishing = false; + ui->startButton->setEnabled(true); + ui->stopButton->setEnabled(false); + ui->stopButton->setText("Stop"); + setRemoteState(error.empty() ? "stopped" : "error"); + statusBar()->showMessage(error.empty() ? (fallback + ? QStringLiteral("Stopped — file finalized; %1 stream(s) ended without confirming the cutoff").arg(fallback) + : QStringLiteral("Stopped — file finalized")) + : QStringLiteral("Finalization failed — recording may be incomplete")); + if (!error.empty()) { + closeWhenFinished = false; + QMessageBox::critical(this, "Recording incomplete", + QString::fromStdString(error) + "\n" + recordingPath); + } + if (closeWhenFinished) close(); + } else { + const auto progress = currentRecording->finalizationProgress(); + const bool stalled = progress.idle >= std::chrono::seconds(5); + setRemoteState(stalled ? "stalled" : progress.catching_up ? "catching_up" + : progress.collecting ? "waiting" : "closing"); + statusBar()->showMessage(stalled + ? QStringLiteral("Waiting without progress — file is not finalized. Wait, or choose Force quit.") + : progress.catching_up + ? QStringLiteral("Catching up to stop time — %1 stream(s) still collecting").arg(progress.collecting) + : progress.collecting + ? QStringLiteral("Waiting for pre-stop data — %1 stream(s) still open").arg(progress.collecting) + : QStringLiteral("Closing recording file…")); + ui->stopButton->setText(stalled ? QStringLiteral("Force quit…") + : progress.collecting ? QStringLiteral("Finish now…") : QStringLiteral("Stop")); + ui->stopButton->setEnabled(stalled || progress.collecting); + } + return; + } + const auto elapsed = static_cast(lsl::local_clock() - startTime); + const QFileInfo fileinfo(recordingPath); + statusBar()->showMessage(QStringLiteral("Recording to %1 (%2; %3kb)") + .arg(QDir::toNativeSeparators(recordingPath), + QTime(0, 0).addSecs(elapsed).toString("hh:mm:ss"), QString::number(fileinfo.size() / 1000))); +} + +void MainWindow::confirmForceQuit() { + QMessageBox dialog(QMessageBox::Warning, "Recording is still finishing", + "The file has not been finalized. Force quitting may lose buffered samples or leave " + "an incomplete recording. Finish now stops collecting additional data and then " + "finalizes the file with the samples already saved.\n" + recordingPath, QMessageBox::NoButton, this); + auto *wait = dialog.addButton("Keep waiting", QMessageBox::RejectRole); + auto *finish = currentRecording && currentRecording->finalizationProgress().collecting + ? dialog.addButton("Finish now (may miss data)", QMessageBox::ActionRole) : nullptr; + auto *quit = currentRecording && currentRecording->finalizationProgress().idle >= std::chrono::seconds(5) + ? dialog.addButton("Force quit", QMessageBox::DestructiveRole) : nullptr; + dialog.setDefaultButton(wait); + dialog.setEscapeButton(wait); + dialog.exec(); + if (finish && dialog.clickedButton() == finish && currentRecording) { + currentRecording->finishCollecting(); + statusUpdate(); } + if (quit && dialog.clickedButton() == quit && currentRecording && !currentRecording->isFinished()) + std::_Exit(3); } void MainWindow::closeEvent(QCloseEvent *ev) { - if (currentRecording) ev->ignore(); + if (!currentRecording) { ev->accept(); return; } + ev->ignore(); + closeWhenFinished = true; + if (!finishing) stopRecording(); + else if (currentRecording->finalizationProgress().idle >= std::chrono::seconds(5) || + currentRecording->finalizationProgress().collecting) confirmForceQuit(); } void MainWindow::blockSelected(const QString &block) { @@ -466,8 +529,18 @@ void MainWindow::startRecording() { } qInfo() << "Missing: " << missingStreams; - currentRecording = std::make_unique(recFilename.toStdString(), - requestedAndAvailableStreams, watchfor, syncOptionsByStreamName, true); + try { + currentRecording = std::make_unique(recFilename.toStdString(), + requestedAndAvailableStreams, watchfor, syncOptionsByStreamName, true); + } catch (const std::exception &e) { + setRemoteState("error"); + QMessageBox::critical(this, "Cannot start recording", QString::fromStdString(e.what())); + return; + } + recordingPath = recFilename; + finishing = false; + closeWhenFinished = false; + setRemoteState("recording"); ui->stopButton->setEnabled(true); ui->startButton->setEnabled(false); startTime = (int)lsl::local_clock(); @@ -479,18 +552,18 @@ void MainWindow::startRecording() { } void MainWindow::stopRecording() { - - if (currentRecording) { - try { - currentRecording = nullptr; - } catch (std::exception &e) { qWarning() << "exception on stop: " << e.what(); } - ui->startButton->setEnabled(true); - ui->stopButton->setEnabled(false); - statusBar()->showMessage("Stopped"); - } else if (!hideWarnings) { - QMessageBox::information( - this, "Not recording", "There is not ongoing recording", QMessageBox::Ok); + if (!currentRecording) return; + if (finishing) { + if (currentRecording->finalizationProgress().idle >= std::chrono::seconds(5) || + currentRecording->finalizationProgress().collecting) confirmForceQuit(); + return; } + finishing = true; + currentRecording->requestStop(); + ui->startButton->setEnabled(false); + ui->stopButton->setEnabled(false); + setRemoteState("finishing"); + statusBar()->showMessage(QStringLiteral("Finishing recording…")); } void MainWindow::selectAllStreams() { @@ -640,6 +713,8 @@ void MainWindow::enableRcs(bool bEnable) { } else if (bEnable) { uint16_t port = ui->rcsport->value(); rcs = std::make_unique(port); + setRemoteState(!currentRecording ? "stopped" : !finishing ? "recording" + : currentRecording->finalizationProgress().idle >= std::chrono::seconds(5) ? "stalled" : "finishing"); // TODO: Add some method to RemoteControlSocket to report if its server is listening (i.e. was successful). connect(rcs.get(), &RemoteControlSocket::refresh_streams, this, &MainWindow::refreshStreams); connect(rcs.get(), &RemoteControlSocket::start, this, &MainWindow::rcsStartRecording); @@ -669,6 +744,7 @@ void MainWindow::rcsStartRecording() { } void MainWindow::rcsStopRecording() { + if (finishing) return; // Remote retries must never open a force-quit dialog. hideWarnings = true; stopRecording(); } diff --git a/src/mainwindow.h b/src/mainwindow.h index abeaad4..043ac65 100644 --- a/src/mainwindow.h +++ b/src/mainwindow.h @@ -1,6 +1,7 @@ #ifndef MAINWINDOW_H #define MAINWINDOW_H #include +#include #include #include #include @@ -42,7 +43,7 @@ class MainWindow : public QMainWindow { ~MainWindow() noexcept override; private slots: - void statusUpdate(void) const; + void statusUpdate(void); void closeEvent(QCloseEvent *ev) override; void blockSelected(const QString &block); std::vector refreshStreams(void); @@ -69,6 +70,11 @@ private slots: void save_config(QString filename); std::unique_ptr currentRecording; + bool finishing = false; + bool closeWhenFinished = false; + QString recordingPath; + void confirmForceQuit(); + void setRemoteState(const QString &state); std::unique_ptr rcs; int startTime; diff --git a/src/recording.cpp b/src/recording.cpp index 0b8b43e..75e3ca0 100644 --- a/src/recording.cpp +++ b/src/recording.cpp @@ -1,194 +1,628 @@ #include "recording.h" //#include "conversions.h" +#include "xdfwriter.h" +#include +#include +#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; +// initial grace for workers to drain their inlets and write their footers +const auto teardown_grace = std::chrono::milliseconds(300); +// time before reporting a slow worker; finalization still waits for it to finish +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 until its body returns + w->thread = std::thread([task] { (*task)(); }); + return w; +} + +// pointer to a stream inlet +using inlet_p = std::shared_ptr; +// Per-stream stop flag shared with the offset worker until it has been joined. +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 { + +std::mutex log_mut; + +/// 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'; + 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( + 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; +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 join_worker Join the worker, reporting when it exceeds the expected duration + * @param w unique_ptr to a worker. Will be reset either way + * @param duration time before reporting that finalization is still waiting + */ +inline void join_worker(worker_p &w, std::chrono::milliseconds duration = max_join_wait) { + if (!timed_join(w, duration)) { + log_err("Waiting for a recording worker to finish before closing the file."); + w->thread.join(); + w.reset(); } - return false; } /** - * @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_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 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(500)); +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; } - 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 join_workers Join all workers before their writer can be destroyed + * @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( - 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 join_workers( + std::list &workers, std::chrono::milliseconds duration = max_join_wait) { + timed_join_some(workers, duration); + if (!workers.empty()) { + log_out(workers.size(), " stream threads still running!"); + for (auto &w : workers) w->thread.join(); + workers.clear(); } } +// Shared control and progress only. No file or inlet operation runs under this mutex. +struct recording::completion { + std::mutex mutex; + std::condition_variable changed; + bool startup_finished = false; + bool stop_requested = false; + std::atomic cutoff{0}; + std::atomic finish_collecting{false}; + std::atomic fallback_streams{0}; + Clock::time_point stopped{}, changed_at{}; + struct stream_progress { + bool collecting = true, done = false, advancing = false; + Clock::time_point advanced = Clock::now(); + }; + std::map streams; + std::promise result; + + void note(streamid_t id, bool collecting, bool done = false, bool advancing = false) { + std::lock_guard lock(mutex); + streams[id] = {collecting, done, advancing, Clock::now()}; + changed_at = Clock::now(); + } +}; + /** - * @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 + * Workers and the background finalizer retain this state. The UI handle owns only + * control/progress and completion. The finalizer joins all workers and closes the + * file before publishing completion; callers can enforce their own inactivity limit. */ -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(500)); +struct recording::impl : std::enable_shared_from_this { + impl(const std::string &filename, std::map syncOptions, + bool collect_offsets, std::shared_ptr control) + : control_(std::move(control)), 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)) {} + + /// stop_and_join() leaves the worker containers empty before this state is destroyed. + ~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 join them all before closing the file. + /// Called only from the background finalizer, never from the UI or a recording worker. + std::string stop_and_join(); + + void requestStop() noexcept; + + std::shared_ptr control_; + // 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 recording_failed_{false}; + 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 + + // 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, bool clocksync, + std::string &end_reason); + + // === 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); } - if (!threads.empty()) { - std::cout << threads.size() << " stream threads still running!" << std::endl; - for (auto &t : threads) t->detach(); - threads.clear(); + + /// 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); + + // === 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. */ } -} -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)) { + /// 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) { + // Each worker owns its state until it has finished; stop_and_join() joins them all. + auto self = shared_from_this(); // create a recording thread for each stream for (const auto &stream : streams) stream_threads_.emplace_back( - new std::thread(&recording::record_from_streaminfo, this, 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( - new std::thread(&recording::record_from_query_results, this, query)); + spawn_worker([self, query] { self->record_from_query_results(query); })); // create a boundary chunk writer thread - boundary_thread_ = std::make_unique(&recording::record_boundaries, this); + boundary_thread_ = spawn_worker([self] { self->record_boundaries(); }); +} + +std::string recording::impl::stop_and_join() { + requestStop(); + // This runs exclusively on the background finalizer. A slow worker can delay + // completion, but it cannot block the UI, its timeout, or its force-quit action. + timed_join_some(stream_threads_, teardown_grace); + if (!stream_threads_.empty()) { + // Keep draining rather than closing every inlet and discarding buffered samples. + join_workers(stream_threads_, max_join_wait); + } + join_worker(boundary_thread_, max_join_wait); + file_.close(); + return recording_failed_ ? "A recording worker failed; the file may be incomplete. See the log." + : ""; } -recording::~recording() { +recording::recording(const std::string &filename, const std::vector &streams, + const std::vector &watchfor, std::map syncOptions, + bool collect_offsets) + : completion_(std::make_shared()), result_(completion_->result.get_future().share()) { + auto state = std::make_shared(filename, std::move(syncOptions), collect_offsets, completion_); + // Start the finalizer before workers so partial startup failures can also be cleaned + // up off the caller's thread. The control mutex protects only short state updates. + std::thread([state, done = completion_]() mutable { + { + std::unique_lock lock(done->mutex); + done->changed.wait(lock, [&] { return done->startup_finished && done->stop_requested; }); + } + std::string error; + try { + error = state->stop_and_join(); + } catch (const std::exception &e) { + error = e.what(); + } catch (...) { + error = "Unknown error finalizing the recording."; + } + state.reset(); + // Includes writer close/flush and inlet destruction, not merely thread completion. + done->result.set_value(std::move(error)); + }).detach(); try { - // set the shutdown flag (from now on no more new streams) - shutdown_ = true; + state->start(streams, watchfor); + } catch (...) { + { + std::lock_guard lock(completion_->mutex); + completion_->startup_finished = true; + completion_->cutoff = lsl::local_clock(); + completion_->stopped = completion_->changed_at = Clock::now(); + completion_->stop_requested = true; + } + completion_->changed.notify_one(); + throw; + } + { + std::lock_guard lock(completion_->mutex); + completion_->startup_finished = true; + } + completion_->changed.notify_one(); +} - // stop the threads - timed_join_or_detach(stream_threads_, max_join_wait); - if (!timed_join(boundary_thread_, max_join_wait + boundary_interval)) { - std::cout << "boundary_thread didn't finish in time!" << std::endl; - boundary_thread_->detach(); +recording::~recording() { requestStop(); } + +void recording::requestStop() noexcept { + { + std::lock_guard lock(completion_->mutex); + if (!completion_->stop_requested) { + completion_->cutoff = lsl::local_clock(); + completion_->stopped = completion_->changed_at = Clock::now(); + completion_->stop_requested = true; } - std::cout << "Closing the file." << std::endl; - } catch (std::exception &e) { - std::cout << "Error while closing the recording: " << e.what() << std::endl; } + completion_->changed.notify_one(); +} + +void recording::finishCollecting() noexcept { + requestStop(); + completion_->finish_collecting = true; +} + +recording::FinalizationProgress recording::finalizationProgress() const { + FinalizationProgress result; + result.fallback_streams = completion_->fallback_streams; + std::lock_guard lock(completion_->mutex); + if (!completion_->stop_requested) return result; + const auto now = Clock::now(); + auto oldest = std::max(completion_->stopped, completion_->changed_at); + for (const auto &entry : completion_->streams) { + const auto &stream = entry.second; + if (stream.done) continue; + const auto advanced = std::max(completion_->stopped, stream.advanced); + oldest = std::min(oldest, advanced); + if (stream.collecting) { + ++result.collecting; + if (stream.advancing && now - advanced < std::chrono::milliseconds(500)) ++result.catching_up; + } + } + result.idle = std::chrono::duration_cast(now - oldest); + return result; +} + +bool recording::waitForFinished(std::chrono::milliseconds timeout) const { + return result_.wait_for(timeout) == std::future_status::ready; +} + +std::string recording::finalizationError() const { + if (!isFinished()) throw std::logic_error("Recording is still finalizing"); + return result_.get(); +} + +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 + // 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(); } -void recording::requestStop() noexcept -{ - shutdown_ = true; +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::record_from_query_results(const std::string &query) { +void recording::impl::stop_offsets(const offset_flag_p &offset_shutdown) noexcept { + { + std::lock_guard lock(shutdown_mut_); + *offset_shutdown = true; + } + shutdown_cv_.notify_all(); +} + +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); + while (Clock::now() < deadline && !shutdown_) { + try { + in->open_stream(network_poll_interval); + return true; + } catch (lsl::timeout_error &) {} + } + return false; +} + +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 + // 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::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; + std::list threads; // our spawned threads + log_out("Watching for a stream with properties ", query); 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()))) { - std::cout << "Found a new stream named " << result.name() - << ", adding it to the recording." << std::endl; + (!known_source_ids.count(result.source_id()))) { + log_out("Found a new stream named ", result.name(), ", adding it to the recording."); // start a new recording thread - threads.emplace_back(new std::thread( - &recording::record_from_streaminfo, this, 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()) 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); + join_workers(threads, max_join_wait); } catch (std::exception &e) { - std::cout << "Error in the record_from_query_results thread: " << e.what() << std::endl; + recording_failed_ = true; + 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) { + // obtain a fresh streamid + streamid_t streamid = fresh_streamid(); + control_->note(streamid, true); + // Publish completion even on an exception; no writer or inlet lock is held here. + auto progress_guard = std::shared_ptr(nullptr, [this, streamid](void *) { + control_->note(streamid, false, true); + }); + 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; - // obtain a fresh streamid - streamid_t streamid = fresh_streamid(); - - inlet_p in; + double nominal_srate = 0; + bool clocksync = false; + std::string end_reason = "transfer_error"; // --- 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)); + in = std::make_shared(src); 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); - std::cout << "Opened the stream " << src.name() << "." << std::endl; - } catch (lsl::timeout_error &) { - std::cout - << "Subscribing to the stream " << src.name() - << " is taking relatively long; collection from this stream will be delayed." - << std::endl; + if (it != sync_options_by_stream_.end()) { + in->set_postprocessing(it->second); + clocksync = (it->second & lsl::post_clocksync) != 0; } - // retrieve the stream header & get its XML version - file_.write_stream_header(streamid, in->info().as_xml()); - std::cout << "Received header for stream " << src.name() << "." << std::endl; + if (open_inlet(in)) + log_out("Opened the stream ", src.name(), "."); + else if (!shutdown_) + 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()); + log_out("Received header for stream ", src.name(), "."); leave_headers_phase(phase_locked); } catch (std::exception &) { @@ -205,35 +639,39 @@ 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; - - const double nominal_srate = in->info().nominal_srate(); + log_out("Started data collection for stream ", src.name(), "."); // 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, + clocksync, end_reason); 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, + clocksync, end_reason); 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, + clocksync, end_reason); 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, + clocksync, end_reason); 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, + clocksync, end_reason); 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, + clocksync, end_reason); break; default: // unsupported channel format @@ -242,9 +680,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 + recording_failed_ = true; + log_err("Error while recording from ", src.name(), ": ", e.what()); } // --- footers phase @@ -258,6 +699,10 @@ void recording::record_from_streaminfo(const lsl::stream_info &src, bool phase_l footer << "" << first_timestamp << "" << last_timestamp << "" << sample_count << ""; + footer << "" << end_reason + << "" << control_->cutoff.load() + << "recorder" + << "2"; footer << ""; { // including the clock_offset list @@ -270,66 +715,84 @@ 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) { + 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; + recording_failed_ = true; + log_out("Error in the record_from_streaminfo thread: ", e.what()); } } -void recording::record_boundaries() { +void recording::impl::record_boundaries() { try { auto next_boundary = Clock::now() + boundary_interval; while (!shutdown_) { - std::this_thread::sleep_for(std::chrono::milliseconds(500)); - if (Clock::now() > next_boundary) { + if (wait_for_shutdown(std::chrono::seconds(1))) break; + if (Clock::now() >= next_boundary) { file_.write_boundary_chunk(); next_boundary = Clock::now() + boundary_interval; } + file_.checkpoint(); } } catch (std::exception &e) { - std::cout << "Error in the record_boundaries thread: " << e.what() << std::endl; + recording_failed_ = true; + log_out("Error in the record_boundaries thread: ", e.what()); } } -void recording::record_offsets( - streamid_t streamid, const inlet_p &in, std::atomic &offset_shutdown) noexcept { +void recording::impl::record_offsets( + 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::this_thread::sleep_for(offset_interval); - // query the time offset - double offset, now; - try { - offset = in->time_correction(2); - now = lsl::local_clock(); - } catch (lsl::timeout_error &) { - std::cerr << "Timeout in time correction query for stream " << streamid - << std::endl; + if (wait_for_shutdown(offset_interval, offset_shutdown.get())) break; + + // liblsl's background measurement survives a time_correction() timeout. Polling + // waits for that same result; it does not restart the packet exchange. + const auto deadline = Clock::now() + seconds_to_duration(max_time_correction_wait); + double offset = 0, now = 0; + bool have_offset = false; + while (!shutdown_ && !*offset_shutdown && Clock::now() < deadline) { + const double remaining = std::chrono::duration(deadline - Clock::now()).count(); + try { + offset = in->time_correction(std::max(0.0, std::min(network_poll_interval, remaining))); + now = lsl::local_clock(); + have_offset = true; + break; + } catch (lsl::timeout_error &) {} } + if (shutdown_ || *offset_shutdown) break; + if (!have_offset) { + log_err("Timeout in time correction query for stream ", streamid); + continue; + } + file_.write_stream_offset(streamid, now, offset); // also append to the offset lists std::lock_guard lock(offset_mut_); offset_lists_[streamid].emplace_back(now - offset, offset); } } catch (std::exception &e) { - std::cout << "Error in the record_offsets thread: " << e.what() << std::endl; + recording_failed_ = true; + 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_--; @@ -338,16 +801,19 @@ 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_); - 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_++; } } -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_--; @@ -356,22 +822,27 @@ 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_); - 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(); }); } } 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) { +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, bool clocksync, + std::string &end_reason) { // 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); + auto self = shared_from_this(); + worker_p offset_thread(offsets_enabled_ + ? spawn_worker([self, streamid, in, offset_shutdown] { + self->record_offsets(streamid, in, offset_shutdown); + }) + : nullptr); try { double sample_interval = srate ? 1.0 / srate : 0; @@ -379,41 +850,109 @@ 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, 4.0); - if (!shutdown_) { - 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 (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; + last_timestamp = ts; ts = 0; } else last_timestamp = ts; } - // write the actual chunk 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); + }; + + // Bounded batches prevent a continuously producing outlet from trapping a pull + // inside liblsl's vector pull_chunk helper and delaying stop/cutoff checks. + std::vector sample; + bool stopping = false, have_correction = clocksync, saw_cutoff = false; + bool unmapped_samples = false, invalid_timestamps = false; + double correction = 0, high_water = -std::numeric_limits::infinity(); + auto last_advance = Clock::now(); + const auto inactivity_grace = std::chrono::seconds(2); + auto observe_stop = [&] { + if (control_->cutoff.load() && !stopping) { + stopping = true; + last_advance = Clock::now(); + } + if (stopping && !have_correction) { + try { + correction = in->time_correction(0.0); + have_correction = std::isfinite(correction); + } catch (const std::exception &) { + // Losing clock service must not discard samples still available from + // the data connection. Preserve them during the fallback grace. + } + } + }; + for (;;) { + // Mapping is needed even if offset chunks are disabled. A zero-timeout + // query starts/reuses liblsl's measurement without waiting. + observe_stop(); + if (stopping && control_->finish_collecting) { + end_reason = "user_requested"; + break; + } + chunk.clear(); + timestamps.clear(); + bool advanced = false; + for (size_t n = 0; n < 1024; ++n) { + const double ts = in->pull_sample(sample, n == 0 ? network_poll_interval : 0.0); + // Stop can arrive while pull_sample is waiting. Compare this sample + // too, without losing data pulled just as the request arrived. + if (!stopping && control_->cutoff.load()) observe_stop(); + if (!ts) break; + const double stop = control_->cutoff.load(); + const double local_ts = ts + correction; + if (stopping && !have_correction) unmapped_samples = true; + if (stopping && !std::isfinite(local_ts)) invalid_timestamps = true; + if (stopping && have_correction && std::isfinite(local_ts)) { + if (local_ts > stop) { + saw_cutoff = true; + continue; + } + if (local_ts > high_water) { + high_water = local_ts; + advanced = true; + } + } + // Preserve inlet timestamps; correction is for comparison only. If the + // clock mapping is unavailable, preserve arrivals during a finite grace. + timestamps.push_back(ts); + chunk.insert(chunk.end(), sample.begin(), sample.end()); + } + write_chunk(); + if (!stopping || advanced) { + last_advance = Clock::now(); + control_->note(streamid, true, false, advanced); + } + if (stopping && Clock::now() - last_advance >= inactivity_grace) { + // Even a post-cutoff sample is not proof of completeness for reordered + // or sparse streams. Leave the inlet open for the same reorder grace. + end_reason = invalid_timestamps ? "unreliable_timestamps" + : (!have_correction || unmapped_samples) ? "clock_unavailable" + : saw_cutoff ? "cutoff_observed" : "inactivity_timeout"; + break; + } + if (!stopping) wait_for_shutdown(chunk_interval); } - } catch (std::exception &e) { - std::cerr << "Error in transfer thread: " << e.what() << std::endl; - offset_shutdown = true; - timed_join_or_detach(offset_thread); + control_->note(streamid, false); + if (end_reason != "cutoff_observed") ++control_->fallback_streams; + log_out("Collection ended for stream ", streamid, ": ", end_reason); + } catch (std::exception &) { + stop_offsets(offset_shutdown); + join_worker(offset_thread, teardown_grace); throw; } - timed_join_or_detach(offset_thread); + stop_offsets(offset_shutdown); + join_worker(offset_thread, teardown_grace); } diff --git a/src/recording.h b/src/recording.h index 0b198ba..3abf92a 100644 --- a/src/recording.h +++ b/src/recording.h @@ -1,48 +1,13 @@ #ifndef RECORDING_H #define RECORDING_H -#include "xdfwriter.h" -#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; -// 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 time that we wait to join a thread, in seconds -const std::chrono::seconds max_join_wait(5); - -using streamid_t = uint32_t; - -// pointer to a thread -using thread_p = std::unique_ptr; -// pointer to a stream inlet -using inlet_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; - +#include +#include +#include /** * A recording process using the lab streaming layer. @@ -55,11 +20,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. */ @@ -67,101 +32,38 @@ class recording { const std::vector &watchfor, std::map syncOptions, bool collect_offsets = true); - /** Destructor. - * Stops the recording and closes the file. - */ + /// Requests shutdown without waiting. Callers must observe completion before normal exit. ~recording(); + /// Fix the recorder-clock cutoff and request catch-up/finalization. Returns immediately. void requestStop() noexcept; + struct FinalizationProgress { + size_t collecting = 0; + size_t catching_up = 0; + size_t fallback_streams = 0; + // Longest inactivity of any outstanding worker (other streams cannot hide a stall). + std::chrono::milliseconds idle{0}; + }; + FinalizationProgress finalizationProgress() const; + /// End catch-up early, preserving footers and recording the explicit truncation reason. + void finishCollecting() noexcept; + /// Wait at most timeout for all workers AND the output file to finish. Does not request stop. + bool waitForFinished(std::chrono::milliseconds timeout) const; + bool isFinished() const { return waitForFinished(std::chrono::milliseconds(0)); } + /// Available after completion: empty on success, otherwise a finalization error. + std::string finalizationError() const; + /// Retain a completion receipt when releasing the nonblocking recording handle. + std::shared_future completionResult() const { return result_; } + recording(const recording &) = delete; + recording &operator=(const recording &) = delete; 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 - 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 - - // 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 - - // 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, const inlet_p &in, std::atomic &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); - - // === 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're 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 - 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; + struct completion; + // The finalizer owns impl, never the UI handle. Dropping the handle cannot close a file + // or join a thread on the caller; completion is published only after impl is destroyed. + std::shared_ptr completion_; + std::shared_future result_; }; #endif diff --git a/src/tcpinterface.cpp b/src/tcpinterface.cpp index 044486b..4422b22 100644 --- a/src/tcpinterface.cpp +++ b/src/tcpinterface.cpp @@ -17,6 +17,14 @@ void RemoteControlSocket::addClient() { void RemoteControlSocket::handleLine(QString s, QTcpSocket *sock) { qInfo() << s; + if (s == "status") { + sock->write(recordingState.toUtf8() + '\n'); + return; + } + if (s == "start" && recordingState != "stopped" && recordingState != "error") { + sock->write("ERROR " + recordingState.toUtf8() + '\n'); + return; + } if (s == "start") emit start(); else if (s == "stop") diff --git a/src/tcpinterface.h b/src/tcpinterface.h index 3eeef1e..080cf41 100644 --- a/src/tcpinterface.h +++ b/src/tcpinterface.h @@ -13,6 +13,7 @@ class RemoteControlSocket : public QObject { QList clients; public: RemoteControlSocket(uint16_t port); + QString recordingState = "stopped"; signals: void refresh_streams(); diff --git a/tests/cli_finalization.py b/tests/cli_finalization.py new file mode 100644 index 0000000..fb7d192 --- /dev/null +++ b/tests/cli_finalization.py @@ -0,0 +1,77 @@ +"""Exercise process exit with the real CLI/recorder and a controlled LSL inlet.""" +import os +from pathlib import Path +import queue +import subprocess +import sys +import tempfile +import threading +import time + + +def run_case(binary, directory, stalled): + path = Path(directory) / ("stalled.xdf" if stalled else "complete.xdf") + env = dict(os.environ) + if stalled: + env["LSL_TEST_STALL"] = "1" + else: + env.pop("LSL_TEST_STALL", None) + proc = subprocess.Popen( + [binary, "--stop-timeout", "0.5" if stalled else "5", str(path), "test"], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + text=True, env=env, + ) + lines = queue.Queue() + output = [] + + def read(): + for line in proc.stdout: + output.append(line) + lines.put(line) + + reader = threading.Thread(target=read, daemon=True) + reader.start() + try: + needle = "TEST: offset query stalled" if stalled else "Started data collection" + deadline = time.monotonic() + 12 + while True: + line = lines.get(timeout=max(0.01, deadline - time.monotonic())) + if needle in line: + break + start = time.monotonic() + proc.stdin.write("\n") + proc.stdin.flush() + proc.wait(timeout=5) + reader.join(timeout=1) + elapsed = time.monotonic() - start + assert proc.returncode == (3 if stalled else 0), "".join(output) + assert elapsed < (1.5 if stalled else 4), f"CLI did not honor its timeout: {elapsed}" + data = path.read_bytes() + assert data.startswith(b"XDF:"), "no recoverable XDF header was flushed" + if stalled: + assert "Finalization timed out" in "".join(output) + # A permanently stalled worker cannot close the writer. Periodic checkpoints + # must nevertheless have preserved complete header and sample chunks. + tags = [] + pos = 4 + while pos < len(data): + width = data[pos] + assert width in (1, 4, 8) + length = int.from_bytes(data[pos + 1:pos + 1 + width], "little") + pos += 1 + width + assert pos + length <= len(data), "checkpoint ended inside an XDF chunk" + tags.append(int.from_bytes(data[pos:pos + 2], "little")) + pos += length + assert 2 in tags and 3 in tags, "recorded samples were not checkpointed" + else: + assert b"" in data, "success reported before footer flush" + print(f"{'stalled' if stalled else 'normal'}: exit={proc.returncode}, {elapsed:.3f}s") + finally: + if proc.poll() is None: + proc.kill() + proc.wait() + + +with tempfile.TemporaryDirectory() as directory: + run_case(sys.argv[1], directory, False) + run_case(sys.argv[1], directory, True) diff --git a/tests/fake_lsl/lsl_cpp.h b/tests/fake_lsl/lsl_cpp.h new file mode 100644 index 0000000..ea48fb0 --- /dev/null +++ b/tests/fake_lsl/lsl_cpp.h @@ -0,0 +1,166 @@ +#pragma once + +// A deterministic inlet for recording finalization tests. Only this test target +// sees it; the recorder and XDF writer are compiled unchanged, without a live +// network dependency. +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace lsl { +using clock = std::chrono::steady_clock; +constexpr int post_clocksync = 1; +enum channel_format_t { cf_int8, cf_int16, cf_int32, cf_float32, cf_double64, cf_string }; +struct timeout_error : std::runtime_error { + timeout_error() : std::runtime_error("test timeout") {} +}; +struct inlet_state { + enum mode { + delayed_result, + unavailable, + stalled_worker, + permanent_stall, + failed_transfer, + queued, + queued_unknown, + stalled_transfer + } behavior; + std::atomic query_started{false}, query_finished{false}; + std::atomic query_calls{0}; + clock::time_point result_ready; + std::mutex mutex; + std::deque> samples; + std::atomic pulled{0}; + void enqueue(double delay, double timestamp) { + std::lock_guard lock(mutex); + samples.emplace_back(clock::now() + std::chrono::duration_cast( + std::chrono::duration(delay)), timestamp); + } + explicit inlet_state(mode behavior) : behavior(behavior) {} +}; + +class stream_info { + public: + std::shared_ptr state; + explicit stream_info(std::shared_ptr state) : state(std::move(state)) {} + std::string name() const { return "FinalizationTest"; } + std::string hostname() const { return "localhost"; } + std::string uid() const { return "finalization-test"; } + std::string source_id() const { return uid(); } + channel_format_t channel_format() const { return cf_float32; } + bool matches_query(const std::string &) const { return true; } + double nominal_srate() const { return 100; } + std::string as_xml() const { + return "FinalizationTest1" + "float32100"; + } +}; + +inline double local_clock() { + return std::chrono::duration(clock::now().time_since_epoch()).count(); +} +inline std::vector resolve_stream(const std::string &, int, double) { return {}; } +inline std::vector resolve_streams() { + return {stream_info(std::make_shared(std::getenv("LSL_TEST_STALL") + ? inlet_state::permanent_stall + : inlet_state::delayed_result))}; +} + +class stream_inlet { + stream_info info_; + bool sent_sample_ = false; + bool clocksync_ = false; + + public: + explicit stream_inlet(const stream_info &info) : info_(info) {} + void open_stream(double) {} + void close_stream() {} + void set_postprocessing(int flags) { clocksync_ = (flags & post_clocksync) != 0; } + stream_info info(double) { return info_; } + int get_channel_count() const { return 1; } + template double pull_sample(std::vector &sample, double timeout) { + if (info_.state->behavior == inlet_state::queued || + info_.state->behavior == inlet_state::queued_unknown) { + const auto until = clock::now() + std::chrono::duration(timeout); + do { + { + std::lock_guard lock(info_.state->mutex); + auto &samples = info_.state->samples; + if (!samples.empty() && samples.front().first <= clock::now()) { + const auto ts = samples.front().second; + samples.pop_front(); + sample.assign(1, T{}); + ++info_.state->pulled; + return ts + (clocksync_ ? 100.0 : 0.0); + } + } + if (timeout == 0) break; + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } while (clock::now() < until); + return 0; + } + if (info_.state->behavior == inlet_state::failed_transfer) { + info_.state->query_started = true; + throw std::runtime_error("simulated transfer failure"); + } + if (!sent_sample_) { + sent_sample_ = true; + if (info_.state->behavior == inlet_state::stalled_transfer) { + info_.state->query_started = true; + std::this_thread::sleep_for(std::chrono::seconds(3)); + } + sample.assign(1, T{}); + return 123; + } + std::this_thread::sleep_for(std::chrono::duration(timeout)); + return 0; + } + template + void pull_chunk_multiplexed(std::vector &chunk, std::vector *timestamps, double) { + chunk.clear(); + timestamps->clear(); + } + double time_correction(double timeout) { + auto &state = *info_.state; + if (state.behavior == inlet_state::queued || state.behavior == inlet_state::stalled_transfer) return 100.0; + if (state.behavior == inlet_state::queued_unknown) { + if (state.query_calls++ % 2) throw std::runtime_error("clock service lost"); + throw timeout_error(); + } + std::lock_guard lock(state.mutex); + if (state.query_calls++ == 0) + state.result_ready = clock::now() + std::chrono::milliseconds(600); + state.query_started = true; + if (state.behavior == inlet_state::permanent_stall) { + std::cout << "TEST: offset query stalled" << std::endl; + for (;;) + std::this_thread::sleep_for(std::chrono::hours(1)); + } + if (state.behavior == inlet_state::stalled_worker) { + // Deliberately exceed both the offset grace and outer join deadline. Even + // an unexpectedly slow worker must finish before the caller can exit the + // process. + std::this_thread::sleep_for(std::chrono::milliseconds(2600)); + } else { + const auto deadline = clock::now() + std::chrono::duration_cast( + std::chrono::duration(timeout)); + if (state.behavior == inlet_state::unavailable || deadline < state.result_ready) { + std::this_thread::sleep_until(deadline); + throw timeout_error(); + } + std::this_thread::sleep_until(state.result_ready); + } + state.query_finished = true; + return 0.0123; + } +}; +} // namespace lsl diff --git a/tests/gui_finalization.cpp b/tests/gui_finalization.cpp new file mode 100644 index 0000000..4039170 --- /dev/null +++ b/tests/gui_finalization.cpp @@ -0,0 +1,126 @@ +#include "mainwindow.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +extern std::atomic release_finalization; +extern std::atomic recording_starts; +extern std::atomic simulate_catchup; + +static void require(bool value, const char *message) { + if (!value) throw std::runtime_error(message); +} +static void pump(int milliseconds) { + QElapsedTimer timer; + timer.start(); + do { + QApplication::processEvents(); + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } while (timer.elapsed() < milliseconds); +} +static QByteArray command(QTcpSocket &socket, const char *text) { + socket.write(text); + socket.flush(); + QElapsedTimer deadline; + deadline.start(); + while (!socket.bytesAvailable() && deadline.elapsed() < 3000) + pump(1); + return socket.readAll(); +} + +int main(int argc, char **argv) { + QApplication app(argc, argv); + try { + QTemporaryDir directory; + QTcpServer reservation; + require(reservation.listen(QHostAddress::LocalHost, 0), "cannot reserve port"); + const auto port = reservation.serverPort(); + reservation.close(); + const auto config = directory.filePath("test.cfg"); + QFile file(config); + require(file.open(QIODevice::WriteOnly), "cannot write config"); + file.write(("StudyRoot=" + directory.path() + + "\nPathTemplate=test.xdf\nRCSEnabled=1\nRCSPort=" + QString::number(port) + + '\n') + .toUtf8()); + file.close(); + const auto configBytes = config.toUtf8(); + MainWindow window(nullptr, configBytes.constData()); + window.show(); + QTcpSocket socket; + socket.connectToHost(QHostAddress::LocalHost, port); + require(socket.waitForConnected(1000), "cannot connect remote control"); + require(command(socket, "start\n") == "OK", "start was not accepted"); + require(recording_starts == 1, "recording was not started"); + int heartbeats = 0; + QTimer heartbeat; + QObject::connect(&heartbeat, &QTimer::timeout, [&] { ++heartbeats; }); + heartbeat.start(10); + QElapsedTimer stop; + stop.start(); + require(command(socket, "stop\n") == "OK", "stop was not accepted"); + require(stop.elapsed() < 200, "stop blocked the GUI"); + require(command(socket, "status\n") == "finishing\n", "premature stopped status"); + require(command(socket, "start\n") == "ERROR finishing\n", "start allowed while finishing"); + auto *start = window.findChild("startButton"); + auto *stopButton = window.findChild("stopButton"); + require(start && stopButton && !start->isEnabled(), "Start enabled before completion"); + simulate_catchup = true; + pump(5200); + require(command(socket, "status\n") == "catching_up\n", "progressing catch-up timed out"); + require(stopButton->text().contains("Finish now"), "controlled finish action unavailable"); + if (argc > 1 && std::string(argv[1]) == "--finish-now") { + QTimer::singleShot(0, [] { + for (auto *widget : QApplication::topLevelWidgets()) + if (auto *dialog = qobject_cast(widget)) + for (auto *button : dialog->buttons()) + if (button->text().startsWith("Finish now")) button->click(); + }); + stopButton->click(); + pump(300); + require(release_finalization, "Finish now did not reach the backend"); + require(command(socket, "status\n") == "stopped\n", "Finish now did not finalize"); + return 0; + } + simulate_catchup = false; + pump(5200); + require(heartbeats > 100, "GUI event loop stopped during stalled finalization"); + require(command(socket, "status\n") == "stalled\n", "stall was not reported"); + require(stopButton->isEnabled() && stopButton->text().contains("Force quit"), + "force-quit action unavailable"); + // Closing while stalled must offer an escape, with Keep waiting as the safe default. + const bool forceQuit = argc > 1 && std::string(argv[1]) == "--force-quit"; + QTimer::singleShot(0, [forceQuit] { + for (auto *widget : QApplication::topLevelWidgets()) + if (auto *dialog = qobject_cast(widget)) { + if (forceQuit) { + for (auto *button : dialog->buttons()) + if (button->text() == "Force quit") button->click(); + } else dialog->reject(); + } + }); + window.close(); + require(!forceQuit, "Force quit did not terminate the process"); + require(window.isVisible(), "Keep waiting closed the window"); + release_finalization = true; + pump(300); + require(command(socket, "status\n") == "stopped\n", "completion not reported"); + require(!window.isVisible(), "pending close was not completed"); + std::cout << "GUI stayed responsive, rejected restart, and closed only after completion\n"; + return 0; + } catch (const std::exception &e) { + release_finalization = true; + std::cerr << e.what() << '\n'; + return 1; + } +} diff --git a/tests/gui_force_exit.py b/tests/gui_force_exit.py new file mode 100644 index 0000000..17e527c --- /dev/null +++ b/tests/gui_force_exit.py @@ -0,0 +1,6 @@ +"""A GUI with permanently pending completion must honor confirmed Force quit.""" +import subprocess +import sys + +result = subprocess.run([sys.argv[1], "--force-quit"], timeout=20) +assert result.returncode == 3, f"expected forced-exit status 3, got {result.returncode}" diff --git a/tests/gui_recording_stub.cpp b/tests/gui_recording_stub.cpp new file mode 100644 index 0000000..3d76692 --- /dev/null +++ b/tests/gui_recording_stub.cpp @@ -0,0 +1,49 @@ +// Isolate the Qt state machine from stream discovery and recording timing. Backend +// completion semantics are tested separately against the real recording.cpp. +#include "recording.h" +#include +#include + +std::atomic release_finalization{false}; +std::atomic recording_starts{0}; +std::atomic simulate_catchup{false}; + +struct recording::completion { + std::atomic stopping{false}; + std::chrono::steady_clock::time_point stopped; + std::promise result; +}; + +recording::recording(const std::string &, const std::vector &, + const std::vector &, std::map, bool) + : completion_(std::make_shared()), + result_(completion_->result.get_future().share()) { + ++recording_starts; +} +recording::~recording() { requestStop(); } +void recording::requestStop() noexcept { + if (completion_->stopping.exchange(true)) return; + completion_->stopped = std::chrono::steady_clock::now(); + std::thread([done = completion_] { + while (!release_finalization) + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + done->result.set_value(""); + }).detach(); +} +bool recording::waitForFinished(std::chrono::milliseconds timeout) const { + return result_.wait_for(timeout) == std::future_status::ready; +} +std::string recording::finalizationError() const { return result_.get(); } + +recording::FinalizationProgress recording::finalizationProgress() const { + FinalizationProgress progress; + if (simulate_catchup) { + completion_->stopped = std::chrono::steady_clock::now(); + progress.collecting = progress.catching_up = 1; + return progress; + } + progress.idle = std::chrono::duration_cast( + std::chrono::steady_clock::now() - completion_->stopped); + return progress; +} +void recording::finishCollecting() noexcept { release_finalization = true; } diff --git a/tests/recording_finalization.cpp b/tests/recording_finalization.cpp new file mode 100644 index 0000000..e7bf154 --- /dev/null +++ b/tests/recording_finalization.cpp @@ -0,0 +1,171 @@ +#include "recording.h" +#include +#include +#include +#include +#include +#include +#include + +static void require(bool value, const char *message) { + if (!value) throw std::runtime_error(message); +} + +static void independent_progress_test() { + auto stalled = std::make_shared(lsl::inlet_state::stalled_transfer); + auto active = std::make_shared(lsl::inlet_state::queued); + const std::string filename = "finalization-independent.xdf"; + recording rec(filename, {lsl::stream_info(stalled), lsl::stream_info(active)}, {}, {}, false); + const auto deadline = lsl::clock::now() + std::chrono::seconds(2); + while (!stalled->query_started && lsl::clock::now() < deadline) + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + require(stalled->query_started, "stalled transfer never started"); + for (int i = 0; i < 7; ++i) active->enqueue(0.1 + i * 0.5, lsl::local_clock() - 110 + i); + rec.requestStop(); + std::this_thread::sleep_for(std::chrono::milliseconds(2500)); + require(rec.finalizationProgress().idle >= std::chrono::milliseconds(2400), + "activity on one inlet concealed another inlet's stall"); + require(rec.waitForFinished(std::chrono::seconds(6)), "workers did not eventually finish"); + std::filesystem::remove(filename); +} + +static void catchup_test(const std::string &scenario) { + const bool unknown = scenario == "unknown_clock"; + auto state = std::make_shared(unknown + ? lsl::inlet_state::queued_unknown : lsl::inlet_state::queued); + const std::string filename = "finalization-" + scenario + ".xdf"; + std::map options; + if (scenario == "clocksync") options["FinalizationTest (localhost)"] = lsl::post_clocksync; + // Deliberately disable offset chunks: cutoff comparison must still use clock mapping. + recording rec(filename, {lsl::stream_info(state)}, {}, options, false); + const double raw_now = lsl::local_clock() - 100.0; + const bool slow = scenario == "catchup"; + const bool early = scenario == "finish_now"; + const bool duplicates = scenario == "nonadvancing"; + if (duplicates) { + for (int i = 0; i < 15; ++i) state->enqueue(0.1 + i * 0.3, raw_now - 1); + } else if (slow) { + for (int i = 0; i < 7; ++i) state->enqueue(0.1 + i, raw_now - 10 + i); + } else { + state->enqueue(0.1, raw_now - 1); + state->enqueue(0.3, raw_now + 0.2); // excluded, even after repeated Stop + state->enqueue(0.6, raw_now - 0.5); // older sample arriving AFTER crossing cutoff + } + rec.requestStop(); + const auto started = lsl::clock::now(); + bool intervened = false; + while (!rec.waitForFinished(std::chrono::milliseconds(20))) { + const auto elapsed = lsl::clock::now() - started; + if (!intervened && elapsed > std::chrono::milliseconds(400)) { + if (early) rec.finishCollecting(); + else rec.requestStop(); + intervened = true; + } + if (slow && elapsed < std::chrono::seconds(7)) + require(rec.finalizationProgress().idle < std::chrono::milliseconds(1500), + "advancing backlog was mistaken for a stall"); + require(elapsed < std::chrono::seconds(12), "catch-up did not finish"); + } + require(rec.finalizationError().empty(), "catch-up failed"); + if (slow) require(lsl::clock::now() - started > std::chrono::seconds(7), + "catch-up stopped before delayed upstream samples arrived"); + std::ifstream file(filename, std::ios::binary); + const std::string data((std::istreambuf_iterator(file)), {}); + const int count = duplicates ? state->pulled.load() : slow ? 7 : early ? 1 : unknown ? 3 : 2; + if (duplicates) { + require(count < 15, "non-advancing timestamps kept collection open indefinitely"); + require(lsl::clock::now() - started < std::chrono::seconds(3), "fallback grace was not bounded"); + } + require(data.find("" + std::to_string(count) + "") != std::string::npos, + "cutoff lost pre-stop data or retained post-stop data"); + const auto reason = early ? "user_requested" : unknown ? "clock_unavailable" + : (slow || duplicates) ? "inactivity_timeout" : "cutoff_observed"; + require(data.find(std::string("") + reason + "") != std::string::npos, + "footer does not explain why collection ended"); + file.close(); + std::filesystem::remove(filename); +} + +int main(int argc, char **argv) { + try { + require(argc == 2, "expected a finalization test scenario"); + const std::string scenario = argv[1]; + if (scenario == "independent_progress") { + independent_progress_test(); + return 0; + } + if (scenario == "catchup" || scenario == "cutoff" || scenario == "clocksync" || + scenario == "unknown_clock" || scenario == "finish_now" || scenario == "nonadvancing") { + catchup_test(scenario); + return 0; + } + const auto mode = scenario == "delayed" ? lsl::inlet_state::delayed_result + : scenario == "unavailable" ? lsl::inlet_state::unavailable + : scenario == "failed" ? lsl::inlet_state::failed_transfer + : lsl::inlet_state::stalled_worker; + auto state = std::make_shared(mode); + const auto filename = "finalization-" + scenario + ".xdf"; + auto rec = std::make_unique( + filename, std::vector{lsl::stream_info(state)}, + std::vector{}, std::map{}); + const auto deadline = lsl::clock::now() + std::chrono::seconds(10); + while ( + !(scenario == "delayed" ? state->query_finished.load() : state->query_started.load()) && + lsl::clock::now() < deadline) + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + require(state->query_started, "offset worker never started"); + const auto stop = lsl::clock::now(); + if (scenario == "abandoned") { + const auto done = rec->completionResult(); + rec.reset(); + require(lsl::clock::now() - stop < std::chrono::milliseconds(100), + "dropping a handle blocked on a stalled worker"); + require(done.wait_for(std::chrono::seconds(6)) == std::future_status::ready, + "background cleanup did not finish"); + require(done.get().empty() && state->query_finished, + "background cleanup did not preserve worker lifetime"); + std::filesystem::remove(filename); + return 0; + } + rec->requestStop(); + rec->requestStop(); // Idempotent, including while a worker is stalled. + require(lsl::clock::now() - stop < std::chrono::milliseconds(100), + "Stop blocked the caller"); + if (scenario == "stalled") + require(!rec->waitForFinished(std::chrono::milliseconds(100)), + "stalled worker reported done"); + require(rec->waitForFinished(std::chrono::seconds(6)), "finalization did not finish"); + require(rec->finalizationError().empty() == (scenario != "failed"), + "finalization did not accurately report worker failure"); + const auto elapsed = std::chrono::duration(lsl::clock::now() - stop).count(); + rec.reset(); + + // Completion must include closing and flushing the writer, not just requesting stop. + std::ifstream file(filename, std::ios::binary); + const std::string contents((std::istreambuf_iterator(file)), {}); + require(contents.substr(0, 4) == "XDF:", + "writer was not flushed before completion"); + require(contents.find(scenario == "failed" + ? "0" + : "1") != std::string::npos, + "stream footer is missing or inconsistent"); + require(contents.find("") != std::string::npos, + "stream footer was not completely flushed"); + if (scenario == "delayed") { + require(state->query_finished, "short waits never obtained the delayed offset"); + require(state->query_calls > 1, "test did not exercise polling across timeouts"); + require(contents.find("") != std::string::npos, "offset missing from footer"); + } else if (scenario == "unavailable") { + require(elapsed < 3.0, "stop waited for the full offset query budget"); + } else if (scenario == "stalled") { + require(state->query_finished, "destructor abandoned a running writer owner"); + } + file.close(); + std::filesystem::remove(filename); + std::cout << scenario << ": file finalized; stop took " << elapsed << " s\n"; + return 0; + } catch (const std::exception &e) { + std::cerr << e.what() << '\n'; + return 1; + } +} diff --git a/xdfwriter/xdfwriter.cpp b/xdfwriter/xdfwriter.cpp index 2e89314..4e72901 100644 --- a/xdfwriter/xdfwriter.cpp +++ b/xdfwriter/xdfwriter.cpp @@ -28,6 +28,7 @@ XDFWriter::XDFWriter(const std::string &filename) file_.push( boost::iostreams::file_descriptor_sink(filename, std::ios::binary | std::ios::trunc)); #endif + file_.exceptions(std::ios::badbit | std::ios::failbit); // [MagicCode] file_ << "XDF:"; // [FileHeader] chunk @@ -46,6 +47,33 @@ void XDFWriter::_write_chunk( _write_chunk_header(tag, content.length(), streamid_p); // [Content] file_ << content; + flush_if_due(tag == chunk_tag_t::fileheader || tag == chunk_tag_t::streamheader || + tag == chunk_tag_t::streamfooter); +} + +void XDFWriter::flush_if_due(bool force) { + const auto now = std::chrono::steady_clock::now(); + if (force || now - last_flush_ >= std::chrono::seconds(1)) { + // Only called after a complete chunk, under write_mut (or during construction). + // This preserves recoverable output on force-quit; it is not a disk fsync. + file_.flush(); + last_flush_ = now; + } +} + +void XDFWriter::close() { + std::lock_guard lock(write_mut); + flush_if_due(true); +#ifdef XDFZ_SUPPORT + file_.reset(); +#else + file_.close(); +#endif +} + +void XDFWriter::checkpoint() { + std::lock_guard lock(write_mut); + flush_if_due(true); } void XDFWriter::_write_chunk_header( @@ -80,6 +108,7 @@ void XDFWriter::write_stream_offset(streamid_t streamid, double now, double offs write_little_endian(file_, now - offset); // [OffsetValue] write_little_endian(file_, offset); + flush_if_due(); } void XDFWriter::write_boundary_chunk() { @@ -89,4 +118,5 @@ void XDFWriter::write_boundary_chunk() { 0xD5, 0x46, 0x73, 0x83, 0xCB, 0xE4}; _write_chunk_header(chunk_tag_t::boundary, sizeof(boundary_uuid)); write_sample_values(file_, boundary_uuid, sizeof(boundary_uuid)); + flush_if_due(true); } diff --git a/xdfwriter/xdfwriter.h b/xdfwriter/xdfwriter.h index 5d48012..d627239 100644 --- a/xdfwriter/xdfwriter.h +++ b/xdfwriter/xdfwriter.h @@ -3,6 +3,7 @@ #include "conversions.h" #include +#include #include #include #include @@ -36,6 +37,8 @@ class XDFWriter { void _write_chunk_header( chunk_tag_t tag, std::size_t length, const streamid_t *streamid_p = nullptr); std::mutex write_mut; + std::chrono::steady_clock::time_point last_flush_ = std::chrono::steady_clock::now(); + void flush_if_due(bool force = false); // write a generic chunk void _write_chunk( @@ -47,6 +50,10 @@ class XDFWriter { * @param filename Filename to write to */ XDFWriter(const std::string &filename); + /// Flush and close after all writing threads have finished; reports output failures. + void close(); + /// Flush completed chunks to the OS while the recording remains open. + void checkpoint(); template void write_data_chunk(streamid_t streamid, const std::vector ×tamps,