diff --git a/scripts/test_recording_teardown.py b/scripts/test_recording_teardown.py new file mode 100644 index 0000000..209d112 --- /dev/null +++ b/scripts/test_recording_teardown.py @@ -0,0 +1,121 @@ +#!/usr/bin/env python +""" +Automated integration test for LabRecorder teardown and XDF integrity. +Tests that LabRecorder stops cleanly and instantly (< 500 ms) and produces valid XDF footers. +""" + +import argparse +import os +import subprocess +import sys +import time +import pylsl +import pyxdf + + +def run_test(cli_path, output_xdf="test_recording.xdf"): + if not os.path.exists(cli_path): + print(f"Error: LabRecorderCLI binary not found at '{cli_path}'") + return False + + if os.path.exists(output_xdf): + os.remove(output_xdf) + + print(f"--- Starting LSL test streams ---") + info_eeg = pylsl.StreamInfo("TestEEG", "EEG", 8, 100, "float32", "test_eeg_source_123") + outlet_eeg = pylsl.StreamOutlet(info_eeg) + + info_marker = pylsl.StreamInfo("TestMarker", "Markers", 1, 0, "string", "test_marker_source_123") + outlet_marker = pylsl.StreamOutlet(info_marker) + + time.sleep(0.5) + + print(f"--- Launching LabRecorderCLI ({cli_path}) ---") + proc = subprocess.Popen( + [cli_path, output_xdf, "name='TestEEG'", "name='TestMarker'"], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + + print(f"--- Streaming samples for 2 seconds ---") + start_time = time.time() + sample_val = 0.0 + while time.time() - start_time < 2.0: + outlet_eeg.push_sample([sample_val] * 8) + sample_val += 1.0 + time.sleep(0.01) + + print(f"--- Triggering shutdown (Enter key to stdin) ---") + t0 = time.perf_counter() + try: + stdout, stderr = proc.communicate(input="\n", timeout=4.0) + except subprocess.TimeoutExpired: + proc.kill() + print("FAIL: LabRecorderCLI hung during shutdown (> 4.0s)!") + return False + + stop_duration = time.perf_counter() - t0 + print(f"--- Teardown completed in {stop_duration:.3f} seconds ---") + + if stop_duration > 1.5: + print(f"FAIL: Shutdown took too long ({stop_duration:.3f}s > 1.5s)") + return False + else: + print(f"PASS: Instant shutdown verified (< 1.5s)") + + if not os.path.exists(output_xdf): + print(f"FAIL: Output file '{output_xdf}' was not created!") + return False + + file_size_kb = os.path.getsize(output_xdf) / 1024.0 + print(f"--- Output XDF file size: {file_size_kb:.2f} KB ---") + + print(f"--- Validating XDF file with pyxdf ---") + try: + streams, header = pyxdf.load_xdf(output_xdf) + except Exception as e: + print(f"FAIL: pyxdf failed to load XDF: {e}") + return False + + if len(streams) != 2: + print(f"FAIL: Expected 2 streams in XDF, got {len(streams)}") + return False + + eeg_stream = next((s for s in streams if s["info"]["name"][0] == "TestEEG"), None) + if not eeg_stream: + print("FAIL: TestEEG stream not found in XDF") + return False + + if len(eeg_stream["time_series"]) == 0: + print("FAIL: TestEEG has 0 recorded samples!") + return False + + print(f"PASS: TestEEG has {len(eeg_stream['time_series'])} samples recorded.") + + # Check footer + if "footer" not in eeg_stream or eeg_stream["footer"]["info"] is None: + print("FAIL: TestEEG is missing footer info!") + return False + + print("PASS: Stream footers are present and valid.") + print("=== ALL INTEGRATION TESTS PASSED ===") + + # Cleanup + if os.path.exists(output_xdf): + os.remove(output_xdf) + + return True + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Test LabRecorder teardown and XDF validity") + parser.add_argument( + "--bin", + default="./build/install/bin/LabRecorderCLI", + help="Path to LabRecorderCLI binary", + ) + args = parser.parse_args() + success = run_test(args.bin) + sys.exit(0 if success else 1) diff --git a/src/mainwindow.cpp b/src/mainwindow.cpp index be82724..c4783b2 100644 --- a/src/mainwindow.cpp +++ b/src/mainwindow.cpp @@ -105,6 +105,21 @@ void MainWindow::statusUpdate() const { .arg(QDir::toNativeSeparators(recFilename), QTime(0,0).addSecs(elapsed).toString("hh:mm:ss"), QString::number(size / 1000)); + + // Check stream statuses for connection warnings + const auto statuses = currentRecording->get_stream_status(); + QStringList warnings; + for (const auto &s : statuses) { + if (s.connection_failed) { + warnings << QString::fromStdString(s.name) + " (TCP timeout)"; + } + } + for (const auto &m : missingStreams) { + warnings << m + " (offline)"; + } + if (!warnings.isEmpty()) { + timeString += QStringLiteral(" | WARNING: Cannot connect to %1").arg(warnings.join(", ")); + } statusBar()->showMessage(timeString); } } diff --git a/src/recording.cpp b/src/recording.cpp index 0b8b43e..2c1599b 100644 --- a/src/recording.cpp +++ b/src/recording.cpp @@ -1,6 +1,7 @@ #include "recording.h" //#include "conversions.h" +#include #include #include #ifdef XDFZ_SUPPORT @@ -36,7 +37,7 @@ inline bool timed_join(thread_p &thread, std::chrono::milliseconds duration = ma const auto start = Clock::now(); while (Clock::now() - start < duration) { if (try_join_once(thread)) return true; - std::this_thread::sleep_for(std::chrono::milliseconds(500)); + std::this_thread::sleep_for(std::chrono::milliseconds(20)); } return false; } @@ -72,7 +73,7 @@ inline void timed_join_or_detach( else ++it; } - std::this_thread::sleep_for(std::chrono::milliseconds(500)); + std::this_thread::sleep_for(std::chrono::milliseconds(20)); } if (!threads.empty()) { std::cout << threads.size() << " stream threads still running!" << std::endl; @@ -87,6 +88,14 @@ recording::recording(const std::string &filename, const std::vector lock(inlets_mut_); + for (auto &in : active_inlets_) { + if (in) { + try { in->close_stream(); } catch (...) {} + } + } + } // stop the threads timed_join_or_detach(stream_threads_, max_join_wait); - if (!timed_join(boundary_thread_, max_join_wait + boundary_interval)) { + if (!timed_join(boundary_thread_, max_join_wait)) { std::cout << "boundary_thread didn't finish in time!" << std::endl; boundary_thread_->detach(); } @@ -119,6 +139,15 @@ recording::~recording() { void recording::requestStop() noexcept { shutdown_ = true; + shutdown_cv_.notify_all(); + { + std::lock_guard lock(inlets_mut_); + for (auto &in : active_inlets_) { + if (in) { + try { in->close_stream(); } catch (...) {} + } + } + } } void recording::record_from_query_results(const std::string &query) { @@ -165,6 +194,29 @@ void recording::record_from_streaminfo(const lsl::stream_info &src, bool phase_l // obtain a fresh streamid streamid_t streamid = fresh_streamid(); + // Find or create telemetry tracker for this stream + std::shared_ptr> sample_counter; + std::shared_ptr> conn_failed; + { + std::lock_guard lock(telemetry_mut_); + auto it = std::find_if(stream_telemetry_.begin(), stream_telemetry_.end(), + [&](const StreamTelemetry &t) { + return t.uid == src.uid() || (t.name == src.name() && t.host == src.hostname()); + }); + if (it != stream_telemetry_.end()) { + sample_counter = it->sample_count; + conn_failed = it->connection_failed; + } else { + StreamTelemetry t; + t.name = src.name(); + t.host = src.hostname(); + t.uid = src.uid(); + sample_counter = t.sample_count; + conn_failed = t.connection_failed; + stream_telemetry_.push_back(t); + } + } + inlet_p in; // --- headers phase @@ -173,6 +225,10 @@ void recording::record_from_streaminfo(const lsl::stream_info &src, bool phase_l // open an inlet to read from (and subscribe to data immediately) in.reset(new lsl::stream_inlet(src)); + { + std::lock_guard lock(inlets_mut_); + active_inlets_.push_back(in); + } auto it = sync_options_by_stream_.find(src.name() + " (" + src.hostname() + ")"); if (it != sync_options_by_stream_.end()) in->set_postprocessing(it->second); @@ -180,6 +236,7 @@ void recording::record_from_streaminfo(const lsl::stream_info &src, bool phase_l in->open_stream(max_open_wait); std::cout << "Opened the stream " << src.name() << "." << std::endl; } catch (lsl::timeout_error &) { + if (conn_failed) conn_failed->store(true); std::cout << "Subscribing to the stream " << src.name() << " is taking relatively long; collection from this stream will be delayed." @@ -213,27 +270,27 @@ void recording::record_from_streaminfo(const lsl::stream_info &src, bool phase_l switch (src.channel_format()) { case lsl::cf_int8: typed_transfer_loop(streamid, nominal_srate, in, first_timestamp, - last_timestamp, sample_count); + last_timestamp, sample_count, sample_counter); break; case lsl::cf_int16: typed_transfer_loop(streamid, nominal_srate, in, first_timestamp, - last_timestamp, sample_count); + last_timestamp, sample_count, sample_counter); break; case lsl::cf_int32: typed_transfer_loop(streamid, nominal_srate, in, first_timestamp, - last_timestamp, sample_count); + last_timestamp, sample_count, sample_counter); break; case lsl::cf_float32: typed_transfer_loop(streamid, nominal_srate, in, first_timestamp, - last_timestamp, sample_count); + last_timestamp, sample_count, sample_counter); break; case lsl::cf_double64: typed_transfer_loop(streamid, nominal_srate, in, first_timestamp, - last_timestamp, sample_count); + last_timestamp, sample_count, sample_counter); break; case lsl::cf_string: typed_transfer_loop(streamid, nominal_srate, in, - first_timestamp, last_timestamp, sample_count); + first_timestamp, last_timestamp, sample_count, sample_counter); break; default: // unsupported channel format @@ -276,6 +333,12 @@ void recording::record_from_streaminfo(const lsl::stream_info &src, bool phase_l leave_footers_phase(phase_locked); throw; } + if (in) { + std::lock_guard lock(inlets_mut_); + active_inlets_.erase( + std::remove(active_inlets_.begin(), active_inlets_.end(), in), + active_inlets_.end()); + } } catch (std::exception &e) { std::cout << "Error in the record_from_streaminfo thread: " << e.what() << std::endl; } @@ -285,7 +348,15 @@ void recording::record_boundaries() { try { auto next_boundary = Clock::now() + boundary_interval; while (!shutdown_) { - std::this_thread::sleep_for(std::chrono::milliseconds(500)); + { + std::unique_lock cv_lock(shutdown_mut_); + if (shutdown_cv_.wait_for(cv_lock, std::chrono::milliseconds(500), [this] { + return shutdown_.load(); + })) { + break; + } + } + if (Clock::now() > next_boundary) { file_.write_boundary_chunk(); next_boundary = Clock::now() + boundary_interval; @@ -301,7 +372,15 @@ void recording::record_offsets( try { while (!shutdown_ && !offset_shutdown) { // sleep for the interval - std::this_thread::sleep_for(offset_interval); + { + std::unique_lock cv_lock(shutdown_mut_); + if (shutdown_cv_.wait_for(cv_lock, offset_interval, [this, &offset_shutdown] { + return shutdown_.load() || offset_shutdown.load(); + })) { + break; + } + } + // query the time offset double offset, now; try { @@ -311,9 +390,10 @@ void recording::record_offsets( std::cerr << "Timeout in time correction query for stream " << streamid << std::endl; } + if (shutdown_ || offset_shutdown) break; file_.write_stream_offset(streamid, now, offset); // also append to the offset lists - std::lock_guard lock(offset_mut_); + std::lock_guard offset_lock(offset_mut_); offset_lists_[streamid].emplace_back(now - offset, offset); } } catch (std::exception &e) { @@ -366,7 +446,8 @@ void recording::enter_footers_phase(bool phase_locked) { 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) { + double &first_timestamp, double &last_timestamp, uint64_t &sample_count, + std::shared_ptr> sample_counter) { // 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, @@ -382,11 +463,12 @@ void recording::typed_transfer_loop(streamid_t streamid, double srate, const inl // Pull the first sample first_timestamp = 0.0; while(!shutdown_ && first_timestamp == 0.0) - first_timestamp = last_timestamp = in->pull_sample(chunk, 4.0); - if (!shutdown_) { + first_timestamp = last_timestamp = in->pull_sample(chunk, 0.1); + if (!shutdown_ && first_timestamp != 0.0) { timestamps.push_back(first_timestamp); file_.write_data_chunk(streamid, timestamps, chunk, (uint32_t)in->get_channel_count()); sample_count += timestamps.size(); + if (sample_counter) sample_counter->store(sample_count); } auto next_pull = Clock::now(); @@ -403,17 +485,36 @@ void recording::typed_transfer_loop(streamid_t streamid, double srate, const inl last_timestamp = ts; } // write the actual chunk - file_.write_data_chunk(streamid, timestamps, chunk, in->get_channel_count()); - sample_count += timestamps.size(); + if (!timestamps.empty()) { + file_.write_data_chunk(streamid, timestamps, chunk, in->get_channel_count()); + sample_count += timestamps.size(); + if (sample_counter) sample_counter->store(sample_count); + } next_pull += chunk_interval; - std::this_thread::sleep_until(next_pull); + std::unique_lock cv_lock(shutdown_mut_); + if (shutdown_cv_.wait_until(cv_lock, next_pull, [this] { return shutdown_.load(); })) { + break; + } } } catch (std::exception &e) { std::cerr << "Error in transfer thread: " << e.what() << std::endl; offset_shutdown = true; + shutdown_cv_.notify_all(); timed_join_or_detach(offset_thread); throw; } + offset_shutdown = true; + shutdown_cv_.notify_all(); timed_join_or_detach(offset_thread); } + +std::vector recording::get_stream_status() const { + std::lock_guard lock(telemetry_mut_); + std::vector result; + for (const auto &t : stream_telemetry_) { + result.push_back({t.name, t.host, t.sample_count ? t.sample_count->load() : 0, + t.connection_failed ? t.connection_failed->load() : false}); + } + return result; +} diff --git a/src/recording.h b/src/recording.h index 0b198ba..edf0dfa 100644 --- a/src/recording.h +++ b/src/recording.h @@ -29,8 +29,8 @@ const auto max_footers_wait = std::chrono::seconds(2); // maximum waiting time for subscribing to a stream, in seconds (if exceeded, stream subscription // will take place later) const double max_open_wait = 5; -// maximum time that we wait to join a thread, in seconds -const std::chrono::seconds max_join_wait(5); +// maximum time that we wait to join a thread +const auto max_join_wait = std::chrono::seconds(2); using streamid_t = uint32_t; @@ -43,6 +43,12 @@ using offset_list = std::list>; // a map from streamid to offset_list using offset_lists = std::map; +struct StreamStatusInfo { + std::string name; + std::string host; + uint64_t sample_count; + bool connection_failed; +}; /** * A recording process using the lab streaming layer. @@ -74,7 +80,21 @@ class recording { void requestStop() noexcept; + /// Get current status and sample count for each stream being recorded + std::vector get_stream_status() const; + private: + struct StreamTelemetry { + std::string name; + std::string host; + std::string uid; + std::shared_ptr> sample_count = std::make_shared>(0); + std::shared_ptr> connection_failed = std::make_shared>(false); + }; + + std::vector stream_telemetry_; + mutable std::mutex telemetry_mut_; + // the file stream XDFWriter file_; // the file output stream // static information @@ -87,6 +107,10 @@ class recording { // phase-of-recording state (headers, streaming data, or footers) std::atomic shutdown_; // whether we are trying to shut down + std::condition_variable shutdown_cv_; // condition variable to wake threads immediately on shutdown + std::mutex shutdown_mut_; // mutex for shutdown condition variable + std::vector active_inlets_; // active inlets to abort on teardown + std::mutex inlets_mut_; // mutex to protect active inlets list uint32_t headers_to_finish_; // the number of streams that still need to write their header // (i.e., are not yet ready to write streaming content) uint32_t streaming_to_finish_; // the number of streams that still need to finish the streaming @@ -137,7 +161,8 @@ class recording { // 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); + double &first_timestamp, double &last_timestamp, uint64_t &sample_count, + std::shared_ptr> sample_counter = nullptr); // === phase registration & condition checks === // writing is coordinated across threads in three phases to keep the file chunks sorted