C #: Asynchronous NamedPipeServerStream Trumpet closes exception

Question

My previous question on the same topic: C #: understanding asynchronous NamedPipeServerStream Now I have the following:

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 get

The pipe is being closed Exception every time I receive a message from a client

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:

 /// <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(); } 

Answer 1

It seems you need a separate NamedPipeServerStream for each client. (Note that I'm not the one who discovered this, see Other Answers.) I would suggest that the working server side would look something like this (draft code):

 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 noted in the commentary on your question, you lose memory when you initiate a new asynchronous call every 3 seconds, calling BeginWaitForConnection in a loop (the only case where it won't be “t when new connections are made at intervals of less than 3 seconds, but I doubt that you know that for sure.) You see, basically every 3 seconds you initiate a new asynchronous call, regardless of whether the last one is still unfinished or completed.In addition, it - again - does not take into account that each client requires a separate NamedPipeServerStream .

To fix this problem, you need to eliminate the BeginWaitForConnection loop and chain of calls using the callback method. This is a similar pattern that you often see in async-I / O when using .NET. Draft code:

 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 thread will be something like this:

• [main thread] StartListeningPipes (): created by NamedPipeServerStream initiated by BeginWaitForConnection ()
• [threadpool thread 1] client # 1 connection, BeginWaitForConnection () callback: EndWaitForConnection (), then StartListeningPipes ()
• [threadpool thread 1] StartListeningPipes (): created a new call to NamedPipeServerStream, BeginWaitForConnection ()
• [threadpool thread 1] return to the callback BeginWaitForConnection (): go to work with the connected client (# 1)
• [threadpool thread 2] client # 2 connection, BeginWaitForConnection () callback: ...
• ...

I think this is a lot more complicated than using blocking I / O - in fact, I'm not quite sure I understood correctly, please indicate if you see any errors - and this is also much more confusing.

To pause the server in both examples, you would obviously set the this.isServerRunning flag to false .

Answer 2

Ok I fell silent. For each client, there must be one NamedPipeServerStream. Therefore, if the Async operation was completed, then you need to recreate the NamedPipeServerStream. Thanks to this multithreaded NamePipeServer in C #

Must be:

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