Using NamedPipeServerStream and NamedPipeClientStream asynchronously

4
votes

I have the following requirements for a server/client architecture:

  1. Write a server/client that works asynchronously.

  2. The communication needs to be a duplex, i.e., reads and writes on both ends.

  3. Multiple clients can connect to the server at any given time.

  4. Server/client should wait until they become available and finally make a connection.

  5. Once a client connects it should write to the stream.

  6. Then the server should read from the stream and write response back to the client.

  7. Finally, the client should read the response and the communication should end.

So with the following requirements in mind I've written the following code but I'm not too sure about it because the docs for pipes are somewhat lacking, unfortunately and the code doesn't seems to work correctly, it hangs at a certain point.

namespace PipesAsyncAwait471
{
    using System;
    using System.Collections.Generic;
    using System.IO.Pipes;
    using System.Linq;
    using System.Threading.Tasks;

    internal class Program
    {
        private static async Task Main()
        {
            List<Task> tasks = new List<Task> {
                HandleRequestAsync(),
            };

            tasks.AddRange(Enumerable.Range(0, 10).Select(i => SendRequestAsync(i, 0, 5)));

            await Task.WhenAll(tasks);
        }

        private static async Task HandleRequestAsync()
        {
            using (NamedPipeServerStream server = new NamedPipeServerStream("MyPipe",
                                                                            PipeDirection.InOut,
                                                                            NamedPipeServerStream.MaxAllowedServerInstances,
                                                                            PipeTransmissionMode.Message,
                                                                            PipeOptions.Asynchronous))
            {
                Console.WriteLine("Waiting...");

                await server.WaitForConnectionAsync().ConfigureAwait(false);

                if (server.IsConnected)
                {
                    Console.WriteLine("Connected");

                    if (server.CanRead) {
                        // Read something...
                    }

                    if (server.CanWrite) {
                        // Write something... 

                        await server.FlushAsync().ConfigureAwait(false);

                        server.WaitForPipeDrain();
                    }

                    server.Disconnect();

                    await HandleRequestAsync().ConfigureAwait(false);
                }
            }
        }

        private static async Task SendRequestAsync(int index, int counter, int max)
        {
            using (NamedPipeClientStream client = new NamedPipeClientStream(".", "MyPipe", PipeDirection.InOut, PipeOptions.Asynchronous))
            {
                await client.ConnectAsync().ConfigureAwait(false);

                if (client.IsConnected)
                {
                    Console.WriteLine($"Index: {index} Counter: {counter}");

                    if (client.CanWrite) {
                        // Write something...

                        await client.FlushAsync().ConfigureAwait(false);

                        client.WaitForPipeDrain();
                    }

                    if (client.CanRead) {
                        // Read something...
                    }
                }

                if (counter <= max) {
                    await SendRequestAsync(index, ++counter, max).ConfigureAwait(false);
                }
                else {
                    Console.WriteLine($"{index} Done!");
                }
            }
        }
    }
}

Assumptions:

The way I expect it to work is for all the requests I make when I call SendRequestAsync to execute concurrently where each request then makes additional requests until it reaches 6 and finally, it should print "Done!".

Remarks:

  1. I've tested it on .NET Framework 4.7.1 and .NET Core 2.0 and I get the same results.

  2. The communication between clients and the server is always local to the machine where clients are web applications that can queue some jobs like launching 3rd-party processes and the server is going to be deployed as a Windows service on the same machine as the web server that these clients are deployed on.

