Fully Adaptive Stateless Worker (#9378)

Co-authored-by: Reuben Bond <[email protected]>
Co-authored-by: Reuben Bond <[email protected]>

Author: ledjon-behluli

src/Orleans.Core.Abstractions/Concurrency/GrainAttributeConcurrency.cs

@@ -4,6 +4,7 @@
 using Orleans.Runtime;
 using System;
 using System.Collections.Generic;
+using System.Globalization;
 
 namespace Orleans.Concurrency
 {
@@ -57,14 +58,21 @@ public sealed class StatelessWorkerAttribute : PlacementAttribute, IGrainPropert
         /// <summary>
         /// Initializes a new instance of the <see cref="StatelessWorkerAttribute"/> class.
         /// </summary>
-        /// <param name="maxLocalWorkers">
-        /// The maximum local workers.
-        /// </param>
+        /// <param name="maxLocalWorkers">The maximum local workers.</param>
         public StatelessWorkerAttribute(int maxLocalWorkers)
             : base(new StatelessWorkerPlacement(maxLocalWorkers))
         {
         }
 
+        /// <summary>
+        /// Initializes a new instance of the <see cref="StatelessWorkerAttribute"/> class.
+        /// </summary>
+        /// <param name="maxLocalWorkers">The maximum local workers.</param>
+        public StatelessWorkerAttribute(int maxLocalWorkers, bool removeIdleWorkers)
+            : base(new StatelessWorkerPlacement(maxLocalWorkers, removeIdleWorkers))
+        {
+        }
+
         /// <summary>
         /// Initializes a new instance of the <see cref="StatelessWorkerAttribute"/> class.
         /// </summary>

src/Orleans.Core.Abstractions/Core/IGrainBase.cs

@@ -290,14 +290,19 @@ public enum DeactivationReasonCode : byte
         ApplicationError,
 
         /// <summary>
-        /// The application requested that this activation deactivate.
+        /// The application requested to deactivate this activation.
         /// </summary>
         ApplicationRequested,
 
         /// <summary>
         /// This activation is migrating to a new location.
         /// </summary>
         Migrating,
+
+        /// <summary>
+        /// The runtime requested to deactivate this activation.
+        /// </summary>
+        RuntimeRequested
     }
 
     internal static class DeactivationReasonCodeExtensions

src/Orleans.Core.Abstractions/Placement/StatelessWorkerPlacement.cs

@@ -12,6 +12,8 @@ namespace Orleans.Runtime
     internal sealed class StatelessWorkerPlacement : PlacementStrategy
     {
         private const string MaxLocalPropertyKey = "max-local-instances";
+        private const string RemoveIdleWorkersPropertyKey = "remove-idle-workers";
+
         private static readonly int DefaultMaxStatelessWorkers = Environment.ProcessorCount;
 
         /// <inheritdoc/>
@@ -23,17 +25,38 @@ internal sealed class StatelessWorkerPlacement : PlacementStrategy
         [Id(0)]
         public int MaxLocal { get; private set; }
 
+        /// <summary>
+        /// When set to <see langword="true"/>, idle workers will be proactively deactivated by the runtime.
+        /// Otherwise if <see langword="false"/>, than the workers will be deactivated according to collection age.
+        /// </summary>
+        [Id(1)]
+        public bool RemoveIdleWorkers { get; private set; } = true;
+
+        /// <summary>
+        /// Initializes a new instance of the <see cref="StatelessWorkerPlacement"/> class.
+        /// </summary>
+        /// <param name="maxLocal">
+        /// The maximum number of local instances which can be simultaneously active for a given grain.
+        /// </param>
+        internal StatelessWorkerPlacement(int maxLocal) : this(maxLocal, true)
+        {
+        }
+
         /// <summary>
         /// Initializes a new instance of the <see cref="StatelessWorkerPlacement"/> class.
         /// </summary>
         /// <param name="maxLocal">
         /// The maximum number of local instances which can be simultaneously active for a given grain.
         /// </param>
-        internal StatelessWorkerPlacement(int maxLocal)
+        /// <param name="removeIdleWorkers">
+        /// Whether idle workers will be proactively deactivated by the runtime instead of only being deactivated according to collection age.
+        /// </param>
+        internal StatelessWorkerPlacement(int maxLocal, bool removeIdleWorkers)
         {
             // If maxLocal was not specified on the StatelessWorkerAttribute,
             // we will use the defaultMaxStatelessWorkers, which is System.Environment.ProcessorCount.
             this.MaxLocal = maxLocal > 0 ? maxLocal : DefaultMaxStatelessWorkers;
+            this.RemoveIdleWorkers = removeIdleWorkers;
         }
 
         /// <summary>
@@ -44,28 +67,39 @@ public StatelessWorkerPlacement() : this(-1)
         }
 
         /// <inheritdoc/>
-        public override string ToString() => $"StatelessWorkerPlacement(max={MaxLocal})";
+        public override string ToString() => $"StatelessWorkerPlacement(MaxLocal={MaxLocal}, RemoveIdleWorkers={RemoveIdleWorkers})";
 
         /// <inheritdoc/>
         public override void Initialize(GrainProperties properties)
         {
             base.Initialize(properties);
-            if (properties.Properties.TryGetValue(MaxLocalPropertyKey, out var value)
-                && !string.IsNullOrWhiteSpace(value))
+
+            if (properties.Properties.TryGetValue(MaxLocalPropertyKey, out var maxLocalValue) &&
+                !string.IsNullOrWhiteSpace(maxLocalValue))
+            {
+                if (int.TryParse(maxLocalValue, out var maxLocal))
+                {
+                    MaxLocal = maxLocal;
+                }
+            }
+
+            if (properties.Properties.TryGetValue(RemoveIdleWorkersPropertyKey, out var removeIdleValue) &&
+                !string.IsNullOrWhiteSpace(removeIdleValue))
             {
-                if (int.TryParse(value, out var maxLocal))
+                if (bool.TryParse(removeIdleValue, out var removeIdle))
                 {
-                    this.MaxLocal = maxLocal;
+                    RemoveIdleWorkers = removeIdle;
                 }
             }
         }
 
         /// <inheritdoc/>
         public override void PopulateGrainProperties(IServiceProvider services, Type grainClass, GrainType grainType, Dictionary<string, string> properties)
         {
-            properties[MaxLocalPropertyKey] = this.MaxLocal.ToString(CultureInfo.InvariantCulture);
+            properties[MaxLocalPropertyKey] = MaxLocal.ToString(CultureInfo.InvariantCulture);
+            properties[RemoveIdleWorkersPropertyKey] = RemoveIdleWorkers.ToString(CultureInfo.InvariantCulture);
 
             base.PopulateGrainProperties(services, grainClass, grainType, properties);
         }
     }
