GithubStar/SSH.NET
1
0
Fork 0
mirror of https://github.com/sshnet/SSH.NET.git synced 2026-10-24 10:10:48 +08:00
Code Issues Packages Projects Releases Wiki Activity

Build the read-ahead mechanism into SftpFileStream (#1705)

Browse Source 
* Build the read-ahead mechanism into SftpFileStream

This change unifies the SFTP download implementations that exist via DownloadFile and
via SftpFileStream, by rewriting SftpFileStream to perform the same "read-aheads" as
DownloadFile. This brings the performance of downloads via SftpFileStream in line with
DownloadFile, such that the latter is now effectively SftpFileStream.CopyTo. It also
brings the recently added DownloadFileAsync up to speed since that was implemented via
SftpFileStream.CopyToAsync.

The methodology is a mix of the previous one and that within OpenSSH: the first call to
SftpFileStream.Read sends one read request to the server. The second sends two and when
not interrupted by Write or similar, the number of in-flight read requests continues to
scale up in this fashion.

I have measured CopyTo to be 3-20x faster than before, depending on file size and server
round-trip time.

* Check CanSeek in ReadAllBytes

* Squeeze out some performance
This commit is contained in:
Rob Hague
2025-10-22 19:55:13 +02:00
committed by GitHub
parent c5c6f28d60
commit dccdedc36e
38 changed files with 1026 additions and 4764 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Renci.SshNet/Common/Extensions.cs
+24
View File
@@ -3,6 +3,7 @@ using System.Collections.Generic;
using System.Globalization;
#if !NET
using System.IO;
using System.Threading.Tasks;
#endif
using System.Net;
using System.Net.Sockets;
@@ -398,6 +399,29 @@ namespace Renci.SshNet.Common
totalRead += read;
}
}
internal static Task<T> WaitAsync<T>(this Task<T> task, CancellationToken cancellationToken)
{
if (task.IsCompleted || !cancellationToken.CanBeCanceled)
{
return task;
}
return WaitCore();
async Task<T> WaitCore()
{
TaskCompletionSource<T> tcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
using var reg = cancellationToken.Register(
() => tcs.TrySetCanceled(cancellationToken),
useSynchronizationContext: false);
var completedTask = await Task.WhenAny(task, tcs.Task).ConfigureAwait(false);
return await completedTask.ConfigureAwait(false);
}
}
#endif
}
}
src/Renci.SshNet/IServiceFactory.cs
-12
View File
@@ -83,18 +83,6 @@ namespace Renci.SshNet
/// <exception cref="SshConnectionException">No key exchange algorithm is supported by both client and server.</exception>
IKeyExchange CreateKeyExchange(IDictionary<string, Func<IKeyExchange>> clientAlgorithms, string[] serverAlgorithms);
/// <summary>
/// Creates an <see cref="ISftpFileReader"/> for the specified file and with the specified
/// buffer size.
/// </summary>
/// <param name="fileName">The file to read.</param>
/// <param name="sftpSession">The SFTP session to use.</param>
/// <param name="bufferSize">The size of buffer.</param>
/// <returns>
/// An <see cref="ISftpFileReader"/>.
/// </returns>
ISftpFileReader CreateSftpFileReader(string fileName, ISftpSession sftpSession, uint bufferSize);
/// <summary>
/// Creates a new <see cref="ISftpResponseFactory"/> instance.
/// </summary>
src/Renci.SshNet/ServiceFactory.cs
-47
View File
@@ -4,8 +4,6 @@ using System.Linq;
using System.Net.Sockets;
using System.Text;
using Microsoft.Extensions.Logging;
using Renci.SshNet.Common;
using Renci.SshNet.Connection;
using Renci.SshNet.Messages.Transport;
@@ -118,51 +116,6 @@ namespace Renci.SshNet
return new NetConfSession(session, operationTimeout);
}
/// <summary>
/// Creates an <see cref="ISftpFileReader"/> for the specified file and with the specified
/// buffer size.
/// </summary>
/// <param name="fileName">The file to read.</param>
/// <param name="sftpSession">The SFTP session to use.</param>
/// <param name="bufferSize">The size of buffer.</param>
/// <returns>
/// An <see cref="ISftpFileReader"/>.
/// </returns>
public ISftpFileReader CreateSftpFileReader(string fileName, ISftpSession sftpSession, uint bufferSize)
{
const int defaultMaxPendingReads = 10;
// Issue #292: Avoid overlapping SSH_FXP_OPEN and SSH_FXP_LSTAT requests for the same file as this
// causes a performance degradation on Sun SSH
var openAsyncResult = sftpSession.BeginOpen(fileName, Flags.Read, callback: null, state: null);
var handle = sftpSession.EndOpen(openAsyncResult);
var statAsyncResult = sftpSession.BeginLStat(fileName, callback: null, state: null);
long? fileSize;
int maxPendingReads;
var chunkSize = sftpSession.CalculateOptimalReadLength(bufferSize);
// fallback to a default maximum of pending reads when remote server does not allow us to obtain
// the attributes of the file
try
{
var fileAttributes = sftpSession.EndLStat(statAsyncResult);
fileSize = fileAttributes.Size;
maxPendingReads = Math.Min(100, (int)Math.Ceiling((double)fileAttributes.Size / chunkSize) + 1);
}
catch (SshException ex)
{
fileSize = null;
maxPendingReads = defaultMaxPendingReads;
sftpSession.SessionLoggerFactory.CreateLogger<ServiceFactory>().LogInformation(ex, "Failed to obtain size of file. Allowing maximum {MaxPendingReads} pending reads", maxPendingReads);
}
return sftpSession.CreateFileReader(handle, sftpSession, chunkSize, maxPendingReads, fileSize);
}
/// <summary>
/// Creates a new <see cref="ISftpResponseFactory"/> instance.
/// </summary>
src/Renci.SshNet/Sftp/ISftpFileReader.cs
-23
View File
@@ -1,23 +0,0 @@
using System;
using Renci.SshNet.Common;
namespace Renci.SshNet.Sftp
{
/// <summary>
/// Reads a given file.
/// </summary>
internal interface ISftpFileReader : IDisposable
{
/// <summary>
/// Reads a sequence of bytes from the current file and advances the position within the file by the number of bytes read.
/// </summary>
/// <returns>
/// The sequence of bytes read from the file, or a zero-length array if the end of the file
/// has been reached.
/// </returns>
/// <exception cref="ObjectDisposedException">The current <see cref="ISftpFileReader"/> is disposed.</exception>
/// <exception cref="SshException">Attempting to read beyond the end of the file.</exception>
byte[] Read();
}
}
src/Renci.SshNet/Sftp/ISftpSession.cs
+1 -16
View File
@@ -67,11 +67,10 @@ namespace Renci.SshNet.Sftp
/// Asynchronously performs a <c>SSH_FXP_FSTAT</c> request.
/// </summary>
/// <param name="handle">The handle.</param>
/// <param name="nullOnError">If set to <see langword="true"/>, <see langword="null"/> is returned in case of an error.</param>
/// <returns>
/// The file attributes.
/// </returns>
SftpFileAttributes RequestFStat(byte[] handle, bool nullOnError);
SftpFileAttributes RequestFStat(byte[] handle);
/// <summary>
/// Asynchronously performs a <c>SSH_FXP_FSTAT</c> request.
@@ -522,19 +521,5 @@ namespace Renci.SshNet.Sftp
/// Currently, we do not take the remote window size into account.
/// </remarks>
uint CalculateOptimalWriteLength(uint bufferSize, byte[] handle);
/// <summary>
/// Creates an <see cref="ISftpFileReader"/> for reading the content of the file represented by a given <paramref name="handle"/>.
/// </summary>
/// <param name="handle">The handle of the file to read.</param>
/// <param name="sftpSession">The SFTP session.</param>
/// <param name="chunkSize">The maximum number of bytes to read with each chunk.</param>
/// <param name="maxPendingReads">The maximum number of pending reads.</param>
/// <param name="fileSize">The size of the file or <see langword="null"/> when the size could not be determined.</param>
/// <returns>
/// An <see cref="ISftpFileReader"/> for reading the content of the file represented by the
/// specified <paramref name="handle"/>.
/// </returns>
ISftpFileReader CreateFileReader(byte[] handle, ISftpSession sftpSession, uint chunkSize, int maxPendingReads, long? fileSize);
}
}
src/Renci.SshNet/Sftp/SftpFileReader.cs
+150 -414
View File
@@ -1,469 +1,205 @@
using System;
#nullable enable
using System;
using System.Collections.Generic;
using System.Globalization;
using System.Diagnostics;
using System.Runtime.ExceptionServices;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Renci.SshNet.Abstractions;
#if !NET
using Renci.SshNet.Common;
#endif
namespace Renci.SshNet.Sftp
{
internal sealed class SftpFileReader : ISftpFileReader
public sealed partial class SftpFileStream
{
private const int ReadAheadWaitTimeoutInMilliseconds = 1000;
private readonly byte[] _handle;
private readonly ISftpSession _sftpSession;
private readonly uint _chunkSize;
private readonly SemaphoreSlim _semaphore;
private readonly object _readLock;
private readonly ManualResetEvent _disposingWaitHandle;
private readonly ManualResetEvent _readAheadCompleted;
private readonly Dictionary<int, BufferedRead> _queue;
private readonly WaitHandle[] _waitHandles;
private readonly ILogger _logger;
/// <summary>
/// Holds the size of the file, when available.
/// </summary>
private readonly long? _fileSize;
private ulong _offset;
private int _readAheadChunkIndex;
private ulong _readAheadOffset;
private int _nextChunkIndex;
/// <summary>
/// Holds a value indicating whether EOF has already been signaled by the SSH server.
/// </summary>
private bool _endOfFileReceived;
/// <summary>
/// Holds a value indicating whether the client has read up to the end of the file.
/// </summary>
private bool _isEndOfFileRead;
private bool _disposingOrDisposed;
private Exception _exception;
/// <summary>
/// Initializes a new instance of the <see cref="SftpFileReader"/> class with the specified handle,
/// <see cref="ISftpSession"/> and the maximum number of pending reads.
/// </summary>
/// <param name="handle">The file handle.</param>
/// <param name="sftpSession">The SFT session.</param>
/// <param name="chunkSize">The size of a individual read-ahead chunk.</param>
/// <param name="maxPendingReads">The maximum number of pending reads.</param>
/// <param name="fileSize">The size of the file, if known; otherwise, <see langword="null"/>.</param>
public SftpFileReader(byte[] handle, ISftpSession sftpSession, uint chunkSize, int maxPendingReads, long? fileSize)
private sealed class SftpFileReader : IDisposable
{
_handle = handle;
_sftpSession = sftpSession;
_chunkSize = chunkSize;
_fileSize = fileSize;
_semaphore = new SemaphoreSlim(maxPendingReads);
_queue = new Dictionary<int, BufferedRead>(maxPendingReads);
_readLock = new object();
_readAheadCompleted = new ManualResetEvent(initialState: false);
_disposingWaitHandle = new ManualResetEvent(initialState: false);
_waitHandles = _sftpSession.CreateWaitHandleArray(_disposingWaitHandle, _semaphore.AvailableWaitHandle);
_logger = sftpSession.SessionLoggerFactory.CreateLogger<SftpFileReader>();
private readonly byte[] _handle;
private readonly ISftpSession _sftpSession;
private readonly int _maxPendingReads;
private readonly ulong? _fileSize;
private readonly Dictionary<ulong, Request> _requests = [];
private readonly CancellationTokenSource _cts;
StartReadAhead();
}
private uint _chunkSize;
public byte[] Read()
{
ThrowHelper.ThrowObjectDisposedIf(_disposingOrDisposed, this);
private ulong _offset;
private ulong _readAheadOffset;
private int _currentMaxRequests;
private ExceptionDispatchInfo? _exception;
if (_exception is not null)
/// <summary>
/// Initializes a new instance of the <see cref="SftpFileReader"/> class with the specified handle,
/// <see cref="ISftpSession"/> and the maximum number of pending reads.
/// </summary>
/// <param name="handle">The file handle.</param>
/// <param name="sftpSession">The SFTP session.</param>
/// <param name="chunkSize">The size of a individual read-ahead chunk.</param>
/// <param name="position">The starting offset in the file.</param>
/// <param name="maxPendingReads">The maximum number of pending reads.</param>
/// <param name="fileSize">The size of the file, if known; otherwise, <see langword="null"/>.</param>
/// <param name="initialMaxRequests">The initial number of pending reads.</param>
public SftpFileReader(byte[] handle, ISftpSession sftpSession, int chunkSize, long position, int maxPendingReads, ulong? fileSize = null, int initialMaxRequests = 1)
{
ExceptionDispatchInfo.Capture(_exception).Throw();
Debug.Assert(chunkSize > 0);
Debug.Assert(position >= 0);
Debug.Assert(initialMaxRequests >= 1);
_handle = handle;
_sftpSession = sftpSession;
_chunkSize = (uint)chunkSize;
_offset = _readAheadOffset = (ulong)position;
_maxPendingReads = maxPendingReads;
_fileSize = fileSize;
_currentMaxRequests = initialMaxRequests;
_cts = new CancellationTokenSource();
}
if (_isEndOfFileRead)
public async Task<byte[]> ReadAsync(CancellationToken cancellationToken)
{
throw new SshException("Attempting to read beyond the end of the file.");
}
_exception?.Throw();
BufferedRead nextChunk;
lock (_readLock)
{
// wait until either the next chunk is available, an exception has occurred or the current
// instance is already disposed
while (!_queue.TryGetValue(_nextChunkIndex, out nextChunk) && _exception is null)
try
{
_ = Monitor.Wait(_readLock);
}
// Fill up the requests buffer with as many requests as we currently allow.
// On the first call to Read, that number is 1. On the second it is 2 etc.
while (_requests.Count < _currentMaxRequests)
{
AddRequest(_readAheadOffset, _chunkSize);
// throw when exception occured in read-ahead, or the current instance is already disposed
if (_exception != null)
{
ExceptionDispatchInfo.Capture(_exception).Throw();
}
_readAheadOffset += _chunkSize;
}
var data = nextChunk.Data;
var request = _requests[_offset];
var data = await request.Task.WaitAsync(cancellationToken).ConfigureAwait(false);
if (nextChunk.Offset == _offset)
{
// have we reached EOF?
if (data.Length == 0)
{
// PERF: we do not bother updating all of the internal state when we've reached EOF
_isEndOfFileRead = true;
// EOF. We effectively disable this instance - further reads will
// continue to return EOF.
_currentMaxRequests = 0;
return data;
}
else
_ = _requests.Remove(_offset);
_offset += (ulong)data.Length;
if (data.Length < request.Count)
{
// remove processed chunk
_ = _queue.Remove(_nextChunkIndex);
// We didn't receive all the data we requested.
// update offset
_offset += (ulong)data.Length;
// If we've read exactly up to our known file size and the next
// request is already in-flight, then wait for it and if it signals
// EOF (as is likely), then call EOF here and omit a final round-trip.
// This optimisation is mostly only beneficial to smaller files on
// higher latency connections.
// move to next chunk
_nextChunkIndex++;
var nextRequestOffset = _offset - (ulong)data.Length + request.Count;
if (_offset == _fileSize &&
_requests.TryGetValue(nextRequestOffset, out var nextRequest))
{
var nextRequestData = await nextRequest.Task.WaitAsync(cancellationToken).ConfigureAwait(false);
if (nextRequestData.Length == 0)
{
_offset = nextRequestOffset;
_currentMaxRequests = 0;
return data;
}
}
// Otherwise, add another request to fill in the gap.
AddRequest(_offset, request.Count - (uint)data.Length);
if (data.Length < _chunkSize)
{
// Right-size the buffer to match the amount that the server
// is willing to return.
// Note that this also happens near EOF.
_chunkSize = Math.Max(512, (uint)data.Length);
}
}
// unblock wait in read-ahead
_ = _semaphore.Release();
if (_currentMaxRequests > 0)
{
if (_readAheadOffset > _fileSize + _chunkSize)
{
// If the file size is known and we've got requests
// out beyond that (plus a buffer for EOD read), then
// restrict the number of outgoing requests.
// This does nothing for the performance of this download
// but may reduce traffic for other downloads.
_currentMaxRequests = 1;
}
else if (_currentMaxRequests < _maxPendingReads)
{
_currentMaxRequests++;
}
}
return data;
}
// When we received an EOF for the next chunk and the size of the file is known, then
// we only complete the current chunk if we haven't already read up to the file size.
// This way we save an extra round-trip to the server.
if (data.Length == 0 && _fileSize.HasValue && _offset == (ulong)_fileSize.Value)
catch (Exception ex) when (!(ex is OperationCanceledException oce && oce.CancellationToken == cancellationToken))
{
// avoid future reads
_isEndOfFileRead = true;
// unblock wait in read-ahead
_ = _semaphore.Release();
// signal EOF to caller
return nextChunk.Data;
// If the wait was cancelled then we will allow subsequent reads as normal.
// For any other errors, we prevent further read requests, effectively disabling
// this instance.
_currentMaxRequests = 0;
_exception = ExceptionDispatchInfo.Capture(ex);
throw;
}
}
/*
* When the server returned less bytes than requested (for the previous chunk)
* we'll synchronously request the remaining data.
*
* Due to the optimization above, we'll only get here in one of the following cases:
* - an EOF situation for files for which we were unable to obtain the file size
* - fewer bytes that requested were returned
*
* According to the SSH specification, this last case should never happen for normal
* disk files (but can happen for device files). In practice, OpenSSH - for example -
* returns less bytes than requested when requesting more than 64 KB.
*
* Important:
* To avoid a deadlock, this read must be done outside of the read lock.
*/
var bytesToCatchUp = nextChunk.Offset - _offset;
/*
* TODO: break loop and interrupt blocking wait in case of exception
*/
var read = _sftpSession.RequestRead(_handle, _offset, (uint)bytesToCatchUp);
if (read.Length == 0)
private void AddRequest(ulong offset, uint count)
{
// process data in read lock to avoid ObjectDisposedException while releasing semaphore
lock (_readLock)
{
// a zero-length (EOF) response is only valid for the read-back when EOF has
// been signaled for the next read-ahead chunk
if (nextChunk.Data.Length == 0)
{
_isEndOfFileRead = true;
// ensure we've not yet disposed the current instance
if (!_disposingOrDisposed)
{
// unblock wait in read-ahead
_ = _semaphore.Release();
}
// signal EOF to caller
return read;
}
// move reader to error state
_exception = new SshException("Unexpectedly reached end of file.");
// ensure we've not yet disposed the current instance
if (!_disposingOrDisposed)
{
// unblock wait in read-ahead
_ = _semaphore.Release();
}
// notify caller of error
throw _exception;
}
_requests.Add(
offset,
new Request(
offset,
count,
_sftpSession.RequestReadAsync(_handle, offset, count, _cts.Token)));
}
_offset += (uint)read.Length;
return read;
}
public void Dispose()
{
Dispose(disposing: true);
GC.SuppressFinalize(this);
}
/// <summary>
/// Releases unmanaged and - optionally - managed resources.
/// </summary>
/// <param name="disposing"><see langword="true"/> to release both managed and unmanaged resources; <see langword="false"/> to release only unmanaged resources.</param>
private void Dispose(bool disposing)
{
if (_disposingOrDisposed)
public void Dispose()
{
return;
}
_exception ??= ExceptionDispatchInfo.Capture(new ObjectDisposedException(GetType().FullName));
// transition to disposing state
_disposingOrDisposed = true;
if (disposing)
{
// record exception to break prevent future Read()
_exception = new ObjectDisposedException(GetType().FullName);
// signal that we're disposing to interrupt wait in read-ahead
_ = _disposingWaitHandle.Set();
// wait until the read-ahead thread has completed
_ = _readAheadCompleted.WaitOne();
// unblock the Read()
lock (_readLock)
if (_requests.Count > 0)
{
// dispose semaphore in read lock to ensure we don't run into an ObjectDisposedException
// in Read()
_semaphore.Dispose();
// Cancel outstanding requests and observe the exception on them
// as an effort to prevent unhandled exceptions.
// awake Read
Monitor.PulseAll(_readLock);
_cts.Cancel();
foreach (var request in _requests.Values)
{
_ = request.Task.Exception;
}
_requests.Clear();
}
_readAheadCompleted.Dispose();
_disposingWaitHandle.Dispose();
if (_sftpSession.IsOpen)
{
try
{
var closeAsyncResult = _sftpSession.BeginClose(_handle, callback: null, state: null);
_sftpSession.EndClose(closeAsyncResult);
}
catch (Exception ex)
{
_logger.LogInformation(ex, "Failure closing handle");
}
}
_cts.Dispose();
}
}
private void StartReadAhead()
{
ThreadAbstraction.ExecuteThread(() =>
private sealed class Request
{
while (!_endOfFileReceived && _exception is null)
public Request(ulong offset, uint count, Task<byte[]> task)
{
// check if we should continue with the read-ahead loop
// note that the EOF and exception check are not included
// in this check as they do not require Read() to be
// unblocked (or have already done this)
if (!ContinueReadAhead())
{
// unblock the Read()
lock (_readLock)
{
Monitor.PulseAll(_readLock);
}
// break the read-ahead loop
break;
}
// attempt to obtain the semaphore; this may time out when all semaphores are
// in use due to pending read-aheads (which in turn can happen when the server
// is slow to respond or when the session is broken)
if (!_semaphore.Wait(ReadAheadWaitTimeoutInMilliseconds))
{
// re-evaluate whether an exception occurred, and - if not - wait again
continue;
}
// don't bother reading any more chunks if we received EOF, an exception has occurred
// or the current instance is disposed
if (_endOfFileReceived || _exception != null)
{
break;
}
// start reading next chunk
var bufferedRead = new BufferedRead(_readAheadChunkIndex, _readAheadOffset);
try
{
// even if we know the size of the file and have read up to EOF, we still want
// to keep reading (ahead) until we receive zero bytes from the remote host as
// we do not want to rely purely on the reported file size
//
// if the offset of the read-ahead chunk is greater than that file size, then
// we can expect to be reading the last (zero-byte) chunk and switch to synchronous
// mode to avoid having multiple read-aheads that read beyond EOF
if (_fileSize != null && (long)_readAheadOffset > _fileSize.Value)
{
var asyncResult = _sftpSession.BeginRead(_handle, _readAheadOffset, _chunkSize, callback: null, bufferedRead);
var data = _sftpSession.EndRead(asyncResult);
ReadCompletedCore(bufferedRead, data);
}
else
{
_ = _sftpSession.BeginRead(_handle, _readAheadOffset, _chunkSize, ReadCompleted, bufferedRead);
}
}
catch (Exception ex)
{
HandleFailure(ex);
break;
}
// advance read-ahead offset
_readAheadOffset += _chunkSize;
// increment index of read-ahead chunk
_readAheadChunkIndex++;
Offset = offset;
Count = count;
Task = task;
}
_ = _readAheadCompleted.Set();
});
}
/// <summary>
/// Returns a value indicating whether the read-ahead loop should be continued.
/// </summary>
/// <returns>
/// <see langword="true"/> if the read-ahead loop should be continued; otherwise, <see langword="false"/>.
/// </returns>
private bool ContinueReadAhead()
{
try
{
var waitResult = _sftpSession.WaitAny(_waitHandles, _sftpSession.OperationTimeout);
switch (waitResult)
{
case 0: // disposing
return false;
case 1: // semaphore available
return true;
default:
throw new NotImplementedException(string.Format(CultureInfo.InvariantCulture, "WaitAny return value '{0}' is not implemented.", waitResult));
}
}
catch (Exception ex)
{
_ = Interlocked.CompareExchange(ref _exception, ex, comparand: null);
return false;
}
}
private void ReadCompleted(IAsyncResult result)
{
if (_disposingOrDisposed)
{
// skip further processing if we're disposing the current instance
// to avoid accessing disposed members
return;
}
var readAsyncResult = (SftpReadAsyncResult)result;
byte[] data;
try
{
data = readAsyncResult.EndInvoke();
}
catch (Exception ex)
{
HandleFailure(ex);
return;
}
// a read that completes with a zero-byte result signals EOF
// but there may be pending reads before that read
var bufferedRead = (BufferedRead)readAsyncResult.AsyncState;
ReadCompletedCore(bufferedRead, data);
}
private void ReadCompletedCore(BufferedRead bufferedRead, byte[] data)
{
bufferedRead.Complete(data);
lock (_readLock)
{
// add item to queue
_queue.Add(bufferedRead.ChunkIndex, bufferedRead);
// Signal that a chunk has been read or EOF has been reached.
// In both cases, Read() will eventually also unblock the "read-ahead" thread.
Monitor.PulseAll(_readLock);
}
// check if server signaled EOF
if (data.Length == 0)
{
// set a flag to stop read-aheads
_endOfFileReceived = true;
}
}
private void HandleFailure(Exception cause)
{
_ = Interlocked.CompareExchange(ref _exception, cause, comparand: null);
// unblock read-ahead
_ = _semaphore.Release();
// unblock Read()
lock (_readLock)
{
Monitor.PulseAll(_readLock);
}
}
internal sealed class BufferedRead
{
public int ChunkIndex { get; }
public byte[] Data { get; private set; }
public ulong Offset { get; }
public BufferedRead(int chunkIndex, ulong offset)
{
ChunkIndex = chunkIndex;
Offset = offset;
}
public void Complete(byte[] data)
{
Data = data;
public ulong Offset { get; }
public uint Count { get; }
public Task<byte[]> Task { get; }
}
}
}
src/Renci.SshNet/Sftp/SftpFileStream.cs
+478 -868
View File
File diff suppressed because it is too large Load Diff
src/Renci.SshNet/Sftp/SftpSession.cs
+5 -27
View File
@@ -251,23 +251,6 @@ namespace Renci.SshNet.Sftp
return canonizedPath + slash + pathParts[pathParts.Length - 1];
}
/// <summary>
/// Creates an <see cref="ISftpFileReader"/> for reading the content of the file represented by a given <paramref name="handle"/>.
/// </summary>
/// <param name="handle">The handle of the file to read.</param>
/// <param name="sftpSession">The SFTP session.</param>
/// <param name="chunkSize">The maximum number of bytes to read with each chunk.</param>
/// <param name="maxPendingReads">The maximum number of pending reads.</param>
/// <param name="fileSize">The size of the file or <see langword="null"/> when the size could not be determined.</param>
/// <returns>
/// An <see cref="ISftpFileReader"/> for reading the content of the file represented by the
/// specified <paramref name="handle"/>.
/// </returns>
public ISftpFileReader CreateFileReader(byte[] handle, ISftpSession sftpSession, uint chunkSize, int maxPendingReads, long? fileSize)
{
return new SftpFileReader(handle, sftpSession, chunkSize, maxPendingReads, fileSize);
}
internal string GetFullRemotePath(string path)
{
var fullPath = path;
@@ -820,6 +803,8 @@ namespace Renci.SshNet.Sftp
/// </returns>
public Task<byte[]> RequestReadAsync(byte[] handle, ulong offset, uint length, CancellationToken cancellationToken)
{
Debug.Assert(length > 0, "This implementation cannot distinguish between EOF and zero-length reads");
if (cancellationToken.IsCancellationRequested)
{
return Task.FromCanceled<byte[]>(cancellationToken);
@@ -1075,15 +1060,8 @@ namespace Renci.SshNet.Sftp
}
}
/// <summary>
/// Performs SSH_FXP_FSTAT request.
/// </summary>
/// <param name="handle">The handle.</param>
/// <param name="nullOnError">If set to <see langword="true"/>, returns <see langword="null"/> instead of throwing an exception.</param>
/// <returns>
/// File attributes.
/// </returns>
public SftpFileAttributes RequestFStat(byte[] handle, bool nullOnError)
/// <inheritdoc/>
public SftpFileAttributes RequestFStat(byte[] handle)
{
SshException exception = null;
SftpFileAttributes attributes = null;
@@ -1109,7 +1087,7 @@ namespace Renci.SshNet.Sftp
WaitOnHandle(wait, OperationTimeout);
}
if (!nullOnError && exception is not null)
if (exception is not null)
{
throw exception;
}
src/Renci.SshNet/SftpClient.cs
+122 -47
View File
@@ -1,5 +1,6 @@
#nullable enable
using System;
using System.Buffers;
using System.Collections.Generic;
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
@@ -899,17 +900,33 @@ namespace Renci.SshNet
/// <inheritdoc />
public void DownloadFile(string path, Stream output, Action<ulong>? downloadCallback = null)
{
ThrowHelper.ThrowIfNullOrWhiteSpace(path);
ThrowHelper.ThrowIfNull(output);
CheckDisposed();
InternalDownloadFile(path, output, asyncResult: null, downloadCallback);
InternalDownloadFile(
path,
output,
asyncResult: null,
downloadCallback,
isAsync: false,
CancellationToken.None).GetAwaiter().GetResult();
}
/// <inheritdoc />
public Task DownloadFileAsync(string path, Stream output, CancellationToken cancellationToken = default)
{
ThrowHelper.ThrowIfNullOrWhiteSpace(path);
ThrowHelper.ThrowIfNull(output);
CheckDisposed();
return InternalDownloadFileAsync(path, output, cancellationToken);
return InternalDownloadFile(
path,
output,
asyncResult: null,
downloadCallback: null,
isAsync: true,
cancellationToken);
}
/// <summary>
@@ -976,17 +993,25 @@ namespace Renci.SshNet
/// </remarks>
public IAsyncResult BeginDownloadFile(string path, Stream output, AsyncCallback? asyncCallback, object? state, Action<ulong>? downloadCallback = null)
{
CheckDisposed();
ThrowHelper.ThrowIfNullOrWhiteSpace(path);
ThrowHelper.ThrowIfNull(output);
CheckDisposed();
var asyncResult = new SftpDownloadAsyncResult(asyncCallback, state);
ThreadAbstraction.ExecuteThread(() =>
_ = DoDownloadAndSetResult();
async Task DoDownloadAndSetResult()
{
try
{
InternalDownloadFile(path, output, asyncResult, downloadCallback);
await InternalDownloadFile(
path,
output,
asyncResult,
downloadCallback,
isAsync: true,
CancellationToken.None).ConfigureAwait(false);
asyncResult.SetAsCompleted(exception: null, completedSynchronously: false);
}
@@ -994,7 +1019,7 @@ namespace Renci.SshNet
{
asyncResult.SetAsCompleted(exp, completedSynchronously: false);
}
});
}
return asyncResult;
}
@@ -1050,7 +1075,7 @@ namespace Renci.SshNet
asyncResult: null,
uploadCallback,
isAsync: false,
default).GetAwaiter().GetResult();
CancellationToken.None).GetAwaiter().GetResult();
}
/// <inheritdoc />
@@ -1689,8 +1714,20 @@ namespace Renci.SshNet
{
using (var stream = OpenRead(path))
{
var buffer = new byte[stream.Length];
stream.ReadExactly(buffer, 0, buffer.Length);
byte[] buffer;
if (stream.CanSeek)
{
buffer = new byte[stream.Length];
stream.ReadExactly(buffer, 0, buffer.Length);
}
else
{
MemoryStream ms = new();
stream.CopyTo(ms);
buffer = ms.ToArray();
}
return buffer;
}
}
@@ -2233,32 +2270,59 @@ namespace Renci.SshNet
return result;
}
/// <summary>
/// Internals the download file.
/// </summary>
/// <param name="path">The path.</param>
/// <param name="output">The output.</param>
/// <param name="asyncResult">An <see cref="IAsyncResult"/> that references the asynchronous request.</param>
/// <param name="downloadCallback">The download callback.</param>
/// <exception cref="ArgumentNullException"><paramref name="output" /> is <see langword="null"/>.</exception>
/// <exception cref="ArgumentException"><paramref name="path" /> is <see langword="null"/> or contains whitespace.</exception>
/// <exception cref="SshConnectionException">Client not connected.</exception>
private void InternalDownloadFile(string path, Stream output, SftpDownloadAsyncResult? asyncResult, Action<ulong>? downloadCallback)
#pragma warning disable S6966 // Awaitable method should be used
private async Task InternalDownloadFile(
string path,
Stream output,
SftpDownloadAsyncResult? asyncResult,
Action<ulong>? downloadCallback,
bool isAsync,
CancellationToken cancellationToken)
{
ThrowHelper.ThrowIfNull(output);
ThrowHelper.ThrowIfNullOrWhiteSpace(path);
Debug.Assert(!string.IsNullOrWhiteSpace(path));
Debug.Assert(output is not null);
Debug.Assert(isAsync || cancellationToken == default);
if (_sftpSession is null)
{
throw new SshConnectionException("Client not connected.");
}
var fullPath = _sftpSession.GetCanonicalPath(path);
SftpFileStream sftpStream;
using (var fileReader = ServiceFactory.CreateSftpFileReader(fullPath, _sftpSession, _bufferSize))
if (isAsync)
{
var totalBytesRead = 0UL;
var fullPath = await _sftpSession.GetCanonicalPathAsync(path, cancellationToken).ConfigureAwait(false);
sftpStream = await SftpFileStream.OpenAsync(
_sftpSession,
fullPath,
FileMode.Open,
FileAccess.Read,
(int)_bufferSize,
cancellationToken,
isDownloadFile: true).ConfigureAwait(false);
}
else
{
var fullPath = _sftpSession.GetCanonicalPath(path);
sftpStream = SftpFileStream.Open(
_sftpSession,
fullPath,
FileMode.Open,
FileAccess.Read,
(int)_bufferSize,
isDownloadFile: true);
}
// The below is effectively sftpStream.CopyTo{Async}(output) with consideration
// for downloadCallback/asyncResult.
var buffer = ArrayPool<byte>.Shared.Rent(81920);
try
{
ulong totalBytesRead = 0;
while (true)
{
// Cancel download
@@ -2267,15 +2331,33 @@ namespace Renci.SshNet
break;
}
var data = fileReader.Read();
if (data.Length == 0)
var bytesRead = isAsync
#if NET
? await sftpStream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false)
#else
? await sftpStream.ReadAsync(buffer, 0, buffer.Length, cancellationToken).ConfigureAwait(false)
#endif
: sftpStream.Read(buffer, 0, buffer.Length);
if (bytesRead == 0)
{
break;
}
output.Write(data, 0, data.Length);
if (isAsync)
{
#if NET
await output.WriteAsync(buffer.AsMemory(0, bytesRead), cancellationToken).ConfigureAwait(false);
#else
await output.WriteAsync(buffer, 0, bytesRead, cancellationToken).ConfigureAwait(false);
#endif
}
else
{
output.Write(buffer, 0, bytesRead);
}
totalBytesRead += (ulong)data.Length;
totalBytesRead += (ulong)bytesRead;
asyncResult?.Update(totalBytesRead);
@@ -2289,28 +2371,21 @@ namespace Renci.SshNet
}
}
}
}
private async Task InternalDownloadFileAsync(string path, Stream output, CancellationToken cancellationToken)
{
ThrowHelper.ThrowIfNull(output);
ThrowHelper.ThrowIfNullOrWhiteSpace(path);
if (_sftpSession is null)
finally
{
throw new SshConnectionException("Client not connected.");
}
ArrayPool<byte>.Shared.Return(buffer);
cancellationToken.ThrowIfCancellationRequested();
var fullPath = await _sftpSession.GetCanonicalPathAsync(path, cancellationToken).ConfigureAwait(false);
var openStreamTask = SftpFileStream.OpenAsync(_sftpSession, fullPath, FileMode.Open, FileAccess.Read, (int)_bufferSize, cancellationToken);
using (var input = await openStreamTask.ConfigureAwait(false))
{
await input.CopyToAsync(output, 81920, cancellationToken).ConfigureAwait(false);
if (isAsync)
{
await sftpStream.DisposeAsync().ConfigureAwait(false);
}
else
{
sftpStream.Dispose();
}
}
}
#pragma warning restore S6966 // Awaitable method should be used
#pragma warning disable S6966 // Awaitable method should be used
private async Task InternalUploadFile(