diff --git a/docs/server-listener.md b/docs/server-listener.md index 0707114..d53caf9 100644 --- a/docs/server-listener.md +++ b/docs/server-listener.md @@ -120,7 +120,7 @@ This queries the database, loads recent sessions, and monitors for EndTime growt | Client can't connect | Firewall blocking the port | Open the Server Listener port (TCP+UDP) on both machines | | `FindLiveSession` says "no live sessions" | No active recording, or data hasn't flushed yet | Start a recorder on the ADS; use `--wait 90` for slow flush intervals | | `CreateParameterDataAccess` throws | Session loaded before parameters were flushed | Wait for the first data flush (~60–70s), then reload the session | -| Data values don't change between polls | Cached session data from initial load | Reload the session (`SessionManager.Load()`) each poll cycle — see `poll-live-samples.cs` | +| Data values don't change between polls, or arrive ~5 minutes behind the session end | Session loaded from the database instead of the Server Listener (state **LiveNotInServer**) | Enable the Server Listener on the recorder and load the session with the connection string from `SessionSummary.GetConnectionString()`; check `State == Live` first. See `poll-live-samples.cs` | ## Related diff --git a/notebooks/06-live-data-monitoring.ipynb b/notebooks/06-live-data-monitoring.ipynb index 02f98b3..2a154cf 100644 --- a/notebooks/06-live-data-monitoring.ipynb +++ b/notebooks/06-live-data-monitoring.ipynb @@ -75,6 +75,16 @@ "metadata": {}, "outputs": [], "source": [ + "# CreateParameterDataAccess can return None (e.g. for a composite session); skip those parameters.\n", + "pdas = {}\n", + "for param_id in monitor_params:\n", + " pda = session.CreateParameterDataAccess(param_id)\n", + " if pda is None:\n", + " print(f\"Skipping {param_id}: could not create a ParameterDataAccess.\")\n", + " else:\n", + " pdas[param_id] = pda\n", + "monitor_params = list(pdas)\n", + "\n", "history = {p: [] for p in monitor_params}\n", "timestamps = []\n", "\n", @@ -90,18 +100,14 @@ " start_time = max(session.StartTime, end_time - window_ns)\n", "\n", " latest = {}\n", - " for param_id in monitor_params:\n", - " pda = session.CreateParameterDataAccess(param_id)\n", - " try:\n", - " samples = pda.GetSamplesBetween(start_time, end_time)\n", - " if samples.SampleCount > 0:\n", - " val = float(samples.Data(samples.SampleCount - 1))\n", - " latest[param_id] = val\n", - " history[param_id].append(val)\n", - " else:\n", - " latest[param_id] = float('nan')\n", - " finally:\n", - " pda.Dispose()\n", + " for param_id, pda in pdas.items():\n", + " samples = pda.GetSamplesBetween(start_time, end_time)\n", + " if samples.SampleCount > 0:\n", + " val = float(samples.Data[samples.SampleCount - 1])\n", + " latest[param_id] = val\n", + " history[param_id].append(val)\n", + " else:\n", + " latest[param_id] = float('nan')\n", "\n", " timestamps.append(poll * POLL_INTERVAL)\n", "\n", @@ -117,7 +123,9 @@ "except KeyboardInterrupt:\n", " print(\"\\nPolling stopped by user.\")\n", "\n", - "print(f\"\\nCollected {len(timestamps)} samples.\")" + "print(f\"\\nCollected {len(timestamps)} samples.\")\n", + "for pda in pdas.values():\n", + " pda.Dispose()" ] }, { diff --git a/notebooks/sqlrace_helpers.py b/notebooks/sqlrace_helpers.py index 09b8024..37697f7 100644 --- a/notebooks/sqlrace_helpers.py +++ b/notebooks/sqlrace_helpers.py @@ -111,7 +111,18 @@ def load_session(session_manager, session_guid, connection_string=None): key = SessionKey.Parse(session_guid) cs = connection_string or get_connection_string() - client_session = session_manager.Load(key, cs) + summary = session_manager.FindSummaryBy(key, cs) + if summary is None: + raise LookupError(f"Session {session_guid} not found in {cs}") + + # For a Live session this is the recorder's Server Listener connection string. + # A LiveNotInServer session has no listener, so its database copy trails the + # leading edge by the recorder's flush delay (5 minutes by default). + if str(summary.State) == "LiveNotInServer": + print("WARNING: Session is LiveNotInServer - enable the recorder's Server Listener " + "for live data (docs/server-listener.md).") + + client_session = session_manager.Load(summary.Key, summary.GetConnectionString()) return client_session, client_session.Session diff --git a/projects/csharp/LiveDashboard/Program.cs b/projects/csharp/LiveDashboard/Program.cs index 6acd91b..001ac52 100644 --- a/projects/csharp/LiveDashboard/Program.cs +++ b/projects/csharp/LiveDashboard/Program.cs @@ -1,9 +1,12 @@ +using LiveDashboard; + using MESL.SqlRace.Domain; -using MESL.SqlRace.Domain.Infrastructure.Enumerators; +using MESL.SqlRace.Domain.Infrastructure.DataPipeline; using MESL.SqlRace.Domain.Query; + using Microsoft.Extensions.Configuration; + using SqlRace.Examples.Core; -using LiveDashboard; // ─── Configuration ─────────────────────────────────────────── var configuration = new ConfigurationBuilder() @@ -48,7 +51,12 @@ var sessionManager = SessionManager.CreateSessionManager(); IClientSession? clientSession = null; Session? session = null; -dynamic? loadKey = null; // Store the key for reloading + +// A LiveNotInServer session is still recording but has no Server Listener, so the only +// copy is the database's, which trails the leading edge by the recorder's flush delay +// (5 minutes by default). Its EndTime is also fixed at load, so the dashboard would freeze. +const string NoListenerHelp = + "Enable the Server Listener on the recorder for live data - see docs/server-listener.md."; if (!string.IsNullOrEmpty(configuredSessionId)) { @@ -57,13 +65,22 @@ { Console.WriteLine($"Session '{configuredSessionId}' not found. Available:"); foreach (var s in sessions.OrderByDescending(s => s.TimeOfRecording).Take(5)) + { Console.WriteLine($" {s.Identifier} ({s.Key})"); + } + return; } - clientSession = sessionManager.Load(match.Key, connectionString); + if (match.State == SessionState.LiveNotInServer) + { + Console.WriteLine($"Session '{configuredSessionId}' is LiveNotInServer. {NoListenerHelp}"); + return; + } + + var sessionConnectionString = match.GetConnectionString(); + clientSession = sessionManager.Load(match.Key, sessionConnectionString); session = clientSession.Session; - loadKey = match.Key; Console.WriteLine($"Loading configured session: {match.Identifier} ({match.Key})"); } else @@ -71,40 +88,51 @@ // Try sessions newest-first, skip any with no sample data foreach (var candidate in sessions.OrderByDescending(s => s.TimeOfRecording)) { - var tempClient = sessionManager.Load(candidate.Key, connectionString); + if (candidate.State == SessionState.LiveNotInServer) + { + Console.WriteLine($"Skipping session {candidate.Identifier} (LiveNotInServer). {NoListenerHelp}"); + continue; + } + + var sessionConnectionString = candidate.GetConnectionString(); + var tempClient = sessionManager.Load(candidate.Key, sessionConnectionString); var tempSession = tempClient.Session; // Check if the first configured parameter has any data var testParam = paramIds.FirstOrDefault(id => tempSession.Parameters.Any(p => p.Identifier == id)); - if (testParam != null) + if (testParam == null) { - using var testPda = tempSession.CreateParameterDataAccess(testParam); - testPda.GoTo(long.MaxValue); - var testSamples = testPda.GetNextSamples(1, StepDirection.Reverse); - - if (testSamples.SampleCount > 0) - { - clientSession = tempClient; - session = tempSession; - loadKey = candidate.Key; - Console.WriteLine($"Loading session: {candidate.Identifier} ({candidate.Key})"); - break; - } + tempClient.Close(); + Console.WriteLine($"Skipping session {candidate.Identifier} (test parameter not found)"); + continue; } - tempClient.Dispose(); - Console.WriteLine($"Skipping session {candidate.Identifier} (no sample data)"); - } + using var testPda = tempSession.CreateParameterDataAccess(testParam); + var sampleCount = testPda.GetSamplesCount(tempSession.StartTime, tempSession.EndTime); - if (session == null) - { - Console.WriteLine("No sessions with sample data found."); - return; + if (sampleCount == 0) + { + Console.WriteLine($"Skipping session {candidate.Identifier} (no sample data)"); + tempClient.Close(); + continue; + } + + clientSession = tempClient; + session = tempSession; + Console.WriteLine($"Loading session: {candidate.Identifier} ({candidate.Key})"); + break; } } +if (clientSession == null || + session == null) +{ + Console.WriteLine("No sessions with sample data found."); + return; +} + // ─── Resolve available parameters ──────────────────────────── var availableParams = paramIds .Where(id => session.Parameters.Any(p => p.Identifier == id)) @@ -121,61 +149,98 @@ var renderer = new DashboardRenderer(pollingMs); var sessionId = session.Identifier; +var lastEndTime = long.MinValue; +// Create a list of PDAs for the available parameters. +var parameterPdas = new List(availableParams.Count); +foreach (var param in availableParams) +{ + var pda = session.CreateParameterDataAccess(param); + parameterPdas.Add(pda); +} + while (!cts.Token.IsCancellationRequested) { - // Reload session each cycle to pick up newly flushed data - clientSession?.Dispose(); - clientSession = sessionManager.Load(loadKey!, connectionString); - session = clientSession.Session; + try + { + await Task.Delay(pollingMs, cts.Token); + } + catch (OperationCanceledException) + { + break; + } var stats = new List(); - - foreach (var paramId in availableParams) + if (lastEndTime >= session.EndTime) { - // Create a fresh PDA each cycle so we always see the latest live data - using var pda = session.CreateParameterDataAccess(paramId); - pda.GoTo(long.MaxValue); - var samples = pda.GetNextSamples(windowSize, StepDirection.Reverse); + // -- No updates to the session, skip this polling cycle -- + continue; + } - if (samples.SampleCount == 0) + lastEndTime = session.EndTime; + + foreach (var pda in parameterPdas) + { + // Query the PDA for new samples + var pdaEndTime = pda.GetEndTime(); + if (pdaEndTime == null) { - stats.Add(new DashboardRenderer.ParameterStats( - paramId, double.NaN, double.NaN, double.NaN, double.NaN, 0)); + // No samples are available for the PDA. + stats.Add( + new DashboardRenderer.ParameterStats( + pda.ParameterIdentifier, + double.NaN, + double.NaN, + double.NaN, + double.NaN, + 0)); continue; } - var latest = samples.Data[0]; - var min = double.MaxValue; - var max = double.MinValue; - var sum = 0.0; + var maxSampleIntervalNs = (long)(1_000_000_000 / pda.GetMaximumFrequency()); + var duration = maxSampleIntervalNs * windowSize; + var startTime = pdaEndTime.Value - duration; + // Grab specifically the end, Max, Min, and Average values for the parameter over the window. + var dataStatistics = pda.GetDataStatistics( + startTime, + duration, + statisticOption: StatisticOption.End | StatisticOption.Max | StatisticOption.Min | StatisticOption.Mean); - for (var j = 0; j < samples.SampleCount; j++) + if (dataStatistics.NumberOfSamples == 0) { - var v = samples.Data[j]; - if (v < min) min = v; - if (v > max) max = v; - sum += v; + // No samples are available for the PDA in the specified window. + stats.Add( + new DashboardRenderer.ParameterStats( + pda.ParameterIdentifier, + double.NaN, + double.NaN, + double.NaN, + double.NaN, + 0)); + continue; } - stats.Add(new DashboardRenderer.ParameterStats( - paramId, latest, min, max, sum / samples.SampleCount, samples.SampleCount)); + var latest = dataStatistics.EndValue; + var min = dataStatistics.MinimumValue; + var max = dataStatistics.MaximumValue; + + stats.Add( + new DashboardRenderer.ParameterStats( + pda.ParameterIdentifier, + latest, + min, + max, + dataStatistics.MeanValue, + dataStatistics.NumberOfSamples)); } renderer.Render( sessionId, session.EndTime > session.StartTime ? "Data" : "Empty", stats); - - try - { - await Task.Delay(pollingMs, cts.Token); - } - catch (OperationCanceledException) - { - break; - } } Console.SetCursorPosition(0, Console.CursorTop + 1); Console.WriteLine("Dashboard stopped."); -clientSession?.Dispose(); +// Dispose all PDAs once we are done. +parameterPdas.ForEach(x => x.Dispose()); +clientSession.Close(); \ No newline at end of file diff --git a/snippets/csharp/live-data/poll-live-samples.cs b/snippets/csharp/live-data/poll-live-samples.cs index 2826cc1..889a893 100644 --- a/snippets/csharp/live-data/poll-live-samples.cs +++ b/snippets/csharp/live-data/poll-live-samples.cs @@ -3,8 +3,10 @@ // ───────────────────────────────────────────────────────────── // // Demonstrates: Polling a live session's latest samples with async delay and cancellation -// Prerequisites: MESL.SQLRace.API NuGet package, .NET 8, a live session -// Input: A live session key and parameter identifier +// Prerequisites: MESL.SQLRace.API NuGet package, .NET 8, a live session recorded with the +// Server Listener enabled (see docs/server-listener.md) +// Input: A live session key, the connection string of the database it is recording to, +// and a parameter identifier // Output: Latest sample values printed at each poll interval // // Related: snippets/csharp/live-data/subscribe-to-lap-events.cs @@ -32,58 +34,62 @@ var paramId = args.Length > 2 ? args[2] : "Temperature:Sensors"; var sessionManager = SessionManager.CreateSessionManager(); +var sessionSummary = sessionManager.FindSummaryBy(sessionKey, connectionString); -using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); - -Console.WriteLine($"Polling {paramId} every 2s (30s timeout)...\n"); - -// --- Verify the parameter exists before polling --- +// --- Live data only comes from the recorder's Server Listener --- +// For a Live session GetConnectionString() returns the listener's "SqlRaceServer:..." string. +// A LiveNotInServer session has no listener: its database copy trails the leading edge by +// the recorder's flush delay (5 minutes by default), so it never gives live data. +var liveConnectionString = sessionSummary?.State == SessionState.Live ? sessionSummary.GetConnectionString() : null; +if (liveConnectionString is null || !liveConnectionString.StartsWith("SqlRaceServer", StringComparison.OrdinalIgnoreCase)) { - using var initClient = sessionManager.Load(sessionKey, connectionString); - if (!initClient.Session.Parameters.Any(p => p.Identifier == paramId)) - { - Console.WriteLine($"ERROR: Parameter '{paramId}' not found in session."); - Console.WriteLine("Available (first 20):"); - foreach (var p in initClient.Session.Parameters.Take(20)) - Console.WriteLine($" {p.Identifier}"); - return; - } + Console.WriteLine($"ERROR: Session is {sessionSummary?.State.ToString() ?? "not found"}, not Live with a Server Listener."); + Console.WriteLine("Enable the Server Listener on the recorder (docs/server-listener.md) and record a new session."); + return; } -var sessionSummary = sessionManager.FindSummaryBy(sessionKey, connectionString); +using var clientSession = sessionManager.Load(sessionSummary.Key, liveConnectionString); +var session = clientSession.Session; -var pollCount = 0; +if (!session.Parameters.Any(p => p.Identifier == paramId)) +{ + Console.WriteLine($"ERROR: Parameter '{paramId}' not found in session."); + Console.WriteLine("Available (first 20):"); + foreach (var p in session.Parameters.Take(20)) + Console.WriteLine($" {p.Identifier}"); + return; +} -// --- Load the session with server listener so we can pick up live data --- -using var clientSession = sessionManager.Load(sessionSummary.Key, sessionSummary.GetConnectionString()); -var session = clientSession.Session; +using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); +Console.WriteLine($"Polling {paramId} every 2s (30s timeout)...\n"); using var pda = session.CreateParameterDataAccess(paramId); +var pollCount = 0; var lastEndTime = long.MinValue; -while (!cts.Token.IsCancellationRequested) +while (!cts.Token.IsCancellationRequested && session.State == SessionState.Live) { try { await Task.Delay(2000, cts.Token); } catch (OperationCanceledException) { break; } - - // --- Check if the end time is updated. --- + + // --- EndTime is kept up to date by the Server Listener while the session is live --- var endTime = session.EndTime; if (lastEndTime >= endTime) { - // --- End time has not been changed. No new data. --- - Console.WriteLine($" Poll {++pollCount}: no data yet"); + Console.WriteLine($" Poll {++pollCount}: no new data"); continue; } - // --- End time has changed. Update lastEndTime and fetch the latest samples. --- lastEndTime = endTime; - pda.GoTo(lastEndTime); + pda.GoTo(endTime); var samples = pda.GetNextSamples(5, StepDirection.Reverse); if (samples.SampleCount > 0) { + // Reverse reads return newest first. var latest = samples.Data[0]; var latestTime = samples.Timestamp[0]; - Console.WriteLine($" Poll {++pollCount}: latest={latest:F2} at {latestTime} ns ({samples.SampleCount} recent)"); + var behindMs = (endTime - latestTime) / 1_000_000.0; + Console.WriteLine($" Poll {++pollCount}: latest={latest:F2} at {latestTime} ns, {behindMs:F0} ms behind session end"); } else {