diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 8b29136..6f5048e 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -38,10 +38,10 @@ jobs: fail-fast: false matrix: config: - - { name: "Ubuntu 22.04", os: ubuntu-22.04 } - - { name: "Ubuntu 24.04", os: ubuntu-24.04 } - - { name: "macOS", os: macos-14, cmake_extra: '-DCMAKE_OSX_ARCHITECTURES="x86_64;arm64"' } - - { name: "Windows", os: windows-latest } + - { name: "Ubuntu 22.04", os: ubuntu-22.04, cli: "install/bin/LabRecorderCLI" } + - { name: "Ubuntu 24.04", os: ubuntu-24.04, cli: "install/bin/LabRecorderCLI" } + - { name: "macOS", os: macos-14, cmake_extra: '-DCMAKE_OSX_ARCHITECTURES="x86_64;arm64"', cli: "install/LabRecorderCLI" } + - { name: "Windows", os: windows-latest, cli: "install/LabRecorderCLI.exe" } steps: - name: Checkout @@ -97,17 +97,27 @@ jobs: # ----------------------------------------------------------------------- # Test CLI # ----------------------------------------------------------------------- - - name: Test CLI (Linux) - if: runner.os == 'Linux' - run: ./install/bin/LabRecorderCLI --help || true + - name: Test CLI + shell: bash + run: ./${{ matrix.config.cli }} --help || true + + # ----------------------------------------------------------------------- + # Integration test: teardown latency and XDF integrity + # ----------------------------------------------------------------------- + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.12' - - name: Test CLI (macOS) - if: runner.os == 'macOS' - run: ./install/LabRecorderCLI --help || true + - name: Install integration test dependencies + run: python -m pip install --upgrade pip pylsl pyxdf - - name: Test CLI (Windows) - if: runner.os == 'Windows' - run: ./install/LabRecorderCLI.exe --help || true + # pylsl brings its own liblsl; it only has to speak the same wire protocol as the liblsl + # bundled with the recorder, not be the same build. Set PYLSL_LIB here if that ever stops + # holding. + - name: Test recording teardown + shell: bash + run: python scripts/test_recording_teardown.py --bin "${{ matrix.config.cli }}" # ----------------------------------------------------------------------- # Package diff --git a/.gitignore b/.gitignore index 2333b58..deafc5b 100644 --- a/.gitignore +++ b/.gitignore @@ -18,3 +18,4 @@ liblsl.deb install-qt.sh .DS_Store .codegraph/ +__pycache__/ diff --git a/scripts/test_recording_teardown.py b/scripts/test_recording_teardown.py new file mode 100644 index 0000000..aa0ceee --- /dev/null +++ b/scripts/test_recording_teardown.py @@ -0,0 +1,431 @@ +#!/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 (1.0 s by default) and with status 0, +* every recorded stream has a stream footer whose sample count matches its data, +* pyxdf does not report the file as damaged, and +* nothing that was sent before the stop is missing from the file. + +The cases cover a plain stop, a stop before any sample arrives, a stop while the recorder is +still subscribing to a source that has gone away, and repeated start/stop cycles. + +Requires ``pylsl`` and ``pyxdf``. +""" + +import argparse +import contextlib +import logging +import os +import subprocess +import sys +import tempfile +import threading +import time + +import pylsl +import pyxdf + +EEG_NAME = "TeardownTestEEG" +MARKER_NAME = "TeardownTestMarkers" +NAMES = (EEG_NAME, MARKER_NAME) +EEG_CHANNELS = 8 +EEG_RATE = 100.0 + +# 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 + + +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), +] + + +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=1.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..55772fa 100644 --- a/src/clirecorder.cpp +++ b/src/clirecorder.cpp @@ -1,6 +1,8 @@ #include "recording.h" #include "xdfwriter.h" +#include + int main(int argc, char **argv) { if (argc < 3 || (argc == 2 && std::string(argv[1]) == "-h")) { std::cout << "Usage: " << argv[0] << " outputfile.xdf 'searchstr' ['searchstr2' ...]\n\n" diff --git a/src/recording.cpp b/src/recording.cpp index 0b8b43e..ea84fba 100644 --- a/src/recording.cpp +++ b/src/recording.cpp @@ -1,194 +1,560 @@ #include "recording.h" //#include "conversions.h" +#include "xdfwriter.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include #include #include +#include +#include #ifdef XDFZ_SUPPORT #include #include #include #endif +// timings in the recording process (e.g., rate of boundary chunks and for cases where a stream +// hangs) approx. interval between boundary chunks +const auto boundary_interval = std::chrono::seconds(10); +// approx. interval between offset measurements +const auto offset_interval = std::chrono::seconds(5); +// approx. interval between resolves for outstanding streams on the watchlist, in seconds +const double resolve_interval = 5; +// timeout of a single resolve attempt, in seconds; the rest of resolve_interval is spent in an +// interruptible wait so that a shutdown request need not wait out a resolve +const double resolve_timeout = 1; +// approx. interval between pulling chunks from outlets +const auto chunk_interval = std::chrono::milliseconds(500); +// maximum waiting time for moving past the headers phase while recording +const auto max_headers_wait = std::chrono::seconds(10); +// maximum waiting time for moving into the footers phase while recording +const auto max_footers_wait = std::chrono::seconds(2); +// maximum waiting time for subscribing to a stream, in seconds (if exceeded, stream subscription +// will take place later) +const double max_open_wait = 5; +// maximum waiting time for a single time correction query, in seconds +const double max_time_correction_wait = 2; +// blocking network calls are issued in slices of this length (in seconds) so that a shutdown +// request is noticed promptly instead of after the full timeout +const double network_poll_interval = 0.2; +// time granted to the stream threads to drain their inlets and write their footers before the +// inlets are forcibly closed +const auto teardown_grace = std::chrono::milliseconds(300); +// maximum time that we wait to join a thread +const auto max_join_wait = std::chrono::seconds(2); + +// steady_clock (not high_resolution_clock, which is an alias for the wall clock in some standard +// libraries) so that waits are unaffected by clock adjustments +using Clock = std::chrono::steady_clock; + +using streamid_t = uint32_t; + +/// thrown by the interruptible helpers when the recording is being torn down +class shutdown_requested : public std::runtime_error { +public: + explicit shutdown_requested(const std::string &what) : std::runtime_error(what) {} +}; + +/** + * A thread paired with a future that becomes ready once the thread body has returned. + * + * std::thread::join() blocks indefinitely, so polling it cannot enforce a deadline: a single call + * against a hung thread never comes back. The future can be waited on with a timeout, and only + * once it is ready do we join (which then returns promptly). A std::packaged_task future is used + * rather than std::async because the latter blocks in its future destructor. + */ +struct worker { + std::thread thread; + std::future done; +}; +// pointer to a worker thread +using worker_p = std::unique_ptr; + +/// start a worker thread running fn +template worker_p spawn_worker(F &&fn) { + auto task = std::make_shared>(std::forward(fn)); + auto w = std::make_unique(); + w->done = task->get_future(); + // the task is kept alive by the lambda, so the worker may be detached safely + w->thread = std::thread([task] { (*task)(); }); + return w; +} + +// pointer to a stream inlet +using inlet_p = std::shared_ptr; +// pointer to a per-stream flag asking that stream's offset thread to finish. Shared rather than +// referenced so that an offset thread which had to be detached cannot outlive its flag. +using offset_flag_p = std::shared_ptr>; +// a list of clock offset estimates (time,value) +using offset_list = std::list>; +// a map from streamid to offset_list +using offset_lists = std::map; + +namespace { + +/// Write one line, atomically with respect to the other recording threads. +/// +/// Every stream has its own thread and they all report progress; an unsynchronised chain of << +/// lets two of them interleave in the middle of a line, which garbles the log and defeats anything +/// that reads it. +template void log_line(std::ostream &out, Args &&...args) { + std::ostringstream line; + (line << ... << std::forward(args)); + line << '\n'; + static std::mutex log_mut; + std::lock_guard lock(log_mut); + out << line.str() << std::flush; +} + +template void log_out(Args &&...args) { + log_line(std::cout, std::forward(args)...); +} + +template void log_err(Args &&...args) { + log_line(std::cerr, std::forward(args)...); +} + +// time spent waiting between two resolves of a watchlist query; the resolve itself already takes +// resolve_timeout, so together they keep the resolve_interval cadence +const auto resolve_pause = std::chrono::duration_cast( + 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 timed_join(worker_p &w, std::chrono::milliseconds duration = max_join_wait) { + if (!w) return true; + // wait on the future rather than calling join() directly: join() has no timeout, so a single + // call against a hung thread would never return and no deadline could be enforced + if (w->done.wait_for(duration) != std::future_status::ready) return false; + w->thread.join(); + w.reset(); + return true; +} + +/** + * @brief timed_join_or_detach Join the worker or detach it if not possible within specified + * duration + * @param w unique_ptr to a worker. Will be reset either way + * @param duration max duration to wait */ -inline bool try_join_once(std::unique_ptr &thread) { - if (thread && thread->joinable()) { - thread->join(); - thread.reset(); - return true; +inline void timed_join_or_detach(worker_p &w, std::chrono::milliseconds duration = max_join_wait) { + if (!timed_join(w, duration)) { + w->thread.detach(); + w.reset(); + log_err("Thread didn't join in time!"); } - 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 timed_join_or_detach Join the workers or detach those that don't finish in time + * @param workers list of workers. Guaranteed to be empty afterwards. + * @param duration duration to wait, shared across all workers */ inline void timed_join_or_detach( - 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; + 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.detach(); + workers.clear(); } } /** - * @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 + * The recording state, and the thread bodies that operate on it. + * + * Every recording thread holds a shared_ptr to this, as does the recording object. A thread that + * could not be joined within the teardown deadline is left running rather than blocking the + * caller, which is typically the UI thread, so the state it writes into has to be able to outlive + * the recording object. Whoever drops the last reference destroys it, and that is what closes the + * file. */ -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) + : file_(filename), offsets_enabled_(collect_offsets), unsorted_(false), streamid_(0), + shutdown_(false), headers_to_finish_(0), streaming_to_finish_(0), + sync_options_by_stream_(std::move(syncOptions)) {} + + /// Deliberately joins nothing: the last recording thread to finish drops the final reference, + /// so this runs on that very thread and joining here would be a self-join. stop_and_join() + /// leaves the worker containers empty, so there is nothing left to clean up. + ~impl() = default; + + /// Spawn the recording threads. Separate from the constructor because the threads need a + /// shared_ptr to this, which shared_from_this() cannot hand out during construction. + void start( + const std::vector &streams, const std::vector &watchfor); + + /// Ask the threads to finish and wait a bounded amount of time for them, leaving any that are + /// still stuck running. Called from the recording object, never from a recording thread. + void stop_and_join() noexcept; + + void requestStop() noexcept; + + // the file stream + XDFWriter file_; // the file output stream + // static information + bool offsets_enabled_; // whether to collect time offset information alongside with the stream + // contents + bool unsorted_; // whether this file may contain unsorted chunks (e.g., of late streams) + + // streamid allocation + std::atomic streamid_; // the highest streamid allocated so far + + // phase-of-recording state (headers, streaming data, or footers) + std::atomic shutdown_; // whether we are trying to shut down + std::condition_variable + shutdown_cv_; // signals shutdown so that every interruptible wait returns at once + std::mutex shutdown_mut_; // protects publication of shutdown_ and of the per-stream offset + // shutdown flags, which the shutdown_cv_ predicates read under it + uint32_t headers_to_finish_; // the number of streams that still need to write their header + // (i.e., are not yet ready to write streaming content) + uint32_t streaming_to_finish_; // the number of streams that still need to finish the streaming + // phase (i.e., are not yet ready for writing their footer) + std::condition_variable + ready_for_streaming_; // condition variable signaling that all streams have finished writing + // their headers and are now ready to write streaming content + std::condition_variable + ready_for_footers_; // condition variable signaling that all streams have finished their + // recording jobs and are now ready to write a footer + std::mutex phase_mut_; // a mutex to protect the phase state + + // inlets with potentially pending network I/O, to be aborted if their thread does not stop in + // time + std::vector active_inlets_; + std::mutex inlets_mut_; // a mutex to protect the active inlet list + + // data structure to collect the time offsets for every stream + offset_lists + offset_lists_; // the clock offset lists for each stream (to be written into the footer) + std::mutex offset_mut_; // a mutex to protect the offset lists + + // data for shutdown / final joining + std::list stream_threads_; // the spawned stream handling threads + worker_p boundary_thread_; // the spawned boundary-recording thread + + // for enabling online sync options + std::map sync_options_by_stream_; + + // === recording thread functions === + + /// record from results of a query (spawn a recording thread for every result produced by the + /// query) + /// @param query The query string + void record_from_query_results(const std::string &query); + + /// record from a given stream (identified by its streaminfo) + /// @param src the stream_info from which to record + /// @param phase_locked whether this is a stream that is locked to the phases (1. Headers, 2. + /// Streaming Content, 3. Footers) + /// Late-added streams (e.g. forgotten devices) are not phase-locked. + void record_from_streaminfo(const lsl::stream_info &src, bool phase_locked); + + /// record boundary markers every few seconds + void record_boundaries(); + + // record ClockOffset chunks from a given stream + void record_offsets(streamid_t streamid, inlet_p in, offset_flag_p offset_shutdown) noexcept; + + // sample collection loop for a numeric stream + template + void typed_transfer_loop(streamid_t streamid, double srate, const inlet_p &in, + double &first_timestamp, double &last_timestamp, uint64_t &sample_count); + + // === interruptible waiting & bounded network calls === + + /// wait until deadline, returning true if the wait was cut short by a shutdown request + /// @param extra an optional additional flag (e.g. a per-stream offset shutdown) that also ends + /// the wait + bool wait_until_shutdown(Clock::time_point deadline, const std::atomic *extra = nullptr); + + /// wait for timeout, returning true if the wait was cut short by a shutdown request + bool wait_for_shutdown(Clock::duration timeout, const std::atomic *extra = nullptr) { + return wait_until_shutdown(Clock::now() + timeout, extra); } - 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); + + // === inlet bookkeeping === + + void register_inlet(const inlet_p &in); + void unregister_inlet(const inlet_p &in) noexcept; + /// close every registered inlet, aborting any blocking socket call in progress + void close_active_inlets() noexcept; + + // === phase registration & condition checks === + // writing is coordinated across threads in three phases to keep the file chunks sorted + + void enter_headers_phase(bool phase_locked); + void leave_headers_phase(bool phase_locked); + void enter_streaming_phase(bool phase_locked); + void leave_streaming_phase(bool phase_locked); + void enter_footers_phase(bool phase_locked); + void leave_footers_phase(bool) { /* Nothing to do. Ignore warning. */ } -} -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) { + // the threads hold a reference to us, so the state they write into outlives a teardown that + // had to leave one of them running + auto self = shared_from_this(); // create a recording thread for each stream for (const auto &stream : streams) stream_threads_.emplace_back( - 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(); }); } -recording::~recording() { +void recording::impl::stop_and_join() noexcept { try { - // set the shutdown flag (from now on no more new streams) + // set the shutdown flag (from now on no more new streams) and wake every waiting thread + requestStop(); + + // give the stream threads a moment to drain their inlets and write their footers by + // themselves; closing an inlet discards what it still holds, so that is a last resort + timed_join_some(stream_threads_, teardown_grace); + if (!stream_threads_.empty()) { + // a thread is stuck in a blocking socket call; closing its inlet aborts that call + close_active_inlets(); + timed_join_or_detach(stream_threads_, max_join_wait); + } + timed_join_or_detach(boundary_thread_, max_join_wait); + log_out("Closing the file."); + } catch (std::exception &e) { + log_out("Error while closing the recording: ", e.what()); + } +} + +recording::recording(const std::string &filename, const std::vector &streams, + const std::vector &watchfor, std::map syncOptions, + bool collect_offsets) + : impl_(std::make_shared(filename, std::move(syncOptions), collect_offsets)) { + try { + impl_->start(streams, watchfor); + } catch (...) { + // some threads may already be running, and our destructor will not run if we throw + impl_->stop_and_join(); + throw; + } +} + +recording::~recording() { impl_->stop_and_join(); } + +void recording::requestStop() noexcept { impl_->requestStop(); } + +void recording::impl::requestStop() noexcept { + { + // publish the flag under the mutex that the shutdown_cv_ predicates read it under: a + // waiter that has just evaluated its predicate as false would otherwise miss the + // notification below and sleep out its full interval + std::lock_guard lock(shutdown_mut_); shutdown_ = true; + } + shutdown_cv_.notify_all(); + + // the phase gates test shutdown_ under phase_mut_, so take and release it for the same reason + { std::lock_guard lock(phase_mut_); } + ready_for_streaming_.notify_all(); + ready_for_footers_.notify_all(); +} + +bool recording::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::impl::stop_offsets(const offset_flag_p &offset_shutdown) noexcept { + { + std::lock_guard lock(shutdown_mut_); + *offset_shutdown = true; + } + shutdown_cv_.notify_all(); +} + +void recording::impl::register_inlet(const inlet_p &in) { + std::lock_guard lock(inlets_mut_); + active_inlets_.push_back(in); +} + +void recording::impl::unregister_inlet(const inlet_p &in) noexcept { + if (!in) return; + std::lock_guard lock(inlets_mut_); + active_inlets_.erase( + std::remove(active_inlets_.begin(), active_inlets_.end(), in), active_inlets_.end()); +} - // 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(); +void recording::impl::close_active_inlets() noexcept { + std::lock_guard lock(inlets_mut_); + for (auto &in : active_inlets_) { + try { + in->close_stream(); + } catch (std::exception &e) { + log_err("Error while closing an inlet: ", e.what()); } - std::cout << "Closing the file." << std::endl; - } catch (std::exception &e) { - std::cout << "Error while closing the recording: " << e.what() << std::endl; } } -void recording::requestStop() noexcept -{ - shutdown_ = true; +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; } -void recording::record_from_query_results(const std::string &query) { +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); } catch (std::exception &e) { - std::cout << "Error in the record_from_query_results thread: " << e.what() << std::endl; + log_out("Error in the record_from_query_results thread: ", e.what()); } } -void recording::record_from_streaminfo(const lsl::stream_info &src, bool phase_locked) { +void recording::impl::record_from_streaminfo(const lsl::stream_info &src, bool phase_locked) { + inlet_p in; try { - double first_timestamp, last_timestamp; + // initialised here because a stream that fails mid-recording still writes a footer + double first_timestamp = 0.0, last_timestamp = 0.0; uint64_t sample_count = 0; + double nominal_srate = 0; // obtain a fresh streamid streamid_t streamid = fresh_streamid(); - inlet_p in; - // --- headers phase try { enter_headers_phase(phase_locked); // open an inlet to read from (and subscribe to data immediately) - in.reset(new lsl::stream_inlet(src)); + in = std::make_shared(src); + register_inlet(in); auto it = sync_options_by_stream_.find(src.name() + " (" + src.hostname() + ")"); if (it != sync_options_by_stream_.end()) in->set_postprocessing(it->second); - try { - in->open_stream(max_open_wait); - 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 (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 - file_.write_stream_header(streamid, in->info().as_xml()); - std::cout << "Received header for stream " << src.name() << "." << std::endl; + // 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 +571,33 @@ 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); break; case lsl::cf_int16: - typed_transfer_loop(streamid, nominal_srate, in, first_timestamp, - last_timestamp, sample_count); + typed_transfer_loop( + streamid, nominal_srate, in, first_timestamp, last_timestamp, sample_count); break; case lsl::cf_int32: - typed_transfer_loop(streamid, nominal_srate, in, first_timestamp, - last_timestamp, sample_count); + typed_transfer_loop( + streamid, nominal_srate, in, first_timestamp, last_timestamp, sample_count); break; case lsl::cf_float32: - typed_transfer_loop(streamid, nominal_srate, in, first_timestamp, - last_timestamp, sample_count); + typed_transfer_loop( + streamid, nominal_srate, in, first_timestamp, last_timestamp, sample_count); break; case lsl::cf_double64: - typed_transfer_loop(streamid, nominal_srate, in, first_timestamp, - last_timestamp, sample_count); + typed_transfer_loop( + streamid, nominal_srate, in, first_timestamp, last_timestamp, sample_count); break; case lsl::cf_string: - typed_transfer_loop(streamid, nominal_srate, in, - first_timestamp, last_timestamp, sample_count); + typed_transfer_loop( + streamid, nominal_srate, in, first_timestamp, last_timestamp, sample_count); break; default: // unsupported channel format @@ -242,9 +606,11 @@ 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 + log_err("Error while recording from ", src.name(), ": ", e.what()); } // --- footers phase @@ -270,66 +636,71 @@ 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; + log_out("Error in the record_from_streaminfo thread: ", e.what()); } + unregister_inlet(in); } -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) { - file_.write_boundary_chunk(); - next_boundary = Clock::now() + boundary_interval; - } + if (wait_for_shutdown(boundary_interval)) break; + file_.write_boundary_chunk(); } } catch (std::exception &e) { - std::cout << "Error in the record_boundaries thread: " << e.what() << std::endl; + 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 + if (wait_for_shutdown(offset_interval, offset_shutdown.get())) break; + + // Query the time offset in one call with the whole budget, not in short slices: the + // query needs a round trip to complete, and restarting it every network_poll_interval + // means it never finishes, so no offset is ever recorded. Teardown does not depend on + // this returning quickly -- the transfer thread stops waiting for us after + // teardown_grace and leaves us running, and we keep the file alive while we do. double offset, now; try { - offset = in->time_correction(2); + offset = in->time_correction(max_time_correction_wait); now = lsl::local_clock(); } catch (lsl::timeout_error &) { - std::cerr << "Timeout in time correction query for stream " << streamid - << std::endl; + 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; + 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 +709,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 +730,26 @@ 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, +void recording::impl::typed_transfer_loop(streamid_t streamid, double srate, const inlet_p &in, double &first_timestamp, double &last_timestamp, uint64_t &sample_count) { // optionally start an offset collection thread for this stream - 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,22 +757,16 @@ 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; @@ -402,18 +774,44 @@ void recording::typed_transfer_loop(streamid_t streamid, double srate, const inl } else last_timestamp = ts; } - // write the actual chunk file_.write_data_chunk(streamid, timestamps, chunk, in->get_channel_count()); sample_count += timestamps.size(); + }; + + // Wait for the first sample, unless the stop got here first. A stream held at the headers + // gate reaches this point with the shutdown already set, having pulled nothing, while its + // inlet has been subscribed and buffering the whole time -- the drain below picks that up. + first_timestamp = 0.0; + while (!shutdown_ && first_timestamp == 0.0) { + const double ts = in->pull_sample(chunk, network_poll_interval); + if (ts == 0.0) continue; + timestamps.assign(1, ts); + write_chunk(); + } + auto next_pull = Clock::now() + chunk_interval; + while (!shutdown_) { + // get a chunk from the stream + in->pull_chunk_multiplexed(chunk, ×tamps, 1e-6); + write_chunk(); + if (wait_until_shutdown(next_pull)) break; next_pull += chunk_interval; - std::this_thread::sleep_until(next_pull); } - } catch (std::exception &e) { - std::cerr << "Error in transfer thread: " << e.what() << std::endl; - offset_shutdown = true; - timed_join_or_detach(offset_thread); + + // one final non-blocking pull, so that samples already buffered in the inlet when the stop + // arrived end up in the file rather than being dropped + try { + in->pull_chunk_multiplexed(chunk, ×tamps, 0.0); + write_chunk(); + } catch (std::exception &e) { + // the inlet was closed under us during teardown; the footer matters more + log_err("Could not drain stream ", streamid, " on stop: ", e.what()); + } + } catch (std::exception &) { + stop_offsets(offset_shutdown); + timed_join_or_detach(offset_thread, teardown_grace); throw; } - timed_join_or_detach(offset_thread); + stop_offsets(offset_shutdown); + timed_join_or_detach(offset_thread, teardown_grace); } diff --git a/src/recording.h b/src/recording.h index 0b198ba..5eee784 100644 --- a/src/recording.h +++ b/src/recording.h @@ -1,48 +1,11 @@ #ifndef RECORDING_H #define RECORDING_H -#include "xdfwriter.h" -#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 +18,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. */ @@ -68,100 +31,19 @@ class recording { bool collect_offsets = true); /** Destructor. - * Stops the recording and closes the file. + * Asks the recording threads to finish and waits a bounded amount of time for them. A thread + * that is still stuck after that is left running; the file is closed once it finishes. */ ~recording(); + /// Ask all recording threads to wrap up. Returns immediately. void requestStop() noexcept; private: - // the file stream - XDFWriter file_; // the file output stream - // static information - bool offsets_enabled_; // whether to collect time offset information alongside with the stream - // contents - bool unsorted_; // whether this file may contain unsorted chunks (e.g., of late streams) - - // streamid allocation - std::atomic streamid_; // the highest streamid allocated so far - - // phase-of-recording state (headers, streaming data, or footers) - std::atomic shutdown_; // whether we are trying to shut down - 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; + /// Shared rather than unique: a recording thread that had to be left running keeps the state + /// it writes into -- the file, the mutexes, the offset lists -- alive until it is done. + std::shared_ptr impl_; }; #endif