perf(runtime): reduce invokable pooling overhead

Author: ReubenBond

src/Orleans.CodeGenerator/InvokableGenerator.cs

@@ -897,6 +897,7 @@ private List<InvokerFieldDescription> GetFieldDescriptions(
                 constructor.MethodKind == MethodKind.Constructor
                 && constructor.HasAttribute(LibraryTypes.GeneratedActivatorConstructorAttribute));
         if (method.MethodTypeParameters.Count == 0
+            && method.Method.Parameters.Length >= 2
             && method.CustomInitializerMethods.Count == 0
             && !requiresDependencyInjection
             && IsPoolableBaseType(baseClassType))

src/Orleans.Core/Messaging/ClientMessageCenter.cs

@@ -191,8 +191,8 @@ public void SendMessage(Message msg)
                     return;
                 }
 
-                connection.Send(msg);
                 LogSendingMessage(msg, connection.RemoteEndPoint);
+                connection.Send(msg);
             }
             else
             {
@@ -211,9 +211,8 @@ async Task SendAsync(ValueTask<Connection?> task, Message message)
                             return;
                         }
 
-                        connection.Send(message);
-
                         LogSendingMessage(message, connection.RemoteEndPoint);
+                        connection.Send(message);
                     }
                     catch (Exception exception)
                     {

src/Orleans.Core/Messaging/MessageFactory.cs

@@ -32,37 +32,22 @@ public MessageFactory(DeepCopier deepCopier, ILogger<MessageFactory> logger, Mes
 
         public Message CreateMessage(object? body, InvokeMethodOptions options)
         {
-            var oneWay = (options & InvokeMethodOptions.OneWay) != 0;
-            var messageBody = body;
-            var ownsBodyObject = false;
-            if (body is RequestBase)
-            {
-                messageBody = CopyBodyObject(body);
-                ownsBodyObject = true;
-            }
-            else if (body is IInvokable)
-            {
-                ownsBodyObject = true;
-            }
-
             var message = new Message
             {
-                Direction = oneWay ? Message.Directions.OneWay : Message.Directions.Request,
+                Direction = (options & InvokeMethodOptions.OneWay) != 0 ? Message.Directions.OneWay : Message.Directions.Request,
                 Id = GetNextCorrelationId(),
                 IsReadOnly = (options & InvokeMethodOptions.ReadOnly) != 0,
                 IsUnordered = (options & InvokeMethodOptions.Unordered) != 0,
                 IsAlwaysInterleave = (options & InvokeMethodOptions.AlwaysInterleave) != 0,
-                BodyObject = messageBody,
-                OwnsBodyObject = ownsBodyObject,
+                BodyObject = body,
+                OwnsBodyObject = body is IInvokable,
                 RequestContextData = RequestContextExtensions.Export(_deepCopier),
             };
 
             _messagingTrace.OnCreateMessage(message);
             return message;
         }
 
-        private object CopyBodyObject(object body) => _deepCopier.Copy(body)!;
-
         private CorrelationId GetNextCorrelationId()
         {
             var id = _seed ^ Interlocked.Increment(ref _nextId);

src/Orleans.Core/Runtime/CallbackData.cs

@@ -210,6 +210,20 @@ public void OnHostShutdown()
             this.context.Complete(Response.FromException(exception));
         }
 
+        public void OnSendFailure(Exception exception)
+        {
+            if (!TryComplete())
+            {
+                return;
+            }
+
+            this.stopwatch.Stop();
+            this.shared.Unregister(this.Message);
+            DisposeCancellationRegistration();
+            _applicationRequestInstruments.OnAppRequestsEnd((long)this.stopwatch.Elapsed.TotalMilliseconds);
+            this.context.Complete(Response.FromException(exception));
+        }
+
         public void DoCallback(Message response)
         {
             if (!TryComplete())

src/Orleans.Core/Runtime/GrainReferenceRuntime.cs

@@ -6,6 +6,7 @@
 using Orleans.CodeGeneration;
 using Orleans.GrainReferences;
 using Orleans.Metadata;
+using Orleans.Serialization;
 using Orleans.Serialization.Invocation;
 
 namespace Orleans.Runtime
@@ -17,20 +18,23 @@ internal class GrainReferenceRuntime : IGrainReferenceRuntime
         private readonly IGrainCancellationTokenRuntime cancellationTokenRuntime;
         private readonly IOutgoingGrainCallFilter[] filters;
         private readonly Action<GrainReference, IResponseCompletionSource, IInvokable, InvokeMethodOptions> sendRequest;
+        private readonly DeepCopier requestCopier;
 
         public GrainReferenceRuntime(
             IRuntimeClient runtimeClient,
             IGrainCancellationTokenRuntime cancellationTokenRuntime,
             IEnumerable<IOutgoingGrainCallFilter> outgoingCallFilters,
             GrainReferenceActivator referenceActivator,
-            GrainInterfaceTypeResolver interfaceTypeResolver)
+            GrainInterfaceTypeResolver interfaceTypeResolver,
+            DeepCopier requestCopier)
         {
             this.RuntimeClient = runtimeClient;
             this.cancellationTokenRuntime = cancellationTokenRuntime;
             this.referenceActivator = referenceActivator;
             this.interfaceTypeResolver = interfaceTypeResolver;
             this.filters = outgoingCallFilters.ToArray();
-            this.sendRequest = (GrainReference reference, IResponseCompletionSource callback, IInvokable body, InvokeMethodOptions options) => RuntimeClient.SendRequest(reference, body, callback, options);
+            this.requestCopier = requestCopier;
+            this.sendRequest = SendFilteredRequest;
         }
 
         public IRuntimeClient RuntimeClient { get; private set; }
@@ -64,25 +68,26 @@ public ValueTask InvokeMethodAsync(GrainReference reference, IInvokable request,
         public void InvokeMethod(GrainReference reference, IInvokable request, InvokeMethodOptions options)
         {
             Debug.Assert((options & InvokeMethodOptions.OneWay) != 0);
-            var disposeRequest = true;
+            var requestTransferred = false;
 
             try
             {
                 // TODO: Remove expensive interface type check
                 if (filters.Length == 0 && request is not IOutgoingGrainCallFilter)
                 {
                     SetGrainCancellationTokensTarget(reference, request);
+                    requestTransferred = true;
                     this.RuntimeClient.SendRequest(reference, request, context: null, options);
                 }
                 else
                 {
                     InvokeMethodWithFiltersAsync(reference, request, options).AsTask().Ignore();
-                    disposeRequest = false;
+                    requestTransferred = true;
                 }
             }
             finally
             {
-                if (disposeRequest)
+                if (!requestTransferred)
                 {
                     DisposeRequest(request);
                 }
@@ -121,60 +126,48 @@ private async ValueTask InvokeMethodWithFiltersAsync(GrainReference reference, I
         private ValueTask<TResult?> InvokeMethodAsyncCore<TResult>(GrainReference reference, IInvokable request, InvokeMethodOptions options)
         {
             ResponseCompletionSource<TResult>? responseCompletionSource = null;
+            var requestTransferred = false;
             try
             {
                 SetGrainCancellationTokensTarget(reference, request);
                 responseCompletionSource = ResponseCompletionSourcePool.Get<TResult>();
+                requestTransferred = true;
                 this.RuntimeClient.SendRequest(reference, request, responseCompletionSource, options);
-                return CompleteInvokeAsync(responseCompletionSource, request, options);
+                return responseCompletionSource.AsValueTask();
             }
             catch
             {
                 responseCompletionSource?.Reset();
-                DisposeRequest(request);
-                throw;
-            }
-        }
+                if (!requestTransferred)
+                {
+                    DisposeRequest(request);
+                }
 
-        private static async ValueTask<TResult?> CompleteInvokeAsync<TResult>(ResponseCompletionSource<TResult> responseCompletionSource, IInvokable request, InvokeMethodOptions options)
-        {
-            try
-            {
-                return await responseCompletionSource.AsValueTask();
-            }
-            finally
-            {
-                DisposeRequest(request);
+                throw;
             }
         }
 
         private ValueTask InvokeMethodAsyncCore(GrainReference reference, IInvokable request, InvokeMethodOptions options)
         {
             ResponseCompletionSource? responseCompletionSource = null;
+            var requestTransferred = false;
             try
             {
                 SetGrainCancellationTokensTarget(reference, request);
                 responseCompletionSource = ResponseCompletionSourcePool.Get();
+                requestTransferred = true;
                 this.RuntimeClient.SendRequest(reference, request, responseCompletionSource, options);
-                return CompleteInvokeAsync(responseCompletionSource, request, options);
+                return responseCompletionSource.AsVoidValueTask();
             }
             catch
             {
                 responseCompletionSource?.Reset();
-                DisposeRequest(request);
-                throw;
-            }
-        }
+                if (!requestTransferred)
+                {
+                    DisposeRequest(request);
+                }
 
-        private static async ValueTask CompleteInvokeAsync(ResponseCompletionSource responseCompletionSource, IInvokable request, InvokeMethodOptions options)
-        {
-            try
-            {
-                await responseCompletionSource.AsVoidValueTask();
-            }
-            finally
-            {
-                DisposeRequest(request);
+                throw;
             }
         }
 
@@ -186,6 +179,18 @@ private static void DisposeRequest(IInvokable request)
             }
         }
 
+        private void SendFilteredRequest(
+            GrainReference reference,
+            IResponseCompletionSource callback,
+            IInvokable request,
+            InvokeMethodOptions options)
+        {
+            var messageRequest = request is RequestBase
+                ? (IInvokable)this.requestCopier.Copy(request)!
+                : request;
+            RuntimeClient.SendRequest(reference, messageRequest, callback, options);
+        }
+
         public object Cast(IAddressable grain, Type grainInterface)
         {
             var grainId = grain.GetGrainId();

src/Orleans.Core/Runtime/OutgoingCallInvoker.cs

@@ -113,8 +113,22 @@ public async Task Invoke()
                     // Finally call the root-level invoker.
                     stage++;
                     var responseCompletionSource = ResponseCompletionSourcePool.Get();
-                    this.sendRequest(this.grainReference, responseCompletionSource, this.request, this.options);
-                    this.Response = await responseCompletionSource.AsValueTask().ConfigureAwait(false);
+                    var requestSent = false;
+                    try
+                    {
+                        this.sendRequest(this.grainReference, responseCompletionSource, this.request, this.options);
+                        requestSent = true;
+                        this.Response = await responseCompletionSource.AsValueTask().ConfigureAwait(false);
+                    }
+                    catch
+                    {
+                        if (!requestSent)
+                        {
+                            responseCompletionSource.Reset();
+                        }
+
+                        throw;
+                    }
 
                     return;
                 }