Using NamedPipeServerStream and NamedPipeClientStream asynchronously

Here is the complete code after some iterations:

PipeServer.cs:

namespace AsyncPipes;

using System.Diagnostics.CodeAnalysis;
using System.IO.Pipes;

public static class PipeServer
{
    public static void WaitForConnection()
        => WaitForConnectionInitializer();

    private static void WaitForConnectionInitializer()
    {
        var context = new ServerContext();
        var server = context.Server;
    
        try
        {
            Console.WriteLine($"Waiting a client...");

            server.BeginWaitForConnection(WaitForConnectionCallback, context);
        }
        catch
        {
            // We need to cleanup here only when something goes wrong.
            context.Dispose();

            throw;
        }

        static void WaitForConnectionCallback(IAsyncResult result)
        {
            var (context, server, _) = ServerContext.FromResult(result);

            server.EndWaitForConnection(result);

            WaitForConnectionInitializer();

            BeginRead(context);
        }

        static void BeginRead(ServerContext context)
        {
            var (_, server, requestBuffer) = context;
            
            server.BeginRead(requestBuffer, 0, requestBuffer.Length, ReadCallback, context);
        }

        static void BeginWrite(ServerContext context)
        {
            var (_, server, responseBuffer) = context;
            
            server.BeginWrite(responseBuffer, 0, responseBuffer.Length, WriteCallback, context);
        }

        static void ReadCallback(IAsyncResult result)
        {
            var (context, server, requestBuffer) = ServerContext.FromResult(result);

            var bytesRead = server.EndRead(result);

            if (bytesRead > 0)
            {
                if (!server.IsMessageComplete)
                {
                    BeginRead(context);
                }
                else
                {
                    var index = BitConverter.ToInt32(requestBuffer, 0);
                    Console.WriteLine($"{index} Request.");

                    BeginWrite(context);
                }
            }
        }

        static void WriteCallback(IAsyncResult result)
        {
            var (context, server, responseBuffer) = ServerContext.FromResult(result);
            var index = -1;
            
            try
            {
                server.EndWrite(result);
                server.WaitForPipeDrain();
                
                index = BitConverter.ToInt32(responseBuffer, 0);
                Console.WriteLine($"{index} Pong.");
            }
            finally
            {
                context.Dispose();
                Console.WriteLine($"{index} Disposed.");
            }
        }
    }

    private sealed class ServerContext : IDisposable
    {
        [NotNull]
        public byte[]? Buffer { get; private set; } = new byte[4];

        [NotNull]
        public NamedPipeServerStream? Server { get; private set; } = new ("PipesDemo",
                                                                        PipeDirection.InOut,
                                                                        NamedPipeServerStream.MaxAllowedServerInstances,
                                                                        PipeTransmissionMode.Message,
                                                                        PipeOptions.Asynchronous);

        public void Deconstruct(out ServerContext context, out NamedPipeServerStream server, out byte[] buffer)
            => (context, server, buffer) = (this, Server, Buffer);

        public static ServerContext FromResult(IAsyncResult result)
        {
            ArgumentNullException.ThrowIfNull(result.AsyncState);
            
            return (ServerContext)result.AsyncState;
        }
        
        public void Dispose()
        {
            if (Server is not null)
            {
                if (Server.IsConnected)
                {
                    Server.Disconnect();
                }
                
                Server.Dispose();
            }
            
            Server = null;
            Buffer = null;
        }
    }
}

PipeClient:

public static class PipeClient
{
    public static void CreateConnection(int index)
    {
        using var client = new NamedPipeClientStream(".", "PipesDemo", PipeDirection.InOut, PipeOptions.None);
        client.Connect();

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

        var responseBuffer = new byte[4];
        var bytesRead = client.Read(responseBuffer, 0, responseBuffer.Length);

        while (bytesRead > 0)
        {
            bytesRead = client.Read(responseBuffer, bytesRead - 1, responseBuffer.Length - bytesRead);
        }

        index = BitConverter.ToInt32(responseBuffer, 0);
        Console.WriteLine($"{index} Response.");
    }
}

Program.cs:

namespace AsyncPipes;

internal class Program
{
    private const int MaxRequests = 1000;

    private static void Main()
    {
        var tasks = new List<Task>
        {
            Task.Run(PipeServer.WaitForConnection)
        };

        tasks.AddRange(Enumerable.Range(0, MaxRequests - 1)
                                .Select(i => Task.Factory.StartNew(() => PipeClient.CreateConnection(i),
                                                                    TaskCreationOptions.LongRunning)));

        Task.WaitAll(tasks.ToArray());

        Console.ReadKey();
    }
}

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:

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.

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. Fixed all of the BeginXXX/EndXXX methods.

4. Removed unnecessary try/catch blocks.

5. Removed unnecessary messages.

6. Refactor the code a bit to make it more readable and concise.

7. Removed the async/await version of the server as I refactored the code and didn't have time to update the async/await version but with the above version you can have an idea of how to do it and the new APIs are much more friendly and easy to deal with.

I hope it helps.