Slightly reworked forced disconnection case

andrei.hilevich · andrei.hilevich · commit 2b498ea16996 · 2024-07-15T12:27:26.000+03:00
diff --git a/src/SharpRpc.Builder/SharpRpc.Builder.csproj b/src/SharpRpc.Builder/SharpRpc.Builder.csproj
@@ -4,7 +4,7 @@
    <TargetFramework>netstandard2.0</TargetFramework>
 	 <LangVersion>8.0</LangVersion>
 	 <Configurations>Debug;Release;DebugBuilder;ReleasePfCounters;ReleaseIde</Configurations>
-	 <Version>1.9.8</Version>
+	 <Version>1.9.9</Version>
 	 <Authors>Andrei Hilevich</Authors>
 	 <Company>Soft-Fx</Company>
 	 <Product>SharpRpc</Product>
@@ -16,7 +16,7 @@
 	 <PackageReleaseNotes></PackageReleaseNotes>
 	 <PackageTags>#RPC RPC</PackageTags>
    <IncludeBuildOutput>false</IncludeBuildOutput>
-   <AssemblyVersion>1.9.8.0</AssemblyVersion>
+   <AssemblyVersion>1.9.9.0</AssemblyVersion>
   </PropertyGroup>
 
   <PropertyGroup Condition="'$(Configuration)'=='DebugBuilder'">
diff --git a/src/SharpRpc/Channel.cs b/src/SharpRpc/Channel.cs
@@ -221,7 +221,7 @@ private void TriggerClose(RpcResult reason, bool isConnectionLost, out Task clos
             }
 
             if (forceDisconnect)
-                DropTransport();
+                AbortConnection();
 
             if (invokeDisconnect)
                 DoDisconnect();
@@ -349,7 +349,7 @@ private async void ConnectRoutine(CancellationToken cancelToken)
         private void OnLogoutTimeout()
         {
             Logger.Warn(Id, "Logout operation timed out!");
-            DropTransport();
+            AbortConnection();
         }
 
         //private Task TriggerCloseComponents(bool isAbortion)
@@ -406,7 +406,7 @@ private async Task CloseComponents()
         private void OnCloseComponentsTimeout()
         {
             Logger.Warn(Id, "Close operation timed out!");
-            DropTransport();
+            AbortConnection();
         }
 
         private void DisposeTransport()
@@ -445,18 +445,28 @@ private async void DoDisconnect()
             _disconnectEvent.SetResult(RpcResult.Ok);
         }
 
-        private void DropTransport()
+        // In case of connection loss, timeout, fatal error, or something ungraceful
+        private void AbortConnection()
         {
+            bool callDispose = false;
+
             lock (_stateSyncObj)
             {
                 _coordinator.AbortCoordination();
-                if (_isTransportDisposed)
-                    return;
-                _isTransportDisposed = true;
+                if (!_isTransportDisposed)
+                {
+                    _isTransportDisposed = true;
+                    callDispose = true;
+                }
             }
+            
+            Dispatcher.Abort(_channelFault);
 
-            Logger.Verbose(Id, "Disposing the transport...");
-            _transport?.Dispose();
+            if (callDispose)
+            {
+                Logger.Verbose(Id, "Disposing the transport...");
+                _transport?.Dispose();
+            }
         }
 
         private void SetClosedState()
diff --git a/src/SharpRpc/Disptaching/MessageDispatcher.NoThreading.cs b/src/SharpRpc/Disptaching/MessageDispatcher.NoThreading.cs
@@ -7,11 +7,7 @@
 
 using SharpRpc.Disptaching;
 using SharpRpc.Lib;
-using System;
 using System.Collections.Generic;
-using System.Diagnostics;
-using System.Linq;
-using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -24,7 +20,8 @@ private class NoThreading : MessageDispatcher
             private readonly object _lockObj = new object();
             private bool _isStarted;
             private bool _isProcessing;
-            private bool _closed;
+            private bool _isClosed;
+            private bool _isEnabled = true;
             private TaskCompletionSource<bool> _startedEvent = new TaskCompletionSource<bool>();
             private TaskCompletionSource<bool> _closeCompletion = new TaskCompletionSource<bool>();
 
@@ -50,30 +47,55 @@ public override RpcResult Start()
 
             public override Task Stop(RpcResult fault)
             {
+                List<IDispatcherOperation> operationsToCancel = null;
+
                 lock (_lockObj)
                 {
-                    if (_closed)
-                        return _closeCompletion.Task;
-
-                    _closed = true;
-
                     if (!_isStarted)
                     {
                         _isStarted = true;
                         _startedEvent.SetResult(true);
                     }
 
-                    Core.OnStop(fault);
+                    if (_isEnabled)
+                    {
+                        _isEnabled = false;
+                        operationsToCancel = Core.OnStop(fault);
+                    }
+
+                    if (_isClosed)
+                        return _closeCompletion.Task;
+
+                    _isClosed = true;
 
                     if (!_isProcessing)
                         CompleteClose();
                 }
 
-                Core.CompleteStop();
+                if (operationsToCancel != null)
+                    Core.CancelOperations(operationsToCancel);
 
                 return _closeCompletion.Task;
             }
 
