From 106dccc04f240d1ec153032d62ad90ad94866118 Mon Sep 17 00:00:00 2001 From: "J. Ritchie Carroll" Date: Thu, 17 Sep 2026 13:57:33 -0500 Subject: [PATCH] Remove quadratic lookups from subscription and metadata paths Large subscriptions (100K+ signals) spent minutes in nested linear scans during subscribe and metadata refresh. Because the cached-measurement replay runs on the client's command processing thread, that delay also deferred the subsequent ConfirmSignalIndexCache command, which gates a v2+ subscriber's time to first usable measurement. DataPublisher.HandleSubscribeRequest: filter the latest-measurement cache through a HashSet of subscribed signal IDs instead of scanning InputMeasurementKeys per cached measurement, and materialize the result. QueueMeasurementsForProcessing enumerates its parameter several times and ImmediateMeasurements copies the entire cache on every enumeration, so the lazy query paid for both repeatedly. DataPublisher.AcquireMetadata: index the device/phasor/measurement join keys once rather than issuing a DataTable.Compute call per row, each of which parsed a fresh filter expression and rescanned the table. Acronym sets use OrdinalIgnoreCase to match the case-insensitive comparison DataTable filter expressions perform by default. SignalIndexCache and DataSubscriber.FixExpectedMeasurementCounts: replace the per-signal DataTable.Select with a dictionary built once over ActiveMeasurements. DataPublisher.UpdateSignalIndexCache: return the authorized MeasurementKey[] instead of Guid[]. Callers were joining those IDs into a multi-megabyte string and re-parsing it to recover keys they already held. MeasurementKey interns every instance by signal ID and updates it in place rather than replacing it, so re-resolving a held key can never yield anything different. Keys are now collected in subscription order, making the result deterministic where the previous ConcurrentDictionary projection was unordered. Co-Authored-By: Claude Opus 5 --- src/lib/sttp.core/DataPublisher.cs | 87 ++++++++++++++++++++++---- src/lib/sttp.core/DataSubscriber.cs | 24 +++++-- src/lib/sttp.core/SignalIndexCache.cs | 22 +++++-- src/lib/sttp.core/SubscriberAdapter.cs | 17 +++-- 4 files changed, 125 insertions(+), 25 deletions(-) diff --git a/src/lib/sttp.core/DataPublisher.cs b/src/lib/sttp.core/DataPublisher.cs index f4b99318..7e381301 100644 --- a/src/lib/sttp.core/DataPublisher.cs +++ b/src/lib/sttp.core/DataPublisher.cs @@ -2154,9 +2154,14 @@ public virtual void SendNotification(string message) /// Client ID of connection over which to update signal index cache. /// New signal index cache. /// Subscribed measurement keys. - public Guid[] UpdateSignalIndexCache(Guid clientID, SignalIndexCache? signalIndexCache, MeasurementKey[]? inputMeasurementKeys) + /// + /// The subset of the subscriber is authorized to receive, in the order + /// provided. These are the same key instances that were passed in, so callers can assign the result directly. + /// + public MeasurementKey[] UpdateSignalIndexCache(Guid clientID, SignalIndexCache? signalIndexCache, MeasurementKey[]? inputMeasurementKeys) { ConcurrentDictionary reference = new(); + List authorizedKeys = []; List unauthorizedKeys = []; int index = 0; @@ -2175,9 +2180,17 @@ public Guid[] UpdateSignalIndexCache(Guid clientID, SignalIndexCache? signalInde // Validate that subscriber has rights to this signal if (signalID != Guid.Empty && hasRightsFunc(signalID)) + { reference.TryAdd(index++, key); + + // Track authorized keys in subscription order - the reference collection is unordered, + // and callers need the authorized subset to assign back as their input measurement keys + authorizedKeys.Add(key); + } else + { unauthorizedKeys.Add(key.SignalID); + } } } @@ -2246,7 +2259,7 @@ public Guid[] UpdateSignalIndexCache(Guid clientID, SignalIndexCache? signalInde } } - return reference.Select(kvp => kvp.Value.SignalID).ToArray(); + return authorizedKeys.ToArray(); } /// @@ -3255,9 +3268,18 @@ private void HandleSubscribeRequest(SubscriberConnection connection, byte[] buff // If client has subscribed to any cached measurements, queue them up for the client if (TryGetAdapterByName(nameof(LatestMeasurementCache), out IActionAdapter? cacheAdapter)) { - if (cacheAdapter is LatestMeasurementCache cache && subscription.InputMeasurementKeys is not null) + if (cacheAdapter is LatestMeasurementCache cache && subscription.InputMeasurementKeys is { } subscribedKeys) { - IEnumerable cachedMeasurements = cache.LatestMeasurements.Where(measurement => subscription.InputMeasurementKeys.Any(key => key.SignalID == measurement.ID)); + // Hash the subscribed signal IDs so cache filtering is O(cached) instead of O(cached * subscribed). + // A large subscription (100K+ signals) against a populated cache otherwise spent minutes here, and + // since this runs on the client's command processing thread, it also delayed the subsequent + // ConfirmSignalIndexCache command -- which gates the subscriber's time to first usable measurement. + HashSet subscribedSignalIDs = [..subscribedKeys.Select(key => key.SignalID)]; + + // Materialize the filter: QueueMeasurementsForProcessing enumerates its parameter several times, + // and each pass over LatestMeasurements copies the entire cache before filtering it + IMeasurement[] cachedMeasurements = cache.LatestMeasurements.Where(measurement => subscribedSignalIDs.Contains(measurement.ID)).ToArray(); + subscription.QueueMeasurementsForProcessing(cachedMeasurements); } } @@ -3480,35 +3502,78 @@ protected virtual DataSet AcquireMetadata(SubscriberConnection connection, Dicti List rowsToRemove = []; string deviceAcronym; + // Each association check below was a DataTable.Compute call with a freshly interpolated filter expression, + // i.e., an expression parse plus a full table scan per row, making this analysis O(rows * rows). With 100K+ + // measurement records that dominated metadata refresh time. The join keys are indexed once instead. + // Note: DataTable filter expressions compare strings case-insensitively by default, so the sets match on + // acronym without regard to case as well. + DataTable measurementDetail = metadata.Tables["MeasurementDetail"]!; + DataTable deviceDetail = metadata.Tables["DeviceDetail"]!; + + HashSet measurementDeviceAcronyms = new(StringComparer.OrdinalIgnoreCase); + + foreach (DataRow row in measurementDetail.Rows) + measurementDeviceAcronyms.Add(row["DeviceAcronym"].ToNonNullString()); + // Remove device records where no associated measurement records exist - foreach (DataRow row in metadata.Tables["DeviceDetail"]!.Rows) + foreach (DataRow row in deviceDetail.Rows) { deviceAcronym = row["Acronym"].ToNonNullString(); - if (!string.IsNullOrEmpty(deviceAcronym) && (int)metadata.Tables["MeasurementDetail"]!.Compute("Count(DeviceAcronym)", $"DeviceAcronym = '{deviceAcronym}'") == 0) + if (!string.IsNullOrEmpty(deviceAcronym) && !measurementDeviceAcronyms.Contains(deviceAcronym)) rowsToRemove.Add(row); } if (metadata.Tables.Contains("PhasorDetail") && metadata.Tables["PhasorDetail"]!.Columns.Contains("DeviceAcronym")) { + DataTable phasorDetail = metadata.Tables["PhasorDetail"]!; + + HashSet definedDeviceAcronyms = new(StringComparer.OrdinalIgnoreCase); + + foreach (DataRow row in deviceDetail.Rows) + definedDeviceAcronyms.Add(row["Acronym"].ToNonNullString()); + // Remove phasor records where no associated device records exist - foreach (DataRow row in metadata.Tables["PhasorDetail"]!.Rows) + foreach (DataRow row in phasorDetail.Rows) { deviceAcronym = row["DeviceAcronym"].ToNonNullString(); - if (!string.IsNullOrEmpty(deviceAcronym) && (int)metadata.Tables["DeviceDetail"]!.Compute("Count(Acronym)", $"Acronym = '{deviceAcronym}'") == 0) + if (!string.IsNullOrEmpty(deviceAcronym) && !definedDeviceAcronyms.Contains(deviceAcronym)) rowsToRemove.Add(row); } - if (metadata.Tables["PhasorDetail"]!.Columns.Contains("SourceIndex") && metadata.Tables["MeasurementDetail"]!.Columns.Contains("PhasorSourceIndex")) + if (phasorDetail.Columns.Contains("SourceIndex") && measurementDetail.Columns.Contains("PhasorSourceIndex")) { + // Index the defined phasor source indexes per device acronym + Dictionary> phasorSourceIndexes = new(StringComparer.OrdinalIgnoreCase); + + foreach (DataRow row in phasorDetail.Rows) + { + deviceAcronym = row["DeviceAcronym"].ToNonNullString(); + int? sourceIndex = row.ConvertField("SourceIndex"); + + if (string.IsNullOrEmpty(deviceAcronym) || sourceIndex is null) + continue; + + if (!phasorSourceIndexes.TryGetValue(deviceAcronym, out HashSet? sourceIndexes)) + { + sourceIndexes = []; + phasorSourceIndexes.Add(deviceAcronym, sourceIndexes); + } + + sourceIndexes.Add(sourceIndex.Value); + } + // Remove measurement records where no associated phasor records exist - foreach (DataRow row in metadata.Tables["MeasurementDetail"]!.Rows) + foreach (DataRow row in measurementDetail.Rows) { deviceAcronym = row["DeviceAcronym"].ToNonNullString(); int? phasorSourceIndex = row.ConvertField("PhasorSourceIndex"); - if (!string.IsNullOrEmpty(deviceAcronym) && phasorSourceIndex is not null && (int)metadata.Tables["PhasorDetail"]!.Compute("Count(DeviceAcronym)", $"DeviceAcronym = '{deviceAcronym}' AND SourceIndex = {phasorSourceIndex}") == 0) + if (string.IsNullOrEmpty(deviceAcronym) || phasorSourceIndex is null) + continue; + + if (!phasorSourceIndexes.TryGetValue(deviceAcronym, out HashSet? sourceIndexes) || !sourceIndexes.Contains(phasorSourceIndex.Value)) rowsToRemove.Add(row); } } diff --git a/src/lib/sttp.core/DataSubscriber.cs b/src/lib/sttp.core/DataSubscriber.cs index 01594031..282a6655 100644 --- a/src/lib/sttp.core/DataSubscriber.cs +++ b/src/lib/sttp.core/DataSubscriber.cs @@ -3794,6 +3794,22 @@ private void FixExpectedMeasurementCounts() if (!measurementTable.Columns.Contains("FramesPerSecond")) return; + // Index measurement rows by signal ID up front: a DataTable.Select per authorized signal parses a new + // filter expression and rescans the entire table, making this O(authorized * measurements). This runs + // right after a subscription is established, so for large subscriptions it directly delays startup. + Dictionary measurementRows = new(measurementTable.Rows.Count); + + foreach (DataRow measurementRow in measurementTable.Rows) + { + // Rows without a parsable signal ID could never have matched the original filter expression + if (!Guid.TryParse(measurementRow["SignalID"].ToNonNullString(), out Guid rowSignalID)) + continue; + + // First row wins, matching the prior behavior of taking the first filtered row + if (!measurementRows.ContainsKey(rowSignalID)) + measurementRows.Add(rowSignalID, measurementRow); + } + // Get expected measurement counts IEnumerable, Guid>> groups = signalIndexCache.AuthorizedSignalIDs .Where(signalID => subscribedDevicesLookup.TryGetValue(signalID, out _)) @@ -3804,7 +3820,7 @@ private void FixExpectedMeasurementCounts() foreach (IGrouping, Guid> group in groups) { int[] frameRates = group - .Select(signalID => GetFramesPerSecond(measurementTable, signalID)) + .Select(signalID => GetFramesPerSecond(measurementRows, signalID)) .Where(frameRate => frameRate != 0) .ToArray(); @@ -3818,11 +3834,9 @@ private void FixExpectedMeasurementCounts() } } - private static int GetFramesPerSecond(DataTable measurementTable, Guid signalID) + private static int GetFramesPerSecond(Dictionary measurementRows, Guid signalID) { - DataRow? row = measurementTable.Select($"SignalID = '{signalID}'").FirstOrDefault(); - - if (row is null) + if (!measurementRows.TryGetValue(signalID, out DataRow? row)) return 0; return row.Field("SignalType")?.ToUpperInvariant() switch diff --git a/src/lib/sttp.core/SignalIndexCache.cs b/src/lib/sttp.core/SignalIndexCache.cs index 34a82c5d..c12a7b27 100644 --- a/src/lib/sttp.core/SignalIndexCache.cs +++ b/src/lib/sttp.core/SignalIndexCache.cs @@ -106,16 +106,30 @@ public SignalIndexCache(DataSet? dataSource, SignalIndexCache remoteCache) DataTable activeMeasurements = dataSource.Tables["ActiveMeasurements"]!; ConcurrentDictionary reference = new(); + // Index measurement IDs by signal ID up front: a DataTable.Select per referenced signal parses a new + // filter expression and rescans the entire table, making cache translation O(referenced * measurements). + // For a large subscription against a large configuration that is the difference between a moment and minutes. + Dictionary measurementIDs = new(activeMeasurements.Rows.Count); + + foreach (DataRow measurementRow in activeMeasurements.Rows) + { + // Rows without a parsable signal ID could never have matched the original filter expression + if (!Guid.TryParse(measurementRow["SignalID"].ToNonNullString(), out Guid rowSignalID)) + continue; + + // First row wins, matching the prior behavior of taking the first filtered row + if (!measurementIDs.ContainsKey(rowSignalID)) + measurementIDs.Add(rowSignalID, measurementRow["ID"].ToNonNullString(MeasurementKey.Undefined.ToString())); + } + foreach (KeyValuePair signalIndex in remoteCache.Reference) { Guid signalID = signalIndex.Value.SignalID; - DataRow[] filteredRows = activeMeasurements.Select($"SignalID = '{signalID}'"); - if (filteredRows.Length == 0) + if (!measurementIDs.TryGetValue(signalID, out string? measurementID)) continue; - DataRow row = filteredRows[0]; - MeasurementKey key = MeasurementKey.LookUpOrCreate(signalID, row["ID"].ToNonNullString(MeasurementKey.Undefined.ToString())); + MeasurementKey key = MeasurementKey.LookUpOrCreate(signalID, measurementID); reference.TryAdd(signalIndex.Key, key); } diff --git a/src/lib/sttp.core/SubscriberAdapter.cs b/src/lib/sttp.core/SubscriberAdapter.cs index a93f0001..d2c60f00 100644 --- a/src/lib/sttp.core/SubscriberAdapter.cs +++ b/src/lib/sttp.core/SubscriberAdapter.cs @@ -243,10 +243,15 @@ public override MeasurementKey[]? InputMeasurementKeys value.Length > 0 && !new HashSet(base.InputMeasurementKeys ?? []).SetEquals(value)) { // Safe: no lock required for signal index cache here - Guid[] authorizedSignalIDs = m_parent.UpdateSignalIndexCache(ClientID, m_connection.SignalIndexCache, value); + MeasurementKey[] authorizedKeys = m_parent.UpdateSignalIndexCache(ClientID, m_connection.SignalIndexCache, value); + // The authorized keys are the very instances just passed in, so they can be assigned directly. This + // previously joined the authorized signal IDs into a delimited string and re-parsed it, which for a + // large subscription built a multi-megabyte string and re-resolved every signal ID only to arrive + // back at these same keys - MeasurementKey interns each instance by signal ID and updates it in + // place rather than replacing it, so re-resolving can never yield anything different. if (DataSource is not null && m_connection.SignalIndexCache is not null) - value = ParseInputMeasurementKeys(DataSource, false, string.Join("; ", authorizedSignalIDs)); + value = authorizedKeys; } base.InputMeasurementKeys = value; @@ -637,7 +642,7 @@ public void ConfirmSignalIndexCache(Guid clientID) { try { - Guid[] authorizedSignalIDs; + MeasurementKey[] authorizedKeys; lock (m_connection.PendingCacheUpdateLock) { @@ -648,11 +653,13 @@ public void ConfirmSignalIndexCache(Guid clientID) m_connection.PendingSignalIndexCache = null; OnStatusMessage(MessageLevel.Info, $"Applying pending signal cache update for subscriber {clientID} with {nextSignalIndexCache.Reference.Count:N0} records...", nameof(ConfirmSignalIndexCache)); - authorizedSignalIDs = m_parent.UpdateSignalIndexCache(ClientID, nextSignalIndexCache, InputMeasurementKeys); + authorizedKeys = m_parent.UpdateSignalIndexCache(ClientID, nextSignalIndexCache, InputMeasurementKeys); } + // See the InputMeasurementKeys setter: the authorized keys are the instances just passed in, so they + // are assigned directly rather than re-parsed from a joined string of their signal IDs if (DataSource is not null) - base.InputMeasurementKeys = ParseInputMeasurementKeys(DataSource, false, string.Join("; ", authorizedSignalIDs)); + base.InputMeasurementKeys = authorizedKeys; } catch (Exception ex) {