Language workers concurrency

Author: alrod

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

@@ -10,6 +10,7 @@
 using System.Linq;
 using System.Reactive.Linq;
 using System.Text;
+using System.Threading;
 using System.Threading.Tasks;
 using System.Threading.Tasks.Dataflow;
 using Microsoft.Azure.WebJobs.Script.Description;
@@ -40,6 +41,8 @@ internal class GrpcWorkerChannel : IRpcWorkerChannel, IDisposable
         private readonly IEnvironment _environment;
         private readonly IOptionsMonitor<ScriptApplicationHostOptions> _applicationHostOptions;
         private readonly ISharedMemoryManager _sharedMemoryManager;
+        private readonly List<TimeSpan> _workerStatusLatencyHistory = new List<TimeSpan>();
+        private readonly IOptions<WorkerConcurrencyOptions> _workerConcurrencyOptions;
 
         private IDisposable _functionLoadRequestResponseEvent;
         private bool _disposed;
@@ -68,6 +71,9 @@ internal class GrpcWorkerChannel : IRpcWorkerChannel, IDisposable
         private TimeSpan _functionLoadTimeout = TimeSpan.FromMinutes(10);
         private bool _isSharedMemoryDataTransferEnabled;
 
+        private object _syncLock = new object();
+        private System.Timers.Timer _timer;
+
         internal GrpcWorkerChannel(
            string workerId,
            IScriptEventManager eventManager,
@@ -79,7 +85,8 @@ internal GrpcWorkerChannel(
            IEnvironment environment,
            IOptionsMonitor<ScriptApplicationHostOptions> applicationHostOptions,
            ISharedMemoryManager sharedMemoryManager,
-           IFunctionDataCache functionDataCache)
+           IFunctionDataCache functionDataCache,
+           IOptions<WorkerConcurrencyOptions> workerConcurrencyOptions)
         {
             _workerId = workerId;
             _eventManager = eventManager;
@@ -91,6 +98,7 @@ internal GrpcWorkerChannel(
             _environment = environment;
             _applicationHostOptions = applicationHostOptions;
             _sharedMemoryManager = sharedMemoryManager;
+            _workerConcurrencyOptions = workerConcurrencyOptions;
 
             _workerCapabilities = new GrpcCapabilities(_workerChannelLogger);
 
@@ -134,7 +142,7 @@ public bool IsChannelReadyForInvocations()
             return !_disposing && !_disposed && _state.HasFlag(RpcWorkerChannelState.InvocationBuffersInitialized | RpcWorkerChannelState.Initialized);
         }
 
-        public async Task StartWorkerProcessAsync()
+        public async Task StartWorkerProcessAsync(CancellationToken cancellationToken)
         {
             _startSubscription = _inboundWorkerEvents.Where(msg => msg.MessageType == MsgType.StartStream)
                 .Timeout(_workerConfig.CountOptions.ProcessStartupTimeout)
@@ -174,6 +182,12 @@ public async Task<WorkerStatus> GetWorkerStatusAsync()
                 }
             }
 
+            workerStatus.IsReady = IsChannelReadyForInvocations();
+            if (_environment.IsWorkerDynamicConcurrencyEnabled())
+            {
+                workerStatus.LatencyHistory = GetLatencies();
+            }
+
             return workerStatus;
         }
 
@@ -750,6 +764,7 @@ protected virtual void Dispose(bool disposing)
                 {
                     _startLatencyMetric?.Dispose();
                     _startSubscription?.Dispose();
+                    _timer?.Dispose();
 
                     // unlink function inputs
                     foreach (var link in _inputLinks)
@@ -843,5 +858,64 @@ internal bool IsSharedMemoryDataTransferEnabled()
             _workerChannelLogger.LogDebug("IsSharedMemoryDataTransferEnabled: {SharedMemoryDataTransferEnabled}", capabilityEnabled);
             return capabilityEnabled;
         }
