From 47eabe7574322c9722f3e255f16fb0d537dafd0d Mon Sep 17 00:00:00 2001 From: Rob Hague Date: Tue, 6 Feb 2024 12:16:11 +0000 Subject: [PATCH] Tweak semaphore usage in Session (#1304) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Change _connectAndLazySemaphoreInitLock to a SemaphoreSlim and use it in ConnectAsync. - Rename it to _connectLock and only use it for connecting. Replace its other usages (on SessionSemaphore and NextChannelNumber) with Interlocked operations. - Remove AuthenticationConnection semaphore. This static member placed a process-wide limit on the number of connections an application can make. I agree with the argument in https://github.com/sshnet/SSH.NET/issues/409#issuecomment-457415542 (and in several other issues/PRs) that this should not be something that the library attempts to control. The last change broke a few tests which do things like making 100 connections. I was tempted to delete these tests as I don't think they have much value, but instead I just limited their concurrency. Co-authored-by: Wojciech Nagórski --- src/Renci.SshNet/Session.cs | 422 +++++++++--------- .../OldIntegrationTests/SshCommandTest.cs | 41 +- .../SftpTests.cs | 2 +- 3 files changed, 215 insertions(+), 250 deletions(-) diff --git a/src/Renci.SshNet/Session.cs b/src/Renci.SshNet/Session.cs index 122e28d3..193b1002 100644 --- a/src/Renci.SshNet/Session.cs +++ b/src/Renci.SshNet/Session.cs @@ -1,4 +1,5 @@ using System; +using System.Diagnostics; using System.Globalization; using System.Linq; using System.Net.Sockets; @@ -81,14 +82,6 @@ namespace Renci.SshNet /// internal static readonly TimeSpan InfiniteTimeSpan = new TimeSpan(0, 0, 0, 0, -1); - /// - /// Controls how many authentication attempts can take place at the same time. - /// - /// - /// Some server may restrict number to prevent authentication attacks. - /// - private static readonly SemaphoreSlim AuthenticationConnection = new SemaphoreSlim(3); - /// /// Holds the factory to use for creating new services. /// @@ -123,9 +116,9 @@ namespace Renci.SshNet /// /// Holds an object that is used to ensure only a single thread can connect - /// and lazy initialize the at any given time. + /// at any given time. /// - private readonly object _connectAndLazySemaphoreInitLock = new object(); + private readonly SemaphoreSlim _connectLock = new SemaphoreSlim(1, 1); /// /// Holds metadata about session messages. @@ -195,7 +188,7 @@ namespace Renci.SshNet private bool _isDisconnectMessageSent; - private uint _nextChannelNumber; + private int _nextChannelNumber; /// /// Holds connection socket. @@ -212,12 +205,18 @@ namespace Renci.SshNet { get { - if (_sessionSemaphore is null) + if (_sessionSemaphore is SemaphoreSlim sessionSemaphore) { - lock (_connectAndLazySemaphoreInitLock) - { - _sessionSemaphore ??= new SemaphoreSlim(ConnectionInfo.MaxSessions); - } + return sessionSemaphore; + } + + sessionSemaphore = new SemaphoreSlim(ConnectionInfo.MaxSessions); + + if (Interlocked.CompareExchange(ref _sessionSemaphore, sessionSemaphore, comparand: null) is not null) + { + // Another thread has set _sessionSemaphore. Dispose our one. + Debug.Assert(_sessionSemaphore != sessionSemaphore); + sessionSemaphore.Dispose(); } return _sessionSemaphore; @@ -234,14 +233,7 @@ namespace Renci.SshNet { get { - uint result; - - lock (_connectAndLazySemaphoreInitLock) - { - result = _nextChannelNumber++; - } - - return result; + return (uint)Interlocked.Increment(ref _nextChannelNumber); } } @@ -583,128 +575,116 @@ namespace Renci.SshNet return; } + _connectLock.Wait(); + try { - AuthenticationConnection.Wait(); - if (IsConnected) { return; } - lock (_connectAndLazySemaphoreInitLock) + // Reset connection specific information + Reset(); + + // Build list of available messages while connecting + _sshMessageFactory = new SshMessageFactory(); + + _socket = _serviceFactory.CreateConnector(ConnectionInfo, _socketFactory) + .Connect(ConnectionInfo); + + var serverIdentification = _serviceFactory.CreateProtocolVersionExchange() + .Start(ClientVersion, _socket, ConnectionInfo.Timeout); + + // Set connection versions + ServerVersion = ConnectionInfo.ServerVersion = serverIdentification.ToString(); + ConnectionInfo.ClientVersion = ClientVersion; + + DiagnosticAbstraction.Log(string.Format("Server version '{0}'.", serverIdentification)); + + if (!(serverIdentification.ProtocolVersion.Equals("2.0") || serverIdentification.ProtocolVersion.Equals("1.99"))) { - // If connected don't connect again - if (IsConnected) - { - return; - } - - // Reset connection specific information - Reset(); - - // Build list of available messages while connecting - _sshMessageFactory = new SshMessageFactory(); - - _socket = _serviceFactory.CreateConnector(ConnectionInfo, _socketFactory) - .Connect(ConnectionInfo); - - var serverIdentification = _serviceFactory.CreateProtocolVersionExchange() - .Start(ClientVersion, _socket, ConnectionInfo.Timeout); - - // Set connection versions - ServerVersion = ConnectionInfo.ServerVersion = serverIdentification.ToString(); - ConnectionInfo.ClientVersion = ClientVersion; - - DiagnosticAbstraction.Log(string.Format("Server version '{0}'.", serverIdentification)); - - if (!(serverIdentification.ProtocolVersion.Equals("2.0") || serverIdentification.ProtocolVersion.Equals("1.99"))) - { - throw new SshConnectionException(string.Format(CultureInfo.CurrentCulture, "Server version '{0}' is not supported.", serverIdentification.ProtocolVersion), - DisconnectReason.ProtocolVersionNotSupported); - } - - ServerIdentificationReceived?.Invoke(this, new SshIdentificationEventArgs(serverIdentification)); - - // Register Transport response messages - RegisterMessage("SSH_MSG_DISCONNECT"); - RegisterMessage("SSH_MSG_IGNORE"); - RegisterMessage("SSH_MSG_UNIMPLEMENTED"); - RegisterMessage("SSH_MSG_DEBUG"); - RegisterMessage("SSH_MSG_SERVICE_ACCEPT"); - RegisterMessage("SSH_MSG_KEXINIT"); - RegisterMessage("SSH_MSG_NEWKEYS"); - - // Some server implementations might sent this message first, prior to establishing encryption algorithm - RegisterMessage("SSH_MSG_USERAUTH_BANNER"); - - // Send our key exchange init. - // We need to do this before starting the message listener to avoid the case where we receive the server - // key exchange init and we continue the key exchange before having sent our own init. - SendMessage(ClientInitMessage); - - // Mark the message listener threads as started - _ = _messageListenerCompleted.Reset(); - - // Start incoming request listener - // ToDo: Make message pump async, to not consume a thread for every session - _ = ThreadAbstraction.ExecuteThreadLongRunning(MessageListener); - - // Wait for key exchange to be completed - WaitOnHandle(_keyExchangeCompletedWaitHandle.WaitHandle); - - // If sessionId is not set then its not connected - if (SessionId is null) - { - Disconnect(); - return; - } - - // Request user authorization service - SendMessage(new ServiceRequestMessage(ServiceName.UserAuthentication)); - - // Wait for service to be accepted - WaitOnHandle(_serviceAccepted); - - if (string.IsNullOrEmpty(ConnectionInfo.Username)) - { - throw new SshException("Username is not specified."); - } - - // Some servers send a global request immediately after successful authentication - // Avoid race condition by already enabling SSH_MSG_GLOBAL_REQUEST before authentication - RegisterMessage("SSH_MSG_GLOBAL_REQUEST"); - - ConnectionInfo.Authenticate(this, _serviceFactory); - _isAuthenticated = true; - - // Register Connection messages - RegisterMessage("SSH_MSG_REQUEST_SUCCESS"); - RegisterMessage("SSH_MSG_REQUEST_FAILURE"); - RegisterMessage("SSH_MSG_CHANNEL_OPEN_CONFIRMATION"); - RegisterMessage("SSH_MSG_CHANNEL_OPEN_FAILURE"); - RegisterMessage("SSH_MSG_CHANNEL_WINDOW_ADJUST"); - RegisterMessage("SSH_MSG_CHANNEL_EXTENDED_DATA"); - RegisterMessage("SSH_MSG_CHANNEL_REQUEST"); - RegisterMessage("SSH_MSG_CHANNEL_SUCCESS"); - RegisterMessage("SSH_MSG_CHANNEL_FAILURE"); - RegisterMessage("SSH_MSG_CHANNEL_DATA"); - RegisterMessage("SSH_MSG_CHANNEL_EOF"); - RegisterMessage("SSH_MSG_CHANNEL_CLOSE"); + throw new SshConnectionException(string.Format(CultureInfo.CurrentCulture, "Server version '{0}' is not supported.", serverIdentification.ProtocolVersion), + DisconnectReason.ProtocolVersionNotSupported); } + + ServerIdentificationReceived?.Invoke(this, new SshIdentificationEventArgs(serverIdentification)); + + // Register Transport response messages + RegisterMessage("SSH_MSG_DISCONNECT"); + RegisterMessage("SSH_MSG_IGNORE"); + RegisterMessage("SSH_MSG_UNIMPLEMENTED"); + RegisterMessage("SSH_MSG_DEBUG"); + RegisterMessage("SSH_MSG_SERVICE_ACCEPT"); + RegisterMessage("SSH_MSG_KEXINIT"); + RegisterMessage("SSH_MSG_NEWKEYS"); + + // Some server implementations might sent this message first, prior to establishing encryption algorithm + RegisterMessage("SSH_MSG_USERAUTH_BANNER"); + + // Send our key exchange init. + // We need to do this before starting the message listener to avoid the case where we receive the server + // key exchange init and we continue the key exchange before having sent our own init. + SendMessage(ClientInitMessage); + + // Mark the message listener threads as started + _ = _messageListenerCompleted.Reset(); + + // Start incoming request listener + // ToDo: Make message pump async, to not consume a thread for every session + _ = ThreadAbstraction.ExecuteThreadLongRunning(MessageListener); + + // Wait for key exchange to be completed + WaitOnHandle(_keyExchangeCompletedWaitHandle.WaitHandle); + + // If sessionId is not set then its not connected + if (SessionId is null) + { + Disconnect(); + return; + } + + // Request user authorization service + SendMessage(new ServiceRequestMessage(ServiceName.UserAuthentication)); + + // Wait for service to be accepted + WaitOnHandle(_serviceAccepted); + + if (string.IsNullOrEmpty(ConnectionInfo.Username)) + { + throw new SshException("Username is not specified."); + } + + // Some servers send a global request immediately after successful authentication + // Avoid race condition by already enabling SSH_MSG_GLOBAL_REQUEST before authentication + RegisterMessage("SSH_MSG_GLOBAL_REQUEST"); + + ConnectionInfo.Authenticate(this, _serviceFactory); + _isAuthenticated = true; + + // Register Connection messages + RegisterMessage("SSH_MSG_REQUEST_SUCCESS"); + RegisterMessage("SSH_MSG_REQUEST_FAILURE"); + RegisterMessage("SSH_MSG_CHANNEL_OPEN_CONFIRMATION"); + RegisterMessage("SSH_MSG_CHANNEL_OPEN_FAILURE"); + RegisterMessage("SSH_MSG_CHANNEL_WINDOW_ADJUST"); + RegisterMessage("SSH_MSG_CHANNEL_EXTENDED_DATA"); + RegisterMessage("SSH_MSG_CHANNEL_REQUEST"); + RegisterMessage("SSH_MSG_CHANNEL_SUCCESS"); + RegisterMessage("SSH_MSG_CHANNEL_FAILURE"); + RegisterMessage("SSH_MSG_CHANNEL_DATA"); + RegisterMessage("SSH_MSG_CHANNEL_EOF"); + RegisterMessage("SSH_MSG_CHANNEL_CLOSE"); } finally { - _ = AuthenticationConnection.Release(); + _ = _connectLock.Release(); } } /// /// Asynchronously connects to the server. /// - /// - /// Please note this function is NOT thread safe.
- /// The caller SHOULD limit the number of simultaneous connection attempts to a server to a single connection attempt. /// The to observe. /// A that represents the asynchronous connect operation. /// Socket connection to the SSH server or proxy server could not be established, or an error occurred while resolving the hostname. @@ -719,97 +699,111 @@ namespace Renci.SshNet return; } - // Reset connection specific information - Reset(); + await _connectLock.WaitAsync(cancellationToken).ConfigureAwait(false); - // Build list of available messages while connecting - _sshMessageFactory = new SshMessageFactory(); - - _socket = await _serviceFactory.CreateConnector(ConnectionInfo, _socketFactory) - .ConnectAsync(ConnectionInfo, cancellationToken).ConfigureAwait(false); - - var serverIdentification = await _serviceFactory.CreateProtocolVersionExchange() - .StartAsync(ClientVersion, _socket, cancellationToken).ConfigureAwait(false); - - // Set connection versions - ServerVersion = ConnectionInfo.ServerVersion = serverIdentification.ToString(); - ConnectionInfo.ClientVersion = ClientVersion; - - DiagnosticAbstraction.Log(string.Format("Server version '{0}'.", serverIdentification)); - - if (!(serverIdentification.ProtocolVersion.Equals("2.0") || serverIdentification.ProtocolVersion.Equals("1.99"))) + try { - throw new SshConnectionException(string.Format(CultureInfo.CurrentCulture, "Server version '{0}' is not supported.", serverIdentification.ProtocolVersion), - DisconnectReason.ProtocolVersionNotSupported); + if (IsConnected) + { + return; + } + + // Reset connection specific information + Reset(); + + // Build list of available messages while connecting + _sshMessageFactory = new SshMessageFactory(); + + _socket = await _serviceFactory.CreateConnector(ConnectionInfo, _socketFactory) + .ConnectAsync(ConnectionInfo, cancellationToken).ConfigureAwait(false); + + var serverIdentification = await _serviceFactory.CreateProtocolVersionExchange() + .StartAsync(ClientVersion, _socket, cancellationToken).ConfigureAwait(false); + + // Set connection versions + ServerVersion = ConnectionInfo.ServerVersion = serverIdentification.ToString(); + ConnectionInfo.ClientVersion = ClientVersion; + + DiagnosticAbstraction.Log(string.Format("Server version '{0}'.", serverIdentification)); + + if (!(serverIdentification.ProtocolVersion.Equals("2.0") || serverIdentification.ProtocolVersion.Equals("1.99"))) + { + throw new SshConnectionException(string.Format(CultureInfo.CurrentCulture, "Server version '{0}' is not supported.", serverIdentification.ProtocolVersion), + DisconnectReason.ProtocolVersionNotSupported); + } + + ServerIdentificationReceived?.Invoke(this, new SshIdentificationEventArgs(serverIdentification)); + + // Register Transport response messages + RegisterMessage("SSH_MSG_DISCONNECT"); + RegisterMessage("SSH_MSG_IGNORE"); + RegisterMessage("SSH_MSG_UNIMPLEMENTED"); + RegisterMessage("SSH_MSG_DEBUG"); + RegisterMessage("SSH_MSG_SERVICE_ACCEPT"); + RegisterMessage("SSH_MSG_KEXINIT"); + RegisterMessage("SSH_MSG_NEWKEYS"); + + // Some server implementations might sent this message first, prior to establishing encryption algorithm + RegisterMessage("SSH_MSG_USERAUTH_BANNER"); + + // Send our key exchange init. + // We need to do this before starting the message listener to avoid the case where we receive the server + // key exchange init and we continue the key exchange before having sent our own init. + SendMessage(ClientInitMessage); + + // Mark the message listener threads as started + _ = _messageListenerCompleted.Reset(); + + // Start incoming request listener + // ToDo: Make message pump async, to not consume a thread for every session + _ = ThreadAbstraction.ExecuteThreadLongRunning(MessageListener); + + // Wait for key exchange to be completed + WaitOnHandle(_keyExchangeCompletedWaitHandle.WaitHandle); + + // If sessionId is not set then its not connected + if (SessionId is null) + { + Disconnect(); + return; + } + + // Request user authorization service + SendMessage(new ServiceRequestMessage(ServiceName.UserAuthentication)); + + // Wait for service to be accepted + WaitOnHandle(_serviceAccepted); + + if (string.IsNullOrEmpty(ConnectionInfo.Username)) + { + throw new SshException("Username is not specified."); + } + + // Some servers send a global request immediately after successful authentication + // Avoid race condition by already enabling SSH_MSG_GLOBAL_REQUEST before authentication + RegisterMessage("SSH_MSG_GLOBAL_REQUEST"); + + ConnectionInfo.Authenticate(this, _serviceFactory); + _isAuthenticated = true; + + // Register Connection messages + RegisterMessage("SSH_MSG_REQUEST_SUCCESS"); + RegisterMessage("SSH_MSG_REQUEST_FAILURE"); + RegisterMessage("SSH_MSG_CHANNEL_OPEN_CONFIRMATION"); + RegisterMessage("SSH_MSG_CHANNEL_OPEN_FAILURE"); + RegisterMessage("SSH_MSG_CHANNEL_WINDOW_ADJUST"); + RegisterMessage("SSH_MSG_CHANNEL_EXTENDED_DATA"); + RegisterMessage("SSH_MSG_CHANNEL_REQUEST"); + RegisterMessage("SSH_MSG_CHANNEL_SUCCESS"); + RegisterMessage("SSH_MSG_CHANNEL_FAILURE"); + RegisterMessage("SSH_MSG_CHANNEL_DATA"); + RegisterMessage("SSH_MSG_CHANNEL_EOF"); + RegisterMessage("SSH_MSG_CHANNEL_CLOSE"); } - - ServerIdentificationReceived?.Invoke(this, new SshIdentificationEventArgs(serverIdentification)); - - // Register Transport response messages - RegisterMessage("SSH_MSG_DISCONNECT"); - RegisterMessage("SSH_MSG_IGNORE"); - RegisterMessage("SSH_MSG_UNIMPLEMENTED"); - RegisterMessage("SSH_MSG_DEBUG"); - RegisterMessage("SSH_MSG_SERVICE_ACCEPT"); - RegisterMessage("SSH_MSG_KEXINIT"); - RegisterMessage("SSH_MSG_NEWKEYS"); - - // Some server implementations might sent this message first, prior to establishing encryption algorithm - RegisterMessage("SSH_MSG_USERAUTH_BANNER"); - - // Send our key exchange init. - // We need to do this before starting the message listener to avoid the case where we receive the server - // key exchange init and we continue the key exchange before having sent our own init. - SendMessage(ClientInitMessage); - - // Mark the message listener threads as started - _ = _messageListenerCompleted.Reset(); - - // Start incoming request listener - // ToDo: Make message pump async, to not consume a thread for every session - _ = ThreadAbstraction.ExecuteThreadLongRunning(MessageListener); - - // Wait for key exchange to be completed - WaitOnHandle(_keyExchangeCompletedWaitHandle.WaitHandle); - - // If sessionId is not set then its not connected - if (SessionId is null) + finally { - Disconnect(); - return; + _ = _connectLock.Release(); } - - // Request user authorization service - SendMessage(new ServiceRequestMessage(ServiceName.UserAuthentication)); - - // Wait for service to be accepted - WaitOnHandle(_serviceAccepted); - - if (string.IsNullOrEmpty(ConnectionInfo.Username)) - { - throw new SshException("Username is not specified."); - } - - // Some servers send a global request immediately after successful authentication - // Avoid race condition by already enabling SSH_MSG_GLOBAL_REQUEST before authentication - RegisterMessage("SSH_MSG_GLOBAL_REQUEST"); - - ConnectionInfo.Authenticate(this, _serviceFactory); - _isAuthenticated = true; - - // Register Connection messages - RegisterMessage("SSH_MSG_REQUEST_SUCCESS"); - RegisterMessage("SSH_MSG_REQUEST_FAILURE"); - RegisterMessage("SSH_MSG_CHANNEL_OPEN_CONFIRMATION"); - RegisterMessage("SSH_MSG_CHANNEL_OPEN_FAILURE"); - RegisterMessage("SSH_MSG_CHANNEL_WINDOW_ADJUST"); - RegisterMessage("SSH_MSG_CHANNEL_EXTENDED_DATA"); - RegisterMessage("SSH_MSG_CHANNEL_REQUEST"); - RegisterMessage("SSH_MSG_CHANNEL_SUCCESS"); - RegisterMessage("SSH_MSG_CHANNEL_FAILURE"); - RegisterMessage("SSH_MSG_CHANNEL_DATA"); - RegisterMessage("SSH_MSG_CHANNEL_EOF"); - RegisterMessage("SSH_MSG_CHANNEL_CLOSE"); } /// diff --git a/test/Renci.SshNet.IntegrationTests/OldIntegrationTests/SshCommandTest.cs b/test/Renci.SshNet.IntegrationTests/OldIntegrationTests/SshCommandTest.cs index e5e78a76..aefe1d6d 100644 --- a/test/Renci.SshNet.IntegrationTests/OldIntegrationTests/SshCommandTest.cs +++ b/test/Renci.SshNet.IntegrationTests/OldIntegrationTests/SshCommandTest.cs @@ -442,47 +442,18 @@ namespace Renci.SshNet.IntegrationTests.OldIntegrationTests } } - [TestMethod] - public void Test_MultipleThread_Example_MultipleConnections() - { - try - { -#region Example SshCommand RunCommand Parallel - Parallel.For(0, 100, - () => - { - var client = new SshClient(SshServerHostName, SshServerPort, User.UserName, User.Password); - client.Connect(); - return client; - }, - (int counter, ParallelLoopState pls, SshClient client) => - { - var result = client.RunCommand("echo 123"); - Debug.WriteLine(string.Format("TestMultipleThreadMultipleConnections #{0}", counter)); - return client; - }, - (SshClient client) => - { - client.Disconnect(); - client.Dispose(); - } - ); -#endregion - - } - catch (Exception exp) - { - Assert.Fail(exp.ToString()); - } - } - [TestMethod] public void Test_MultipleThread_100_MultipleConnections() { try { - Parallel.For(0, 100, + var options = new ParallelOptions() + { + MaxDegreeOfParallelism = 8 + }; + + Parallel.For(0, 100, options, () => { var client = new SshClient(SshServerHostName, SshServerPort, User.UserName, User.Password); diff --git a/test/Renci.SshNet.IntegrationTests/SftpTests.cs b/test/Renci.SshNet.IntegrationTests/SftpTests.cs index ee00bb15..87b59c7a 100644 --- a/test/Renci.SshNet.IntegrationTests/SftpTests.cs +++ b/test/Renci.SshNet.IntegrationTests/SftpTests.cs @@ -83,7 +83,7 @@ namespace Renci.SshNet.IntegrationTests public void Sftp_ConnectDisconnect_Parallel() { const int iterations = 10; - const int threads = 20; + const int threads = 5; var startEvent = new ManualResetEvent(false);