Skip to content

Commit 9e47a01

Browse files
ReubenBondReubenBond
committed
perf(runtime): key silo callbacks by correlation id
1 parent 963842c commit 9e47a01

3 files changed

Lines changed: 30 additions & 46 deletions

File tree

‎src/Orleans.Core/Messaging/StripedCallbackDictionary.cs‎

Lines changed: 7 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -55,38 +55,38 @@ private Stripe GetStripe(CorrelationId correlationId)
5555
/// Attempts to add the specified key and value to the dictionary.
5656
/// </summary>
5757
[MethodImpl(MethodImplOptions.AggressiveInlining)]
58-
public bool TryAdd(GrainId owner, CorrelationId id, TValue value)
58+
public bool TryAdd(CorrelationId id, TValue value)
5959
{
6060
var stripe = GetStripe(id);
6161
lock (stripe.Lock)
6262
{
63-
return stripe.Dictionary.TryAdd(new(owner, id), value);
63+
return stripe.Dictionary.TryAdd(id, value);
6464
}
6565
}
6666

6767
/// <summary>
6868
/// Attempts to get the value associated with the specified key.
6969
/// </summary>
7070
[MethodImpl(MethodImplOptions.AggressiveInlining)]
71-
public bool TryGetValue(GrainId owner, CorrelationId id, [NotNullWhen(true)] out TValue? value)
71+
public bool TryGetValue(CorrelationId id, [NotNullWhen(true)] out TValue? value)
7272
{
7373
var stripe = GetStripe(id);
7474
lock (stripe.Lock)
7575
{
76-
return stripe.Dictionary.TryGetValue(new(owner, id), out value);
76+
return stripe.Dictionary.TryGetValue(id, out value);
7777
}
7878
}
7979

8080
/// <summary>
8181
/// Attempts to remove the value with the specified key.
8282
/// </summary>
8383
[MethodImpl(MethodImplOptions.AggressiveInlining)]
84-
public bool TryRemove(GrainId owner, CorrelationId id, [NotNullWhen(true)] out TValue? value)
84+
public bool TryRemove(CorrelationId id, [NotNullWhen(true)] out TValue? value)
8585
{
8686
var stripe = GetStripe(id);
8787
lock (stripe.Lock)
8888
{
89-
return stripe.Dictionary.Remove(new(owner, id), out value);
89+
return stripe.Dictionary.Remove(id, out value);
9090
}
9191
}
9292

@@ -180,8 +180,6 @@ private sealed class Stripe
180180
#else
181181
public readonly object Lock = new();
182182
#endif
183-
public readonly Dictionary<CallbackKey, TValue> Dictionary = new();
183+
public readonly Dictionary<CorrelationId, TValue> Dictionary = new();
184184
}
185-
186-
private readonly record struct CallbackKey(GrainId Owner, CorrelationId Id);
187185
}

‎src/Orleans.Runtime/Core/InsideRuntimeClient.cs‎

