From 90b1cfd7cc55a1a7cafacbf51620ec60f9980a77 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Fri, 24 Feb 2023 11:09:23 -0600 Subject: [PATCH 1/3] optimize `ActorMaterializer` to only inject HOCON configuration once creating lots of materializer instances (i.e. in actors that own their own streams) results in a tremendous amount of unreclaimable HOCON-related memory - on the order of 11Gb in https://github.com/Aaronontheweb/AkkaSqlQueryCrushTest This tries to avoid injecting the top level Akka.Streams fallback every time a materializer is used. In addition to that, we try to cache the default materializer settings parsed from the `ActorSystem`s HOCON if none are provided. --- src/core/Akka.Streams/ActorMaterializer.cs | 23 ++++++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/src/core/Akka.Streams/ActorMaterializer.cs b/src/core/Akka.Streams/ActorMaterializer.cs index b71e1f05938..4dbeb667cec 100644 --- a/src/core/Akka.Streams/ActorMaterializer.cs +++ b/src/core/Akka.Streams/ActorMaterializer.cs @@ -42,6 +42,18 @@ public static Config DefaultConfig() => DefaultMaterializerConfig; #region static + + /// + /// Injecting the top-level Materializer HOCON configuration over and over again is expensive, so we want to avoid + /// doing it each time a materializer is instantiated. This flag will be set to true once the configuration has been + /// injected the first time. + /// + private static volatile bool _injectedConfig = false; + + /// + /// Cache the default materializer settings so we don't constantly parse them + /// + private static ActorMaterializerSettings _defaultSettings = null; /// /// @@ -75,9 +87,16 @@ public static ActorMaterializer Create(IActorRefFactory context, ActorMaterializ var haveShutDown = new AtomicBoolean(); var system = ActorSystemOf(context); - system.Settings.InjectTopLevelFallback(DefaultConfig()); + if(!_injectedConfig) + { + // Inject the top-level fallback config for the Materializer once, and only once. + // This is a performance optimization to avoid having to do this on every materialization. + system.Settings.InjectTopLevelFallback(DefaultConfig()); + _injectedConfig = true; + } - settings = settings ?? ActorMaterializerSettings.Create(system); + // use the default settings if none have been passed in + settings ??= (_defaultSettings ??= ActorMaterializerSettings.Create(system)); return new ActorMaterializerImpl( system: system, From 7f9411ed3248d85c36cb7ccaa3ca54313a019391 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Fri, 24 Feb 2023 11:10:08 -0600 Subject: [PATCH 2/3] modernized syntax --- src/core/Akka.Streams/ActorMaterializer.cs | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/src/core/Akka.Streams/ActorMaterializer.cs b/src/core/Akka.Streams/ActorMaterializer.cs index 4dbeb667cec..af27196c8c9 100644 --- a/src/core/Akka.Streams/ActorMaterializer.cs +++ b/src/core/Akka.Streams/ActorMaterializer.cs @@ -109,14 +109,14 @@ public static ActorMaterializer Create(IActorRefFactory context, ActorMaterializ private static ActorSystem ActorSystemOf(IActorRefFactory context) { - if (context is ExtendedActorSystem) - return (ActorSystem)context; - if (context is IActorContext) - return ((IActorContext)context).System; - if (context == null) - throw new ArgumentNullException(nameof(context), "IActorRefFactory must be defined"); - - throw new ArgumentException($"ActorRefFactory context must be a ActorSystem or ActorContext, got [{context.GetType()}]"); + return context switch + { + ExtendedActorSystem system => system, + IActorContext actorContext => actorContext.System, + null => throw new ArgumentNullException(nameof(context), "IActorRefFactory must be defined"), + _ => throw new ArgumentException( + $"ActorRefFactory context must be a ActorSystem or ActorContext, got [{context.GetType()}]") + }; } #endregion From 73eb93e4559af9f5cd27e375854a458b7bfc8e49 Mon Sep 17 00:00:00 2001 From: Aaron Stannard Date: Fri, 24 Feb 2023 11:58:23 -0600 Subject: [PATCH 3/3] improve cache to detect multiple `ActorSystem` instances --- src/core/Akka.Streams/ActorMaterializer.cs | 26 +++++++++++++++++----- 1 file changed, 20 insertions(+), 6 deletions(-) diff --git a/src/core/Akka.Streams/ActorMaterializer.cs b/src/core/Akka.Streams/ActorMaterializer.cs index af27196c8c9..e4212faa6e8 100644 --- a/src/core/Akka.Streams/ActorMaterializer.cs +++ b/src/core/Akka.Streams/ActorMaterializer.cs @@ -6,7 +6,9 @@ //----------------------------------------------------------------------- using System; +using System.Collections.Concurrent; using System.Runtime.Serialization; +using System.Threading.Tasks; using Akka.Actor; using Akka.Configuration; using Akka.Dispatch; @@ -42,18 +44,18 @@ public static Config DefaultConfig() => DefaultMaterializerConfig; #region static - + /// /// Injecting the top-level Materializer HOCON configuration over and over again is expensive, so we want to avoid /// doing it each time a materializer is instantiated. This flag will be set to true once the configuration has been /// injected the first time. /// - private static volatile bool _injectedConfig = false; + private static readonly ConcurrentDictionary InjectedConfig = new(); /// /// Cache the default materializer settings so we don't constantly parse them /// - private static ActorMaterializerSettings _defaultSettings = null; + private static readonly ConcurrentDictionary DefaultSettings = new(); /// /// @@ -87,16 +89,28 @@ public static ActorMaterializer Create(IActorRefFactory context, ActorMaterializ var haveShutDown = new AtomicBoolean(); var system = ActorSystemOf(context); - if(!_injectedConfig) + if(!InjectedConfig.TryGetValue(system, out _) && InjectedConfig.TryAdd(system, true)) { // Inject the top-level fallback config for the Materializer once, and only once. // This is a performance optimization to avoid having to do this on every materialization. system.Settings.InjectTopLevelFallback(DefaultConfig()); - _injectedConfig = true; + + static async Task CleanUp(ActorSystem sys) + { + // remove ActorSystem from cache when it terminates so we don't leak memory + await sys.WhenTerminated.ConfigureAwait(false); + InjectedConfig.TryRemove(sys, out _); + DefaultSettings.TryRemove(sys, out _); + } + +#pragma warning disable CS4014 + CleanUp(system); +#pragma warning restore CS4014 + } // use the default settings if none have been passed in - settings ??= (_defaultSettings ??= ActorMaterializerSettings.Create(system)); + settings ??= DefaultSettings.GetOrAdd(system, ActorMaterializerSettings.Create); return new ActorMaterializerImpl( system: system,