10.4.1

URocket/Connection/Connection.Write.cs

@@ -30,13 +30,16 @@ public unsafe partial class Connection : IBufferWriter<byte>, IDisposable
     private readonly UnmanagedMemoryManager _manager;
 
     /// <summary>Pointer to the unmanaged write buffer.</summary>
-    public byte* WriteBuffer { get; private set; }
+    public byte* WriteBuffer { get; }
 
     /// <summary>Logical start of valid data (currently unused; reserved for future streaming).</summary>
     public int WriteHead { get; set; }
 
     /// <summary>Logical end of written data in <see cref="WriteBuffer"/>.</summary>
-    public int WriteTail { get; set; }
+    public int WriteTail { get; private set; }
+
+    /// <summary> Number of bytes that were sent to kernel, we need this value to validate whether all flushed data was processed by the kernel. </summary>
+    internal int WriteInFlight { get; set; }
 
     /// <summary>Indicates that the buffer may be flushed by the reactor.</summary>
     public volatile bool CanFlush = true;
@@ -59,7 +62,7 @@ public Connection(int writeSlabSize = 1024 * 16) {
     /// Copies unmanaged memory into the write slab.
     /// </summary>
     [MethodImpl(MethodImplOptions.AggressiveInlining)]
-    public void Write(byte* ptr, int length)
+    internal void Write(byte* ptr, int length)
     {
         if ((uint)length > (uint)_writeSlabSize) // also rejects negative
             throw new ArgumentOutOfRangeException(nameof(length));

URocket/Engine/Engine.Reactor.HandleSubmitAndWaitSingleCall.cs

@@ -143,8 +143,14 @@ internal void HandleSubmitAndWaitSingleCall()
                                 if (connection.WriteHead < connection.WriteTail) 
                                 {
                                     Console.WriteLine("Oddness");
-                                    connection.CanFlush = false;
-                                    SubmitSend(io_uring_instance, connection.ClientFd, connection.WriteBuffer, (uint)connection.WriteHead, (uint)connection.WriteTail);
+                                    //connection.CanFlush = false; This is unnecessary? 
+                                    SubmitSend(
+                                        io_uring_instance, 
+                                        connection.ClientFd,
+                                        connection.WriteBuffer, 
+                                        (uint)connection.WriteHead, 
+                                        (uint)(connection.WriteTail - connection.WriteHead));
+
                                     continue;
                                     // queued SQE; flushed next loop
                                 }

URocket/Engine/Engine.Reactor.cs

@@ -225,9 +225,16 @@ private void DrainWriteQ()
                 if (_engine.Connections[Id].TryGetValue(fd, out var connection))
                 {
                     connection.CanWrite = false; // Reset write flag for each drained connection
-
-                    if(connection.CanFlush)
-                        Send(connection.ClientFd, connection.WriteBuffer, (uint)connection.WriteHead, (uint)connection.WriteTail);
+
+                    if (connection.CanFlush)
+                    {
+                        Send(connection.ClientFd, connection.WriteBuffer, (uint)connection.WriteHead,
+                            (uint)connection.WriteTail);
+
+                        // Data is sent to kernel, until we don't have confirmation that
+                        // this data was fully processed by the kernel we don't send more data to it 
+                        connection.CanFlush = false;
+                    }
                 }
             }
         }

URocket/URocket.csproj

@@ -13,9 +13,9 @@
         <Authors>Diogo Martins</Authors>
         <RepositoryUrl>https://github.com/MDA2AV/uRocket</RepositoryUrl>
         <RepositoryType>git</RepositoryType>
-        <AssemblyVersion>10.4.0</AssemblyVersion>
-        <FileVersion>10.4.0</FileVersion>
-        <Version>10.4.0</Version>
+        <AssemblyVersion>10.4.1</AssemblyVersion>
+        <FileVersion>10.4.1</FileVersion>
+        <Version>10.4.1</Version>
         <PackageReadmeFile>README.md</PackageReadmeFile>
         <PackageTags>io_uring</PackageTags>
         <PackageLicenseFile>LICENSE</PackageLicenseFile>