+
+        internal void EnsureTimerStarted()
+        {
+            if (_environment.IsWorkerDynamicConcurrencyEnabled())
+            {
+                lock (_syncLock)
+                {
+                    if (_timer == null)
+                    {
+                        _timer = new System.Timers.Timer()
+                        {
+                            AutoReset = false,
+                            Interval = _workerConcurrencyOptions.Value.CheckInterval.TotalMilliseconds,
+                        };
+
+                        _timer.Elapsed += OnTimer;
+                        _timer.Start();
+                    }
+                }
+            }
+        }
+
+        internal IEnumerable<TimeSpan> GetLatencies()
+        {
+            EnsureTimerStarted();
+            return _workerStatusLatencyHistory;
+        }
+
+        internal async void OnTimer(object sender, System.Timers.ElapsedEventArgs e)
+        {
+            if (_disposed)
+            {
+                return;
+            }
+
+            try
+            {
+                WorkerStatus workerStatus = await GetWorkerStatusAsync();
+                AddSample(_workerStatusLatencyHistory, workerStatus.Latency);
+            }
+            catch
+            {
+                // Don't allow background execptions to escape
+                // E.g. when a rpc channel is shutting down we can process exceptions
+            }
+            _timer.Start();
+        }
+
+        private void AddSample<T>(List<T> samples, T sample)
+        {
+            lock (_syncLock)
+            {
+                if (samples.Count == _workerConcurrencyOptions.Value.HistorySize)
+                {
+                    samples.RemoveAt(0);
+                }
+                samples.Add(sample);
+            }
+        }
     }
 }

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

@@ -24,9 +24,10 @@ public class GrpcWorkerChannelFactory : IRpcWorkerChannelFactory
         private readonly IOptionsMonitor<ScriptApplicationHostOptions> _applicationHostOptions = null;
         private readonly ISharedMemoryManager _sharedMemoryManager = null;
         private readonly IFunctionDataCache _functionDataCache = null;
+        private readonly IOptions<WorkerConcurrencyOptions> _workerConcurrencyOptions;
 
         public GrpcWorkerChannelFactory(IScriptEventManager eventManager, IEnvironment environment, IRpcServer rpcServer, ILoggerFactory loggerFactory, IOptionsMonitor<LanguageWorkerOptions> languageWorkerOptions,
-            IOptionsMonitor<ScriptApplicationHostOptions> applicationHostOptions, IRpcWorkerProcessFactory rpcWorkerProcessManager, ISharedMemoryManager sharedMemoryManager, IFunctionDataCache functionDataCache)
+            IOptionsMonitor<ScriptApplicationHostOptions> applicationHostOptions, IRpcWorkerProcessFactory rpcWorkerProcessManager, ISharedMemoryManager sharedMemoryManager, IFunctionDataCache functionDataCache, IOptions<WorkerConcurrencyOptions> workerConcurrencyOptions)
         {
             _eventManager = eventManager;
             _loggerFactory = loggerFactory;
@@ -35,6 +36,7 @@ public GrpcWorkerChannelFactory(IScriptEventManager eventManager, IEnvironment e
             _applicationHostOptions = applicationHostOptions;
             _sharedMemoryManager = sharedMemoryManager;
             _functionDataCache = functionDataCache;
+            _workerConcurrencyOptions = workerConcurrencyOptions;
         }
 
         public IRpcWorkerChannel Create(string scriptRootPath, string runtime, IMetricsLogger metricsLogger, int attemptCount, IEnumerable<RpcWorkerConfig> workerConfigs)
@@ -58,7 +60,8 @@ public IRpcWorkerChannel Create(string scriptRootPath, string runtime, IMetricsL
                          _environment,
                          _applicationHostOptions,
                          _sharedMemoryManager,
-                         _functionDataCache);
+                         _functionDataCache,
+                         _workerConcurrencyOptions);
         }
     }
 }

src/WebJobs.Script.WebHost/DependencyInjection/DependencyValidator/DependencyValidator.cs

@@ -12,6 +12,7 @@
 using Microsoft.Azure.WebJobs.Script.Scale;
 using Microsoft.Azure.WebJobs.Script.WebHost.Diagnostics;
 using Microsoft.Azure.WebJobs.Script.Workers;
+using Microsoft.Azure.WebJobs.Script.Workers.Rpc;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
@@ -46,6 +47,7 @@ private static ExpectedDependencyBuilder CreateExpectedDependencies()
                 .ExpectFactory<IFileMonitoringService>()
                 .Expect<WorkerConsoleLogService>()
                 .Expect<FunctionInvocationDispatcherShutdownManager>()
+                .Expect<WorkerConcurrencyManager>()
                 .Optional<FunctionsScaleMonitorService>()
                 .Optional<FuncAppFileProvisioningService>() // Used by powershell.
                 .Optional<JobHostService>() // Missing when host is offline.

src/WebJobs.Script/Config/HostJsonFileConfigurationSource.cs

