Avoiding initialization of multiple processes when a WebHost channel is avaialable and worker indexing is enabled (#8810)

github-actions[bot] · fabiocav · web-flow · commit a0fe5b7d91c3 · 2022-09-30T04:08:17.000-07:00
Co-authored-by: Fabio Cavalcante &lt;facaval@microsoft.com&gt;
diff --git a/src/WebJobs.Script.Grpc/Channel/GrpcWorkerChannel.cs b/src/WebJobs.Script.Grpc/Channel/GrpcWorkerChannel.cs
@@ -367,6 +367,7 @@ internal FunctionLoadRequestCollection GetFunctionLoadRequestCollection(IEnumera
 
         public Task SendFunctionEnvironmentReloadRequest()
         {
+            _functionsIndexingTask = new TaskCompletionSource<List<RawFunctionMetadata>>(TaskCreationOptions.RunContinuationsAsynchronously);
             _workerChannelLogger.LogDebug("Sending FunctionEnvironmentReloadRequest to WorkerProcess with Pid: '{0}'", _rpcWorkerProcess.Id);
             IDisposable latencyEvent = _metricsLogger.LatencyEvent(MetricEventNames.SpecializationEnvironmentReloadRequestResponse);
 
diff --git a/src/WebJobs.Script/Workers/Rpc/FunctionRegistration/RpcFunctionInvocationDispatcher.cs b/src/WebJobs.Script/Workers/Rpc/FunctionRegistration/RpcFunctionInvocationDispatcher.cs
@@ -312,7 +312,7 @@ public async Task InitializeAsync(IEnumerable<FunctionMetadata> functions, Cance
                 Dictionary<string, TaskCompletionSource<IRpcWorkerChannel>> webhostLanguageWorkerChannels = _webHostLanguageWorkerChannelManager.GetChannels(_workerRuntime);
                 if (webhostLanguageWorkerChannels != null)
                 {
-                    int countOfReadyChannels = 0;
+                    int workerProcessCount = 0;
                     foreach (string workerId in webhostLanguageWorkerChannels.Keys.ToList())
                     {
                         if (webhostLanguageWorkerChannels.TryGetValue(workerId, out TaskCompletionSource<IRpcWorkerChannel> initializedLanguageWorkerChannelTask))
@@ -327,8 +327,9 @@ public async Task InitializeAsync(IEnumerable<FunctionMetadata> functions, Cance
                                 {
                                     initializedLanguageWorkerChannel.SetupFunctionInvocationBuffers(_functions);
                                     initializedLanguageWorkerChannel.SendFunctionLoadRequests(_managedDependencyOptions.Value, _scriptOptions.FunctionTimeout);
-                                    ++countOfReadyChannels;
                                 }
+
+                                ++workerProcessCount;
                             }
                             catch (Exception ex)
                             {
@@ -337,7 +338,7 @@ public async Task InitializeAsync(IEnumerable<FunctionMetadata> functions, Cance
                             }
                         }
                     }
-                    StartWorkerProcesses(countOfReadyChannels, InitializeWebhostLanguageWorkerChannel, true, _webHostLanguageWorkerChannelManager.GetChannels(_workerRuntime));
+                    StartWorkerProcesses(workerProcessCount, InitializeWebhostLanguageWorkerChannel, true, _webHostLanguageWorkerChannelManager.GetChannels(_workerRuntime));
                 }
                 else
                 {

Original file line number	Diff line number	Diff line change
`@@ -367,6 +367,7 @@ internal FunctionLoadRequestCollection GetFunctionLoadRequestCollection(IEnumera`
`367`	`367`
`368`	`368`	`public Task SendFunctionEnvironmentReloadRequest()`
`369`	`369`	`{`
	`370`	`+ _functionsIndexingTask = new TaskCompletionSource<List<RawFunctionMetadata>>(TaskCreationOptions.RunContinuationsAsynchronously);`
`370`	`371`	`_workerChannelLogger.LogDebug("Sending FunctionEnvironmentReloadRequest to WorkerProcess with Pid: '{0}'", _rpcWorkerProcess.Id);`
`371`	`372`	`IDisposable latencyEvent = _metricsLogger.LatencyEvent(MetricEventNames.SpecializationEnvironmentReloadRequestResponse);`
`372`	`373`
Original file line number	Diff line number	Diff line change
`@@ -312,7 +312,7 @@ public async Task InitializeAsync(IEnumerable<FunctionMetadata> functions, Cance`
`312`	`312`	`Dictionary<string, TaskCompletionSource<IRpcWorkerChannel>> webhostLanguageWorkerChannels = _webHostLanguageWorkerChannelManager.GetChannels(_workerRuntime);`
`313`	`313`	`if (webhostLanguageWorkerChannels != null)`
`314`	`314`	`{`
`315`		`- int countOfReadyChannels = 0;`
	`315`	`+ int workerProcessCount = 0;`
`316`	`316`	`foreach (string workerId in webhostLanguageWorkerChannels.Keys.ToList())`
`317`	`317`	`{`
`318`	`318`	`if (webhostLanguageWorkerChannels.TryGetValue(workerId, out TaskCompletionSource<IRpcWorkerChannel> initializedLanguageWorkerChannelTask))`
`@@ -327,8 +327,9 @@ public async Task InitializeAsync(IEnumerable<FunctionMetadata> functions, Cance`
`327`	`327`	`{`
`328`	`328`	`initializedLanguageWorkerChannel.SetupFunctionInvocationBuffers(_functions);`
`329`	`329`	`initializedLanguageWorkerChannel.SendFunctionLoadRequests(_managedDependencyOptions.Value, _scriptOptions.FunctionTimeout);`
`330`		`- ++countOfReadyChannels;`
`331`	`330`	`}`
	`331`	`+`
	`332`	`+ ++workerProcessCount;`
`332`	`333`	`}`
`333`	`334`	`catch (Exception ex)`
`334`	`335`	`{`
`@@ -337,7 +338,7 @@ public async Task InitializeAsync(IEnumerable<FunctionMetadata> functions, Cance`
`337`	`338`	`}`
`338`	`339`	`}`
`339`	`340`	`}`
`340`		`- StartWorkerProcesses(countOfReadyChannels, InitializeWebhostLanguageWorkerChannel, true, _webHostLanguageWorkerChannelManager.GetChannels(_workerRuntime));`
	`341`	`+ StartWorkerProcesses(workerProcessCount, InitializeWebhostLanguageWorkerChannel, true, _webHostLanguageWorkerChannelManager.GetChannels(_workerRuntime));`
`341`	`342`	`}`
`342`	`343`	`else`
`343`	`344`	`{`