Activation repartitioner: use null for no-op message observer to avoid interface call (#9056)

ReubenBond · web-flow · commit a85e3372bb81 · 2024-07-09T19:42:20.000-07:00
diff --git a/src/Orleans.Core/Networking/Connection.cs b/src/Orleans.Core/Networking/Connection.cs
@@ -353,7 +353,7 @@ private async Task ProcessOutgoing()
 
             Exception error = default;
             var serializer = this.shared.ServiceProvider.GetRequiredService<MessageSerializer>();
-            var messageStatisticsSink = this.shared.MessageStatisticsSink;
+            var messageObserver = this.shared.MessageStatisticsSink.GetMessageObserver();
             try
             {
                 var output = this._transport.Output;
@@ -375,7 +375,7 @@ private async Task ProcessOutgoing()
                             inflight.Add(message);
                             var (headerLength, bodyLength) = serializer.Write(output, message);
                             RecordMessageSend(message, headerLength + bodyLength, headerLength);
-                            messageStatisticsSink.RecordMessage(message);
+                            messageObserver?.Invoke(message);
                             message = null;
                         }
                     }
diff --git a/src/Orleans.Core/Placement/Repartitioning/IMessageStatisticsSink.cs b/src/Orleans.Core/Placement/Repartitioning/IMessageStatisticsSink.cs
@@ -1,13 +1,15 @@
+#nullable enable
+using System;
 using Orleans.Runtime;
 
 namespace Orleans.Placement.Repartitioning;
 
 internal interface IMessageStatisticsSink
 {
-    void RecordMessage(Message message);
+    Action<Message>? GetMessageObserver();
 }
 
 internal sealed class NoOpMessageStatisticsSink : IMessageStatisticsSink
 {
-    public void RecordMessage(Message message) { }
+    public Action<Message>? GetMessageObserver() => null;
 }
diff --git a/src/Orleans.Runtime/Hosting/DefaultSiloServices.cs b/src/Orleans.Runtime/Hosting/DefaultSiloServices.cs
@@ -139,7 +139,7 @@ internal static void AddDefaultServices(ISiloBuilder builder)
             services.TryAddSingleton<MessageCenter>();
             services.TryAddFromExisting<IMessageCenter, MessageCenter>();
             services.TryAddSingleton(FactoryUtility.Create<MessageCenter, Gateway>);
-            services.TryAddSingleton<IConnectedClientCollection>(sp => (IConnectedClientCollection)sp.GetRequiredService<MessageCenter>().Gateway ?? new EmptyConnectedClientCollection());
+            services.TryAddSingleton<IConnectedClientCollection>(sp => sp.GetRequiredService<MessageCenter>().Gateway as IConnectedClientCollection ?? new EmptyConnectedClientCollection());
             services.TryAddSingleton<InternalGrainRuntime>();
             services.TryAddSingleton<InsideRuntimeClient>();
             services.TryAddFromExisting<IRuntimeClient, InsideRuntimeClient>();
diff --git a/src/Orleans.Runtime/Messaging/MessageCenter.cs b/src/Orleans.Runtime/Messaging/MessageCenter.cs
@@ -1,3 +1,4 @@
+#nullable enable
 using System;
 using System.Collections.Generic;
 using System.Diagnostics;
@@ -22,12 +23,12 @@ internal class MessageCenter : IMessageCenter, IAsyncDisposable
         private readonly SiloMessagingOptions messagingOptions;
         private readonly PlacementService placementService;
         private readonly GrainLocator _grainLocator;
-        private readonly IMessageStatisticsSink _messageStatisticsSink;
+        private readonly Action<Message>? _messageObserver;
         private readonly ILogger log;
         private readonly Catalog catalog;
         private bool stopped;
