C#：异步 NamedPipeServerStream 管道正在关闭异常

C#: Asynchronous NamedPipeServerStream The pipe is being closed exception

提问by Ksice

My previous question on the same theme: C#: Asynchronous NamedPipeServerStream understandingNow I have next:

我之前关于同一主题的问题：C#：Asynchronous NamedPipeServerStream 理解现在我有下一个：

private void StartListeningPipes()
{
    try
    {
        isPipeWorking = true;
                namedPipeServerStream = new NamedPipeServerStream(PIPENAME, PipeDirection.InOut, 1, PipeTransmissionMode.Byte, PipeOptions.Asynchronous, BUFFERSIZE, BUFFERSIZE);
                Console.Write("Waiting for client connection...");
                while(isPipeWorking)
                {
            IAsyncResult asyncResult = namedPipeServerStream.BeginWaitForConnection(this.WaitForConnectionAsyncCallback, null);
                        Thread.Sleep(3*1000);
                }
        }
        //// Catch the IOException that is raised if the pipe is broken or disconnected.
        catch (IOException e)
        {
        Console.WriteLine("IOException: {0}. Restart pipe server...", e.Message);
                StopListeningPipes();
                StartListeningPipes();
        }
        //// Catch ObjectDisposedException if server was stopped. Then do nothing.
        catch (ObjectDisposedException)
        {
        }
}

private void WaitForConnectionAsyncCallback(IAsyncResult result)
{
    try
    {
        namedPipeServerStream.EndWaitForConnection(result);
        Console.WriteLine("Client connected.");
        namedPipeServerStream.WaitForPipeDrain();
                byte[] buff = new byte[BUFFERSIZE];
                namedPipeServerStream.Read(buff, 0, BUFFERSIZE);
                string recStr = TrimNulls(buff);
                Array.Clear(buff, 0, buff.Length);
                Console.WriteLine();
                Console.WriteLine("'"+recStr+"'");
    }
    catch (Exception e)
    {
        Console.WriteLine("Error: " + e.Message);            
        }
}

But I'm getting

但我得到

The pipe is being closed Exceptioneverytime I receive a message from client

The pipe is being closed Exception每次我收到客户的消息

Why?

为什么？

My client:

我的客户：

 using (NamedPipeClientStream pipeStream = new NamedPipeClientStream(General.PIPENAME))
{
    try
        {
        byte[] bytes = General.Iso88591Encoding.GetBytes(sendingMessage);
                pipeStream.Write(bytes, 0, bytes.Length);
                pipeStream.Flush();
                pipeStream.WaitForPipeDrain();
        }
        catch (TimeoutException)
        {
        Console.WriteLine("Timeout error!");
        }
    catch (Exception e)
        {
        Console.WriteLine(string.Format("Error! ", e.Message));
        }
}

---

---

Final code at the moment is:

目前的最终代码是：

/// <summary>
        /// Create new NamedPipeServerStream for listening to pipe client connection
        /// </summary>
        private void ListenForPipeClients()
        {
            if (!this.isListeningToClients)
                return;

            try
            {
                PipeSecurity ps = new PipeSecurity();
                PipeAccessRule par = new PipeAccessRule("Everyone", PipeAccessRights.ReadWrite, System.Security.AccessControl.AccessControlType.Allow);
                ps.AddAccessRule(par);
                pipeClientConnection = new NamedPipeServerStream(General.PIPENAME, PipeDirection.InOut, 1, PipeTransmissionMode.Byte, PipeOptions.Asynchronous, General.BUFFERSIZE, General.BUFFERSIZE, ps);
                Console.Write("Waiting for client connection...");
                /*namedPipeServerStream.WaitForConnection();
                OnPipeConnected(namedPipeServerStream);*/
                IAsyncResult result = pipeClientConnection.BeginWaitForConnection(OnPipeConnected, pipeClientConnection);
            }
            catch (ObjectDisposedException)
            {
                //// Catch ObjectDisposedException if server was stopped. Then do nothing.
            }
            catch (Exception e)
            {
                Console.WriteLine("Error occures: {0}. Restart pipe server...", e.Message);
                this.logger.Add(LogLevel.Warning, string.Format("Error occures: {0}. Restart pipe server...", e.Message));
                ListenForPipeClients();
            }
        }

        /// <summary>
        /// Async callback on client connected action
        /// </summary>
        /// <param name="asyncResult">Async result</param>
        private void OnPipeConnected(IAsyncResult asyncResult)
        {
            using (var conn = (NamedPipeServerStream)asyncResult.AsyncState)
            {
                try
                {
                    conn.EndWaitForConnection(asyncResult);
                    Console.WriteLine("Client connected.");
                    PipeClientConnection clientConnection = new PipeClientConnection(conn, notifierSenderCache, defaultStorageTime);
                }
                catch (Exception e)
                {
                    Console.WriteLine(e.Message);
                    this.logger.Add(LogLevel.Warning, e.Message);
                }
            }

            ListenForPipeClients();
        }

采纳答案by ShdNx

It appears that you need a separate NamedPipeServerStreamfor each client. (Note that I was not the one to discover this, see the other answers.) I'd imagine the working server-side would look something like this (draft code):

看来您需要NamedPipeServerStream为每个客户端单独创建一个。（请注意，我不是发现这一点的人，请参阅其他答案。）我想工作的服务器端看起来像这样（草稿代码）：

