Allow ThreadedSocketAcceptor to be restarted after calling Stop() - fixes issue #196PR 586 Continued (connamara#959)

History:
* Original issue connamara#196
* First PR attempt connamara#351
* that PR was updated to PR connamara#586
* and then *that* PR was updated to PR connamara#959

Description from all the squashed commits:

* 196 - modifications to allow the acceptor to be restarted after a call to stop has been made
* 196 - changing to using ManulResetEvent rather than thread.sleep and ensuring it's disposed of properly
* 196 - need to call session.logon when starting the acceptor as sessions are disabled during the call to logout during acceptor.stop()
* 196 - changing ThreadedSocketReactor to use BeginAcceptTcpClient/EndAcceptTcpClient
* 196 - fixing issues introduced by dispose changes for 340
* 196 - modifications to allow the acceptor to be restarted after a call to stop has been made
* 196 - need to call session.logon when starting the acceptor as sessions are disabled during the call to logout during acceptor.stop()
* 196 - changing ThreadedSocketReactor to use BeginAcceptTcpClient/EndAcceptTcpClient
* Fixing appveyor issue with default parameters in a test
* Issue 196 - fixing test
* specify debug file location - Attempt to eliminate flakiness (connamara#586)
* Fix rebase of RestartingTheThreadedSocketAcceptorTests
* Remove duplicate function that popped up after rebase completed
* Updating tests to match latest remote updates
* Continued fixing test with latest remote changes
* Fixing more tests
* fix warnings and other minor changes
* fix dupe/unused usings
* fix nullable warnings
* remove unused vars
* refactor out the silly LogEvent function
* delete #region around the sole ctor
* change some properties to be expression-bodied
* use ObjectDisposedException.ThrowIf
* remove unneeded SocketOption assignments
* hard-wrap some long lines
* make AreSocketsRunning internal; only UT uses it
* release notes for connamara#195 / PR connamara#596 / PR connamara#959
* delete non-error write to the NonSessionLog


---------

Co-authored-by: Huw Jenkins <[email protected]>
Co-authored-by: Vlad Yourtchenko <[email protected]>
Co-authored-by: Grant Birchmeier <[email protected]>

Author: 3 people

QuickFIXn/Exceptions.cs

@@ -120,6 +120,17 @@ public ConnectionResetByPeerException(System.Exception innerException)
         { }
     }
 
+    public class ConnectionShutdownRequestedException : QuickFIXException
+    {
+        public ConnectionShutdownRequestedException()
+            : base("Connection shutdown requested")
+        { }
+
+        public ConnectionShutdownRequestedException(System.Exception innerException)
+            : base("Connection shutdown requested", innerException)
+        { }
+    }
+
     public class MessageParseError : QuickFIXException
     {
         public MessageParseError()

QuickFIXn/ThreadedSocketAcceptor.cs

@@ -4,6 +4,7 @@
 using System;
 using QuickFix.Logger;
 using QuickFix.Store;
+using System.Threading;
 
 namespace QuickFix
 {
@@ -13,17 +14,16 @@ namespace QuickFix
     /// </summary>
     public class ThreadedSocketAcceptor : IAcceptor
     {
+        private const int TenSecondsInTicks = 10000;
+
         private readonly Dictionary<SessionID, Session> _sessions = new();
         private readonly SessionSettings _settings;
         private readonly Dictionary<IPEndPoint, AcceptorSocketDescriptor> _socketDescriptorForAddress = new();
         private readonly SessionFactory _sessionFactory;
-        private bool _isStarted = false;
         private bool _disposed = false;
         private readonly object _sync = new();
         private readonly NonSessionLog _nonSessionLog;
 
-        #region Constructors
-
         /// <summary>
         /// Create a ThreadedSocketAcceptor
         /// </summary>
@@ -59,13 +59,11 @@ public ThreadedSocketAcceptor(
             }
         }
 
-        #endregion
-
         #region Private Methods
 
         private AcceptorSocketDescriptor GetAcceptorSocketDescriptor(SettingsDictionary dict)
         {
-            int port = System.Convert.ToInt32(dict.GetLong(SessionSettings.SOCKET_ACCEPT_PORT));
+            int port = Convert.ToInt32(dict.GetLong(SessionSettings.SOCKET_ACCEPT_PORT));
             SocketSettings socketSettings = new SocketSettings();
 
             IPEndPoint socketEndPoint;
@@ -133,7 +131,7 @@ private bool CreateSession(SessionID sessionId, SettingsDictionary dict)
 
                     // start SocketReactor if it was created via AddSession call
                     // and if acceptor is already started
-                    if (_isStarted && !_disposed)
+                    if (IsStarted && !_disposed)
                     {
                         descriptor.SocketReactor.Start();
                     }
@@ -166,6 +164,21 @@ private void StopAcceptingConnections()
             }
         }
 
+        private void LogonAllSessions()
+        {
+            foreach (Session session in _sessions.Values)
+            {
+                try
+                {
+                    session.Logon();
+                }
+                catch (Exception e)
+                {
+                    session.Log.OnEvent($"Error during logon of Session {session.SessionID}: {e.Message}");
+                }
+            }
+        }
+
         private void LogoutAllSessions(bool force)
         {
             foreach (Session session in _sessions.Values)
@@ -182,55 +195,40 @@ private void LogoutAllSessions(bool force)
 
             if (force && IsLoggedOn)
             {
-                foreach (Session session in _sessions.Values)
-                {
-                    try
-                    {
-                        if (session.IsLoggedOn)
-                            session.Disconnect("Forcibly disconnecting session");
-                    }
-                    catch (Exception e)
-                    {
-                        session.Log.OnEvent($"Error during disconnect of Session {session.SessionID}: {e.Message}");
-                    }
-                }
+                DisconnectSessions("Forcibly disconnecting sessions");
             }
 
             if (!force)
                 WaitForLogout();
         }
 
-        /// <summary>
-        /// TODO implement WaitForLogout
-        /// </summary>
         private void WaitForLogout()
         {
-            /*
-            int start = System.Environment.TickCount;
-            HashSet<Session> sessions = new HashSet<Session>(sessions_.Values);
-            while(sessions.Count > 0)
+            int start = Environment.TickCount;
+            using( var resetEvent = new ManualResetEvent( false ) )
             {
-                Thread.Sleep(100);
-                
-                int elapsed = System.Environment.TickCount - start;
-                Iterator<Session> sessionItr = loggedOnSessions.iterator();
-                while (sessionItr.hasNext())
+                while (IsLoggedOn && (Environment.TickCount - start) < TenSecondsInTicks)
                 {
-                    Session session = sessionItr.next();
-                    if (elapsed >= session.getLogoutTimeout() * 1000L)
-                    {
-                        session.disconnect("Logout timeout, force disconnect", false);
-                        sessionItr.remove();
-                    }
+                    resetEvent.WaitOne( 100 );
+                }
+            }
+            DisconnectSessions("Logout timeout, force disconnect");
+        }
+
+        private void DisconnectSessions(string disconnectMessage)
+        {
+            foreach (Session session in _sessions.Values)
+            {
+                try
+                {
+                    if (session.IsLoggedOn)
+                        session.Disconnect(disconnectMessage);
                 }
-                // Be sure we don't look forever
-                if (elapsed > 60000)
+                catch (Exception e)
                 {
-                    log.warn("Stopping session logout wait after 1 minute");
-                    break;
+                    session.Log.OnEvent($"Error during disconnect of Session {session.SessionID}: {e.Message}");
                 }
             }
-            */
         }
 
         private void DisposeSessions()
@@ -252,10 +250,11 @@ public void Start()
 
             lock (_sync)
             {
-                if (!_isStarted)
+                if (!IsStarted)
                 {
+                    LogonAllSessions();
                     StartAcceptingConnections();
-                    _isStarted = true;
+                    IsStarted = true;
                 }
             }
         }
