From 1e97890eff283cc1ceb3c9356d049f3640a6cd05 Mon Sep 17 00:00:00 2001
From: Aaron Stannard <aaron@petabridge.com>
Date: Fri, 11 Mar 2016 09:08:53 -0800
Subject: [PATCH] Fixed major bug with EndpointWriter adaptive backoff Fixed
 issue with PlayerFSM that broke Throttle in MNTK; Added TestConductorSpec
 cleaned up referential equality inside RemoteWatcher

---
 .../CoreAPISpec.ApproveCore.approved.txt      |   1 +
 .../Akka.MultiNodeTestRunner.csproj           |   4 +
 src/core/Akka.MultiNodeTestRunner/Program.cs  |   5 +
 src/core/Akka.Remote.TestKit/Conductor.cs     |  35 ++--
 .../Internals/Reference.conf                  |   4 +-
 src/core/Akka.Remote.TestKit/MsgEncoder.cs    |   5 +-
 src/core/Akka.Remote.TestKit/MultiNodeSpec.cs |   6 +-
 src/core/Akka.Remote.TestKit/Player.cs        |  21 ++-
 .../Akka.Remote.Tests.MultiNode.csproj        |   1 +
 .../RemoteRandomSpec.cs                       |   2 +-
 .../RemoteRoundRobinSpec.cs                   |   2 +-
 .../TestConductor/TestConductorSpec.cs        | 155 ++++++++++++++++++
 .../Transport/AkkaProtocolStressTest.cs       |   2 +-
 src/core/Akka.Remote/Endpoint.cs              |   2 +-
 src/core/Akka.Remote/RemoteWatcher.cs         |   6 +-
 .../Transport/ThrottleTransportAdapter.cs     |   8 +-
 .../Akka.Tests/Actor/FSMTransitionSpec.cs     |   2 +
 src/core/Akka/Configuration/Config.cs         |   5 +
 .../Samples.Cluster.Transformation/App.config |   2 +-
 .../Cluster/Samples.Cluster.Simple/App.config |   2 +-
 20 files changed, 229 insertions(+), 41 deletions(-)
 create mode 100644 src/core/Akka.Remote.Tests.MultiNode/TestConductor/TestConductorSpec.cs

