Intelligent worker concurrency can start more than 10 workers (#8854).

alrod · web-flow · commit a625f30035f0 · 2022-10-21T14:37:18.000-07:00
diff --git a/src/WebJobs.Script/Workers/WorkerConcurrencyManager.cs b/src/WebJobs.Script/Workers/WorkerConcurrencyManager.cs
@@ -118,7 +118,7 @@ public Task StopAsync(CancellationToken cancellationToken)
 
         internal async void OnTimer(object sender, System.Timers.ElapsedEventArgs e)
         {
-            if (_disposed)
+            if (_environment.IsPlaceholderModeEnabled() || _disposed)
             {
                 return;
             }
@@ -131,10 +131,9 @@ internal async void OnTimer(object sender, System.Timers.ElapsedEventArgs e)
                 {
                     if (_functionInvocationDispatcher is RpcFunctionInvocationDispatcher rpcDispatcher)
                     {
-                        // Check memory
-                        var currentChannels = (await rpcDispatcher.GetAllWorkerChannelsAsync()).Select(x => x.WorkerProcess.Process.PrivateMemorySize64);
-                        if (IsEnoughMemoryToScale(Process.GetCurrentProcess().PrivateMemorySize64,
-                            currentChannels,
+                        var allWorkerChannels = await rpcDispatcher.GetAllWorkerChannelsAsync();
+                        if (CanScale(allWorkerChannels) && IsEnoughMemoryToScale(Process.GetCurrentProcess().PrivateMemorySize64,
+                            allWorkerChannels.Select(x => x.WorkerProcess.Process.PrivateMemorySize64),
                             _memoryLimit))
                         {
                             await _functionInvocationDispatcher.StartWorkerChannel();
@@ -176,10 +175,11 @@ private void OnActivationTimer(object sender, System.Timers.ElapsedEventArgs e)
                     return;
                 }
             }
-            catch (Exception)
+            catch (Exception ex)
             {
                 // Best effort
                 // return and do not start the activation timmer again
+                _logger.LogError(ex, "Error activating worker concurrency");
                 return;
             }
 
@@ -318,6 +318,33 @@ internal bool IsEnoughMemoryToScale(long hostProcessSize, IEnumerable<long> work
             return true;
         }
 
+        /// <summary>
+        /// Checks if a new worker can be added.
+        /// </summary>
+        /// <param name="workerChannels">All current worker channels.</param>
+        /// <returns>True if a new worker can be started.</returns>
+        internal bool CanScale(IEnumerable<IRpcWorkerChannel> workerChannels)
+        {
+            // Cancel if there is any "non-ready" channel.
+            // A "ready" channel means it's ready for invocations.
+            var nonReadyWorkerChannels = workerChannels.Where(x => x.IsChannelReadyForInvocations() == false);
+            if (nonReadyWorkerChannels.Any())
+            {
+                _logger.LogDebug($"Starting new language worker canceled as there is atleast one non ready channel: TotalChannels={workerChannels.Count()}, NonReadyChannels={nonReadyWorkerChannels.Count()}");
+                return false;
+            }
+
+            // Cancel if MaxWorkerCount is reached.
+            int workersCount = workerChannels.Count();
+            if (workersCount >= _workerConcurrencyOptions.Value.MaxWorkerCount)
+            {
+                _logger.LogDebug($"Starting new language worker canceled as the count of total channels reaches the maximum limit: TotalChannels={workersCount}, MaxWorkerCount={_workerConcurrencyOptions.Value.MaxWorkerCount}");
+                return false;
+            }
+
+            return true;
+        }
+
         internal class WorkerStatusDetails
         {
             public string WorkerId { get; set; }
diff --git a/test/WebJobs.Script.Tests/Workers/Rpc/WorkerConcurrencyManagerTest.cs b/test/WebJobs.Script.Tests/Workers/Rpc/WorkerConcurrencyManagerTest.cs
@@ -308,10 +308,10 @@ public async Task AddWorkerIfNeeded_Returns_Expected(WorkerConcurrencyOptions op
                 LatencyHistory = latencies2.Select(x => TimeSpan.FromMilliseconds(x))
             });
 
-            WorkerConcurrencyManager concurrancyManager = new WorkerConcurrencyManager(functionInvocationDispatcherFactory.Object, _testEnvironment, Options.Create(options),
+            WorkerConcurrencyManager concurrencyManager = new WorkerConcurrencyManager(functionInvocationDispatcherFactory.Object, _testEnvironment, Options.Create(options),
                 _functionsHostingConfigurations, _applicationLifetime, _loggerFactory);
-            await concurrancyManager.StartAsync(CancellationToken.None);
-            bool value = concurrancyManager.NewWorkerIsRequired(workerStatuses, elapsedFromLastAdding);
+            await concurrencyManager.StartAsync(CancellationToken.None);
+            bool value = concurrencyManager.NewWorkerIsRequired(workerStatuses, elapsedFromLastAdding);
 
             Assert.Equal(value, expected);
         }
@@ -323,9 +323,9 @@ public async Task StartAsync_DoesNotGetDispatcher(string workerRuntime)
         {
             _testEnvironment.SetEnvironmentVariable(EnvironmentSettingNames.FunctionWorkerRuntime, workerRuntime);
             Mock<IFunctionInvocationDispatcherFactory> functionInvocationDispatcherFactory = new Mock<IFunctionInvocationDispatcherFactory>(MockBehavior.Strict);
-            WorkerConcurrencyManager concurrancyManager = new WorkerConcurrencyManager(functionInvocationDispatcherFactory.Object, _testEnvironment, Options.Create(new WorkerConcurrencyOptions()),
+            WorkerConcurrencyManager concurrencyManager = new WorkerConcurrencyManager(functionInvocationDispatcherFactory.Object, _testEnvironment, Options.Create(new WorkerConcurrencyOptions()),
                 _functionsHostingConfigurations, _applicationLifetime, _loggerFactory);
-            await concurrancyManager.StartAsync(CancellationToken.None);
+            await concurrencyManager.StartAsync(CancellationToken.None);
         }
 
         [Theory]
@@ -339,9 +339,9 @@ public async Task StartAsync_GetsDispatcher(string workerRuntime)
             Mock<IFunctionInvocationDispatcherFactory> functionInvocationDispatcherFactory = new Mock<IFunctionInvocationDispatcherFactory>(MockBehavior.Strict);
             functionInvocationDispatcherFactory.Setup(x => x.GetFunctionDispatcher()).Returns(functionInvocationDispatcher.Object);
 
-            WorkerConcurrencyManager concurrancyManager = new WorkerConcurrencyManager(functionInvocationDispatcherFactory.Object, _testEnvironment, Options.Create(new WorkerConcurrencyOptions()),
+            WorkerConcurrencyManager concurrencyManager = new WorkerConcurrencyManager(functionInvocationDispatcherFactory.Object, _testEnvironment, Options.Create(new WorkerConcurrencyOptions()),
                 _functionsHostingConfigurations, _applicationLifetime, _loggerFactory);
-            await concurrancyManager.StartAsync(CancellationToken.None);
+            await concurrencyManager.StartAsync(CancellationToken.None);
         }
 
         [Fact]
@@ -356,10 +356,10 @@ public async Task ActivateWorkerConcurency_FunctionsHostingConfiguration_WorkAsE
             conf.Setup(x => x.FunctionsWorkerDynamicConcurrencyEnabled).Returns(true);
             WorkerConcurrencyOptions options = new WorkerConcurrencyOptions();
 
-            WorkerConcurrencyManager concurrancyManager = new WorkerConcurrencyManager(functionInvocationDispatcherFactory.Object, testEnvironment, Options.Create(options), conf.Object,
+            WorkerConcurrencyManager concurrencyManager = new WorkerConcurrencyManager(functionInvocationDispatcherFactory.Object, testEnvironment, Options.Create(options), conf.Object,
                 _applicationLifetime, _loggerFactory);
-            concurrancyManager.ActivationTimerInterval = TimeSpan.FromMilliseconds(100);
-            await concurrancyManager.StartAsync(CancellationToken.None);
+            concurrencyManager.ActivationTimerInterval = TimeSpan.FromMilliseconds(100);
+            await concurrencyManager.StartAsync(CancellationToken.None);
             await TestHelpers.Await(() => _loggerProvider.GetAllLogMessages().SingleOrDefault(x => x.FormattedMessage.StartsWith("Dynamic worker concurrency monitoring was started by activation timer.")) != null, timeout: 1000, pollingInterval: 100);
             conf.Setup(x => x.FunctionsWorkerDynamicConcurrencyEnabled).Returns(false);
             await TestHelpers.Await(() => _loggerProvider.GetAllLogMessages().SingleOrDefault(x => x.FormattedMessage.StartsWith("Dynamic worker concurrency monitoring is disabled after activation. Shutting down Functions Host.")) != null, timeout: 1000, pollingInterval: 100);
@@ -377,14 +377,55 @@ public void IsEnoughMemory_WorkAsExpected(long availableMemory, long hostProcess
             Mock<IFunctionsHostingConfiguration> conf = new Mock<IFunctionsHostingConfiguration>();
             WorkerConcurrencyOptions options = new WorkerConcurrencyOptions();
 
-            WorkerConcurrencyManager concurrancyManager = new WorkerConcurrencyManager(functionInvocationDispatcherFactory.Object, testEnvironment,
+            WorkerConcurrencyManager concurrencyManager = new WorkerConcurrencyManager(functionInvocationDispatcherFactory.Object, testEnvironment,
                 Options.Create(options), conf.Object, _applicationLifetime, _loggerFactory);
 
-            Assert.True(concurrancyManager.IsEnoughMemoryToScale(hostProcessSize, languageWorkerSizes, availableMemory) == result);
+            Assert.True(concurrencyManager.IsEnoughMemoryToScale(hostProcessSize, languageWorkerSizes, availableMemory) == result);
             if (!result)
             {
                 Assert.Contains(_loggerProvider.GetAllLogMessages().Select(x => x.FormattedMessage), x => x.StartsWith("Starting new language worker canceled:"));
             }
         }
+
+        [Theory]
+        [InlineData(4, new bool[] { true, true, true }, true)]
+        [InlineData(3, new bool[] { true, true, true }, false)]
+        [InlineData(4, new bool[] { true, false, true }, false)]
+        public void CanScale_ReturnsExpected(int maxWorkerCount, bool[] isReadyArray, bool result)
+        {
+            TestEnvironment testEnvironment = new TestEnvironment();
+            Mock<IFunctionInvocationDispatcherFactory> functionInvocationDispatcherFactory = new Mock<IFunctionInvocationDispatcherFactory>(MockBehavior.Strict);
+            Mock<IFunctionsHostingConfiguration> conf = new Mock<IFunctionsHostingConfiguration>();
+            WorkerConcurrencyOptions options = new WorkerConcurrencyOptions();
+            options.MaxWorkerCount = maxWorkerCount;
+
+            WorkerConcurrencyManager concurrencyManager = new WorkerConcurrencyManager(functionInvocationDispatcherFactory.Object, testEnvironment,
+                Options.Create(options), conf.Object, _applicationLifetime, _loggerFactory);
+
+            List<IRpcWorkerChannel> workerChannels = new List<IRpcWorkerChannel>();
+            foreach (bool isReady in isReadyArray)
+            {
+                Mock<IRpcWorkerChannel> mock = new Mock<IRpcWorkerChannel>();
+                mock.Setup(x => x.IsChannelReadyForInvocations()).Returns(isReady);
+                workerChannels.Add(mock.Object);
+            }
+            Assert.Equal(concurrencyManager.CanScale(workerChannels), result);
+        }
+
+        [Fact]
+        public void OnTimer_WorksAsExpected_IfPlaceholderMode_Enabled()
+        {
+            TestEnvironment testEnvironment = new TestEnvironment();
+            Mock<IFunctionInvocationDispatcherFactory> functionInvocationDispatcherFactory = new Mock<IFunctionInvocationDispatcherFactory>(MockBehavior.Strict);
+            Mock<IFunctionsHostingConfiguration> conf = new Mock<IFunctionsHostingConfiguration>();
+            WorkerConcurrencyOptions options = new WorkerConcurrencyOptions();
+
+            WorkerConcurrencyManager concurrencyManager = new WorkerConcurrencyManager(functionInvocationDispatcherFactory.Object, testEnvironment,
+                Options.Create(options), conf.Object, _applicationLifetime, _loggerFactory);
+
+            testEnvironment.SetEnvironmentVariable(EnvironmentSettingNames.AzureWebsitePlaceholderMode, "1");
+            concurrencyManager.OnTimer(null, null);
+            Assert.Empty(_loggerProvider.GetAllLogMessages().Where(x => x.FormattedMessage == "Error monitoring worker concurrency"));
+        }
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -118,7 +118,7 @@ public Task StopAsync(CancellationToken cancellationToken)`
`118`	`118`
`119`	`119`	`internal async void OnTimer(object sender, System.Timers.ElapsedEventArgs e)`
`120`	`120`	`{`
`121`		`- if (_disposed)`
	`121`	`+ if (_environment.IsPlaceholderModeEnabled() \|\| _disposed)`
`122`	`122`	`{`
`123`	`123`	`return;`
`124`	`124`	`}`
`@@ -131,10 +131,9 @@ internal async void OnTimer(object sender, System.Timers.ElapsedEventArgs e)`
`131`	`131`	`{`
`132`	`132`	`if (_functionInvocationDispatcher is RpcFunctionInvocationDispatcher rpcDispatcher)`
`133`	`133`	`{`
`134`		`- // Check memory`
`135`		`- var currentChannels = (await rpcDispatcher.GetAllWorkerChannelsAsync()).Select(x => x.WorkerProcess.Process.PrivateMemorySize64);`
`136`		`- if (IsEnoughMemoryToScale(Process.GetCurrentProcess().PrivateMemorySize64,`
`137`		`- currentChannels,`
	`134`	`+ var allWorkerChannels = await rpcDispatcher.GetAllWorkerChannelsAsync();`
	`135`	`+ if (CanScale(allWorkerChannels) && IsEnoughMemoryToScale(Process.GetCurrentProcess().PrivateMemorySize64,`
	`136`	`+ allWorkerChannels.Select(x => x.WorkerProcess.Process.PrivateMemorySize64),`
`138`	`137`	`_memoryLimit))`
`139`	`138`	`{`
`140`	`139`	`await _functionInvocationDispatcher.StartWorkerChannel();`
`@@ -176,10 +175,11 @@ private void OnActivationTimer(object sender, System.Timers.ElapsedEventArgs e)`
`176`	`175`	`return;`
`177`	`176`	`}`
`178`	`177`	`}`
`179`		`- catch (Exception)`
	`178`	`+ catch (Exception ex)`
`180`	`179`	`{`
`181`	`180`	`// Best effort`
`182`	`181`	`// return and do not start the activation timmer again`
	`182`	`+ _logger.LogError(ex, "Error activating worker concurrency");`
`183`	`183`	`return;`
`184`	`184`	`}`
`185`	`185`
`@@ -318,6 +318,33 @@ internal bool IsEnoughMemoryToScale(long hostProcessSize, IEnumerable<long> work`
`318`	`318`	`return true;`
`319`	`319`	`}`
`320`	`320`
	`321`	`+ /// <summary>`
	`322`	`+ /// Checks if a new worker can be added.`
	`323`	`+ /// </summary>`
	`324`	`+ /// <param name="workerChannels">All current worker channels.</param>`
	`325`	`+ /// <returns>True if a new worker can be started.</returns>`
	`326`	`+ internal bool CanScale(IEnumerable<IRpcWorkerChannel> workerChannels)`
	`327`	`+ {`
	`328`	`+ // Cancel if there is any "non-ready" channel.`
	`329`	`+ // A "ready" channel means it's ready for invocations.`
	`330`	`+ var nonReadyWorkerChannels = workerChannels.Where(x => x.IsChannelReadyForInvocations() == false);`
	`331`	`+ if (nonReadyWorkerChannels.Any())`
	`332`	`+ {`
	`333`	`+ _logger.LogDebug($"Starting new language worker canceled as there is atleast one non ready channel: TotalChannels={workerChannels.Count()}, NonReadyChannels={nonReadyWorkerChannels.Count()}");`
	`334`	`+ return false;`
	`335`	`+ }`
	`336`	`+`
	`337`	`+ // Cancel if MaxWorkerCount is reached.`
	`338`	`+ int workersCount = workerChannels.Count();`
	`339`	`+ if (workersCount >= _workerConcurrencyOptions.Value.MaxWorkerCount)`
	`340`	`+ {`
	`341`	`+ _logger.LogDebug($"Starting new language worker canceled as the count of total channels reaches the maximum limit: TotalChannels={workersCount}, MaxWorkerCount={_workerConcurrencyOptions.Value.MaxWorkerCount}");`
	`342`	`+ return false;`
	`343`	`+ }`
	`344`	`+`
	`345`	`+ return true;`
	`346`	`+ }`
	`347`	`+`
`321`	`348`	`internal class WorkerStatusDetails`
`322`	`349`	`{`
`323`	`350`	`public string WorkerId { get; set; }`