@@ -267,17 +266,17 @@ public void Stop()
 
         public void Stop(bool force)
         {
-            if (_disposed)
-                throw new ObjectDisposedException(GetType().Name);
-
-            StopAcceptingConnections();
-            LogoutAllSessions(force);
-            DisposeSessions();
-            _sessions.Clear();
-            _isStarted = false;
+            ObjectDisposedException.ThrowIf(_disposed, this);
 
-            // FIXME StopSessionTimer();
-            // FIXME Session.UnregisterSessions(GetSessions());
+            lock( _sync )
+            {
+                if( IsStarted )
+                {
+                    IsStarted = false;
+                    LogoutAllSessions(force);
+                    StopAcceptingConnections();
+                }
+            }
         }
 
         /// <summary>
@@ -292,6 +291,29 @@ public bool IsLoggedOn
             }
         }
 
+        /// <summary>
+        /// Gets a value indicating whether this instance is started.
+        /// </summary>
+        /// <value>
+        /// <c>true</c> if this instance is started; otherwise, <c>false</c>.
+        /// </value>
+        public bool IsStarted { get; private set; } = false;
+
+        /// <summary>
+        /// (For use by Unit Tests)
+        /// Gets a value indicating whether this instance is started.
+        /// </summary>
+        /// <value>
+        /// <c>true</c> if this instance is started; otherwise, <c>false</c>.
+        /// </value>
+        internal bool AreSocketsRunning
+        {
+            get
+            {
+                return _socketDescriptorForAddress.All( s => s.Value.SocketReactor.IsRunning );
+            }
+        }
+
         /// <summary>
         /// Get the SessionIDs for the sessions managed by this acceptor.
         /// </summary>
@@ -363,22 +385,21 @@ public bool RemoveSession(SessionID sessionId, bool terminateActiveSession)
         /// Any subclasses of ThreadedSocketAcceptor should override this if they have resources to dispose
         /// Any override should call base.Dispose(disposing).
         /// </summary>
-        /// <param name="disposing"></param>
+        /// <param name="disposing">true if called from Dispose()</param>
         protected void Dispose(bool disposing)
         {
-            if (_disposed) return;
-            if (disposing)
+            if(_disposed) { return; }
+            try
             {
-                try
-                {
-                    Stop();
-                }
-                catch (ObjectDisposedException)
-                {
-                    // ignore
-                }
+                Stop();
+                DisposeSessions();
+                _sessions.Clear();
+                _disposed = true;
+            }
+            catch (ObjectDisposedException)
+            {
+                // ignore
             }
-            _disposed = true;
         }
         /// <summary>
         /// Disposes created sessions
@@ -388,8 +409,8 @@ protected void Dispose(bool disposing)
         /// </remarks>
         public void Dispose()
         {
-            Dispose(true);
-            GC.SuppressFinalize(this);
+            Dispose( true );
+            GC.SuppressFinalize( this );
         }
 
         ~ThreadedSocketAcceptor() => Dispose(false);