@@ -51,7 +51,7 @@ public class HostJsonFileConfigurationProvider : ConfigurationProvider
             {
                 "version", "functionTimeout", "retry", "functions", "http", "watchDirectories", "watchFiles", "queues", "serviceBus",
                 "eventHub", "singleton", "logging", "aggregator", "healthMonitor", "extensionBundle", "managedDependencies",
-                "customHandler", "httpWorker", "extensions"
+                "customHandler", "httpWorker", "extensions", "concurrency"
             };
 
             private readonly HostJsonFileConfigurationSource _configurationSource;

src/WebJobs.Script/Config/WorkerConcurrencyOptions.cs

@@ -0,0 +1,59 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System;
+using System.ComponentModel.DataAnnotations;
+using Newtonsoft.Json;
+using Newtonsoft.Json.Linq;
+
+namespace Microsoft.Azure.WebJobs.Script
+{
+    public class WorkerConcurrencyOptions
+    {
+        public WorkerConcurrencyOptions()
+        {
+            // Setting default values
+            LatencyThreshold = TimeSpan.FromMilliseconds(100);
+            AdjustmentPeriod = TimeSpan.FromSeconds(10);
+            CheckInterval = TimeSpan.FromSeconds(1);
+            HistorySize = 10;
+            NewWorkerThreshold = 0.2F;
+            MaxWorkerCount = 10;
+        }
+
+        /// <summary>
+        /// Gets or sets the latency threshold dictating when worker channel is overloaded.
+        /// </summary>
+        public TimeSpan LatencyThreshold { get; set; }
+
+        /// <summary>
+        /// Gets or sets the time when worker
+        /// channels will start to monitor again after adding a new worker.
+        /// </summary>
+        public TimeSpan AdjustmentPeriod { get; set; }
+
+        /// <summary>
+        /// Gets or sets interval to check worker channel state.
+        /// </summary>
+        public TimeSpan CheckInterval { get; set; }
+
+        /// <summary>
+        /// Gets or sets the history size to store workers channel states.
+        /// </summary>
+        public int HistorySize { get; set; }
+
+        /// <summary>
+        /// Gets or sets the threshold dictating when a new worker will be added.
+        /// Value should be between 0 and 1 indicating the percentage of overloaded channel latency samples required to trigger a addition of a new worker
+        /// </summary>
+        [Range(typeof(float), "0F", "1F", ErrorMessage = "Value for {0} must be between {1} and {2}.")]
+        public float NewWorkerThreshold { get; set; }
+
+        /// <summary>
+        /// Gets or sets the max count of workers.
+        /// It will be set depending on SKU if equal to 0.
+        /// </summary>
+        [Range(typeof(int), "1", "100", ErrorMessage = "Value for {0} must be between {1} and {2}.")]
+        public int MaxWorkerCount { get; set; }
+    }
+}

src/WebJobs.Script/Environment/EnvironmentExtensions.cs

@@ -472,5 +472,14 @@ public static bool DrainOnApplicationStoppingEnabled(this IEnvironment environme
             return !string.IsNullOrEmpty(environment.GetEnvironmentVariable(KubernetesServiceHost)) ||
                 (bool.TryParse(environment.GetEnvironmentVariable(DrainOnApplicationStopping), out bool v) && v);
         }
+
+        public static bool IsWorkerDynamicConcurrencyEnabled(this IEnvironment environment)
+        {
+            if (bool.TryParse(environment.GetEnvironmentVariable(RpcWorkerConstants.FunctionsWorkerDynamicConcurrencyEnabled), out bool concurrencyEnabled))
+            {
+                return concurrencyEnabled && string.IsNullOrEmpty(environment.GetEnvironmentVariable(RpcWorkerConstants.FunctionsWorkerProcessCountSettingName));
+            }
+            return false;
+        }
     }
 }

src/WebJobs.Script/ScriptHostBuilderExtensions.cs

