Multithreaded waveform data file reads for scopesession loading

Author: azonenberg

lib

@@ -5,6 +5,7 @@ This is a running list of significant bug fixes and new features since the last
 ## New features since v0.1.1
 
 * Core: Changed rate limiting sleep in InstrumentThread loop from 10ms to 1ms to avoid bogging down high performance instruments like the ThunderScope
+* Core: Scopesession loading now uses multithreaded IO for significant performance gains especially when many channels and deep history are involved
 * Drivers: ThunderScope now overlaps socket IO and GPU processing of waveforms giving a significant increase in WFM/s rate
 * Filters: Added GPU acceleration for several filters including CDR PLL (7.5x speedup), 100baseTX (10x speedup), eye pattern (25x speedup), histogram (12x speedup), TIE (5.3x speedup) and more (https://github.com/ngscopeclient/scopehal/issues/977).
 * Filters: CDR PLL now outputs the input signal sampled by the recovered clock in a second data stream (https://github.com/ngscopeclient/scopehal/issues/991)

release-notes/CHANGELOG.md

@@ -214,6 +214,33 @@ void HistoryManager::LoadEmptyHistoryToSession(Session& session)
 	}
 }
 
+/**
+	@brief Retroactively modify history for an instrument
+
+	This is used during session loading if we find waveforms in certain legacy formats that need to be converted
+	ex post facto
+ */
+void HistoryManager::Retcon(shared_ptr<Oscilloscope> scope, size_t chan, size_t stream, WaveformBase* wfm)
+{
+	TimePoint tp(wfm->m_startTimestamp, wfm->m_startFemtoseconds);
+
+	for(auto& point : m_history)
+	{
+		if(point->m_time != tp)
+			continue;
+
+		//Make sure we have history for this scope
+		auto jt = point->m_history.find(scope);
+		if(jt == point->m_history.end())
+			return;
+
+		//Overwrite the waveform.
+		//Does not free the old one, it's assumed this is done by SetData() on the scope before calling this function
+		jt->second[StreamDescriptor(scope->GetChannel(chan), stream)] = wfm;
+		break;
+	}
+}
+
 /**
 	@brief Adds new data to the history
 
@@ -262,6 +289,7 @@ void HistoryManager::AddHistory(
 
 	//If we already have a history point for the same exact timestamp, do nothing
 	//Either a bug or we're in append mode
+	//TODO: when loading multiscope scopesessions, do we want to add stuff here??
 	if(HasHistory(tp))
 		return;
 

src/ngscopeclient/HistoryManager.cpp

@@ -2,7 +2,7 @@
 *                                                                                                                      *
 * ngscopeclient                                                                                                        *
 *                                                                                                                      *
-* Copyright (c) 2012-2024 Andrew D. Zonenberg and contributors                                                         *
+* Copyright (c) 2012-2026 Andrew D. Zonenberg and contributors                                                         *
 * All rights reserved.                                                                                                 *
 *                                                                                                                      *
 * Redistribution and use in source and binary forms, with or without modification, are permitted provided that the     *
@@ -77,6 +77,8 @@ class HistoryManager
 
 	bool OnMemoryPressure(MemoryPressureLevel level, MemoryPressureType type, size_t requestedSize);
 
+	void Retcon(std::shared_ptr<Oscilloscope> scope, size_t chan, size_t stream, WaveformBase* wfm);
+
 	void AddHistory(
 		const std::vector<std::shared_ptr<Oscilloscope>>& scopes,
 		bool deleteOld = true,

src/ngscopeclient/HistoryManager.h

@@ -438,11 +438,82 @@ bool Session::LoadWaveformData(int version, const string& dataDir)
 		}
 	}
 
+	//Third pass: history and refresh filters
+	double hstart = GetTime();
+	double cdt = 0;
+	bool converted = false;
+	for(auto point : m_history.m_history)
+	{
+		point->LoadHistoryToSession(*this);
+
+		double cstart = GetTime();
+		if(ConvertLegacyUniformWaveforms())
+			converted = true;
+		cdt += GetTime() - cstart;
+
+		RefreshAllFilters();
+	}
+	double hdt = GetTime() - hstart;
+	LogTrace("History replay took %.3f ms\n", hdt * 1000);
+	if(converted)
+		LogTrace("Legacy waveform conversion took %.3f ms\n", cdt * 1000);
+
 	m_history.SetMaxToCurrentDepth();
 
 	return true;
 }
 
+/**
+	@brief Check for any legacy waveforms stored in sparsev1 format that are actually uniform
+ */
+bool Session::ConvertLegacyUniformWaveforms()
+{
+	bool converted = false;
+
+	for(auto scope : m_oscilloscopes)
+	{
+		for(size_t i=0; i < scope->GetChannelCount(); i++)
+		{
+			auto oc = scope->GetOscilloscopeChannel(i);
+			if(!oc)
+				continue;
+
+			for(size_t j=0; j<oc->GetStreamCount(); j++)
+			{
+				auto data = oc->GetData(j);
+				if(!data)
+					continue;
+
+				auto sacap = dynamic_cast<SparseAnalogWaveform*>(data);
+
+				//Quickly check if the waveform is dense packed, even if it was stored as sparse.
+				//Since we know samples must be monotonic and non-overlapping, we don't have to check every single one!
+				if(sacap)
+				{
+					int64_t nlast = sacap->size() - 1;
+					if( (sacap->m_offsets[0] == 0) &&
+						(sacap->m_offsets[nlast] == nlast) &&
+						(sacap->m_durations[nlast] == 1) )
+					{
+						LogDebug("Found legacy uniform waveform stored on disk as sparse, converting to uniform\n");
+
+						//Waveform was actually uniform, so convert it
+						auto cap = new UniformAnalogWaveform(*sacap);
+						oc->SetData(cap, j);
+
+						//Update history
+						m_history.Retcon(scope, i, j, cap);
+
+						converted = true;
+					}
+				}
+			}
+		}
+	}
+
+	return converted;
+}
+
 /**
 	@brief Loads waveform data for filters that need to be preserved
  */
