Skip to content

Remove quadratic lookups from subscription and metadata paths #35

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ritchiecarroll wants to merge 1 commit into
base: development
Choose a base branch
from
+125 −25
Open

Remove quadratic lookups from subscription and metadata paths #35

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
87 changes: 76 additions & 11 deletions src/lib/sttp.core/DataPublisher.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2154,9 +2154,14 @@ public virtual void SendNotification(string message)
/// <param name="clientID">Client ID of connection over which to update signal index cache.</param>
/// <param name="signalIndexCache">New signal index cache.</param>
/// <param name="inputMeasurementKeys">Subscribed measurement keys.</param>
public Guid[] UpdateSignalIndexCache(Guid clientID, SignalIndexCache? signalIndexCache, MeasurementKey[]? inputMeasurementKeys)
/// <returns>
/// The subset of <paramref name="inputMeasurementKeys"/> 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.
/// </returns>
public MeasurementKey[] UpdateSignalIndexCache(Guid clientID, SignalIndexCache? signalIndexCache, MeasurementKey[]? inputMeasurementKeys)
{
ConcurrentDictionary<int, MeasurementKey> reference = new();
List<MeasurementKey> authorizedKeys = [];
List<Guid> unauthorizedKeys = [];
int index = 0;

Expand All @@ -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);
}
}
}

Expand Down Expand Up @@ -2246,7 +2259,7 @@ public Guid[] UpdateSignalIndexCache(Guid clientID, SignalIndexCache? signalInde
}
}

return reference.Select(kvp => kvp.Value.SignalID).ToArray();
return authorizedKeys.ToArray();
}

/// <summary>
Expand Down Expand Up @@ -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<IMeasurement> 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<Guid> 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);
}
}
Expand Down Expand Up @@ -3480,35 +3502,78 @@ protected virtual DataSet AcquireMetadata(SubscriberConnection connection, Dicti
List<DataRow> 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<string> 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<string> 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<string, HashSet<int>> phasorSourceIndexes = new(StringComparer.OrdinalIgnoreCase);

foreach (DataRow row in phasorDetail.Rows)
{
deviceAcronym = row["DeviceAcronym"].ToNonNullString();
int? sourceIndex = row.ConvertField<int?>("SourceIndex");

if (string.IsNullOrEmpty(deviceAcronym) || sourceIndex is null)
continue;

if (!phasorSourceIndexes.TryGetValue(deviceAcronym, out HashSet<int>? 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<int?>("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<int>? sourceIndexes) || !sourceIndexes.Contains(phasorSourceIndex.Value))
rowsToRemove.Add(row);
}
}
Expand Down
24 changes: 19 additions & 5 deletions src/lib/sttp.core/DataSubscriber.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Guid, DataRow> 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<IGrouping<DeviceStatisticsHelper<SubscribedDevice>, Guid>> groups = signalIndexCache.AuthorizedSignalIDs
.Where(signalID => subscribedDevicesLookup.TryGetValue(signalID, out _))
Expand All @@ -3804,7 +3820,7 @@ private void FixExpectedMeasurementCounts()
foreach (IGrouping<DeviceStatisticsHelper<SubscribedDevice>, Guid> group in groups)
{
int[] frameRates = group
.Select(signalID => GetFramesPerSecond(measurementTable, signalID))
.Select(signalID => GetFramesPerSecond(measurementRows, signalID))
.Where(frameRate => frameRate != 0)
.ToArray();

Expand All @@ -3818,11 +3834,9 @@ private void FixExpectedMeasurementCounts()
}
}

private static int GetFramesPerSecond(DataTable measurementTable, Guid signalID)
private static int GetFramesPerSecond(Dictionary<Guid, DataRow> 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<string>("SignalType")?.ToUpperInvariant() switch
Expand Down
22 changes: 18 additions & 4 deletions src/lib/sttp.core/SignalIndexCache.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,16 +106,30 @@ public SignalIndexCache(DataSet? dataSource, SignalIndexCache remoteCache)
DataTable activeMeasurements = dataSource.Tables["ActiveMeasurements"]!;
ConcurrentDictionary<int, MeasurementKey> 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<Guid, string> 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<int, MeasurementKey> 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);
}

Expand Down
17 changes: 12 additions & 5 deletions src/lib/sttp.core/SubscriberAdapter.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,10 +243,15 @@ public override MeasurementKey[]? InputMeasurementKeys
value.Length > 0 && !new HashSet<MeasurementKey>(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;
Expand Down Expand Up @@ -637,7 +642,7 @@ public void ConfirmSignalIndexCache(Guid clientID)
{
try
{
Guid[] authorizedSignalIDs;
MeasurementKey[] authorizedKeys;

lock (m_connection.PendingCacheUpdateLock)
{
Expand All @@ -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)
{
Expand Down