Merge pull request #881 from jpdillingham/chaos

jpdillingham · web-flow · commit c3ec1d8445a5 · 2025-10-09T19:14:28.000-05:00
Number of bug fixes and improvements around connection writes
diff --git a/src/Network/Tcp/Connection.cs b/src/Network/Tcp/Connection.cs
@@ -51,6 +51,10 @@ public Connection(IPEndPoint ipEndPoint, ConnectionOptions options = null, ITcpC
             // is pretty much the only option.
             Options.ConfigureSocket(TcpClient.Client);
 
+            // this should call SetSocketOptions on the socket (Client)
+            TcpClient.Client.ReceiveTimeout = Options.InactivityTimeout;
+            TcpClient.Client.SendTimeout = Options.InactivityTimeout;
+
             WriteQueueSemaphore = new SemaphoreSlim(Options.WriteQueueSize);
 
             if (Options.InactivityTimeout > 0)
@@ -89,7 +93,13 @@ public Connection(IPEndPoint ipEndPoint, ConnectionOptions options = null, ITcpC
                 State = ConnectionState.Connected;
                 InactivityTimer?.Start();
                 WatchdogTimer.Start();
+
                 Stream = TcpClient.GetStream();
+
+                // these should also call SetSocketOptions on the socket. doing it twice to make sure!
+                // read and write timeouts can cause the client to hang, even if the inactivity timer disconnects
+                Stream.WriteTimeout = Options.InactivityTimeout;
+                Stream.ReadTimeout = Options.InactivityTimeout;
             }
         }
 
@@ -191,6 +201,7 @@ public Connection(IPEndPoint ipEndPoint, ConnectionOptions options = null, ITcpC
         private TaskCompletionSource<string> DisconnectTaskCompletionSource { get; } = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
         private SemaphoreSlim WriteSemaphore { get; set; } = new SemaphoreSlim(initialCount: 1, maxCount: 1);
         private SemaphoreSlim WriteQueueSemaphore { get; set; }
+        private bool WriteQueueFull { get; set; }
 
         /// <summary>
         ///     Asynchronously connects the client to the configured <see cref="IPEndPoint"/>.
@@ -279,8 +290,14 @@ public async Task ConnectAsync(CancellationToken? cancellationToken = null)
 
                 InactivityTimer?.Start();
                 WatchdogTimer.Start();
+
                 Stream = TcpClient.GetStream();
 
+                // these should also call SetSocketOptions on the socket. doing it twice to make sure!
+                // read and write timeouts can cause the client to hang, even if the inactivity timer disconnects
+                Stream.ReadTimeout = Options.InactivityTimeout;
+                Stream.WriteTimeout = Options.InactivityTimeout;
+
                 ChangeState(ConnectionState.Connected, $"Connected to {IPEndPoint}");
             }
             catch (Exception ex)
