diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 3d082d73d41..32b539b1109 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -1,3 +1,76 @@ +#### 1.4.25 September 08 2021 #### +**Maintenance Release for Akka.NET 1.4** +Akka.NET v1.4.25 includes some _significant_ performance improvements for Akka.Remote and a number of important bug fixes and improvements. + +**Bug Fixes and Improvements** +* [Akka.IO.Tcp: connecting to an unreachable DnsEndpoint never times out](https://github.com/akkadotnet/akka.net/issues/5154) +* [Akka.Actor: need to enforce `stdout-loglevel = off` all the way through ActorSystem lifecycle](https://github.com/akkadotnet/akka.net/issues/5246) +* [Akka.Actor: `Ask` should push unhandled answers into deadletter](https://github.com/akkadotnet/akka.net/pull/5259) +* [Akka.Routing: Make Router.Route` virtual](https://github.com/akkadotnet/akka.net/pull/5238) +* [Akka.Actor: Improve performance on `IActorRef.Child` API](https://github.com/akkadotnet/akka.net/pull/5242) - _signficantly_ improves performance of many Akka.NET functions, but includes a public API change on `IActorRef` that is source compatible but not necessarily binary-compatible. `IActorRef GetChild(System.Collections.Generic.IEnumerable name)` is now `IActorRef GetChild(System.Collections.Generic.IReadOnlyList name)`. This API is almost never called directly by user code (it's almost always called via the internals of the `ActorSystem` when resolving `ActorSelection`s or remote messages) so this change should be safe. +* [Akka.Actor: `IsNobody` throws NRE](https://github.com/akkadotnet/akka.net/issues/5213) +* [Akka.Cluster.Tools: singleton fix cleanup of overdue _removed members](https://github.com/akkadotnet/akka.net/pull/5229) +* [Akka.DistributedData: ddata exclude `Exiting` members in Read/Write `MajorityPlus`](https://github.com/akkadotnet/akka.net/pull/5227) + +**Performance Improvements** +Using our standard `RemotePingPong` benchmark, the difference between v1.4.24 and v1.4.24 is significant: + +_v1.4.24_ + +``` +OSVersion: Microsoft Windows NT 6.2.9200.0 +ProcessorCount: 16 +ClockSpeed: 0 MHZ +Actor Count: 32 +Messages sent/received per client: 200000 (2e5) +Is Server GC: True +Thread count: 111 + +Num clients, Total [msg], Msgs/sec, Total [ms] + 1, 200000, 96994, 2062.08 + 5, 1000000, 194818, 5133.93 + 10, 2000000, 198966, 10052.93 + 15, 3000000, 199455, 15041.56 + 20, 4000000, 198177, 20184.53 + 25, 5000000, 197613, 25302.80 + 30, 6000000, 197349, 30403.82 +``` + +_v1.4.25_ + +``` +OSVersion: Microsoft Windows NT 6.2.9200.0 +ProcessorCount: 16 +ClockSpeed: 0 MHZ +Actor Count: 32 +Messages sent/received per client: 200000 (2e5) +Is Server GC: True +Thread count: 111 + +Num clients, Total [msg], Msgs/sec, Total [ms] + 1, 200000, 130634, 1531.54 + 5, 1000000, 246975, 4049.20 + 10, 2000000, 244499, 8180.16 + 15, 3000000, 244978, 12246.39 + 20, 4000000, 245159, 16316.37 + 25, 5000000, 243333, 20548.09 + 30, 6000000, 241644, 24830.55 +``` + +This represents a 24% overall throughput improvement in Akka.Remote across the board. We have additional PRs staged that should get aggregate performance improvements above 40% for Akka.Remote over v1.4.24 but they didn't make it into the Akka.NET v1.4.25 release. + +You can [see the full set of changes introduced in Akka.NET v1.4.25 here](https://github.com/akkadotnet/akka.net/milestone/56?closed=1) + +| COMMITS | LOC+ | LOC- | AUTHOR | +| --- | --- | --- | --- | +| 32 | 1301 | 400 | Aaron Stannard | +| 4 | 358 | 184 | Andreas Dirnberger | +| 3 | 414 | 149 | Gregorius Soedharmo | +| 3 | 3 | 3 | dependabot[bot] | +| 2 | 43 | 10 | zbynek001 | +| 1 | 14 | 13 | tometchy | +| 1 | 139 | 3 | carlcamilleri | + #### 1.4.24 August 17 2021 #### **Maintenance Release for Akka.NET 1.4** diff --git a/docs/articles/utilities/logging.md b/docs/articles/utilities/logging.md index 055b7abe8d5..097212a5499 100644 --- a/docs/articles/utilities/logging.md +++ b/docs/articles/utilities/logging.md @@ -25,6 +25,28 @@ Akka.NET comes with two built in loggers. * __StandardOutLogger__ * __BusLogging__ +### StandardOutLogger +`StandardOutLogger` is considered as a minimal logger and implements the `MinimalLogger` abstract +class. Its job is simply to output all `LogEvent`s emitted by the `EventBus` onto the console. +Since it is not an actual actor, ie. it doesn't need the `ActorSystem` to operate, it is also +used to log other loggers activity at the very start and very end of the `ActorSystem` life cycle. +You can change the minimal logger start and end life cycle behaviour by changing the +`akka.stdout-loglevel` HOCON settings to `OFF` if you do not need these feature in your application. + +### Advanced MinimalLogger Setup +You can also replace `StandardOutLogger` by making your own logger class with an empty constructor +that inherits/implements the `MinimalLogger` abstract class and passing the fully qualified class +name into the `akka.stdout-logger-class` HOCON settings. + +> [!WARNING] +> Be aware that `MinimalLogger` implementations are __NOT__ real actors and will __NOT__ have any +> access to the `ActorSystem` and all of its extensions. All logging done inside a `MinimalLogger` +> have to be done in as simple as possible manner since it is used to log how other loggers are +> behaving at the very start and very end of the `ActorSystem` life cycle. +> +> Note that `MinimalLogger` are __NOT__ interchangeable with other Akka.NET loggers and there can +> only be one `MinimalLogger` registered with the `ActorSystem` in the HOCON settings. + ## Contrib Loggers These loggers are also available as separate nuget packages @@ -64,17 +86,18 @@ akka { ``` ## Example configuration ```hocon -akka { - stdout-loglevel = DEBUG - loglevel = DEBUG - log-config-on-start = on - actor { - debug { - receive = on - autoreceive = on - lifecycle = on - event-stream = on - unhandled = on - } - } +akka { + stdout-loglevel = DEBUG + loglevel = DEBUG + log-config-on-start = on + actor { + debug { + receive = on + autoreceive = on + lifecycle = on + event-stream = on + unhandled = on + } + } +} ``` diff --git a/src/Akka.sln b/src/Akka.sln index 52dbe1dacd7..bf0782f7236 100644 --- a/src/Akka.sln +++ b/src/Akka.sln @@ -246,6 +246,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "SerializationBenchmarks", " EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "DDataStressTest", "examples\Cluster\DData\DDataStressTest\DDataStressTest.csproj", "{44B3DDD6-6103-4E8F-8AC2-0F4BA3CF6B50}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Akka.Cluster.Benchmarks", "benchmark\Akka.Cluster.Benchmarks\Akka.Cluster.Benchmarks.csproj", "{3CEBB0AE-6A88-4C32-A1D3-A8FB1E7E236B}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -1137,6 +1139,18 @@ Global {44B3DDD6-6103-4E8F-8AC2-0F4BA3CF6B50}.Release|x64.Build.0 = Release|Any CPU {44B3DDD6-6103-4E8F-8AC2-0F4BA3CF6B50}.Release|x86.ActiveCfg = Release|Any CPU {44B3DDD6-6103-4E8F-8AC2-0F4BA3CF6B50}.Release|x86.Build.0 = Release|Any CPU + {3CEBB0AE-6A88-4C32-A1D3-A8FB1E7E236B}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {3CEBB0AE-6A88-4C32-A1D3-A8FB1E7E236B}.Debug|Any CPU.Build.0 = Debug|Any CPU + {3CEBB0AE-6A88-4C32-A1D3-A8FB1E7E236B}.Debug|x64.ActiveCfg = Debug|Any CPU + {3CEBB0AE-6A88-4C32-A1D3-A8FB1E7E236B}.Debug|x64.Build.0 = Debug|Any CPU + {3CEBB0AE-6A88-4C32-A1D3-A8FB1E7E236B}.Debug|x86.ActiveCfg = Debug|Any CPU + {3CEBB0AE-6A88-4C32-A1D3-A8FB1E7E236B}.Debug|x86.Build.0 = Debug|Any CPU + {3CEBB0AE-6A88-4C32-A1D3-A8FB1E7E236B}.Release|Any CPU.ActiveCfg = Release|Any CPU + {3CEBB0AE-6A88-4C32-A1D3-A8FB1E7E236B}.Release|Any CPU.Build.0 = Release|Any CPU + {3CEBB0AE-6A88-4C32-A1D3-A8FB1E7E236B}.Release|x64.ActiveCfg = Release|Any CPU + {3CEBB0AE-6A88-4C32-A1D3-A8FB1E7E236B}.Release|x64.Build.0 = Release|Any CPU + {3CEBB0AE-6A88-4C32-A1D3-A8FB1E7E236B}.Release|x86.ActiveCfg = Release|Any CPU + {3CEBB0AE-6A88-4C32-A1D3-A8FB1E7E236B}.Release|x86.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -1245,6 +1259,7 @@ Global {D62F4AD6-318F-4ECC-B875-83FA9933A81B} = {162F5991-EA57-4221-9B70-F9B6FEC18036} {2E4B9584-42CC-4D17-B719-9F462B16C94D} = {73108242-625A-4D7B-AA09-63375DBAE464} {44B3DDD6-6103-4E8F-8AC2-0F4BA3CF6B50} = {C50E1A9E-820C-4E75-AE39-6F96A99AC4A7} + {3CEBB0AE-6A88-4C32-A1D3-A8FB1E7E236B} = {73108242-625A-4D7B-AA09-63375DBAE464} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {03AD8E21-7507-4E68-A4E9-F4A7E7273164} diff --git a/src/benchmark/Akka.Benchmarks/Actor/ActorRefBenchmarks.cs b/src/benchmark/Akka.Benchmarks/Actor/ActorRefBenchmarks.cs new file mode 100644 index 00000000000..60a5ecc9c9f --- /dev/null +++ b/src/benchmark/Akka.Benchmarks/Actor/ActorRefBenchmarks.cs @@ -0,0 +1,67 @@ +// //----------------------------------------------------------------------- +// // +// // Copyright (C) 2009-2021 Lightbend Inc. +// // Copyright (C) 2013-2021 .NET Foundation +// // +// //----------------------------------------------------------------------- + +using System; +using System.Threading.Tasks; +using Akka.Actor; +using Akka.Benchmarks.Configurations; +using BenchmarkDotNet.Attributes; + +namespace Akka.Benchmarks.Actor +{ + [Config(typeof(MicroBenchmarkConfig))] // need memory diagnosis + public class ActorRefBenchmarks + { + [Params(10000)] + public int Iterations { get; set; } + private TimeSpan _timeout; + private ActorSystem _system; + private IActorRef _echo; + private IActorRef _echo2; + + [GlobalSetup] + public void Setup() + { + _timeout = TimeSpan.FromMinutes(1); + _system = ActorSystem.Create("system"); + _echo = _system.ActorOf(Props.Create(() => new EchoActor()), "echo"); + _echo2 = _system.ActorOf(Props.Create(() => new EchoActor()), "echo2"); + } + + [Benchmark] + public int ActorRefGetHashCode() + { + return _echo.GetHashCode(); + } + + [Benchmark] + public bool ActorRefEqualsSelf() + { + return _echo.Equals(_echo); + } + + [Benchmark] + public bool ActorRefEqualsSomeoneElse() + { + return _echo.Equals(_echo2); + } + + [GlobalCleanup] + public void Cleanup() + { + _system.Terminate().Wait(); + } + + public class EchoActor : UntypedActor + { + protected override void OnReceive(object message) + { + Sender.Tell(message); + } + } + } +} \ No newline at end of file diff --git a/src/benchmark/Akka.Benchmarks/Actor/GetChildBenchmark.cs b/src/benchmark/Akka.Benchmarks/Actor/GetChildBenchmark.cs new file mode 100644 index 00000000000..95a8318a188 --- /dev/null +++ b/src/benchmark/Akka.Benchmarks/Actor/GetChildBenchmark.cs @@ -0,0 +1,159 @@ +// //----------------------------------------------------------------------- +// // +// // Copyright (C) 2009-2021 Lightbend Inc. +// // Copyright (C) 2013-2021 .NET Foundation +// // +// //----------------------------------------------------------------------- + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using Akka.Actor; +using Akka.Benchmarks.Configurations; +using BenchmarkDotNet.Attributes; +using BenchmarkDotNet.Engines; +using FluentAssertions; + +namespace Akka.Benchmarks.Actor +{ + /// + /// Used to measure how quickly an call can be executed in the wild. + /// + [Config(typeof(MicroBenchmarkConfig))] + public class GetChildBenchmark + { + #region classes + public sealed class Child : UntypedActor + { + protected override void OnReceive(object message) + { + + } + + protected override void PreStart() + { + if (Self.Path.Name.Length > 1) + { + // recursively create children using the previous name segments + var nextName = new string(Self.Path.Name.Skip(1).ToArray()); + Context.ActorOf(Props.Create(() => new Child()), nextName); + } + } + } + + public sealed class ActorWithChild : UntypedActor + { + public sealed class Get + { + public Get(string name) + { + Name = name; + } + + public string Name { get; } + } + + public sealed class Create + { + public Create(string name) + { + Name = name; + } + + public string Name { get; } + } + + protected override void OnReceive(object message) + { + switch (message) + { + case Get g: + { + var child = Context.Child(g.Name); + Sender.Tell(child); + break; + } + case Create c: + { + var child = Context.ActorOf(Props.Create(() => new Child()), c.Name); + Sender.Tell(child); + break; + } + default: + Unhandled(message); + break; + } + } + } + + #endregion + + private TimeSpan _timeout; + private ActorSystem _system; + private IActorRef _parentActor; + + private ActorWithChild.Get _getMessage = new ActorWithChild.Get("food"); + private ActorWithChild.Create _createMessage = new ActorWithChild.Create("food"); + + private IActorContext _cell; + private RepointableActorRef _repointableActorRef; + private LocalActorRef _localActorRef; + private VirtualPathContainer _virtualPathContainer; + + private List _rpChildQueryPath = new List() { "food", "ood", "od" }; + private List _lclChildQueryPath = new List() { "ood", "od", "d" }; + private List _virtualPathContainerQueryPath = new List() { "foo" }; + + [GlobalSetup] + public async Task Setup() + { + _timeout = TimeSpan.FromMinutes(1); + _system = ActorSystem.Create("system"); + _parentActor = _system.ActorOf(Props.Create(() => new ActorWithChild()), "parent"); + _localActorRef = (LocalActorRef)await _parentActor.Ask(_createMessage, _timeout); + + _cell = _parentActor.As().Underlying.As(); + _repointableActorRef = (RepointableActorRef)_parentActor; + + var exp = _system.As(); + + var vPath = exp.Guardian.Path / "testTemp"; + _virtualPathContainer = + new VirtualPathContainer(exp.Provider, vPath, exp.Guardian, exp.Log); + + _virtualPathContainer.AddChild("foo", + new EmptyLocalActorRef(exp.Provider, vPath / "foo", exp.EventStream)); + } + + [Benchmark] + public void ResolveChild() + { + _cell.Child(_getMessage.Name); + } + + [Benchmark] + public void Resolve3DeepChildRepointableActorRef() + { + _repointableActorRef.GetChild(_rpChildQueryPath); + } + + [Benchmark] + public void Resolve3DeepChildLocalActorRef() + { + _localActorRef.GetChild(_lclChildQueryPath); + } + + [Benchmark] + public void ResolveVirtualPathContainer() + { + _virtualPathContainer.GetChild(_virtualPathContainerQueryPath); + } + + [GlobalCleanup] + public void Cleanup() + { + _system.Terminate().Wait(); + } + } +} \ No newline at end of file diff --git a/src/benchmark/Akka.Benchmarks/Akka.Benchmarks.csproj b/src/benchmark/Akka.Benchmarks/Akka.Benchmarks.csproj index 1c53c562aaa..bb7f7c03c95 100644 --- a/src/benchmark/Akka.Benchmarks/Akka.Benchmarks.csproj +++ b/src/benchmark/Akka.Benchmarks/Akka.Benchmarks.csproj @@ -3,13 +3,11 @@ Exe - $(NetCoreTestVersion) + $(NetTestVersion);$(NetCoreTestVersion) - - - + diff --git a/src/benchmark/Akka.Benchmarks/Remoting/LruBoundedCacheBenchmarks.cs b/src/benchmark/Akka.Benchmarks/Remoting/LruBoundedCacheBenchmarks.cs new file mode 100644 index 00000000000..d01ca7eda69 --- /dev/null +++ b/src/benchmark/Akka.Benchmarks/Remoting/LruBoundedCacheBenchmarks.cs @@ -0,0 +1,108 @@ +// //----------------------------------------------------------------------- +// // +// // Copyright (C) 2009-2021 Lightbend Inc. +// // Copyright (C) 2013-2021 .NET Foundation +// // +// //----------------------------------------------------------------------- + +using System.Threading; +using System.Threading.Tasks; +using Akka.Actor; +using Akka.Actor.Dsl; +using Akka.Benchmarks.Configurations; +using Akka.Configuration; +using Akka.Remote.Serialization; +using Akka.Util; +using BenchmarkDotNet.Attributes; + +namespace Akka.Benchmarks.Remoting +{ + [Config(typeof(MicroBenchmarkConfig))] + [SimpleJob( invocationCount: 10_000_000)] + public class LruBoundedCacheBenchmarks + { + private ActorSystem _sys1; + private Config _config = @"akka.actor.provider = remote + akka.remote.dot-netty.tcp.port = 0"; + + private ActorRefResolveThreadLocalCache _resolveCache; + private ActorPathThreadLocalCache _pathCache; + private AddressThreadLocalCache _addressCache; + + private string _cacheMissPath; + private IActorRef _cacheMissActorRef; + + private string _cacheHitPath; + private int _cacheHitPathCount = 0; + private IActorRef _cacheHitActorRef; + + private Address _addr1; + private string _addr1String; + + [GlobalSetup] + public async Task Setup() + { + _sys1 = ActorSystem.Create("BenchSys", _config); + _resolveCache = ActorRefResolveThreadLocalCache.For(_sys1); + _pathCache = ActorPathThreadLocalCache.For(_sys1); + _addressCache = AddressThreadLocalCache.For(_sys1); + + var es = (ExtendedActorSystem)_sys1; + _addr1 = es.Provider.DefaultAddress; + _addr1String = _addr1.ToString(); + + var name = "target" + ++_cacheHitPathCount; + _cacheHitActorRef = _sys1.ActorOf(act => + { + act.ReceiveAny((o, context) => context.Sender.Tell(context.Sender)); + }, name); + + _cacheMissActorRef = await _cacheHitActorRef.Ask("hit", CancellationToken.None); + + _cacheHitPath = _cacheHitActorRef.Path.ToString(); + _cacheMissPath = _cacheMissActorRef.Path.ToString(); + } + + [IterationSetup] + public void IterationSetup() + { + _cacheMissPath = $"/user/f/{_cacheHitPathCount++}"; + } + + [Benchmark] + public void ActorRefResolveMissBenchmark() + { + _resolveCache.Cache.GetOrCompute("/user/ignore"); + } + + [Benchmark] + public void ActorRefResolveHitBenchmark() + { + _resolveCache.Cache.GetOrCompute(_cacheHitPath); + } + + [Benchmark] + public void AddressHitBenchmark() + { + _addressCache.Cache.GetOrCompute(_addr1String); + } + + [Benchmark] + public void ActorPathCacheHitBenchmark() + { + _pathCache.Cache.GetOrCompute(_cacheHitPath); + } + + [Benchmark] + public void ActorPathCacheMissBenchmark() + { + _pathCache.Cache.GetOrCompute(_cacheMissPath); + } + + [GlobalCleanup] + public async Task Cleanup() + { + await _sys1.Terminate(); + } + } +} \ No newline at end of file diff --git a/src/benchmark/Akka.Cluster.Benchmarks/Akka.Cluster.Benchmarks.csproj b/src/benchmark/Akka.Cluster.Benchmarks/Akka.Cluster.Benchmarks.csproj new file mode 100644 index 00000000000..21ccbdcb825 --- /dev/null +++ b/src/benchmark/Akka.Cluster.Benchmarks/Akka.Cluster.Benchmarks.csproj @@ -0,0 +1,23 @@ + + + + Exe + $(NetTestVersion);$(NetCoreTestVersion) + + + + + + + + + + + + + + Configs.cs + + + + diff --git a/src/benchmark/Akka.Cluster.Benchmarks/Program.cs b/src/benchmark/Akka.Cluster.Benchmarks/Program.cs new file mode 100644 index 00000000000..df08f62bf68 --- /dev/null +++ b/src/benchmark/Akka.Cluster.Benchmarks/Program.cs @@ -0,0 +1,20 @@ +using System; +using System.Reflection; +using BenchmarkDotNet.Configs; +using BenchmarkDotNet.Running; + +namespace Akka.Cluster.Benchmarks +{ + class Program + { + static void Main(string[] args) + { +#if (DEBUG) + BenchmarkSwitcher.FromAssembly(Assembly.GetExecutingAssembly()).Run(args, new DebugInProcessConfig()); +#else + BenchmarkSwitcher.FromAssembly(Assembly.GetExecutingAssembly()).Run(args); +#endif + + } + } +} \ No newline at end of file diff --git a/src/benchmark/Akka.Cluster.Benchmarks/Sharding/HashCodeMessageExtractorBenchmarks.cs b/src/benchmark/Akka.Cluster.Benchmarks/Sharding/HashCodeMessageExtractorBenchmarks.cs new file mode 100644 index 00000000000..ee8d2a9d1ba --- /dev/null +++ b/src/benchmark/Akka.Cluster.Benchmarks/Sharding/HashCodeMessageExtractorBenchmarks.cs @@ -0,0 +1,37 @@ +//----------------------------------------------------------------------- +// +// Copyright (C) 2009-2021 Lightbend Inc. +// Copyright (C) 2013-2021 .NET Foundation +// +//----------------------------------------------------------------------- + +using Akka.Benchmarks.Configurations; +using Akka.Cluster.Sharding; +using BenchmarkDotNet.Attributes; + +namespace Akka.Cluster.Benchmarks.Sharding +{ + [Config(typeof(MicroBenchmarkConfig))] + public class HashCodeMessageExtractorBenchmarks + { + private ShardingEnvelope _m1 = new ShardingEnvelope("foo", 1); + private ShardedMessage _m2 = new ShardedMessage("foo", 1); + private IMessageExtractor _extractor = new ShardMessageExtractor(); + + [Benchmark] + public void RouteShardEnvelope() + { + _extractor.EntityId(_m1); + _extractor.EntityMessage(_m1); + _extractor.ShardId(_m1); + } + + [Benchmark] + public void RouteTypedMessage() + { + _extractor.EntityId(_m2); + _extractor.EntityMessage(_m2); + _extractor.ShardId(_m2); + } + } +} \ No newline at end of file diff --git a/src/benchmark/Akka.Cluster.Benchmarks/Sharding/ShardMessageRoutingBenchmarks.cs b/src/benchmark/Akka.Cluster.Benchmarks/Sharding/ShardMessageRoutingBenchmarks.cs new file mode 100644 index 00000000000..c2fc8d0ead0 --- /dev/null +++ b/src/benchmark/Akka.Cluster.Benchmarks/Sharding/ShardMessageRoutingBenchmarks.cs @@ -0,0 +1,194 @@ +//----------------------------------------------------------------------- +// +// Copyright (C) 2009-2021 Lightbend Inc. +// Copyright (C) 2013-2021 .NET Foundation +// +//----------------------------------------------------------------------- + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using Akka.Actor; +using Akka.Benchmarks.Configurations; +using Akka.Cluster.Sharding; +using Akka.Routing; +using BenchmarkDotNet.Attributes; +using static Akka.Cluster.Benchmarks.Sharding.ShardingHelper; + +namespace Akka.Cluster.Benchmarks.Sharding +{ + [Config(typeof(MonitoringConfig))] + public class ShardMessageRoutingBenchmarks + { + [Params(StateStoreMode.Persistence, StateStoreMode.DData)] + public StateStoreMode StateMode; + + [Params(10000)] + public int MsgCount; + + public int BatchSize = 20; + + private ActorSystem _sys1; + private ActorSystem _sys2; + + private IActorRef _shardRegion1; + private IActorRef _shardRegion2; + private IActorRef _localRouter; + + private string _entityOnSys1; + private string _entityOnSys2; + + private ShardedMessage _messageToSys1; + private ShardedMessage _messageToSys2; + + private IActorRef _batchActor; + private Task _batchComplete; + +#if (DEBUG) + [GlobalSetup] + public void Setup() + { + var config = StateMode switch + { + StateStoreMode.Persistence => CreatePersistenceConfig(), + StateStoreMode.DData => CreateDDataConfig(), + _ => null + }; + + _sys1 = ActorSystem.Create("BenchSys", config); + _sys2 = ActorSystem.Create("BenchSys", config); + + var c1 = Cluster.Get(_sys1); + var c2 = Cluster.Get(_sys2); + + c1.JoinAsync(c1.SelfAddress).Wait(); + c2.JoinAsync(c1.SelfAddress).Wait(); + + _shardRegion1 = StartShardRegion(_sys1); + _shardRegion2 = StartShardRegion(_sys2); + + _localRouter = _sys1.ActorOf(Props.Create(_shardRegion1).WithRouter(new RoundRobinPool(50))); + + var s1Asks = new List>(20); + var s2Asks = new List>(20); + + foreach (var i in Enumerable.Range(0, 20)) + { + s1Asks.Add(_shardRegion1.Ask(new ShardingEnvelope(i.ToString(), ShardedEntityActor.Resolve.Instance), TimeSpan.FromSeconds(3))); + s2Asks.Add(_shardRegion2.Ask(new ShardingEnvelope(i.ToString(), ShardedEntityActor.Resolve.Instance), TimeSpan.FromSeconds(3))); + } + + // wait for all Ask operations to complete + Task.WhenAll(s1Asks.Concat(s2Asks)).Wait(); + + _entityOnSys2 = s1Asks.First(x => x.Result.Addr.Equals(c2.SelfAddress)).Result.EntityId; + _entityOnSys1 = s2Asks.First(x => x.Result.Addr.Equals(c1.SelfAddress)).Result.EntityId; + + _messageToSys1 = new ShardedMessage(_entityOnSys1, 10); + _messageToSys2 = new ShardedMessage(_entityOnSys2, 10); + } +#else + [GlobalSetup] + public async Task Setup() + { + var config = StateMode switch + { + StateStoreMode.Persistence => CreatePersistenceConfig(), + StateStoreMode.DData => CreateDDataConfig(), + _ => null + }; + + _sys1 = ActorSystem.Create("BenchSys", config); + _sys2 = ActorSystem.Create("BenchSys", config); + + var c1 = Cluster.Get(_sys1); + var c2 = Cluster.Get(_sys2); + + await c1.JoinAsync(c1.SelfAddress); + await c2.JoinAsync(c1.SelfAddress); + + _shardRegion1 = StartShardRegion(_sys1); + _shardRegion2 = StartShardRegion(_sys2); + + _localRouter = _sys1.ActorOf(Props.Create(_shardRegion1).WithRouter(new RoundRobinPool(1000))); + + var s1Asks = new List>(20); + var s2Asks = new List>(20); + + foreach (var i in Enumerable.Range(0, 20)) + { + s1Asks.Add(_shardRegion1.Ask(new ShardingEnvelope(i.ToString(), ShardedEntityActor.Resolve.Instance), TimeSpan.FromSeconds(3))); + s2Asks.Add(_shardRegion2.Ask(new ShardingEnvelope(i.ToString(), ShardedEntityActor.Resolve.Instance), TimeSpan.FromSeconds(3))); + } + + // wait for all Ask operations to complete + await Task.WhenAll(s1Asks.Concat(s2Asks)); + + _entityOnSys2 = s1Asks.First(x => x.Result.Addr.Equals(c2.SelfAddress)).Result.EntityId; + _entityOnSys1 = s2Asks.First(x => x.Result.Addr.Equals(c1.SelfAddress)).Result.EntityId; + + _messageToSys1 = new ShardedMessage(_entityOnSys1, 10); + _messageToSys2 = new ShardedMessage(_entityOnSys2, 10); + } +#endif + + [IterationSetup] + public void PerIteration() + { + var tcs = new TaskCompletionSource(); + _batchComplete = tcs.Task; + _batchActor = _sys1.ActorOf(Props.Create(() => new BulkSendActor(tcs, MsgCount))); + } + + [Benchmark] + public async Task SingleRequestResponseToLocalEntity() + { + for (var i = 0; i < MsgCount; i++) + await _shardRegion1.Ask(_messageToSys1); + } + + [Benchmark] + public async Task StreamingToLocalEntity() + { + _batchActor.Tell(new BulkSendActor.BeginSend(_messageToSys1, _shardRegion1, BatchSize)); + await _batchComplete; + } + + [Benchmark] + public async Task SingleRequestResponseToRemoteEntity() + { + for (var i = 0; i < MsgCount; i++) + await _shardRegion1.Ask(_messageToSys2); + } + + + [Benchmark] + public async Task SingleRequestResponseToRemoteEntityWithLocalProxy() + { + for (var i = 0; i < MsgCount; i++) + await _localRouter.Ask(new SendShardedMessage(_messageToSys2.EntityId, _messageToSys2)); + } + + [Benchmark] + public async Task StreamingToRemoteEntity() + { + _batchActor.Tell(new BulkSendActor.BeginSend(_messageToSys2, _shardRegion1, BatchSize)); + await _batchComplete; + } + + [IterationCleanup] + public void IterationCleanup() + { + _sys1.Stop(_batchActor); + } + + [GlobalCleanup] + public async Task Cleanup() + { + var t1 = _sys1.Terminate(); + var t2 = _sys2.Terminate(); + await Task.WhenAll(t1, t2); + } + } +} \ No newline at end of file diff --git a/src/benchmark/Akka.Cluster.Benchmarks/Sharding/ShardSpawnBenchmarks.cs b/src/benchmark/Akka.Cluster.Benchmarks/Sharding/ShardSpawnBenchmarks.cs new file mode 100644 index 00000000000..18c904728a2 --- /dev/null +++ b/src/benchmark/Akka.Cluster.Benchmarks/Sharding/ShardSpawnBenchmarks.cs @@ -0,0 +1,93 @@ +// //----------------------------------------------------------------------- +// // +// // Copyright (C) 2009-2021 Lightbend Inc. +// // Copyright (C) 2013-2021 .NET Foundation +// // +// //----------------------------------------------------------------------- + +using System.Threading.Tasks; +using Akka.Actor; +using Akka.Benchmarks.Configurations; +using Akka.Cluster.Sharding; +using BenchmarkDotNet.Attributes; +using BenchmarkDotNet.Engines; +using static Akka.Cluster.Benchmarks.Sharding.ShardingHelper; + +namespace Akka.Cluster.Benchmarks.Sharding +{ + [Config(typeof(MonitoringConfig))] + [SimpleJob(RunStrategy.ColdStart, targetCount:1, warmupCount:0, launchCount:5)] + public class ShardSpawnBenchmarks + { + [Params(StateStoreMode.Persistence, StateStoreMode.DData)] + public StateStoreMode StateMode; + + [Params(1000)] + public int EntityCount; + + [Params(true, false)] + public bool RememberEntities; + + public int BatchSize = 20; + + private ActorSystem _sys1; + private ActorSystem _sys2; + + private IActorRef _shardRegion1; + private IActorRef _shardRegion2; + + public static int _shardRegionId = 0; + + + [GlobalSetup] + public async Task Setup() + { + var config = StateMode switch + { + StateStoreMode.Persistence => CreatePersistenceConfig(RememberEntities), + StateStoreMode.DData => CreateDDataConfig(RememberEntities), + _ => null + }; + + _sys1 = ActorSystem.Create("BenchSys", config); + _sys2 = ActorSystem.Create("BenchSys", config); + + var c1 = Cluster.Get(_sys1); + var c2 = Cluster.Get(_sys2); + + await c1.JoinAsync(c1.SelfAddress); + await c2.JoinAsync(c1.SelfAddress); + } + + [IterationSetup] + public void IterationSetup() + { + /* + * Create a new set of shard regions each time, so all of the shards are freshly allocated + * on each benchmark run. + */ + _shardRegion1 = StartShardRegion(_sys1, "entities" + _shardRegionId); + _shardRegion2 = StartShardRegion(_sys2, "entities" + _shardRegionId); + _shardRegionId++; + } + + [Benchmark] + public async Task SpawnEntities() + { + for (var i = 0; i < EntityCount; i++) + { + var msg = new ShardedMessage(i.ToString(), i); + await _shardRegion1.Ask(msg); + } + } + + [GlobalCleanup] + public async Task Cleanup() + { + var t2 = CoordinatedShutdown.Get(_sys2).Run(CoordinatedShutdown.ActorSystemTerminateReason.Instance); + var t1 = CoordinatedShutdown.Get(_sys1).Run(CoordinatedShutdown.ActorSystemTerminateReason.Instance); + + await Task.WhenAll(t1, t2); + } + } +} \ No newline at end of file diff --git a/src/benchmark/Akka.Cluster.Benchmarks/Sharding/ShardingInfrastructure.cs b/src/benchmark/Akka.Cluster.Benchmarks/Sharding/ShardingInfrastructure.cs new file mode 100644 index 00000000000..e016ea70052 --- /dev/null +++ b/src/benchmark/Akka.Cluster.Benchmarks/Sharding/ShardingInfrastructure.cs @@ -0,0 +1,267 @@ +// //----------------------------------------------------------------------- +// // +// // Copyright (C) 2009-2021 Lightbend Inc. +// // Copyright (C) 2013-2021 .NET Foundation +// // +// //----------------------------------------------------------------------- + +using System.Threading.Tasks; +using Akka.Actor; +using Akka.Cluster.Sharding; +using Akka.Configuration; +using Akka.Util.Internal; + +namespace Akka.Cluster.Benchmarks.Sharding +{ + public sealed class ShardedEntityActor : ReceiveActor + { + public sealed class Resolve + { + public static readonly Resolve Instance = new Resolve(); + private Resolve(){} + } + + public sealed class ResolveResp + { + public ResolveResp(string entityId, Address addr) + { + EntityId = entityId; + Addr = addr; + } + + public string EntityId { get; } + + public Address Addr { get; } + } + + public ShardedEntityActor() + { + Receive(e => + { + Sender.Tell(new ResolveResp(e.EntityId, Cluster.Get(Context.System).SelfAddress)); + }); + + ReceiveAny(_ => Sender.Tell(_)); + } + } + + + + public sealed class ShardedProxyEntityActor : ReceiveActor, IWithUnboundedStash + { + private IActorRef _shardRegion; + private IActorRef _sender; + + + + + + public ShardedProxyEntityActor(IActorRef shardRegion) + { + _shardRegion = shardRegion; + WaitRequest(); + } + + public void WaitRequest() + { + + Receive(e => + { + _sender = Sender; + _shardRegion.Tell(e.Message); + Become(WaitResult); + + }); + + ReceiveAny(x => { + Sender.Tell(x); + }); + + + } + + + public void WaitResult() + { + Receive((msg) => { + _sender.Tell(msg); + Stash.UnstashAll(); + Become(WaitRequest); + }); + + ReceiveAny(msg => + Stash.Stash() + ); + + + } + + public IStash Stash { get; set; } + + } + + + + public sealed class BulkSendActor : ReceiveActor + { + public sealed class BeginSend + { + public BeginSend(ShardedMessage msg, IActorRef target, int batchSize) + { + Msg = msg; + Target = target; + BatchSize = batchSize; + } + + public ShardedMessage Msg { get; } + + public IActorRef Target { get; } + + public int BatchSize { get; } + } + + private int _remaining; + + private readonly TaskCompletionSource _tcs; + + public BulkSendActor(TaskCompletionSource tcs, int remaining) + { + _tcs = tcs; + _remaining = remaining; + + Receive(b => + { + for (var i = 0; i < b.BatchSize; i++) + { + b.Target.Tell(b.Msg); + } + }); + + Receive(s => + { + if (--remaining > 0) + { + Sender.Tell(s); + } + else + { + Context.Stop(Self); // shut ourselves down + _tcs.TrySetResult(true); + } + }); + } + } + + public sealed class ShardedMessage + { + public ShardedMessage(string entityId, int message) + { + EntityId = entityId; + Message = message; + } + + public string EntityId { get; } + + public int Message { get; } + } + + + + public sealed class SendShardedMessage + { + public SendShardedMessage(string entityId, ShardedMessage message) + { + EntityId = entityId; + Message = message; + } + + public string EntityId { get; } + + public ShardedMessage Message { get; } + } + + + /// + /// Use a default even though it takes extra work to setup the benchmark + /// + public sealed class ShardMessageExtractor : HashCodeMessageExtractor + { + /// + /// We only ever run with a maximum of two nodes, so ~10 shards per node + /// + public ShardMessageExtractor(int shardCount = 20) : base(shardCount) + { + } + + public override string EntityId(object message) + { + if(message is ShardedMessage sharded) + { + return sharded.EntityId; + } + + if (message is ShardingEnvelope e) + { + return e.EntityId; + } + + return null; + } + } + + public static class ShardingHelper + { + public static AtomicCounter DbId = new AtomicCounter(0); + + internal static string BoolToToggle(bool val) + { + return val ? "on" : "off"; + } + + public static Config CreatePersistenceConfig(bool rememberEntities = false) + { + var connectionString = + "Filename=file:memdb-journal-" + DbId.IncrementAndGet() + ".db;Mode=Memory;Cache=Shared"; + var config = $@" + akka.actor.provider = cluster + akka.remote.dot-netty.tcp.port = 0 + akka.cluster.sharding.state-store-mode=persistence + akka.cluster.sharding.remember-entities = {BoolToToggle(rememberEntities)} + akka.persistence.journal.plugin = ""akka.persistence.journal.sqlite"" + akka.persistence.journal.sqlite {{ + class = ""Akka.Persistence.Sqlite.Journal.SqliteJournal, Akka.Persistence.Sqlite"" + auto-initialize = on + connection-string = ""{connectionString}"" + }} + akka.persistence.snapshot-store {{ + plugin = ""akka.persistence.snapshot-store.sqlite"" + sqlite {{ + class = ""Akka.Persistence.Sqlite.Snapshot.SqliteSnapshotStore, Akka.Persistence.Sqlite"" + auto-initialize = on + connection-string = ""{connectionString}"" + }} + }}"; + + return config; + } + + public static Config CreateDDataConfig(bool rememberEntities = false) + { + var config = $@" + akka.actor.provider = cluster + akka.remote.dot-netty.tcp.port = 0 + akka.cluster.sharding.state-store-mode=ddata + akka.cluster.sharding.remember-entities = {BoolToToggle(rememberEntities)}"; + + return config; + } + + public static IActorRef StartShardRegion(ActorSystem system, string entityName = "entities") + { + var props = Props.Create(() => new ShardedEntityActor()); + var sharding = ClusterSharding.Get(system); + return sharding.Start(entityName, s => props, ClusterShardingSettings.Create(system), + new ShardMessageExtractor()); + } + } +} \ No newline at end of file diff --git a/src/common.props b/src/common.props index 2ab4cd5d294..aeb909faf56 100644 --- a/src/common.props +++ b/src/common.props @@ -10,17 +10,18 @@ 2.4.1 - 16.10.0 + 16.11.0 0.11.1 [12.0.3,) 2.0.1 3.17.3 + 0.13.1 netcoreapp3.1 net5.0 net471 netstandard2.0 5.10.3 - 2.15.3 + 2.16.3 2.0.3 4.7.0 akka;actors;actor model;Akka;concurrency diff --git a/src/contrib/cluster/Akka.Cluster.Metrics/Routing/ClusterMetricsRouting.cs b/src/contrib/cluster/Akka.Cluster.Metrics/Routing/ClusterMetricsRouting.cs index d360d6ce80a..ffe4d5fb256 100644 --- a/src/contrib/cluster/Akka.Cluster.Metrics/Routing/ClusterMetricsRouting.cs +++ b/src/contrib/cluster/Akka.Cluster.Metrics/Routing/ClusterMetricsRouting.cs @@ -22,12 +22,17 @@ namespace Akka.Cluster.Metrics { + public interface IClusterMetricsRoutingLogic + { + void MetricsChanged(ClusterMetricsChanged @event); + } + /// /// Load balancing of messages to cluster nodes based on cluster metric data. /// /// It uses random selection of routees based on probabilities derived from the remaining capacity of corresponding node. /// - public sealed class AdaptiveLoadBalancingRoutingLogic : RoutingLogic + public sealed class AdaptiveLoadBalancingRoutingLogic : RoutingLogic, IClusterMetricsRoutingLogic { private readonly ActorSystem _system; private readonly IMetricsSelector _metricsSelector; @@ -176,7 +181,7 @@ public override Router CreateRouter(ActorSystem system) /// public override Props RoutingLogicController(RoutingLogic routingLogic) { - return Actor.Props.Create(() => new AdaptiveLoadBalancingMetricsListener(routingLogic as AdaptiveLoadBalancingRoutingLogic)); + return Actor.Props.Create(() => new AdaptiveLoadBalancingMetricsListener((IClusterMetricsRoutingLogic)routingLogic)); } /// @@ -328,7 +333,7 @@ public override Router CreateRouter(ActorSystem system) /// public override Props RoutingLogicController(RoutingLogic routingLogic) { - return Actor.Props.Create(() => new AdaptiveLoadBalancingMetricsListener(routingLogic as AdaptiveLoadBalancingRoutingLogic)); + return Actor.Props.Create(() => new AdaptiveLoadBalancingMetricsListener((IClusterMetricsRoutingLogic)routingLogic)); } /// @@ -392,10 +397,10 @@ public ISurrogated FromSurrogate(ActorSystem system) [InternalApi] public class AdaptiveLoadBalancingMetricsListener : ActorBase { - private readonly AdaptiveLoadBalancingRoutingLogic _routingLogic; + private readonly IClusterMetricsRoutingLogic _routingLogic; private readonly ClusterMetrics _extension = ClusterMetrics.Get(Context.System); - public AdaptiveLoadBalancingMetricsListener(AdaptiveLoadBalancingRoutingLogic routingLogic) + public AdaptiveLoadBalancingMetricsListener(IClusterMetricsRoutingLogic routingLogic) { _routingLogic = routingLogic; } diff --git a/src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/AsyncWriteProxyEx.cs b/src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/AsyncWriteProxyEx.cs index 730a383d749..a26fc94daab 100644 --- a/src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/AsyncWriteProxyEx.cs +++ b/src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/AsyncWriteProxyEx.cs @@ -161,7 +161,7 @@ protected override Task> WriteMessagesAsync(IEnumerabl if (_store == null) return StoreNotInitialized>(); - return _store.AskEx(sender => new WriteMessages(messages, sender, 1), Timeout) + return _store.Ask(sender => new WriteMessages(messages, sender, 1), Timeout, CancellationToken.None) .ContinueWith(r => { if (r.IsCanceled) @@ -197,7 +197,7 @@ protected override Task DeleteMessagesToAsync(string persistenceId, long toSeque var result = new TaskCompletionSource(); - _store.AskEx(sender => new DeleteMessagesTo(persistenceId, toSequenceNr, sender), Timeout).ContinueWith(r => + _store.Ask(sender => new DeleteMessagesTo(persistenceId, toSequenceNr, sender), Timeout, CancellationToken.None).ContinueWith(r => { if (r.IsFaulted) result.TrySetException(r.Exception); @@ -252,7 +252,7 @@ public override Task ReadHighestSequenceNrAsync(string persistenceId, long var result = new TaskCompletionSource(); - _store.AskEx(sender => new ReplayMessages(0, 0, 0, persistenceId, sender), Timeout) + _store.Ask(sender => new ReplayMessages(0, 0, 0, persistenceId, sender), Timeout, CancellationToken.None) .ContinueWith(t => { if (t.IsFaulted) @@ -358,110 +358,4 @@ protected override bool Receive(object message) return false; } } - - internal static class FuturesEx - { - public static Task AskEx(this ICanTell self, Func messageFactory, TimeSpan? timeout = null) - { - return self.AskEx(messageFactory, timeout, CancellationToken.None); - } - - public static Task AskEx(this ICanTell self, Func messageFactory, CancellationToken cancellationToken) - { - return self.AskEx(messageFactory, null, cancellationToken); - } - - public static Task AskEx(this ICanTell self, Func messageFactory, TimeSpan? timeout, CancellationToken cancellationToken) - { - return self.AskEx(messageFactory, timeout, cancellationToken); - } - - public static Task AskEx(this ICanTell self, Func messageFactory, TimeSpan? timeout = null) - { - return self.AskEx(messageFactory, timeout, CancellationToken.None); - } - - public static Task AskEx(this ICanTell self, Func messageFactory, CancellationToken cancellationToken) - { - return self.AskEx(messageFactory, null, cancellationToken); - } - - public static async Task AskEx(this ICanTell self, Func messageFactory, TimeSpan? timeout, CancellationToken cancellationToken) - { - IActorRefProvider provider = ResolveProvider(self); - if (provider == null) - throw new ArgumentException("Unable to resolve the target Provider", nameof(self)); - - return (T)await AskEx(self, messageFactory, provider, timeout, cancellationToken); - } - internal static IActorRefProvider ResolveProvider(ICanTell self) - { - if (InternalCurrentActorCellKeeper.Current != null) - return InternalCurrentActorCellKeeper.Current.SystemImpl.Provider; - - if (self is IInternalActorRef iar) - return iar.Provider; - - if (self is ActorSelection asel) - return ResolveProvider(asel.Anchor); - - return null; - } - - private static async Task AskEx(ICanTell self, Func messageFactory, IActorRefProvider provider, TimeSpan? timeout, CancellationToken cancellationToken) - { - var result = new TaskCompletionSource(); - - CancellationTokenSource timeoutCancellation = null; - timeout = timeout ?? provider.Settings.AskTimeout; - var ctrList = new List(2); - - if (timeout != Timeout.InfiniteTimeSpan && timeout.Value > default(TimeSpan)) - { - timeoutCancellation = new CancellationTokenSource(); - - ctrList.Add(timeoutCancellation.Token.Register(() => - { - result.TrySetException(new AskTimeoutException($"Timeout after {timeout} seconds")); - })); - - timeoutCancellation.CancelAfter(timeout.Value); - } - - if (cancellationToken.CanBeCanceled) - { - ctrList.Add(cancellationToken.Register(() => result.TrySetCanceled())); - } - - //create a new tempcontainer path - ActorPath path = provider.TempPath(); - - var future = new FutureActorRef(result, t => { }, path); - //The future actor needs to be registered in the temp container - provider.RegisterTempActor(future, path); - - self.Tell(messageFactory(future), future); - - try - { - return await result.Task; - } - finally - { - //callback to unregister from tempcontainer - - provider.UnregisterTempActor(path); - - for (var i = 0; i < ctrList.Count; i++) - { - ctrList[i].Dispose(); - } - - if (timeoutCancellation != null) - { - timeoutCancellation.Dispose(); - } - } - } - } } diff --git a/src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/ClusterShardingLeavingSpec.cs b/src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/ClusterShardingLeavingSpec.cs index d5ceded4f21..17560102de1 100644 --- a/src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/ClusterShardingLeavingSpec.cs +++ b/src/contrib/cluster/Akka.Cluster.Sharding.Tests.MultiNode/ClusterShardingLeavingSpec.cs @@ -193,15 +193,7 @@ private void Join(RoleName from, RoleName to) { Cluster.Join(Node(to).Address); StartSharding(); - Within(TimeSpan.FromSeconds(15), () => - { - AwaitAssert(() => - { - Cluster.State.Members.Should().Contain(i => i.UniqueAddress == Cluster.SelfUniqueAddress && i.Status == MemberStatus.Up); - }); - }); }, from); - EnterBarrier(from.Name + "-joined"); } private void StartSharding() @@ -261,6 +253,13 @@ public void ClusterSharding_with_leaving_member_should_join_cluster() Join(_config.Second, _config.First); Join(_config.Third, _config.First); Join(_config.Fourth, _config.First); + + AwaitAssert(() => + { + Cluster.State.Members.Count.Should().Be(Roles.Count); + Cluster.State.Members.Select(x => x.Status).ToImmutableHashSet().Should() + .BeEquivalentTo(ImmutableHashSet.Empty.Add(MemberStatus.Up)); + }); EnterBarrier("after-2"); }); @@ -281,6 +280,7 @@ public void ClusterSharding_with_leaving_member_should_initialize_shards() .ToImmutableDictionary(kv => kv.Key, kv => kv.Value); shardLocations.Tell(new Locations(locations)); + Sys.Log.Debug("Original locations: {0}", string.Join(",", locations.Select(x => $"{x.Key}->{x.Value}"))); }, _config.First); EnterBarrier("after-3"); } @@ -303,6 +303,8 @@ public void ClusterSharding_with_leaving_member_should__recover_after_leaving_co ExpectTerminated(region, TimeSpan.FromSeconds(15)); }, _config.First); EnterBarrier("stopped"); + + // more stress by not having the barrier here RunOn(() => { @@ -318,7 +320,14 @@ public void ClusterSharding_with_leaving_member_should__recover_after_leaving_co var r = kv.Value; region.Tell(new Ping(id), probe.Ref); if (r.Path.Address.Equals(firstAddress)) - probe.ExpectMsg(TimeSpan.FromSeconds(1)).Should().NotBe(r); + { + var newRef = probe.ExpectMsg(TimeSpan.FromSeconds(1)); + + // have to log before we assert + Sys.Log.Debug("Moved [{0}] from [{1}] to [{2}]", id, r, newRef); + newRef.Should().NotBe(r); + + } else probe.ExpectMsg(r, TimeSpan.FromSeconds(1)); // should not move } diff --git a/src/contrib/cluster/Akka.Cluster.Sharding/ClusterSharding.cs b/src/contrib/cluster/Akka.Cluster.Sharding/ClusterSharding.cs index c961f9d354c..44860d928e4 100644 --- a/src/contrib/cluster/Akka.Cluster.Sharding/ClusterSharding.cs +++ b/src/contrib/cluster/Akka.Cluster.Sharding/ClusterSharding.cs @@ -60,6 +60,8 @@ public abstract class HashCodeMessageExtractor : IMessageExtractor /// public readonly int MaxNumberOfShards; + private readonly Dictionary _cachedIds; + /// /// TBD /// @@ -67,6 +69,11 @@ public abstract class HashCodeMessageExtractor : IMessageExtractor protected HashCodeMessageExtractor(int maxNumberOfShards) { MaxNumberOfShards = maxNumberOfShards; + _cachedIds = new Dictionary(MaxNumberOfShards); + foreach (var c in Enumerable.Range(0, maxNumberOfShards)) + { + _cachedIds[c] = c.ToString(); + } } /// @@ -98,8 +105,8 @@ public virtual string ShardId(object message) id = se.EntityId; else id = EntityId(message); - - return (Math.Abs(MurmurHash.StringHash(id)) % MaxNumberOfShards).ToString(); + + return _cachedIds[(Math.Abs(MurmurHash.StringHash(id)) % MaxNumberOfShards)]; } } diff --git a/src/contrib/cluster/Akka.Cluster.Sharding/ShardRegion.cs b/src/contrib/cluster/Akka.Cluster.Sharding/ShardRegion.cs index 784271accba..678adc7e5af 100644 --- a/src/contrib/cluster/Akka.Cluster.Sharding/ShardRegion.cs +++ b/src/contrib/cluster/Akka.Cluster.Sharding/ShardRegion.cs @@ -974,7 +974,7 @@ private void SendGracefulShutdownToCoordinatorIfInProgress() if (GracefulShutdownInProgress) { var actorSelections = CoordinatorSelection; - Log.Debug("Sending graceful shutdown to {0}", actorSelections); + Log.Debug("Sending graceful shutdown to {0}", string.Join(",", actorSelections.Select(c => c.ToString()))); actorSelections.ForEach(c => c.Tell(new PersistentShardCoordinator.GracefulShutdownRequest(Self))); } } diff --git a/src/contrib/cluster/Akka.Cluster.Tools/Singleton/ClusterSingletonManager.cs b/src/contrib/cluster/Akka.Cluster.Tools/Singleton/ClusterSingletonManager.cs index 074060b5ddd..2e85b93d4bb 100644 --- a/src/contrib/cluster/Akka.Cluster.Tools/Singleton/ClusterSingletonManager.cs +++ b/src/contrib/cluster/Akka.Cluster.Tools/Singleton/ClusterSingletonManager.cs @@ -751,7 +751,7 @@ private void AddRemoved(UniqueAddress node) private void CleanupOverdueNotMemberAnyMore() { - _removed = _removed.Where(kv => kv.Value.IsOverdue).ToImmutableDictionary(); + _removed = _removed.Where(kv => !kv.Value.IsOverdue).ToImmutableDictionary(); } private ActorSelection Peer(Address at) diff --git a/src/contrib/cluster/Akka.DistributedData/ReadAggregator.cs b/src/contrib/cluster/Akka.DistributedData/ReadAggregator.cs index f8a77c5915c..6ec6f8490e6 100644 --- a/src/contrib/cluster/Akka.DistributedData/ReadAggregator.cs +++ b/src/contrib/cluster/Akka.DistributedData/ReadAggregator.cs @@ -234,7 +234,8 @@ public override int GetHashCode() /// /// but with the given number of nodes added to the majority count. At most - /// all nodes. + /// all nodes. Exiting nodes are excluded using `ReadMajorityPlus` because those are typically + /// about to be removed and will not be able to respond. /// public sealed class ReadMajorityPlus : IReadConsistency, IEquatable { diff --git a/src/contrib/cluster/Akka.DistributedData/Replicator.cs b/src/contrib/cluster/Akka.DistributedData/Replicator.cs index 53d2c2cdf3a..abaf692d960 100644 --- a/src/contrib/cluster/Akka.DistributedData/Replicator.cs +++ b/src/contrib/cluster/Akka.DistributedData/Replicator.cs @@ -307,6 +307,11 @@ public static Props Props(ReplicatorSettings settings) => /// private ImmutableSortedSet
_joiningNodes = ImmutableSortedSet
.Empty; + /// + /// cluster exiting nodes, doesn't contain selfAddress + /// + private ImmutableSortedSet
_exitingNodes = ImmutableSortedSet
.Empty; + private ImmutableDictionary _removedNodes = ImmutableDictionary.Empty; private ImmutableDictionary _pruningPerformed = ImmutableDictionary.Empty; private ImmutableHashSet _tombstonedNodes = ImmutableHashSet.Empty; @@ -423,12 +428,22 @@ public Replicator(ReplicatorSettings settings) else Become(NormalReceive); } - private IImmutableList
NodesForReadWrite() + private IImmutableList
NodesForReadWrite(bool excludeExiting) { - if (_settings.PreferOldest) - return _membersByAge.Select(i => i.Address).ToImmutableList(); + if (excludeExiting && !_exitingNodes.IsEmpty) + { + if (_settings.PreferOldest) + return _membersByAge.Where(i => !_exitingNodes.Contains(i.Address)).Select(i => i.Address).ToImmutableList(); + else + return _nodes.Except(_exitingNodes).ToImmutableList(); + } else - return _nodes.ToImmutableList(); + { + if (_settings.PreferOldest) + return _membersByAge.Select(i => i.Address).ToImmutableList(); + else + return _nodes.ToImmutableList(); + } } protected override void PreStart() @@ -585,6 +600,7 @@ private bool NormalReceive(object message) case ClusterEvent.MemberJoined m: ReceiveMemberJoining(m.Member); return true; case ClusterEvent.MemberWeaklyUp m: ReceiveMemberWeaklyUp(m.Member); return true; case ClusterEvent.MemberUp m: ReceiveMemberUp(m.Member); return true; + case ClusterEvent.MemberExited m: ReceiveMemberExiting(m.Member); return true; case ClusterEvent.MemberRemoved m: ReceiveMemberRemoved(m.Member); return true; case ClusterEvent.IMemberEvent m: ReceiveOtherMemberEvent(m.Member); return true; @@ -613,8 +629,12 @@ private void ReceiveGet(IKey key, IReadConsistency consistency, object req) else Sender.Tell(new GetSuccess(key, req, localValue.Data)); } else - Context.ActorOf(ReadAggregator.Props(key, consistency, req, NodesForReadWrite(), _unreachable, !_settings.PreferOldest, localValue, Sender) + { + var excludeExiting = consistency is ReadMajorityPlus || consistency is ReadAll; + + Context.ActorOf(ReadAggregator.Props(key, consistency, req, NodesForReadWrite(excludeExiting), _unreachable, !_settings.PreferOldest, localValue, Sender) .WithDispatcher(Context.Props.Dispatcher)); + } } private bool IsLocalGet(IReadConsistency consistency) @@ -727,8 +747,10 @@ private void ReceiveUpdate(IKey key, Func modi // The order is also kept when prefer-oldest is enabled. var shuffle = !(_settings.PreferOldest || (writeDelta?.RequiresCausalDeliveryOfDeltas) == true); + var excludeExiting = consistency is WriteMajorityPlus || consistency is WriteAll; + var writeAggregator = Context.ActorOf(WriteAggregator - .Props(key, writeEnvelope, writeDelta, consistency, request, NodesForReadWrite(), _unreachable, shuffle, Sender, durable) + .Props(key, writeEnvelope, writeDelta, consistency, request, NodesForReadWrite(excludeExiting), _unreachable, shuffle, Sender, durable) .WithDispatcher(Context.Props.Dispatcher)); if (durable) @@ -851,8 +873,10 @@ private void ReceiveDelete(IKey key, IWriteConsistency consistency, object reque } else { + var excludeExiting = consistency is WriteMajorityPlus || consistency is WriteAll; + var writeAggregator = Context.ActorOf(WriteAggregator - .Props(key, DeletedEnvelope, null, consistency, request, NodesForReadWrite(), _unreachable, !_settings.PreferOldest, Sender, durable) + .Props(key, DeletedEnvelope, null, consistency, request, NodesForReadWrite(excludeExiting), _unreachable, !_settings.PreferOldest, Sender, durable) .WithDispatcher(Context.Props.Dispatcher)); if (durable) @@ -1302,6 +1326,12 @@ private void ReceiveMemberUp(Member m) } } + private void ReceiveMemberExiting(Member m) + { + if (MatchingRole(m) && m.Address != _selfAddress) + _exitingNodes = _exitingNodes.Add(m.Address); + } + private void ReceiveMemberRemoved(Member m) { if (m.Address == _selfAddress) Context.Stop(Self); @@ -1315,6 +1345,7 @@ private void ReceiveMemberRemoved(Member m) _nodes = _nodes.Remove(m.Address); _weaklyUpNodes = _weaklyUpNodes.Remove(m.Address); _joiningNodes = _joiningNodes.Remove(m.Address); + _exitingNodes = _exitingNodes.Remove(m.Address); _removedNodes = _removedNodes.SetItem(m.UniqueAddress, _allReachableClockTime); _unreachable = _unreachable.Remove(m.Address); diff --git a/src/contrib/cluster/Akka.DistributedData/WriteAggregator.cs b/src/contrib/cluster/Akka.DistributedData/WriteAggregator.cs index bdf9a1a010e..d4a87ea89d0 100644 --- a/src/contrib/cluster/Akka.DistributedData/WriteAggregator.cs +++ b/src/contrib/cluster/Akka.DistributedData/WriteAggregator.cs @@ -267,7 +267,8 @@ public override int GetHashCode() /// /// but with the given number of nodes added to the majority count. At most - /// all nodes. + /// all nodes. Exiting nodes are excluded using `WriteMajorityPlus` because those are typically + /// about to be removed and will not be able to respond. /// public sealed class WriteMajorityPlus : IWriteConsistency, IEquatable { diff --git a/src/contrib/persistence/Akka.Persistence.Sql.Common/Akka.Persistence.Sql.Common.csproj b/src/contrib/persistence/Akka.Persistence.Sql.Common/Akka.Persistence.Sql.Common.csproj index f45af5faf11..6c74759db1c 100644 --- a/src/contrib/persistence/Akka.Persistence.Sql.Common/Akka.Persistence.Sql.Common.csproj +++ b/src/contrib/persistence/Akka.Persistence.Sql.Common/Akka.Persistence.Sql.Common.csproj @@ -7,7 +7,6 @@ $(NetStandardLibVersion) $(AkkaPackageTags);persistence;eventsource;sql true - beta diff --git a/src/contrib/persistence/Akka.Persistence.Sql.Common/Snapshot/SqlSnapshotStore.cs b/src/contrib/persistence/Akka.Persistence.Sql.Common/Snapshot/SqlSnapshotStore.cs index f63d246e7b3..644c19420e9 100644 --- a/src/contrib/persistence/Akka.Persistence.Sql.Common/Snapshot/SqlSnapshotStore.cs +++ b/src/contrib/persistence/Akka.Persistence.Sql.Common/Snapshot/SqlSnapshotStore.cs @@ -123,23 +123,27 @@ private async Task Initialize() } catch (Exception e) { - return new Failure {Exception = e}; + return new Status.Failure(e); } } - private bool WaitingForInitialization(object message) => message.Match() - .With(_ => - { - UnbecomeStacked(); - Stash.UnstashAll(); - }) - .With(failure => + private bool WaitingForInitialization(object message) + { + switch(message) { - Log.Error(failure.Exception, "Error during snapshot store initialization"); - Context.Stop(Self); - }) - .Default(_ => Stash.Stash()) - .WasHandled; + case Initialized _: + UnbecomeStacked(); + Stash.UnstashAll(); + return true; + case Status.Failure msg: + Log.Error(msg.Cause, "Error during snapshot store initialization"); + Context.Stop(Self); + return true; + default: + Stash.Stash(); + return true; + } + } /// /// TBD diff --git a/src/contrib/persistence/Akka.Persistence.Sqlite.Tests/Bugfix4360Spec.cs b/src/contrib/persistence/Akka.Persistence.Sqlite.Tests/Bugfix4360Spec.cs index 8f9411c5b3e..a09f1895697 100644 --- a/src/contrib/persistence/Akka.Persistence.Sqlite.Tests/Bugfix4360Spec.cs +++ b/src/contrib/persistence/Akka.Persistence.Sqlite.Tests/Bugfix4360Spec.cs @@ -57,9 +57,9 @@ public class DoSnapshot private DoSnapshot() { } } - public RecoverActor(IActorRef reply) + public RecoverActor(IActorRef reply, string persistenceId) { - PersistenceId = Self.Path.Name; + PersistenceId = persistenceId; Reply = reply; } @@ -120,7 +120,8 @@ public Bugfix4360Spec(ITestOutputHelper output) : base(TestConf, output: output) [Fact] public void Should_recover_persistentactor_sqlite() { - var recoveryActor = Sys.ActorOf(Props.Create(() => new RecoverActor(TestActor)), ThreadLocalRandom.Current.Next(0, 100000).ToString()); + var id = ThreadLocalRandom.Current.Next(0, 100000).ToString(); + var recoveryActor = Sys.ActorOf(Props.Create(() => new RecoverActor(TestActor, id))); var r1 = ExpectMsg>(); r1.Count().Should().Be(0); // empty recovery @@ -136,14 +137,10 @@ public void Should_recover_persistentactor_sqlite() ExpectTerminated(recoveryActor); // recreate the actor and recover + var recoveryActor2 = Sys.ActorOf(Props.Create(() => new RecoverActor(TestActor, id))); - AwaitAssert(() => - { - var recoveryActor2 = Sys.ActorOf(Props.Create(() => new RecoverActor(TestActor)), recoveryActor.Path.Name); - - var r2 = ExpectMsg>(TimeSpan.FromMilliseconds(100)); - r2.Should().Contain(new[] { "foo", "bar" }); - }, interval:TimeSpan.FromMilliseconds(150)); + var r2 = ExpectMsg>(); + r2.Should().Contain(new[] { "foo", "bar" }); } [Fact] diff --git a/src/contrib/persistence/Akka.Persistence.Sqlite.Tests/Query/SqliteCurrentAllEventsSpec.cs b/src/contrib/persistence/Akka.Persistence.Sqlite.Tests/Query/SqliteCurrentAllEventsSpec.cs index e95723e5467..e3d3bf99073 100644 --- a/src/contrib/persistence/Akka.Persistence.Sqlite.Tests/Query/SqliteCurrentAllEventsSpec.cs +++ b/src/contrib/persistence/Akka.Persistence.Sqlite.Tests/Query/SqliteCurrentAllEventsSpec.cs @@ -30,7 +30,7 @@ class = ""Akka.Persistence.Sqlite.Journal.SqliteJournal, Akka.Persistence.Sqlite table-name = event_journal metadata-table-name = journal_metadata auto-initialize = on - connection-string = ""Filename=file:memdb-journal-eventsbytag-{id}.db;Mode=Memory;Cache=Shared"" + connection-string = ""Filename=file:memdb-journal-allevents-{id}.db;Mode=Memory;Cache=Shared"" refresh-interval = 1s }} akka.test.single-expect-default = 10s") diff --git a/src/contrib/testkits/Akka.TestKit.Xunit/Internals/Loggers.cs b/src/contrib/testkits/Akka.TestKit.Xunit/Internals/Loggers.cs index 66244c36e28..6b64b006915 100644 --- a/src/contrib/testkits/Akka.TestKit.Xunit/Internals/Loggers.cs +++ b/src/contrib/testkits/Akka.TestKit.Xunit/Internals/Loggers.cs @@ -5,6 +5,7 @@ // //----------------------------------------------------------------------- +using System; using Akka.Actor; using Akka.Event; using Xunit.Abstractions; @@ -16,20 +17,44 @@ namespace Akka.TestKit.Xunit.Internals /// public class TestOutputLogger : ReceiveActor { + private readonly ITestOutputHelper _output; + /// /// Initializes a new instance of the class. /// /// The provider used to write test output. public TestOutputLogger(ITestOutputHelper output) { - Receive(e => output.WriteLine(e.ToString())); - Receive(e => output.WriteLine(e.ToString())); - Receive(e => output.WriteLine(e.ToString())); - Receive(e => output.WriteLine(e.ToString())); + _output = output; + + Receive(Write); + Receive(Write); + Receive(Write); + Receive(Write); Receive(e => { e.LoggingBus.Subscribe(Self, typeof (LogEvent)); }); } + + private void Write(LogEvent e) + { + try + { + _output.WriteLine(e.ToString()); + } + catch (FormatException ex) + { + if (e.Message is LogMessage msg) + { + var message = + $"Received a malformed formatted message. Log level: [{e.LogLevel()}], Template: [{msg.Format}], args: [{string.Join(",", msg.Args)}]"; + if(e.Cause != null) + throw new AggregateException(message, ex, e.Cause); + throw new FormatException(message, ex); + } + throw; + } + } } } diff --git a/src/core/Akka.API.Tests/CoreAPISpec.ApproveClusterMetrics.approved.txt b/src/core/Akka.API.Tests/CoreAPISpec.ApproveClusterMetrics.approved.txt index c527ba6bf0f..4fab075a254 100644 --- a/src/core/Akka.API.Tests/CoreAPISpec.ApproveClusterMetrics.approved.txt +++ b/src/core/Akka.API.Tests/CoreAPISpec.ApproveClusterMetrics.approved.txt @@ -23,7 +23,7 @@ namespace Akka.Cluster.Metrics [Akka.Annotations.InternalApiAttribute()] public class AdaptiveLoadBalancingMetricsListener : Akka.Actor.ActorBase { - public AdaptiveLoadBalancingMetricsListener(Akka.Cluster.Metrics.AdaptiveLoadBalancingRoutingLogic routingLogic) { } + public AdaptiveLoadBalancingMetricsListener(Akka.Cluster.Metrics.IClusterMetricsRoutingLogic routingLogic) { } protected override void PostStop() { } protected override void PreStart() { } protected override bool Receive(object message) { } @@ -52,7 +52,7 @@ namespace Akka.Cluster.Metrics public Akka.Util.ISurrogated FromSurrogate(Akka.Actor.ActorSystem system) { } } } - public sealed class AdaptiveLoadBalancingRoutingLogic : Akka.Routing.RoutingLogic + public sealed class AdaptiveLoadBalancingRoutingLogic : Akka.Routing.RoutingLogic, Akka.Cluster.Metrics.IClusterMetricsRoutingLogic { public AdaptiveLoadBalancingRoutingLogic(Akka.Actor.ActorSystem system, Akka.Cluster.Metrics.IMetricsSelector metricsSelector = null) { } public void MetricsChanged(Akka.Cluster.Metrics.Events.ClusterMetricsChanged @event) { } @@ -120,6 +120,10 @@ namespace Akka.Cluster.Metrics public CpuMetricsSelector() { } public override System.Collections.Immutable.IImmutableDictionary Capacity(System.Collections.Immutable.IImmutableSet nodeMetrics) { } } + public interface IClusterMetricsRoutingLogic + { + void MetricsChanged(Akka.Cluster.Metrics.Events.ClusterMetricsChanged @event); + } public interface IMetricsCollector : System.IDisposable { Akka.Cluster.Metrics.Serialization.NodeMetrics Sample(); diff --git a/src/core/Akka.API.Tests/CoreAPISpec.ApproveCore.approved.txt b/src/core/Akka.API.Tests/CoreAPISpec.ApproveCore.approved.txt index 12863425b8c..d2b56896a88 100644 --- a/src/core/Akka.API.Tests/CoreAPISpec.ApproveCore.approved.txt +++ b/src/core/Akka.API.Tests/CoreAPISpec.ApproveCore.approved.txt @@ -93,7 +93,6 @@ namespace Akka.Actor public System.Collections.Generic.IEnumerable GetChildren() { } public static Akka.Actor.IActorRef GetCurrentSelfOrNoSender() { } public static Akka.Actor.IActorRef GetCurrentSenderOrNoSender() { } - [System.ObsoleteAttribute("Use TryGetSingleChild [0.7.1]")] public Akka.Actor.IInternalActorRef GetSingleChild(string name) { } public void Init(bool sendSupervise, Akka.Dispatch.MailboxType mailboxType) { } public Akka.Actor.Internal.ChildRestartStats InitChild(Akka.Actor.IInternalActorRef actor) { } @@ -127,7 +126,6 @@ namespace Akka.Actor public void TerminatedQueuedFor(Akka.Actor.IActorRef subject, Akka.Util.Option customMessage) { } public bool TryGetChildStatsByName(string name, out Akka.Actor.Internal.IChildStats child) { } protected bool TryGetChildStatsByRef(Akka.Actor.IActorRef actor, out Akka.Actor.Internal.ChildRestartStats child) { } - public bool TryGetSingleChild(string name, out Akka.Actor.IInternalActorRef child) { } public void UnbecomeStacked() { } protected void UnreserveChild(string name) { } public Akka.Actor.IActorRef Unwatch(Akka.Actor.IActorRef subject) { } @@ -853,23 +851,24 @@ namespace Akka.Actor } public delegate void TransitionHandler(TState initialState, TState nextState); } + [System.ObsoleteAttribute("Use Akka.Actor.Status.Failure")] public class Failure { public Failure() { } public System.Exception Exception { get; set; } public System.DateTime Timestamp { get; set; } } + [System.ObsoleteAttribute("Use List of Akka.Actor.Status.Failure")] public class Failures { public Failures() { } public System.Collections.Generic.List Entries { get; } } - public class FutureActorRef : Akka.Actor.MinimalActorRef + public sealed class FutureActorRef : Akka.Actor.MinimalActorRef { - public FutureActorRef(System.Threading.Tasks.TaskCompletionSource result, System.Action unregister, Akka.Actor.ActorPath path) { } + public FutureActorRef(System.Threading.Tasks.TaskCompletionSource result, Akka.Actor.ActorPath path, Akka.Actor.IActorRefProvider provider) { } public override Akka.Actor.ActorPath Path { get; } public override Akka.Actor.IActorRefProvider Provider { get; } - public override void SendSystemMessage(Akka.Dispatch.SysMsg.ISystemMessage message) { } protected override void TellInternal(object message, Akka.Actor.IActorRef sender) { } } public class static Futures @@ -1081,7 +1080,7 @@ namespace Akka.Actor bool IsTerminated { get; } Akka.Actor.IInternalActorRef Parent { get; } Akka.Actor.IActorRefProvider Provider { get; } - Akka.Actor.IActorRef GetChild(System.Collections.Generic.IEnumerable name); + Akka.Actor.IActorRef GetChild(System.Collections.Generic.IReadOnlyList name); void Restart(System.Exception cause); void Resume(System.Exception causedByFailure = null); [System.ObsoleteAttribute("Use SendSystemMessage(message) [1.1.0]")] @@ -1202,7 +1201,7 @@ namespace Akka.Actor public abstract bool IsTerminated { get; } public abstract Akka.Actor.IInternalActorRef Parent { get; } public abstract Akka.Actor.IActorRefProvider Provider { get; } - public abstract Akka.Actor.IActorRef GetChild(System.Collections.Generic.IEnumerable name); + public abstract Akka.Actor.IActorRef GetChild(System.Collections.Generic.IReadOnlyList name); public abstract void Restart(System.Exception cause); public abstract void Resume(System.Exception causedByFailure = null); [System.ObsoleteAttribute("Use SendSystemMessage(message) instead [1.1.0]")] @@ -1245,7 +1244,7 @@ namespace Akka.Actor protected Akka.Actor.IInternalActorRef Supervisor { get; } protected Akka.Actor.ActorSystem System { get; } public override Akka.Actor.ICell Underlying { get; } - public override Akka.Actor.IActorRef GetChild(System.Collections.Generic.IEnumerable name) { } + public override Akka.Actor.IActorRef GetChild(System.Collections.Generic.IReadOnlyList name) { } public override Akka.Actor.IInternalActorRef GetSingleChild(string name) { } protected virtual Akka.Actor.ActorCell NewActorCell(Akka.Actor.Internal.ActorSystemImpl system, Akka.Actor.IInternalActorRef self, Akka.Actor.Props props, Akka.Dispatch.MessageDispatcher dispatcher, Akka.Actor.IInternalActorRef supervisor) { } public override void Restart(System.Exception cause) { } @@ -1317,7 +1316,7 @@ namespace Akka.Actor [System.ObsoleteAttribute("Use Context.Watch and Receive [1.1.0]")] public override bool IsTerminated { get; } public override Akka.Actor.IInternalActorRef Parent { get; } - public override Akka.Actor.IActorRef GetChild(System.Collections.Generic.IEnumerable name) { } + public override Akka.Actor.IActorRef GetChild(System.Collections.Generic.IReadOnlyList name) { } public override void Restart(System.Exception cause) { } public override void Resume(System.Exception causedByFailure = null) { } public override void SendSystemMessage(Akka.Dispatch.SysMsg.ISystemMessage message) { } @@ -1533,7 +1532,7 @@ namespace Akka.Actor public override Akka.Actor.ActorPath Path { get; } public override Akka.Actor.IActorRefProvider Provider { get; } public override Akka.Actor.ICell Underlying { get; } - public override Akka.Actor.IActorRef GetChild(System.Collections.Generic.IEnumerable name) { } + public override Akka.Actor.IActorRef GetChild(System.Collections.Generic.IReadOnlyList name) { } public override Akka.Actor.IInternalActorRef GetSingleChild(string name) { } public Akka.Actor.RepointableActorRef Initialize(bool async) { } protected virtual Akka.Actor.ActorCell NewCell() { } @@ -1699,6 +1698,7 @@ namespace Akka.Actor public bool SerializeAllMessages { get; } public Akka.Actor.Setup.ActorSystemSetup Setup { get; } public string StdoutLogLevel { get; } + public Akka.Event.MinimalLogger StdoutLogger { get; } public string SupervisorStrategyClass { get; } public Akka.Actor.ActorSystem System { get; } public System.TimeSpan UnstartedPushTimeout { get; } @@ -1720,16 +1720,20 @@ namespace Akka.Actor public abstract class Status { protected Status() { } - public class Failure : Akka.Actor.Status + public sealed class Failure : Akka.Actor.Status { public readonly System.Exception Cause; + public readonly object State; public Failure(System.Exception cause) { } + public Failure(System.Exception cause, object state) { } public override string ToString() { } } - public class Success : Akka.Actor.Status + public sealed class Success : Akka.Actor.Status { + public static readonly Akka.Actor.Status.Success Instance; public readonly object Status; public Success(object status) { } + public override string ToString() { } } } public class StoppingSupervisorStrategy : Akka.Actor.SupervisorStrategyConfigurator @@ -3092,7 +3096,7 @@ namespace Akka.Event public static string FromType(System.Type t, Akka.Actor.ActorSystem system) { } public static System.Type SourceType(object o) { } } - public class LoggerInitialized : Akka.Actor.INoSerializationVerificationNeeded + public class LoggerInitialized : Akka.Actor.INoSerializationVerificationNeeded, Akka.Event.IDeadLetterSuppression { public LoggerInitialized() { } } @@ -3102,7 +3106,6 @@ namespace Akka.Event } public class static Logging { - public static readonly Akka.Event.StandardOutLogger StandardOutLogger; public static System.Type ClassFor(this Akka.Event.LogLevel logLevel) { } public static Akka.Event.ILoggingAdapter GetLogger(this Akka.Actor.IActorContext context, Akka.Event.ILogMessageFormatter logMessageFormatter = null) { } public static Akka.Event.ILoggingAdapter GetLogger(Akka.Actor.ActorSystem system, object logSourceObj, Akka.Event.ILogMessageFormatter logMessageFormatter = null) { } @@ -3155,6 +3158,14 @@ namespace Akka.Event public void SetLogLevel(Akka.Event.LogLevel logLevel) { } public void StartStdoutLogger(Akka.Actor.Settings config) { } } + public abstract class MinimalLogger : Akka.Actor.MinimalActorRef + { + protected MinimalLogger() { } + public virtual Akka.Actor.ActorPath Path { get; } + public virtual Akka.Actor.IActorRefProvider Provider { get; } + protected abstract void Log(object message); + protected virtual void TellInternal(object message, Akka.Actor.IActorRef sender) { } + } public sealed class NoLogger : Akka.Event.ILoggingAdapter { public static readonly Akka.Event.ILoggingAdapter Instance; @@ -3175,18 +3186,15 @@ namespace Akka.Event public void Warning(string format, params object[] args) { } public void Warning(System.Exception cause, string format, params object[] args) { } } - public class StandardOutLogger : Akka.Actor.MinimalActorRef + public class StandardOutLogger : Akka.Event.MinimalLogger { public StandardOutLogger() { } public static System.ConsoleColor DebugColor { get; set; } public static System.ConsoleColor ErrorColor { get; set; } public static System.ConsoleColor InfoColor { get; set; } - public override Akka.Actor.ActorPath Path { get; } - public override Akka.Actor.IActorRefProvider Provider { get; } public static bool UseColors { get; set; } public static System.ConsoleColor WarningColor { get; set; } - public static void PrintLogEvent(Akka.Event.LogEvent logEvent) { } - protected override void TellInternal(object message, Akka.Actor.IActorRef sender) { } + protected override void Log(object message) { } } public class Subscription { @@ -3297,10 +3305,13 @@ namespace Akka.IO } public class Resolved : Akka.IO.Dns.Command { + public Resolved(string name, System.Exception ex) { } public Resolved(string name, System.Collections.Generic.IEnumerable ipv4, System.Collections.Generic.IEnumerable ipv6) { } public System.Net.IPAddress Addr { get; } + public System.Exception Exception { get; } public System.Collections.Generic.IEnumerable Ipv4 { get; } public System.Collections.Generic.IEnumerable Ipv6 { get; } + public bool IsSuccess { get; } public string Name { get; } public static Akka.IO.Dns.Resolved Create(string name, System.Collections.Generic.IEnumerable addresses) { } } @@ -4480,7 +4491,7 @@ namespace Akka.Routing public virtual Akka.Routing.Router RemoveRoutee(Akka.Routing.Routee routee) { } public Akka.Routing.Router RemoveRoutee(Akka.Actor.IActorRef routee) { } public Akka.Routing.Router RemoveRoutee(Akka.Actor.ActorSelection routee) { } - public void Route(object message, Akka.Actor.IActorRef sender) { } + public virtual void Route(object message, Akka.Actor.IActorRef sender) { } protected virtual void Send(Akka.Routing.Routee routee, object message, Akka.Actor.IActorRef sender) { } protected object UnWrap(object message) { } public virtual Akka.Routing.Router WithRoutees(params Akka.Routing.Routee[] routees) { } diff --git a/src/core/Akka.API.Tests/CoreAPISpec.ApproveRemote.approved.txt b/src/core/Akka.API.Tests/CoreAPISpec.ApproveRemote.approved.txt index 0d3f12af45c..3b90d4f2deb 100644 --- a/src/core/Akka.API.Tests/CoreAPISpec.ApproveRemote.approved.txt +++ b/src/core/Akka.API.Tests/CoreAPISpec.ApproveRemote.approved.txt @@ -171,7 +171,7 @@ namespace Akka.Remote public override Akka.Actor.IInternalActorRef Parent { get; } public override Akka.Actor.ActorPath Path { get; } public override Akka.Actor.IActorRefProvider Provider { get; } - public override Akka.Actor.IActorRef GetChild(System.Collections.Generic.IEnumerable name) { } + public override Akka.Actor.IActorRef GetChild(System.Collections.Generic.IReadOnlyList name) { } public bool IsWatchIntercepted(Akka.Actor.IActorRef watchee, Akka.Actor.IActorRef watcher) { } public override void Restart(System.Exception cause) { } public override void Resume(System.Exception causedByFailure = null) { } diff --git a/src/core/Akka.Persistence.TCK/Query/PersistenceIdsSpec.cs b/src/core/Akka.Persistence.TCK/Query/PersistenceIdsSpec.cs index 47b82f41f06..0a16b1099a8 100644 --- a/src/core/Akka.Persistence.TCK/Query/PersistenceIdsSpec.cs +++ b/src/core/Akka.Persistence.TCK/Query/PersistenceIdsSpec.cs @@ -177,7 +177,7 @@ public virtual void ReadJournal_AllPersistenceIds_should_deliver_persistenceId_o }); } - [Fact] + [Fact(Skip = "Not a good test - tightly couples to private implementation details")] public virtual async Task ReadJournal_should_deallocate_AllPersistenceIds_publisher_when_the_last_subscriber_left() { if (AllocatesAllPersistenceIDsPublisher) diff --git a/src/core/Akka.Persistence.Tests/ManyRecoveriesSpec.cs b/src/core/Akka.Persistence.Tests/ManyRecoveriesSpec.cs index c5d2f6485d7..2c7b8e13c42 100644 --- a/src/core/Akka.Persistence.Tests/ManyRecoveriesSpec.cs +++ b/src/core/Akka.Persistence.Tests/ManyRecoveriesSpec.cs @@ -60,7 +60,7 @@ protected override void OnCommand(object message) public ManyRecoveriesSpec() : base(ConfigurationFactory.ParseString(@" akka.actor.default-dispatcher.Type = ForkJoinDispatcher akka.actor.default-dispatcher.dedicated-thread-pool.thread-count = 5 - akka.persistence.max-concurrent-recoveries = 3 + akka.persistence.max-concurrent-recoveries = 1 akka.persistence.journal.plugin = ""akka.persistence.journal.inmem"" # snapshot store plugin is NOT defined, things should still work diff --git a/src/core/Akka.Persistence.Tests/ReceivePersistentActorAsyncAwaitSpec.cs b/src/core/Akka.Persistence.Tests/ReceivePersistentActorAsyncAwaitSpec.cs index 4ffcdbd4aca..d42af7d9a36 100644 --- a/src/core/Akka.Persistence.Tests/ReceivePersistentActorAsyncAwaitSpec.cs +++ b/src/core/Akka.Persistence.Tests/ReceivePersistentActorAsyncAwaitSpec.cs @@ -621,7 +621,7 @@ public void ActorTaskScheduler_reentrancy_should_not_be_possible() ExpectNoMsg(1000); } - [Fact] + [Fact(Skip = "Racy - both Tasks run inside the same scheduler with max concurrency == 1")] public void Actor_PipeTo_should_not_be_delayed_by_async_receive() { var actor = Sys.ActorOf(Props.Create(() => new AsyncPipeToDelayActor("pid"))); diff --git a/src/core/Akka.Remote/RemoteActorRef.cs b/src/core/Akka.Remote/RemoteActorRef.cs index 098f9f6c846..976b6956ec5 100644 --- a/src/core/Akka.Remote/RemoteActorRef.cs +++ b/src/core/Akka.Remote/RemoteActorRef.cs @@ -12,6 +12,7 @@ using Akka.Annotations; using Akka.Dispatch.SysMsg; using Akka.Event; +using Akka.Util.Internal.Collections; namespace Akka.Remote { @@ -106,17 +107,16 @@ public RemoteActorRef(RemoteTransport remote, Address localAddressToUse, ActorPa /// The name. /// ActorRef. /// TBD - public override IActorRef GetChild(IEnumerable name) + public override IActorRef GetChild(IReadOnlyList name) { - var items = name.ToList(); - switch (items.FirstOrDefault()) + switch (name.FirstOrDefault()) { case null: return this; case "..": - return Parent.GetChild(items.Skip(1)); + return Parent.GetChild(name.NoCopySlice(1)); default: - return new RemoteActorRef(Remote, LocalAddressToUse, Path / items, ActorRefs.Nobody, Props.None, Deploy.None); + return new RemoteActorRef(Remote, LocalAddressToUse, Path / name, ActorRefs.Nobody, Props.None, Deploy.None); } } diff --git a/src/core/Akka.Remote/RemoteActorRefProvider.cs b/src/core/Akka.Remote/RemoteActorRefProvider.cs index 2ad062aef56..9014b35857f 100644 --- a/src/core/Akka.Remote/RemoteActorRefProvider.cs +++ b/src/core/Akka.Remote/RemoteActorRefProvider.cs @@ -435,7 +435,7 @@ public Deploy LookUpRemotes(IEnumerable p) public bool HasAddress(Address address) { - return address == _local.RootPath.Address || address == RootPath.Address || Transport.Addresses.Any(a => a == address); + return address.Equals(_local.RootPath.Address) || address.Equals(RootPath.Address) || Transport.Addresses.Contains(address); } /// @@ -490,7 +490,7 @@ public IInternalActorRef ResolveActorRefWithLocalAddress(string path, Address lo { if (actorPath is RootActorPath) return RootGuardian; - return _local.ResolveActorRef(RootGuardian, actorPath.ElementsWithUid); + return (IInternalActorRef)ResolveActorRef(path); // so we can use caching } return CreateRemoteRef(new RootActorPath(actorPath.Address) / actorPath.ElementsWithUid, localAddress); diff --git a/src/core/Akka.Remote/RemoteSystemDaemon.cs b/src/core/Akka.Remote/RemoteSystemDaemon.cs index 1a820b62fb3..2e5fce15820 100644 --- a/src/core/Akka.Remote/RemoteSystemDaemon.cs +++ b/src/core/Akka.Remote/RemoteSystemDaemon.cs @@ -156,11 +156,9 @@ protected override void TellInternal(object message, IActorRef sender) } } - var t = Rec(ImmutableList.Empty); - var concatenatedChildNames = t.Item1; - var m = t.Item2; + var (concatenatedChildNames, m) = Rec(ImmutableList.Empty); - var child = GetChild(concatenatedChildNames); + var child = GetChild(concatenatedChildNames.ToList()); if (child.IsNobody()) { var emptyRef = new EmptyLocalActorRef(_system.Provider, @@ -302,7 +300,7 @@ private void HandleDaemonMsgCreate(DaemonMsgCreate message) /// /// The name. /// ActorRef. - public override IActorRef GetChild(IEnumerable name) + public override IActorRef GetChild(IReadOnlyList name) { var path = name.Join("/"); var n = 0; @@ -313,7 +311,7 @@ public override IActorRef GetChild(IEnumerable name) { if (uid != ActorCell.UndefinedUid && uid != child.Path.Uid) return Nobody.Instance; - return n == 0 ? child : child.GetChild(name.TakeRight(n)); + return n == 0 ? child : child.GetChild(name.TakeRight(n).ToList()); } var last = path.LastIndexOf("/", StringComparison.Ordinal); diff --git a/src/core/Akka.Remote/Serialization/ActorRefResolveCache.cs b/src/core/Akka.Remote/Serialization/ActorRefResolveCache.cs index 2410572dcdb..2bd7de5b453 100644 --- a/src/core/Akka.Remote/Serialization/ActorRefResolveCache.cs +++ b/src/core/Akka.Remote/Serialization/ActorRefResolveCache.cs @@ -65,7 +65,8 @@ protected override int Hash(string k) protected override bool IsCacheable(IActorRef v) { - return !(v is EmptyLocalActorRef); + // don't cache any FutureActorRefs, et al + return !(v is MinimalActorRef && !(v is FunctionRef)); } } } diff --git a/src/core/Akka.Streams.Tests/Dsl/GraphStageTimersSpec.cs b/src/core/Akka.Streams.Tests/Dsl/GraphStageTimersSpec.cs index a143ff2ef17..b0a245a823a 100644 --- a/src/core/Akka.Streams.Tests/Dsl/GraphStageTimersSpec.cs +++ b/src/core/Akka.Streams.Tests/Dsl/GraphStageTimersSpec.cs @@ -81,8 +81,8 @@ public void GraphStage_timer_support_must_correctly_cancel_a_named_timer() { var driver = SetupIsolatedStage(); driver.Tell(TestCancelTimer.Instance); - Within(TimeSpan.FromMilliseconds(500), () => ExpectMsg()); - Within(TimeSpan.FromMilliseconds(300), TimeSpan.FromSeconds(1), () => ExpectMsg(new Tick(1))); + Within(TimeSpan.FromMilliseconds(5000), () => ExpectMsg()); + Within(TimeSpan.FromMilliseconds(200), TimeSpan.FromSeconds(3000), () => ExpectMsg(new Tick(1))); ExpectNoMsg(TimeSpan.FromSeconds(1)); driver.StopStage(); } diff --git a/src/core/Akka.Streams.Tests/Dsl/SubstreamSubscriptionTimeoutSpec.cs b/src/core/Akka.Streams.Tests/Dsl/SubstreamSubscriptionTimeoutSpec.cs index b63cc8791e1..9b3f584e861 100644 --- a/src/core/Akka.Streams.Tests/Dsl/SubstreamSubscriptionTimeoutSpec.cs +++ b/src/core/Akka.Streams.Tests/Dsl/SubstreamSubscriptionTimeoutSpec.cs @@ -39,7 +39,7 @@ public SubstreamSubscriptionTimeoutSpec(ITestOutputHelper helper) : base(Config, Materializer = ActorMaterializer.Create(Sys, settings); } - [Fact] + [Fact(Skip = "Racy")] public void GroupBy_and_SplitWhen_must_timeout_and_cancel_substream_publisher_when_no_one_subscribes_to_them_after_some_time() { this.AssertAllStagesStopped(() => @@ -57,6 +57,13 @@ public void GroupBy_and_SplitWhen_must_timeout_and_cancel_substream_publisher_wh publisherProbe.SendNext(1); publisherProbe.SendNext(2); publisherProbe.SendNext(3); + + /* + * Why this spec is skipped: in the event that subscriber.ExpectSubscription() or (subscriber.ExpectNext() + * + s1SubscriberProbe.ExpectSubscription()) exceeds 300ms, the next call to subscriber.ExpectNext will + * fail. This test is too tightly fitted to the timeout duration to be reliable, although somewhat ironically + * it does validate that the underlying cancellation does work! + */ var s1 = subscriber.ExpectNext().Item2; // should not break normal usage diff --git a/src/core/Akka.Streams.Tests/Dsl/UnfoldFlowSpec.cs b/src/core/Akka.Streams.Tests/Dsl/UnfoldFlowSpec.cs index 0811853cd4e..1211e3dd320 100644 --- a/src/core/Akka.Streams.Tests/Dsl/UnfoldFlowSpec.cs +++ b/src/core/Akka.Streams.Tests/Dsl/UnfoldFlowSpec.cs @@ -134,7 +134,7 @@ public void UnfoldFlow_should_increment_integers_and_handle_KillSwitch_and_fail_ snk.ExpectError().Should().Be(kill); } - [Fact(Skip ="Racy")] + [Fact] public void UnfoldFlow_should_increment_integers_and_handle_KillSwitch_and_fail_after_timeout_when_aborted() { var t = _source.ToMaterialized(this.SinkProbe(), Keep.Both).Run(Sys.Materializer()); @@ -147,7 +147,6 @@ public void UnfoldFlow_should_increment_integers_and_handle_KillSwitch_and_fail_ pub.EnsureSubscription(); snk.EnsureSubscription(); sub.Cancel(); - snk.ExpectNoMsg(_timeout.DivideBy(2)); snk.ExpectError(); } @@ -165,7 +164,6 @@ public void UnfoldFlow_should_increment_integers_and_handle_KillSwitch_and_fail_ snk.EnsureSubscription(); sub.Cancel(); snk.Request(1); - snk.ExpectNoMsg(_timeout.DivideBy(2)); snk.ExpectError(); } @@ -364,7 +362,6 @@ public void UnfoldFlow_should_increment_integers_and_handle_KillSwitch_and_fail_ snk.EnsureSubscription(); sub.Cancel(); snk.Request(1); - snk.ExpectNoMsg(_timeout.DivideBy(2)); snk.ExpectError(); } @@ -445,7 +442,6 @@ public void UnfoldFlow_should_increment_integers_and_handle_KillSwitch_and_compl pub.EnsureSubscription(); snk.EnsureSubscription(); sub.Cancel(); - snk.ExpectNoMsg(_timeout.DivideBy(2)); pub.SendComplete(); snk.ExpectComplete(); } diff --git a/src/core/Akka.TestKit/EventFilter/TestEventListener.cs b/src/core/Akka.TestKit/EventFilter/TestEventListener.cs index a959d8c9da3..a42ca7b2fd5 100644 --- a/src/core/Akka.TestKit/EventFilter/TestEventListener.cs +++ b/src/core/Akka.TestKit/EventFilter/TestEventListener.cs @@ -30,57 +30,65 @@ public class TestEventListener : DefaultLogger /// TBD protected override bool Receive(object message) { - if(message is InitializeLogger) + switch (message) { - var initLogger = (InitializeLogger)message; - var bus = initLogger.LoggingBus; - var self = Context.Self; - bus.Subscribe(self, typeof(Mute)); - bus.Subscribe(self, typeof(Unmute)); - bus.Subscribe(self, typeof(DeadLetter)); - bus.Subscribe(self, typeof(UnhandledMessage)); - Sender.Tell(new LoggerInitialized()); - } - else if(message is Mute) - { - var mute = (Mute)message; - foreach(var filter in mute.Filters) + case InitializeLogger initLogger: { - AddFilter(filter); + base.Receive(message); + var bus = initLogger.LoggingBus; + var self = Context.Self; + bus.Subscribe(self, typeof(Mute)); + bus.Subscribe(self, typeof(Unmute)); + bus.Subscribe(self, typeof(DeadLetter)); + bus.Subscribe(self, typeof(UnhandledMessage)); + Sender.Tell(new LoggerInitialized()); + break; } - } - else if(message is Unmute) - { - var unmute = (Unmute)message; - foreach(var filter in unmute.Filters) + case Mute mute: { - RemoveFilter(filter); + foreach(var filter in mute.Filters) + { + AddFilter(filter); + } + + break; } - } - else if(message is LogEvent) - { - var logEvent = (LogEvent)message; - if(!ShouldFilter(logEvent)) + case Unmute unmute: { - Print(logEvent); + foreach(var filter in unmute.Filters) + { + RemoveFilter(filter); + } + + break; } + case LogEvent logEvent: + { + if(!ShouldFilter(logEvent)) + { + Print(logEvent); + } + + break; + } + case DeadLetter letter: + HandleDeadLetter(letter); + break; + + case UnhandledMessage un: + { + var rcp = un.Recipient; + var warning = new Warning(rcp.Path.ToString(), rcp.GetType(), "Unhandled message from " + un.Sender + ": " + un.Message); + if(!ShouldFilter(warning)) + Print(warning); + break; + } + + default: + Print(new Debug(Context.System.Name,GetType(),message)); + break; } - else if(message is DeadLetter) - { - HandleDeadLetter((DeadLetter)message); - } - else if(message is UnhandledMessage) - { - var un = (UnhandledMessage) message; - var rcp = un.Recipient; - var warning = new Warning(rcp.Path.ToString(), rcp.GetType(), "Unhandled message from " + un.Sender + ": " + un.Message); - if(!ShouldFilter(warning)) - Print(warning); - } - else - { - Print(new Debug(Context.System.Name,GetType(),message)); - } + return true; } diff --git a/src/core/Akka.TestKit/TestActorRefBase.cs b/src/core/Akka.TestKit/TestActorRefBase.cs index b440c75f89e..a557d2545fc 100644 --- a/src/core/Akka.TestKit/TestActorRefBase.cs +++ b/src/core/Akka.TestKit/TestActorRefBase.cs @@ -258,7 +258,7 @@ ISurrogate ISurrogated.ToSurrogate(ActorSystem system) bool IInternalActorRef.IsTerminated { get { return _internalRef.IsTerminated; } } - IActorRef IInternalActorRef.GetChild(IEnumerable name) + IActorRef IInternalActorRef.GetChild(IReadOnlyList name) { return _internalRef.GetChild(name); } diff --git a/src/core/Akka.TestKit/TestProbe.cs b/src/core/Akka.TestKit/TestProbe.cs index 31fd374010c..ef586d69042 100644 --- a/src/core/Akka.TestKit/TestProbe.cs +++ b/src/core/Akka.TestKit/TestProbe.cs @@ -124,7 +124,7 @@ ISurrogate ISurrogated.ToSurrogate(ActorSystem system) bool IInternalActorRef.IsTerminated { get { return ((IInternalActorRef)TestActor).IsTerminated; } } - IActorRef IInternalActorRef.GetChild(IEnumerable name) + IActorRef IInternalActorRef.GetChild(IReadOnlyList name) { return ((IInternalActorRef)TestActor).GetChild(name); } diff --git a/src/core/Akka.Tests/Actor/ActorRefSpec.cs b/src/core/Akka.Tests/Actor/ActorRefSpec.cs index 0cb77f298db..05d012521ad 100644 --- a/src/core/Akka.Tests/Actor/ActorRefSpec.cs +++ b/src/core/Akka.Tests/Actor/ActorRefSpec.cs @@ -12,6 +12,7 @@ using Akka.Serialization; using Akka.TestKit; using Akka.TestKit.TestActors; +using Akka.Util; using Xunit; namespace Akka.Tests.Actor @@ -290,6 +291,16 @@ public void An_ActorRef_should_never_have_a_null_Sender_Bug_1212() ExpectNoMsg(); } + [Fact] + public void An_ActorRef_Mock_should_be_like_Nobody() + { + var mock = new ActorRefMock(); + + mock.Tell("dummy"); + + mock.IsNobody().ShouldBeTrue(); + } + private void VerifyActorTermination(IActorRef actorRef) { var watcher = CreateTestProbe(); @@ -297,6 +308,37 @@ private void VerifyActorTermination(IActorRef actorRef) watcher.ExpectTerminated(actorRef, TimeSpan.FromSeconds(20)); } + private sealed class ActorRefMock : IActorRef + { + public ActorRefMock() { } + + ActorPath IActorRef.Path => null; + + int IComparable.CompareTo(IActorRef other) + { + throw new NotSupportedException(); + } + + int IComparable.CompareTo(object obj) + { + throw new NotSupportedException(); + } + + bool IEquatable.Equals(IActorRef other) + { + throw new NotSupportedException(); + } + + void ICanTell.Tell(object message, IActorRef sender) + { + } + + ISurrogate ISurrogated.ToSurrogate(ActorSystem system) + { + throw new NotSupportedException(); + } + } + private class NestingActor : ActorBase { internal readonly IActorRef Nested; diff --git a/src/core/Akka.Tests/Actor/AskSpec.cs b/src/core/Akka.Tests/Actor/AskSpec.cs index c7ec057a2fb..0e0fca26f47 100644 --- a/src/core/Akka.Tests/Actor/AskSpec.cs +++ b/src/core/Akka.Tests/Actor/AskSpec.cs @@ -15,6 +15,7 @@ using Akka.Util.Internal; using FluentAssertions; using Nito.AsyncEx; +using Akka.Dispatch.SysMsg; namespace Akka.Tests.Actor { @@ -37,6 +38,29 @@ protected override void OnReceive(object message) { Sender.Tell("answer"); } + + if (message.Equals("delay")) + { + Thread.Sleep(3000); + Sender.Tell("answer"); + } + + if (message.Equals("many")) + { + Sender.Tell("answer1"); + Sender.Tell("answer2"); + Sender.Tell("answer2"); + } + + if (message.Equals("invalid")) + { + Sender.Tell(123); + } + + if (message.Equals("system")) + { + Sender.Tell(new DummySystemMessage()); + } } } @@ -83,6 +107,10 @@ protected override void OnReceive(object message) } } + public sealed class DummySystemMessage : ISystemMessage + { + } + [Fact] public async Task Can_Ask_Response_actor() { @@ -114,6 +142,60 @@ public async Task Can_get_timeout_when_asking_actor() await Assert.ThrowsAsync(async () => await actor.Ask("timeout", TimeSpan.FromSeconds(3))); } + [Fact] + public async Task Ask_should_put_timeout_answer_into_deadletter() + { + var actor = Sys.ActorOf(); + + await EventFilter.DeadLetter().ExpectOneAsync(TimeSpan.FromSeconds(5), async () => + { + await Assert.ThrowsAsync(async () => await actor.Ask("delay", TimeSpan.FromSeconds(1))); + }); + } + + [Fact] + public async Task Ask_should_put_too_many_answers_into_deadletter() + { + var actor = Sys.ActorOf(); + + await EventFilter.DeadLetter().ExpectAsync(2, async () => + { + var result = await actor.Ask("many", TimeSpan.FromSeconds(1)); + result.ShouldBe("answer1"); + }); + } + + [Fact] + public async Task Ask_should_not_put_canceled_answer_into_deadletter() + { + var actor = Sys.ActorOf(); + + await EventFilter.DeadLetter().ExpectAsync(0, async () => + { + using (var cts = new CancellationTokenSource(TimeSpan.FromSeconds(1))) + await Assert.ThrowsAsync(async () => await actor.Ask("delay", Timeout.InfiniteTimeSpan, cts.Token)); + }); + } + + [Fact] + public async Task Ask_should_put_invalid_answer_into_deadletter() + { + var actor = Sys.ActorOf(); + + await EventFilter.DeadLetter().ExpectOne(async () => + { + await Assert.ThrowsAsync(async () => await actor.Ask("invalid", TimeSpan.FromSeconds(1))); + }); + } + + [Fact] + public async Task Ask_should_fail_on_system_message() + { + var actor = Sys.ActorOf(); + + await Assert.ThrowsAsync(async () => await actor.Ask("system", TimeSpan.FromSeconds(1))); + } + [Fact] public async Task Can_cancel_when_asking_actor() { diff --git a/src/core/Akka.Tests/Configuration/ConfigurationSpec.cs b/src/core/Akka.Tests/Configuration/ConfigurationSpec.cs index 7074746fa98..18654b6f9f9 100644 --- a/src/core/Akka.Tests/Configuration/ConfigurationSpec.cs +++ b/src/core/Akka.Tests/Configuration/ConfigurationSpec.cs @@ -42,6 +42,8 @@ public void The_default_configuration_file_contain_all_configuration_properties( settings.LoggerStartTimeout.Seconds.ShouldBe(5); settings.LogLevel.ShouldBe("INFO"); settings.StdoutLogLevel.ShouldBe("WARNING"); + settings.StdoutLogger.Should().NotBeNull(); + settings.StdoutLogger.Should().BeOfType(); settings.LogConfigOnStart.ShouldBeFalse(); settings.LogDeadLetters.ShouldBe(10); settings.LogDeadLettersDuringShutdown.ShouldBeFalse(); diff --git a/src/core/Akka.Tests/IO/TcpIntegrationSpec.cs b/src/core/Akka.Tests/IO/TcpIntegrationSpec.cs index 0b56cb5c43a..acaa67a06ba 100644 --- a/src/core/Akka.Tests/IO/TcpIntegrationSpec.cs +++ b/src/core/Akka.Tests/IO/TcpIntegrationSpec.cs @@ -22,6 +22,8 @@ using Xunit.Abstractions; using FluentAssertions; using System.Runtime.InteropServices; +using System.Threading; +using Akka.Event; namespace Akka.Tests.IO { @@ -470,6 +472,18 @@ public void The_TCP_transport_implementation_dont_report_Connected_when_endpoint replies.Count.ShouldBe(0); } + [Fact] + public void Should_report_Error_only_once_when_connecting_to_unreachable_DnsEndpoint() + { + var probe = CreateTestProbe(); + var endpoint = new DnsEndPoint("fake", 1000); + Sys.Tcp().Tell(new Tcp.Connect(endpoint), probe.Ref); + + // expecting CommandFailed or no reply (within timeout) + var replies = probe.ReceiveWhile(TimeSpan.FromSeconds(5), x => x as Tcp.CommandFailed); + replies.Count.ShouldBe(1); + } + [Fact] public void The_TCP_transport_implementation_handle_tcp_connection_actor_death_properly() { diff --git a/src/core/Akka.Tests/Loggers/LoggerSpec.cs b/src/core/Akka.Tests/Loggers/LoggerSpec.cs index 72681e225a6..d7079a1de5a 100644 --- a/src/core/Akka.Tests/Loggers/LoggerSpec.cs +++ b/src/core/Akka.Tests/Loggers/LoggerSpec.cs @@ -2,7 +2,10 @@ using System.Collections.Generic; using System.Linq; using System.Threading; +using System.Threading.Tasks; using Akka.Actor; +using Akka.Actor.Internal; +using Akka.Actor.Setup; using Akka.Configuration; using Akka.Event; using Akka.TestKit; @@ -142,5 +145,26 @@ private class FakeException : Exception public FakeException(string message) : base(message) { } } + + [Fact] + public async Task ShouldBeAbleToReplaceStandardOutLoggerWithCustomMinimalLogger() + { + var config = ConfigurationFactory + .ParseString("akka.stdout-logger-class = \"Akka.Tests.Loggers.LoggerSpec+CustomLogger, Akka.Tests\"") + .WithFallback(ConfigurationFactory.Default()); + + var system = ActorSystem.Create("MinimalLoggerTest", config); + system.Settings.StdoutLogger.Should().BeOfType(); + await system.Terminate(); + } + + public class CustomLogger : MinimalLogger + { + protected override void Log(object message) + { + Console.WriteLine(message); + } + } + } } diff --git a/src/core/Akka.Tests/Loggers/ShutdownLoggerSpec.cs b/src/core/Akka.Tests/Loggers/ShutdownLoggerSpec.cs new file mode 100644 index 00000000000..796a4404d26 --- /dev/null +++ b/src/core/Akka.Tests/Loggers/ShutdownLoggerSpec.cs @@ -0,0 +1,84 @@ +// //----------------------------------------------------------------------- +// // +// // Copyright (C) 2009-2021 Lightbend Inc. +// // Copyright (C) 2013-2021 .NET Foundation +// // +// //----------------------------------------------------------------------- + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Reflection; +using System.Threading.Tasks; +using Akka.Actor; +using Akka.Configuration; +using Akka.Event; +using Akka.TestKit; +using FluentAssertions; +using Xunit; +using Xunit.Abstractions; + +namespace Akka.Tests.Loggers +{ + public class ShutdownLoggerSpec: AkkaSpec + { + private static readonly Config Config = ConfigurationFactory.ParseString(@" +akka.loglevel = OFF +akka.stdout-loglevel = OFF +akka.stdout-logger-class = ""Akka.Tests.Loggers.ThrowingLogger, Akka.Tests"""); + + public ShutdownLoggerSpec(ITestOutputHelper output) : base(Config, output) + { + } + + [Fact(DisplayName = "StandardOutLogger should not be called on shutdown when stdout-loglevel is set to OFF")] + public async Task StandardOutLoggerShouldNotBeCalled() + { + Sys.Settings.StdoutLogger.Should().BeOfType(); + + var probeRef = Sys.ActorOf(Props.Create(() => new LogProbe())); + probeRef.Tell(new InitializeLogger(Sys.EventStream)); + var probe = await probeRef.Ask("hey"); + + await Sys.Terminate(); + + await Task.Delay(RemainingOrDefault); + if (probe.Events.Any(o => o is Error err && err.Cause is ShutdownLogException)) + throw new Exception("Test failed, log should not be called after shutdown."); + } + } + + internal class LogProbe : ReceiveActor + { + public List Events { get; } = new List(); + + public LogProbe() + { + Receive(msg => Sender.Tell(this)); + Receive(Events.Add); + Receive(e => + { + e.LoggingBus.Subscribe(Self, typeof (LogEvent)); + Sender.Tell(new LoggerInitialized()); + }); + + } + } + + internal class ShutdownLogException : Exception + { + public ShutdownLogException() + { } + + public ShutdownLogException(string msg) : base(msg) + { } + } + + internal class ThrowingLogger : MinimalLogger + { + protected override void Log(object message) + { + throw new ShutdownLogException("This logger should never be called."); + } + } +} \ No newline at end of file diff --git a/src/core/Akka/Actor/ActorBase.cs b/src/core/Akka/Actor/ActorBase.cs index 251cfb76157..0e8c000155e 100644 --- a/src/core/Akka/Actor/ActorBase.cs +++ b/src/core/Akka/Actor/ActorBase.cs @@ -20,8 +20,10 @@ public abstract class Status /// /// Indicates the success of some operation which has been performed /// - public class Success : Status + public sealed class Success : Status { + public static readonly Success Instance = new Success(null); + /// /// TBD /// @@ -35,19 +37,27 @@ public Success(object status) { Status = status; } + + public override string ToString() => Status is null ? "Success" : $"Success: {Status}"; } /// /// Indicates the failure of some operation that was requested and includes an /// describing the underlying cause of the problem. /// - public class Failure : Status + public sealed class Failure : Status { /// /// The cause of the failure /// public readonly Exception Cause; + /// + /// The source state of the failure + /// It can be used to send the command message back + /// + public readonly object State; + /// /// Initializes a new instance of the class. /// @@ -55,13 +65,23 @@ public class Failure : Status public Failure(Exception cause) { Cause = cause; + State = null; } - /// - public override string ToString() + /// + /// Initializes a new instance of the class. + /// + /// The cause of the failure + /// The source state of the failure + public Failure(Exception cause, object state) { - return $"Failure: {Cause}"; + Cause = cause; + State = state; } + + /// + public override string ToString() + => State is null ? $"Failure: {Cause}" : $"Failure[{State}]: {Cause}"; } } diff --git a/src/core/Akka/Actor/ActorCell.Children.cs b/src/core/Akka/Actor/ActorCell.Children.cs index db426b09e5f..163418c39db 100644 --- a/src/core/Akka/Actor/ActorCell.Children.cs +++ b/src/core/Akka/Actor/ActorCell.Children.cs @@ -265,8 +265,7 @@ protected void SetTerminated() { get { - var terminating = ChildrenContainer as TerminatingChildrenContainer; - return terminating != null && terminating.Reason is SuspendReason.IWaitingForChildren; + return ChildrenContainer is TerminatingChildrenContainer terminating && terminating.Reason is SuspendReason.IWaitingForChildren; } } @@ -325,8 +324,7 @@ public bool TryGetChildStatsByName(string name, out IChildStats child) //This /// private bool TryGetChildRestartStatsByName(string name, out ChildRestartStats child) { - IChildStats stats; - if (ChildrenContainer.TryGetByName(name, out stats)) + if (ChildrenContainer.TryGetByName(name, out var stats)) { child = stats as ChildRestartStats; if (child != null) @@ -355,56 +353,37 @@ protected bool TryGetChildStatsByRef(IActorRef actor, out ChildRestartStats chil /// /// N/A /// N/A - [Obsolete("Use TryGetSingleChild [0.7.1]")] public IInternalActorRef GetSingleChild(string name) - { - IInternalActorRef child; - return TryGetSingleChild(name, out child) ? child : ActorRefs.Nobody; - } - - /// - /// TBD - /// - /// TBD - /// TBD - /// TBD - public bool TryGetSingleChild(string name, out IInternalActorRef child) { if (name.IndexOf('#') < 0) { // optimization for the non-uid case - if (TryGetChildRestartStatsByName(name, out var stats)) + if (ChildrenContainer.TryGetByName(name, out var stats) && stats is ChildRestartStats r) { - child = stats.Child; - return true; + return r.Child; } - else if (TryGetFunctionRef(name, out var functionRef)) + + if (TryGetFunctionRef(name, out var functionRef)) { - child = functionRef; - return true; + return functionRef; } } else { var (s, uid) = GetNameAndUid(name); - if (TryGetChildRestartStatsByName(s, out var stats)) + if (TryGetChildRestartStatsByName(s, out var stats) && (uid == ActorCell.UndefinedUid || uid == stats.Uid)) { - if (uid == ActorCell.UndefinedUid || uid == stats.Uid) - { - child = stats.Child; - return true; - } + return stats.Child; } - else if (TryGetFunctionRef(s, uid, out var functionRef)) + + if (TryGetFunctionRef(s, uid, out var functionRef)) { - child = functionRef; - return true; + return functionRef; } } - child = ActorRefs.Nobody; - return false; + return ActorRefs.Nobody; } - + /// /// TBD /// diff --git a/src/core/Akka/Actor/ActorCell.cs b/src/core/Akka/Actor/ActorCell.cs index 959faa67f75..d8eabb6c149 100644 --- a/src/core/Akka/Actor/ActorCell.cs +++ b/src/core/Akka/Actor/ActorCell.cs @@ -236,12 +236,15 @@ public void Init(bool sendSupervise, MailboxType mailboxType) [Obsolete("Use TryGetChildStatsByName [0.7.1]", true)] public IInternalActorRef GetChildByName(string name) //TODO: Should return Option[ChildStats] { - return TryGetSingleChild(name, out var child) ? child : ActorRefs.Nobody; + return GetSingleChild(name); } IActorRef IActorContext.Child(string name) { - return TryGetSingleChild(name, out var child) ? child : ActorRefs.Nobody; + if (TryGetChildStatsByName(name, out var child) && child is ChildRestartStats s) + return s.Child; + + return ActorRefs.Nobody; } /// diff --git a/src/core/Akka/Actor/ActorRef.Extensions.cs b/src/core/Akka/Actor/ActorRef.Extensions.cs index e876d9744a4..ef4be2f9988 100644 --- a/src/core/Akka/Actor/ActorRef.Extensions.cs +++ b/src/core/Akka/Actor/ActorRef.Extensions.cs @@ -23,7 +23,8 @@ public static class ActorRefExtensions /// true if the is valid; otherwise false. public static bool IsNobody(this IActorRef actorRef) { - return actorRef == null || actorRef is Nobody || actorRef is DeadLetterActorRef || (actorRef.Path.Uid == 0 && actorRef.Path.Name == "deadLetters"); + return actorRef is null || actorRef is Nobody || actorRef is DeadLetterActorRef + || actorRef.Path is null || (actorRef.Path.Uid == 0 && actorRef.Path.Name == "deadLetters"); } /// diff --git a/src/core/Akka/Actor/ActorRef.cs b/src/core/Akka/Actor/ActorRef.cs index 959da6607cf..7208047ff44 100644 --- a/src/core/Akka/Actor/ActorRef.cs +++ b/src/core/Akka/Actor/ActorRef.cs @@ -19,6 +19,7 @@ using Akka.Event; using Akka.Util; using Akka.Util.Internal; +using Akka.Util.Internal.Collections; namespace Akka.Actor { @@ -68,51 +69,34 @@ public interface IRepointableRef : IActorRefScope /// /// ActorRef implementation used for one-off tasks. /// - public class FutureActorRef : MinimalActorRef + public sealed class FutureActorRef : MinimalActorRef { private readonly TaskCompletionSource _result; private readonly ActorPath _path; + private readonly IActorRefProvider _provider; /// /// INTERNAL API /// /// TBD - /// TBD /// TBD - public FutureActorRef(TaskCompletionSource result, Action unregister, ActorPath path) + /// TBD + public FutureActorRef(TaskCompletionSource result, ActorPath path, IActorRefProvider provider) { - if (ActorCell.Current != null) - { - _actorAwaitingResultSender = ActorCell.Current.Sender; - } _result = result; _path = path; - - _result.Task.ContinueWith(unregister); + _provider = provider; } /// /// TBD /// - public override ActorPath Path - { - get { return _path; } - } + public override ActorPath Path => _path; /// /// TBD /// - /// TBD - public override IActorRefProvider Provider - { - get { throw new NotImplementedException(); } - } - - - private const int INITIATED = 0; - private const int COMPLETED = 1; - private int status = INITIATED; - private readonly IActorRef _actorAwaitingResultSender; + public override IActorRefProvider Provider => _provider; /// /// TBD @@ -121,43 +105,36 @@ public override IActorRefProvider Provider /// TBD protected override void TellInternal(object message, IActorRef sender) { + var handled = false; - if (message is ISystemMessage sysM) //we have special handling for system messages - { - SendSystemMessage(sysM); - } - else + switch (message) { - if (Interlocked.Exchange(ref status, COMPLETED) == INITIATED) - { - if (message is T t) - { - _result.TrySetResult(t); - } - else if (message == null) //special case: https://github.com/akkadotnet/akka.net/issues/5204 - { - _result.TrySetResult(default); - } - else if (message is Failure f) - { - _result.TrySetException(f.Exception ?? new TaskCanceledException("Task cancelled by actor via Failure message.")); - } - else - { - _result.TrySetException(new ArgumentException( - $"Received message of type [{message.GetType()}] - Ask expected message of type [{typeof(T)}]")); - } - } + case ISystemMessage msg: + handled = _result.TrySetException(new InvalidOperationException($"system message of type '{msg.GetType().Name}' is invalid for {nameof(FutureActorRef)}")); + break; + case T t: + handled = _result.TrySetResult(t); + break; + case null: + handled = _result.TrySetResult(default); + break; + case Status.Failure f: + handled = _result.TrySetException(f.Cause + ?? new TaskCanceledException("Task cancelled by actor via Failure message.")); + break; + case Failure f: + handled = _result.TrySetException(f.Exception + ?? new TaskCanceledException("Task cancelled by actor via Failure message.")); + break; + default: + _ = _result.TrySetException(new ArgumentException( + $"Received message of type [{message.GetType()}] - Ask expected message of type [{typeof(T)}]")); + break; } - } - - /// - /// TBD - /// - /// TBD - public override void SendSystemMessage(ISystemMessage message) - { - base.SendSystemMessage(message); + + //ignore canceled ask and put unhandled answers into deadletter + if (!handled && !_result.Task.IsCanceled) + _provider.DeadLetters.Tell(message ?? default(T), this); } } @@ -422,7 +399,7 @@ public interface IInternalActorRef : IActorRef, IActorRefScope /// /// The path elements. /// The , or if the requested path does not exist, returns . - IActorRef GetChild(IEnumerable name); + IActorRef GetChild(IReadOnlyList name); /// /// Resumes an actor if it has been suspended. @@ -481,7 +458,7 @@ public abstract class InternalActorRefBase : ActorRefBase, IInternalActorRef public abstract IActorRefProvider Provider { get; } /// - public abstract IActorRef GetChild(IEnumerable name); //TODO: Refactor this to use an IEnumerator instead as this will be faster instead of enumerating multiple times over name, as the implementations currently do. + public abstract IActorRef GetChild(IReadOnlyList name); //TODO: Refactor this to use an IEnumerator instead as this will be faster instead of enumerating multiple times over name, as the implementations currently do. /// public abstract void Resume(Exception causedByFailure = null); @@ -530,9 +507,9 @@ public override IInternalActorRef Parent } /// - public override IActorRef GetChild(IEnumerable name) + public override IActorRef GetChild(IReadOnlyList name) { - if (name.All(string.IsNullOrEmpty)) + if (name.All(x => string.IsNullOrEmpty(x))) return this; return ActorRefs.Nobody; } @@ -726,16 +703,16 @@ public abstract class ActorRefWithCell : InternalActorRefBase private IEnumerable SelfAndChildren() { yield return this; - foreach(var child in Children.SelectMany(x => - { - switch(x) - { - case ActorRefWithCell cell: - return cell.SelfAndChildren(); - default: - return new[] { x }; - } - })) + foreach (var child in Children.SelectMany(x => + { + switch (x) + { + case ActorRefWithCell cell: + return cell.SelfAndChildren(); + default: + return new[] { x }; + } + })) { yield return child; } @@ -874,20 +851,17 @@ override def getChild(name: Iterator[String]): InternalActorRef = { /// /// TBD /// TBD - public override IActorRef GetChild(IEnumerable name) + public override IActorRef GetChild(IReadOnlyList name) { //Using enumerator to avoid multiple enumerations of name. - var enumerator = name.GetEnumerator(); - if (!enumerator.MoveNext()) - { - //name was empty + if (name.Count == 0) return this; - } - var firstName = enumerator.Current; + + var firstName = name[0]; if (string.IsNullOrEmpty(firstName)) return this; if (_children.TryGetValue(firstName, out var child)) - return child.GetChild(new Enumerable(enumerator)); + return child.GetChild(name.NoCopySlice(1)); return ActorRefs.Nobody; } diff --git a/src/core/Akka/Actor/ActorRefProvider.cs b/src/core/Akka/Actor/ActorRefProvider.cs index 0c9cd6dd3f1..aa6e21adb95 100644 --- a/src/core/Akka/Actor/ActorRefProvider.cs +++ b/src/core/Akka/Actor/ActorRefProvider.cs @@ -462,7 +462,7 @@ public IActorRef ResolveActorRef(ActorPath path) /// TBD /// TBD /// TBD - internal IInternalActorRef ResolveActorRef(IInternalActorRef actorRef, IReadOnlyCollection pathElements) + internal IInternalActorRef ResolveActorRef(IInternalActorRef actorRef, IReadOnlyList pathElements) { if (pathElements.Count == 0) { diff --git a/src/core/Akka/Actor/Address.cs b/src/core/Akka/Actor/Address.cs index 8ff18069456..81398855cc6 100644 --- a/src/core/Akka/Actor/Address.cs +++ b/src/core/Akka/Actor/Address.cs @@ -10,6 +10,7 @@ using System.Diagnostics; using System.Linq; using System.Text; +using System.Xml.Xsl; using Akka.Util; namespace Akka.Actor @@ -154,7 +155,7 @@ public bool Equals(Address other) } /// - public override bool Equals(object obj) => obj is Address && Equals((Address)obj); + public override bool Equals(object obj) => Equals(obj as Address); /// public override int GetHashCode() @@ -245,7 +246,7 @@ public Address WithPort(int? port = null) /// true if both addresses are equal; otherwise false public static bool operator ==(Address left, Address right) { - return Equals(left, right); + return left?.Equals(right) ?? ReferenceEquals(right, null); } /// @@ -256,7 +257,7 @@ public Address WithPort(int? port = null) /// true if both addresses are not equal; otherwise false public static bool operator !=(Address left, Address right) { - return !Equals(left, right); + return !(left == right); } /// diff --git a/src/core/Akka/Actor/Futures.cs b/src/core/Akka/Actor/Futures.cs index 20dbcedd613..c49cc8b74e4 100644 --- a/src/core/Akka/Actor/Futures.cs +++ b/src/core/Akka/Actor/Futures.cs @@ -149,16 +149,19 @@ public static Task Ask(this ICanTell self, Func message } //create a new tempcontainer path - ActorPath path = provider.TempPath(); + var path = provider.TempPath(); - var future = new FutureActorRef(result, t => + var future = new FutureActorRef(result, path, provider); + + //The future actor needs to be unregistered in the temp container + _ = result.Task.ContinueWith(t => { provider.UnregisterTempActor(path); ctr1?.Dispose(); ctr2?.Dispose(); timeoutCancellation?.Dispose(); - }, path); + }, TaskContinuationOptions.ExecuteSynchronously); //The future actor needs to be registered in the temp container provider.RegisterTempActor(future, path); diff --git a/src/core/Akka/Actor/LocalActorRef.cs b/src/core/Akka/Actor/LocalActorRef.cs index 10e4f298b46..11e9f6cad9d 100644 --- a/src/core/Akka/Actor/LocalActorRef.cs +++ b/src/core/Akka/Actor/LocalActorRef.cs @@ -12,6 +12,7 @@ using Akka.Dispatch; using Akka.Dispatch.SysMsg; using Akka.Util.Internal; +using Akka.Util.Internal.Collections; namespace Akka.Actor { @@ -52,7 +53,8 @@ public class LocalActorRef : ActorRefWithCell, ILocalRef /// TBD /// TBD /// TBD - public LocalActorRef(ActorSystemImpl system, Props props, MessageDispatcher dispatcher, MailboxType mailboxType, IInternalActorRef supervisor, ActorPath path) + public LocalActorRef(ActorSystemImpl system, Props props, MessageDispatcher dispatcher, MailboxType mailboxType, + IInternalActorRef supervisor, ActorPath path) { _system = system; _props = props; @@ -61,18 +63,19 @@ public LocalActorRef(ActorSystemImpl system, Props props, MessageDispatcher disp _supervisor = supervisor; _path = path; - /* - * Safe publication of this class’s fields is guaranteed by Mailbox.SetActor() - * which is called indirectly from ActorCell.init() (if you’re wondering why - * this is at all important, remember that under the CLR readonly fields are only - * frozen at the _end_ of the constructor, but we are publishing "this" before - * that is reached). - * This means that the result of NewActorCell needs to be written to the field - * _cell before we call init and start, since we can start using "this" - * object from another thread as soon as we run init. - */ + /* + * Safe publication of this class’s fields is guaranteed by Mailbox.SetActor() + * which is called indirectly from ActorCell.init() (if you’re wondering why + * this is at all important, remember that under the CLR readonly fields are only + * frozen at the _end_ of the constructor, but we are publishing "this" before + * that is reached). + * This means that the result of NewActorCell needs to be written to the field + * _cell before we call init and start, since we can start using "this" + * object from another thread as soon as we run init. + */ // ReSharper disable once VirtualMemberCallInConstructor - _cell = NewActorCell(_system, this, _props, _dispatcher, _supervisor); // _cell needs to be assigned before Init is called. + _cell = NewActorCell(_system, this, _props, _dispatcher, + _supervisor); // _cell needs to be assigned before Init is called. _cell.Init(true, MailboxType); } @@ -206,19 +209,17 @@ protected override void TellInternal(object message, IActorRef sender) /// public override IInternalActorRef GetSingleChild(string name) { - IInternalActorRef child; - return _cell.TryGetSingleChild(name, out child) ? child : ActorRefs.Nobody; + return _cell.GetSingleChild(name); } /// - public override IActorRef GetChild(IEnumerable name) + public override IActorRef GetChild(IReadOnlyList name) { - var current = (IActorRef)this; + IActorRef current = this; int index = 0; - foreach (string element in name) + foreach (var element in name) { - var currentLocalActorRef = current as LocalActorRef; - if (currentLocalActorRef != null) + if (current is LocalActorRef currentLocalActorRef) { switch (element) { @@ -232,20 +233,21 @@ public override IActorRef GetChild(IEnumerable name) break; } } - else + else if (current is IInternalActorRef internalActorRef) { //Current is not a LocalActorRef - if (current != null) - { - var rest = name.Skip(index).ToList(); - return current.AsInstanceOf().GetChild(rest); - } + var rest = name.NoCopySlice(index); + return internalActorRef.GetChild(rest); + } + else // not a LocalActorRef or an IInternalActorRef + { throw new NotSupportedException("Bug, we should not get here"); } + index++; } + return current; } } -} - +} \ No newline at end of file diff --git a/src/core/Akka/Actor/RepointableActorRef.cs b/src/core/Akka/Actor/RepointableActorRef.cs index 6933f23da21..31ccb2f872e 100644 --- a/src/core/Akka/Actor/RepointableActorRef.cs +++ b/src/core/Akka/Actor/RepointableActorRef.cs @@ -15,6 +15,7 @@ using Akka.Dispatch.SysMsg; using Akka.Event; using Akka.Pattern; +using Akka.Util.Internal.Collections; namespace Akka.Actor { @@ -278,17 +279,16 @@ protected override void TellInternal(object message, IActorRef sender) /// /// TBD /// TBD - public override IActorRef GetChild(IEnumerable name) + public override IActorRef GetChild(IReadOnlyList name) { - var current = (IActorRef)this; - if (!name.Any()) return current; + if (name.Count == 0) return this; - var next = name.FirstOrDefault() ?? ""; + var next = name[0]; switch (next) { case "..": - return Parent.GetChild(name.Skip(1)); + return Parent.GetChild(name.NoCopySlice(1)); case "": return ActorRefs.Nobody; default: @@ -297,8 +297,8 @@ public override IActorRef GetChild(IEnumerable name) { if (stats is ChildRestartStats crs && (uid == ActorCell.UndefinedUid || uid == crs.Uid)) { - if (name.Skip(1).Any()) - return crs.Child.GetChild(name.Skip(1)); + if (name.Count > 1) + return crs.Child.GetChild(name.NoCopySlice(1)); else return crs.Child; } diff --git a/src/core/Akka/Actor/Settings.cs b/src/core/Akka/Actor/Settings.cs index 906c016e3a5..1ecf387cbdc 100644 --- a/src/core/Akka/Actor/Settings.cs +++ b/src/core/Akka/Actor/Settings.cs @@ -11,6 +11,7 @@ using Akka.Actor.Setup; using Akka.Configuration; using Akka.Dispatch; +using Akka.Event; using Akka.Routing; using ConfigurationFactory = Akka.Configuration.ConfigurationFactory; @@ -109,6 +110,31 @@ public Settings(ActorSystem system, Config config, ActorSystemSetup setup) LogLevel = Config.GetString("akka.loglevel", null); StdoutLogLevel = Config.GetString("akka.stdout-loglevel", null); + + var stdoutClassName = Config.GetString("akka.stdout-logger-class", null); + if (string.IsNullOrWhiteSpace(stdoutClassName)) + { + StdoutLogger = new StandardOutLogger(); + } + else + { + var stdoutLoggerType = Type.GetType(stdoutClassName); + if (stdoutLoggerType == null) + throw new ArgumentException($"Could not load type of {stdoutClassName} for standard out logger."); + if(!typeof(MinimalLogger).IsAssignableFrom(stdoutLoggerType)) + throw new ArgumentException("Standard out logger type must inherit from the MinimalLogger abstract class."); + + try + { + StdoutLogger = (MinimalLogger)Activator.CreateInstance(stdoutLoggerType); + } + catch (MissingMethodException) + { + throw new MissingMethodException( + "Standard out logger type must inherit from the MinimalLogger abstract class and have an empty constructor."); + } + } + Loggers = Config.GetStringList("akka.loggers", new string[] { }); LoggersDispatcher = Config.GetString("akka.loggers-dispatcher", null); LoggerStartTimeout = Config.GetTimeSpan("akka.logger-startup-timeout", null); @@ -247,6 +273,11 @@ public Settings(ActorSystem system, Config config, ActorSystemSetup setup) /// The stdout log level. public string StdoutLogLevel { get; private set; } + /// + /// Returns a singleton instance of the standard out logger. + /// + public MinimalLogger StdoutLogger { get; } + /// /// Gets the loggers. /// diff --git a/src/core/Akka/Actor/SupervisorStrategy.cs b/src/core/Akka/Actor/SupervisorStrategy.cs index b3faecb02eb..7d2e5d529b4 100644 --- a/src/core/Akka/Actor/SupervisorStrategy.cs +++ b/src/core/Akka/Actor/SupervisorStrategy.cs @@ -750,6 +750,7 @@ public override int GetHashCode() /// /// Collection of failures, used to keep track of how many times a given actor has failed. /// + [Obsolete("Use List of Akka.Actor.Status.Failure")] public class Failures { /// @@ -769,6 +770,7 @@ public Failures() /// /// Represents a single failure. /// + [Obsolete("Use Akka.Actor.Status.Failure")] public class Failure { /// diff --git a/src/core/Akka/Configuration/Pigeon.conf b/src/core/Akka/Configuration/Pigeon.conf index a2d17b96a50..49a1fce3478 100644 --- a/src/core/Akka/Configuration/Pigeon.conf +++ b/src/core/Akka/Configuration/Pigeon.conf @@ -42,6 +42,10 @@ akka { # Log level for the very basic logger activated during AkkaApplication startup # Options: OFF, ERROR, WARNING, INFO, DEBUG stdout-loglevel = "WARNING" + + # Fully qualified class name (FQCN) of the very basic logger used on startup and shutdown. + # You can substitute the logger by supplying an FQCN to a custom class that implements Akka.Event.MinimumLogger + stdout-logger-class = "Akka.Event.StandardOutLogger" # Log the complete configuration at INFO level when the actor system is started. # This is useful when you are uncertain of what configuration is used. diff --git a/src/core/Akka/Event/DefaultLogger.cs b/src/core/Akka/Event/DefaultLogger.cs index 7d03e27dfef..730b238138b 100644 --- a/src/core/Akka/Event/DefaultLogger.cs +++ b/src/core/Akka/Event/DefaultLogger.cs @@ -16,6 +16,8 @@ namespace Akka.Event /// public class DefaultLogger : ActorBase, IRequiresMessageQueue { + private MinimalLogger _stdoutLogger; + /// /// TBD /// @@ -23,17 +25,18 @@ public class DefaultLogger : ActorBase, IRequiresMessageQueueTBD protected override bool Receive(object message) { - if (message is InitializeLogger) + switch (message) { - Sender.Tell(new LoggerInitialized()); - return true; + case InitializeLogger _: + _stdoutLogger = Context.System.Settings.StdoutLogger; + Sender.Tell(new LoggerInitialized()); + return true; + case LogEvent logEvent: + Print(logEvent); + return true; + default: + return false; } - var logEvent = message as LogEvent; - if (logEvent == null) - return false; - - Print(logEvent); - return true; } /// @@ -42,7 +45,10 @@ protected override bool Receive(object message) /// The log event that is to be output. protected virtual void Print(LogEvent logEvent) { - StandardOutLogger.PrintLogEvent(logEvent); + if (_stdoutLogger == null) + throw new Exception("Logger has not been initialized yet."); + + _stdoutLogger.Tell(logEvent); } } } diff --git a/src/core/Akka/Event/LoggerInitialized.cs b/src/core/Akka/Event/LoggerInitialized.cs index d7aa17a92ec..2fe9b3e0222 100644 --- a/src/core/Akka/Event/LoggerInitialized.cs +++ b/src/core/Akka/Event/LoggerInitialized.cs @@ -12,7 +12,7 @@ namespace Akka.Event /// /// This class represents a message used to notify subscribers that a logger has been initialized. /// - public class LoggerInitialized : INoSerializationVerificationNeeded + public class LoggerInitialized : INoSerializationVerificationNeeded, IDeadLetterSuppression { } } diff --git a/src/core/Akka/Event/LoggerMailbox.cs b/src/core/Akka/Event/LoggerMailbox.cs index fb8e5d7b518..51e32e25e87 100644 --- a/src/core/Akka/Event/LoggerMailbox.cs +++ b/src/core/Akka/Event/LoggerMailbox.cs @@ -82,15 +82,15 @@ void IMessageQueue.CleanUp(IActorRef owner, IMessageQueue deadletters) { if (HasMessages) { - Envelope envelope; - + var logger = _system.Settings.StdoutLogger; + // Drain all remaining messages to the StandardOutLogger. // CleanUp is called after switching out the mailbox, which is why // this kind of look works without a limit. - while (TryDequeue(out envelope)) + while (TryDequeue(out var envelope)) { // Logging.StandardOutLogger is a MinimalActorRef, i.e. not a "real" actor - Logging.StandardOutLogger.Tell(envelope.Message, envelope.Sender); + logger.Tell(envelope.Message, envelope.Sender); } } MessageQueue.CleanUp(owner, deadletters); diff --git a/src/core/Akka/Event/Logging.cs b/src/core/Akka/Event/Logging.cs index 05d334b8eac..a1104ed3dd5 100644 --- a/src/core/Akka/Event/Logging.cs +++ b/src/core/Akka/Event/Logging.cs @@ -147,11 +147,6 @@ public static string SimpleName(Type t) private const string Off = "OFF"; private const LogLevel OffLogLevel = (LogLevel) int.MaxValue; - /// - /// Returns a singleton instance of the standard out logger. - /// - public static readonly StandardOutLogger StandardOutLogger = new StandardOutLogger(); - /// /// Retrieves the log event class associated with the specified log level. /// diff --git a/src/core/Akka/Event/LoggingBus.cs b/src/core/Akka/Event/LoggingBus.cs index 2282eb6b3f8..fbe8e70149b 100644 --- a/src/core/Akka/Event/LoggingBus.cs +++ b/src/core/Akka/Event/LoggingBus.cs @@ -102,7 +102,7 @@ internal void StartDefaultLoggers(ActorSystemImpl system) throw new ConfigurationException($@"Logger specified in config cannot be found: ""{strLoggerType}"""); } - if (loggerType == typeof(StandardOutLogger)) + if (typeof(MinimalLogger).IsAssignableFrom(loggerType)) { shouldRemoveStandardOutLogger = false; continue; @@ -143,8 +143,9 @@ internal void StartDefaultLoggers(ActorSystemImpl system) if (shouldRemoveStandardOutLogger) { - Publish(new Debug(logName, GetType(), "StandardOutLogger being removed")); - Unsubscribe(Logging.StandardOutLogger); + var stdOutLogger = system.Settings.StdoutLogger; + Publish(new Debug(logName, GetType(), $"{Logging.SimpleName(stdOutLogger)} being removed")); + Unsubscribe(stdOutLogger); } Publish(new Debug(logName, GetType(), "Default Loggers started")); @@ -157,19 +158,18 @@ internal void StartDefaultLoggers(ActorSystemImpl system) internal void StopDefaultLoggers(ActorSystem system) { var level = LogLevel; // volatile access before reading loggers - if (!_loggers.Any(c => c is StandardOutLogger)) + if (!_loggers.Any(c => c is MinimalLogger)) { SetUpStdoutLogger(system.Settings); - Publish(new Debug(SimpleName(this), GetType(), "Shutting down: StandardOutLogger started")); + Publish(new Debug(SimpleName(this), GetType(), $"Shutting down: {Logging.SimpleName(system.Settings.StdoutLogger)} started")); } foreach (var logger in _loggers) { - if (!(logger is StandardOutLogger)) + if (!(logger is MinimalLogger)) { Unsubscribe(logger); - var internalActorRef = logger as IInternalActorRef; - if (internalActorRef != null) + if (logger is IInternalActorRef internalActorRef) { internalActorRef.Stop(); } @@ -214,9 +214,9 @@ private string CreateLoggerName(Type actorClass) } /// - /// Starts the logger. + /// Starts the logger. /// - /// The configuration used to configure the . + /// The configuration used to configure the . public void StartStdoutLogger(Settings config) { SetUpStdoutLogger(config); @@ -226,7 +226,7 @@ public void StartStdoutLogger(Settings config) private void SetUpStdoutLogger(Settings config) { var logLevel = Logging.LogLevelFor(config.StdoutLogLevel); - SubscribeLogLevelAndAbove(logLevel, Logging.StandardOutLogger); + SubscribeLogLevelAndAbove(logLevel, config.StdoutLogger); } /// diff --git a/src/core/Akka/Event/StandardOutLogger.cs b/src/core/Akka/Event/StandardOutLogger.cs index 6493c136fc4..b213d0e4bac 100644 --- a/src/core/Akka/Event/StandardOutLogger.cs +++ b/src/core/Akka/Event/StandardOutLogger.cs @@ -13,6 +13,38 @@ namespace Akka.Event { + public abstract class MinimalLogger : MinimalActorRef + { + /// + /// N/A + /// + /// This exception is automatically thrown since does not support this property. + public sealed override IActorRefProvider Provider + { + get { throw new NotSupportedException("This logger does not provide."); } + } + + /// + /// The path where this logger currently resides. + /// + public sealed override ActorPath Path { get; } = new RootActorPath(Address.AllSystems, "/StandardOutLogger"); + + protected sealed override void TellInternal(object message, IActorRef sender) + { + switch (message) + { + case null: + throw new ArgumentNullException(nameof(message), "The message to log must not be null."); + + default: + Log(message); + break; + } + } + + protected abstract void Log(object message); + } + /// /// This class represents an event logger that logs its messages to standard output (e.g. the console). /// @@ -21,9 +53,8 @@ namespace Akka.Event /// even before normal logging is started. /// /// - public class StandardOutLogger : MinimalActorRef + public class StandardOutLogger : MinimalLogger { - private readonly ActorPath _path = new RootActorPath(Address.AllSystems, "/StandardOutLogger"); /// /// Initializes the class. @@ -37,23 +68,6 @@ static StandardOutLogger() UseColors = true; } - /// - /// N/A - /// - /// This exception is automatically thrown since does not support this property. - public override IActorRefProvider Provider - { - get { throw new NotSupportedException("This logger does not provide."); } - } - - /// - /// The path where this logger currently resides. - /// - public override ActorPath Path - { - get { return _path; } - } - /// /// Handles incoming log events by printing them to the console. /// @@ -62,22 +76,20 @@ public override ActorPath Path /// /// This exception is thrown if the given is undefined. /// - protected override void TellInternal(object message, IActorRef sender) + protected override void Log(object message) { - if(message == null) - throw new ArgumentNullException(nameof(message), "The message to log must not be null."); - - var logEvent = message as LogEvent; - if (logEvent != null) + switch (message) { - PrintLogEvent(logEvent); - } - else - { - Console.WriteLine(message); + case LogEvent logEvent: + PrintLogEvent(logEvent); + break; + + default: + Console.WriteLine(message); + break; } } - + /// /// The foreground color to use when printing Debug events to the console. /// @@ -107,7 +119,7 @@ protected override void TellInternal(object message, IActorRef sender) /// Prints a specified event to the console. /// /// The event to print - public static void PrintLogEvent(LogEvent logEvent) + internal static void PrintLogEvent(LogEvent logEvent) { try { diff --git a/src/core/Akka/IO/Dns.cs b/src/core/Akka/IO/Dns.cs index b510f51331f..7d31041321b 100644 --- a/src/core/Akka/IO/Dns.cs +++ b/src/core/Akka/IO/Dns.cs @@ -7,9 +7,11 @@ using System; using System.Collections.Generic; +using System.Collections.Immutable; using System.Linq; using System.Net; using System.Net.Sockets; +using System.Runtime.ExceptionServices; using Akka.Actor; using Akka.Configuration; using Akka.Routing; @@ -94,6 +96,11 @@ public class Resolved : Command { private readonly IPAddress _addr; + public Resolved(string name, Exception ex) : this(name, null, null) + { + Exception = ex; + } + /// /// TBD /// @@ -103,24 +110,28 @@ public class Resolved : Command public Resolved(string name, IEnumerable ipv4, IEnumerable ipv6) { Name = name; - Ipv4 = ipv4; - Ipv6 = ipv6; + Ipv4 = ipv4?.ToImmutableList() ?? ImmutableList.Empty; + Ipv6 = ipv6?.ToImmutableList() ?? ImmutableList.Empty; - _addr = ipv4.FirstOrDefault() ?? ipv6.FirstOrDefault(); + _addr = Ipv4.FirstOrDefault() ?? Ipv6.FirstOrDefault(); } + public bool IsSuccess => Exception == null; + + public Exception Exception { get; } + /// /// TBD /// - public string Name { get; private set; } + public string Name { get; } /// /// TBD /// - public IEnumerable Ipv4 { get; private set; } + public IEnumerable Ipv4 { get; } /// /// TBD /// - public IEnumerable Ipv6 { get; private set; } + public IEnumerable Ipv6 { get; } /// /// TBD @@ -129,8 +140,11 @@ public IPAddress Addr { get { - //TODO: Throw better exception - if (_addr == null) throw new Exception("Unknown host"); + if(Exception != null) + ExceptionDispatchInfo.Capture(Exception).Throw(); + else + if (_addr == null) throw new Exception("Unknown host"); + return _addr; } } diff --git a/src/core/Akka/IO/InetAddressDnsResolver.cs b/src/core/Akka/IO/InetAddressDnsResolver.cs index 91d4ed68b4e..b57aec6b5ad 100644 --- a/src/core/Akka/IO/InetAddressDnsResolver.cs +++ b/src/core/Akka/IO/InetAddressDnsResolver.cs @@ -8,6 +8,8 @@ using System.Linq; using System.Net; using System.Net.Sockets; +using System.Runtime.ExceptionServices; +using System.Threading.Tasks; using Akka.Actor; using Akka.Configuration; @@ -43,28 +45,33 @@ public InetAddressDnsResolver(SimpleDnsCache cache, Config config) /// TBD protected override bool Receive(object message) { - var resolve = message as Dns.Resolve; - if (resolve != null) + if (message is Dns.Resolve resolve) { + var replyTo = Sender; var answer = _cache.Cached(resolve.Name); - if (answer == null) + if (answer != null) { - try - { - //TODO: IP6 - answer = Dns.Resolved.Create(resolve.Name, System.Net.Dns.GetHostEntryAsync(resolve.Name).Result.AddressList.Where(x => - x.AddressFamily == AddressFamily.InterNetwork - || _useIpv6 && x.AddressFamily == AddressFamily.InterNetworkV6)); - _cache.Put(answer, _positiveTtl); - } - catch (SocketException ex) + replyTo.Tell(answer); + return true; + } + + System.Net.Dns.GetHostEntryAsync(resolve.Name).ContinueWith(t => + { + if (t.IsFaulted) { - if (ex.SocketErrorCode != SocketError.HostNotFound) throw; - answer = new Dns.Resolved(resolve.Name, Enumerable.Empty(), Enumerable.Empty()); - _cache.Put(answer, _negativeTtl); + var flattened = t.Exception.Flatten().InnerExceptions; + return flattened.Count == 1 + ? new Dns.Resolved(resolve.Name, flattened[0]) + : new Dns.Resolved(resolve.Name, t.Exception); } - } - Sender.Tell(answer); + + answer = Dns.Resolved.Create(resolve.Name, t.Result.AddressList.Where(x => + x.AddressFamily == AddressFamily.InterNetwork + || _useIpv6 && x.AddressFamily == AddressFamily.InterNetworkV6)); + _cache.Put(answer, _positiveTtl); + return answer; + + }, TaskContinuationOptions.ExecuteSynchronously).PipeTo(replyTo); return true; } return false; diff --git a/src/core/Akka/Routing/Router.cs b/src/core/Akka/Routing/Router.cs index 1457d1baa80..b9a2916d9c3 100644 --- a/src/core/Akka/Routing/Router.cs +++ b/src/core/Akka/Routing/Router.cs @@ -327,7 +327,7 @@ protected object UnWrap(object message) /// /// The message to send. /// The sender of this message - which will be propagated to the routee. - public void Route(object message, IActorRef sender) + public virtual void Route(object message, IActorRef sender) { if (message is Broadcast) { diff --git a/src/core/Akka/Util/Internal/Collections/ListSlice.cs b/src/core/Akka/Util/Internal/Collections/ListSlice.cs new file mode 100644 index 00000000000..5614b6a05b7 --- /dev/null +++ b/src/core/Akka/Util/Internal/Collections/ListSlice.cs @@ -0,0 +1,182 @@ +// //----------------------------------------------------------------------- +// // +// // Copyright (C) 2009-2021 Lightbend Inc. +// // Copyright (C) 2013-2021 .NET Foundation +// // +// //----------------------------------------------------------------------- + +using System; +using System.Collections; +using System.Collections.Generic; +using System.Linq; + +namespace Akka.Util.Internal.Collections +{ + internal static class SliceExtensions + { + public static IReadOnlyList NoCopySlice(this IReadOnlyList list, int offset) + { + switch (list) + { + // slice of a slice + case ListSlice slice: + { + return new ListSlice(slice.Array, slice.Offset + offset, slice.Array.Count - slice.Offset - offset); + } + default: + { + return new ListSlice(list, offset, list.Count - offset); + } + } + } + } + + /// + /// but for + /// + /// + internal struct ListSlice : IList, IReadOnlyList + { + private sealed class SliceEnumerator : IEnumerator + { + private readonly IReadOnlyList _array; + private readonly int _start; + private readonly int _end; + private int _current; + + public SliceEnumerator(ListSlice array) + { + _array = array._array; + _start = array.Offset; + _end = _start + array.Count; + _current = _start - 1; + } + + public bool MoveNext() + { + if (_current < _end) + { + _current++; + return (_current < _end); + } + return false; + } + + public void Reset() + { + _current = _start - 1; + } + + public T Current + { + get + { + if (_current < _start) throw new InvalidOperationException("Enumeration not started."); + if (_current >= _end) throw new InvalidOperationException("Enumeration ended."); + return _array[_current]; + } + } + + object IEnumerator.Current => Current; + + public void Dispose() + { + + } + } + + private readonly IReadOnlyList _array; + + public ListSlice(IReadOnlyList array) + { + + if (array == null) + throw new ArgumentNullException(nameof(array)); + + _array = array; + Offset = 0; + Count = array.Count; + } + + public ListSlice(IReadOnlyList array, int offset, int count) + { + if (array == null) + throw new ArgumentNullException(nameof(array)); + if (offset < 0) + throw new ArgumentOutOfRangeException(nameof(offset), "Cannot be below zero."); + if (count < 0) + throw new ArgumentOutOfRangeException(nameof(count), "Cannot be below zero."); + + _array = array; + Offset = offset; + Count = count; + } + + public int Offset { get; } + + public IReadOnlyList Array => _array; + + public IEnumerator GetEnumerator() + { + return new SliceEnumerator(this); + } + + IEnumerator IEnumerable.GetEnumerator() + { + return GetEnumerator(); + } + + public void Add(T item) + { + throw new System.NotImplementedException(); + } + + public void Clear() + { + throw new System.NotImplementedException(); + } + + public bool Contains(T item) + { + throw new System.NotImplementedException(); + } + + public void CopyTo(T[] array, int arrayIndex) + { + var n = 0; + foreach (var i in _array.Skip(Offset).Take(Count)) + { + array[arrayIndex + n++] = i; + } + } + + public bool Remove(T item) + { + throw new System.NotImplementedException(); + } + + public int Count { get; } + + public bool IsReadOnly => true; + public int IndexOf(T item) + { + throw new System.NotImplementedException(); + } + + public void Insert(int index, T item) + { + throw new System.NotImplementedException(); + } + + public void RemoveAt(int index) + { + throw new System.NotImplementedException(); + } + + public T this[int index] + { + get => _array[Offset + index]; + set => throw new System.NotImplementedException(); + } + } +} \ No newline at end of file