@@ -520,7 +591,7 @@ bool Session::LoadWaveformDataForFilters(
 
 			//Actually load the waveform
 			string fname = datdir + "/stream" + to_string(i) + ".bin";
-			DoLoadWaveformDataForStream(f, i, fmt, fname);
+			DoLoadWaveformDataForStream(cap, fmt, fname);
 		}
 	}
 
@@ -538,6 +609,7 @@ bool Session::LoadWaveformDataForScope(
 {
 	LogTrace("Loading waveform data for scope \"%s\"\n", scope->m_nickname.c_str());
 	LogIndenter li;
+	double start = GetTime();
 
 	TimePoint time(0, 0);
 	TimePoint newest(0, 0);
@@ -563,7 +635,16 @@ bool Session::LoadWaveformDataForScope(
 			chan->SetData(nullptr, j);
 	}
 
-	//Load the data for each waveform
+	//Save info for each waveform so we know where to load it from
+	struct WaveformLoadInfo
+	{
+		WaveformBase* wfm;
+		string format;
+		string fname;
+	};
+	vector<WaveformLoadInfo> waveformsToLoad;
+
+	//First pass: Load metadata and allocate waveforms etc
 	for(auto it : wavenode)
 	{
 		//Top level metadata
@@ -706,32 +787,49 @@ bool Session::LoadWaveformDataForScope(
 					nstream);
 			}
 
-			DoLoadWaveformDataForStream(
-				scope->GetOscilloscopeChannel(nchan),
-				nstream,
-				formats[i],
-				tmp);
+			//Figure out what needs to be loaded
+			WaveformLoadInfo info;
+			info.wfm = scope->GetOscilloscopeChannel(nchan)->GetData(nstream);
+			info.format = formats[i];
+			info.fname = tmp;
+			waveformsToLoad.push_back(info);
 		}
 
+		//TODO: merge history in multi scope sessions?
 		vector<shared_ptr<Oscilloscope>> temp;
 		temp.push_back(scope);
 		m_history.AddHistory(temp, false, pinned, label);
+	}
 
-		//TODO: this is not good for multiscope
-		//TODO: handle eye patterns (need to know window size for it to work right)
-		RefreshAllFilters();
+	//Second pass: Actually load the waveform data in parallel
+	#pragma omp parallel for
+	for(size_t i=0; i<waveformsToLoad.size(); i++)
+	{
+		auto info = waveformsToLoad[i];
+		DoLoadWaveformDataForStream(info.wfm, info.format, info.fname);
 	}
+
+	//Copy it all to the GPU in one go
+	//For now, use the global transfer queue for this
+	{
+		std::lock_guard<std::mutex> lock(g_vkTransferMutex);
+		g_vkTransferCommandBuffer->begin({});
+
+		for(size_t i=0; i<waveformsToLoad.size(); i++)
+			waveformsToLoad[i].wfm->PrepareForGpuAccessNonblocking(*g_vkTransferCommandBuffer);
+
+		g_vkTransferCommandBuffer->end();
+		g_vkTransferQueue->SubmitAndBlock(*g_vkTransferCommandBuffer);
+	}
+
+	double dt = GetTime() - start;
+	LogTrace("Scope data loaded in %.3f ms\n", dt * 1000);
+
 	return true;
 }
 
-void Session::DoLoadWaveformDataForStream(
-	OscilloscopeChannel* chan,
-	int stream,
-	string format,
-	string fname
-	)
+void Session::DoLoadWaveformDataForStream(WaveformBase* cap, string format, string fname)
 {
-	auto cap = chan->GetData(stream);
 	auto sacap = dynamic_cast<SparseAnalogWaveform*>(cap);
 	auto uacap = dynamic_cast<UniformAnalogWaveform*>(cap);
 	auto sdcap = dynamic_cast<SparseDigitalWaveform*>(cap);
@@ -837,21 +935,6 @@ void Session::DoLoadWaveformDataForStream(
 				ccap->m_durations[j] = stime[1];
 			}
 		}
-
-		//Quickly check if the waveform is dense packed, even if it was stored as sparse.
-		//Since we know samples must be monotonic and non-overlapping, we don't have to check every single one!
-		int64_t nlast = nsamples - 1;
-		if(sacap)
-		{
-			if( (sacap->m_offsets[0] == 0) &&
-				(sacap->m_offsets[nlast] == nlast) &&
-				(sacap->m_durations[nlast] == 1) )
-			{
-				//Waveform was actually uniform, so convert it
-				cap = new UniformAnalogWaveform(*sacap);
-				chan->SetData(cap, stream);
-			}
-		}
 	}
 
 	//Dense packed
@@ -880,7 +963,6 @@ void Session::DoLoadWaveformDataForStream(
 	}
 
 	cap->MarkModifiedFromCpu();
-	cap->PrepareForGpuAccess();
 
 	#ifdef _WIN32
 		delete[] buf;

src/ngscopeclient/Session.cpp

@@ -387,11 +387,8 @@ class Session
 		int version,
 		const YAML::Node& node,
 		const std::string& dataDir);
-	void DoLoadWaveformDataForStream(
-		OscilloscopeChannel* chan,
-		int stream,
-		std::string format,
-		std::string fname);
+	void DoLoadWaveformDataForStream(WaveformBase* cap, std::string format, std::string fname);
+	bool ConvertLegacyUniformWaveforms();
 
 	///@brief Version of the file being loaded
 	int m_fileLoadVersion;