-        private HostedClient hostedClient;
-        private Action<Message> sniffIncomingMessageHandler;
+        private HostedClient? hostedClient;
+        private Action<Message>? sniffIncomingMessageHandler;
 
         public MessageCenter(
             ILocalSiloDetails siloDetails,
@@ -50,7 +51,7 @@ public MessageCenter(
             this.messagingTrace = messagingTrace;
             this.placementService = placementService;
             _grainLocator = grainLocator;
-            _messageStatisticsSink = messageStatisticsSink;
+            _messageObserver = messageStatisticsSink.GetMessageObserver();
             this.log = logger;
             this.messageFactory = messageFactory;
             this._siloAddress = siloDetails.SiloAddress;
@@ -61,19 +62,19 @@ public MessageCenter(
             }
         }
 
-        public Gateway Gateway { get; }
+        public Gateway? Gateway { get; }
 
         internal bool IsBlockingApplicationMessages { get; private set; }
 
-        public void SetHostedClient(HostedClient client) => this.hostedClient = client;
+        public void SetHostedClient(HostedClient? client) => this.hostedClient = client;
 
         public bool TryDeliverToProxy(Message msg)
         {
             if (!msg.TargetGrain.IsClient()) return false;
             if (this.Gateway is Gateway gateway && gateway.TryDeliverToProxy(msg)
                 || this.hostedClient is HostedClient client && client.TryDispatchToClient(msg))
             {
-                _messageStatisticsSink.RecordMessage(msg);
+                _messageObserver?.Invoke(msg);
                 return true;
             }
 
@@ -125,7 +126,7 @@ public async Task StopAcceptingClientMessages()
             }
         }
 