diff --git a/src/core/Akka.API.Tests/CoreAPISpec.ApproveCore.approved.txt b/src/core/Akka.API.Tests/CoreAPISpec.ApproveCore.approved.txt
index 5c9a0d7af01..fc395ee5843 100644
--- a/src/core/Akka.API.Tests/CoreAPISpec.ApproveCore.approved.txt
+++ b/src/core/Akka.API.Tests/CoreAPISpec.ApproveCore.approved.txt
@@ -1928,6 +1928,7 @@ namespace Akka.Configuration
 {
     public class Config
     {
+        public static readonly Akka.Configuration.Config Empty;
         public Config() { }
         public Config(Akka.Configuration.Hocon.HoconRoot root) { }
         public Config(Akka.Configuration.Config source, Akka.Configuration.Config fallback) { }
diff --git a/src/core/Akka.MultiNodeTestRunner/Akka.MultiNodeTestRunner.csproj b/src/core/Akka.MultiNodeTestRunner/Akka.MultiNodeTestRunner.csproj
index 3c2826de55f..a37b2a17269 100644
--- a/src/core/Akka.MultiNodeTestRunner/Akka.MultiNodeTestRunner.csproj
+++ b/src/core/Akka.MultiNodeTestRunner/Akka.MultiNodeTestRunner.csproj
@@ -90,6 +90,10 @@
     <Error Condition="!Exists('$(SolutionDir)\.nuget\NuGet.targets')" Text="$([System.String]::Format('$(ErrorText)', '$(SolutionDir)\.nuget\NuGet.targets'))" />
     <Error Condition="!Exists('..\..\packages\xunit.core.2.0.0\build\portable-net45+win+wpa81+wp80+monotouch+monoandroid+Xamarin.iOS\xunit.core.props')" Text="$([System.String]::Format('$(ErrorText)', '..\..\packages\xunit.core.2.0.0\build\portable-net45+win+wpa81+wp80+monotouch+monoandroid+Xamarin.iOS\xunit.core.props'))" />
   </Target>
+  <PropertyGroup>
+    <PreBuildEvent>
+    </PreBuildEvent>
+  </PropertyGroup>
   <!-- To modify your build process, add your task inside one of the targets below and uncomment it. 
        Other similar extension points exist, see Microsoft.Common.targets.
   <Target Name="BeforeBuild">
diff --git a/src/core/Akka.MultiNodeTestRunner/Program.cs b/src/core/Akka.MultiNodeTestRunner/Program.cs
index 7b35d523868..49cfc303471 100644
--- a/src/core/Akka.MultiNodeTestRunner/Program.cs
+++ b/src/core/Akka.MultiNodeTestRunner/Program.cs
@@ -200,6 +200,11 @@ static void Main(string[] args)
             //Block until all Sinks have been terminated.
             TestRunSystem.WhenTerminated.Wait(TimeSpan.FromMinutes(1));
 
+            if (Debugger.IsAttached)
+            {
+                Console.ReadLine(); //block when debugging
+            }
+
             //Return the proper exit code
             Environment.Exit(ExitCodeContainer.ExitCode);
         }
diff --git a/src/core/Akka.Remote.TestKit/Conductor.cs b/src/core/Akka.Remote.TestKit/Conductor.cs
index 635b961181b..3b5cf66cf34 100644
--- a/src/core/Akka.Remote.TestKit/Conductor.cs
+++ b/src/core/Akka.Remote.TestKit/Conductor.cs
@@ -8,6 +8,7 @@
 using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.Threading;
 using System.Threading.Tasks;
 using Akka.Actor;
 using Akka.Configuration;
@@ -65,17 +66,23 @@ public IActorRef Controller
         public async Task<INode> StartController(int participants, RoleName name, INode controllerPort)
         {
             if(_controller != null) throw new IllegalStateException("TestConductorServer was already started");
-            _controller = _system.ActorOf(new Props(typeof (Controller), new object[] {participants, controllerPort}),
-                "controller");
+            _controller = _system.ActorOf(Props.Create(() => new Controller(participants, controllerPort)),
+               "controller");
             //TODO: Need to review this async stuff
-            var node = await _controller.Ask<INode>(TestKit.Controller.GetSockAddr.Instance).ConfigureAwait(false);
+            var node = await _controller.Ask<INode>(TestKit.Controller.GetSockAddr.Instance, Settings.QueryTimeout).ConfigureAwait(false);
             await StartClient(name, node).ConfigureAwait(false);
             return node;
         }
 
+        /// <summary>
+        /// Obtain the port to which the controller’s socket is actually bound. This
+        /// will deviate from the configuration in `akka.testconductor.port` in case
+        /// that was given as zero.
+        /// </summary>
+        /// <returns>The address of the controller's socket endpoint</returns>
         public Task<INode> SockAddr()
         {
-            return _controller.Ask<INode>(TestKit.Controller.GetSockAddr.Instance);
+            return Controller.Ask<INode>(TestKit.Controller.GetSockAddr.Instance, Settings.QueryTimeout);
         }
 
         /// <summary>
@@ -104,7 +111,7 @@ public Task<Done> Throttle(RoleName node, RoleName target, ThrottleTransportAdap
             float rateMBit)
         {
             RequireTestConductorTransport();
-            return Controller.Ask<Done>(new Throttle(node, target, direction, rateMBit));
+            return Controller.Ask<Done>(new Throttle(node, target, direction, rateMBit), Settings.QueryTimeout);
         }
 
         /// <summary>
@@ -128,9 +135,10 @@ public Task<Done> Blackhole(RoleName node, RoleName target, ThrottleTransportAda
 
         private void RequireTestConductorTransport()
         {
+            // Verifies that the Throttle and  FailureInjector TransportAdapters are active
             if(!Transport.DefaultAddress.Protocol.Contains(".trttl.gremlin."))
                 throw new ConfigurationException("To use this feature you must activate the failure injector adapters " +
-                    "(trttl, gremlin) by specifying `testTransport(on = true)` in your MultiNodeConfig.");
+                    "(trttl, gremlin) by specifying `TestTransport(on = true)` in your MultiNodeConfig.");
         }
 
         /// <summary>
@@ -160,7 +168,7 @@ public Task<Done> PassThrough(RoleName node, RoleName target, ThrottleTransportA
         /// <returns></returns>
         public Task<Done> Disconnect(RoleName node, RoleName target)
         {
-            return Controller.Ask<Done>(new Disconnect(node, target, false));
+            return Controller.Ask<Done>(new Disconnect(node, target, false), Settings.QueryTimeout);
         }
 
         /// <summary>
@@ -173,7 +181,7 @@ public Task<Done> Disconnect(RoleName node, RoleName target)
         /// <returns></returns>
         public Task<Done> Abort(RoleName node, RoleName target)
         {
-            return Controller.Ask<Done>(new Disconnect(node, target, true));
+            return Controller.Ask<Done>(new Disconnect(node, target, true), Settings.QueryTimeout);
         }
 
         /// <summary>
@@ -187,7 +195,7 @@ public Task<Done> Exit(RoleName node, int exitValue)
         {
             // the recover is needed to handle ClientDisconnectedException exception,
             // which is normal during shutdown
-            return Controller.Ask(new Terminate(node, new Right<bool, int>(exitValue))).ContinueWith(t =>
+            return Controller.Ask(new Terminate(node, new Right<bool, int>(exitValue)), Settings.QueryTimeout).ContinueWith(t =>
             {
                 if(t.Result is Done) return Done.Instance;
                 var failure = t.Result as FSMBase.Failure;
@@ -209,7 +217,7 @@ public Task<Done> Shutdown(RoleName node, bool abort = false)
         {
             // the recover is needed to handle ClientDisconnectedException exception,
             // which is normal during shutdown
-            return Controller.Ask(new Terminate(node, new Left<bool, int>(abort))).ContinueWith(t =>
+            return Controller.Ask(new Terminate(node, new Left<bool, int>(abort)), Settings.QueryTimeout).ContinueWith(t =>
             {
                 if (t.Result is Done) return Done.Instance;
                 var failure = t.Result as FSMBase.Failure;
@@ -224,7 +232,7 @@ public Task<Done> Shutdown(RoleName node, bool abort = false)
         /// </summary>
         public Task<IEnumerable<RoleName>> GetNodes()
         {
-            return Controller.Ask<IEnumerable<RoleName>>(TestKit.Controller.GetNodes.Instance);
+            return Controller.Ask<IEnumerable<RoleName>>(TestKit.Controller.GetNodes.Instance, Settings.QueryTimeout);
         }
 
         /// <summary>
@@ -237,7 +245,7 @@ public Task<IEnumerable<RoleName>> GetNodes()
         /// <returns></returns>
         public Task<Done> RemoveNode(RoleName node)
         {
-            return Controller.Ask<Done>(new Remove(node));
+            return Controller.Ask<Done>(new Remove(node), Settings.QueryTimeout);
         }
     }
 
@@ -262,7 +270,8 @@ public ConductorHandler(IActorRef controller, ILoggingAdapter log)
         public async void OnConnect(INode remoteAddress, IConnection responseChannel)
         {
             _log.Debug("connection from {0}", responseChannel.RemoteHost);
-            //TODO: Seems wrong to create new RemoteConnection here
+            
+            // Duration of this Ask operation needs to be infinite
             var fsm = await _controller.Ask<IActorRef>(new Controller.CreateServerFSM(new RemoteConnection(responseChannel, this)), TimeSpan.FromMilliseconds(Int32.MaxValue));
             _clients.AddOrUpdate(responseChannel, fsm, (connection, @ref) => fsm);
         }
diff --git a/src/core/Akka.Remote.TestKit/Internals/Reference.conf b/src/core/Akka.Remote.TestKit/Internals/Reference.conf
index 18309fa6b7c..d1fb1a8de89 100644
--- a/src/core/Akka.Remote.TestKit/Internals/Reference.conf
+++ b/src/core/Akka.Remote.TestKit/Internals/Reference.conf
@@ -43,7 +43,7 @@ akka {
         pool-size-factor = 1.0
 
         # Max number of threads to cap factor-based number to
-        pool-size-max = 2
+        pool-size-max = 1
       }
 
       # (I&O) Used to configure the number of I/O worker threads on client sockets
@@ -58,7 +58,7 @@ akka {
         pool-size-factor = 1.0
 
         # Max number of threads to cap factor-based number to
-        pool-size-max = 2
+        pool-size-max = 1
       }
     }
   }
diff --git a/src/core/Akka.Remote.TestKit/MsgEncoder.cs b/src/core/Akka.Remote.TestKit/MsgEncoder.cs
index 88925188c7c..33f22c0d2a0 100644
--- a/src/core/Akka.Remote.TestKit/MsgEncoder.cs
+++ b/src/core/Akka.Remote.TestKit/MsgEncoder.cs
@@ -74,12 +74,15 @@ public void Encode(IConnection connection, object message, out List<IByteBuf> en
                                 .SetOp(BarrierOp.Fail)))
                     .With<ThrottleMsg>(
                         throttle =>
+                        {
                             w.SetFailure(
                                 InjectFailure.CreateBuilder()
                                     .SetFailure(TCP.FailType.Throttle)
                                     .SetAddress(Address2Proto(throttle.Target))
+                                    .SetFailure(TCP.FailType.Throttle)
                                     .SetDirection(Direction2Proto(throttle.Direction))
-                                    .SetRateMBit(throttle.RateMBit)))
+                                    .SetRateMBit(throttle.RateMBit));
+                        })
                     .With<DisconnectMsg>(
                         disconnect =>
                             w.SetFailure(
diff --git a/src/core/Akka.Remote.TestKit/MultiNodeSpec.cs b/src/core/Akka.Remote.TestKit/MultiNodeSpec.cs
index a703778c782..2c2d7ff5bcd 100644
--- a/src/core/Akka.Remote.TestKit/MultiNodeSpec.cs
+++ b/src/core/Akka.Remote.TestKit/MultiNodeSpec.cs
@@ -28,7 +28,11 @@ namespace Akka.Remote.TestKit
     /// </summary>
     public abstract class MultiNodeConfig
     {
-        Config _commonConf = null;
+        // allows us to avoid NullReferenceExceptions if we make this empty rather than null
+        // so that way if a MultiNodeConfig doesn't explicitly set CommonConfig to some value
+        // it will remain safe by defaut
+        Config _commonConf = Akka.Configuration.Config.Empty; 
+
         ImmutableDictionary<RoleName, Config> _nodeConf = ImmutableDictionary.Create<RoleName, Config>();
         ImmutableList<RoleName> _roles = ImmutableList.Create<RoleName>();
         ImmutableDictionary<RoleName, ImmutableList<string>> _deployments = ImmutableDictionary.Create<RoleName, ImmutableList<string>>();
diff --git a/src/core/Akka.Remote.TestKit/Player.cs b/src/core/Akka.Remote.TestKit/Player.cs
index c3aa6a1fa1f..739afd85a67 100644
--- a/src/core/Akka.Remote.TestKit/Player.cs
+++ b/src/core/Akka.Remote.TestKit/Player.cs
@@ -52,10 +52,9 @@ public IActorRef Client
         public Task<Done> StartClient(RoleName name, INode controllerAddr)
         {
             if(_client != null) throw new IllegalStateException("TestConductorClient already started");
-            _client =
-                _system.ActorOf(new Props(typeof (ClientFSM),
-                    new object[] {name, controllerAddr}), "TestConductorClient");
-            
+                _client =
+                _system.ActorOf(Props.Create(() => new ClientFSM(name, controllerAddr)), "TestConductorClient");
+
             //TODO: IRequiresMessageQueue
             var a = _system.ActorOf(Props.Create<WaitForClientFSMToConnect>());
 
@@ -132,22 +131,26 @@ public void Enter(TimeSpan timeout, ImmutableList<string> names)
                 try
                 {
                     var askTimeout = barrierTimeout + Settings.QueryTimeout;
-                    //TODO: Wait?
+                    // Need to force barrier to wait here, so we can pass along a "fail barrier" message in the event
+                    // of a failed operation
                     _client.Ask(new ToServer<EnterBarrier>(new EnterBarrier(name, barrierTimeout)), askTimeout).Wait();
                 }
-                catch (OperationCanceledException)
+                catch (AggregateException)
                 {
                     _client.Tell(new ToServer<FailBarrier>(new FailBarrier(name)));
                     throw new TimeoutException("Client timed out while waiting for barrier " + name);
                 }
+                catch (OperationCanceledException)
+                {
+                   _system.Log.Debug("OperationCanceledException was thrown instead of AggregateException");
+                }
                 _system.Log.Debug("passed barrier {0}", name);
             }
         }
 
         public Task<Address> GetAddressFor(RoleName name)
         {
-            //TODO: QueryTimeout implicit?
-            return _client.Ask<Address>(new ToServer<GetAddress>(new GetAddress(name)));
+            return _client.Ask<Address>(new ToServer<GetAddress>(new GetAddress(name)), Settings.QueryTimeout);
         }
     }
 
@@ -363,7 +366,7 @@ public void InitFSM()
                     _log.Info("disconnected from TestConductor");
                     throw new ConnectionFailure("disconnect");
                 }
-                if(@event.FsmEvent is ToServer<Done> && @event.StateData.Channel != null && @event.StateData.RunningOp == null)
+                if(@event.FsmEvent is ToServer<Done> && @event.StateData.Channel != null)
                 {
                     @event.StateData.Channel.Write(Done.Instance);
                     return Stay();
diff --git a/src/core/Akka.Remote.Tests.MultiNode/Akka.Remote.Tests.MultiNode.csproj b/src/core/Akka.Remote.Tests.MultiNode/Akka.Remote.Tests.MultiNode.csproj
index 03d002c8432..9e3ee4d5e87 100644
--- a/src/core/Akka.Remote.Tests.MultiNode/Akka.Remote.Tests.MultiNode.csproj
+++ b/src/core/Akka.Remote.Tests.MultiNode/Akka.Remote.Tests.MultiNode.csproj
@@ -72,6 +72,7 @@
     <Compile Include="RemoteNodeShutdownAndComesBackSpec.cs" />
     <Compile Include="RemoteRoundRobinSpec.cs" />
     <Compile Include="RemoteRandomSpec.cs" />
+    <Compile Include="TestConductor\TestConductorSpec.cs" />
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\..\contrib\testkits\Akka.TestKit.Xunit2\Akka.TestKit.Xunit2.csproj">
diff --git a/src/core/Akka.Remote.Tests.MultiNode/RemoteRandomSpec.cs b/src/core/Akka.Remote.Tests.MultiNode/RemoteRandomSpec.cs
index 29a8240f253..51afa3736ef 100644
--- a/src/core/Akka.Remote.Tests.MultiNode/RemoteRandomSpec.cs
+++ b/src/core/Akka.Remote.Tests.MultiNode/RemoteRandomSpec.cs
@@ -91,7 +91,7 @@ protected override void OnReceive(object message)
             }
         }
 
-        [MultiNodeFact]
+        [MultiNodeFact()]
         public void RemoteRandomSpecs()
         {
             A_remote_random_pool_must_be_locally_instantiated_on_a_remote_node_and_be_able_to_communicate_through_its_remote_actor_ref();
diff --git a/src/core/Akka.Remote.Tests.MultiNode/RemoteRoundRobinSpec.cs b/src/core/Akka.Remote.Tests.MultiNode/RemoteRoundRobinSpec.cs
index e2e73139795..35377ce2195 100644
--- a/src/core/Akka.Remote.Tests.MultiNode/RemoteRoundRobinSpec.cs
+++ b/src/core/Akka.Remote.Tests.MultiNode/RemoteRoundRobinSpec.cs
@@ -114,7 +114,7 @@ public override int Resize(IEnumerable<Routee> currentRoutees)
             }
         }
 
-        [MultiNodeFact]
+        [MultiNodeFact()]
         public void RemoteRoundRobinSpecs()
         {
             A_remote_round_robin_must_be_locally_instantiated_on_a_remote_node_and_be_able_to_communicate_through_its_remote_actor_ref();
diff --git a/src/core/Akka.Remote.Tests.MultiNode/TestConductor/TestConductorSpec.cs b/src/core/Akka.Remote.Tests.MultiNode/TestConductor/TestConductorSpec.cs
new file mode 100644
index 00000000000..1b34c5119a4
--- /dev/null
+++ b/src/core/Akka.Remote.Tests.MultiNode/TestConductor/TestConductorSpec.cs
@@ -0,0 +1,155 @@
+using System;
+using System.Collections.Generic;
+using System.Collections.Immutable;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Akka.Actor;
+using Akka.Actor.Dsl;
+using Akka.Configuration;
+using Akka.Remote.TestKit;
+using Akka.Remote.Transport;
+using Akka.TestKit;
+
+namespace Akka.Remote.Tests.MultiNode.TestConductor
+{
+    public class TestConductorSpecConfig : MultiNodeConfig
+    {
+        public RoleName Master { get; private set; }
+
+        public RoleName Slave { get; private set; }
+
+        public TestConductorSpecConfig()
+        {
+            Master = Role("master");
+            Slave = Role("slave");
+            CommonConfig = DebugConfig(true);
+            TestTransport = true;
+        }
+    }
+
+    public class TestConductorSpecNode1 : TestConductorSpec { }
+    public class TestConductorSpecNode2 : TestConductorSpec { }
+
+    public abstract class TestConductorSpec : MultiNodeSpec
+    {
+        private TestConductorSpecConfig _config;
+
+        public TestConductorSpec() : this(new TestConductorSpecConfig()) { }
+
+        protected TestConductorSpec(TestConductorSpecConfig config) : base(config)
+        {
+            _config = config;
+        }
+
+        protected override int InitialParticipantsValueFactory
+        {
+            get
+            {
+                return 2;
+            } 
+        }
+
+        private IActorRef _echo;
+
+        protected IActorRef GetEchoActorRef()
+        {
+            if (_echo == null)
+            {
+                Sys.ActorSelection(Node(_config.Master).Root / "user" / "echo").Tell(new Identify(null));
+                _echo = ExpectMsg<ActorIdentity>().Subject;
+            }
+            return _echo;
+        }
+
+        [MultiNodeFact]
+        public void ATestConductorMust()
+        {
+            Enter_a_Barrier();
+            Support_Throttling_of_Network_Connections();
+        }
+
+        public void Enter_a_Barrier()
+        {
+            RunOn(() =>
+            {
+                Sys.ActorOf(c => c.ReceiveAny((m, ctx) =>
+                {
+                    TestActor.Tell(m);
+                    ctx.Sender.Tell(m);
+                }), "echo");
+            }, _config.Master);
+
+            EnterBarrier("name");
+        }
+
+        public void Support_Throttling_of_Network_Connections()
+        {
+            RunOn(() =>
+            {
+                // start remote network connection so that it can be throttled
+                GetEchoActorRef().Tell("start");
+            }, _config.Slave);
+
+            ExpectMsg("start");
+
+            RunOn(() =>
+            {
+                TestConductor.Throttle(_config.Slave, _config.Master, ThrottleTransportAdapter.Direction.Send, 0.01f).Wait();
+            }, _config.Master);
+
+            EnterBarrier("throttled_send");
+
+            RunOn(() =>
+            {
+                foreach(var i in Enumerable.Range(0, 10))
+                {
+                    GetEchoActorRef().Tell(i);
+                }
+            }, _config.Slave);
+
+            // fudged the value to 0.5,since messages are a different size in Akka.NET
+            Within(TimeSpan.FromSeconds(0.5), TimeSpan.FromSeconds(2), () =>
+            {
+                ExpectMsg(0, TimeSpan.FromMilliseconds(500));
+                ReceiveN(9).ShouldOnlyContainInOrder(Enumerable.Range(1,9).Cast<object>().ToArray());
+            });
+
+            EnterBarrier("throttled_send2");
+            RunOn(() =>
+            {
+                TestConductor.Throttle(_config.Slave, _config.Master, ThrottleTransportAdapter.Direction.Send, -1).Wait();
+                TestConductor.Throttle(_config.Slave, _config.Master, ThrottleTransportAdapter.Direction.Receive, 0.01F).Wait();
+            }, _config.Master);
+
+            EnterBarrier("throttled_recv");
+
+            RunOn(() =>
+            {
+                foreach (var i in Enumerable.Range(10, 10))
+                {
+                    GetEchoActorRef().Tell(i);
+                }
+            }, _config.Slave);
+
+            var minMax = IsNode(_config.Master)
+                ? new Tuple<TimeSpan, TimeSpan>(TimeSpan.Zero, TimeSpan.FromMilliseconds(500))
+                : new Tuple<TimeSpan, TimeSpan>(TimeSpan.FromSeconds(0.3), TimeSpan.FromSeconds(3));
+
+            Within(minMax.Item1, minMax.Item2, () =>
+            {
+                ExpectMsg(10, TimeSpan.FromMilliseconds(500));
+                ReceiveN(9).ShouldOnlyContainInOrder(Enumerable.Range(11, 9).Cast<object>().ToArray());
+            });
+
+            EnterBarrier("throttled_recv2");
+
+            RunOn(() =>
+            {
+                TestConductor.Throttle(_config.Slave, _config.Master, ThrottleTransportAdapter.Direction.Receive, -1).Wait();
+            }, _config.Master);
+
+            EnterBarrier("after");
+        }
+    }
+}
diff --git a/src/core/Akka.Remote.Tests/Transport/AkkaProtocolStressTest.cs b/src/core/Akka.Remote.Tests/Transport/AkkaProtocolStressTest.cs
index 6a7fd6f3e4e..b5017f618cf 100644
--- a/src/core/Akka.Remote.Tests/Transport/AkkaProtocolStressTest.cs
+++ b/src/core/Akka.Remote.Tests/Transport/AkkaProtocolStressTest.cs
@@ -178,7 +178,7 @@ public AkkaProtocolStressTest() : base(AkkaProtocolStressTestConfig)
 
         #region Tests
 
-        [Fact(Skip = "fails due to out-of-order processing as a result of Helios eventing")]
+        [Fact(Skip = "Still have intermittent timeouts")]
         public void AkkaProtocolTransport_must_guarantee_at_most_once_delivery_and_message_ordering_despite_packet_loss()
         {
             //todo mute both systems for deadletters for any type of message
diff --git a/src/core/Akka.Remote/Endpoint.cs b/src/core/Akka.Remote/Endpoint.cs
index 3669b224cf1..ccfac508577 100644
--- a/src/core/Akka.Remote/Endpoint.cs
+++ b/src/core/Akka.Remote/Endpoint.cs
@@ -1100,7 +1100,7 @@ private void ScheduleBackoffTimer()
             else
             {
                 _smallBackoffCount += 1;
-                var backoffDeadlineNanoTime = TimeSpan.FromTicks((MonotonicClock.GetNanos() + _adaptiveBackoffNanos).ToTicks());
+                var backoffDeadlineNanoTime = TimeSpan.FromTicks(_adaptiveBackoffNanos.ToTicks());
 
                 Context.System.Scheduler.ScheduleTellOnce(backoffDeadlineNanoTime, Self, BackoffTimer.Instance, Self);
             }
diff --git a/src/core/Akka.Remote/RemoteWatcher.cs b/src/core/Akka.Remote/RemoteWatcher.cs
index ff3cdf9982f..08e487b04de 100644
--- a/src/core/Akka.Remote/RemoteWatcher.cs
+++ b/src/core/Akka.Remote/RemoteWatcher.cs
@@ -427,13 +427,13 @@ private void AddWatching(IActorRef watchee, IActorRef watcher)
 
         protected void ProcessUnwatchRemote(IActorRef watchee, IActorRef watcher)
         {
-            if (watcher != Self)
+            if (!Equals(watcher, Self))
             {
                 _log.Debug("Unwatching: [{0} -> {1}]", watcher.Path, watchee.Path);
                 _watching.Remove(Tuple.Create(watchee, watcher));
 
                 // clean up self watch when no more watchers of this watchee
-                if (_watching.All(t => t.Item1 != watchee || t.Item2 == Self))
+                if (_watching.All(t => !Equals(t.Item1, watchee) || Equals(t.Item2, Self)))
                 {
                     _log.Debug("Cleanup self watch of [{0}]", watchee.Path);
                     Context.Unwatch(watchee);
@@ -456,7 +456,7 @@ private void ProcessTerminated(IActorRef watchee, bool existenceConfirmed, bool
             var toProcess = _watching.Where(t => t.Item1.Equals(watchee)).ToList();
             foreach (var t in toProcess)
             {
-                if (!addressTerminated && t.Item2 != Self)
+                if (!addressTerminated && !Equals(t.Item2, Self))
                     t.Item2.Tell(new DeathWatchNotification(watchee, existenceConfirmed, false));
             }
 
diff --git a/src/core/Akka.Remote/Transport/ThrottleTransportAdapter.cs b/src/core/Akka.Remote/Transport/ThrottleTransportAdapter.cs
index 033855a3424..94eec9a2ebb 100644
--- a/src/core/Akka.Remote/Transport/ThrottleTransportAdapter.cs
+++ b/src/core/Akka.Remote/Transport/ThrottleTransportAdapter.cs
@@ -79,17 +79,13 @@ public override Task<bool> ManagementCommand(object message)
                 {
                     return r.Result is SetThrottleAck;
                 }, 
-                    TaskContinuationOptions.AttachedToParent | 
-                    TaskContinuationOptions.ExecuteSynchronously |
-                    TaskContinuationOptions.OnlyOnRanToCompletion);
+                    TaskContinuationOptions.ExecuteSynchronously);
             }
 
             if (message is ForceDisassociate || message is ForceDisassociateExplicitly)
             {
                 return manager.Ask(message, AskTimeout).ContinueWith(r => r.Result is ForceDisassociateAck,
-                    TaskContinuationOptions.AttachedToParent |
-                    TaskContinuationOptions.ExecuteSynchronously |
-                    TaskContinuationOptions.OnlyOnRanToCompletion);
+                    TaskContinuationOptions.ExecuteSynchronously);
             }
 
             return WrappedTransport.ManagementCommand(message);
diff --git a/src/core/Akka.Tests/Actor/FSMTransitionSpec.cs b/src/core/Akka.Tests/Actor/FSMTransitionSpec.cs
index afcfecb4382..870a18ed031 100644
--- a/src/core/Akka.Tests/Actor/FSMTransitionSpec.cs
+++ b/src/core/Akka.Tests/Actor/FSMTransitionSpec.cs
@@ -122,6 +122,8 @@ public MyFSM(IActorRef target)
                 Initialize();
             }
 
+            
+
             public IActorRef Target { get; private set; }
 
             protected override void PreRestart(Exception reason, object message)
diff --git a/src/core/Akka/Configuration/Config.cs b/src/core/Akka/Configuration/Config.cs
index 9a653825ef1..b04e1d5226f 100644
--- a/src/core/Akka/Configuration/Config.cs
+++ b/src/core/Akka/Configuration/Config.cs
@@ -483,6 +483,11 @@ public virtual IEnumerable<KeyValuePair<string, HoconValue>> AsEnumerable()
                 current = current.Fallback;
             }
         }
+
+        /// <summary>
+        /// A static "Empty" configuration we can use instead of <c>null</c> in some key areas.
+        /// </summary>
+        public static readonly Config Empty = ConfigurationFactory.Empty;
     }
 
     /// <summary>
