RFC: Rewrite PipeStream and SshCommand stream handling

src/Renci.SshNet/Common/LinkedListQueue.cs

@@ -0,0 +1,141 @@
+namespace Renci.SshNet.Common
+{
+    using System;
+    using System.Threading;
+
+    /// <summary>
+    /// Fast concurrent generic linked list queue.
+    /// </summary>
+    internal class LinkedListQueue<T> : IDisposable
+    {
+        sealed class Entry<E>
+        {
+            public E Item;
+            public Entry<E> Next;
+        }
+
+        private readonly object _lock = new object();
+
+        private Entry<T> _first;
+        private Entry<T> _last;
+
+        private bool _isAddingCompleted;
+
+        /// <summary>
+        /// Gets whether this <see cref="T:Renci.SshNet.Common.LinkedListQueue`1"/> has been marked as complete for adding and is empty.
+        /// </summary>
+        /// <value>Whether this queue has been marked as complete for adding and is empty.</value>
+        public bool IsCompleted
+        {
+            get { return _isAddingCompleted && _first == null && _last == null; }
+        }
+
+        /// <summary>
+        /// Gets whether this <see cref="T:Renci.SshNet.Common.LinkedListQueue`1"/> has been marked as complete for adding.
+        /// </summary>
+        /// <value>Whether this queue has been marked as complete for adding.</value>
+        public bool IsAddingCompleted
+        {
+            get { return _isAddingCompleted; }
+            set
+            {
+                lock (_lock)
+                {
+                    _isAddingCompleted = value;
+                }
+            }
+        }
+
+        /// <summary>
+        /// Adds the item to <see cref="T:Renci.SshNet.Common.LinkedListQueue`1"/>.
+        /// </summary>
+        /// <param name="item">The item to be added to the queue. The value can be a null reference.</param>
+        public void Add(T item)
+        {
+            lock (_lock)
+            {
+                if (_isAddingCompleted)
+                    return;
+
+                var entry = new Entry<T>();
+                entry.Item = item;
+
+                if (_last != null)
+                {
+                    _last.Next = entry;
+                }
+
+                _last = entry;
+
+                if (_first == null)
+                {
+                    _first = entry;
+                }
+
+                Monitor.PulseAll(_lock);
+            }
+        }
+
+        /// <summary>
+        /// Marks the <see cref="T:Renci.SshNet.Common.LinkedListQueue`1"/> instances as not accepting any more additions.
+        /// </summary>
+        public void CompleteAdding()
+        {
+            lock (_lock)
+            {
+                IsAddingCompleted = true;
+                Monitor.PulseAll(_lock);
+            }
+        }
+
+        /// <summary>
+        /// Tries to remove an item from the <see cref="T:Renci.SshNet.Common.LinkedListQueue`1"/>.
+        /// </summary>
+        /// <returns><c>true</c>, if an item could be removed; otherwise <c>false</c>.</returns>
+        /// <param name="item">The item to be removed from the queue.</param>
+        /// <param name="wait">Wait for data or fail immediately if empty.</param>
+        public bool TryTake(out T item, bool wait)
+        {
+            lock (_lock)
+            {
+                if (_first == null && !wait)
+                {
+                    item = default(T);
+                    return false;
+                }
+
+                while (_first == null && !_isAddingCompleted)
+                    Monitor.Wait(_lock);
+
+                if (_first == null && _isAddingCompleted)
+                {
+                    item = default(T);
+                    return false;
+                }
+
+                item = _first.Item;
+                _first = _first.Next;
+                return true;
+            }
+        }
+
+        /// <summary>
+        /// Releases all resource used by the <see cref="T:Renci.SshNet.Common.LinkedListQueue`1"/> object.
+        /// </summary>
+        /// <remarks>Call <see cref="Dispose"/> when you are finished using the
+        /// <see cref="T:Renci.SshNet.Common.LinkedListQueue`1"/>. The <see cref="Dispose"/> method leaves the
+        /// <see cref="T:Renci.SshNet.Common.LinkedListQueue`1"/> in an unusable state. After calling
+        /// <see cref="Dispose"/>, you must release all references to the
+        /// <see cref="T:Renci.SshNet.Common.LinkedListQueue`1"/> so the garbage collector can reclaim the memory that
+        /// the <see cref="T:Renci.SshNet.Common.LinkedListQueue`1"/> was occupying.</remarks>
+        public void Dispose()
+        {
+            lock (_lock)
+            {
+                _first = null;
+                _last = null;
+                _isAddingCompleted = true;
+            }
+        }
+    }
+}

src/Renci.SshNet/Common/Pipe.cs