Lines changed: 8 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ internal sealed partial class InsideRuntimeClient : IRuntimeClient, ILifecyclePa
3030
private readonly ILogger invokeExceptionLogger;
3131
private readonly ILoggerFactory loggerFactory;
3232
private readonly SiloMessagingOptions messagingOptions;
33+
// MessageFactory assigns unique correlation ids to every request created by this runtime client.
3334
private readonly StripedCallbackDictionary<CallbackData> callbacks;
3435
private readonly InterfaceToImplementationMappingCache interfaceToImplementationMapping;
3536
private readonly SharedCallbackData sharedCallbackData;
@@ -87,15 +88,15 @@ public InsideRuntimeClient(
8788

8889
var callbackDataLogger = loggerFactory.CreateLogger<CallbackData>();
8990
this.sharedCallbackData = new SharedCallbackData(
90-
msg => this.UnregisterCallback(msg.SendingGrain, msg.Id),
91+
msg => this.UnregisterCallback(msg.Id),
9192
callbackDataLogger,
9293
this.messagingOptions.ResponseTimeout,
9394
this.messagingOptions.CancelRequestOnTimeout,
9495
this.messagingOptions.WaitForCancellationAcknowledgement,
9596
cancellationManager: null!);
9697

9798
this.systemSharedCallbackData = new SharedCallbackData(
98-
msg => this.UnregisterCallback(msg.SendingGrain, msg.Id),
99+
msg => this.UnregisterCallback(msg.Id),
99100
callbackDataLogger,
100101
this.messagingOptions.SystemResponseTimeout,
101102
cancelOnTimeout: false,
@@ -194,7 +195,7 @@ public void SendRequest(
194195
return;
195196
}
196197

197-
callbacks.TryAdd(message.SendingGrain, message.Id, callbackData);
198+
callbacks.TryAdd(message.Id, callbackData);
198199
callbackData.SubscribeForCancellation(cancellationToken);
199200
}
200201
else
@@ -235,9 +236,9 @@ public void SendResponse(Message request, Response response)
235236
/// <summary>
236237
/// UnRegister a callback.
237238
/// </summary>
238-
private void UnregisterCallback(GrainId owner, CorrelationId correlationId)
239+
private void UnregisterCallback(CorrelationId correlationId)
239240
{
240-
callbacks.TryRemove(owner, correlationId, out _);
241+
callbacks.TryRemove(correlationId, out _);
241242
}
242243

243244
public void SniffIncomingMessage(Message message)
@@ -467,7 +468,7 @@ public void ReceiveResponse(Message message)
467468

468469
private void ProcessResponseCallback(Message message)
469470
{
470-
if (callbacks.TryRemove(message.TargetGrain, message.Id, out var callbackData))
471+
if (callbacks.TryRemove(message.Id, out var callbackData))
471472
{
472473
// IMPORTANT: we do not schedule the response callback via the scheduler, since the only thing it does
473474
// is to resolve/break the resolver. The continuations/waits that are based on this resolution will be scheduled as work items.
@@ -482,7 +483,7 @@ private void ProcessResponseCallback(Message message)
482483
private void ProcessStatusResponse(Message message)
483484
{
484485
var status = (StatusResponse)message.BodyObject!;
485-
callbacks.TryGetValue(message.TargetGrain, message.Id, out var callback);
486+
callbacks.TryGetValue(message.Id, out var callback);
486487
var request = callback?.Message;
487488
if (request is not null)
488489
{

‎test/Orleans.Runtime.Tests/StripedCallbackDictionaryTests.cs‎

Lines changed: 15 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,6 @@ namespace Tester;
88
[TestCategory("BVT")]
99
public class StripedCallbackDictionaryTests
1010
{
11-
private static readonly GrainId Owner = GrainId.Create("test", "owner");
1211
private static readonly Action<int, object?> EmptyVisitor = static (_, _) => { };
1312

1413
[Fact]
@@ -36,27 +35,13 @@ public void AddGetAndRemovePreserveValue()
3635
var dictionary = new StripedCallbackDictionary<string>();
3736
var id = new CorrelationId(42);
3837

39-
Assert.True(dictionary.TryAdd(Owner, id, "value"));
40-
Assert.False(dictionary.TryAdd(Owner, id, "duplicate"));
41-
Assert.True(dictionary.TryGetValue(Owner, id, out var value));
38+
Assert.True(dictionary.TryAdd(id, "value"));
39+
Assert.False(dictionary.TryAdd(id, "duplicate"));
40+
Assert.True(dictionary.TryGetValue(id, out var value));
4241
Assert.Equal("value", value);
43-
Assert.True(dictionary.TryRemove(Owner, id, out value));
44-
Assert.Equal("value", value);
45-
Assert.False(dictionary.TryGetValue(Owner, id, out _));
46-
}
47-
48-
[Fact]
49-
public void CallbackOwnerIsPartOfTheKey()
50-
{
51-
var dictionary = new StripedCallbackDictionary<string>();
52-
var otherOwner = GrainId.Create("test", "other-owner");
53-
var id = new CorrelationId(42);
54-
55-
Assert.True(dictionary.TryAdd(Owner, id, "value"));
56-
Assert.False(dictionary.TryGetValue(otherOwner, id, out _));
57-
Assert.False(dictionary.TryRemove(otherOwner, id, out _));
58-
Assert.True(dictionary.TryGetValue(Owner, id, out var value));
42+
Assert.True(dictionary.TryRemove(id, out value));
5943
Assert.Equal("value", value);
44+
Assert.False(dictionary.TryGetValue(id, out _));
6045
}
6146

6247
[Fact]
@@ -66,7 +51,7 @@ public void EnumerationReturnsSnapshotValues()
6651
for (var i = 0; i < 32; i++)
6752
{
6853
var id = new CorrelationId(i);
69-
Assert.True(dictionary.TryAdd(Owner, id, i));
54+
Assert.True(dictionary.TryAdd(id, i));
7055
}
7156

7257
var values = new List<int>();
@@ -83,8 +68,8 @@ public void ConcurrentOperationsPreserveCountAndValues()
8368
Parallel.For(0, 10_000, i =>
8469
{
8570
var id = new CorrelationId(i);
86-
Assert.True(dictionary.TryAdd(Owner, id, i));
87-
Assert.True(dictionary.TryGetValue(Owner, id, out var value));
71+
Assert.True(dictionary.TryAdd(id, i));
72+
Assert.True(dictionary.TryGetValue(id, out var value));
8873
Assert.Equal(i, value);
8974
});
9075

@@ -93,7 +78,7 @@ public void ConcurrentOperationsPreserveCountAndValues()
9378

9479
Parallel.For(0, 10_000, i =>
9580
{
96-
Assert.True(dictionary.TryRemove(Owner, new CorrelationId(i), out var value));
81+
Assert.True(dictionary.TryRemove(new CorrelationId(i), out var value));
9782
Assert.Equal(i, value);
9883
});
9984

@@ -107,20 +92,20 @@ public void ConcurrentLookupAndRemovalRemainConsistent()
10792
var dictionary = new StripedCallbackDictionary<int>();
10893
for (var i = 0; i < count; i++)
10994
{
110-
Assert.True(dictionary.TryAdd(Owner, new CorrelationId(i), i));
95+
Assert.True(dictionary.TryAdd(new CorrelationId(i), i));
11196
}
11297

11398
Parallel.Invoke(
11499
() => Parallel.For(0, count, i =>
115100
{
116-
if (dictionary.TryGetValue(Owner, new CorrelationId(i), out var value))
101+
if (dictionary.TryGetValue(new CorrelationId(i), out var value))
117102
{
118103
Assert.Equal(i, value);
119104
}
120105
}),
121106
() => Parallel.For(0, count, i =>
122107
{
123-
Assert.True(dictionary.TryRemove(Owner, new CorrelationId(i), out var value));
108+
Assert.True(dictionary.TryRemove(new CorrelationId(i), out var value));
124109
Assert.Equal(i, value);
125110
}));
126111

@@ -133,12 +118,12 @@ public void SnapshotVisitorAllowsValuesToRemoveThemselves()
133118
var dictionary = new StripedCallbackDictionary<int>();
134119
for (var i = 0; i < 32; i++)
135120
{
136-
Assert.True(dictionary.TryAdd(Owner, new CorrelationId(i), i));
121+
Assert.True(dictionary.TryAdd(new CorrelationId(i), i));
137122
}
138123

139-
dictionary.ForEach((Dictionary: dictionary, Owner), static (value, state) =>
124+
dictionary.ForEach(dictionary, static (value, dictionary) =>
140125
{
141-
Assert.True(state.Dictionary.TryRemove(state.Owner, new CorrelationId(value), out var removed));
126+
Assert.True(dictionary.TryRemove(new CorrelationId(value), out var removed));
142127
Assert.Equal(value, removed);
143128
});
144129

0 commit comments

Comments
 (0)