Pipelines

Author: danielmarbach

Diff for: Async.Netcore.csproj

@@ -12,4 +12,8 @@
     </None>
   </ItemGroup>
 
+  <ItemGroup>
+    <PackageReference Include="System.IO.Pipelines" Version="4.6.0-preview5.19224.8" />
+  </ItemGroup>
+
 </Project>

Diff for: PipelineExtensions.cs

@@ -0,0 +1,99 @@
+using System;
+using System.Buffers;
+using System.IO;
+using System.IO.Pipelines;
+using System.Net;
+using System.Net.Sockets;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+
+static class PipelinesExtensions
+{
+    public static void Explain(this Pipelines runnable, TextWriter writer)
+    {
+        writer.WriteLine(@"
+- All buffer management is delegated to the `PipeReader`/`PipeWriter` implementations.`
+- Besides handling the memory management, the other core pipelines feature is the ability to peek at data in the Pipe without actually consuming it.
+- `FlushAsync` provides back pressure and flow control. PipeWriter.FlushAsync “blocks” when the amount of data in the Pipe crosses PauseWriterThreshold and “unblocks” when it becomes lower than ResumeWriterThreshold.
+- `PipeScheduler` gives fine grained control over scheduling the IO.
+");
+    }
+
+    public static void ProcessLine(this Pipelines runnable, Socket socket, in ReadOnlySequence<byte> buffer)
+    {
+        Console.Write($"[{socket.RemoteEndPoint}]: ");
+        foreach (var segment in buffer)
+        {
+            Console.Write(Encoding.UTF8.GetString(segment.Span));
+        }
+        Console.WriteLine();
+    }
+
+    public static void ReadUntilEOLAndOutputToConsole(this Pipelines runnable, Socket socket, ReadOnlySequence<byte> buffer)
+    {
+        SequencePosition? position = null;
+        do
+        {
+            // Find the EOL
+            position = buffer.PositionOf((byte)'\n');
+
+            if (position != null)
+            {
+                var line = buffer.Slice(0, position.Value);
+                runnable.ProcessLine(socket, line);
+
+                // This is equivalent to position + 1
+                var next = buffer.GetPosition(1, position.Value);
+
+                // Skip what we've already processed including \n
+                buffer = buffer.Slice(next);
+            }
+        }
+        while (position != null);
+    }
+
+    public static async Task Send(this Pipelines runnable, CancellationTokenSource tokenSource)
+    {
+        var clientSocket = new Socket(SocketType.Stream, ProtocolType.Tcp);
+        await clientSocket.ConnectAsync(new IPEndPoint(IPAddress.Loopback, 8087));
+        Console.WriteLine("Connecting to port 8087");
+
+        string line;
+        while ((line = Console.ReadLine()) != "exit")
+        {
+            var buffer = Encoding.ASCII.GetBytes(line.Replace(Environment.NewLine, string.Empty) + Environment.NewLine);
+
+            await clientSocket.SendAsync(new ArraySegment<byte>(buffer), SocketFlags.None);
+        }
+
+        tokenSource.Cancel();
+
+        Console.WriteLine("Send done");
+    }
+
+    public static async Task Receive(this Pipelines runnable, Pipe pipe, CancellationToken token)
+    {
+        var listenSocket = new Socket(SocketType.Stream, ProtocolType.Tcp);
+        listenSocket.Bind(new IPEndPoint(IPAddress.Loopback, 8087));
+        Console.WriteLine("Listening on port 8087");
+        listenSocket.Listen(120);
+
+        using (token.Register(() => listenSocket.Close()))
+        {
+            while (!token.IsCancellationRequested)
+            {
+                Socket socket;
+                try
+                {
+                    socket = await listenSocket.AcceptAsync();
+                }
+                catch (SocketException)
+                {
+                    return;
+                }
+                _ = runnable.ProcessLinesAsync(socket, pipe, token);
+            }
+        }
+    }
+}

Diff for: Pipelines.cs

@@ -0,0 +1,98 @@
+using System;
+using System.Buffers;
+using System.IO.Pipelines;
+using System.Net;
+using System.Net.Sockets;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+
+class Pipelines : IRunnable
+{
+    public async Task Run()
+    {
+        var tokenSource = new CancellationTokenSource();
+        var token = tokenSource.Token;
+
+        var pipe = new Pipe(new PipeOptions(useSynchronizationContext: false));
+
+        var receiveTask = this.Receive(pipe, token);
+        var sendTask = this.Send(tokenSource);
+
+        await Task.WhenAll(receiveTask, sendTask);
+    }
+
+    public async Task ProcessLinesAsync(Socket socket, Pipe pipe, CancellationToken token)
+    {
+        Console.WriteLine($"[{socket.RemoteEndPoint}]: connected");
+
+        Task writing = FillPipeAsync(socket, pipe.Writer, token);
+        Task reading = ReadPipeAsync(socket, pipe.Reader, token);
+
+        await Task.WhenAll(reading, writing).IgnoreCancellation();
+
+        Console.WriteLine($"[{socket.RemoteEndPoint}]: disconnected");
+    }
+
+    private static async Task FillPipeAsync(Socket socket, PipeWriter writer, CancellationToken token)
+    {
+        const int minimumBufferSize = 512;
+
+        while (!token.IsCancellationRequested)
+        {
+            try
+            {
+                // Request a minimum of 512 bytes from the PipeWriter
+                Memory<byte> memory = writer.GetMemory(minimumBufferSize);
+                int bytesRead = await socket.ReceiveAsync(memory, SocketFlags.None, token);
+                if (bytesRead == 0)
+                {
+                    break;
+                }
+
+                // Tell the PipeWriter how much was read
+                writer.Advance(bytesRead);
+            }
+            catch
+            {
+                break;
+            }
+
+            // Make the data available to the PipeReader
+            FlushResult result = await writer.FlushAsync(token);
+
+            if (result.IsCompleted)
+            {
+                break;
+            }
+        }
+
+        // Signal to the reader that we're done writing
+        writer.Complete();
+    }
+
+    async Task ReadPipeAsync(Socket socket, PipeReader reader, CancellationToken token)
+    {
+        while (!token.IsCancellationRequested)
+        {
+            ReadResult result = await reader.ReadAsync(token);
+
+            ReadOnlySequence<byte> buffer = result.Buffer;
+
+            this.ReadUntilEOLAndOutputToConsole(socket, buffer);
+
+            // We sliced the buffer until no more data could be processed
+            // Tell the PipeReader how much we consumed and how much we left to process
+            reader.AdvanceTo(buffer.Start, buffer.End);
+
+            if (result.IsCompleted)
+            {
+                break;
+            }
+        }
+
+        reader.Complete();
+    }
+
+
+}

Diff for: README.md

@@ -6,7 +6,14 @@
 
 ## AsyncEnumerable
 
-- `Stream`, `Utf8JsonWriter`, `System.Threading.Timer`, `CancellationTokenRegistration`, `BinaryWriter`, `TextWriter` and `IAsyncEnumerator<T>` implement `IAsyncDisposable`
-- `Stream`, `BinaryWriter`, `TextWriter`  calls `.Dispose` synchronously
-- `Stream` `FlushAsync` calls `Flush` on another thread which is bad behavior that should be overwritten
+- `IAsyncEnumerable<T>` allows to write asynchronous pull based streams, similar to regular enumerables with `yield return` and `yield break`
+- `WithCancellation` only adds the token to the enumerator but doesn't influence the state machine
+- `WithCancellation` in combination with `[EnumeratorCancellation]` can be used to ceate a combined token
+
+## Pipelines
+
+- All buffer management is delegated to the `PipeReader`/`PipeWriter` implementations.`
+- Besides handling the memory management, the other core pipelines feature is the ability to peek at data in the Pipe without actually consuming it.
+- `FlushAsync` provides back pressure and flow control. PipeWriter.FlushAsync “blocks” when the amount of data in the Pipe crosses PauseWriterThreshold and “unblocks” when it becomes lower than ResumeWriterThreshold.
+- `PipeScheduler` gives fine grained control over scheduling the IO.
 

Diff for: TaskExtensions.cs

@@ -0,0 +1,16 @@
+using System;
+using System.Threading.Tasks;
+
+static class TaskExtensions
+{
+    public static async Task IgnoreCancellation(this Task task)
+    {
+        try
+        {
+            await task.ConfigureAwait(false);
+        }
+        catch (OperationCanceledException)
+        {
+        }
+    }
+}