Perf optimizations with function ID resolution (#9959) (#10447)

* Perf optimizations with function ID resolution

* Minor perf and code improvements to RpcFunctionInvocationDispatcher

Co-authored-by: Fabio Cavalcante <[email protected]>

Author: kshyju

src/WebJobs.Script.Grpc/Channel/GrpcWorkerChannel.cs

@@ -586,10 +586,11 @@ public void SetupFunctionInvocationBuffers(IEnumerable<FunctionMetadata> functio
             _functions = functions;
             foreach (FunctionMetadata metadata in functions)
             {
-                _workerChannelLogger.LogDebug("Setting up FunctionInvocationBuffer for function: '{functionName}' with functionId: '{functionId}'", metadata.Name, metadata.GetFunctionId());
-                _functionInputBuffers[metadata.GetFunctionId()] = new BufferBlock<ScriptInvocationContext>();
+                string functionId = metadata.GetFunctionId();
+                _workerChannelLogger.LogDebug("Setting up FunctionInvocationBuffer for function: '{functionName}' with functionId: '{functionId}'", metadata.Name, functionId);
+                _functionInputBuffers[functionId] = new BufferBlock<ScriptInvocationContext>();
             }
-            _state = _state | RpcWorkerChannelState.InvocationBuffersInitialized;
+            _state |= RpcWorkerChannelState.InvocationBuffersInitialized;
         }
 
         public void SendFunctionLoadRequests(ManagedDependencyOptions managedDependencyOptions, TimeSpan? functionTimeout)
@@ -742,7 +743,11 @@ internal FunctionEnvironmentReloadRequest GetFunctionEnvironmentReloadRequest(ID
         internal void SendFunctionLoadRequest(FunctionMetadata metadata, ManagedDependencyOptions managedDependencyOptions)
         {
             _functionLoadRequestResponseEvent = _metricsLogger.LatencyEvent(MetricEventNames.FunctionLoadRequestResponse);
-            _workerChannelLogger.LogDebug("Sending FunctionLoadRequest for function: '{functionName}' with functionId: '{functionId}'", metadata.Name, metadata.GetFunctionId());
+
+            if (_workerChannelLogger.IsEnabled(LogLevel.Debug))
+            {
+                _workerChannelLogger.LogDebug("Sending FunctionLoadRequest for function: '{functionName}' with functionId: '{functionId}'", metadata.Name, metadata.GetFunctionId());
+            }
 
             // send a load request for the registered function
             SendStreamingMessage(new StreamingMessage
@@ -838,20 +843,21 @@ internal async Task SendInvocationRequest(ScriptInvocationContext context)
         {
             try
             {
-                var invocationId = context.ExecutionContext.InvocationId.ToString();
+                string invocationId = context.ExecutionContext.InvocationId.ToString();
+                string functionId = context.FunctionMetadata.GetFunctionId();
 
                 // do not send an invocation request for functions that failed to load or could not be indexed by the worker
-                if (_functionLoadErrors.ContainsKey(context.FunctionMetadata.GetFunctionId()))
+                if (_functionLoadErrors.TryGetValue(functionId, out Exception exception))
                 {
                     _workerChannelLogger.LogDebug("Function {functionName} failed to load", context.FunctionMetadata.Name);
-                    context.ResultSource.TrySetException(_functionLoadErrors[context.FunctionMetadata.GetFunctionId()]);
+                    context.ResultSource.TrySetException(exception);
                     RemoveExecutingInvocation(invocationId);
                     return;
                 }
-                else if (_metadataRequestErrors.ContainsKey(context.FunctionMetadata.GetFunctionId()))
+                else if (_metadataRequestErrors.TryGetValue(functionId, out exception))
                 {
                     _workerChannelLogger.LogDebug("Worker failed to load metadata for {functionName}", context.FunctionMetadata.Name);
-                    context.ResultSource.TrySetException(_metadataRequestErrors[context.FunctionMetadata.GetFunctionId()]);
+                    context.ResultSource.TrySetException(exception);
                     RemoveExecutingInvocation(invocationId);
                     return;
                 }

src/WebJobs.Script/Extensions/FunctionMetadataExtensions.cs

@@ -2,6 +2,7 @@
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
 using System;
+using System.Collections.Generic;
 using System.Linq;
 using Microsoft.Azure.WebJobs.Script.Description;
 using Microsoft.Azure.WebJobs.Script.Extensions;
@@ -92,13 +93,16 @@ public static string GetFunctionGroup(this FunctionMetadata metadata)
 
         public static string GetFunctionId(this FunctionMetadata metadata)
         {
-            if (!metadata.Properties.TryGetValue(FunctionIdKey, out object idObj)
-                || !(idObj is string))
+            if (metadata.Properties.TryGetValue(FunctionIdKey, out object idObj)
+                && idObj is string id)
             {
-                metadata.Properties[FunctionIdKey] = Guid.NewGuid().ToString();
+                return id;
             }
 
-            return metadata.Properties[FunctionIdKey] as string;
+            id = Guid.NewGuid().ToString();
+            return metadata.Properties.TryAdd(FunctionIdKey, id)
+                ? id
+                : metadata.Properties[FunctionIdKey] as string;
         }
 
         internal static void SetFunctionId(this FunctionMetadata metadata, string functionId)

src/WebJobs.Script/Workers/Rpc/FunctionRegistration/RpcFunctionInvocationDispatcher.cs

@@ -18,7 +18,6 @@
 using Microsoft.Azure.WebJobs.Script.ManagedDependencies;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
-
 using FunctionMetadata = Microsoft.Azure.WebJobs.Script.Description.FunctionMetadata;
 
 namespace Microsoft.Azure.WebJobs.Script.Workers.Rpc
@@ -196,7 +195,7 @@ private void SetDispatcherStateToInitialized(IDictionary<string, TaskCompletionS
             // RanToCompletion indicates successful process startup
             if (State != FunctionInvocationDispatcherState.Initialized
                 && webhostLanguageWorkerChannel != null
-                && webhostLanguageWorkerChannel.Where(a => a.Value.Task.Status == TaskStatus.RanToCompletion).Any())
+                && webhostLanguageWorkerChannel.Any(a => a.Value.Task.Status == TaskStatus.RanToCompletion))
             {
                 SetFunctionDispatcherStateToInitializedAndLog();
             }
@@ -253,7 +252,7 @@ public async Task InitializeAsync(IEnumerable<FunctionMetadata> functions, Cance
                 return;
             }
 
-            _workerRuntime = _workerRuntime ?? Utility.GetWorkerRuntime(functions, _environment);
+            _workerRuntime ??= Utility.GetWorkerRuntime(functions, _environment);
 
             // In case of multi language runtime, _workerRuntime has no significance, thus skipping this check for multi language runtime environment
             if ((string.IsNullOrEmpty(_workerRuntime) || _workerRuntime.Equals(RpcWorkerConstants.DotNetLanguageWorkerName, StringComparison.InvariantCultureIgnoreCase)) && !_environment.IsMultiLanguageRuntimeEnvironment())
@@ -294,7 +293,7 @@ public async Task InitializeAsync(IEnumerable<FunctionMetadata> functions, Cance
                 return;
             }
 
-            _functions = functions ?? new List<FunctionMetadata>();
+            _functions = functions;
 
             if (_environment.IsMultiLanguageRuntimeEnvironment())
             {
@@ -351,6 +350,7 @@ public async Task InitializeAsync(IEnumerable<FunctionMetadata> functions, Cance
                     }
                 }
             }
+
             AddLogUserCategory(functions);
         }
 
@@ -402,10 +402,15 @@ public async Task InvokeAsync(ScriptInvocationContext invocationContext)
             // This could throw if no initialized workers are found. Shut down instance and retry.
             IEnumerable<IRpcWorkerChannel> workerChannels = await GetInitializedWorkerChannelsAsync(invocationContext.FunctionMetadata.Language ?? _workerRuntime);
             var rpcWorkerChannel = _functionDispatcherLoadBalancer.GetLanguageWorkerChannel(workerChannels);
-            if (rpcWorkerChannel.FunctionInputBuffers.TryGetValue(invocationContext.FunctionMetadata.GetFunctionId(), out BufferBlock<ScriptInvocationContext> bufferBlock))
+            string functionId = invocationContext.FunctionMetadata.GetFunctionId();
+            if (rpcWorkerChannel.FunctionInputBuffers.TryGetValue(functionId, out BufferBlock<ScriptInvocationContext> bufferBlock))
             {
-                _logger.LogTrace("Posting invocation id:{InvocationId} on workerId:{workerChannelId}", invocationContext.ExecutionContext.InvocationId, rpcWorkerChannel.Id);
-                rpcWorkerChannel.FunctionInputBuffers[invocationContext.FunctionMetadata.GetFunctionId()].Post(invocationContext);
+                if (_logger.IsEnabled(LogLevel.Trace))
+                {
+                    _logger.LogTrace("Posting invocation id:{InvocationId} on workerId:{workerChannelId}", invocationContext.ExecutionContext.InvocationId, rpcWorkerChannel.Id);
+                }
+
+                bufferBlock.Post(invocationContext);
             }
             else
             {
@@ -539,11 +544,12 @@ private async Task DisposeAndRestartWorkerChannel(string runtime, string workerI
             if (ShouldRestartWorkerChannel(runtime, isWebHostChannelDisposed, isJobHostChannelDisposed))
             {
                 // Set state to "WorkerProcessRestarting" if there are no other workers to handle work
-                if ((await GetInitializedWorkerChannelsAsync()).Count() == 0)
+                if (!(await GetInitializedWorkerChannelsAsync()).Any())
                 {
                     State = FunctionInvocationDispatcherState.WorkerProcessRestarting;
                     _logger.LogDebug("No initialized worker channels for runtime '{runtime}'. Delaying future invocations", runtime);
                 }
+
                 // Restart worker channel
                 _logger.LogDebug("Restarting worker channel for runtime: '{runtime}'", runtime);
                 await StartWorkerChannel(runtime);