while(this.isServerRunning)
{
     var pipeClientConnection = new NamedPipeServerStream(...);

     try
     {
         pipeClientConnection.WaitForConnection();
     }
     catch(...)
     {
         ...
         continue;
     }

     ThreadPool.QueueUserWorkItem(state =>
          {
               // we need a separate variable here, so as not to make the lambda capture the pipeClientConnection variable, which is not recommended in multi-threaded scenarios
               using(var pipeClientConn = (NamedPipeServerStream)state)
               {
                    // do stuff
                    ...
               }
          }, pipeClientConnection);
}

As a side note, as it was pointed out in a comment to your question, you're wasting memory with initiating a new async call every 3 seconds by calling BeginWaitForConnectionin a loop (the only case where this wouldn't waste memory is when new connections are made in intervals smaller than 3 seconds, but I doubt that you can know this for sure). You see, basically every 3 seconds you're initiating a new async call, regardless of whether the last one is still pending or has completed. Furthermore, it - once again - does not take into account that you need a separate NamedPipeServerStreamfor each client.

作为旁注，正如在对您的问题的评论中指出的那样，您通过BeginWaitForConnection循环调用每 3 秒启动一次新的异步调用是在浪费内存（这不会浪费内存的唯一情况是 new连接的时间间隔小于 3 秒，但我怀疑您是否可以肯定地知道这一点）。你看，基本上每 3 秒你就会发起一个新的异步调用，不管最后一个调用是未决还是已经完成。此外，它 - 再一次 - 没有考虑到您需要NamedPipeServerStream为每个客户端单独设置。

To fix this issue, you need to eliminate the loop, and "chain" the BeginWaitForConnection calls using the callback method. This is a similar pattern you'll see quite often in async I/O when using .NET. Draft code:

要解决此问题，您需要消除循环，并使用回调方法“链接”BeginWaitForConnection 调用。这是使用 .NET 时在异步 I/O 中经常会看到的类似模式。草稿代码：

private void StartListeningPipes()
{
    if(!this.isServerRunning)
    {
        return;
    }

    var pipeClientConnection = new NamedPipeServerStream(...);

    try
    {
        pipeClientConnection.BeginWaitForConnection(asyncResult =>
            {
                // note that the body of the lambda is not part of the outer try... catch block!
                using(var conn = (NamedPipeServerStream)asyncResult.AsyncState)
                {
                    try
                    {
                        conn.EndWaitForConnection(asyncResult);
                    }
                    catch(...)
                    {
                        ...
                    }

                    // we have a connection established, time to wait for new ones while this thread does its business with the client
                    // this may look like a recursive call, but it is not: remember, we're in a lambda expression
                    // if this bothers you, just export the lambda into a named private method, like you did in your question
                    StartListeningPipes();

                    // do business with the client
                    conn.WaitForPipeDrain();
                    ...
                }
            }, pipeClientConnection);
    }
    catch(...)
    {
        ...
    }
}

The control flow will be something like this:

控制流将是这样的：

• [main thread] StartListeningPipes(): created NamedPipeServerStream, initiated BeginWaitForConnection()
• [threadpool thread 1] client #1 connecting, BeginWaitForConnection() callback: EndWaitForConnection() then StartListeningPipes()
• [threadpool thread 1] StartListeningPipes(): created new NamedPipeServerStream, BeginWaitForConnection() call
• [threadpool thread 1] back to the BeginWaitForConnection() callback: getting down to business with the connected client (#1)
• [threadpool thread 2] client #2 connecting, BeginWaitForConnection() callback: ...
• ...

• 【主线程】StartListeningPipes()：创建NamedPipeServerStream，发起BeginWaitForConnection()
• [线程池线程 1] 客户端 #1 连接，BeginWaitForConnection() 回调：EndWaitForConnection() 然后 StartListeningPipes()
• [线程池线程 1] StartListeningPipes()：创建新的 NamedPipeServerStream，BeginWaitForConnection() 调用
• [线程池线程 1] 回到 BeginWaitForConnection() 回调：开始处理连接的客户端（#1）
• [线程池线程 2] 客户端 #2 连接，BeginWaitForConnection() 回调：...
• ...

I think that this is a lot more difficult than using blocking I/O - in fact, I'm not quite certain I got it right, please point it out if you see any mistakes - and it's also a lot more confusing.

我认为这比使用阻塞 I/O 困难得多——事实上，我不太确定我做对了，如果你看到任何错误，请指出——而且它也更令人困惑。

To pause the server in either examples, you obviously would set the this.isServerRunningflag to false.

要在任一示例中暂停服务器，您显然会将this.isServerRunning标志设置为false。

回答by Ksice

Ok. Stupied me. There should be one NamedPipeServerStream for each client. So if Async operation was completed, then have to recreate NamedPipeServerStream. Thanks this Multithreaded NamePipeServer in C#

好的。把我搞糊涂了。每个客户端应该有一个 NamedPipeServerStream。所以如果 Async 操作完成，那么必须重新创建 NamedPipeServerStream。感谢这个C# 中的多线程 NamePipeServer

Should be:

应该：

while(isPipeWorking)
            {
                IAsyncResult asyncResult = namedPipeServerStream.BeginWaitForConnection(this.WaitForConnectionAsyncCallback, null);
                Thread.Sleep(3*1000);
                if (asyncResult.IsCompleted)
                {
                    RestartPipeServer();
                    break;
                }
            }