From 612bc3c2e1ef51826f545b540a3aac9815011813 Mon Sep 17 00:00:00 2001 From: Rob Hague Date: Thu, 22 Feb 2024 21:36:57 +0100 Subject: [PATCH] Fix a few issues with PipeStream 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. --- src/Renci.SshNet/Common/PipeStream.cs | 413 +++++------------- src/Renci.SshNet/ScpClient.cs | 6 + src/Renci.SshNet/ShellStream.cs | 4 +- src/Renci.SshNet/SshCommand.cs | 107 ++--- .../SshClientBenchmark.cs | 12 + .../SshClientTests.cs | 22 + .../Renci.SshNet.IntegrationTests/SshTests.cs | 24 +- test/Renci.SshNet.Tests/.editorconfig | 10 +- .../Classes/Common/PipeStreamTest.cs | 128 ++++-- .../Common/PipeStream_Close_BlockingRead.cs | 52 --- .../Common/PipeStream_Close_BlockingWrite.cs | 64 --- ...ipeStream_Flush_BytesRemainingAfterRead.cs | 28 -- ...eStream_Flush_NoBytesRemainingAfterRead.cs | 61 --- .../Classes/PipeStreamTest_Dispose.cs | 64 +-- .../Classes/SshCommandTest_Dispose.cs | 18 +- .../SshCommandTest_EndExecute_ChannelOpen.cs | 5 +- 16 files changed, 332 insertions(+), 686 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 da7b89cba..db211bf19 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(); - - /// - /// Indicates that the input stream has been flushed and that - /// all remaining data should be written to the output stream. - /// - private bool _isFlushed; - - /// - /// 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; + private readonly object _sync = new object(); - /// - /// 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; + 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; - /// - /// 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"); - } + while (_head == _tail && !_disposed) + { + _ = Monitor.Wait(_sync); + } - if (buffer is null) - { - throw new ArgumentNullException(nameof(buffer)); - } + AssertValid(); - if (offset + count > buffer.Length) - { - throw new ArgumentException("The sum of offset and count is greater than the buffer length."); - } + var bytesRead = Math.Min(count, _tail - _head); - if (offset < 0 || count < 0) - { - throw new ArgumentOutOfRangeException(nameof(offset), "offset or count is negative."); - } + Buffer.BlockCopy(_buffer, _head, buffer, offset, bytesRead); - if (BlockLastReadBuffer && count >= MaxBufferLength) - { - throw new ArgumentException(string.Format(CultureInfo.CurrentCulture, "count({0}) > mMaxBufferLength({1})", count, MaxBufferLength)); - } + _head += bytesRead; -#if NET7_0_OR_GREATER - ObjectDisposedException.ThrowIf(_isDisposed, this); -#else - if (_isDisposed) - { - throw CreateObjectDisposedException(); + AssertValid(); + + return bytesRead; } -#endif // NET7_0_OR_GREATER + } - if (count == 0) + /// + public override void Write(byte[] buffer, int offset, int count) + { + lock (_sync) { - return 0; - } + ThrowIfDisposed(); - var readLength = 0; + AssertValid(); - lock (_buffer) - { - while (!_isDisposed && !ReadAvailable(count)) - { - _ = Monitor.Wait(_buffer); - } + // Ensure sufficient buffer space and copy the new data in. - // return zero when the read is interrupted by a close/dispose of the stream - if (_isDisposed) + if (_buffer.Length - _tail >= count) { - return 0; + // 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; } - - // fill the read buffer - for (; readLength < count && _buffer.Count > 0; readLength++) + else { - buffer[readLength] = _buffer.Dequeue(); - } + // We can't fit the new data after _tail. - Monitor.Pulse(_buffer); - } - - return readLength; - } + var newLength = _tail - _head + count; - /// - /// 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); - } + 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; + } - /// - /// 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) - { - throw new ArgumentNullException(nameof(buffer)); - } + // Copy the new data into the freed-up space. + Buffer.BlockCopy(buffer, offset, _buffer, _tail - _head, count); - if (offset + count > buffer.Length) - { - throw new ArgumentException("The sum of offset and count is greater than the buffer length."); - } + _head = 0; + _tail = newLength; + } - if (offset < 0 || count < 0) - { - throw new ArgumentOutOfRangeException(nameof(offset), "offset or count is negative."); - } + AssertValid(); -#if NET7_0_OR_GREATER - ObjectDisposedException.ThrowIf(_isDisposed, this); -#else - if (_isDisposed) - { - throw CreateObjectDisposedException(); + Monitor.PulseAll(_sync); } -#endif // NET7_0_OR_GREATER + } - if (count == 0) + /// + 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 @@ public override bool CanSeek /// 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 7aa57c62b..6a211f442 100644 --- a/src/Renci.SshNet/ScpClient.cs +++ b/src/Renci.SshNet/ScpClient.cs @@ -234,6 +234,7 @@ public void Upload(Stream source, string path) 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 @@ -273,6 +274,7 @@ public void Upload(FileInfo fileInfo, string path) 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 @@ -324,6 +326,7 @@ public void Upload(DirectoryInfo directoryInfo, string path) 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: @@ -367,6 +370,7 @@ public void Download(string filename, FileInfo fileInfo) 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 @@ -407,6 +411,7 @@ public void Download(string directoryName, DirectoryInfo directoryInfo) 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 @@ -447,6 +452,7 @@ public void Download(string filename, Stream destination) 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 193badd31..d903ee7e7 100644 --- a/src/Renci.SshNet/ShellStream.cs +++ b/src/Renci.SshNet/ShellStream.cs @@ -182,9 +182,7 @@ public override bool CanWrite 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 df61cda7e..09fca2380 100644 --- a/src/Renci.SshNet/SshCommand.cs +++ b/src/Renci.SshNet/SshCommand.cs @@ -28,8 +28,8 @@ public class SshCommand : IDisposable 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 @@ public string Result { get { - _result ??= new StringBuilder(); + if (_result is not null) + { + return _result; + } - if (OutputStream != null && OutputStream.Length > 0) + if (OutputStream is null) { - using (var sr = new StreamReader(OutputStream, - _encoding, - detectEncodingFromByteOrderMarks: true, - bufferSize: 1024, - leaveOpen: true)) - { - _ = _result.Append(sr.ReadToEnd()); - } + return string.Empty; } - return _result.ToString(); + using (var sr = new StreamReader(OutputStream, + _encoding, + detectEncodingFromByteOrderMarks: true)) + { + return _result = sr.ReadToEnd(); + } } } @@ -134,26 +135,22 @@ public string Error { get { - if (_hasError) + if (_error is not null) + { + return _error; + } + + if (ExtendedOutputStream is null || !_hasError) { - _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 string.Empty; } - return string.Empty; + using (var sr = new StreamReader(ExtendedOutputStream, + _encoding, + detectEncodingFromByteOrderMarks: true)) + { + return _error = sr.ReadToEnd(); + } } } @@ -265,19 +262,8 @@ public IAsyncResult BeginExecute(AsyncCallback callback, object state) 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 @@ public IAsyncResult BeginExecute(AsyncCallback callback, object state) _result = null; _error = null; + _hasError = false; _callback = callback; _channel = CreateChannel(); @@ -341,13 +328,21 @@ public string EndExecute(IAsyncResult asyncResult) _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 @@ private void Session_ErrorOccured(object sender, ExceptionEventArgs e) private void SetAsyncComplete() { - OutputStream?.Flush(); - ExtendedOutputStream?.Flush(); + OutputStream?.Dispose(); + ExtendedOutputStream?.Dispose(); _asyncResult.IsCompleted = true; @@ -480,11 +475,7 @@ private void Channel_RequestReceived(object sender, ChannelRequestEventArgs e) 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 @@ private void Channel_ExtendedDataReceived(object sender, ChannelExtendedDataEven 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 68c8cf034..47c169e74 100644 --- a/test/Renci.SshNet.IntegrationBenchmarks/SshClientBenchmark.cs +++ b/test/Renci.SshNet.IntegrationBenchmarks/SshClientBenchmark.cs @@ -72,6 +72,18 @@ public string RunCommand() 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 2cfeb53c3..4a0d4df65 100644 --- a/test/Renci.SshNet.IntegrationTests/SshClientTests.cs +++ b/test/Renci.SshNet.IntegrationTests/SshClientTests.cs @@ -22,6 +22,28 @@ public void Echo_Command_with_all_characters() 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 3cf4a9a10..1f9be9a0e 100644 --- a/test/Renci.SshNet.IntegrationTests/SshTests.cs +++ b/test/Renci.SshNet.IntegrationTests/SshTests.cs @@ -172,7 +172,7 @@ public void Ssh_CreateShell() } [TestMethod] - public void Ssh_Command_IntermittendOutput_EndExecute() + public void Ssh_Command_IntermittentOutput_EndExecute() { const string remoteFile = "/home/sshnet/test.sh"; @@ -229,16 +229,8 @@ public void Ssh_Command_IntermittendOutput_EndExecute() } } - /// - /// 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 @@ public void Ssh_Command_IntermittendOutput_OutputStream() 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 2e0c27477..a152e9e53 100644 --- a/test/Renci.SshNet.Tests/.editorconfig +++ b/test/Renci.SshNet.Tests/.editorconfig @@ -309,6 +309,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 @@ -401,4 +405,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 6d56807fd..6d93355f4 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 @@ public void Test_PipeStream_Write_Read_Buffer() using (var stream = new PipeStream()) { - stream.Write(testBuffer, 0, testBuffer.Length); - - Assert.AreEqual(stream.Length, testBuffer.Length); + stream.Write(testBuffer, 0, 512); - _ = stream.Read(outputBuffer, 0, outputBuffer.Length); + Assert.AreEqual(512, stream.Length); - Assert.AreEqual(stream.Length, 0); + Assert.AreEqual(128, stream.Read(outputBuffer, 64, 128)); + + 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 @@ public void Test_PipeStream_Write_Read_Byte() 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 @@ public void Read() 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 @@ public void CanReadTest() [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 @@ public void LengthTest() 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 ed25b260a..000000000 --- a/test/Renci.SshNet.Tests/Classes/Common/PipeStream_Close_BlockingRead.cs +++ /dev/null @@ -1,52 +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 eb314d4d7..000000000 --- a/test/Renci.SshNet.Tests/Classes/Common/PipeStream_Close_BlockingWrite.cs +++ /dev/null @@ -1,64 +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 415466af2..a92890494 100644 --- a/test/Renci.SshNet.Tests/Classes/Common/PipeStream_Flush_BytesRemainingAfterRead.cs +++ b/test/Renci.SshNet.Tests/Classes/Common/PipeStream_Flush_BytesRemainingAfterRead.cs @@ -88,33 +88,5 @@ public void ReadingMoreBytesThanAvailableDoesNotBlock() 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 f306edf1d..000000000 --- a/test/Renci.SshNet.Tests/Classes/Common/PipeStream_Flush_NoBytesRemainingAfterRead.cs +++ /dev/null @@ -1,61 +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 6ce5e73ff..a1e25cb8e 100644 --- a/test/Renci.SshNet.Tests/Classes/PipeStreamTest_Dispose.cs +++ b/test/Renci.SshNet.Tests/Classes/PipeStreamTest_Dispose.cs @@ -32,47 +32,19 @@ private void Act() [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) - { - } - } - - [TestMethod] - public void MaxBufferLength_Getter_ShouldReturnTwoHundredMegabyte() - { - Assert.AreEqual(200 * 1024 * 1024, _pipeStream.MaxBufferLength); + _pipeStream.Flush(); } [TestMethod] - public void MaxBufferLength_Setter_ShouldModifyMaxBufferLength() + public void Length_ShouldNotThrow() { - 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] @@ -95,33 +67,15 @@ public void Position_Setter_ShouldThrowNotSupportedException() } [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 c51f8499f..f0f6192fb 100644 --- a/test/Renci.SshNet.Tests/Classes/SshCommandTest_Dispose.cs +++ b/test/Renci.SshNet.Tests/Classes/SshCommandTest_Dispose.cs @@ -70,14 +70,7 @@ public void OutputStreamShouldReturnNull() [TestMethod] public void OutputStreamShouldHaveBeenDisposed() { - try - { - _outputStream.ReadByte(); - Assert.Fail(); - } - catch (ObjectDisposedException) - { - } + Assert.AreEqual(-1, _outputStream.ReadByte()); } [TestMethod] @@ -89,14 +82,7 @@ public void ExtendedOutputStreamShouldReturnNull() [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 0001a9184..90fe2e9a8 100644 --- a/test/Renci.SshNet.Tests/Classes/SshCommandTest_EndExecute_ChannelOpen.cs +++ b/test/Renci.SshNet.Tests/Classes/SshCommandTest_EndExecute_ChannelOpen.cs @@ -53,8 +53,7 @@ private void Arrange() _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); @@ -70,6 +69,8 @@ private void Arrange() 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()