+            public override void Abort(RpcResult fault)
+            {
+                List<IDispatcherOperation> operationsToCancel = null;
+
+                lock (_lockObj)
+                {
+                    if (_isEnabled)
+                    {
+                        _isEnabled = false;
+                        operationsToCancel = Core.OnStop(fault);
+                    }
+                }
+
+                if (operationsToCancel != null)
+                    Core.CancelOperations(operationsToCancel);
+            }
+
+
 #if NET5_0_OR_GREATER
             public override async ValueTask OnMessages()
 #else
@@ -84,7 +106,7 @@ public override async Task OnMessages()
 
                 lock (_lockObj)
                 {
-                    if (_closed)
+                    if (!_isEnabled)
                         return;
 
                     if (!_isStarted)
@@ -100,10 +122,11 @@ public override async Task OnMessages()
 
                     lock (_lockObj)
                     {
-                        if (_closed)
+                        if (!_isEnabled)
                         {
                             _isProcessing = false;
-                            CompleteClose();
+                            if (_isClosed)
+                                CompleteClose();
                             return;
                         }
                     }
@@ -116,7 +139,7 @@ public override async Task OnMessages()
                 {
                     _isProcessing = false;
 
-                    if (_closed)
+                    if (_isClosed)
                         CompleteClose();
                 }
             }
diff --git a/src/SharpRpc/Disptaching/MessageDispatcher.cs b/src/SharpRpc/Disptaching/MessageDispatcher.cs
@@ -70,9 +70,11 @@ protected void OnError(RpcRetCode code, string message)
 
         public void TriggerStop(RpcResult fault) => Stop(fault);
 
-        public abstract Task Stop(RpcResult fault);
+        public abstract Task Stop(RpcResult fault); // graceful close
         public abstract RpcResult Register(IDispatcherOperation operation);
         public abstract void Unregister(IDispatcherOperation operation);
+        public abstract void Abort(RpcResult fault); // In case of connection loss, timeout, or something ungraceful
+
         protected abstract void CancelOutgoingCall(IDispatcherOperation callObj);
         protected abstract void DoCall(IRequestMessage requestMsg, IDispatcherOperation callObj, CancellationToken cToken);
 
diff --git a/src/SharpRpc/Disptaching/MessageDispatcherCore.cs b/src/SharpRpc/Disptaching/MessageDispatcherCore.cs
@@ -20,7 +20,6 @@ internal class MessageDispatcherCore
         private readonly Dictionary<string, IDispatcherOperation> _operations = new Dictionary<string, IDispatcherOperation>();
         private bool _completed;
         private RpcResult _fault;
-        private List<IDispatcherOperation> _tasksToCancel;
 
         public MessageDispatcherCore(TxPipeline txNode, RpcCallHandler msgHandler, Action<RpcRetCode, string> onErrorAction)
         {
@@ -67,21 +66,22 @@ public void ProcessMessage(IMessage message)
                 ProcessOneWayMessage(message);
         }
 
-        public void OnStop(RpcResult fault)
+        public List<IDispatcherOperation> OnStop(RpcResult fault)
         {
             lock (_operations)
             {
                 _completed = true;
                 _fault = fault;
 
-                _tasksToCancel = _operations.Values.ToList();
+                var tasksToCancel = _operations.Values.ToList();
                 _operations.Clear();
+                return tasksToCancel;
             }
         }
 
-        public void CompleteStop()
+        public void CancelOperations(IEnumerable<IDispatcherOperation> operations)
         {
-            foreach (var task in _tasksToCancel)
+            foreach (var task in operations)
                 task.Terminate(_fault);
         }
 
diff --git a/src/SharpRpc/SharpRpc.csproj b/src/SharpRpc/SharpRpc.csproj
@@ -3,7 +3,7 @@
   <PropertyGroup>
     <TargetFrameworks>net6.0;net472;netstandard2.0</TargetFrameworks>
     <Configurations>Debug;Release;ReleasePfCounters;ReleaseIde</Configurations>
-    <Version>1.9.8</Version>
+    <Version>1.9.9</Version>
     <Company>Soft-Fx</Company>
     <Authors>Andrei Hilevich</Authors>
     <PackageId>SharpRpc.Core</PackageId>

