Skip to content

Commit 73f9785

Browse files
martincostellomartincostello
authored
[OpenTelemetry] Use AlternateLookup (#7467)
1 parent d84d2af commit 73f9785

5 files changed

Lines changed: 415 additions & 90 deletions

File tree

‎src/OpenTelemetry/Metrics/AggregatorStore.cs‎

Lines changed: 114 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,22 @@ internal sealed class AggregatorStore
3737
private readonly Queue<int>? availableMetricPoints;
3838

3939
private readonly ConcurrentDictionary<Tags, int> tagsToMetricPointIndexDictionary =
40+
#if NET9_0_OR_GREATER
41+
new(TagsComparer.Instance);
42+
#else
4043
new();
44+
#endif
45+
46+
#if NET9_0_OR_GREATER
47+
// Alternate lookup that resolves a MetricPoint directly from the incoming
48+
// tags span, avoiding the copy of the tags into thread-static storage.
49+
// Only used for cumulative aggregation (the non-reclaim lookup path).
50+
private readonly ConcurrentDictionary<Tags, int>.AlternateLookup<ReadOnlySpan<KeyValuePair<string, object?>>> tagsToMetricPointIndexLookup;
51+
52+
// Alternate lookup over the delta-with-reclaim dictionary. Only valid (and
53+
// only initialized) when OutputDelta is true.
54+
private readonly ConcurrentDictionary<Tags, LookupData>.AlternateLookup<ReadOnlySpan<KeyValuePair<string, object?>>> tagsToMetricPointIndexLookupDelta;
55+
#endif
4156

4257
private readonly string name;
4358
private readonly MetricPoint[] metricPoints;
@@ -108,16 +123,30 @@ internal AggregatorStore(
108123
// Newer attributes should be added starting at the index: 2
109124
this.metricPointIndex = 1;
110125

126+
#if NET9_0_OR_GREATER
127+
this.tagsToMetricPointIndexLookup =
128+
this.tagsToMetricPointIndexDictionary.GetAlternateLookup<ReadOnlySpan<KeyValuePair<string, object?>>>();
129+
#endif
130+
111131
// Always reclaim unused MetricPoints for Delta aggregation temporality
112132
if (this.OutputDelta)
113133
{
114134
this.availableMetricPoints = new Queue<int>(cardinalityLimit);
115135

116136
// There is no overload which only takes capacity as the parameter
117-
// Using the DefaultConcurrencyLevel defined in the ConcurrentDictionary class: https://github.com/dotnet/runtime/blob/v7.0.5/src/libraries/System.Collections.Concurrent/src/System/Collections/Concurrent/ConcurrentDictionary.cs#L2020
137+
// Using the DefaultConcurrencyLevel defined in the ConcurrentDictionary class: https://github.com/dotnet/runtime/blob/v10.0.0/src/libraries/System.Collections.Concurrent/src/System/Collections/Concurrent/ConcurrentDictionary.cs#L2054
118138
// We expect at the most (user provided cardinality limit) * 2 entries- one for sorted and one for unsorted input
119-
this.TagsToMetricPointIndexDictionaryDelta =
120-
new ConcurrentDictionary<Tags, LookupData>(concurrencyLevel: Environment.ProcessorCount, capacity: cardinalityLimit * 2);
139+
var concurrencyLevel = Environment.ProcessorCount;
140+
var capacity = cardinalityLimit * 2;
141+
142+
#if NET9_0_OR_GREATER
143+
this.TagsToMetricPointIndexDictionaryDelta = new(concurrencyLevel, capacity, comparer: TagsComparer.Instance);
144+
145+
this.tagsToMetricPointIndexLookupDelta =
146+
this.TagsToMetricPointIndexDictionaryDelta.GetAlternateLookup<ReadOnlySpan<KeyValuePair<string, object?>>>();
147+
#else
148+
this.TagsToMetricPointIndexDictionaryDelta = new(concurrencyLevel, capacity);
149+
#endif
121150

122151
// Add all the indices except for the reserved ones to the queue so that threads have
123152
// readily available access to these MetricPoints for their use.
@@ -685,57 +714,69 @@ private int LookupAggregatorStoreForDeltaWithReclaim(KeyValuePair<string, object
685714

686715
// If the running thread created a new MetricPoint, then the Snapshot method cannot reclaim that MetricPoint because MetricPoint is initialized with a ReferenceCount of 1.
687716
// It can simply return the index.
688-
689717
if (!newMetricPointCreated)
690718
{
691-
// If the running thread did not create the MetricPoint, it could be working on an index that has been reclaimed by Snapshot method.
692-
// This could happen if the thread get switched out by CPU after it retrieves the index but the Snapshot method reclaims it before the thread wakes up again.
719+
index = this.ResolveExistingDeltaMetricPoint(lookupData, length);
720+
}
693721

694-
ref var metricPointAtIndex = ref this.metricPoints[index];
695-
var referenceCount = Interlocked.Increment(ref metricPointAtIndex.ReferenceCount);
722+
return index;
723+
}
696724

697-
if (referenceCount < 0)
698-
{
699-
// Rare case: Snapshot method had already marked the MetricPoint available for reuse as it has not been updated in last collect cycle.
700-
701-
// Example scenario:
702-
// Thread T1 wants to record a measurement for (k1,v1).
703-
// Thread T1 creates a new MetricPoint at index 100 and adds an entry for (k1,v1) in the dictionary with the relevant LookupData value; ReferenceCount of the MetricPoint is 1 at this point.
704-
// Thread T1 completes the update and decrements the ReferenceCount to 0.
705-
// Later, another update thread (could be T1 as well) wants to record a measurement for (k1,v1)
706-
// It looks up the dictionary and retrieves the index as 100. ReferenceCount for the MetricPoint is 0 at this point.
707-
// This update thread gets switched out by the CPU.
708-
// With the reclaim behavior, Snapshot method reclaims the index 100 as the MetricPoint for the index has NoCollectPending and has a ReferenceCount of 0.
709-
// Snapshot thread sets the ReferenceCount to int.MinValue.
710-
// The update thread wakes up and increments the ReferenceCount but finds the value to be negative.
711-
712-
// Retry attempt to get a MetricPoint.
713-
index = this.RemoveStaleEntriesAndGetAvailableMetricPointRare(lookupData, length);
714-
}
715-
else if (metricPointAtIndex.LookupData != lookupData)
716-
{
717-
// Rare case: Another thread with different input tags could have reclaimed this MetricPoint if it was freed up by Snapshot method.
718-
719-
// Example scenario:
720-
// Thread T1 wants to record a measurement for (k1,v1).
721-
// Thread T1 creates a new MetricPoint at index 100 and adds an entry for (k1,v1) in the dictionary with the relevant LookupData value; ReferenceCount of the MetricPoint is 1 at this point.
722-
// Thread T1 completes the update and decrements the ReferenceCount to 0.
723-
// Later, another update thread T2 (could be T1 as well) wants to record a measurement for (k1,v1)
724-
// It looks up the dictionary and retrieves the index as 100. ReferenceCount for the MetricPoint is 0 at this point.
725-
// This update thread T2 gets switched out by the CPU.
726-
// With the reclaim behavior, Snapshot method reclaims the index 100 as the MetricPoint for the index has NoCollectPending and has a ReferenceCount of 0.
727-
// Snapshot thread sets the ReferenceCount to int.MinValue.
728-
// An update thread T3 wants to record a measurement for (k2,v2).
729-
// Thread T3 looks for an available index from the queue and finds index 100.
730-
// Thread T3 creates a new MetricPoint at index 100 and adds an entry for (k2,v2) in the dictionary with the LookupData value for (k2,v2). ReferenceCount of the MetricPoint is 1 at this point.
731-
// The update thread T2 wakes up and increments the ReferenceCount and finds the value to be positive but the LookupData value does not match the one for (k1,v1).
732-
733-
// Remove reference since its not the right MetricPoint.
734-
Interlocked.Decrement(ref metricPointAtIndex.ReferenceCount);
735-
736-
// Retry attempt to get a MetricPoint.
737-
index = this.RemoveStaleEntriesAndGetAvailableMetricPointRare(lookupData, length);
738-
}
725+
// Increments the ReferenceCount for an existing (not newly created) delta MetricPoint
726+
// resolved from the dictionary and validates that it is still the right one, retrying
727+
// via the rare path if the Snapshot method reclaimed it in the meantime.
728+
[MethodImpl(MethodImplOptions.AggressiveInlining)]
729+
private int ResolveExistingDeltaMetricPoint(LookupData lookupData, int length)
730+
{
731+
var index = lookupData.Index;
732+
733+
// If the running thread did not create the MetricPoint, it could be working on an index that has been reclaimed by Snapshot method.
734+
// This could happen if the thread get switched out by CPU after it retrieves the index but the Snapshot method reclaims it before the thread wakes up again.
735+
736+
ref var metricPointAtIndex = ref this.metricPoints[index];
737+
var referenceCount = Interlocked.Increment(ref metricPointAtIndex.ReferenceCount);
738+
739+
if (referenceCount < 0)
740+
{
741+
// Rare case: Snapshot method had already marked the MetricPoint available for reuse as it has not been updated in last collect cycle.
742+
743+
// Example scenario:
744+
// Thread T1 wants to record a measurement for (k1,v1).
745+
// Thread T1 creates a new MetricPoint at index 100 and adds an entry for (k1,v1) in the dictionary with the relevant LookupData value; ReferenceCount of the MetricPoint is 1 at this point.
746+
// Thread T1 completes the update and decrements the ReferenceCount to 0.
747+
// Later, another update thread (could be T1 as well) wants to record a measurement for (k1,v1)
748+
// It looks up the dictionary and retrieves the index as 100. ReferenceCount for the MetricPoint is 0 at this point.
749+
// This update thread gets switched out by the CPU.
750+
// With the reclaim behavior, Snapshot method reclaims the index 100 as the MetricPoint for the index has NoCollectPending and has a ReferenceCount of 0.
751+
// Snapshot thread sets the ReferenceCount to int.MinValue.
752+
// The update thread wakes up and increments the ReferenceCount but finds the value to be negative.
753+
754+
// Retry attempt to get a MetricPoint.
755+
index = this.RemoveStaleEntriesAndGetAvailableMetricPointRare(lookupData, length);
756+
}
757+
else if (metricPointAtIndex.LookupData != lookupData)
758+
{
759+
// Rare case: Another thread with different input tags could have reclaimed this MetricPoint if it was freed up by Snapshot method.
760+
761+
// Example scenario:
762+
// Thread T1 wants to record a measurement for (k1,v1).
763+
// Thread T1 creates a new MetricPoint at index 100 and adds an entry for (k1,v1) in the dictionary with the relevant LookupData value; ReferenceCount of the MetricPoint is 1 at this point.
764+
// Thread T1 completes the update and decrements the ReferenceCount to 0.
765+
// Later, another update thread T2 (could be T1 as well) wants to record a measurement for (k1,v1)
766+
// It looks up the dictionary and retrieves the index as 100. ReferenceCount for the MetricPoint is 0 at this point.
767+
// This update thread T2 gets switched out by the CPU.
768+
// With the reclaim behavior, Snapshot method reclaims the index 100 as the MetricPoint for the index has NoCollectPending and has a ReferenceCount of 0.
769+
// Snapshot thread sets the ReferenceCount to int.MinValue.
770+
// An update thread T3 wants to record a measurement for (k2,v2).
771+
// Thread T3 looks for an available index from the queue and finds index 100.
772+
// Thread T3 creates a new MetricPoint at index 100 and adds an entry for (k2,v2) in the dictionary with the LookupData value for (k2,v2). ReferenceCount of the MetricPoint is 1 at this point.
773+
// The update thread T2 wakes up and increments the ReferenceCount and finds the value to be positive but the LookupData value does not match the one for (k1,v1).
774+
775+
// Remove reference since its not the right MetricPoint.
776+
Interlocked.Decrement(ref metricPointAtIndex.ReferenceCount);
777+
778+
// Retry attempt to get a MetricPoint.
779+
index = this.RemoveStaleEntriesAndGetAvailableMetricPointRare(lookupData, length);
739780
}
740781

741782
return index;
@@ -1033,6 +1074,29 @@ private int FindMetricAggregatorsDefault(ReadOnlySpan<KeyValuePair<string, objec
10331074
return 0;
10341075
}
10351076

1077+
#if NET9_0_OR_GREATER
1078+
// Fast path: look up an existing MetricPoint directly from the incoming
1079+
// tags span. The dictionary stores keys in the given (unsorted) order too,
1080+
// so a previously-seen tag set in its original order resolves here without
1081+
// copying the tags into thread-static storage, constructing a Tags key, or
1082+
// sorting. Misses (new tag sets) fall through to the slower copy-and-add
1083+
// path below.
1084+
if (this.OutputDelta)
1085+
{
1086+
// For delta-with-reclaim a hit must still run the ReferenceCount
1087+
// bookkeeping (and possible reclaim retry), so resolve via the shared
1088+
// helper rather than returning the index directly.
1089+
if (this.tagsToMetricPointIndexLookupDelta.TryGetValue(tags, out var lookupData))
1090+
{
1091+
return this.ResolveExistingDeltaMetricPoint(lookupData, tagLength);
1092+
}
1093+
}
1094+
else if (this.tagsToMetricPointIndexLookup.TryGetValue(tags, out var existingIndex))
1095+
{
1096+
return existingIndex;
1097+
}
1098+
#endif
1099+
10361100
var storage = ThreadStaticStorage.GetStorage();
10371101

10381102
storage.SplitToKeysAndValues(tags, tagLength, out var tagKeysAndValues);

‎src/OpenTelemetry/Metrics/Tags.cs‎

Lines changed: 48 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
// Copyright The OpenTelemetry Authors
22
// SPDX-License-Identifier: Apache-2.0
33

4-
using System.Diagnostics;
54
using System.Runtime.CompilerServices;
65

76
namespace OpenTelemetry.Metrics;
@@ -37,14 +36,58 @@ public readonly bool Equals(Tags other)
3736
return true;
3837
}
3938

40-
var length = ourKvps.Length;
41-
42-
if (length != theirKvps.Length)
39+
if (this.hashCode != other.hashCode)
4340
{
4441
return false;
4542
}
4643

47-
if (this.hashCode != other.hashCode)
44+
return SequenceEqual(ourKvps, theirKvps);
45+
}
46+
47+
public readonly bool Equals(ReadOnlySpan<KeyValuePair<string, object?>> other)
48+
=> SequenceEqual(this.KeyValuePairs, other);
49+
50+
public override readonly int GetHashCode() => this.hashCode;
51+
52+
[MethodImpl(MethodImplOptions.AggressiveInlining)]
53+
internal static int ComputeHashCode(ReadOnlySpan<KeyValuePair<string, object?>> keyValuePairs)
54+
{
55+
#if NET || NETSTANDARD2_1_OR_GREATER
56+
HashCode hashCode = default;
57+
58+
for (var i = 0; i < keyValuePairs.Length; i++)
59+
{
60+
ref readonly var item = ref keyValuePairs[i];
61+
hashCode.Add(item.Key.GetHashCode(StringComparison.Ordinal));
62+
hashCode.Add(item.Value);
63+
}
64+
65+
return hashCode.ToHashCode();
66+
#else
67+
var hash = 17;
68+
69+
for (var i = 0; i < keyValuePairs.Length; i++)
70+
{
71+
ref readonly var item = ref keyValuePairs[i];
72+
unchecked
73+
{
74+
hash = (hash * 31) + item.Key.GetHashCode();
75+
hash = (hash * 31) + (item.Value?.GetHashCode() ?? 0);
76+
}
77+
}
78+
79+
return hash;
80+
#endif
81+
}
82+
83+
[MethodImpl(MethodImplOptions.AggressiveInlining)]
84+
private static bool SequenceEqual(
85+
ReadOnlySpan<KeyValuePair<string, object?>> ourKvps,
86+
ReadOnlySpan<KeyValuePair<string, object?>> theirKvps)
87+
{
88+
var length = ourKvps.Length;
89+
90+
if (length != theirKvps.Length)
4891
{
4992
return false;
5093
}
@@ -120,41 +163,6 @@ public readonly bool Equals(Tags other)
120163
}
121164
}
122165

123-
public override readonly int GetHashCode() => this.hashCode;
124-
125-
[MethodImpl(MethodImplOptions.AggressiveInlining)]
126-
private static int ComputeHashCode(KeyValuePair<string, object?>[] keyValuePairs)
127-
{
128-
Debug.Assert(keyValuePairs != null, "keyValuePairs was null");
129-
130-
#if NET || NETSTANDARD2_1_OR_GREATER
131-
HashCode hashCode = default;
132-
133-
for (var i = 0; i < keyValuePairs.Length; i++)
134-
{
135-
ref var item = ref keyValuePairs[i];
136-
hashCode.Add(item.Key.GetHashCode(StringComparison.Ordinal));
137-
hashCode.Add(item.Value);
138-
}
139-
140-
return hashCode.ToHashCode();
141-
#else
142-
var hash = 17;
143-
144-
for (var i = 0; i < keyValuePairs!.Length; i++)
145-
{
146-
ref var item = ref keyValuePairs[i];
147-
unchecked
148-
{
149-
hash = (hash * 31) + item.Key.GetHashCode();
150-
hash = (hash * 31) + (item.Value?.GetHashCode() ?? 0);
151-
}
152-
}
153-
154-
return hash;
155-
#endif
156-
}
157-
158166
[MethodImpl(MethodImplOptions.AggressiveInlining)]
159167
private static bool AreEqual(in KeyValuePair<string, object?> ours, in KeyValuePair<string, object?> theirs)
160168
{

‎src/OpenTelemetry/Metrics/TagsComparer.cs‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
// Copyright The OpenTelemetry Authors
2+
// SPDX-License-Identifier: Apache-2.0
3+
4+
namespace OpenTelemetry.Metrics;
5+
6+
/// <summary>
7+
/// Equality comparer for <see cref="Tags"/>.
8+
/// </summary>
9+
internal sealed class TagsComparer :
10+
#if NET9_0_OR_GREATER
11+
IAlternateEqualityComparer<ReadOnlySpan<KeyValuePair<string, object?>>, Tags>,
12+
#endif
13+
IEqualityComparer<Tags>
14+
{
15+
public static readonly TagsComparer Instance = new();
16+
17+
public bool Equals(Tags x, Tags y) => x.Equals(y);
18+
19+
public int GetHashCode(Tags obj) => obj.GetHashCode();
20+
21+
#if NET9_0_OR_GREATER
22+
public bool Equals(ReadOnlySpan<KeyValuePair<string, object?>> alternate, Tags other)
23+
=> other.Equals(alternate);
24+
25+
public int GetHashCode(ReadOnlySpan<KeyValuePair<string, object?>> alternate)
26+
=> Tags.ComputeHashCode(alternate);
27+
28+
public Tags Create(ReadOnlySpan<KeyValuePair<string, object?>> alternate)
29+
=> new(alternate.ToArray());
30+
#endif
31+
}

0 commit comments

Comments
 (0)