Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -484,6 +484,8 @@ namespace Akka.TestKit
public void SetAutoPilot(Akka.TestKit.AutoPilot pilot) { }
public virtual void Shutdown(System.TimeSpan? duration = default, bool verifySystemShutdown = false) { }
protected virtual void Shutdown(Akka.Actor.ActorSystem system, System.TimeSpan? duration = default, bool verifySystemShutdown = false) { }
public virtual System.Threading.Tasks.Task ShutdownAsync(System.TimeSpan? duration = default, bool verifySystemShutdown = false) { }
protected virtual System.Threading.Tasks.Task ShutdownAsync(Akka.Actor.ActorSystem system, System.TimeSpan? duration = default, bool verifySystemShutdown = false) { }
public bool TryPeekOne(out Akka.TestKit.MessageEnvelope envelope, System.TimeSpan? max, System.Threading.CancellationToken cancellationToken) { }
[return: System.Runtime.CompilerServices.TupleElementNames(new string[] {
"success",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -484,6 +484,8 @@ namespace Akka.TestKit
public void SetAutoPilot(Akka.TestKit.AutoPilot pilot) { }
public virtual void Shutdown(System.TimeSpan? duration = default, bool verifySystemShutdown = false) { }
protected virtual void Shutdown(Akka.Actor.ActorSystem system, System.TimeSpan? duration = default, bool verifySystemShutdown = false) { }
public virtual System.Threading.Tasks.Task ShutdownAsync(System.TimeSpan? duration = default, bool verifySystemShutdown = false) { }
protected virtual System.Threading.Tasks.Task ShutdownAsync(Akka.Actor.ActorSystem system, System.TimeSpan? duration = default, bool verifySystemShutdown = false) { }
public bool TryPeekOne(out Akka.TestKit.MessageEnvelope envelope, System.TimeSpan? max, System.Threading.CancellationToken cancellationToken) { }
[return: System.Runtime.CompilerServices.TupleElementNames(new string[] {
"success",
Expand Down
17 changes: 9 additions & 8 deletions src/core/Akka.Cluster.Tests.MultiNode/StressSpec.cs
Original file line number Diff line number Diff line change
Expand Up @@ -1236,10 +1236,10 @@ await WithinAsync(loopDuration, async () =>

if (activeRoles.Contains(Myself))
{
previousAs.OnSuccess(s =>
{
Shutdown(s);
});
// await the teardown instead of blocking on it: the sync Shutdown pins a thread
// pool thread for the whole wait, starving this node's own heartbeat sender.
if (previousAs.HasValue)
await ShutdownAsync(previousAs.Value);

var sys = ActorSystem.Create(Sys.Name, Sys.Settings.Config);
MuteLog(sys);
Expand Down Expand Up @@ -1279,10 +1279,11 @@ await AwaitMembersUpAsync(NbrUsedRoles + activeRoles.Length,
return await Loop(counter + 1, nextAs, nextAddresses);
}

(await Loop(1, Option<ActorSystem>.None, ImmutableHashSet<Address>.Empty)).OnSuccess(aSys =>
{
Shutdown(aSys);
});
// await the teardown instead of blocking on it: the sync Shutdown pins a thread pool thread for the
// whole wait, which on a busy agent starves this node's own heartbeat sender and gets it downed.
var lastAs = await Loop(1, Option<ActorSystem>.None, ImmutableHashSet<Address>.Empty);
if (lastAs.HasValue)
await ShutdownAsync(lastAs.Value);

await WithinAsync(loopDuration, async () =>
{
Expand Down
87 changes: 87 additions & 0 deletions src/core/Akka.TestKit.Tests/TestKitBaseTests/ShutdownAsyncTests.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,87 @@
//-----------------------------------------------------------------------
// <copyright file="ShutdownAsyncTests.cs" company="Akka.NET Project">
// Copyright (C) 2009-2022 Lightbend Inc. <http://www.lightbend.com>
// Copyright (C) 2013-2025 .NET Foundation <https://github.com/akkadotnet/akka.net>
// </copyright>
//-----------------------------------------------------------------------

using System;
using System.Threading.Tasks;
using Akka.Actor;
using Akka.Event;
using Akka.TestKit.Extensions;
using FluentAssertions;
using Xunit;

namespace Akka.TestKit.Tests.TestKitBaseTests
{
public class ShutdownAsyncTests : AkkaSpec
{
private static readonly TimeSpan ShortTimeout = TimeSpan.FromMilliseconds(200);
private static readonly TimeSpan CleanupTimeout = TimeSpan.FromSeconds(10);

[Fact(DisplayName = "ShutdownAsync should throw TimeoutException when verifySystemShutdown is true and the system will not stop")]
public async Task Should_Throw_When_SystemDoesNotStop_And_VerifyIsSet()
{
var release = NewRelease();
var sys = CreateStuckSystem("shutdown-async-throws", release);

try
{
var ex = await Assert.ThrowsAsync<TimeoutException>(
() => ShutdownAsync(sys, ShortTimeout, verifySystemShutdown: true));

ex.Message.Should().Contain("Failed to stop [shutdown-async-throws]");
}
finally
{
await ReleaseAndCleanUp(sys, release);
}
}

[Fact(DisplayName = "ShutdownAsync should log instead of throwing when verifySystemShutdown is not set")]
public async Task Should_Log_When_SystemDoesNotStop_And_VerifyIsNotSet()
{
var release = NewRelease();
var sys = CreateStuckSystem("shutdown-async-logs", release);
var probe = CreateTestProbe();
sys.EventStream.Subscribe(probe.Ref, typeof(Warning));

try
{
await ShutdownAsync(sys, ShortTimeout);

await probe.FishForMessageAsync<Warning>(
w => w.Message.ToString().Contains("Failed to stop [shutdown-async-logs]"),
TimeSpan.FromSeconds(3));
}
finally
{
await ReleaseAndCleanUp(sys, release);
}
}

private static TaskCompletionSource<Done> NewRelease()
=> new(TaskCreationOptions.RunContinuationsAsynchronously);

/// <summary>
/// Builds a system whose coordinated shutdown parks on a task we control, so
/// <see cref="ActorSystem.Terminate"/> cannot finish until the test releases it.
/// </summary>
private ActorSystem CreateStuckSystem(string name, TaskCompletionSource<Done> release)
{
var sys = ActorSystem.Create(name, Sys.Settings.Config);
CoordinatedShutdown.Get(sys).AddTask(
CoordinatedShutdown.PhaseBeforeActorSystemTerminate,
"block-until-released",
() => release.Task);
return sys;
}

private static async Task ReleaseAndCleanUp(ActorSystem sys, TaskCompletionSource<Done> release)
{
release.TrySetResult(Done.Instance);
await sys.Terminate().AwaitWithTimeout(CleanupTimeout);
}
}
}
85 changes: 74 additions & 11 deletions src/core/Akka.TestKit/TestKitBase.cs
Original file line number Diff line number Diff line change
Expand Up @@ -575,19 +575,82 @@ protected virtual void Shutdown(
{
system ??= _testState.System;

var durationValue = duration.GetValueOrDefault(Dilated(TimeSpan.FromSeconds(5)).Min(TimeSpan.FromSeconds(10)));
var durationValue = ShutdownDurationOrDefault(duration);

var wasShutdownDuringWait = system.Terminate().Wait(durationValue);
if(!wasShutdownDuringWait)
{
// Forcefully close the ActorSystem to make sure we exit the test cleanly
((ExtendedActorSystem) system).Guardian.Stop();

const string msg = "Failed to stop [{0}] within [{1}]. ActorSystem is being forcefully shut down.\n{2}";
if(verifySystemShutdown)
throw new TimeoutException(string.Format(msg, system.Name, durationValue, ((ExtendedActorSystem) system).PrintTree()));
system.Log.Warning(msg, system.Name, durationValue, ((ExtendedActorSystem) system).PrintTree());
}
ForceShutdownIfNeeded(system, wasShutdownDuringWait, durationValue, verifySystemShutdown);
}

/// <summary>
/// Shuts down this system without blocking the calling thread.
/// On failure debug output will be logged about the remaining actors in the system.
/// If verifySystemShutdown is true, then an exception will be thrown on failure.
/// </summary>
/// <remarks>
/// Prefer this over <see cref="Shutdown(TimeSpan?,bool)"/>. The synchronous form pins a thread pool
/// thread for up to the full duration. On a loaded machine that pinned thread can starve the very
/// system work the shutdown is waiting on - cluster heartbeats, coordinated shutdown phases - so the
/// wait itself makes the timeout more likely.
/// </remarks>
/// <param name="duration">Optional. The duration to wait for shutdown. Default is 5 seconds multiplied with the config value "akka.test.timefactor".</param>
/// <param name="verifySystemShutdown">if set to <c>true</c> an exception will be thrown on failure.</param>
/// <exception cref="TimeoutException">Thrown when the system did not stop in time and <paramref name="verifySystemShutdown"/> is <c>true</c>.</exception>
public virtual Task ShutdownAsync(
TimeSpan? duration = null,
bool verifySystemShutdown = false)
=> ShutdownAsync(_testState.System, duration, verifySystemShutdown);

/// <summary>
/// Shuts down the specified system without blocking the calling thread.
/// On failure debug output will be logged about the remaining actors in the system.
/// If verifySystemShutdown is true, then an exception will be thrown on failure.
/// </summary>
/// <remarks>
/// Prefer this over <see cref="Shutdown(ActorSystem,TimeSpan?,bool)"/>. The synchronous form pins a
/// thread pool thread for up to the full duration. On a loaded machine that pinned thread can starve
/// the very system work the shutdown is waiting on - cluster heartbeats, coordinated shutdown phases -
/// so the wait itself makes the timeout more likely.
/// </remarks>
/// <param name="system">The system to shutdown.</param>
/// <param name="duration">The duration to wait for shutdown. Default is 5 seconds multiplied with the config value "akka.test.timefactor"</param>
/// <param name="verifySystemShutdown">if set to <c>true</c> an exception will be thrown on failure.</param>
/// <exception cref="TimeoutException">Thrown when the system did not stop in time and <paramref name="verifySystemShutdown"/> is <c>true</c>.</exception>
protected virtual async Task ShutdownAsync(
ActorSystem system,
TimeSpan? duration = null,
bool verifySystemShutdown = false)
{
system ??= _testState.System;

var durationValue = ShutdownDurationOrDefault(duration);

// race Terminate() against a timer rather than blocking a thread on it
var wasShutdownDuringWait = await system.Terminate().AwaitWithTimeout(durationValue);
ForceShutdownIfNeeded(system, wasShutdownDuringWait, durationValue, verifySystemShutdown);
}

private TimeSpan ShutdownDurationOrDefault(TimeSpan? duration)
=> duration.GetValueOrDefault(Dilated(TimeSpan.FromSeconds(5)).Min(TimeSpan.FromSeconds(10)));

/// <summary>
/// Force-stops a system that failed to terminate on its own inside the timeout.
/// </summary>
private static void ForceShutdownIfNeeded(
ActorSystem system,
bool wasShutdownDuringWait,
TimeSpan durationValue,
bool verifySystemShutdown)
{
if(wasShutdownDuringWait)
return;

// Forcefully close the ActorSystem to make sure we exit the test cleanly
((ExtendedActorSystem) system).Guardian.Stop();

const string msg = "Failed to stop [{0}] within [{1}]. ActorSystem is being forcefully shut down.\n{2}";
if(verifySystemShutdown)
throw new TimeoutException(string.Format(msg, system.Name, durationValue, ((ExtendedActorSystem) system).PrintTree()));
system.Log.Warning(msg, system.Name, durationValue, ((ExtendedActorSystem) system).PrintTree());
}

/// <summary>
Expand Down
Loading