Merge pull request #475 from MUnique/dev/network-improvements

Improvements for the network code

src/Network/Connection.cs

@@ -193,8 +193,8 @@ protected override async ValueTask OnCompleteAsync(Exception? exception)
     /// Reads the mu online packet by raising <see cref="PacketReceived" />.
     /// </summary>
     /// <param name="packet">The mu online packet.</param>
-    /// <returns>The async task.</returns>
-    protected override async ValueTask ReadPacketAsync(ReadOnlySequence<byte> packet)
+    /// <returns><see langword="true" />, if the flush was successful or not required.<see langword="false" />, if the pipe reader is completed and no longer reading data.</returns>
+    protected override async ValueTask<bool> ReadPacketAsync(ReadOnlySequence<byte> packet)
     {
         IncomingBytesCounter.Add(packet.Length);
 
@@ -205,6 +205,7 @@ protected override async ValueTask ReadPacketAsync(ReadOnlySequence<byte> packet
         try
         {
             await this.PacketReceived.SafeInvokeAsync(packet).ConfigureAwait(false);
+            return true;
         }
         finally
         {

src/Network/ExtendedPipeWriter.cs

@@ -9,7 +9,7 @@ namespace MUnique.OpenMU.Network;
 using System.Threading;
 
 /// <summary>
-/// A wrapper for an existing <see cref="PipeWriter"/>, which flushes the advanced bytes automatically.
+/// A wrapper for an existing <see cref="PipeWriter"/>, which has metrics about the written bytes.
 /// </summary>
 public class ExtendedPipeWriter : PipeWriter
 {

src/Network/PacketPipeReaderBase.cs

@@ -11,6 +11,11 @@ namespace MUnique.OpenMU.Network;
 /// <summary>
 /// Base class for all classes which read mu online data packets from a <see cref="PipeReader"/>.
 /// </summary>
+/// <remarks>
+/// Things to consider here:
+///  * Do not call FlushAsync if the reader can't start until FlushAsync finishes, as that may cause a deadlock.
+///  * Ensure that only one context "owns" a PipeReader or PipeWriter or accesses them. These types are not thread-safe.
+/// </remarks>
 public abstract class PacketPipeReaderBase
 {
     private readonly byte[] _headerBuffer = new byte[3];
@@ -24,8 +29,8 @@ public abstract class PacketPipeReaderBase
     /// Reads the mu online packet.
     /// </summary>
     /// <param name="packet">The mu online packet.</param>
-    /// <returns>The async task.</returns>
-    protected abstract ValueTask ReadPacketAsync(ReadOnlySequence<byte> packet);
+    /// <returns><see langword="true" />, if the flush was successful or not required.<see langword="false" />, if the pipe reader is completed and no longer reading data.</returns>
+    protected abstract ValueTask<bool> ReadPacketAsync(ReadOnlySequence<byte> packet);
 
     /// <summary>
     /// Called when the <see cref="Source"/> completed.
@@ -70,11 +75,32 @@ protected async Task ReadSourceAsync()
         await this.OnCompleteAsync(null).ConfigureAwait(false);
     }
 
+    /// <summary>
+    /// Tries to flush the writer.
+    /// </summary>
+    /// <param name="pipeWriter">The pipe writer.</param>
+    /// <returns>
+    ///   <see langword="true" />, if the flush was successful or not required.<see langword="false" />, if the pipe reader is completed and no longer reading data.
+    /// </returns>
+    protected async ValueTask<bool> TryFlushWriterAsync(PipeWriter pipeWriter)
+    {
+        if (pipeWriter is { CanGetUnflushedBytes: true, UnflushedBytes: 0 })
+        {
+            // It was flushed already in the background.
+            return true;
+        }
+
+        // todo: what happens if it was flushed in the background in the meantime? race-condition?
+        var flushResult = await pipeWriter.FlushAsync().ConfigureAwait(false);
+        return !flushResult.IsCompleted;
+    }
+
     private async Task<bool> ReadBufferAsync()
     {
         ReadResult result = await this.Source.ReadAsync().ConfigureAwait(false);
         ReadOnlySequence<byte> buffer = result.Buffer;
         int? length = null;
+        var readingCancelledOrCompleted = false;
         do
         {
             if (buffer.Length > 2)
@@ -97,7 +123,11 @@ private async Task<bool> ReadBufferAsync()
             if (length is > 0 && buffer.Length >= length)
             {
                 var packet = buffer.Slice(0, length.Value);
-                await this.ReadPacketAsync(packet).ConfigureAwait(false);
+                if (!await this.ReadPacketAsync(packet).ConfigureAwait(false))
+                {
+                    readingCancelledOrCompleted = true;
+                    break;
+                }
 
                 buffer = buffer.Slice(buffer.GetPosition(length.Value), buffer.End);
                 length = null;
@@ -121,6 +151,6 @@ private async Task<bool> ReadBufferAsync()
             this.Source.AdvanceTo(buffer.Start);
         }
 
-        return result.IsCompleted;
+        return result.IsCompleted || readingCancelledOrCompleted;
     }
 }

src/Network/SimpleModulus/PipelinedSimpleModulusDecryptor.cs

@@ -80,8 +80,8 @@ protected override ValueTask OnCompleteAsync(Exception? exception)
     /// Decrypts the packet and writes it into our pipe.
     /// </summary>
     /// <param name="packet">The mu online packet.</param>
-    /// <returns>The async task.</returns>
-    protected override async ValueTask ReadPacketAsync(ReadOnlySequence<byte> packet)
+    /// <returns><see langword="true" />, if the flush was successful or not required.<see langword="false" />, if the pipe reader is completed and no longer reading data.</returns>
+    protected override async ValueTask<bool> ReadPacketAsync(ReadOnlySequence<byte> packet)
     {
         // The next line is getting a span from the writer which is at least as big as the packet.
         // As I found out, it's initially about 2 kb in size and gets smaller within further
@@ -92,7 +92,7 @@ protected override async ValueTask ReadPacketAsync(ReadOnlySequence<byte> packet
         {
             // we just have to write-through
             this.CopyDataIntoWriter(this.Pipe.Writer, packet);
-            await this.Pipe.Writer.FlushAsync().ConfigureAwait(false);
+            return await this.TryFlushWriterAsync(this.Pipe.Writer).ConfigureAwait(false);
         }
         else
         {
@@ -105,7 +105,7 @@ protected override async ValueTask ReadPacketAsync(ReadOnlySequence<byte> packet
             }
 
             this.DecryptAndWrite(packet);
-            await this.Pipe.Writer.FlushAsync().ConfigureAwait(false);
+            return await this.TryFlushWriterAsync(this.Pipe.Writer).ConfigureAwait(false);
         }
     }
 

src/Network/SimpleModulus/PipelinedSimpleModulusEncryptor.cs

@@ -73,20 +73,19 @@ protected override ValueTask OnCompleteAsync(Exception? exception)
     }
 
     /// <inheritdoc />
-    protected override async ValueTask ReadPacketAsync(ReadOnlySequence<byte> packet)
+    protected override async ValueTask<bool> ReadPacketAsync(ReadOnlySequence<byte> packet)
     {
         packet.Slice(0, this.HeaderBuffer.Length).CopyTo(this.HeaderBuffer);
 
         if (this.HeaderBuffer[0] < 0xC3)
         {
             // we just have to write-through
             this.CopyDataIntoWriter(this._target, packet);
-            await this._target.FlushAsync().ConfigureAwait(false);
-            return;
+            return await this.TryFlushWriterAsync(this._target).ConfigureAwait(false);
         }
 
         this.EncryptAndWrite(packet);
-        await this._target.FlushAsync().ConfigureAwait(false);
+        return await this.TryFlushWriterAsync(this._target).ConfigureAwait(false);
     }
 
     /// <summary>

src/Network/Xor/PipelinedXor32Decryptor.cs

@@ -56,11 +56,11 @@ protected override ValueTask OnCompleteAsync(Exception? exception)
     /// Decrypts the packet and writes it into our pipe.
     /// </summary>
     /// <param name="packet">The mu online packet.</param>
-    /// <returns>The async task.</returns>
-    protected override async ValueTask ReadPacketAsync(ReadOnlySequence<byte> packet)
+    /// <returns><see langword="true" />, if the flush was successful or not required.<see langword="false" />, if the pipe reader is completed and no longer reading data.</returns>
+    protected override async ValueTask<bool> ReadPacketAsync(ReadOnlySequence<byte> packet)
     {
         this.DecryptAndWrite(packet);
-        await this._pipe.Writer.FlushAsync().ConfigureAwait(false);
+        return await this.TryFlushWriterAsync(this._pipe.Writer).ConfigureAwait(false);
     }
 
     private void DecryptAndWrite(ReadOnlySequence<byte> packet)

src/Network/Xor/PipelinedXor32Encryptor.cs

@@ -58,11 +58,11 @@ protected override ValueTask OnCompleteAsync(Exception? exception)
     /// Encrypts the packet and writes it into the target.
     /// </summary>
     /// <param name="packet">The mu online packet.</param>
-    /// <returns>The async task.</returns>
-    protected override async ValueTask ReadPacketAsync(ReadOnlySequence<byte> packet)
+    /// <returns><see langword="true" />, if the flush was successful or not required.<see langword="false" />, if the pipe reader is completed and no longer reading data.</returns>
+    protected override async ValueTask<bool> ReadPacketAsync(ReadOnlySequence<byte> packet)
     {
         this.EncryptAndWrite(packet);
-        await this._target.FlushAsync().ConfigureAwait(false);
+        return await this.TryFlushWriterAsync(this._target).ConfigureAwait(false);
     }
 
     private void EncryptAndWrite(ReadOnlySequence<byte> packet)