Merge pull request #164 from dotnet-campus/t/lindexi/Ipc

修复线程安全

Author: walterlv

.github/workflows/dotnet-core.yml

@@ -23,7 +23,7 @@ jobs:
       run: dotnet build --configuration Release
 
     - name: Test
-      run: dotnet test --configuration Release -v:n --tl:off --no-build
+      run: dotnet test --configuration Release -v:n --tl:off --no-build --logger:"console;verbosity=detailed"
 
     - name: Pack
       run: dotnet pack --configuration Release --no-build

…s.Ipc/Context/ConnectExistsPeerResult.cs …c/Context/ConnectToExistingPeerResult.cssrc/dotnetCampus.Ipc/Context/ConnectExistsPeerResult.cs renamed to src/dotnetCampus.Ipc/Context/ConnectToExistingPeerResult.cs src/dotnetCampus.Ipc/Context/ConnectExistsPeerResult.cs renamed to src/dotnetCampus.Ipc/Context/ConnectToExistingPeerResult.cs

@@ -13,9 +13,9 @@ namespace dotnetCampus.Ipc.Context;
 /// <summary>
 /// 连接已经存在的 Peer 的结果
 /// </summary>
-public readonly struct ConnectExistsPeerResult
+public readonly struct ConnectToExistingPeerResult
 {
-    internal ConnectExistsPeerResult(PeerProxy? peerProxy, Task peerConnectFinishedTask)
+    internal ConnectToExistingPeerResult(PeerProxy? peerProxy, Task peerConnectFinishedTask)
     {
         PeerProxy = peerProxy;
         PeerConnectFinishedTask = peerConnectFinishedTask;
@@ -37,5 +37,5 @@ internal ConnectExistsPeerResult(PeerProxy? peerProxy, Task peerConnectFinishedT
     [MemberNotNullWhen(true, nameof(PeerProxy))]
     public bool IsSuccess => PeerProxy != null;
 
-    internal static ConnectExistsPeerResult Fail() => new ConnectExistsPeerResult(null, TaskUtils.CompletedTask);
+    internal static ConnectToExistingPeerResult Fail() => new ConnectToExistingPeerResult(null, TaskUtils.CompletedTask);
 }

src/dotnetCampus.Ipc/Context/LoggingContext/LoggerEventIds.cs

@@ -4,6 +4,8 @@ namespace dotnetCampus.Ipc.Context.LoggingContext;
 
 internal static class LoggerEventIds
 {
+    public static EventId CommonDebugEventId => new EventId(0, "Debug");
+
     /// <summary>
     /// 发送消息
     /// </summary>

src/dotnetCampus.Ipc/Internals/PeerReConnector.cs

@@ -63,7 +63,7 @@ private async Task<bool> TryReconnectAsync(IpcClientService ipcClientService)
             {
                 try
                 {
-                    return await ipcClientService.StartInternalAsync(isReConnect: true, shouldRegisterToPeer: true, onlyConnectExistsPeer: false);
+                    return await ipcClientService.StartInternalAsync(isReConnect: true, shouldRegisterToPeer: true, onlyConnectToExistingPeer: false);
                 }
                 // ## 此异常有两种
                 catch (FileNotFoundException)

src/dotnetCampus.Ipc/Pipes/IpcClientService.cs

@@ -107,7 +107,7 @@ public NamedPipeClientStreamResult(Exception exception)
         /// <returns></returns>
         public async Task Start(bool shouldRegisterToPeer = true)
         {
-            var result = await StartInternalAsync(isReConnect: false, shouldRegisterToPeer, onlyConnectExistsPeer: false/*不是只连接存在的对方，如果对方还不存在，则进行等待*/);
+            var result = await StartInternalAsync(isReConnect: false, shouldRegisterToPeer, onlyConnectToExistingPeer: false/*不是只连接存在的对方，如果对方还不存在，则进行等待*/);
 
             if (!result)
             {
@@ -121,22 +121,22 @@ public async Task Start(bool shouldRegisterToPeer = true)
         /// <returns></returns>
         internal Task<bool> TryConnectToExistingPeerAsync()
         {
-            return StartInternalAsync(isReConnect: false, shouldRegisterToPeer: true, onlyConnectExistsPeer: true);
+            return StartInternalAsync(isReConnect: false, shouldRegisterToPeer: false, onlyConnectToExistingPeer: true);
         }
 
         /// <inheritdoc cref="Start"/>
         /// <param name="isReConnect">是否属于重新连接</param>
         /// <param name="shouldRegisterToPeer">是否需要向对方注册</param>
-        /// <param name="onlyConnectExistsPeer">只连接存在的对方</param>
+        /// <param name="onlyConnectToExistingPeer">只连接存在的对方</param>
         /// <returns>True:启动成功</returns>
-        internal async Task<bool> StartInternalAsync(bool isReConnect, bool shouldRegisterToPeer, bool onlyConnectExistsPeer)
+        internal async Task<bool> StartInternalAsync(bool isReConnect, bool shouldRegisterToPeer, bool onlyConnectToExistingPeer)
         {
             var localClient = IpcContext.PipeName;
             var remoteServer = PeerName;
 
             Logger.Trace($"StartInternalAsync Connecting NamedPipe. LocalClient:'{localClient}';RemoteServer:'{remoteServer}'");
 
-            if (onlyConnectExistsPeer)
+            if (onlyConnectToExistingPeer)
             {
                 if (!PipeHelper.IsPipeExists(remoteServer))
                 {
@@ -151,7 +151,7 @@ internal async Task<bool> StartInternalAsync(bool isReConnect, bool shouldRegist
 
             try
             {
-                var result = await ConnectNamedPipeAsync(isReConnect, namedPipeClientStream, onlyConnectExistsPeer);
+                var result = await ConnectNamedPipeAsync(isReConnect, namedPipeClientStream, onlyConnectToExistingPeer);
                 if (!result)
                 {
                     _namedPipeClientStreamTaskCompletionSource.TrySetResult(new NamedPipeClientStreamResult(namedPipeClientStream: null));
@@ -175,7 +175,7 @@ internal async Task<bool> StartInternalAsync(bool isReConnect, bool shouldRegist
             if (shouldRegisterToPeer)
             {
                 // 启动之后，向对方注册，此时对方是服务器
-                await RegisterToPeer();
+                await RegisterToPeerAsync();
             }
 
             return true;
@@ -186,17 +186,17 @@ internal async Task<bool> StartInternalAsync(bool isReConnect, bool shouldRegist
         /// </summary>
         /// <param name="isReConnect">是否属于重新连接</param>
         /// <param name="namedPipeClientStream"></param>
-        /// <param name="onlyConnectExistsPeer">只连接存在的对方</param>
+        /// <param name="onlyConnectToExistingPeer">只连接存在的对方</param>
         /// <returns>True 连接成功</returns>
         /// 独立方法，方便 dnspy 调试
-        private async Task<bool> ConnectNamedPipeAsync(bool isReConnect, NamedPipeClientStream namedPipeClientStream, bool onlyConnectExistsPeer)
+        private async Task<bool> ConnectNamedPipeAsync(bool isReConnect, NamedPipeClientStream namedPipeClientStream, bool onlyConnectToExistingPeer)
         {
             var connector = IpcContext.IpcClientPipeConnector;
 
             if (connector == null)
             {
                 var timeout = Timeout.Infinite;
-                if (onlyConnectExistsPeer)
+                if (onlyConnectToExistingPeer)
                 {
                     // 如果是只连接存在的对方的情况，即使对方存在，也需要设置一个短暂的超时时间
                     // 为什么这里还需要设置超时时间，这是因为可能上一步判断 IsPipeExists 时，对方还是存在的，然而当前准备连接的时候，对方已经挂了，因此不能无限等待，需要设置一个短暂的时间
@@ -209,10 +209,11 @@ private async Task<bool> ConnectNamedPipeAsync(bool isReConnect, NamedPipeClient
                 }
                 catch (Exception e)
                 {
-                    if (onlyConnectExistsPeer && e is IpcPipeConnectionException ipcPipeConnectionException)
+                    if (onlyConnectToExistingPeer && e is IpcPipeConnectionException ipcPipeConnectionException)
                     {
                         if (ipcPipeConnectionException.InnerException is TimeoutException)
                         {
+                            // 如果是只连接存在的对方，且连接超时了，则直接返回 false 证明对方现在无法被连接上，正常就是对方不存在
                             return false;
                         }
                     }
@@ -223,7 +224,7 @@ private async Task<bool> ConnectNamedPipeAsync(bool isReConnect, NamedPipeClient
             }
             else
             {
-                return await CustomConnectNamedPipeAsync(connector, isReConnect, namedPipeClientStream, onlyConnectExistsPeer);
+                return await CustomConnectNamedPipeAsync(connector, isReConnect, namedPipeClientStream, onlyConnectToExistingPeer);
             }
         }
 
@@ -233,16 +234,16 @@ private async Task<bool> ConnectNamedPipeAsync(bool isReConnect, NamedPipeClient
         /// <param name="ipcClientPipeConnector"></param>
         /// <param name="isReConnect">是否属于重新连接</param>
         /// <param name="namedPipeClientStream"></param>
-        /// <param name="onlyConnectExistsPeer"></param>
+        /// <param name="onlyConnectToExistingPeer"></param>
         /// <returns></returns>
         private async Task<bool> CustomConnectNamedPipeAsync(IIpcClientPipeConnector ipcClientPipeConnector,
             bool isReConnect,
-            NamedPipeClientStream namedPipeClientStream, bool onlyConnectExistsPeer)
+            NamedPipeClientStream namedPipeClientStream, bool onlyConnectToExistingPeer)
         {
             Logger.Trace($"Connecting NamedPipe by {nameof(CustomConnectNamedPipeAsync)}. LocalClient:'{IpcContext.PipeName}';RemoteServer:'{PeerName}'");
             var cancellationToken = CancellationToken.None;
 
-            if (onlyConnectExistsPeer)
+            if (onlyConnectToExistingPeer)
             {
                 cancellationToken = new CancellationTokenSource(TimeSpan.FromMilliseconds(10)).Token;
             }
@@ -292,7 +293,7 @@ void ConnectNamedPipe()
             }
         }
 
-        private async Task RegisterToPeer()
+        internal async Task RegisterToPeerAsync()
         {
             Logger.Trace($"[{nameof(IpcClientService)}] StartRegisterToPeer PipeName={IpcContext.PipeName}");
 

src/dotnetCampus.Ipc/Pipes/IpcProvider.cs

@@ -5,10 +5,12 @@
 
 using dotnetCampus.Ipc.CompilerServices.GeneratedProxies;
 using dotnetCampus.Ipc.Context;
+using dotnetCampus.Ipc.Context.LoggingContext;
 using dotnetCampus.Ipc.Exceptions;
 using dotnetCampus.Ipc.Internals;
 using dotnetCampus.Ipc.Utils;
 using dotnetCampus.Ipc.Utils.Extensions;
+using dotnetCampus.Ipc.Utils.Logging;
 
 namespace dotnetCampus.Ipc.Pipes
 {
@@ -80,7 +82,7 @@ public async void StartServer()
                 var ipcServerService = new IpcServerService(IpcContext);
                 _ipcServerService = ipcServerService;
 
-                ipcServerService.PeerConnected += NamedPipeServerStreamPoolPeerConnected;
+                ipcServerService.PeerConnected += IpcServerService_OnPeerConnected;
 
                 // 以下的 Start 是一个循环，不会返回的
                 await ipcServerService.Start().ConfigureAwait(false);
@@ -100,13 +102,17 @@ public async void StartServer()
         /// </summary>
         /// <param name="sender"></param>
         /// <param name="e"></param>
-        private async void NamedPipeServerStreamPoolPeerConnected(object? sender, IpcInternalPeerConnectedArgs e)
+        private async void IpcServerService_OnPeerConnected(object? sender, IpcInternalPeerConnectedArgs e)
         {
             try
             {
+                IpcContext.Logger.Debug($"[OnPeerConnected]IpcProvider.OnPeerConnected PeerName={e.PeerName};CurrentName={IpcContext.PipeName}");
+
                 // 也许是对方反过来连接
                 if (PeerManager.TryGetValue(e.PeerName, out var peerProxy))
                 {
+                    IpcContext.Logger.Debug($"[OnPeerConnected]PeerManager.TryGetValue Success. PeerName={e.PeerName};CurrentName={IpcContext.PipeName}");
+
                     // 如果当前的 Peer 已断开且不需要重新连接，那么重新创建 Peer 反过来连接对方的服务器端
                     if (peerProxy.IsBroken && !IpcContext.IpcConfiguration.AutoReconnectPeers)
                     {
@@ -121,6 +127,8 @@ private async void NamedPipeServerStreamPoolPeerConnected(object? sender, IpcInt
                 }
                 else
                 {
+                    IpcContext.Logger.Debug($"[OnPeerConnected]PeerManager.TryGetValue Fail. ConnectBackToPeer. PeerName={e.PeerName};CurrentName={IpcContext.PipeName}");
+
                     // 其他客户端连接，需要反过来连接对方的服务器端
                     await ConnectBackToPeer(e);
                 }
@@ -136,6 +144,8 @@ private async Task ConnectBackToPeer(IpcInternalPeerConnectedArgs e)
         {
             try
             {
+                IpcContext.Logger.Debug($"[OnPeerConnected] ConnectBackToPeer. PeerName={e.PeerName};CurrentName={IpcContext.PipeName}");
+
                 await ConnectBackToPeerCore(e);
             }
             catch (ObjectDisposedException)
@@ -318,10 +328,10 @@ public async Task<PeerProxy> GetAndConnectToPeerAsync(string peerName)
         /// 尝试获取或连接到已经存在的 Peer 上。如果当前的 Peer 还没起来，则不等待连接，直接返回失败
         /// </summary>
         /// <param name="peerName">对方</param>
-        /// <param name="shouldWaitPeerConnectFinished">是否应该等待对方连接回来完成，完全完成双向连接。如设置为 false 则需要自己通过 <see cref="ConnectExistsPeerResult.PeerConnectFinishedTask"/> 进行等待。默认为 true 表示等待所有准备完成再返回</param>
+        /// <param name="shouldWaitPeerConnectFinished">是否应该等待对方连接回来完成，完全完成双向连接。如设置为 false 则需要自己通过 <see cref="ConnectToExistingPeerResult.PeerConnectFinishedTask"/> 进行等待。默认为 true 表示等待所有准备完成再返回</param>
         /// <returns></returns>
         /// 为什么会存在 <paramref name="shouldWaitPeerConnectFinished"/> 参数，这是为了解决极端情况下，刚好本进程能连接到对方，连接完成瞬间，对方挂了，无法反向连接回来的情况。正常不需要设置此参数
-        public async Task<ConnectExistsPeerResult> TryConnectToExistingPeerAsync(string peerName, bool shouldWaitPeerConnectFinished = true)
+        public async Task<ConnectToExistingPeerResult> TryConnectToExistingPeerAsync(string peerName, bool shouldWaitPeerConnectFinished = true)
         {
             if (PeerManager.TryGetValue(peerName, out var peerProxy))
             {
@@ -336,16 +346,22 @@ public async Task<ConnectExistsPeerResult> TryConnectToExistingPeerAsync(string
 
                 var ipcClientService = CreateIpcClientService(peerName);
 
+                // 尝试连接对方
+                // 连接的时候不会立刻向对方注册自己，只是建立连接关系而已
+                // 这样是因为一旦向对方注册自己，那对方将会反过来向自己注册。然而此时存在多线程安全问题，此时的 PeerManager 还没加入对方。导致以下代码里面的 PeerManager.WaitForPeerConnectFinishedAsync 无法完成等待，导致单元测试失败
                 var result = await ipcClientService.TryConnectToExistingPeerAsync().ConfigureAwait(false);
                 if (!result)
                 {
                     // 对方不存在
-                    return ConnectExistsPeerResult.Fail();
+                    return ConnectToExistingPeerResult.Fail();
                 }
 
                 // 需要确定能连接上对方了，才能加入到 PeerManager 里面。确保不会在下次进来的时候，拿到了一个无法建立连接的 Peer 对象。这里的添加顺序是先确保连接再添加，这就意味着在并行的时候，可能会多次尝试连接。这是符合预期的，本身连接也没有多少损耗，最多只会多创建一个管道而已
                 peerProxy = new PeerProxy(peerName, ipcClientService, IpcContext);
                 PeerManager.TryAdd(peerProxy);
+
+                // 在 PeerProxy 加入到管理之后，才能向对方注册自己，确保对方收到注册之后，反过来向自己注册时，可以从管理里面拿到注册的对方信息，从而让 PeerManager.WaitForPeerConnectFinishedAsync 能够完成
+                await ipcClientService.RegisterToPeerAsync();
             }
 
             // 等待对方回连，建立双向连接
@@ -364,11 +380,11 @@ public async Task<ConnectExistsPeerResult> TryConnectToExistingPeerAsync(string
                 catch (IpcPeerConnectionBrokenException e)
                 {
                     // 对方连接断开了
-                    return ConnectExistsPeerResult.Fail();
+                    return ConnectToExistingPeerResult.Fail();
                 }
             }
 
-            return new ConnectExistsPeerResult(peerProxy, peerConnectFinishedTask);
+            return new ConnectToExistingPeerResult(peerProxy, peerConnectFinishedTask);
         }
 
         /// <summary>

src/dotnetCampus.Ipc/Utils/Extensions/LoggerExtensions.cs

@@ -1,5 +1,6 @@
 using System;
 
+using dotnetCampus.Ipc.Context.LoggingContext;
 using dotnetCampus.Ipc.Utils.Logging;
 
 namespace dotnetCampus.Ipc.Utils.Extensions
@@ -13,7 +14,7 @@ public static void Trace(this ILogger? logger, string message)
 
         public static void Debug(this ILogger? logger, string message)
         {
-            logger?.Log(LogLevel.Debug, default, message, null, FormatOnlyMessage);
+            logger?.Log(LogLevel.Debug, LoggerEventIds.CommonDebugEventId, message, null, FormatOnlyMessage);
         }
 
         public static void Information(this ILogger? logger, string message)