Original file line number	Diff line number	Diff line change
`@@ -221,7 +221,7 @@ private void TriggerClose(RpcResult reason, bool isConnectionLost, out Task clos`
`221`	`221`	`}`
`222`	`222`
`223`	`223`	`if (forceDisconnect)`
`224`		`- DropTransport();`
	`224`	`+ AbortConnection();`
`225`	`225`
`226`	`226`	`if (invokeDisconnect)`
`227`	`227`	`DoDisconnect();`
`@@ -349,7 +349,7 @@ private async void ConnectRoutine(CancellationToken cancelToken)`
`349`	`349`	`private void OnLogoutTimeout()`
`350`	`350`	`{`
`351`	`351`	`Logger.Warn(Id, "Logout operation timed out!");`
`352`		`- DropTransport();`
	`352`	`+ AbortConnection();`
`353`	`353`	`}`
`354`	`354`
`355`	`355`	`//private Task TriggerCloseComponents(bool isAbortion)`
`@@ -406,7 +406,7 @@ private async Task CloseComponents()`
`406`	`406`	`private void OnCloseComponentsTimeout()`
`407`	`407`	`{`
`408`	`408`	`Logger.Warn(Id, "Close operation timed out!");`
`409`		`- DropTransport();`
	`409`	`+ AbortConnection();`
`410`	`410`	`}`
`411`	`411`
`412`	`412`	`private void DisposeTransport()`
`@@ -445,18 +445,28 @@ private async void DoDisconnect()`
`445`	`445`	`_disconnectEvent.SetResult(RpcResult.Ok);`
`446`	`446`	`}`
`447`	`447`
`448`		`- private void DropTransport()`
	`448`	`+ // In case of connection loss, timeout, fatal error, or something ungraceful`
	`449`	`+ private void AbortConnection()`
`449`	`450`	`{`
	`451`	`+ bool callDispose = false;`
	`452`	`+`
`450`	`453`	`lock (_stateSyncObj)`
`451`	`454`	`{`
`452`	`455`	`_coordinator.AbortCoordination();`
`453`		`- if (_isTransportDisposed)`
`454`		`- return;`
`455`		`- _isTransportDisposed = true;`
	`456`	`+ if (!_isTransportDisposed)`
	`457`	`+ {`
	`458`	`+ _isTransportDisposed = true;`
	`459`	`+ callDispose = true;`
	`460`	`+ }`
`456`	`461`	`}`
	`462`	`+`
	`463`	`+ Dispatcher.Abort(_channelFault);`
`457`	`464`
`458`		`- Logger.Verbose(Id, "Disposing the transport...");`
`459`		`- _transport?.Dispose();`
	`465`	`+ if (callDispose)`
	`466`	`+ {`
	`467`	`+ Logger.Verbose(Id, "Disposing the transport...");`
	`468`	`+ _transport?.Dispose();`
	`469`	`+ }`
`460`	`470`	`}`
`461`	`471`
`462`	`472`	`private void SetClosedState()`
Original file line number	Diff line number	Diff line change
`@@ -20,7 +20,6 @@ internal class MessageDispatcherCore`
`20`	`20`	`private readonly Dictionary<string, IDispatcherOperation> _operations = new Dictionary<string, IDispatcherOperation>();`
`21`	`21`	`private bool _completed;`
`22`	`22`	`private RpcResult _fault;`
`23`		`- private List<IDispatcherOperation> _tasksToCancel;`
`24`	`23`
`25`	`24`	`public MessageDispatcherCore(TxPipeline txNode, RpcCallHandler msgHandler, Action<RpcRetCode, string> onErrorAction)`
`26`	`25`	`{`
`@@ -67,21 +66,22 @@ public void ProcessMessage(IMessage message)`
`67`	`66`	`ProcessOneWayMessage(message);`
`68`	`67`	`}`
`69`	`68`
`70`		`- public void OnStop(RpcResult fault)`
	`69`	`+ public List<IDispatcherOperation> OnStop(RpcResult fault)`
`71`	`70`	`{`
`72`	`71`	`lock (_operations)`
`73`	`72`	`{`
`74`	`73`	`_completed = true;`
`75`	`74`	`_fault = fault;`
`76`	`75`
`77`		`- _tasksToCancel = _operations.Values.ToList();`
	`76`	`+ var tasksToCancel = _operations.Values.ToList();`
`78`	`77`	`_operations.Clear();`
	`78`	`+ return tasksToCancel;`
`79`	`79`	`}`
`80`	`80`	`}`
`81`	`81`
`82`		`- public void CompleteStop()`
	`82`	`+ public void CancelOperations(IEnumerable<IDispatcherOperation> operations)`
`83`	`83`	`{`
`84`		`- foreach (var task in _tasksToCancel)`
	`84`	`+ foreach (var task in operations)`
`85`	`85`	`task.Terminate(_fault);`
`86`	`86`	`}`
`87`	`87`