@@ -698,24 +715,25 @@ private async Task WriteInternalAsync(byte[] bytes, CancellationToken cancellati
 
         private async Task WriteInternalAsync(long length, Stream inputStream, Func<int, CancellationToken, Task<int>> governor, Action<int, int, int> reporter, CancellationToken cancellationToken)
         {
+            // a failure to allocate memory will throw, so we need to do it within the try/catch
+            // declare and initialize it here so it's available in the finally block
+            byte[] buffer = Array.Empty<byte>();
+
             // in the case of a bad (or failing) connection, it is possible for us to continue to write data, particularly
             // distributed search requests, to the connection for quite a while before the underlying socket figures out that it
             // is in a bad state. when this happens memory usage skyrockets. see https://github.com/slskd/slskd/issues/251 for
-            // more information
-            if (WriteQueueSemaphore.CurrentCount == 0)
+            // more information.  note that this isn't for synchronization, it's to maintain a count of waiting writes.
+            if (WriteQueueFull || !await WriteQueueSemaphore.WaitAsync(0, cancellationToken).ConfigureAwait(false))
             {
+                // note: the semaphore check and this latch are not atomic! it's possible for one thread to fail to get the semaphore,
+                // and for the next to succeed (if one is released in the finally) before we set this to true. if that happens,
+                // *something* below will fail and the exception will bubble out (and if it doesn't throw, it probably got sent)
+                WriteQueueFull = true;
+
                 Disconnect("The write buffer is full");
                 throw new ConnectionWriteDroppedException($"Dropped buffered message to {IPEndPoint}; the write buffer is full");
             }
 
-            // a failure to allocate memory will throw, so we need to do it within the try/catch
-            // declare and initialize it here so it's available in the finally block
-            byte[] buffer = Array.Empty<byte>();
-
-            // grab a slot on the queue semaphore.  note that this isn't for synchronization, it's to
-            // maintain a count of waiting writes
-            await WriteQueueSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
-
             // obtain the write semaphore for this connection.  this keeps concurrent writes
             // from interleaving, which will mangle the messages on the receiving end
             await WriteSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
@@ -732,8 +750,13 @@ private async Task WriteInternalAsync(long length, Stream inputStream, Func<int,
 
                 long totalBytesWritten = 0;
 
-                while (!Disposed && totalBytesWritten < length)
+                while (totalBytesWritten < length)
                 {
+                    if (Disposed || State == ConnectionState.Disconnecting || State == ConnectionState.Disconnected)
+                    {
+                        throw new ConnectionWriteException($"Write aborted after {totalBytesWritten} bytes written; the connection has been or is being {(Disposed ? "disposed" : "disconnected")}");
+                    }
+
                     var bytesRemaining = length - totalBytesWritten;
                     var bytesToRead = bytesRemaining >= buffer.Length ? buffer.Length : (int)bytesRemaining;
 
@@ -781,12 +804,15 @@ private async Task WriteInternalAsync(long length, Stream inputStream, Func<int,
             }
             finally
             {
-                WriteQueueSemaphore.Release();
-                WriteSemaphore.Release();
-
 #if NETSTANDARD2_1_OR_GREATER
                 System.Buffers.ArrayPool<byte>.Shared.Return(buffer);
 #endif
+
+                if (!Disposed)
+                {
+                    WriteQueueSemaphore.Release();
+                    WriteSemaphore.Release();
+                }
             }
         }
     }
diff --git a/src/Network/Tcp/INetworkStream.cs b/src/Network/Tcp/INetworkStream.cs
@@ -28,6 +28,16 @@ namespace Soulseek.Network.Tcp
     /// </summary>
     internal interface INetworkStream : IDisposable
     {
+        /// <summary>
+        ///     Gets or sets the read timeout for the stream.
+        /// </summary>
+        int ReadTimeout { get; set; }
+
+        /// <summary>
+        ///     Gets or sets the write timeout for the stream.
+        /// </summary>
+        int WriteTimeout { get; set; }
+
         /// <summary>
         ///     Closes the <see cref="NetworkStream"/>.
         /// </summary>
diff --git a/src/Network/Tcp/NetworkStreamAdapter.cs b/src/Network/Tcp/NetworkStreamAdapter.cs
@@ -43,6 +43,42 @@ public NetworkStreamAdapter(NetworkStream networkStream)
             NetworkStream = networkStream;
         }
 
+        /// <summary>
+        ///     Gets or sets the read timeout for the <see cref="NetworkStream"/>.
+        /// </summary>
+        /// <remarks>
+        ///     Uses SetSocketOption under the hood: https://github.com/microsoft/referencesource/blob/main/System/net/System/Net/Sockets/NetworkStream.cs.
+        /// </remarks>
+        public int ReadTimeout
+        {
+            get
+            {
+                return NetworkStream.ReadTimeout;
+            }
+            set
+            {
+                NetworkStream.ReadTimeout = value;
+            }
+        }
+
+        /// <summary>
+        ///     Gets or sets the write timeout for the <see cref="NetworkStream"/>.
+        /// </summary>
+        /// <remarks>
+        ///     Uses SetSocketOption under the hood: https://github.com/microsoft/referencesource/blob/main/System/net/System/Net/Sockets/NetworkStream.cs.
+        /// </remarks>
+        public int WriteTimeout
+        {
+            get
+            {
+                return NetworkStream.WriteTimeout;
+            }
+            set
+            {
+                NetworkStream.WriteTimeout = value;
+            }
+        }
+
         private bool Disposed { get; set; }
         private NetworkStream NetworkStream { get; set; }
 
diff --git a/src/Soulseek.csproj b/src/Soulseek.csproj
@@ -32,7 +32,7 @@
 
   <PropertyGroup>
     <GeneratePackageOnBuild>true</GeneratePackageOnBuild>
-    <Version>8.2.1</Version>
+    <Version>8.3.0</Version>
     <Authors>JP Dillingham</Authors>
     <Product>Soulseek.NET</Product>
     <PackageProjectUrl>https://github.com/jpdillingham/Soulseek.NET</PackageProjectUrl>
diff --git a/src/SoulseekClient.cs b/src/SoulseekClient.cs
@@ -3185,6 +3185,7 @@ void UpdateProgress(long bytesDownloaded)
 
                 var endpoint = await GetUserEndPointAsync(username, cancellationToken).ConfigureAwait(false);
                 var peerConnection = await PeerConnectionManager.GetOrAddMessageConnectionAsync(username, endpoint, cancellationToken).ConfigureAwait(false);
+                Diagnostic.Debug($"Fetched peer connection for download of {Path.GetFileName(download.Filename)} from {username} (id: {peerConnection.Id}, state: {peerConnection.State})");
 
                 // prepare two waits; one for the transfer response to confirm that our request is acknowledged and another for
                 // the eventual transfer request sent when the peer is ready to send the file. the response message should be
@@ -3195,9 +3196,12 @@ void UpdateProgress(long bytesDownloaded)
 
                 // request the file
                 await peerConnection.WriteAsync(new TransferRequest(TransferDirection.Download, token, remoteFilename), cancellationToken).ConfigureAwait(false);
+                Diagnostic.Debug($"Wrote transfer request for download of {Path.GetFileName(download.Filename)} from {username} (id: {peerConnection.Id}, state: {peerConnection.State})");
+
                 UpdateState(TransferStates.Requested);
 
                 var transferRequestAcknowledgement = await transferRequestAcknowledged.ConfigureAwait(false);
+                Diagnostic.Debug($"Received transfer request ACK for download of {Path.GetFileName(download.Filename)} from {username}: allowed: {transferRequestAcknowledgement.IsAllowed}, message: {transferRequestAcknowledgement.Message} (token: {token})");
 
                 if (transferRequestAcknowledgement.IsAllowed)
                 {
@@ -3221,6 +3225,7 @@ void UpdateProgress(long bytesDownloaded)
                     download.Connection = await PeerConnectionManager
                         .GetTransferConnectionAsync(username, endpoint, transferRequestAcknowledgement.Token, cancellationToken)
                         .ConfigureAwait(false);
+                    Diagnostic.Debug($"Fetched transfer connection for download of {Path.GetFileName(download.Filename)} from {username} (id: {download.Connection.Id}, state: {download.Connection.State})");
                 }
                 else if (!string.Equals(transferRequestAcknowledgement.Message.TrimEnd('.'), "Queued", StringComparison.OrdinalIgnoreCase))
                 {
@@ -3252,6 +3257,7 @@ void UpdateProgress(long bytesDownloaded)
                     peerConnection = await PeerConnectionManager
                         .GetOrAddMessageConnectionAsync(username, endpoint, cancellationToken)
                         .ConfigureAwait(false);
+                    Diagnostic.Debug($"Fetched peer connection for download of {Path.GetFileName(download.Filename)} from {username} (id: {peerConnection.Id}, state: {peerConnection.State})");
 
                     // prepare a wait for the eventual transfer connection
                     var connectionTask = PeerConnectionManager
@@ -3263,6 +3269,7 @@ void UpdateProgress(long bytesDownloaded)
                     try
                     {
                         download.Connection = await connectionTask.ConfigureAwait(false);
+                        Diagnostic.Debug($"Fetched transfer connection for download of {Path.GetFileName(download.Filename)} from {username} (id: {download.Connection.Id}, state: {download.Connection.State})");
                     }
                     catch (ConnectionException)
                     {
@@ -4256,17 +4263,22 @@ void UpdateProgress(long bytesUploaded)
                 var messageConnection = await PeerConnectionManager
                     .GetOrAddMessageConnectionAsync(username, endpoint, cancellationToken)
                     .ConfigureAwait(false);
+                Diagnostic.Debug($"Fetched peer connection for upload of {Path.GetFileName(upload.Filename)} to {username} (id: {messageConnection.Id}, state: {messageConnection.State})");
 
                 // prepare a wait for the transfer response
                 var transferRequestAcknowledged = Waiter.Wait<TransferResponse>(
                     new WaitKey(MessageCode.Peer.TransferResponse, upload.Username, upload.Token), Options.PeerConnectionOptions.InactivityTimeout, cancellationToken);
 
                 // request to start the upload
                 var transferRequest = new TransferRequest(TransferDirection.Upload, upload.Token, upload.Filename, size);
+
                 await messageConnection.WriteAsync(transferRequest, cancellationToken).ConfigureAwait(false);
+                Diagnostic.Debug($"Wrote transfer request for upload of {Path.GetFileName(upload.Filename)} to {username} (id: {messageConnection.Id}, state: {messageConnection.State})");
+
                 UpdateState(TransferStates.Requested);
 
                 var transferRequestAcknowledgement = await transferRequestAcknowledged.ConfigureAwait(false);
+                Diagnostic.Debug($"Received transfer request ACK for upload of {Path.GetFileName(upload.Filename)} to {username}: allowed: {transferRequestAcknowledgement.IsAllowed}, message: {transferRequestAcknowledgement.Message} (token: {token})");
 
                 if (!transferRequestAcknowledgement.IsAllowed)
                 {
@@ -4278,6 +4290,7 @@ void UpdateProgress(long bytesUploaded)
                 upload.Connection = await PeerConnectionManager
                     .GetTransferConnectionAsync(upload.Username, endpoint, upload.Token, cancellationToken)
                     .ConfigureAwait(false);
+                Diagnostic.Debug($"Fetched transfer connection for upload of {Path.GetFileName(upload.Filename)} to {username} (id: {upload.Connection.Id}, state: {upload.Connection.State})");
 
                 // create a task completion source that represents the disconnect of the transfer connection. this is one of two tasks that will 'race'
                 // to determine the outcome of the upload.
diff --git a/tests/Soulseek.Tests.Unit/Network/Tcp/ConnectionTests.cs b/tests/Soulseek.Tests.Unit/Network/Tcp/ConnectionTests.cs
