[Backport] Ensure _maxProcessCount is always initialized (#8359)

* backport maxprocesscount

* remove old maxproccount prop

Author: liliankasem

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

@@ -31,15 +31,15 @@ internal class RpcFunctionInvocationDispatcher : IFunctionInvocationDispatcher
         private readonly SemaphoreSlim _startWorkerProcessLock = new SemaphoreSlim(1, 1);
         private readonly TimeSpan _thresholdBetweenRestarts = TimeSpan.FromMinutes(WorkerConstants.WorkerRestartErrorIntervalThresholdInMinutes);
         private readonly IOptions<WorkerConcurrencyOptions> _workerConcurrencyOptions;
+        private readonly IEnumerable<RpcWorkerConfig> _workerConfigs;
+        private readonly Lazy<Task<int>> _maxProcessCount;
 
         private IScriptEventManager _eventManager;
-        private IEnumerable<RpcWorkerConfig> _workerConfigs;
         private IWebHostRpcWorkerChannelManager _webHostLanguageWorkerChannelManager;
         private IJobHostRpcWorkerChannelManager _jobHostLanguageWorkerChannelManager;
         private IDisposable _workerErrorSubscription;
         private IDisposable _workerRestartSubscription;
         private ScriptJobHostOptions _scriptOptions;
-        private int _maxProcessCount;
         private IRpcFunctionInvocationDispatcherLoadBalancer _functionDispatcherLoadBalancer;
         private bool _disposed = false;
         private bool _disposing = false;
@@ -71,12 +71,12 @@ public RpcFunctionInvocationDispatcher(IOptions<ScriptJobHostOptions> scriptHost
         {
             _metricsLogger = metricsLogger;
             _scriptOptions = scriptHostOptions.Value;
-            _environment = environment;
+            _environment = environment ?? throw new ArgumentNullException(nameof(environment));
             _applicationLifetime = applicationLifetime;
             _webHostLanguageWorkerChannelManager = webHostLanguageWorkerChannelManager;
             _jobHostLanguageWorkerChannelManager = jobHostLanguageWorkerChannelManager;
             _eventManager = eventManager;
-            _workerConfigs = languageWorkerOptions.CurrentValue.WorkerConfigs;
+            _workerConfigs = languageWorkerOptions?.CurrentValue?.WorkerConfigs ?? throw new ArgumentNullException(nameof(languageWorkerOptions));
             _managedDependencyOptions = managedDependencyOptions ?? throw new ArgumentNullException(nameof(managedDependencyOptions));
             _logger = loggerFactory.CreateLogger<RpcFunctionInvocationDispatcher>();
             _rpcWorkerChannelFactory = rpcWorkerChannelFactory;
@@ -86,13 +86,13 @@ public RpcFunctionInvocationDispatcher(IOptions<ScriptJobHostOptions> scriptHost
             _workerIndexing = Utility.CanWorkerIndex(_workerConfigs, _environment);
             State = FunctionInvocationDispatcherState.Default;
 
-            _workerErrorSubscription = _eventManager.OfType<WorkerErrorEvent>()
-               .Subscribe(WorkerError);
-            _workerRestartSubscription = _eventManager.OfType<WorkerRestartEvent>()
-               .Subscribe(WorkerRestart);
+            _workerErrorSubscription = _eventManager.OfType<WorkerErrorEvent>().Subscribe(WorkerError);
+            _workerRestartSubscription = _eventManager.OfType<WorkerRestartEvent>().Subscribe(WorkerRestart);
 
             _shutdownStandbyWorkerChannels = ShutdownWebhostLanguageWorkerChannels;
             _shutdownStandbyWorkerChannels = _shutdownStandbyWorkerChannels.Debounce(milliseconds: 5000);
+
+            _maxProcessCount = new Lazy<Task<int>>(GetMaxProcessCount);
         }
 
         public FunctionInvocationDispatcherState State { get; private set; }
@@ -101,12 +101,27 @@ public RpcFunctionInvocationDispatcher(IOptions<ScriptJobHostOptions> scriptHost
 
         public IJobHostRpcWorkerChannelManager JobHostLanguageWorkerChannelManager => _jobHostLanguageWorkerChannelManager;
 
-        internal ConcurrentStack<WorkerErrorEvent> LanguageWorkerErrors => _languageWorkerErrors;
+        internal Task<int> MaxProcessCount => _maxProcessCount.Value;
 
-        internal int MaxProcessCount => _maxProcessCount;
+        internal ConcurrentStack<WorkerErrorEvent> LanguageWorkerErrors => _languageWorkerErrors;
 
         internal IWebHostRpcWorkerChannelManager WebHostLanguageWorkerChannelManager => _webHostLanguageWorkerChannelManager;
 
+        private async Task<int> GetMaxProcessCount()
+        {
+            if (_workerConcurrencyOptions != null && !string.IsNullOrEmpty(_workerRuntime))
+            {
+                var workerConfig = _workerConfigs.Where(c => c.Description.Language.Equals(_workerRuntime, StringComparison.InvariantCultureIgnoreCase))
+                                                 .FirstOrDefault();
+                if (workerConfig != null)
+                {
+                    return _environment.IsWorkerDynamicConcurrencyEnabled() ? _workerConcurrencyOptions.Value.MaxWorkerCount : workerConfig.CountOptions.ProcessCount;
+                }
+            }
+
+            return (await GetAllWorkerChannelsAsync()).Count();
+        }
+
         internal async Task InitializeJobhostLanguageWorkerChannelAsync()
         {
             await InitializeJobhostLanguageWorkerChannelAsync(0);
@@ -131,7 +146,7 @@ internal async Task InitializeJobhostLanguageWorkerChannelAsync(int attemptCount
         private void SetFunctionDispatcherStateToInitializedAndLog()
         {
             State = FunctionInvocationDispatcherState.Initialized;
-            // Do not change this log message. Vs Code relies on this to figure out when to attach debuger to the worker process.
+            // Do not change this log message. Vs Code relies on this to figure out when to attach debugger to the worker process.
             _logger.LogInformation("Worker process started and initialized.");
         }
 
@@ -170,7 +185,7 @@ private void StartWorkerProcesses(int startIndex, Func<Task> startAction, bool i
         {
             Task.Run(async () =>
             {
-                for (var count = startIndex; count < _maxProcessCount
+                for (var count = startIndex; count < (await _maxProcessCount.Value)
                     && !_processStartCancellationToken.IsCancellationRequested; count++)
                 {
                     if (_environment.IsWorkerDynamicConcurrencyEnabled() && count > 0)
@@ -229,6 +244,7 @@ public async Task InitializeAsync(IEnumerable<FunctionMetadata> functions, Cance
             }
 
             var workerConfig = _workerConfigs.Where(c => c.Description.Language.Equals(_workerRuntime, StringComparison.InvariantCultureIgnoreCase)).FirstOrDefault();
+
             if (workerConfig == null && (functions == null || functions.Count() == 0))
             {
                 // Only throw if workerConfig is null AND some functions have been found.
@@ -239,16 +255,15 @@ public async Task InitializeAsync(IEnumerable<FunctionMetadata> functions, Cance
             if ((functions == null || functions.Count() == 0) && !_workerIndexing)
             {
                 // do not initialize function dispatcher if there are no functions, unless the worker is indexing
-                _logger.LogDebug("RpcFunctionInvocationDispatcher received no functions");
+                _logger.LogDebug($"{nameof(RpcFunctionInvocationDispatcher)} received no functions");
                 return;
             }
 
             _functions = functions ?? new List<FunctionMetadata>();
-            _maxProcessCount = workerConfig.CountOptions.ProcessCount;
             _processStartupInterval = workerConfig.CountOptions.ProcessStartupInterval;
             _restartWait = workerConfig.CountOptions.ProcessRestartInterval;
             _shutdownTimeout = workerConfig.CountOptions.ProcessShutdownTimeout;
-            ErrorEventsThreshold = 3 * _maxProcessCount;
+            ErrorEventsThreshold = 3 * (await _maxProcessCount.Value);
 
             if (Utility.IsSupportedRuntime(_workerRuntime, _workerConfigs))
             {
@@ -419,10 +434,11 @@ internal async Task<IEnumerable<IRpcWorkerChannel>> GetInitializedWorkerChannels
         {
             IEnumerable<IRpcWorkerChannel> workerChannels = await GetAllWorkerChannelsAsync();
             IEnumerable<IRpcWorkerChannel> initializedWorkers = workerChannels.Where(ch => ch.IsChannelReadyForInvocations());
-            int workerCount = _environment.IsWorkerDynamicConcurrencyEnabled() ? _workerConcurrencyOptions.Value.MaxWorkerCount : _maxProcessCount;
+            int workerCount = await _maxProcessCount.Value;
+
             if (initializedWorkers.Count() > workerCount)
             {
-                throw new InvalidOperationException($"Number of initialized language workers exceeded:{initializedWorkers.Count()} exceeded maxProcessCount: {_maxProcessCount}");
+                throw new InvalidOperationException($"Number of initialized language workers exceeded:{initializedWorkers.Count()} exceeded maxProcessCount: {workerCount}");
             }
 
             return initializedWorkers;

test/WebJobs.Script.Tests/Workers/Rpc/RpcFunctionInvocationDispatcherTests.cs

@@ -279,6 +279,32 @@ public async Task ShutdownTests_WithInfinitelyRunningTasks_Timesout()
             }
         }
 
+        [Fact]
+        public async Task FunctionDispatcherState_PlaceholderMode_SingleWebHostChannel()
+        {
+            // Arrange
+            int expectedWebHostProcessCount = 1;
+            int expectedJobHostProcessCount = 0;
+
+            var functionDispatcher = GetTestFunctionDispatcher(runtime: RpcWorkerConstants.JavaLanguageWorkerName, placeholder: true, addWebhostChannel: true);
+            await WaitForWebhostWorkerChannelsToStartup(functionDispatcher.WebHostLanguageWorkerChannelManager, expectedWebHostProcessCount, RpcWorkerConstants.JavaLanguageWorkerName);
+            var channels = functionDispatcher.WebHostLanguageWorkerChannelManager.GetChannels(RpcWorkerConstants.JavaLanguageWorkerName).Values;
+            channels.FirstOrDefault().Task.Result.SetupFunctionInvocationBuffers(null);
+
+            // Act
+            await functionDispatcher.InitializeAsync(GetTestFunctionsList(RpcWorkerConstants.JavaLanguageWorkerName));
+            await functionDispatcher.ShutdownAsync();
+
+            // Assert
+            var webHostCount = functionDispatcher.WebHostLanguageWorkerChannelManager.GetChannels(RpcWorkerConstants.JavaLanguageWorkerName).Count();
+            var jobHostCount = functionDispatcher.JobHostLanguageWorkerChannelManager.GetChannels().Count();
+            var maxProcessCount = await functionDispatcher.MaxProcessCount;
+
+            Assert.Equal(expectedWebHostProcessCount, webHostCount);
+            Assert.Equal(expectedJobHostProcessCount, jobHostCount);
+            Assert.Equal(expectedWebHostProcessCount, maxProcessCount);
+        }
+
         [Fact]
         public async Task FunctionDispatcherState_Default_DotNetFunctions()
         {
@@ -640,7 +666,8 @@ public async Task FunctionDispatcher_ErroredWebHostChannel()
         }
 
         private static RpcFunctionInvocationDispatcher GetTestFunctionDispatcher(int maxProcessCountValue = 1, bool addWebhostChannel = false,
-            Mock<IWebHostRpcWorkerChannelManager> mockwebHostLanguageWorkerChannelManager = null, bool throwOnProcessStartUp = false, TimeSpan? startupIntervals = null, string runtime = null, bool workerIndexing = false)
+            Mock<IWebHostRpcWorkerChannelManager> mockwebHostLanguageWorkerChannelManager = null, bool throwOnProcessStartUp = false,
+            TimeSpan? startupIntervals = null, string runtime = null, bool workerIndexing = false, bool placeholder = false)
         {
             var eventManager = new ScriptEventManager();
             var metricsLogger = new Mock<IMetricsLogger>();
@@ -649,6 +676,7 @@ private static RpcFunctionInvocationDispatcher GetTestFunctionDispatcher(int max
             TimeSpan intervals = startupIntervals ?? TimeSpan.FromMilliseconds(100);
 
             testEnv.SetEnvironmentVariable(RpcWorkerConstants.FunctionsWorkerProcessCountSettingName, maxProcessCountValue.ToString());
+
             if (runtime != null)
             {
                 testEnv.SetEnvironmentVariable(EnvironmentSettingNames.FunctionWorkerRuntime, runtime);
@@ -657,6 +685,10 @@ private static RpcFunctionInvocationDispatcher GetTestFunctionDispatcher(int max
             {
                 testEnv.SetEnvironmentVariable(EnvironmentSettingNames.AzureWebJobsFeatureFlags, ScriptConstants.FeatureFlagEnableWorkerIndexing);
             }
+            if (placeholder)
+            {
+                testEnv.SetEnvironmentVariable(EnvironmentSettingNames.AzureWebsitePlaceholderMode, "1");
+            }
 
             var options = new ScriptJobHostOptions
             {
@@ -673,6 +705,7 @@ private static RpcFunctionInvocationDispatcher GetTestFunctionDispatcher(int max
             IRpcWorkerChannelFactory testLanguageWorkerChannelFactory = new TestRpcWorkerChannelFactory(eventManager, _testLogger, scriptOptions.Value.RootScriptPath, throwOnProcessStartUp);
             IWebHostRpcWorkerChannelManager testWebHostLanguageWorkerChannelManager = new TestRpcWorkerChannelManager(eventManager, _testLogger, scriptOptions.Value.RootScriptPath, testLanguageWorkerChannelFactory);
             IJobHostRpcWorkerChannelManager jobHostLanguageWorkerChannelManager = new JobHostRpcWorkerChannelManager(_testLoggerFactory);
+
             if (addWebhostChannel)
             {
                 testWebHostLanguageWorkerChannelManager.InitializeChannelAsync("java");
@@ -681,6 +714,7 @@ private static RpcFunctionInvocationDispatcher GetTestFunctionDispatcher(int max
             {
                 testWebHostLanguageWorkerChannelManager = mockwebHostLanguageWorkerChannelManager.Object;
             }
+
             var mockFunctionDispatcherLoadBalancer = new Mock<IRpcFunctionInvocationDispatcherLoadBalancer>();
 
             _javaTestChannel = new TestRpcWorkerChannel(Guid.NewGuid().ToString(), "java", eventManager, _testLogger, false);
@@ -747,14 +781,14 @@ private IEnumerable<FunctionMetadata> GetTestFunctionsList(string runtime)
             {
                 new FunctionMetadata()
                 {
-                     Language = runtime,
-                     Name = "js1"
+                    Language = runtime,
+                    Name = "js1"
                 },
 
                 new FunctionMetadata()
                 {
-                     Language = runtime,
-                     Name = "js2"
+                    Language = runtime,
+                    Name = "js2"
                 }
             };
         }