diff --git a/src/Renci.SshNet.NET35/Renci.SshNet.NET35.csproj b/src/Renci.SshNet.NET35/Renci.SshNet.NET35.csproj index 3d399f39..eb2af862 100644 --- a/src/Renci.SshNet.NET35/Renci.SshNet.NET35.csproj +++ b/src/Renci.SshNet.NET35/Renci.SshNet.NET35.csproj @@ -1,5 +1,5 @@  - + Debug AnyCPU @@ -12,29 +12,33 @@ Renci.SshNet v3.5 512 + true full false bin\Debug\ - TRACE;DEBUG;FEATURE_REGEX_COMPILE;FEATURE_BINARY_SERIALIZATION;FEATURE_RNG_CREATE;FEATURE_SOCKET_SYNC;FEATURE_SOCKET_EAP;FEATURE_SOCKET_APM;FEATURE_SOCKET_SETSOCKETOPTION;FEATURE_SOCKET_POLL;FEATURE_STREAM_APM;FEATURE_DNS_SYNC;FEATURE_THREAD_THREADPOOL;FEATURE_THREAD_SLEEP;FEATURE_HASH_MD5;FEATURE_HASH_SHA1_CREATE;FEATURE_HASH_SHA256_CREATE;FEATURE_HASH_SHA384_CREATE;FEATURE_HASH_SHA512_CREATE;FEATURE_HASH_RIPEMD160_CREATE;FEATURE_HMAC_MD5;FEATURE_HMAC_SHA1;FEATURE_HMAC_SHA256;FEATURE_HMAC_SHA384;FEATURE_HMAC_SHA512;FEATURE_HMAC_RIPEMD160;FEATURE_MEMORYSTREAM_GETBUFFER;FEATURE_DIAGNOSTICS_TRACESOURCE;FEATURE_ENCODING_ASCII + TRACE;DEBUG;FEATURE_REGEX_COMPILE;FEATURE_BINARY_SERIALIZATION;FEATURE_RNG_CREATE;FEATURE_SOCKET_SYNC;FEATURE_SOCKET_EAP;FEATURE_SOCKET_APM;FEATURE_SOCKET_SETSOCKETOPTION;FEATURE_SOCKET_POLL;FEATURE_SOCKET_SELECT;FEATURE_STREAM_APM;FEATURE_DNS_SYNC;FEATURE_THREAD_THREADPOOL;FEATURE_THREAD_SLEEP;FEATURE_HASH_MD5;FEATURE_HASH_SHA1_CREATE;FEATURE_HASH_SHA256_CREATE;FEATURE_HASH_SHA384_CREATE;FEATURE_HASH_SHA512_CREATE;FEATURE_HASH_RIPEMD160_CREATE;FEATURE_HMAC_MD5;FEATURE_HMAC_SHA1;FEATURE_HMAC_SHA256;FEATURE_HMAC_SHA384;FEATURE_HMAC_SHA512;FEATURE_HMAC_RIPEMD160;FEATURE_MEMORYSTREAM_GETBUFFER;FEATURE_DIAGNOSTICS_TRACESOURCE;FEATURE_ENCODING_ASCII prompt 4 true bin\Debug\Renci.SshNet.xml + 5 + false none true bin\Release\ - TRACE;FEATURE_REGEX_COMPILE;FEATURE_BINARY_SERIALIZATION;FEATURE_RNG_CREATE;FEATURE_SOCKET_SYNC;FEATURE_SOCKET_EAP;FEATURE_SOCKET_APM;FEATURE_SOCKET_SETSOCKETOPTION;FEATURE_SOCKET_POLL;FEATURE_STREAM_APM;FEATURE_DNS_SYNC;FEATURE_THREAD_THREADPOOL;FEATURE_THREAD_SLEEP;FEATURE_HASH_MD5;FEATURE_HASH_SHA1_CREATE;FEATURE_HASH_SHA256_CREATE;FEATURE_HASH_SHA384_CREATE;FEATURE_HASH_SHA512_CREATE;FEATURE_HASH_RIPEMD160_CREATE;FEATURE_HMAC_MD5;FEATURE_HMAC_SHA1;FEATURE_HMAC_SHA256;FEATURE_HMAC_SHA384;FEATURE_HMAC_SHA512;FEATURE_HMAC_RIPEMD160;FEATURE_MEMORYSTREAM_GETBUFFER;FEATURE_DIAGNOSTICS_TRACESOURCE;FEATURE_ENCODING_ASCII + TRACE;FEATURE_REGEX_COMPILE;FEATURE_BINARY_SERIALIZATION;FEATURE_RNG_CREATE;FEATURE_SOCKET_SYNC;FEATURE_SOCKET_EAP;FEATURE_SOCKET_APM;FEATURE_SOCKET_SETSOCKETOPTION;FEATURE_SOCKET_POLL;FEATURE_SOCKET_SELECT;FEATURE_STREAM_APM;FEATURE_DNS_SYNC;FEATURE_THREAD_THREADPOOL;FEATURE_THREAD_SLEEP;FEATURE_HASH_MD5;FEATURE_HASH_SHA1_CREATE;FEATURE_HASH_SHA256_CREATE;FEATURE_HASH_SHA384_CREATE;FEATURE_HASH_SHA512_CREATE;FEATURE_HASH_RIPEMD160_CREATE;FEATURE_HMAC_MD5;FEATURE_HMAC_SHA1;FEATURE_HMAC_SHA256;FEATURE_HMAC_SHA384;FEATURE_HMAC_SHA512;FEATURE_HMAC_RIPEMD160;FEATURE_MEMORYSTREAM_GETBUFFER;FEATURE_DIAGNOSTICS_TRACESOURCE;FEATURE_ENCODING_ASCII prompt 4 bin\Release\Renci.SshNet.xml true + false true @@ -861,6 +865,9 @@ Sftp\SftpFileAttributes.cs + + Sftp\SftpFileReader.cs + Sftp\SftpFileStream.cs @@ -876,6 +883,9 @@ Sftp\SftpMessageTypes.cs + + Sftp\SftpReadAsyncResult.cs + Sftp\SftpSession.cs diff --git a/src/Renci.SshNet.Tests/Renci.SshNet.Tests.csproj b/src/Renci.SshNet.Tests/Renci.SshNet.Tests.csproj index 4801475c..1ceb2872 100644 --- a/src/Renci.SshNet.Tests/Renci.SshNet.Tests.csproj +++ b/src/Renci.SshNet.Tests/Renci.SshNet.Tests.csproj @@ -42,8 +42,8 @@ - - ..\..\packages\Moq.4.2.1409.1722\lib\net40\Moq.dll + + ..\..\..\..\..\Users\Gert Driesen\Dropbox\Privé\Projects\SShNetTests\packages\Moq.4.2.1409.1722\lib\net40\Moq.dll diff --git a/src/Renci.SshNet/Common/AsyncResult.cs b/src/Renci.SshNet/Common/AsyncResult.cs index 70726ed4..51995c14 100644 --- a/src/Renci.SshNet/Common/AsyncResult.cs +++ b/src/Renci.SshNet/Common/AsyncResult.cs @@ -29,10 +29,12 @@ namespace Renci.SshNet.Common private Exception _exception; /// - /// Gets or sets a value indicating whether EndInvoke has been called on the current AsyncResult. + /// Gets or sets a value indicating whether has been called on the current + /// . /// /// - /// true if has been called on the current ; otherwise, false. + /// true if has been called on the current ; + /// otherwise, false. /// public bool EndInvokeCalled { get; private set; } @@ -75,7 +77,7 @@ namespace Renci.SshNet.Common /// /// Waits until the asynchronous operation completes, and then returns. /// - public void EndInvoke() + internal void EndInvoke() { // This method assumes that only 1 thread calls EndInvoke for this object if (!IsCompleted) @@ -93,6 +95,28 @@ namespace Renci.SshNet.Common throw _exception; } + /// + /// Waits until the asynchronous operation completes, and then returns. + /// + internal void EndInvoke(TimeSpan timeout) + { + // This method assumes that only 1 thread calls EndInvoke for this object + if (!IsCompleted) + { + // If the operation isn't done, wait for it + var completed = AsyncWaitHandle.WaitOne(timeout); + _asyncWaitHandle = null; // Allow early GC + AsyncWaitHandle.Dispose(); + + } + + EndInvokeCalled = true; + + // Operation is done: if an exception occurred, throw it + if (_exception != null) + throw _exception; + } + #region Implementation of IAsyncResult /// diff --git a/src/Renci.SshNet/Common/SemaphoreLight.cs b/src/Renci.SshNet/Common/SemaphoreLight.cs index 4ff28acb..19fa7964 100644 --- a/src/Renci.SshNet/Common/SemaphoreLight.cs +++ b/src/Renci.SshNet/Common/SemaphoreLight.cs @@ -64,7 +64,6 @@ namespace Renci.SshNet.Common /// public void Wait() { - lock (_lock) { while (_currentCount < 1) diff --git a/src/Renci.SshNet/Renci.SshNet.csproj b/src/Renci.SshNet/Renci.SshNet.csproj index fd278e17..05b839b7 100644 --- a/src/Renci.SshNet/Renci.SshNet.csproj +++ b/src/Renci.SshNet/Renci.SshNet.csproj @@ -18,7 +18,7 @@ full false bin\Debug\ - TRACE;DEBUG;FEATURE_REGEX_COMPILE;FEATURE_BINARY_SERIALIZATION;FEATURE_RNG_CREATE;FEATURE_SOCKET_SYNC;FEATURE_SOCKET_EAP;FEATURE_SOCKET_APM;FEATURE_SOCKET_SETSOCKETOPTION;FEATURE_SOCKET_POLL;FEATURE_STREAM_APM;FEATURE_DNS_SYNC;FEATURE_THREAD_COUNTDOWNEVENT;FEATURE_THREAD_THREADPOOL;FEATURE_THREAD_SLEEP;FEATURE_HASH_MD5;FEATURE_HASH_SHA1_CREATE;FEATURE_HASH_SHA256_CREATE;FEATURE_HASH_SHA384_CREATE;FEATURE_HASH_SHA512_CREATE;FEATURE_HASH_RIPEMD160_CREATE;FEATURE_HMAC_MD5;FEATURE_HMAC_SHA1;FEATURE_HMAC_SHA256;FEATURE_HMAC_SHA384;FEATURE_HMAC_SHA512;FEATURE_HMAC_RIPEMD160;FEATURE_MEMORYSTREAM_GETBUFFER;FEATURE_DIAGNOSTICS_TRACESOURCE;FEATURE_ENCODING_ASCII + TRACE;DEBUG;FEATURE_REGEX_COMPILE;FEATURE_BINARY_SERIALIZATION;FEATURE_RNG_CREATE;FEATURE_SOCKET_SYNC;FEATURE_SOCKET_EAP;FEATURE_SOCKET_APM;FEATURE_SOCKET_SETSOCKETOPTION;FEATURE_SOCKET_SELECT;FEATURE_SOCKET_POLL;FEATURE_STREAM_APM;FEATURE_DNS_SYNC;FEATURE_THREAD_COUNTDOWNEVENT;FEATURE_THREAD_THREADPOOL;FEATURE_THREAD_SLEEP;FEATURE_HASH_MD5;FEATURE_HASH_SHA1_CREATE;FEATURE_HASH_SHA256_CREATE;FEATURE_HASH_SHA384_CREATE;FEATURE_HASH_SHA512_CREATE;FEATURE_HASH_RIPEMD160_CREATE;FEATURE_HMAC_MD5;FEATURE_HMAC_SHA1;FEATURE_HMAC_SHA256;FEATURE_HMAC_SHA384;FEATURE_HMAC_SHA512;FEATURE_HMAC_RIPEMD160;FEATURE_MEMORYSTREAM_GETBUFFER;FEATURE_DIAGNOSTICS_TRACESOURCE;FEATURE_ENCODING_ASCII prompt 4 bin\Debug\Renci.SshNet.xml @@ -29,13 +29,14 @@ none true bin\Release\ - FEATURE_REGEX_COMPILE;FEATURE_BINARY_SERIALIZATION;FEATURE_RNG_CREATE;FEATURE_SOCKET_SYNC;FEATURE_SOCKET_EAP;FEATURE_SOCKET_APM;FEATURE_SOCKET_SETSOCKETOPTION;FEATURE_SOCKET_POLL;FEATURE_STREAM_APM;FEATURE_DNS_SYNC;FEATURE_THREAD_COUNTDOWNEVENT;FEATURE_THREAD_THREADPOOL;FEATURE_THREAD_SLEEP;FEATURE_HASH_MD5;FEATURE_HASH_SHA1_CREATE;FEATURE_HASH_SHA256_CREATE;FEATURE_HASH_SHA384_CREATE;FEATURE_HASH_SHA512_CREATE;FEATURE_HASH_RIPEMD160_CREATE;FEATURE_HMAC_MD5;FEATURE_HMAC_SHA1;FEATURE_HMAC_SHA256;FEATURE_HMAC_SHA384;FEATURE_HMAC_SHA512;FEATURE_HMAC_RIPEMD160;FEATURE_MEMORYSTREAM_GETBUFFER;FEATURE_DIAGNOSTICS_TRACESOURCE;FEATURE_ENCODING_ASCII + FEATURE_REGEX_COMPILE;FEATURE_BINARY_SERIALIZATION;FEATURE_RNG_CREATE;FEATURE_SOCKET_SYNC;FEATURE_SOCKET_EAP;FEATURE_SOCKET_APM;FEATURE_SOCKET_SETSOCKETOPTION;FEATURE_SOCKET_SELECT;FEATURE_SOCKET_POLL;FEATURE_STREAM_APM;FEATURE_DNS_SYNC;FEATURE_THREAD_COUNTDOWNEVENT;FEATURE_THREAD_THREADPOOL;FEATURE_THREAD_SLEEP;FEATURE_HASH_MD5;FEATURE_HASH_SHA1_CREATE;FEATURE_HASH_SHA256_CREATE;FEATURE_HASH_SHA384_CREATE;FEATURE_HASH_SHA512_CREATE;FEATURE_HASH_RIPEMD160_CREATE;FEATURE_HMAC_MD5;FEATURE_HMAC_SHA1;FEATURE_HMAC_SHA256;FEATURE_HMAC_SHA384;FEATURE_HMAC_SHA512;FEATURE_HMAC_RIPEMD160;FEATURE_MEMORYSTREAM_GETBUFFER;FEATURE_DIAGNOSTICS_TRACESOURCE;FEATURE_ENCODING_ASCII prompt 4 bin\Release\Renci.SshNet.xml true + 5 true @@ -416,6 +417,7 @@ + Code @@ -425,6 +427,7 @@ Code + Code diff --git a/src/Renci.SshNet/Session.cs b/src/Renci.SshNet/Session.cs index e2674309..4cf4c2de 100644 --- a/src/Renci.SshNet/Session.cs +++ b/src/Renci.SshNet/Session.cs @@ -1815,14 +1815,16 @@ namespace Renci.SshNet /// private void MessageListener() { +#if FEATURE_SOCKET_SELECT + var readSockets = new List { _socket }; +#endif // FEATURE_SOCKET_SELECT + try { - var readSockets = new List {_socket}; - // remain in message loop until socket is shut down or until we're disconnecting while (_socket.IsConnected()) { -#if FEATURE_SOCKET_POLL +#if FEATURE_SOCKET_SELECT // if the socket is already disposed when Select is invoked, then a SocketException // stating "An operation was attempted on something that is not a socket" is thrown; // we attempt to avoid this exception by having an IsConnected() that can break the @@ -1835,7 +1837,8 @@ namespace Renci.SshNet // perform a blocking select to determine whether there's is data available to be // read; we do not use a blocking read to allow us to use Socket.Poll to determine - // if the connection is still available (in IsSocketConnected + // if the connection is still available (in IsSocketConnected) + Socket.Select(readSockets, null, null, -1); // the Select invocation will be interrupted in one of the following conditions: @@ -1855,7 +1858,19 @@ namespace Renci.SshNet // break out of the message loop break; } -#endif // FEATURE_SOCKET_POLL +#elif FEATURE_SOCKET_POLL + // when Socket.Select(IList, IList, IList, Int32) is not available or is buggy, we use + // Socket.Poll(Int, SelectMode) to block until either data is available or the socket + // is closed + + var status = _socket.Poll(-1, SelectMode.SelectRead); + if (!_socket.IsConnected()) + { + // connection with SSH server was closed; + // break out of the message loop + break; + } +#endif // FEATURE_SOCKET_SELECT var message = ReceiveMessage(); if (message == null) @@ -1869,7 +1884,7 @@ namespace Renci.SshNet message.Process(this); } - // connection with SSH server was closed + // connection with SSH server was closed or socket was disposed RaiseError(CreateConnectionAbortedByServerException()); } catch (SocketException ex) @@ -2201,6 +2216,7 @@ namespace Renci.SshNet return; } + // "save" exception and set exception wait handle to ensure any waits are interrupted _exception = exp; _exceptionWaitHandle.Set(); diff --git a/src/Renci.SshNet/Sftp/ISftpSession.cs b/src/Renci.SshNet/Sftp/ISftpSession.cs index 56190a9d..1c7660bd 100644 --- a/src/Renci.SshNet/Sftp/ISftpSession.cs +++ b/src/Renci.SshNet/Sftp/ISftpSession.cs @@ -2,11 +2,20 @@ using System.Collections.Generic; using System.Threading; using Renci.SshNet.Sftp.Responses; +using Renci.SshNet.Channels; namespace Renci.SshNet.Sftp { internal interface ISftpSession : ISubsystemSession { + /// + /// Gets the channel associated with this session. + /// + /// + /// The channel associated with this session. + /// + IChannelSession Channel { get; } + /// /// Gets the SFTP protocol version. /// @@ -95,6 +104,10 @@ namespace Renci.SshNet.Sftp /// data array; null if EOF byte[] RequestRead(byte[] handle, ulong offset, uint length); + SftpReadAsyncResult BeginRead(byte[] handle, ulong offset, uint length, AsyncCallback callback, object state); + + byte[] EndRead(IAsyncResult asyncResult); + /// /// Performs SSH_FXP_READDIR request /// diff --git a/src/Renci.SshNet/Sftp/SftpFileReader.cs b/src/Renci.SshNet/Sftp/SftpFileReader.cs new file mode 100644 index 00000000..3c0fad6f --- /dev/null +++ b/src/Renci.SshNet/Sftp/SftpFileReader.cs @@ -0,0 +1,229 @@ +using Renci.SshNet.Abstractions; +using Renci.SshNet.Common; +using System; +using System.Collections.Generic; +using System.Threading; + +namespace Renci.SshNet.Sftp +{ + internal class SftpFileReader + { + private const int MaxReadAhead = 10; + + private readonly byte[] _handle; + private readonly ISftpSession _sftpSession; + private SemaphoreLight _semaphore; + private bool _isCompleted; + private uint _chunkLength; + private int _readAheadChunkIndex; + private int _nextChunkIndex; + private ulong _readAheadOffset; + private ulong _offset; + private ulong _fileSize; + private readonly IDictionary _queue; + private readonly object _readLock; + + public SftpFileReader(byte[] handle, ISftpSession sftpSession) + { + _handle = handle; + _sftpSession = sftpSession; + _chunkLength = 16 * 1024; // TODO ! + _semaphore = new SemaphoreLight(MaxReadAhead); + _queue = new Dictionary(MaxReadAhead); + _readLock = new object(); + + _fileSize = (ulong)_sftpSession.RequestFStat(_handle).Size; + + ThreadAbstraction.ExecuteThread(() => + { + while (!_isCompleted) + { + if (_readAheadOffset >= _fileSize) + break; + + // TODO implement cancellation!? + _semaphore.Wait(); + + // start reading next chunk + _sftpSession.BeginRead(_handle, _readAheadOffset, _chunkLength, ReadCompleted, + new BufferedRead(_readAheadChunkIndex, _readAheadOffset)); + + // advance read-ahead offset + _readAheadOffset += _chunkLength; + + _readAheadChunkIndex++; + } + }); + } + + public byte[] Read() + { + lock (_readLock) + { + BufferedRead nextChunk; + + while (!_queue.TryGetValue(_nextChunkIndex, out nextChunk) && !_isCompleted) + Monitor.Wait(_readLock); + + if (_isCompleted) + return new byte[0]; + + if (nextChunk.Offset == _offset) + { + var data = nextChunk.Data; + _offset += (ulong) data.Length; + + // remove processed chunk + _queue.Remove(_nextChunkIndex); + // move to next chunk + _nextChunkIndex++; + // allow read ahead of a new chunk + _semaphore.Release(); + return data; + } + + // when the server returned less bytes than requested (for the previous chunk) + // we'll synchronously request the remaining data + + var catchUp = new byte[nextChunk.Offset - _offset]; + var bytesCaughtUp = 0L; + + while (bytesCaughtUp < catchUp.Length) + { + // TODO: break loop and interrupt blocking wait in case of exception + var read = _sftpSession.RequestRead(_handle, _offset, (uint) catchUp.Length); + if (read.Length == 0) + { + // break in loop in "read-ahead" thread (once a blocking wait is interrupted) + _isCompleted = true; + // interrupt blocking wait in "read-ahead" thread + lock (_readLock) + Monitor.PulseAll(_readLock); + // signal failure + throw new SshException("Unexpectedly reached end of file."); + } + + bytesCaughtUp += read.Length; + _offset += (ulong) bytesCaughtUp; + } + + return catchUp; + } + } + + private void ReadCompleted(IAsyncResult result) + { + var readAsyncResult = result as SftpReadAsyncResult; + if (readAsyncResult == null) + return; + + var data = readAsyncResult.EndInvoke(); + if (data.Length == 0) + { + _isCompleted = true; + } + else + { + var bufferedRead = (BufferedRead)readAsyncResult.AsyncState; + bufferedRead.Complete(data); + _queue.Add(bufferedRead.ChunkIndex, bufferedRead); + } + + // signal that a chunk has been read or EOF has been reached; + // in both cases, we want to unblock the "read-ahead" thread + lock (_readLock) + { + Monitor.Pulse(_readLock); + } + } + } + + //private class BufferedReadState + //{ + // private BlockingQueue _queue; + // private long _offset; + + // public BufferedReadState(long offset, SemaphoreLight semaphore, BlockingQueue queue) + // { + // _queue = queue; + // _offset = offset; + // _semaphore = semaphore; + // } + + // public long Offset + // { + // get { return _offset; } + // } + + // public BlockingQueue Queue + // { + // get { return _queue; } + // } + + // public SemaphoreLight Semaphore + // { + // get { return _semaphore; } + // } + //} + + //private class BlockingQueue + //{ + // private readonly object _lock = new object(); + // private Queue _queue; + // private bool _isClosed; + + // public BlockingQueue(int capacity) + // { + // _queue = new Queue(capacity); + // } + + // public T Dequeue() + // { + // lock (_lock) + // { + // while (!_isClosed && _queue.Count == 0) + // Monitor.Wait(_lock); + + // if (_queue.Count == 0) + // return default(T); + + // return _queue.Dequeue(); + // } + // } + + // public void Enqueue(T item) + // { + // lock (_lock) + // { + // _queue.Enqueue(item); + // Monitor.PulseAll(_lock); + // } + // } + + // public void Close() + // { + // _isClosed = true; + // Monitor.PulseAll(_lock); + // } + //} + + internal class BufferedRead + { + public int ChunkIndex { get; private set; } + + public byte[] Data { get; private set; } + + public ulong Offset { get; private set; } + + public BufferedRead(int chunkIndex, ulong offset) + { + ChunkIndex = chunkIndex; + Offset = offset; + } + + public void Complete(byte[] data) + { + Data = data; + } + } +} diff --git a/src/Renci.SshNet/Sftp/SftpReadAsyncResult.cs b/src/Renci.SshNet/Sftp/SftpReadAsyncResult.cs new file mode 100644 index 00000000..65911d22 --- /dev/null +++ b/src/Renci.SshNet/Sftp/SftpReadAsyncResult.cs @@ -0,0 +1,12 @@ +using Renci.SshNet.Common; +using System; + +namespace Renci.SshNet.Sftp +{ + internal class SftpReadAsyncResult : AsyncResult + { + public SftpReadAsyncResult(AsyncCallback asyncCallback, object state) : base(asyncCallback, state) + { + } + } +} diff --git a/src/Renci.SshNet/Sftp/SftpSession.cs b/src/Renci.SshNet/Sftp/SftpSession.cs index 8a3c6773..f00842a4 100644 --- a/src/Renci.SshNet/Sftp/SftpSession.cs +++ b/src/Renci.SshNet/Sftp/SftpSession.cs @@ -169,6 +169,8 @@ namespace Renci.SshNet.Sftp var offset = 0; var count = data.Length; + //Console.WriteLine("Data received: " + count); + // improve performance and reduce GC pressure by not buffering channel data if the received // chunk contains the complete packet data. // @@ -184,6 +186,8 @@ namespace Renci.SshNet.Sftp var packetTotalLength = packetDataLength + packetLengthByteCount; + //Console.WriteLine("Current: " + data.Length + " | Total: " + packetTotalLength); + // check if complete packet data (or more) is available if (count >= packetTotalLength) { @@ -241,6 +245,8 @@ namespace Renci.SshNet.Sftp var packetTotalLength = packetDataLength + packetLengthByteCount; + //Console.WriteLine("Current: " + data.Length + " | Total: " + packetTotalLength); + // check if complete packet data is available if (_data.Count < packetTotalLength) { @@ -397,6 +403,46 @@ namespace Renci.SshNet.Sftp } } + public SftpReadAsyncResult BeginRead(byte[] handle, ulong offset, uint length, AsyncCallback callback, object state) + { + var asyncResult = new SftpReadAsyncResult(callback, state); + + var request = new SftpReadRequest(ProtocolVersion, NextRequestId, handle, offset, length, + response => + { + asyncResult.SetAsCompleted(response.Data, false); + }, + response => + { + if (response.StatusCode != StatusCodes.Eof) + { + asyncResult.SetAsCompleted(GetSftpException(response), false); + } + else + { + asyncResult.SetAsCompleted(Array.Empty, false); + } + }); + SendRequest(request); + + return asyncResult; + } + + public byte[] EndRead(IAsyncResult asyncResult) + { + if (asyncResult == null) + throw new ArgumentNullException("asyncResult"); + + var sftpReadAsyncResult = asyncResult as SftpReadAsyncResult; + if (sftpReadAsyncResult == null) + throw new ArgumentException("IDIOT", "asyncResult"); + + if (sftpReadAsyncResult.EndInvokeCalled) + throw new InvalidOperationException("EndRead has already been called."); + + return sftpReadAsyncResult.EndInvoke(); + } + /// /// Performs SSH_FXP_READ request. /// diff --git a/src/Renci.SshNet/SftpClient.cs b/src/Renci.SshNet/SftpClient.cs index c98e3fff..dc04a5c8 100644 --- a/src/Renci.SshNet/SftpClient.cs +++ b/src/Renci.SshNet/SftpClient.cs @@ -1996,42 +1996,122 @@ namespace Renci.SshNet var fullPath = _sftpSession.GetCanonicalPath(path); var handle = _sftpSession.RequestOpen(fullPath, Flags.Read); - - ulong offset = 0; - - var optimalReadLength = _sftpSession.CalculateOptimalReadLength(_bufferSize); - - var data = _sftpSession.RequestRead(handle, offset, optimalReadLength); - - // Read data while available - while (data.Length > 0) + var async = true; + if (async) { - // Cancel download - if (asyncResult != null && asyncResult.IsDownloadCanceled) - break; + var fileReader = new SftpFileReader(handle, _sftpSession); + var totalBytesRead = 0UL; - output.Write(data, 0, data.Length); - - output.Flush(); - - offset += (ulong)data.Length; - - // Call callback to report number of bytes read - if (downloadCallback != null) + while (true) { - // copy offset to ensure it's not modified between now and execution of callback - var downloadOffset = offset; + var data = fileReader.Read(); + if (data.Length == 0) + break; - // Execute callback on different thread - ThreadAbstraction.ExecuteThread(() => { downloadCallback(downloadOffset); }); + output.Write(data, 0, data.Length); + + totalBytesRead += (ulong) data.Length; + + //Console.WriteLine(totalBytesRead); + + if (downloadCallback != null) + downloadCallback(totalBytesRead); } + } + else + { + ulong offset = 0; - data = _sftpSession.RequestRead(handle, offset, optimalReadLength); + var optimalReadLength = _sftpSession.CalculateOptimalReadLength(_bufferSize); + + var data = _sftpSession.RequestRead(handle, offset, optimalReadLength); + + // Read data while available + while (data.Length > 0) + { + // Cancel download + if (asyncResult != null && asyncResult.IsDownloadCanceled) + break; + + output.Write(data, 0, data.Length); + + output.Flush(); + + offset += (ulong)data.Length; + + //Console.WriteLine("" + data.Length + " => " + offset); + + // Call callback to report number of bytes read + if (downloadCallback != null) + { + // copy offset to ensure it's not modified between now and execution of callback + var downloadOffset = offset; + + // Execute callback on different thread + ThreadAbstraction.ExecuteThread(() => { downloadCallback(downloadOffset); }); + } + + data = _sftpSession.RequestRead(handle, offset, optimalReadLength); + } } _sftpSession.RequestClose(handle); } + private class BlockingQueue + { + private readonly object _lock = new object(); + private Queue _queue; + private bool _isClosed; + + public BlockingQueue(int capacity) + { + _queue = new Queue(capacity); + } + + public T Dequeue() + { + lock (_lock) + { + while (!_isClosed && _queue.Count == 0) + Monitor.Wait(_lock); + + if (_queue.Count == 0) + return default(T); + + return _queue.Dequeue(); + } + } + + public void Enqueue(T item) + { + lock (_lock) + { + _queue.Enqueue(item); + Monitor.PulseAll(_lock); + } + } + + public void Close() + { + _isClosed = true; + Monitor.PulseAll(_lock); + } + } + + private class BufferedRead + { + public byte[] Data { get; private set; } + + public ulong Offset { get; private set; } + + public BufferedRead(ulong offset, byte[] data) + { + Offset = offset; + Data = data; + } + } + /// /// Internals the upload file. /// diff --git a/src/Renci.SshNet/SubsystemSession.cs b/src/Renci.SshNet/SubsystemSession.cs index 5cb0e8a0..6bbd783b 100644 --- a/src/Renci.SshNet/SubsystemSession.cs +++ b/src/Renci.SshNet/SubsystemSession.cs @@ -42,7 +42,7 @@ namespace Renci.SshNet /// /// The channel associated with this session. /// - internal IChannelSession Channel + public IChannelSession Channel { get {