@@ -0,0 +1,47 @@
+namespace Renci.SshNet.Common
+{
+    using System;
+    using System.IO;
+
+    /// <summary>
+    /// A generic pipe to pass through data.
+    /// </summary>
+    internal class Pipe : IDisposable
+    {
+        private readonly LinkedListQueue<byte[]> _queue;
+
+        /// <summary>
+        /// Gets the input stream.
+        /// </summary>
+        /// <value>The input stream.</value>
+        public Stream InputStream { get; private set; }
+
+        /// <summary>
+        /// Gets the output stream.
+        /// </summary>
+        /// <value>The output stream.</value>
+        public Stream OutputStream { get; private set; }
+
+        public Pipe()
+        {
+            _queue = new LinkedListQueue<byte[]>();
+            InputStream = new PipeInputStream(_queue);
+            OutputStream = new PipeOutputStream(_queue);
+        }
+
+        /// <summary>
+        /// Releases all resource used by the <see cref="T:Renci.SshNet.Common.Pipe"/> object.
+        /// </summary>
+        /// <remarks>Call <see cref="Dispose"/> when you are finished using the <see cref="T:Renci.SshNet.Common.Pipe"/>. The
+        /// <see cref="Dispose"/> method leaves the <see cref="T:Renci.SshNet.Common.Pipe"/> in an unusable state. After
+        /// calling <see cref="Dispose"/>, you must release all references to the
+        /// <see cref="T:Renci.SshNet.Common.Pipe"/> so the garbage collector can reclaim the memory that the
+        /// <see cref="T:Renci.SshNet.Common.Pipe"/> was occupying.</remarks>
+        public void Dispose()
+        {
+            OutputStream.Dispose();
+            InputStream.Dispose();
+            _queue.Dispose();
+        }
+    }
+}

src/Renci.SshNet/Common/PipeInputStream.cs

@@ -0,0 +1,122 @@
+namespace Renci.SshNet.Common 
+{
+    using System;
+    using System.IO;
+
+    internal class PipeInputStream : Stream
+    {
+        private LinkedListQueue<byte[]> _queue;
+        private byte[] _current;
+        private int _currentPosition;
+        private bool _isDisposed;
+
+        public PipeInputStream(LinkedListQueue<byte[]> queue)
+        {
+            _queue = queue;
+        }
+
+        public override void Flush()
+        {
+        }
+
+        public override long Seek(long offset, SeekOrigin origin)
+        {
+            throw new NotSupportedException();
+        }
+
+        public override void SetLength(long value)
+        {
+            throw new NotSupportedException();
+        }
+
+        public override int Read(byte[] buffer, int offset, int count)
+        {
+            if (buffer == null)
+                throw new ArgumentNullException("buffer");
+            if (offset + count > buffer.Length)
+                throw new ArgumentException("The sum of offset and count is greater than the buffer length.");
+            if (offset < 0 || count < 0)
+                throw new ArgumentOutOfRangeException("offset", "offset or count is negative.");
+            if (_isDisposed)
+                throw CreateObjectDisposedException();
+
+            var bytesRead = 0;
+
+            while (bytesRead < count)
+            {
+                if (_current == null || _currentPosition == _current.Length)
+                {
+                    if (!_queue.TryTake(out _current, (bytesRead == 0)))
+                    {
+                        _current = null;
+                        return bytesRead;
+                    }
+
+                    _currentPosition = 0;
+                }
+
+                var toRead = _current.Length - _currentPosition;
+                if (toRead > count - bytesRead)
+                    toRead = count - bytesRead;
+
+                Buffer.BlockCopy(_current, _currentPosition, buffer, offset + bytesRead, toRead);
+
+                _currentPosition += toRead;
+                bytesRead += toRead;
+            }
+
+            return bytesRead;
+        }
+
+        public override void Write(byte[] buffer, int offset, int count)
+        {
+            throw new NotSupportedException();
+        }
+
+        public override bool CanRead
+        {
+            get { return !_isDisposed; }
+        }
+
+        public override bool CanSeek
+        {
+            get { return false; }
+        }
+
+        public override bool CanWrite
+        {
+            get { return false; }
+        }
+
+        public override long Length
+        {
+            get
+            {
+                throw new NotSupportedException();
+            }
+        }
+
+        public override long Position
+        {
+            get
+            {
+                throw new NotSupportedException();
+            }
+            set
+            {
+                throw new NotSupportedException();
+            }
+        }
+
+        protected override void Dispose(bool disposing)
+        {
+            base.Dispose(disposing);
+            _isDisposed = true;
+        }
+
+        private ObjectDisposedException CreateObjectDisposedException()
+        {
+            return new ObjectDisposedException(GetType().FullName);
+        }
+    }
+}