2
c#.netnamed-pipes
Instead of using pipes you may want to use TCP. See msdn examples : docs.microsoft.com/en-us/dotnet/framework/network-programming/… - jdweng
@jdweng The client and server are processes found on the same machine so TCP would be an overkill for this. - Eyal Alon
Absolutely wrong. Millions of applications use TCP on a local PC. Pipes are used for standard input and standard output but are rarely used to tunnel into an application. Using TCP you can use a sniffer like wireshark or fiddler to debug application. - jdweng
@jdweng TCP is used on local PCs that generally needs to make a remote connection, pipes are used heavily in IPC and I'm using it exactly for that, besides the client and the server there's also 3rd party processes that the server launches where these process have their stdin/out redirected but this isn't related the issue at hand so I didn't bother to go into details about it. - Eyal Alon
The recursive call to SendRequestAsync() is very ugly. That constantly creates new client pipes, none get closed. The show is probably over when it has created too many connections. Throw this away and use the MSDN sample code as a guide to get this right. - Hans Passant

2 Answers

3
votes

When disconnecting, WaitForPipeDrain() can throw an IOException due to a broken pipe.

If this happens in your server Task, then it will never listen for the next connection, and all of the remaining client connections hang on ConnectAsync().

If this happens in one of the client Tasks, then it will not continue to recurse and increment the counter for that index.

If you wrap the call to WaitForPipeDrain() in a try/catch, the program will continue running forever, because your function HandleRequestAsync() is infinitely recursive.

In short, to get this to work:

  1. Handle IOException from WaitForPipeDrain()
  2. HandleRequestAsync() has to finish at some point.
3
votes

Here is the complete code after some iterations:

namespace PipesAsyncAwait471
{
    using System;
    using System.Collections.Generic;
    using System.IO;
    using System.IO.Pipes;
    using System.Linq;
    using System.Threading.Tasks;

    internal class Program
    {
        private const int MAX_REQUESTS = 1000;

        private static void Main()
        {
            var tasks = new List<Task> {
                //Task.Run(() => HandleRequest(0))
                HandleRequestAsync(0)
            };

            tasks.AddRange(Enumerable.Range(0, MAX_REQUESTS).Select(i => Task.Factory.StartNew(() => SendRequest(i), TaskCreationOptions.LongRunning)));

            Task.WhenAll(tasks);

            Console.ReadKey();
        }

        private static void HandleRequest(int counter)
        {
            try {
                var server = new NamedPipeServerStream("MyPipe",
                                                    PipeDirection.InOut,
                                                    NamedPipeServerStream.MaxAllowedServerInstances,
                                                    PipeTransmissionMode.Message,
                                                    PipeOptions.Asynchronous);

                Console.WriteLine($"Waiting a client... {counter}");

                server.BeginWaitForConnection(WaitForConnectionCallback, server);
            }
            catch (Exception ex) {
                Console.WriteLine(ex);
            }

            void WaitForConnectionCallback(IAsyncResult result)
            {
                var server = (NamedPipeServerStream)result.AsyncState;

                int index = -1;

                try {
                    server.EndWaitForConnection(result);

                    HandleRequest(++counter);

                    if (server.IsConnected) {
                        var request = new byte[4];
                        server.BeginRead(request, 0, request.Length, ReadCallback, server);
                        index = BitConverter.ToInt32(request, 0);
                        Console.WriteLine($"{index} Request.");

                        var response = BitConverter.GetBytes(index);
                        server.BeginWrite(response, 0, response.Length, WriteCallback, server);
                        server.Flush();
                        server.WaitForPipeDrain();
                        Console.WriteLine($"{index} Pong.");

                        server.Disconnect();
                        Console.WriteLine($"{index} Disconnected.");
                    }
                }
                catch (IOException ex) {
                    Console.WriteLine($"{index}\n\t{ex}");
                }
                finally {
                    server.Dispose();
                }
            }

            void ReadCallback(IAsyncResult result) 
            {
                var server = (NamedPipeServerStream)result.AsyncState;

                try {
                    server.EndRead(result);
                }
                catch (IOException ex) {
                    Console.WriteLine(ex);
                }
            }

            void WriteCallback(IAsyncResult result) 
            {
                var server = (NamedPipeServerStream)result.AsyncState;

                try {
                    server.EndWrite(result);
                }
                catch (IOException ex) {
                    Console.WriteLine(ex);
                }
            }
        }

