From fdbc4d3e366a48a401b9a0d4ebca44e94d9e4ea1 Mon Sep 17 00:00:00 2001 From: Rob Hague Date: Thu, 23 May 2024 14:30:05 +0100 Subject: [PATCH] Fix a few issues with PipeStream (#1399) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Apply similar treatment to PipeStream as #1322 did to ShellStream PipeStream now behaves much more Stream-like. In particular, it performs partial reads (instead of blocking until a certain amount of data is available), blocks until data is available (instead of returning 0 prematurely) and removes the Stream-unlike properties `BlockLastReadBuffer` and `MaxBufferLength`. Sadly I gave up trying to make a benchmark compatible with all the quirks of the previous implementation, but a dumb throughput test (reading and writing simultaneously) shows about 5.2GB/s with this implementation compared to 140MB/s previously. Some cleanup of its usage in the library followed. Co-authored-by: Wojciech Nagórski --- src/Renci.SshNet/Common/PipeStream.cs | 427 +++++------------- src/Renci.SshNet/ScpClient.cs | 6 + src/Renci.SshNet/ShellStream.cs | 4 +- src/Renci.SshNet/SshCommand.cs | 109 ++--- .../SshClientBenchmark.cs | 12 + .../SshClientTests.cs | 22 + .../Renci.SshNet.IntegrationTests/SshTests.cs | 24 +- test/Renci.SshNet.Tests/.editorconfig | 10 +- .../Classes/Common/PipeStreamTest.cs | 126 ++++-- .../Common/PipeStream_Close_BlockingRead.cs | 54 --- .../Common/PipeStream_Close_BlockingWrite.cs | 66 --- ...ipeStream_Flush_BytesRemainingAfterRead.cs | 28 -- ...eStream_Flush_NoBytesRemainingAfterRead.cs | 63 --- .../Classes/PipeStreamTest_Dispose.cs | 64 +-- .../Classes/SshCommandTest_Dispose.cs | 18 +- .../SshCommandTest_EndExecute_ChannelOpen.cs | 5 +- 16 files changed, 339 insertions(+), 699 deletions(-) delete mode 100644 test/Renci.SshNet.Tests/Classes/Common/PipeStream_Close_BlockingRead.cs delete mode 100644 test/Renci.SshNet.Tests/Classes/Common/PipeStream_Close_BlockingWrite.cs delete mode 100644 test/Renci.SshNet.Tests/Classes/Common/PipeStream_Flush_NoBytesRemainingAfterRead.cs diff --git a/src/Renci.SshNet/Common/PipeStream.cs b/src/Renci.SshNet/Common/PipeStream.cs index da7b89cb..db211bf1 100644 --- a/src/Renci.SshNet/Common/PipeStream.cs +++ b/src/Renci.SshNet/Common/PipeStream.cs @@ -1,6 +1,6 @@ -using System; -using System.Collections.Generic; -using System.Globalization; +#nullable enable +using System; +using System.Diagnostics; using System.IO; using System.Threading; @@ -10,356 +10,175 @@ namespace Renci.SshNet.Common /// PipeStream is a thread-safe read/write data stream for use between two threads in a /// single-producer/single-consumer type problem. /// - /// - /// Copyright (c) 2006 James Kolpack (james dot kolpack at google mail) - /// - /// Permission is hereby granted, free of charge, to any person obtaining a copy of this software and - /// associated documentation files (the "Software"), to deal in the Software without restriction, - /// including without limitation the rights to use, copy, modify, merge, publish, distribute, - /// sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is - /// furnished to do so, subject to the following conditions: - /// - /// The above copyright notice and this permission notice shall be included in all copies or - /// substantial portions of the Software. - /// - /// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, - /// INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR - /// PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE - /// LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT - /// OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR - /// OTHER DEALINGS IN THE SOFTWARE. - /// public class PipeStream : Stream { - /// - /// Queue of bytes provides the datastructure for transmitting from an - /// input stream to an output stream. - /// - /// Possible more effecient ways to accomplish this. - private readonly Queue _buffer = new Queue(); + private readonly object _sync = new object(); - /// - /// Indicates that the input stream has been flushed and that - /// all remaining data should be written to the output stream. - /// - private bool _isFlushed; + private byte[] _buffer = new byte[1024]; + private int _head; // The index from which the data starts in _buffer. + private int _tail; // The index at which to add new data into _buffer. + private bool _disposed; - /// - /// Setting this to true will cause Read() to block if it appears - /// that it will run out of data. - /// - private bool _canBlockLastRead; - - /// - /// Indicates whether the current is disposed. - /// - private bool _isDisposed; - - /// - /// Gets or sets the maximum number of bytes to store in the buffer. - /// - /// The length of the max buffer. - public long MaxBufferLength { get; set; } = 200 * 1024 * 1024; - - /// - /// Gets or sets a value indicating whether to block last read method before the buffer is empty. - /// When true, Read() will block until it can fill the passed in buffer and count. - /// When false, Read() will not block, returning all the available buffer data. - /// - /// - /// Setting to true will remove the possibility of ending a stream reader prematurely. - /// - /// - /// if block last read method before the buffer is empty; otherwise, . - /// - /// Methods were called after the stream was closed. - public bool BlockLastReadBuffer +#pragma warning disable MA0076 // Do not use implicit culture-sensitive ToString in interpolated strings + [Conditional("DEBUG")] + private void AssertValid() { - get - { -#if NET7_0_OR_GREATER - ObjectDisposedException.ThrowIf(_isDisposed, this); -#else - if (_isDisposed) - { - throw CreateObjectDisposedException(); - } -#endif // NET7_0_OR_GREATER - - return _canBlockLastRead; - } - set - { -#if NET7_0_OR_GREATER - ObjectDisposedException.ThrowIf(_isDisposed, this); -#else - if (_isDisposed) - { - throw CreateObjectDisposedException(); - } -#endif // NET7_0_OR_GREATER - - _canBlockLastRead = value; - - // when turning off the block last read, signal Read() that it may now read the rest of the buffer. - if (!_canBlockLastRead) - { - lock (_buffer) - { - Monitor.Pulse(_buffer); - } - } - } + Debug.Assert(Monitor.IsEntered(_sync), $"Should be in lock on {nameof(_sync)}"); + Debug.Assert(_head >= 0, $"{nameof(_head)} should be non-negative but is {_head}"); + Debug.Assert(_tail >= 0, $"{nameof(_tail)} should be non-negative but is {_tail}"); + Debug.Assert(_head <= _buffer.Length, $"{nameof(_head)} should be <= {nameof(_buffer)}.Length but is {_head}"); + Debug.Assert(_tail <= _buffer.Length, $"{nameof(_tail)} should be <= {nameof(_buffer)}.Length but is {_tail}"); + Debug.Assert(_head <= _tail, $"Should have {nameof(_head)} <= {nameof(_tail)} but have {_head} <= {_tail}"); } +#pragma warning restore MA0076 // Do not use implicit culture-sensitive ToString in interpolated strings /// - /// When overridden in a derived class, clears all buffers for this stream and causes any buffered data to be written to the underlying device. + /// This method does nothing. /// - /// An I/O error occurs. - /// Methods were called after the stream was closed. - /// - /// Once flushed, any subsequent read operations no longer block until requested bytes are available. Any write operation reactivates blocking - /// reads. - /// public override void Flush() { -#if NET7_0_OR_GREATER - ObjectDisposedException.ThrowIf(_isDisposed, this); -#else - if (_isDisposed) - { - throw CreateObjectDisposedException(); - } -#endif // NET7_0_OR_GREATER - - _isFlushed = true; - lock (_buffer) - { - // unblock read hereby allowing buffer to be partially filled - Monitor.Pulse(_buffer); - } } /// - /// When overridden in a derived class, sets the position within the current stream. + /// This method always throws . /// - /// - /// The new position within the current stream. - /// - /// A byte offset relative to the origin parameter. + /// A byte offset relative to the parameter. /// A value of type indicating the reference point used to obtain the new position. - /// The stream does not support seeking, such as if the stream is constructed from a pipe or console output. + /// Never. + /// Always. public override long Seek(long offset, SeekOrigin origin) { throw new NotSupportedException(); } /// - /// When overridden in a derived class, sets the length of the current stream. + /// This method always throws . /// /// The desired length of the current stream in bytes. - /// The stream does not support both writing and seeking, such as if the stream is constructed from a pipe or console output. + /// Always. public override void SetLength(long value) { throw new NotSupportedException(); } - /// - /// When overridden in a derived class, reads a sequence of bytes from the current stream and advances the position within the stream by the number of bytes read. - /// - /// - /// The total number of bytes read into the buffer. This can be less than the number of bytes requested if that many bytes are not currently available, or zero if the stream is closed or end of the stream has been reached. - /// - /// An array of bytes. When this method returns, the buffer contains the specified byte array with the values between offset and (offset + count - 1) replaced by the bytes read from the current source. - /// The zero-based byte offset in buffer at which to begin storing the data read from the current stream. - /// The maximum number of bytes to be read from the current stream. - /// The sum of offset and count is larger than the buffer length. - /// Methods were called after the stream was closed. - /// The stream does not support reading. - /// is . - /// An I/O error occurs. - /// offset or count is negative. + /// public override int Read(byte[] buffer, int offset, int count) { - if (offset != 0) + lock (_sync) { - throw new NotSupportedException("Offsets with value of non-zero are not supported"); - } - - if (buffer is null) - { - throw new ArgumentNullException(nameof(buffer)); - } - - if (offset + count > buffer.Length) - { - throw new ArgumentException("The sum of offset and count is greater than the buffer length."); - } - - if (offset < 0 || count < 0) - { - throw new ArgumentOutOfRangeException(nameof(offset), "offset or count is negative."); - } - - if (BlockLastReadBuffer && count >= MaxBufferLength) - { - throw new ArgumentException(string.Format(CultureInfo.CurrentCulture, "count({0}) > mMaxBufferLength({1})", count, MaxBufferLength)); - } - -#if NET7_0_OR_GREATER - ObjectDisposedException.ThrowIf(_isDisposed, this); -#else - if (_isDisposed) - { - throw CreateObjectDisposedException(); - } -#endif // NET7_0_OR_GREATER - - if (count == 0) - { - return 0; - } - - var readLength = 0; - - lock (_buffer) - { - while (!_isDisposed && !ReadAvailable(count)) + while (_head == _tail && !_disposed) { - _ = Monitor.Wait(_buffer); + _ = Monitor.Wait(_sync); } - // return zero when the read is interrupted by a close/dispose of the stream - if (_isDisposed) - { - return 0; - } + AssertValid(); - // fill the read buffer - for (; readLength < count && _buffer.Count > 0; readLength++) - { - buffer[readLength] = _buffer.Dequeue(); - } + var bytesRead = Math.Min(count, _tail - _head); - Monitor.Pulse(_buffer); + Buffer.BlockCopy(_buffer, _head, buffer, offset, bytesRead); + + _head += bytesRead; + + AssertValid(); + + return bytesRead; } - - return readLength; } - /// - /// Returns a value indicating whether data is available. - /// - /// The count. - /// - /// if data is available; otherwise, . - /// - private bool ReadAvailable(int count) - { - var length = Length; - return (_isFlushed || length >= count) && (length >= (count + 1) || !BlockLastReadBuffer); - } - - /// - /// Writes a sequence of bytes to the current stream and advances the current position within this stream by the number of bytes written. - /// - /// An array of bytes. This method copies count bytes from buffer to the current stream. - /// The zero-based byte offset in buffer at which to begin copying bytes to the current stream. - /// The number of bytes to be written to the current stream. - /// An I/O error occurs. - /// The stream does not support writing. - /// Methods were called after the stream was closed. - /// is . - /// The sum of offset and count is greater than the buffer length. - /// offset or count is negative. + /// public override void Write(byte[] buffer, int offset, int count) { - if (buffer is null) + lock (_sync) { - throw new ArgumentNullException(nameof(buffer)); - } + ThrowIfDisposed(); - if (offset + count > buffer.Length) - { - throw new ArgumentException("The sum of offset and count is greater than the buffer length."); - } + AssertValid(); - if (offset < 0 || count < 0) - { - throw new ArgumentOutOfRangeException(nameof(offset), "offset or count is negative."); - } + // Ensure sufficient buffer space and copy the new data in. -#if NET7_0_OR_GREATER - ObjectDisposedException.ThrowIf(_isDisposed, this); -#else - if (_isDisposed) - { - throw CreateObjectDisposedException(); - } -#endif // NET7_0_OR_GREATER + if (_buffer.Length - _tail >= count) + { + // If there is enough space after _tail for the new data, + // then copy the data there. + Buffer.BlockCopy(buffer, offset, _buffer, _tail, count); + _tail += count; + } + else + { + // We can't fit the new data after _tail. - if (count == 0) + var newLength = _tail - _head + count; + + if (newLength <= _buffer.Length) + { + // If there is sufficient space at the start of the buffer, + // then move the current data to the start of the buffer. + Buffer.BlockCopy(_buffer, _head, _buffer, 0, _tail - _head); + } + else + { + // Otherwise, we're gonna need a bigger buffer. + var newBuffer = new byte[Math.Max(newLength, _buffer.Length * 2)]; + Buffer.BlockCopy(_buffer, _head, newBuffer, 0, _tail - _head); + _buffer = newBuffer; + } + + // Copy the new data into the freed-up space. + Buffer.BlockCopy(buffer, offset, _buffer, _tail - _head, count); + + _head = 0; + _tail = newLength; + } + + AssertValid(); + + Monitor.PulseAll(_sync); + } + } + + /// + protected override void Dispose(bool disposing) + { + if (!disposing) { + base.Dispose(disposing); return; } - lock (_buffer) + lock (_sync) { - // wait until the buffer isn't full - while (Length >= MaxBufferLength) + if (_disposed) { - _ = Monitor.Wait(_buffer); + return; } - _isFlushed = false; // if it were flushed before, it soon will not be. + _disposed = true; - // queue up the buffer data - for (var i = offset; i < offset + count; i++) - { - _buffer.Enqueue(buffer[i]); - } - - Monitor.Pulse(_buffer); // signal that write has occurred + Monitor.PulseAll(_sync); } - } - /// - /// Releases the unmanaged resources used by the Stream and optionally releases the managed resources. - /// - /// to release both managed and unmanaged resources; to release only unmanaged resources. - /// - /// Disposing a will interrupt blocking read and write operations. - /// - protected override void Dispose(bool disposing) - { base.Dispose(disposing); - - if (!_isDisposed) - { - lock (_buffer) - { - _isDisposed = true; - Monitor.Pulse(_buffer); - } - } } /// /// Gets a value indicating whether the current stream supports reading. /// /// - /// true if the stream supports reading; otherwise, false. + /// . /// + /// + /// It is safe to read from even after disposal. + /// public override bool CanRead { - get { return !_isDisposed; } + get { return true; } } /// /// Gets a value indicating whether the current stream supports seeking. /// /// - /// if the stream supports seeking; otherwise, . + /// . /// public override bool CanSeek { @@ -370,56 +189,60 @@ namespace Renci.SshNet.Common /// Gets a value indicating whether the current stream supports writing. /// /// - /// if the stream supports writing; otherwise, . + /// if this stream has not been disposed and the underlying channel + /// is still open, otherwise . /// + /// + /// A value of does not necessarily mean a write will succeed. It is possible + /// that the stream is disposed by another thread between a call to and the call to write. + /// public override bool CanWrite { - get { return !_isDisposed; } + get { return !_disposed; } } /// - /// Gets the length in bytes of the stream. + /// Gets the number of bytes currently available for reading. /// - /// - /// A long value representing the length of the stream in bytes. - /// - /// A class derived from Stream does not support seeking. - /// Methods were called after the stream was closed. + /// A long value representing the length of the stream in bytes. public override long Length { get { -#if NET7_0_OR_GREATER - ObjectDisposedException.ThrowIf(_isDisposed, this); -#else - if (_isDisposed) + lock (_sync) { - throw CreateObjectDisposedException(); + AssertValid(); + return _tail - _head; } -#endif // NET7_0_OR_GREATER - - return _buffer.Count; } } /// - /// Gets or sets the position within the current stream. + /// This property always returns 0, and throws + /// when calling the setter. /// /// - /// The current position within the stream. + /// 0. /// - /// The stream does not support seeking. + /// The setter is called. +#pragma warning disable SA1623 // The property's documentation should begin with 'Gets or sets' public override long Position +#pragma warning restore SA1623 // The property's documentation should begin with 'Gets or sets' { get { return 0; } set { throw new NotSupportedException(); } } -#if !NET7_0_OR_GREATER - private ObjectDisposedException CreateObjectDisposedException() + private void ThrowIfDisposed() { - return new ObjectDisposedException(GetType().FullName); +#if NET7_0_OR_GREATER + ObjectDisposedException.ThrowIf(_disposed, this); +#else + if (_disposed) + { + throw new ObjectDisposedException(GetType().FullName); + } +#endif // NET7_0_OR_GREATER } -#endif // !NET7_0_OR_GREATER } } diff --git a/src/Renci.SshNet/ScpClient.cs b/src/Renci.SshNet/ScpClient.cs index 0355a0a5..06476fd9 100644 --- a/src/Renci.SshNet/ScpClient.cs +++ b/src/Renci.SshNet/ScpClient.cs @@ -262,6 +262,7 @@ namespace Renci.SshNet using (var channel = Session.CreateChannelSession()) { channel.DataReceived += (sender, e) => input.Write(e.Data, 0, e.Data.Length); + channel.Closed += (sender, e) => input.Dispose(); channel.Open(); // Pass only the directory part of the path to the server, and use the (hidden) -d option to signal @@ -307,6 +308,7 @@ namespace Renci.SshNet using (var channel = Session.CreateChannelSession()) { channel.DataReceived += (sender, e) => input.Write(e.Data, 0, e.Data.Length); + channel.Closed += (sender, e) => input.Dispose(); channel.Open(); // Pass only the directory part of the path to the server, and use the (hidden) -d option to signal @@ -364,6 +366,7 @@ namespace Renci.SshNet using (var channel = Session.CreateChannelSession()) { channel.DataReceived += (sender, e) => input.Write(e.Data, 0, e.Data.Length); + channel.Closed += (sender, e) => input.Dispose(); channel.Open(); // start copy with the following options: @@ -413,6 +416,7 @@ namespace Renci.SshNet using (var channel = Session.CreateChannelSession()) { channel.DataReceived += (sender, e) => input.Write(e.Data, 0, e.Data.Length); + channel.Closed += (sender, e) => input.Dispose(); channel.Open(); // Send channel command request @@ -459,6 +463,7 @@ namespace Renci.SshNet using (var channel = Session.CreateChannelSession()) { channel.DataReceived += (sender, e) => input.Write(e.Data, 0, e.Data.Length); + channel.Closed += (sender, e) => input.Dispose(); channel.Open(); // Send channel command request @@ -505,6 +510,7 @@ namespace Renci.SshNet using (var channel = Session.CreateChannelSession()) { channel.DataReceived += (sender, e) => input.Write(e.Data, 0, e.Data.Length); + channel.Closed += (sender, e) => input.Dispose(); channel.Open(); // Send channel command request diff --git a/src/Renci.SshNet/ShellStream.cs b/src/Renci.SshNet/ShellStream.cs index 193badd3..d903ee7e 100644 --- a/src/Renci.SshNet/ShellStream.cs +++ b/src/Renci.SshNet/ShellStream.cs @@ -182,9 +182,7 @@ namespace Renci.SshNet get { return !_disposed; } } - /// - /// This method does nothing. - /// + /// public override void Flush() { ThrowIfDisposed(); diff --git a/src/Renci.SshNet/SshCommand.cs b/src/Renci.SshNet/SshCommand.cs index bbdd349c..4bd3abd9 100644 --- a/src/Renci.SshNet/SshCommand.cs +++ b/src/Renci.SshNet/SshCommand.cs @@ -28,8 +28,8 @@ namespace Renci.SshNet private EventWaitHandle _sessionErrorOccuredWaitHandle; private EventWaitHandle _commandCancelledWaitHandle; private Exception _exception; - private StringBuilder _result; - private StringBuilder _error; + private string _result; + private string _error; private bool _hasError; private bool _isDisposed; private bool _isCancelled; @@ -109,21 +109,22 @@ namespace Renci.SshNet { get { - _result ??= new StringBuilder(); - - if (OutputStream != null && OutputStream.Length > 0) + if (_result is not null) { - using (var sr = new StreamReader(OutputStream, - _encoding, - detectEncodingFromByteOrderMarks: true, - bufferSize: 1024, - leaveOpen: true)) - { - _ = _result.Append(sr.ReadToEnd()); - } + return _result; } - return _result.ToString(); + if (OutputStream is null) + { + return string.Empty; + } + + using (var sr = new StreamReader(OutputStream, + _encoding, + detectEncodingFromByteOrderMarks: true)) + { + return _result = sr.ReadToEnd(); + } } } @@ -134,26 +135,22 @@ namespace Renci.SshNet { get { - if (_hasError) + if (_error is not null) { - _error ??= new StringBuilder(); - - if (ExtendedOutputStream != null && ExtendedOutputStream.Length > 0) - { - using (var sr = new StreamReader(ExtendedOutputStream, - _encoding, - detectEncodingFromByteOrderMarks: true, - bufferSize: 1024, - leaveOpen: true)) - { - _ = _error.Append(sr.ReadToEnd()); - } - } - - return _error.ToString(); + return _error; } - return string.Empty; + if (ExtendedOutputStream is null || !_hasError) + { + return string.Empty; + } + + using (var sr = new StreamReader(ExtendedOutputStream, + _encoding, + detectEncodingFromByteOrderMarks: true)) + { + return _error = sr.ReadToEnd(); + } } } @@ -265,19 +262,8 @@ namespace Renci.SshNet throw new ArgumentException("CommandText property is empty."); } - var outputStream = OutputStream; - if (outputStream is not null) - { - outputStream.Dispose(); - OutputStream = null; - } - - var extendedOutputStream = ExtendedOutputStream; - if (extendedOutputStream is not null) - { - extendedOutputStream.Dispose(); - ExtendedOutputStream = null; - } + OutputStream?.Dispose(); + ExtendedOutputStream?.Dispose(); // Initialize output streams OutputStream = new PipeStream(); @@ -285,6 +271,7 @@ namespace Renci.SshNet _result = null; _error = null; + _hasError = false; _callback = callback; _channel = CreateChannel(); @@ -341,13 +328,21 @@ namespace Renci.SshNet _inputStream?.Close(); - // wait for operation to complete (or time out) - WaitOnHandle(_asyncResult.AsyncWaitHandle); + try + { + // wait for operation to complete (or time out) + WaitOnHandle(_asyncResult.AsyncWaitHandle); + } + finally + { + UnsubscribeFromEventsAndDisposeChannel(_channel); + _channel = null; - UnsubscribeFromEventsAndDisposeChannel(_channel); - _channel = null; + OutputStream?.Dispose(); + ExtendedOutputStream?.Dispose(); - commandAsyncResult.EndCalled = true; + commandAsyncResult.EndCalled = true; + } if (!_isCancelled) { @@ -437,8 +432,8 @@ namespace Renci.SshNet private void SetAsyncComplete() { - OutputStream?.Flush(); - ExtendedOutputStream?.Flush(); + OutputStream?.Dispose(); + ExtendedOutputStream?.Dispose(); _asyncResult.IsCompleted = true; @@ -480,11 +475,7 @@ namespace Renci.SshNet private void Channel_ExtendedDataReceived(object sender, ChannelExtendedDataEventArgs e) { - if (ExtendedOutputStream != null) - { - ExtendedOutputStream.Write(e.Data, 0, e.Data.Length); - ExtendedOutputStream.Flush(); - } + ExtendedOutputStream?.Write(e.Data, 0, e.Data.Length); if (e.DataTypeCode == 1) { @@ -494,11 +485,7 @@ namespace Renci.SshNet private void Channel_DataReceived(object sender, ChannelDataEventArgs e) { - if (OutputStream != null) - { - OutputStream.Write(e.Data, 0, e.Data.Length); - OutputStream.Flush(); - } + OutputStream?.Write(e.Data, 0, e.Data.Length); if (_asyncResult != null) { diff --git a/test/Renci.SshNet.IntegrationBenchmarks/SshClientBenchmark.cs b/test/Renci.SshNet.IntegrationBenchmarks/SshClientBenchmark.cs index 68c8cf03..47c169e7 100644 --- a/test/Renci.SshNet.IntegrationBenchmarks/SshClientBenchmark.cs +++ b/test/Renci.SshNet.IntegrationBenchmarks/SshClientBenchmark.cs @@ -72,6 +72,18 @@ namespace Renci.SshNet.IntegrationBenchmarks return _sshClient!.RunCommand("echo $'test !@#$%^&*()_+{}:,./<>[];\\|'").Result; } + [Benchmark] + public string RunBigCommand() + { + using var command = _sshClient!.CreateCommand("head -c 10000000 /dev/urandom | base64"); // 10MB of data please + + var asyncResult = command.BeginExecute(); + + command.OutputStream.CopyTo(Stream.Null); + + return command.EndExecute(asyncResult); + } + [Benchmark] public string ShellStreamReadLine() { diff --git a/test/Renci.SshNet.IntegrationTests/SshClientTests.cs b/test/Renci.SshNet.IntegrationTests/SshClientTests.cs index 2cfeb53c..4a0d4df6 100644 --- a/test/Renci.SshNet.IntegrationTests/SshClientTests.cs +++ b/test/Renci.SshNet.IntegrationTests/SshClientTests.cs @@ -22,6 +22,28 @@ namespace Renci.SshNet.IntegrationTests Assert.AreEqual("test !@#$%^&*()_+{}:,./<>[];\\|\n", response.Result); } + [TestMethod] + public void Test_BigCommand() + { + using var command = _sshClient.CreateCommand("head -c 10000000 /dev/urandom | base64"); // 10MB of data please + + var asyncResult = command.BeginExecute(); + + long totalBytesRead = 0; + int bytesRead; + byte[] buffer = new byte[4096]; + + while ((bytesRead = command.OutputStream.Read(buffer, 0, buffer.Length)) != 0) + { + totalBytesRead += bytesRead; + } + + var result = command.EndExecute(asyncResult); + + Assert.AreEqual(13_508_775, totalBytesRead); + Assert.AreEqual(0, result.Length); + } + [TestMethod] public void Send_InputStream_to_Command() { diff --git a/test/Renci.SshNet.IntegrationTests/SshTests.cs b/test/Renci.SshNet.IntegrationTests/SshTests.cs index ea11818e..f65b8b5f 100644 --- a/test/Renci.SshNet.IntegrationTests/SshTests.cs +++ b/test/Renci.SshNet.IntegrationTests/SshTests.cs @@ -172,7 +172,7 @@ namespace Renci.SshNet.IntegrationTests } [TestMethod] - public void Ssh_Command_IntermittendOutput_EndExecute() + public void Ssh_Command_IntermittentOutput_EndExecute() { const string remoteFile = "/home/sshnet/test.sh"; @@ -229,16 +229,8 @@ namespace Renci.SshNet.IntegrationTests } } - /// - /// Ignored for now, because: - /// * OutputStream.Read(...) does not block when no data is available - /// * SshCommand.(Begin)Execute consumes *OutputStream*, advancing its position. - /// - /// https://github.com/sshnet/SSH.NET/issues/650 - /// [TestMethod] - [Ignore] - public void Ssh_Command_IntermittendOutput_OutputStream() + public void Ssh_Command_IntermittentOutput_OutputStream() { const string remoteFile = "/home/sshnet/test.sh"; @@ -297,8 +289,16 @@ namespace Renci.SshNet.IntegrationTests var actualResult = command.EndExecute(asyncResult); - Assert.AreEqual(expectedResult, actualResult); - Assert.AreEqual(expectedResult, command.Result); + // command.Result (also returned from EndExecute) consumes OutputStream, + // which we've already read from, so Result will be empty. + // TODO consider the suggested changes in https://github.com/sshnet/SSH.NET/issues/650 + + //Assert.AreEqual(expectedResult, actualResult); + //Assert.AreEqual(expectedResult, command.Result); + + // For now just assert the current behaviour. + Assert.AreEqual(0, actualResult.Length); + Assert.AreEqual(0, command.Result.Length); } } finally diff --git a/test/Renci.SshNet.Tests/.editorconfig b/test/Renci.SshNet.Tests/.editorconfig index 496a8321..ae7f4921 100644 --- a/test/Renci.SshNet.Tests/.editorconfig +++ b/test/Renci.SshNet.Tests/.editorconfig @@ -305,6 +305,10 @@ dotnet_diagnostic.MA0110.severity = silent # https://github.com/meziantou/Meziantou.Analyzer/blob/main/docs/Rules/MA0026.md dotnet_diagnostic.MA0026.severity = silent +# MA0042: Do not use blocking calls in an async method +# https://github.com/meziantou/Meziantou.Analyzer/blob/main/docs/Rules/MA0042.md +dotnet_diagnostic.MA0042.severity = silent + #### .NET Compiler Platform analysers rules #### # CA1031: Do not catch general exception types @@ -397,4 +401,8 @@ dotnet_diagnostic.IDE0032.severity = silent # CA1812: Avoid uninstantiated internal classes # https://learn.microsoft.com/en-us/dotnet/fundamentals/code-analysis/quality-rules/ca1812 -dotnet_diagnostic.CA1812.severity = silent \ No newline at end of file +dotnet_diagnostic.CA1812.severity = silent + +# CA1849: Call async methods when in an async method +# https://learn.microsoft.com/en-us/dotnet/fundamentals/code-analysis/quality-rules/CA1849 +dotnet_diagnostic.CA1849.severity = silent diff --git a/test/Renci.SshNet.Tests/Classes/Common/PipeStreamTest.cs b/test/Renci.SshNet.Tests/Classes/Common/PipeStreamTest.cs index 56738dff..30685e98 100644 --- a/test/Renci.SshNet.Tests/Classes/Common/PipeStreamTest.cs +++ b/test/Renci.SshNet.Tests/Classes/Common/PipeStreamTest.cs @@ -1,6 +1,6 @@ using System; using System.IO; -using System.Threading; +using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; @@ -23,15 +23,15 @@ namespace Renci.SshNet.Tests.Classes.Common using (var stream = new PipeStream()) { - stream.Write(testBuffer, 0, testBuffer.Length); + stream.Write(testBuffer, 0, 512); - Assert.AreEqual(stream.Length, testBuffer.Length); + Assert.AreEqual(512, stream.Length); - _ = stream.Read(outputBuffer, 0, outputBuffer.Length); + Assert.AreEqual(128, stream.Read(outputBuffer, 64, 128)); - Assert.AreEqual(stream.Length, 0); + Assert.AreEqual(384, stream.Length); - Assert.IsTrue(testBuffer.IsEqualTo(outputBuffer)); + CollectionAssert.AreEqual(new byte[64].Concat(testBuffer.Take(128)).Concat(new byte[832]), outputBuffer); } } @@ -45,19 +45,17 @@ namespace Renci.SshNet.Tests.Classes.Common using (var stream = new PipeStream()) { stream.Write(testBuffer, 0, testBuffer.Length); - Assert.AreEqual(stream.Length, testBuffer.Length); - _ = stream.ReadByte(); - Assert.AreEqual(stream.Length, testBuffer.Length - 1); - _ = stream.ReadByte(); - Assert.AreEqual(stream.Length, testBuffer.Length - 2); + Assert.AreEqual(1024, stream.Length); + Assert.AreEqual(testBuffer[0], stream.ReadByte()); + Assert.AreEqual(1023, stream.Length); + Assert.AreEqual(testBuffer[1], stream.ReadByte()); + Assert.AreEqual(1022, stream.Length); } } [TestMethod] public void Read() { - const int sleepTime = 100; - var target = new PipeStream(); target.WriteByte(0x0a); target.WriteByte(0x0d); @@ -69,20 +67,88 @@ namespace Renci.SshNet.Tests.Classes.Common Assert.AreEqual(0x0a, readBuffer[0]); Assert.AreEqual(0x0d, readBuffer[1]); - var writeToStreamThread = new Thread( - () => - { - Thread.Sleep(sleepTime); - var writeBuffer = new byte[] { 0x05, 0x03 }; - target.Write(writeBuffer, 0, writeBuffer.Length); - }); - writeToStreamThread.Start(); + var writeBuffer = new byte[] { 0x05, 0x03 }; + target.Write(writeBuffer, 0, writeBuffer.Length); - readBuffer = new byte[2]; + readBuffer = new byte[4]; bytesRead = target.Read(readBuffer, 0, readBuffer.Length); - Assert.AreEqual(2, bytesRead); + Assert.AreEqual(3, bytesRead); Assert.AreEqual(0x09, readBuffer[0]); Assert.AreEqual(0x05, readBuffer[1]); + Assert.AreEqual(0x03, readBuffer[2]); + Assert.AreEqual(0x00, readBuffer[3]); + } + + [TestMethod] + public async Task Read_NonEmptyArray_OnlyReturnsZeroAfterDispose() + { + // When there is no data available, a read should block, + // but then unblock (and return 0) after disposal. + + var pipeStream = new PipeStream(); + + Task readTask = pipeStream.ReadAsync(new byte[16], 0, 16); + + await Task.Delay(50); + + Assert.IsFalse(readTask.IsCompleted); + + pipeStream.Dispose(); + + Assert.AreEqual(0, await readTask); + } + + [TestMethod] + public async Task Read_EmptyArray_OnlyReturnsZeroAfterDispose() + { + // Similarly, zero byte reads should still block until after disposal. + + var pipeStream = new PipeStream(); + + Task readTask = pipeStream.ReadAsync(Array.Empty(), 0, 0); + + await Task.Delay(50); + + Assert.IsFalse(readTask.IsCompleted); + + pipeStream.Dispose(); + + Assert.AreEqual(0, await readTask); + } + + [TestMethod] + public async Task Read_EmptyArray_OnlyReturnsZeroWhenDataAvailable() + { + // And zero byte reads should block but then return 0 once data + // is available. + + var pipeStream = new PipeStream(); + + Task readTask = pipeStream.ReadAsync(Array.Empty(), 0, 0); + + await Task.Delay(50); + + Assert.IsFalse(readTask.IsCompleted); + + pipeStream.Write(new byte[] { 1, 2, 3, 4 }, 0, 4); + + Assert.AreEqual(0, await readTask); + } + + [TestMethod] + public void Read_AfterDispose_StillWorks() + { + var pipeStream = new PipeStream(); + + pipeStream.Write(new byte[] { 1, 2, 3, 4 }, 0, 4); + + pipeStream.Dispose(); +#pragma warning disable S3966 // Objects should not be disposed more than once + pipeStream.Dispose(); // Check that multiple Dispose is OK. +#pragma warning restore S3966 // Objects should not be disposed more than once + + Assert.AreEqual(4, pipeStream.Read(new byte[5], 0, 5)); + Assert.AreEqual(0, pipeStream.Read(new byte[5], 0, 5)); } [TestMethod] @@ -151,7 +217,7 @@ namespace Renci.SshNet.Tests.Classes.Common [TestMethod] public void CanSeekTest() { - var target = new PipeStream(); // TODO: Initialize to an appropriate value + var target = new PipeStream(); Assert.IsFalse(target.CanSeek); } @@ -177,18 +243,6 @@ namespace Renci.SshNet.Tests.Classes.Common Assert.AreEqual(0L, target.Length); } - /// - ///A test for MaxBufferLength - /// - [TestMethod] - public void MaxBufferLengthTest() - { - var target = new PipeStream(); - Assert.AreEqual(200 * 1024 * 1024, target.MaxBufferLength); - target.MaxBufferLength = 0L; - Assert.AreEqual(0L, target.MaxBufferLength); - } - [TestMethod] public void Position_GetterAlwaysReturnsZero() { diff --git a/test/Renci.SshNet.Tests/Classes/Common/PipeStream_Close_BlockingRead.cs b/test/Renci.SshNet.Tests/Classes/Common/PipeStream_Close_BlockingRead.cs deleted file mode 100644 index 41a5f07b..00000000 --- a/test/Renci.SshNet.Tests/Classes/Common/PipeStream_Close_BlockingRead.cs +++ /dev/null @@ -1,54 +0,0 @@ -using System.Threading; - -using Microsoft.VisualStudio.TestTools.UnitTesting; - -using Renci.SshNet.Common; -using Renci.SshNet.Tests.Common; - -namespace Renci.SshNet.Tests.Classes.Common -{ - [TestClass] - public class PipeStream_Close_BlockingRead : TripleATestBase - { - private PipeStream _pipeStream; - private int _bytesRead; - private Thread _readThread; - - protected override void Arrange() - { - _pipeStream = new PipeStream(); - - _pipeStream.WriteByte(10); - _pipeStream.WriteByte(13); - _pipeStream.WriteByte(25); - - _bytesRead = 123; - - _readThread = new Thread(() => _bytesRead = _pipeStream.Read(new byte[4], 0, 4)); - _readThread.Start(); - - // ensure we've started reading - Assert.IsFalse(_readThread.Join(50)); - } - - protected override void Act() - { - _pipeStream.Close(); - - // give async read time to complete - _ = _readThread.Join(100); - } - - [TestMethod] - public void BlockingReadShouldHaveBeenInterrupted() - { - Assert.AreEqual(ThreadState.Stopped, _readThread.ThreadState); - } - - [TestMethod] - public void ReadShouldHaveReturnedZero() - { - Assert.AreEqual(0, _bytesRead); - } - } -} diff --git a/test/Renci.SshNet.Tests/Classes/Common/PipeStream_Close_BlockingWrite.cs b/test/Renci.SshNet.Tests/Classes/Common/PipeStream_Close_BlockingWrite.cs deleted file mode 100644 index b0e3c3c2..00000000 --- a/test/Renci.SshNet.Tests/Classes/Common/PipeStream_Close_BlockingWrite.cs +++ /dev/null @@ -1,66 +0,0 @@ -using System; -using System.Threading; - -using Microsoft.VisualStudio.TestTools.UnitTesting; - -using Renci.SshNet.Common; -using Renci.SshNet.Tests.Common; - -namespace Renci.SshNet.Tests.Classes.Common -{ - [TestClass] - public class PipeStream_Close_BlockingWrite : TripleATestBase - { - private PipeStream _pipeStream; - private Exception _writeException; - private Thread _writehread; - - protected override void Arrange() - { - _pipeStream = new PipeStream { MaxBufferLength = 3 }; - - _writehread = new Thread(() => - { - _pipeStream.WriteByte(10); - _pipeStream.WriteByte(13); - _pipeStream.WriteByte(25); - - // attempting to write more bytes than the max. buffer length should block - // until bytes are read or the stream is closed - try - { - _pipeStream.WriteByte(35); - } - catch (Exception ex) - { - _writeException = ex; - } - }); - _writehread.Start(); - - // ensure we've started writing - Assert.IsFalse(_writehread.Join(50)); - } - - protected override void Act() - { - _pipeStream.Close(); - - // give write time to complete - _ = _writehread.Join(100); - } - - [TestMethod] - public void BlockingWriteShouldHaveBeenInterrupted() - { - Assert.AreEqual(ThreadState.Stopped, _writehread.ThreadState); - } - - [TestMethod] - public void WriteShouldHaveThrownObjectDisposedException() - { - Assert.IsNotNull(_writeException); - Assert.AreEqual(typeof(ObjectDisposedException), _writeException.GetType()); - } - } -} diff --git a/test/Renci.SshNet.Tests/Classes/Common/PipeStream_Flush_BytesRemainingAfterRead.cs b/test/Renci.SshNet.Tests/Classes/Common/PipeStream_Flush_BytesRemainingAfterRead.cs index d765e2fc..60ea1362 100644 --- a/test/Renci.SshNet.Tests/Classes/Common/PipeStream_Flush_BytesRemainingAfterRead.cs +++ b/test/Renci.SshNet.Tests/Classes/Common/PipeStream_Flush_BytesRemainingAfterRead.cs @@ -90,33 +90,5 @@ namespace Renci.SshNet.Tests.Classes.Common Assert.AreEqual(0, buffer[2]); Assert.AreEqual(0, buffer[3]); } -#if NETFRAMEWORK - [TestMethod] - public void WriteCausesSubsequentReadToBlockUntilRequestedNumberOfBytesAreAvailable() - { - _pipeStream.WriteByte(32); - - var buffer = new byte[4]; - int bytesRead = int.MaxValue; - - Thread readThread = new Thread(() => - { - bytesRead = _pipeStream.Read(buffer, 0, buffer.Length); - }); - readThread.Start(); - - Assert.IsFalse(readThread.Join(500)); - - // Thread Abort method is obsolete: https://learn.microsoft.com/en-us/dotnet/core/compatibility/core-libraries/5.0/thread-abort-obsolete - readThread.Abort(); - - - Assert.AreEqual(int.MaxValue, bytesRead); - Assert.AreEqual(0, buffer[0]); - Assert.AreEqual(0, buffer[1]); - Assert.AreEqual(0, buffer[2]); - Assert.AreEqual(0, buffer[3]); - } -#endif } } diff --git a/test/Renci.SshNet.Tests/Classes/Common/PipeStream_Flush_NoBytesRemainingAfterRead.cs b/test/Renci.SshNet.Tests/Classes/Common/PipeStream_Flush_NoBytesRemainingAfterRead.cs deleted file mode 100644 index 0926f2e6..00000000 --- a/test/Renci.SshNet.Tests/Classes/Common/PipeStream_Flush_NoBytesRemainingAfterRead.cs +++ /dev/null @@ -1,63 +0,0 @@ -using System.Threading; - -using Microsoft.VisualStudio.TestTools.UnitTesting; - -using Renci.SshNet.Common; -using Renci.SshNet.Tests.Common; - -namespace Renci.SshNet.Tests.Classes.Common -{ - [TestClass] - public class PipeStream_Flush_NoBytesRemainingAfterRead : TripleATestBase - { - private PipeStream _pipeStream; - private byte[] _readBuffer; - private int _bytesRead; - private Thread _readThread; - - protected override void Arrange() - { - _pipeStream = new PipeStream(); - _pipeStream.WriteByte(10); - _pipeStream.WriteByte(13); - - _bytesRead = 0; - _readBuffer = new byte[4]; - - _readThread = new Thread(() => _bytesRead = _pipeStream.Read(_readBuffer, 0, _readBuffer.Length)); - _readThread.Start(); - - // ensure we've started reading - Assert.IsFalse(_readThread.Join(50)); - } - - protected override void Act() - { - _pipeStream.Flush(); - - // give async read time to complete - _ = _readThread.Join(100); - } - - [TestMethod] - public void AsyncReadShouldHaveFinished() - { - Assert.AreEqual(ThreadState.Stopped, _readThread.ThreadState); - } - - [TestMethod] - public void ReadShouldReturnNumberOfBytesAvailableThatAreWrittenToBuffer() - { - Assert.AreEqual(2, _bytesRead); - } - - [TestMethod] - public void BytesAvailableInStreamShouldHaveBeenWrittenToBuffer() - { - Assert.AreEqual(10, _readBuffer[0]); - Assert.AreEqual(13, _readBuffer[1]); - Assert.AreEqual(0, _readBuffer[2]); - Assert.AreEqual(0, _readBuffer[3]); - } - } -} diff --git a/test/Renci.SshNet.Tests/Classes/PipeStreamTest_Dispose.cs b/test/Renci.SshNet.Tests/Classes/PipeStreamTest_Dispose.cs index 8ccf52d0..77fc2b72 100644 --- a/test/Renci.SshNet.Tests/Classes/PipeStreamTest_Dispose.cs +++ b/test/Renci.SshNet.Tests/Classes/PipeStreamTest_Dispose.cs @@ -34,47 +34,19 @@ namespace Renci.SshNet.Tests.Classes [TestMethod] public void CanRead_ShouldReturnTrue() { - Assert.IsFalse(_pipeStream.CanRead); + Assert.IsTrue(_pipeStream.CanRead); } [TestMethod] - public void Flush_ShouldThrowObjectDisposedException() + public void Flush_ShouldNotThrow() { - try - { - _pipeStream.Flush(); - Assert.Fail(); - } - catch (ObjectDisposedException) - { - } + _pipeStream.Flush(); } [TestMethod] - public void MaxBufferLength_Getter_ShouldReturnTwoHundredMegabyte() + public void Length_ShouldNotThrow() { - Assert.AreEqual(200 * 1024 * 1024, _pipeStream.MaxBufferLength); - } - - [TestMethod] - public void MaxBufferLength_Setter_ShouldModifyMaxBufferLength() - { - var newValue = new Random().Next(1, int.MaxValue); - _pipeStream.MaxBufferLength = newValue; - Assert.AreEqual(newValue, _pipeStream.MaxBufferLength); - } - - [TestMethod] - public void Length_ShouldThrowObjectDisposedException() - { - try - { - var value = _pipeStream.Length; - Assert.Fail("" + value); - } - catch (ObjectDisposedException) - { - } + _ = _pipeStream.Length; } [TestMethod] @@ -97,33 +69,15 @@ namespace Renci.SshNet.Tests.Classes } [TestMethod] - public void Read_ByteArrayAndOffsetAndCount_ShouldThrowObjectDisposedException() + public void Read_ByteArrayAndOffsetAndCount_ShouldNotThrow() { - var buffer = new byte[0]; - const int offset = 0; - const int count = 0; - - try - { - _pipeStream.Read(buffer, offset, count); - Assert.Fail(); - } - catch (ObjectDisposedException) - { - } + Assert.AreEqual(0, _pipeStream.Read(new byte[1], 0, 1)); } [TestMethod] - public void ReadByte_ShouldThrowObjectDisposedException() + public void ReadByte_ShouldNotThrow() { - try - { - _pipeStream.ReadByte(); - Assert.Fail(); - } - catch (ObjectDisposedException) - { - } + Assert.AreEqual(-1, _pipeStream.ReadByte()); } [TestMethod] diff --git a/test/Renci.SshNet.Tests/Classes/SshCommandTest_Dispose.cs b/test/Renci.SshNet.Tests/Classes/SshCommandTest_Dispose.cs index 61f67130..21542f83 100644 --- a/test/Renci.SshNet.Tests/Classes/SshCommandTest_Dispose.cs +++ b/test/Renci.SshNet.Tests/Classes/SshCommandTest_Dispose.cs @@ -73,14 +73,7 @@ namespace Renci.SshNet.Tests.Classes [TestMethod] public void OutputStreamShouldHaveBeenDisposed() { - try - { - _outputStream.ReadByte(); - Assert.Fail(); - } - catch (ObjectDisposedException) - { - } + Assert.AreEqual(-1, _outputStream.ReadByte()); } [TestMethod] @@ -92,14 +85,7 @@ namespace Renci.SshNet.Tests.Classes [TestMethod] public void ExtendedOutputStreamShouldHaveBeenDisposed() { - try - { - _extendedOutputStream.ReadByte(); - Assert.Fail(); - } - catch (ObjectDisposedException) - { - } + Assert.AreEqual(-1, _extendedOutputStream.ReadByte()); } [TestMethod] diff --git a/test/Renci.SshNet.Tests/Classes/SshCommandTest_EndExecute_ChannelOpen.cs b/test/Renci.SshNet.Tests/Classes/SshCommandTest_EndExecute_ChannelOpen.cs index 417219ae..29343968 100644 --- a/test/Renci.SshNet.Tests/Classes/SshCommandTest_EndExecute_ChannelOpen.cs +++ b/test/Renci.SshNet.Tests/Classes/SshCommandTest_EndExecute_ChannelOpen.cs @@ -56,8 +56,7 @@ namespace Renci.SshNet.Tests.Classes _sessionMock.InSequence(seq).Setup(p => p.CreateChannelSession()).Returns(_channelSessionMock.Object); _channelSessionMock.InSequence(seq).Setup(p => p.Open()); _channelSessionMock.InSequence(seq).Setup(p => p.SendExecRequest(_commandText)) - .Returns(true) - .Raises(c => c.Closed += null, new ChannelEventArgs(5)); + .Returns(true); _channelSessionMock.InSequence(seq).Setup(p => p.Dispose()); _sshCommand = new SshCommand(_sessionMock.Object, _commandText, _encoding); @@ -73,6 +72,8 @@ namespace Renci.SshNet.Tests.Classes new ChannelExtendedDataEventArgs(0, _encoding.GetBytes(_extendedDataB), 0)); _channelSessionMock.Raise(c => c.RequestReceived += null, new ChannelRequestEventArgs(new ExitStatusRequestInfo((uint)_expectedExitStatus))); + _channelSessionMock.Raise(c => c.Closed += null, + new ChannelEventArgs(5)); } private void Act()