-}
+}

src/Orleans.Persistence.Memory/Storage/MemoryStorageWithLatency.cs

@@ -52,9 +52,9 @@ public MemoryGrainStorageWithLatency(
             MemoryStorageWithLatencyOptions options,
             ILoggerFactory loggerFactory,
             IGrainFactory grainFactory,
-            IGrainStorageSerializer defaultGrainStorageSerialzier)
+            IGrainStorageSerializer defaultGrainStorageSerializer)
         {
-            this.baseGranStorage = new MemoryGrainStorage(name, options, loggerFactory.CreateLogger<MemoryGrainStorage>(), grainFactory, defaultGrainStorageSerialzier);
+            this.baseGranStorage = new MemoryGrainStorage(name, options, loggerFactory.CreateLogger<MemoryGrainStorage>(), grainFactory, defaultGrainStorageSerializer);
             this.options = options;
         }
 
@@ -101,16 +101,24 @@ private async Task MakeFixedLatencyCall(Func<Task> action)
                 error = exc;
             }
 
-            if (sw.Elapsed < this.options.Latency)
+            do
             {
                 // Work out the remaining time to wait so that this operation exceeds the required Latency.
                 // Also adds an extra fudge factor to account for any system clock resolution edge cases.
                 var extraDelay = TimeSpan.FromTicks(
-                    this.options.Latency.Ticks - sw.Elapsed.Ticks + TimeSpan.TicksPerMillisecond /* round up */ );
+                     5 * TimeSpan.TicksPerMillisecond + this.options.Latency.Ticks - sw.Elapsed.Ticks);
 
-                await Task.Delay(extraDelay);
-            }
+                if (extraDelay > TimeSpan.Zero)
+                {
+                    await Task.Delay(extraDelay);
+                }
+                else
+                {
+                    break;
+                }
+            } while (true);
 
+            Debug.Assert(sw.Elapsed >= this.options.Latency, "sw.Elapsed >= this.options.Latency");
             if (error != null)
             {
                 // Wrap in AggregateException so that the original error stack trace is preserved.

src/Orleans.Runtime/Catalog/GrainTypeSharedContext.cs

@@ -33,6 +33,7 @@ public GrainTypeSharedContext(
         IOptions<SiloMessagingOptions> messagingOptions,
         IOptions<GrainCollectionOptions> collectionOptions,
         IOptions<SchedulingOptions> schedulingOptions,
+        IOptions<StatelessWorkerOptions> statelessWorkerOptions,
         IGrainRuntime grainRuntime,
         ILoggerFactory loggerFactory,
         GrainReferenceActivator grainReferenceActivator,
@@ -56,6 +57,7 @@ public GrainTypeSharedContext(
         var grainDirectoryResolver = serviceProvider.GetRequiredService<GrainDirectoryResolver>();
         GrainDirectory = PlacementStrategy.IsUsingGrainDirectory ? grainDirectoryResolver.Resolve(grainType) : null;
         SchedulingOptions = schedulingOptions.Value;
+        StatelessWorkerOptions = statelessWorkerOptions.Value;
         Runtime = grainRuntime;
         MigrationManager = _serviceProvider.GetService<IActivationMigrationManager>();
 
@@ -181,6 +183,11 @@ public void SetComponent<TComponent>(TComponent? instance)
     /// </summary>
     public SchedulingOptions SchedulingOptions { get; }
 
+    /// <summary>
+    /// Gets the stateless worker options.
+    /// </summary>
+    public StatelessWorkerOptions StatelessWorkerOptions { get; }
+
     /// <summary>
     /// Gets the grain runtime.
     /// </summary>