        private static async Task HandleRequestAsync(int counter)
        {
            NamedPipeServerStream server = null;

            int index = -1;

            try {
                server = new NamedPipeServerStream("MyPipe",
                                                PipeDirection.InOut,
                                                NamedPipeServerStream.MaxAllowedServerInstances,
                                                PipeTransmissionMode.Message,
                                                PipeOptions.Asynchronous);

                Console.WriteLine($"Waiting a client... {counter}");

                await server.WaitForConnectionAsync()
                            .ContinueWith(async t => await HandleRequestAsync(++counter).ConfigureAwait(false))
                            .ConfigureAwait(false);

                if (server.IsConnected) {
                    var request = new byte[4];
                    await server.ReadAsync(request, 0, request.Length).ConfigureAwait(false);
                    index = BitConverter.ToInt32(request, 0);
                    Console.WriteLine($"{index} Request.");

                    var response = BitConverter.GetBytes(index);
                    await server.WriteAsync(response, 0, response.Length).ConfigureAwait(false);
                    await server.FlushAsync().ConfigureAwait(false);
                    server.WaitForPipeDrain();
                    Console.WriteLine($"{index} Pong.");

                    server.Disconnect();
                    Console.WriteLine($"{index} Disconnected.");
                }
            }
            catch (IOException ex) {
                Console.WriteLine($"{index}\n\t{ex}");
            }
            finally {
                server?.Dispose();
            }
        }

        private static void SendRequest(int index)
        {
            NamedPipeClientStream client = null;

            try {
                client = new NamedPipeClientStream(".", "MyPipe", PipeDirection.InOut, PipeOptions.None);

                client.Connect();

                var request = BitConverter.GetBytes(index);
                client.Write(request, 0, request.Length);
                client.Flush();
                client.WaitForPipeDrain();
                Console.WriteLine($"{index} Ping.");

                var response = new byte[4];
                client.Read(response, 0, response.Length);
                index = BitConverter.ToInt32(response, 0);
                Console.WriteLine($"{index} Response.");
            }
            catch (Exception ex) {
                Console.WriteLine($"{index}\n\t{ex}");
            }
            finally {
                client?.Dispose();
            }
        }
    }
}

You can sort the messages and observe the following:

  1. Connections are opened and closed correctly.

  2. Data is sent and received correctly.

  3. Finally, the server still waits for further connections.

Updates:

  1. Changed PipeOptions.Asynchronous to PipeOptions.None otherwise it seems like it hangs for the duration of the requests and only then processing them at once.

    PipeOptions.Asynchronous is simply causing a different order of execution than PipeOptions.None, and that's exposing a race condition / deadlock in your code. You can see the effect of it if you use Task Manager, for example, to monitor the thread count of your process... you should see it creeping up at a rate of appx 1 thread per second, until it gets to around 100 threads (maybe 110 or so), at which point your code runs to completion. Or if you add ThreadPool.SetMinThreads(200, 200) at the beginning. Your code has a problem where if the wrong ordering occurs (and that's made more likely by using Asynchronous), you create a cycle where it can't be satisfied until there are enough threads to run all of the concurrent ConnectAsyncs your main method has queued, which aren't truly async and instead just create a work item to invoke the synchronous Connect method (this is unfortunate, and it's issues like this that are one of the reasons I urge folks not to expose async APIs that simply queue works items to call sync methods). Source.

  2. Revised and simplified the example:

    1. There's no true asynchronous Connect method for pipes, ConnectAsync uses Task.Factory.StartNew behind the scene so you might just as well use Connect and then pass the method (SendRequest in our example) that calls the synchronous Connect version to Task.Factory.StartNew.

    2. The server is completely asynchronous now and as far as I can tell it works with no issues.

    3. Added two implementations for the server one that uses callbacks and another one that takes advantage over the async/await feature just because I couldn't find a good example for these two.

I hope it helps.