-        public Action<Message> SniffIncomingMessage
+        public Action<Message>? SniffIncomingMessage
         {
             set
             {
@@ -251,8 +252,8 @@ static async Task SendAsync(MessageCenter messageCenter, ValueTask<Connection> c
         public void RejectMessage(
             Message message,
             Message.RejectionTypes rejectionType,
-            Exception exc,
-            string rejectInfo = null)
+            Exception? exc,
+            string? rejectInfo = null)
         {
             if (message.Direction == Message.Directions.Request
                 || (message.Direction == Message.Directions.OneWay && message.HasCacheInvalidationHeader))
@@ -271,20 +272,20 @@ public void RejectMessage(
 
         internal void ProcessRequestsToInvalidActivation(
             List<Message> messages,
-            GrainAddress oldAddress,
-            SiloAddress forwardingAddress,
-            string failedOperation = null,
-            Exception exc = null,
+            GrainAddress? oldAddress,
+            SiloAddress? forwardingAddress,
+            string? failedOperation = null,
+            Exception? exc = null,
             bool rejectMessages = false)
         {
             if (rejectMessages)
             {
-                GrainAddress validAddress = forwardingAddress switch
+                GrainAddress? validAddress = forwardingAddress switch
                 {
                     null => null,
                     _ => new()
                     {
-                        GrainId = oldAddress.GrainId,
+                        GrainId = oldAddress?.GrainId ?? default,
                         SiloAddress = forwardingAddress,
                     }
                 };
@@ -304,12 +305,12 @@ internal void ProcessRequestsToInvalidActivation(
             else
             {
                 this.messagingTrace.OnDispatcherForwardingMultiple(messages.Count, oldAddress, forwardingAddress, failedOperation, exc);
-                GrainAddress destination = forwardingAddress switch
+                GrainAddress? destination = forwardingAddress switch
                 {
                     null => null,
                     _ => new()
                     {
-                        GrainId = oldAddress.GrainId,
+                        GrainId = oldAddress?.GrainId ?? default,
                         SiloAddress = forwardingAddress,
                     }
                 };
@@ -323,10 +324,10 @@ internal void ProcessRequestsToInvalidActivation(
 
         private void ProcessRequestToInvalidActivation(
             Message message,
-            GrainAddress oldAddress,
-            SiloAddress forwardingAddress,
+            GrainAddress? oldAddress,
+            SiloAddress? forwardingAddress,
             string failedOperation,
-            Exception exc = null,
+            Exception? exc = null,
             bool rejectMessages = false)
         {
             Debug.Assert(!message.IsLocalOnly);
@@ -344,20 +345,20 @@ private void ProcessRequestToInvalidActivation(
             }
             else
             {
-                GrainAddress destination = forwardingAddress switch
+                GrainAddress? destination = forwardingAddress switch
                 {
                     null => null,
                     _ => new()
                     {
-                        GrainId = oldAddress.GrainId,
+                        GrainId = oldAddress?.GrainId ?? default,
                         SiloAddress = forwardingAddress,
                     }
                 };
                 this.TryForwardRequest(message, oldAddress, destination, failedOperation, exc);
             }
         }
 
-        private void TryForwardRequest(Message message, GrainAddress oldAddress, GrainAddress destination, string failedOperation = null, Exception exc = null)
+        private void TryForwardRequest(Message message, GrainAddress? oldAddress, GrainAddress? destination, string? failedOperation = null, Exception? exc = null)
         {
             Debug.Assert(!message.IsLocalOnly);
 
@@ -415,7 +416,7 @@ internal void RerouteMessage(Message message)
             ResendMessageImpl(message);
         }
 
-        private bool TryForwardMessage(Message message, SiloAddress forwardingAddress)
+        private bool TryForwardMessage(Message message, SiloAddress? forwardingAddress)
         {
             if (!MayForward(message, this.messagingOptions)) return false;
 
@@ -425,7 +426,7 @@ private bool TryForwardMessage(Message message, SiloAddress forwardingAddress)
             return true;
         }
 
-        private void ResendMessageImpl(Message message, SiloAddress forwardingAddress = null)
+        private void ResendMessageImpl(Message message, SiloAddress? forwardingAddress = null)
         {
             if (log.IsEnabled(LogLevel.Debug)) log.LogDebug("Resend {Message}", message);
 
@@ -543,7 +544,7 @@ public void ReceiveMessage(Message msg)
                     }
 
                     targetActivation.ReceiveMessage(msg);
-                    _messageStatisticsSink.RecordMessage(msg);
+                    _messageObserver?.Invoke(msg);
                 }
             }
             catch (Exception ex)
diff --git a/src/Orleans.Runtime/Placement/Repartitioning/ActivationRepartitioner.MessageSink.cs b/src/Orleans.Runtime/Placement/Repartitioning/ActivationRepartitioner.MessageSink.cs
@@ -1,6 +1,7 @@
 #nullable enable
 using System;
 using System.Diagnostics;
+using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
@@ -109,7 +110,9 @@ private async Task ProcessPendingEdges(CancellationToken cancellationToken)
         }
     }
 
-    public void RecordMessage(Message message)
+    public Action<Message>? GetMessageObserver() => RecordMessage;
+
+    private void RecordMessage(Message message)
     {
         if (!_enableMessageSampling || message.IsSystemMessage)
         {
diff --git a/src/Orleans.Runtime/Placement/Repartitioning/RepartitionerMessageFilter.cs b/src/Orleans.Runtime/Placement/Repartitioning/RepartitionerMessageFilter.cs
@@ -1,5 +1,4 @@
 #nullable enable
-using Orleans;
 using Orleans.Metadata;
 using System;
 using System.Collections.Concurrent;

Original file line number	Diff line number	Diff line change
`@@ -353,7 +353,7 @@ private async Task ProcessOutgoing()`
`353`	`353`
`354`	`354`	`Exception error = default;`
`355`	`355`	`var serializer = this.shared.ServiceProvider.GetRequiredService<MessageSerializer>();`
`356`		`- var messageStatisticsSink = this.shared.MessageStatisticsSink;`
	`356`	`+ var messageObserver = this.shared.MessageStatisticsSink.GetMessageObserver();`
`357`	`357`	`try`
`358`	`358`	`{`
`359`	`359`	`var output = this._transport.Output;`
`@@ -375,7 +375,7 @@ private async Task ProcessOutgoing()`
`375`	`375`	`inflight.Add(message);`
`376`	`376`	`var (headerLength, bodyLength) = serializer.Write(output, message);`
`377`	`377`	`RecordMessageSend(message, headerLength + bodyLength, headerLength);`
`378`		`- messageStatisticsSink.RecordMessage(message);`
	`378`	`+ messageObserver?.Invoke(message);`
`379`	`379`	`message = null;`
`380`	`380`	`}`
`381`	`381`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,13 +1,15 @@`
	`1`	`+#nullable enable`
	`2`	`+using System;`
`1`	`3`	`using Orleans.Runtime;`
`2`	`4`
`3`	`5`	`namespace Orleans.Placement.Repartitioning;`
`4`	`6`
`5`	`7`	`internal interface IMessageStatisticsSink`
`6`	`8`	`{`
`7`		`- void RecordMessage(Message message);`
	`9`	`+ Action<Message>? GetMessageObserver();`
`8`	`10`	`}`
`9`	`11`
`10`	`12`	`internal sealed class NoOpMessageStatisticsSink : IMessageStatisticsSink`
`11`	`13`	`{`
`12`		`- public void RecordMessage(Message message) { }`
	`14`	`+ public Action<Message>? GetMessageObserver() => null;`
`13`	`15`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,3 +1,4 @@`
	`1`	`+#nullable enable`
`1`	`2`	`using System;`
`2`	`3`	`using System.Collections.Generic;`
`3`	`4`	`using System.Diagnostics;`
`@@ -22,12 +23,12 @@ internal class MessageCenter : IMessageCenter, IAsyncDisposable`
`22`	`23`	`private readonly SiloMessagingOptions messagingOptions;`
`23`	`24`	`private readonly PlacementService placementService;`
`24`	`25`	`private readonly GrainLocator _grainLocator;`
`25`		`- private readonly IMessageStatisticsSink _messageStatisticsSink;`
	`26`	`+ private readonly Action<Message>? _messageObserver;`
`26`	`27`	`private readonly ILogger log;`
`27`	`28`	`private readonly Catalog catalog;`
`28`	`29`	`private bool stopped;`
`29`		`- private HostedClient hostedClient;`
`30`		`- private Action<Message> sniffIncomingMessageHandler;`
	`30`	`+ private HostedClient? hostedClient;`
	`31`	`+ private Action<Message>? sniffIncomingMessageHandler;`
`31`	`32`
`32`	`33`	`public MessageCenter(`
`33`	`34`	`ILocalSiloDetails siloDetails,`
`@@ -50,7 +51,7 @@ public MessageCenter(`
`50`	`51`	`this.messagingTrace = messagingTrace;`
`51`	`52`	`this.placementService = placementService;`
`52`	`53`	`_grainLocator = grainLocator;`
`53`		`- _messageStatisticsSink = messageStatisticsSink;`
	`54`	`+ _messageObserver = messageStatisticsSink.GetMessageObserver();`
`54`	`55`	`this.log = logger;`
`55`	`56`	`this.messageFactory = messageFactory;`
`56`	`57`	`this._siloAddress = siloDetails.SiloAddress;`
`@@ -61,19 +62,19 @@ public MessageCenter(`
`61`	`62`	`}`
`62`	`63`	`}`
`63`	`64`
`64`		`- public Gateway Gateway { get; }`
	`65`	`+ public Gateway? Gateway { get; }`
`65`	`66`
`66`	`67`	`internal bool IsBlockingApplicationMessages { get; private set; }`
`67`	`68`
`68`		`- public void SetHostedClient(HostedClient client) => this.hostedClient = client;`
	`69`	`+ public void SetHostedClient(HostedClient? client) => this.hostedClient = client;`
`69`	`70`
`70`	`71`	`public bool TryDeliverToProxy(Message msg)`
`71`	`72`	`{`
`72`	`73`	`if (!msg.TargetGrain.IsClient()) return false;`
`73`	`74`	`if (this.Gateway is Gateway gateway && gateway.TryDeliverToProxy(msg)`
`74`	`75`	`\|\| this.hostedClient is HostedClient client && client.TryDispatchToClient(msg))`
`75`	`76`	`{`
`76`		`- _messageStatisticsSink.RecordMessage(msg);`
	`77`	`+ _messageObserver?.Invoke(msg);`
`77`	`78`	`return true;`
`78`	`79`	`}`
`79`	`80`
`@@ -125,7 +126,7 @@ public async Task StopAcceptingClientMessages()`
`125`	`126`	`}`
`126`	`127`	`}`
`127`	`128`
`128`		`- public Action<Message> SniffIncomingMessage`
	`129`	`+ public Action<Message>? SniffIncomingMessage`
`129`	`130`	`{`
`130`	`131`	`set`
`131`	`132`	`{`
`@@ -251,8 +252,8 @@ static async Task SendAsync(MessageCenter messageCenter, ValueTask<Connection> c`
`251`	`252`	`public void RejectMessage(`
`252`	`253`	`Message message,`
`253`	`254`	`Message.RejectionTypes rejectionType,`
`254`		`- Exception exc,`
`255`		`- string rejectInfo = null)`
	`255`	`+ Exception? exc,`
	`256`	`+ string? rejectInfo = null)`
`256`	`257`	`{`
`257`	`258`	`if (message.Direction == Message.Directions.Request`
`258`	`259`	`\|\| (message.Direction == Message.Directions.OneWay && message.HasCacheInvalidationHeader))`
`@@ -271,20 +272,20 @@ public void RejectMessage(`
`271`	`272`
`272`	`273`	`internal void ProcessRequestsToInvalidActivation(`
`273`	`274`	`List<Message> messages,`
`274`		`- GrainAddress oldAddress,`
`275`		`- SiloAddress forwardingAddress,`
`276`		`- string failedOperation = null,`
`277`		`- Exception exc = null,`
	`275`	`+ GrainAddress? oldAddress,`
	`276`	`+ SiloAddress? forwardingAddress,`
	`277`	`+ string? failedOperation = null,`
	`278`	`+ Exception? exc = null,`
`278`	`279`	`bool rejectMessages = false)`
`279`	`280`	`{`
`280`	`281`	`if (rejectMessages)`
`281`	`282`	`{`
`282`		`- GrainAddress validAddress = forwardingAddress switch`
	`283`	`+ GrainAddress? validAddress = forwardingAddress switch`
`283`	`284`	`{`
`284`	`285`	`null => null,`
`285`	`286`	`_ => new()`
`286`	`287`	`{`
`287`		`- GrainId = oldAddress.GrainId,`
	`288`	`+ GrainId = oldAddress?.GrainId ?? default,`
`288`	`289`	`SiloAddress = forwardingAddress,`
`289`	`290`	`}`
`290`	`291`	`};`
`@@ -304,12 +305,12 @@ internal void ProcessRequestsToInvalidActivation(`
`304`	`305`	`else`
`305`	`306`	`{`
`306`	`307`	`this.messagingTrace.OnDispatcherForwardingMultiple(messages.Count, oldAddress, forwardingAddress, failedOperation, exc);`
`307`		`- GrainAddress destination = forwardingAddress switch`
	`308`	`+ GrainAddress? destination = forwardingAddress switch`
`308`	`309`	`{`
`309`	`310`	`null => null,`
`310`	`311`	`_ => new()`
`311`	`312`	`{`
`312`		`- GrainId = oldAddress.GrainId,`
	`313`	`+ GrainId = oldAddress?.GrainId ?? default,`
`313`	`314`	`SiloAddress = forwardingAddress,`
`314`	`315`	`}`
`315`	`316`	`};`
`@@ -323,10 +324,10 @@ internal void ProcessRequestsToInvalidActivation(`
`323`	`324`
`324`	`325`	`private void ProcessRequestToInvalidActivation(`
`325`	`326`	`Message message,`
`326`		`- GrainAddress oldAddress,`
`327`		`- SiloAddress forwardingAddress,`
	`327`	`+ GrainAddress? oldAddress,`
	`328`	`+ SiloAddress? forwardingAddress,`
`328`	`329`	`string failedOperation,`
`329`		`- Exception exc = null,`
	`330`	`+ Exception? exc = null,`
`330`	`331`	`bool rejectMessages = false)`
`331`	`332`	`{`
`332`	`333`	`Debug.Assert(!message.IsLocalOnly);`
`@@ -344,20 +345,20 @@ private void ProcessRequestToInvalidActivation(`
`344`	`345`	`}`
`345`	`346`	`else`
`346`	`347`	`{`
`347`		`- GrainAddress destination = forwardingAddress switch`
	`348`	`+ GrainAddress? destination = forwardingAddress switch`
`348`	`349`	`{`
`349`	`350`	`null => null,`
`350`	`351`	`_ => new()`
`351`	`352`	`{`
`352`		`- GrainId = oldAddress.GrainId,`
	`353`	`+ GrainId = oldAddress?.GrainId ?? default,`
`353`	`354`	`SiloAddress = forwardingAddress,`
`354`	`355`	`}`
`355`	`356`	`};`
`356`	`357`	`this.TryForwardRequest(message, oldAddress, destination, failedOperation, exc);`
`357`	`358`	`}`
`358`	`359`	`}`
`359`	`360`
`360`		`- private void TryForwardRequest(Message message, GrainAddress oldAddress, GrainAddress destination, string failedOperation = null, Exception exc = null)`
	`361`	`+ private void TryForwardRequest(Message message, GrainAddress? oldAddress, GrainAddress? destination, string? failedOperation = null, Exception? exc = null)`
`361`	`362`	`{`
`362`	`363`	`Debug.Assert(!message.IsLocalOnly);`
`363`	`364`
`@@ -415,7 +416,7 @@ internal void RerouteMessage(Message message)`
`415`	`416`	`ResendMessageImpl(message);`
`416`	`417`	`}`
`417`	`418`
`418`		`- private bool TryForwardMessage(Message message, SiloAddress forwardingAddress)`
	`419`	`+ private bool TryForwardMessage(Message message, SiloAddress? forwardingAddress)`
`419`	`420`	`{`
`420`	`421`	`if (!MayForward(message, this.messagingOptions)) return false;`
`421`	`422`
`@@ -425,7 +426,7 @@ private bool TryForwardMessage(Message message, SiloAddress forwardingAddress)`
`425`	`426`	`return true;`
`426`	`427`	`}`
`427`	`428`
`428`		`- private void ResendMessageImpl(Message message, SiloAddress forwardingAddress = null)`
	`429`	`+ private void ResendMessageImpl(Message message, SiloAddress? forwardingAddress = null)`
`429`	`430`	`{`
`430`	`431`	`if (log.IsEnabled(LogLevel.Debug)) log.LogDebug("Resend {Message}", message);`
`431`	`432`
`@@ -543,7 +544,7 @@ public void ReceiveMessage(Message msg)`
`543`	`544`	`}`
`544`	`545`
`545`	`546`	`targetActivation.ReceiveMessage(msg);`
`546`		`- _messageStatisticsSink.RecordMessage(msg);`
	`547`	`+ _messageObserver?.Invoke(msg);`
`547`	`548`	`}`
`548`	`549`	`}`
`549`	`550`	`catch (Exception ex)`
Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,7 @@`
`1`	`1`	`#nullable enable`
`2`	`2`	`using System;`
`3`	`3`	`using System.Diagnostics;`
	`4`	`+using System.Runtime.CompilerServices;`
`4`	`5`	`using System.Threading;`
`5`	`6`	`using System.Threading.Tasks;`
`6`	`7`	`using Microsoft.Extensions.Logging;`
`@@ -109,7 +110,9 @@ private async Task ProcessPendingEdges(CancellationToken cancellationToken)`
`109`	`110`	`}`
`110`	`111`	`}`
`111`	`112`
`112`		`- public void RecordMessage(Message message)`
	`113`	`+ public Action<Message>? GetMessageObserver() => RecordMessage;`
	`114`	`+`
	`115`	`+ private void RecordMessage(Message message)`
`113`	`116`	`{`
`114`	`117`	`if (!_enableMessageSampling \|\| message.IsSystemMessage)`
`115`	`118`	`{`
Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,4 @@`
`1`	`1`	`#nullable enable`
`2`		`-using Orleans;`
`3`	`2`	`using Orleans.Metadata;`
`4`	`3`	`using System;`
`5`	`4`	`using System.Collections.Concurrent;`