From 6d41624f47e905817ba7dabefbd86fb9aa7eb169 Mon Sep 17 00:00:00 2001 From: Nadia Aina <113453463+Nadia-Adaptive@users.noreply.github.com> Date: Fri, 21 Aug 2026 18:14:32 +0100 Subject: [PATCH 1/4] Port AgentRunner close rework from Agrona 2.5.0 --- .../Concurrent/AgentRunnerTest.cs | 169 ++++++++++++++++++ src/Adaptive.Agrona/Concurrent/AgentRunner.cs | 46 +++-- 2 files changed, 202 insertions(+), 13 deletions(-) create mode 100644 src/Adaptive.Agrona.Tests/Concurrent/AgentRunnerTest.cs 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..ddf78d6d 100644 --- a/src/Adaptive.Agrona/Concurrent/AgentRunner.cs +++ b/src/Adaptive.Agrona/Concurrent/AgentRunner.cs @@ -192,6 +192,11 @@ public void Run() /// public void Dispose() { + if (IsClosed) + { + return; + } + _isRunning = false; var thread = _thread.GetAndSet(Tombstone); @@ -210,27 +215,42 @@ public void Dispose() } else if (Tombstone != thread) { - while (true) + var wasInterrupted = false; + try { - try + while (thread.IsAlive) { - thread.Join(RETRY_CLOSE_TIMEOUT_MS); - - if (!thread.IsAlive || IsClosed) + try { - return; - } + if (wasInterrupted) + { + Console.Error.WriteLine( + $"Agent '{_agent.RoleName()}' failed to close due to close interrupted, retrying..." + ); + thread.Interrupt(); + } - Console.Error.WriteLine( - $"Timeout waiting for agent '{_agent.RoleName()}' to close, Retrying..." - ); + thread.Join(RETRY_CLOSE_TIMEOUT_MS); - thread.Interrupt(); + if (thread.IsAlive) + { + Console.Error.WriteLine( + $"Agent '{_agent.RoleName()}' failed to close due to timeout, retrying..." + ); + thread.Interrupt(); + } + } + catch (ThreadInterruptedException) + { + wasInterrupted = true; + } } - catch (ThreadInterruptedException) + } + finally + { + if (wasInterrupted) { System.Threading.Thread.CurrentThread.Interrupt(); - return; } } } From 938a74871afa019277930b92d84bee6466c7e55b Mon Sep 17 00:00:00 2001 From: Nadia Aina <113453463+Nadia-Adaptive@users.noreply.github.com> Date: Mon, 24 Aug 2026 13:46:31 +0100 Subject: [PATCH 2/4] Fix per loop interrupt logging in AgentRunner --- src/Adaptive.Agrona/Concurrent/AgentRunner.cs | 21 +++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/src/Adaptive.Agrona/Concurrent/AgentRunner.cs b/src/Adaptive.Agrona/Concurrent/AgentRunner.cs index ddf78d6d..599f51bd 100644 --- a/src/Adaptive.Agrona/Concurrent/AgentRunner.cs +++ b/src/Adaptive.Agrona/Concurrent/AgentRunner.cs @@ -216,6 +216,7 @@ public void Dispose() else if (Tombstone != thread) { var wasInterrupted = false; + var hasLoggedInterrupt = false; try { while (thread.IsAlive) @@ -224,9 +225,12 @@ public void Dispose() { if (wasInterrupted) { - Console.Error.WriteLine( - $"Agent '{_agent.RoleName()}' failed to close due to close interrupted, retrying..." - ); + if (!hasLoggedInterrupt) + { + LogError("close interrupted"); + hasLoggedInterrupt = true; + } + thread.Interrupt(); } @@ -234,9 +238,7 @@ public void Dispose() if (thread.IsAlive) { - Console.Error.WriteLine( - $"Agent '{_agent.RoleName()}' failed to close due to timeout, retrying..." - ); + LogError("timeout"); thread.Interrupt(); } } @@ -256,6 +258,13 @@ public void Dispose() } } + 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 From 39967f3624344f6832baf33c9d5a4ee6aa224290 Mon Sep 17 00:00:00 2001 From: Nadia Aina <113453463+Nadia-Adaptive@users.noreply.github.com> Date: Tue, 15 Sep 2026 14:53:48 +0100 Subject: [PATCH 3/4] Note interrupt behaviour in Close doc comment --- src/Adaptive.Agrona/Concurrent/AgentRunner.cs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/Adaptive.Agrona/Concurrent/AgentRunner.cs b/src/Adaptive.Agrona/Concurrent/AgentRunner.cs index 599f51bd..49c7dcef 100644 --- a/src/Adaptive.Agrona/Concurrent/AgentRunner.cs +++ b/src/Adaptive.Agrona/Concurrent/AgentRunner.cs @@ -187,6 +187,10 @@ 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. /// /// From 730ee3c9dfa991ea9c6d25809905a6e2d01fa87e Mon Sep 17 00:00:00 2001 From: Nadia Aina <113453463+Nadia-Adaptive@users.noreply.github.com> Date: Wed, 16 Sep 2026 16:55:24 +0100 Subject: [PATCH 4/4] Make AgentRunner.IsClosed volatile --- src/Adaptive.Agrona/Concurrent/AgentRunner.cs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/Adaptive.Agrona/Concurrent/AgentRunner.cs b/src/Adaptive.Agrona/Concurrent/AgentRunner.cs index 49c7dcef..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; } } @@ -209,7 +211,7 @@ public void Dispose() { try { - IsClosed = true; + _isClosed = true; _agent.OnClose(); } catch (Exception ex)