@@ -251,6 +251,7 @@ public static IHostBuilder AddScriptHostCore(this IHostBuilder builder, ScriptAp
                 services.AddSingleton<IScriptJobHost>(p => p.GetRequiredService<ScriptHost>());
                 services.AddSingleton<IJobHost>(p => p.GetRequiredService<ScriptHost>());
                 services.AddSingleton<IFunctionProvider, ProxyFunctionProvider>();
+                services.AddSingleton<IHostedService, WorkerConcurrencyManager>();
 
                 services.AddSingleton<ITypeLocator, ScriptTypeLocator>();
                 services.AddSingleton<ScriptSettingsManager>();
@@ -271,6 +272,7 @@ public static IHostBuilder AddScriptHostCore(this IHostBuilder builder, ScriptAp
                 services.ConfigureOptions<JobHostFunctionTimeoutOptionsSetup>();
                 // LanguageWorkerOptionsSetup should be registered in WebHostServiceCollection as well to enable starting worker processing in placeholder mode.
                 services.ConfigureOptions<LanguageWorkerOptionsSetup>();
+                services.AddOptions<WorkerConcurrencyOptions>();
                 services.ConfigureOptions<HttpWorkerOptionsSetup>();
                 services.ConfigureOptions<ManagedDependencyOptionsSetup>();
                 services.AddOptions<FunctionResultAggregatorOptions>()
@@ -343,6 +345,7 @@ public static void AddCommonServices(IServiceCollection services)
             services.TryAddSingleton<IEnvironment>(SystemEnvironment.Instance);
             services.TryAddSingleton<HostPerformanceManager>();
             services.ConfigureOptions<HostHealthMonitorOptionsSetup>();
+
             AddProcessRegistry(services);
         }
 

src/WebJobs.Script/Utility.cs

@@ -4,6 +4,7 @@
 using System;
 using System.Collections.Generic;
 using System.Collections.ObjectModel;
+using System.Collections.Specialized;
 using System.Dynamic;
 using System.Globalization;
 using System.IO;
@@ -261,6 +262,11 @@ public static IReadOnlyDictionary<string, string> ToStringValues(this IReadOnlyD
             return data.ToDictionary(p => p.Key, p => p.Value != null ? p.Value.ToString() : null, StringComparer.OrdinalIgnoreCase);
         }
 
+        public static string GetValueOrNull(this StringDictionary dictionary, string key)
+        {
+            return dictionary.ContainsKey(key) ? dictionary[key] : null;
+        }
+
         // "Namespace.Class.Method" --> "Method"
         public static string GetFunctionShortName(string functionName)
         {

src/WebJobs.Script/Workers/FunctionInvocationDispatcherFactory.cs

@@ -30,7 +30,8 @@ public FunctionInvocationDispatcherFactory(IOptions<ScriptJobHostOptions> script
             IWebHostRpcWorkerChannelManager webHostLanguageWorkerChannelManager,
             IJobHostRpcWorkerChannelManager jobHostLanguageWorkerChannelManager,
             IOptions<ManagedDependencyOptions> managedDependencyOptions,
-            IRpcFunctionInvocationDispatcherLoadBalancer functionDispatcherLoadBalancer)
+            IRpcFunctionInvocationDispatcherLoadBalancer functionDispatcherLoadBalancer,
+            IOptions<WorkerConcurrencyOptions> workerConcurrencyOptions)
         {
             if (httpWorkerOptions.Value == null)
             {
@@ -53,7 +54,8 @@ public FunctionInvocationDispatcherFactory(IOptions<ScriptJobHostOptions> script
                 webHostLanguageWorkerChannelManager,
                 jobHostLanguageWorkerChannelManager,
                 managedDependencyOptions,
-                functionDispatcherLoadBalancer);
+                functionDispatcherLoadBalancer,
+                workerConcurrencyOptions);
         }
 
         public IFunctionInvocationDispatcher GetFunctionDispatcher() => _functionDispatcher;

src/WebJobs.Script/Workers/Http/HttpFunctionInvocationDispatcher.cs

@@ -59,6 +59,11 @@ public HttpFunctionInvocationDispatcher(IOptions<ScriptJobHostOptions> scriptHos
                .Subscribe(WorkerRestart);
         }
 
+        // For tests
+        internal HttpFunctionInvocationDispatcher()
+        {
+        }
+
         public FunctionInvocationDispatcherState State { get; private set; }
 
         public int ErrorEventsThreshold { get; private set; }
@@ -119,6 +124,12 @@ public async void WorkerRestart(HttpWorkerRestartEvent workerRestart)
             }
         }
 
+        public Task StartWorkerChannel()
+        {
+            // currently only one worker
+            return Task.CompletedTask;
+        }
+
         private async Task DisposeAndRestartWorkerChannel(string workerId)
         {
             // Since we only have one HTTP worker process, as soon as we dispose it, InvokeAsync will fail. Set state to