diff --git a/src/Adaptive.Agrona.Tests/Concurrent/AgentRunnerTest.cs b/src/Adaptive.Agrona.Tests/Concurrent/AgentRunnerTest.cs new file mode 100644 index 00000000..a6789eca --- /dev/null +++ b/src/Adaptive.Agrona.Tests/Concurrent/AgentRunnerTest.cs @@ -0,0 +1,169 @@ +/* + * Copyright 2014 - 2026 Adaptive Financial Consulting Ltd + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +using System; +using System.Threading; +using Adaptive.Agrona.Concurrent; +using FakeItEasy; +using NUnit.Framework; + +namespace Adaptive.Agrona.Tests.Concurrent +{ + public class AgentRunnerTest + { + private IAgent _agent; + private IErrorHandler _errorHandler; + private AgentRunner _runner; + + [SetUp] + public void Setup() + { + _agent = A.Fake(); + _errorHandler = A.Fake(); + A.CallTo(() => _agent.RoleName()).Returns("test-agent"); + + _runner = new AgentRunner(new NoOpIdleStrategy(), _errorHandler, null, _agent); + } + + [Test] + public void ShouldNotInterruptRunnerThreadIfCloseCompletesOnTime() + { + var started = SetUpDoWork(() => 0); + + var runnerThread = AgentRunner.StartOnThread(_runner); + started.Wait(); + + _runner.Dispose(); + + Assert.That(runnerThread.IsAlive, Is.False); + Assert.That(_runner.IsClosed, Is.True); + A.CallTo(() => _agent.OnStart()).MustHaveHappenedOnceExactly(); + A.CallTo(() => _agent.OnClose()).MustHaveHappenedOnceExactly(); + A.CallTo(() => _errorHandler.OnError(A._)).MustNotHaveHappened(); + } + + [Test] + public void ShouldInterruptRunnerThreadIfCloseCallIsItselfInterrupted() + { + var started = SetUpDoWork(() => BlockUntilInterrupted()); + + var runnerThread = AgentRunner.StartOnThread(_runner); + started.Wait(); + + var callerReinterrupted = false; + var enteringClose = new ManualResetEventSlim(false); + var closerThread = new Thread(() => + { + enteringClose.Set(); + _runner.Dispose(); + + try + { + Thread.Sleep(100); + } + catch (ThreadInterruptedException) + { + callerReinterrupted = true; + } + }); + closerThread.Start(); + + enteringClose.Wait(); + closerThread.Interrupt(); + closerThread.Join(); + + Assert.That(closerThread.IsAlive, Is.False); + Assert.That(runnerThread.IsAlive, Is.False); + Assert.That(callerReinterrupted, Is.True); + Assert.That(_runner.IsClosed, Is.True); + } + + [Test] + public void ShouldInterruptRunnerThreadIfCloseDoesNotCompleteWithinCloseTimeout() + { + var started = SetUpDoWork(() => BlockUntilInterrupted()); + + var runnerThread = AgentRunner.StartOnThread(_runner); + started.Wait(); + + _runner.Dispose(); + + Assert.That(runnerThread.IsAlive, Is.False); + Assert.That(_runner.IsClosed, Is.True); + A.CallTo(() => _agent.OnClose()).MustHaveHappenedOnceExactly(); + } + + [Test] + public void ShouldInterruptRunnerThreadIfCloseCompletesOnTimeWhenMainThreadIsInterruptedBeforeTheCloseCall() + { + var agentInterrupted = false; + var started = SetUpDoWork(() => BlockUntilInterrupted(() => agentInterrupted = true)); + + var runnerThread = AgentRunner.StartOnThread(_runner); + started.Wait(); + + var callerReinterrupted = false; + var closerThread = new Thread(() => + { + Thread.CurrentThread.Interrupt(); + + _runner.Dispose(); + + try + { + Thread.Sleep(100); + } + catch (ThreadInterruptedException) + { + callerReinterrupted = true; + } + }); + closerThread.Start(); + closerThread.Join(); + + Assert.That(closerThread.IsAlive, Is.False); + Assert.That(runnerThread.IsAlive, Is.False); + Assert.That(agentInterrupted, Is.True); + Assert.That(callerReinterrupted, Is.True); + Assert.That(_runner.IsClosed, Is.True); + } + + private ManualResetEventSlim SetUpDoWork(Func onDoWork) + { + var started = new ManualResetEventSlim(false); + A.CallTo(() => _agent.DoWork()).ReturnsLazily(() => + { + started.Set(); + return onDoWork(); + }); + return started; + } + + private static int BlockUntilInterrupted(Action onInterrupted = null) + { + try + { + Thread.Sleep(Timeout.Infinite); + } + catch (ThreadInterruptedException) + { + onInterrupted?.Invoke(); + } + + return 0; + } + } +} diff --git a/src/Adaptive.Agrona/Concurrent/AgentRunner.cs b/src/Adaptive.Agrona/Concurrent/AgentRunner.cs index 4780d766..40670c8b 100644 --- a/src/Adaptive.Agrona/Concurrent/AgentRunner.cs +++ b/src/Adaptive.Agrona/Concurrent/AgentRunner.cs @@ -37,10 +37,12 @@ public class AgentRunner : IDisposable private volatile bool _isRunning = true; + private volatile bool _isClosed; + /// /// Has the been closed? /// - public bool IsClosed { get; private set; } + public bool IsClosed => _isClosed; private readonly AtomicCounter _errorCounter; private readonly IErrorHandler _errorHandler; @@ -178,7 +180,7 @@ public void Run() } finally { - IsClosed = true; + _isClosed = true; } } @@ -187,11 +189,20 @@ public void Run() /// performing /// it logic. /// + /// Note: if the caller thread is interrupted while invoking this method then the agent thread will be + /// interrupted as well, but the loop will not exit until the agent thread fully terminates. + /// + /// /// The clean up logic will only be performed once even if close is called from multiple concurrent threads. /// /// public void Dispose() { + if (IsClosed) + { + return; + } + _isRunning = false; var thread = _thread.GetAndSet(Tombstone); @@ -200,7 +211,7 @@ public void Dispose() { try { - IsClosed = true; + _isClosed = true; _agent.OnClose(); } catch (Exception ex) @@ -210,32 +221,56 @@ public void Dispose() } else if (Tombstone != thread) { - while (true) + var wasInterrupted = false; + var hasLoggedInterrupt = false; + try { - try + while (thread.IsAlive) { - thread.Join(RETRY_CLOSE_TIMEOUT_MS); - - if (!thread.IsAlive || IsClosed) + try { - return; - } + if (wasInterrupted) + { + if (!hasLoggedInterrupt) + { + LogError("close interrupted"); + hasLoggedInterrupt = true; + } - Console.Error.WriteLine( - $"Timeout waiting for agent '{_agent.RoleName()}' to close, Retrying..." - ); + thread.Interrupt(); + } - thread.Interrupt(); + thread.Join(RETRY_CLOSE_TIMEOUT_MS); + + if (thread.IsAlive) + { + LogError("timeout"); + thread.Interrupt(); + } + } + catch (ThreadInterruptedException) + { + wasInterrupted = true; + } } - catch (ThreadInterruptedException) + } + finally + { + if (wasInterrupted) { System.Threading.Thread.CurrentThread.Interrupt(); - return; } } } } + private void LogError(string reason) + { + Console.Error.WriteLine( + $"Agent '{_agent.RoleName()}' failed to close due to {reason}, retrying..." + ); + } + private bool DoDutyCycle(IIdleStrategy idleStrategy, IAgent agent) { try