diff --git a/src/examples/Cluster/Roles/Samples.Cluster.Transformation/App.config b/src/examples/Cluster/Roles/Samples.Cluster.Transformation/App.config
index ac5cd466feb..7260eddf0ec 100644
--- a/src/examples/Cluster/Roles/Samples.Cluster.Transformation/App.config
+++ b/src/examples/Cluster/Roles/Samples.Cluster.Transformation/App.config
@@ -27,7 +27,7 @@
                 "akka.tcp://ClusterSystem@127.0.0.1:2551",
                 "akka.tcp://ClusterSystem@127.0.0.1:2552"]
 
-              auto-down-unreachable-after = 10s
+              #auto-down-unreachable-after = 10s
             }
           }
       ]]>
diff --git a/src/examples/Cluster/Samples.Cluster.Simple/App.config b/src/examples/Cluster/Samples.Cluster.Simple/App.config
index 702c085578b..6785f64d844 100644
--- a/src/examples/Cluster/Samples.Cluster.Simple/App.config
+++ b/src/examples/Cluster/Samples.Cluster.Simple/App.config
@@ -27,7 +27,7 @@
                 "akka.tcp://ClusterSystem@127.0.0.1:2551",
                 "akka.tcp://ClusterSystem@127.0.0.1:2552"]
 
-              auto-down-unreachable-after = 30s
+              #auto-down-unreachable-after = 30s
             }
           }
       ]]>