[c#] Handle non-seekable streams in InputStream

InputStream would always try to call Stream.Seek when it needed to skip
over some bytes in the underlying stream, even if the stream didn't
support Seek.

Now, it checks whether the stream can seek. If not, it makes dummy Read
calls to advance the underlying stream.

Regression test added, and all the implementations of IInputStream that
ship with Bond are tested against these tests.

Existing InputStream tests folded into these IInputStreams tests, and
tested against both seekable and non-seekable streams.

Fixes https://github.com/Microsoft/bond/issues/498

CHANGELOG.md

@@ -34,6 +34,16 @@ different versioning scheme, following the Haskell community's
 * Use Newtonsoft's JSON.NET BigInteger support -- when available -- to
   handle the full range of uint64 values in the SimpleJson protocol (.NET
   4.5 or greater, .NET Standard 1.6 or greater).
+* `Bond.IO.Unsafe.InputStream` can now be used with streams that do not
+  implement [`Stream.Seek`][msdn-stream-seek], like
+  [`System.IO.Compression.GzipStream`][msdn-gzipstream].
+  [Issue #498](https://github.com/Microsoft/bond/issues/498)
+    * Such streams are detected by inspecting
+      [`Stream.CanSeek`][msdn-stream-canseek].
+
+[msdn-gzipstream]: https://msdn.microsoft.com/en-us/library/system.io.compression.gzipstream(v=vs.110).aspx
+[msdn-stream-canseek]: https://msdn.microsoft.com/en-us/library/system.io.stream.canseek(v=vs.110).aspx
+[msdn-stream-seek]: https://msdn.microsoft.com/en-us/library/system.io.stream.seek(v=vs.110).aspx
 
 ## 6.0.0: 2017-06-29  ##
 * `gbc` & compiler library: 0.10.0.0

cs/src/io/unsafe/InputStream.cs

@@ -4,6 +4,7 @@
 namespace Bond.IO.Unsafe
 {
     using System;
+    using System.Diagnostics;
     using System.IO;
 
     /// <summary>
@@ -20,14 +21,16 @@ namespace Bond.IO.Unsafe
             get { return activeAllocationChunk; }
             set
             {
-                if (value < 0)
+                if (value <= 0)
                 {
-                    throw new ArgumentOutOfRangeException("value", "Value cannot be negative");
+                    throw new ArgumentOutOfRangeException(nameof(value), "Value must be positive.");
                 }
                 activeAllocationChunk = value;
             }
         }
 
+        static readonly byte[][] EmptyTempBuffers = new byte[0][];
+
         static int activeAllocationChunk;
 
         readonly Stream stream;
@@ -38,9 +41,9 @@ namespace Bond.IO.Unsafe
             ActiveAllocationChunk = DefaultAllocationChunk;
         }
 
-        // When we read more data from the stream we can override existing buffer
-        // only if it hasn't been exposed via ReadBytes or Clone. Otherwise a new
-        // buffer has to be allocated.
+        // When we read more data from the stream we can overwrite the
+        // existing buffer only if it hasn't been exposed via ReadBytes or
+        // Clone. Otherwise a new buffer has to be allocated.
         bool canReuseBuffer;
 
         public override long Length
@@ -93,30 +96,46 @@ namespace Bond.IO.Unsafe
 
         internal override void EndOfStream(int count)
         {
-            var oldBuffer = buffer;
-
+            // The unread bytes left in the buffer. May be negative, which
+            // indicates that this stream has been advanced beyond where we
+            // are in the underlying stream and some bytes will need to be
+            // skipped.
             var remaining = end - position;
 
-            if (remaining < 0)
-            {
-                stream.Seek(-remaining, SeekOrigin.Current);
-                remaining = 0;
-            }
-
             bool failed = false;
-            byte[][] tempBuffers = null;
-            int numTempBuffers = 0;
+            byte[][] tempBuffers = EmptyTempBuffers;
+
+            // Check whether we need to read in chunks to avoid allocating a
+            // ton of memory ahead of time.
             if ((count > buffer.Length && (count - buffer.Length > ActiveAllocationChunk)))
             {
-                // Calculate number of temp buffers -1 in case difference is exactly
-                // multiple of chunk size.
-                numTempBuffers = (count - buffer.Length - 1) / ActiveAllocationChunk;
+                // Calculate number of temp buffers; we round down since the
+                // last chunk is read directly into final buffer. Note:
+                // Difference is adjusted by -1 to round down correctly in
+                // cases where the difference is exactly a multiple of the
+                // allocation chunk size.
+                int numTempBuffers = (count - buffer.Length - 1) / ActiveAllocationChunk;
 
                 tempBuffers = new byte[numTempBuffers][];
-                for (int i = 0; i < numTempBuffers; i++)
+
+                for (int i = 0; i < tempBuffers.Length; i++)
                 {
                     tempBuffers[i] = new byte[ActiveAllocationChunk];
 
+                    if (remaining < 0)
+                    {
+                        // We need to skip ahead in the underlying stream.
+                        // Borrow the buffer to do the skipping before we do
+                        // the real read.
+
+                        // Only should happen for the first iteration, as we
+                        // reset remaining.
+                        Debug.Assert(i == 0);
+
+                        AdvanceUnderlyingStream(-remaining, tempBuffers[i]);
+                        remaining = 0;
+                    }
+
                     var bytesRead = stream.Read(tempBuffers[i], 0, ActiveAllocationChunk);
                     if (bytesRead != ActiveAllocationChunk)
                     {
@@ -128,28 +147,53 @@ namespace Bond.IO.Unsafe
 
             if (!failed)
             {
+                var oldBuffer = buffer;
+
                 if (!canReuseBuffer || count > buffer.Length)
                 {
                     buffer = new byte[Math.Max(bufferLength, count)];
                     canReuseBuffer = true;
                 }
 
+                int offset;
+
                 if (remaining > 0)
                 {
+                    // Copy any remaining bytes from the old buffer into the
+                    // final buffer. This may just move the bytes to the
+                    // beginning of the buffer.
                     Buffer.BlockCopy(oldBuffer, position, buffer, 0, remaining);
+                    offset = remaining;
                 }
-
-                int offset = remaining;
-                if (numTempBuffers > 0)
+                else if (remaining < 0)
                 {
-                    for (int i = 0; i < numTempBuffers; i++)
-                    {
-                        Buffer.BlockCopy(tempBuffers[i], 0, buffer, offset, ActiveAllocationChunk);
-                        offset += ActiveAllocationChunk;
-                    }
+                    // Nothing in the old buffer, but we need to skip ahead
+                    // in the underlying stream.
+                    AdvanceUnderlyingStream(-remaining, buffer);
+                    offset = 0;
+                }
+                else
+                {
+                    // The stars are aligned, so just start at the beginning
+                    // of the final buffer.
+                    offset = 0;
                 }
-                end = offset + stream.Read(buffer, offset, buffer.Length - offset);
 
+                // Copy from any temp buffers into the final buffer. In the
+                // common case, there are no temp buffers.
+                foreach (byte[] tempBuffer in tempBuffers)
+                {
+                    Buffer.BlockCopy(
+                        tempBuffer,
+                        0,
+                        buffer,
+                        offset,
+                        tempBuffer.Length);
+                    offset += tempBuffer.Length;
+                }
+
+                // Read the final block; update valid length and position.
+                end = offset + stream.Read(buffer, offset, buffer.Length - offset);
                 position = 0;
             }
 
@@ -158,5 +202,40 @@ namespace Bond.IO.Unsafe
                 base.EndOfStream(count - end);
             }
         }
+
+        /// <summary>
+        /// Advances the underlying stream by <paramref name="count"/> bytes.
+        /// </summary>
+        /// <remarks>Correctly handles streams that cannot Seek.</remarks>
+        /// <param name="count">The number of bytes to advance.</param>
+        /// <param name="scratchBuffer">
+        /// An already allocated buffer to use if dummy reads need to be
+        /// performed.
+        /// </param>
+        void AdvanceUnderlyingStream(int count, byte[] scratchBuffer)
+        {
+            Debug.Assert(scratchBuffer != null);
+
+            if (stream.CanSeek)
+            {
+                stream.Seek(count, SeekOrigin.Current);
+            }
+            else
+            {
+                while (count > 0)
+                {
+                    int bytesRead = stream.Read(
+                        scratchBuffer,
+                        offset: 0,
+                        count: Math.Min(scratchBuffer.Length, count));
+                    count -= bytesRead;
+
+                    if (bytesRead == 0)
+                    {
+                        base.EndOfStream(count);
+                    }
+                }
+            }
+        }
     }
 }

cs/test/core/StreamTests.cs

@@ -1,34 +1,77 @@
 namespace UnitTest
 {
+    using System;
     using System.IO;
+    using System.Runtime.InteropServices;
     using NUnit.Framework;
     using Bond;
+    using Bond.IO;
     using Bond.Protocols;
     using Bond.IO.Unsafe;
 
+    /// <summary>
+    /// Common test cases for IInputStream implementations. There's a
+    /// concrete implementation for each IInputStream implementation we want
+    /// to test.
+    /// </summary>
     [TestFixture]
-    public class StreamTests
+    public abstract class IInputStreamTestsBase<TInputStream>
+        where TInputStream : IInputStream, ICloneable<TInputStream>
     {
-
-        // Restore the default settings at the start and end of each test
-        [SetUp]
-        [TearDown]
-        public void RestoreDefaults()
+        protected TInputStream MakeInputStream(params byte[] buffer)
         {
-            InputStream.ActiveAllocationChunk = InputStream.DefaultAllocationChunk;
+            return MakeInputStream(new ArraySegment<byte>(buffer));
+        }
+
+        /// <summary>
+        /// Creates an <typeparamref name="TInputStream" /> over the given bytes.
+        /// </summary>
+        protected abstract TInputStream MakeInputStream(ArraySegment<byte> buffer);
+
+        // Not all implementations use an internal buffer, but for those that
+        // do, it needs to be this large.
+        protected const int InternalBufferSize = 9;
+
+        [Test]
+        public void Skip_AdvancesPosition()
+        {
+            var inputStream = MakeInputStream(0, 1, 2);
+
+            inputStream.SkipBytes(2);
+            Assert.AreEqual(2, inputStream.Position);
+
+            Assert.AreEqual(2, inputStream.ReadUInt8());
         }
 
         [Test]
-        public void StreamPositionLengthTest()
+        public void SkipBeyondInternalBufferSize_SeeksCorrectly()
         {
-            StreamPositionLengthTestImpl();
-            InputStream.ActiveAllocationChunk = 8;
-            StreamPositionLengthTestImpl();
+            var buf = MakeSequential(InternalBufferSize*3 + 1);
+
+            var inputStream = MakeInputStream(buf);
+            Assert.AreEqual(0, inputStream.ReadUInt8());
+            inputStream.SkipBytes(InternalBufferSize);
+            var blob = inputStream.ReadBytes(InternalBufferSize*2);
+            for (int j = 0; j < blob.Count; ++j)
+            {
+                byte expectedValue = (byte) ((j + InternalBufferSize + 1)%256);
+                Assert.AreEqual(expectedValue, blob.Array[blob.Offset + j]);
+            }
         }
 
-        internal void StreamPositionLengthTestImpl()
+        [Test]
+        public void SkipBeyondEndOfStream_ThrowsOnNextRead()
         {
-            const int _50MB = 50*1024*1024;
+            var buf = MakeSequential(InternalBufferSize*3);
+            var inputStream = MakeInputStream(buf);
+            inputStream.SkipBytes(InternalBufferSize*3 + 1);
+            Assert.Throws<EndOfStreamException>(() => inputStream.ReadUInt8());
+        }
+
+        [Test]
+        public void Stream_PositionLength_AsExpected()
+        {
+            const int _50MB = 50 * 1024 * 1024;
 
             var from1 = Random.Init<Containers>();
             var from2 = Random.Init<Containers>();
@@ -57,8 +100,8 @@
 
             stream.Position = 0;
 
-            var input = new InputStream(stream);
-            var reader = new CompactBinaryReader<InputStream>(input);
+            var input = MakeInputStream(stream.ToArray());
+            var reader = new CompactBinaryReader<TInputStream>(input);
 
             Assert.IsTrue(input.Position == stream.Position);
             Assert.IsTrue(input.Length == stream.Length);
@@ -77,72 +120,51 @@
         }
 
         [Test]
-        public void StreamBufferReuseTest()
+        public void ReadBytes_DifferentSizesAndPositions_ReadCorrectly()
         {
-            StreamBufferReuseTestImpl();
-            InputStream.ActiveAllocationChunk = 8;
-            StreamBufferReuseTestImpl();
-            InputStream.ActiveAllocationChunk = 2;
-            StreamBufferReuseTestImpl();
-        }
-
-        internal void StreamBufferReuseTestImpl()
-        {
-            var buffer = new byte[5 * 1024];
-            
-            for (var i = 0; i < buffer.Length; ++i)
-                buffer[i] = (byte)(i % 256);
+            var buffer = MakeSequential(5 * 1024);
 
             for (var k = 3; k < 20; ++k)
             {
-                var stream = new MemoryStream(buffer, 0, buffer.Length, false, true);
-                var input = new InputStream(stream, 9);
-                
+                var input = MakeInputStream(buffer);
+
                 while (input.Position + k + sizeof(long) < input.Length)
                 {
                     var x = input.Position;
                     var bytes = input.ReadBytes(k);
                     input.ReadUInt64();
                     for (var j = 0; j < bytes.Count; ++j)
-                        Assert.AreEqual(bytes.Array[bytes.Offset + j], (x + j) % 256);
+                    {
+                        Assert.AreEqual((x + j)%256, bytes.Array[bytes.Offset + j]);
+                    }
                 }
             }
         }
 
-
-        delegate void IntTest<T>(T value);
-
         [Test]
-        public void VarIntTest()
+        public void VarInts_RoundTrip()
         {
-            VarIntTestImpl();
-            InputStream.ActiveAllocationChunk = 8;
-            VarIntTestImpl();
-        }
-
-        internal void VarIntTestImpl()
-        {
-            IntTest<ushort> test16 = (value) =>
+            Action<ushort> test16 = value =>
             {
                 var output = new OutputBuffer();
                 output.WriteVarUInt16(value);
-                var input = new InputBuffer(output.Data);
+                var input = MakeInputStream(output.Data);
                 Assert.AreEqual(value, input.ReadVarUInt16());
             };
 
-            IntTest<uint> test32 = (value) =>
+            Action<uint> test32 = value =>
             {
                 var output = new OutputBuffer();
                 output.WriteVarUInt32(value);
-                var input = new InputBuffer(output.Data);
+                var input = MakeInputStream(output.Data);
                 Assert.AreEqual(value, input.ReadVarUInt32());
             };
 
-            IntTest<ulong> test64 = (value) =>
+            Action<ulong> test64 = value =>
             {
                 var output = new OutputBuffer();
                 output.WriteVarUInt64(value);
-                var input = new InputBuffer(output.Data);
+                var input = MakeInputStream(output.Data);
                 Assert.AreEqual(value, input.ReadVarUInt64());
             };
 
@@ -153,5 +175,204 @@
             test64(ulong.MinValue);
             test64(ulong.MaxValue);
         }
+
+        protected static ArraySegment<byte> EmbedBuffer(ArraySegment<byte> buffer)
+        {
+            if (buffer.Offset != 0 && buffer.Count != buffer.Array.Length)
+            {
+                // it's already embedded in something else
+                return buffer;
+            }
+
+            var largerBuffer = new byte[buffer.Count + 5];
+            largerBuffer[0] = 255;
+            largerBuffer[1] = 255;
+            largerBuffer[2] = 255;
+            Buffer.BlockCopy(
+                src: buffer.Array,
+                srcOffset: buffer.Offset,
+                dst: largerBuffer,
+                dstOffset: 3,
+                count: buffer.Count);
+            largerBuffer[largerBuffer.Length - 2] = 128;
+            largerBuffer[largerBuffer.Length - 1] = 128;
+
+            return new ArraySegment<byte>(largerBuffer, offset: 3, count: buffer.Count);
+        }
+
+        protected static byte[] MakeSequential(int length)
+        {
+            Assert.That(length >= 0);
+
+            var buf = new byte[length];
+            for (int i = 0; i < length; ++i)
+            {
+                buf[i] = (byte) (i%256);
+            }
+
+            return buf;
+        }
+    }
+
+    public class UnsafeInputBufferTests : IInputStreamTestsBase<Bond.IO.Unsafe.InputBuffer>
+    {
+        protected override Bond.IO.Unsafe.InputBuffer MakeInputStream(ArraySegment<byte> buffer)
+        {
+            return new Bond.IO.Unsafe.InputBuffer(buffer);
+        }
+    }
+
+    public class UnsafeInputBufferOffsetTests : IInputStreamTestsBase<Bond.IO.Unsafe.InputBuffer>
+    {
+        protected override Bond.IO.Unsafe.InputBuffer MakeInputStream(ArraySegment<byte> buffer)
+        {
+            return new Bond.IO.Unsafe.InputBuffer(EmbedBuffer(buffer));
+        }
+    }
+
+    public class SafeInputBufferTests : IInputStreamTestsBase<Bond.IO.Safe.InputBuffer>
+    {
+        protected override Bond.IO.Safe.InputBuffer MakeInputStream(ArraySegment<byte> buffer)
+        {
+            return new Bond.IO.Safe.InputBuffer(buffer);
+        }
+    }
+
+    public class SafeInputBufferOffsetTests : IInputStreamTestsBase<Bond.IO.Safe.InputBuffer>
+    {
+        protected override Bond.IO.Safe.InputBuffer MakeInputStream(ArraySegment<byte> buffer)
+        {
+            return new Bond.IO.Safe.InputBuffer(EmbedBuffer(buffer));
+        }
+    }
+
+    public class InputPointerTests : IInputStreamTestsBase<InputPointer>
+    {
+        GCHandle? pinnedBuffer;
+
+        [TearDown]
+        public void TearDown()
+        {
+            pinnedBuffer?.Free();
+            pinnedBuffer = null;
+        }
+
+        protected override InputPointer MakeInputStream(ArraySegment<byte> buffer)
+        {
+            // there may already be a pinnedBuffer, so be sure to free it
+            pinnedBuffer?.Free();
+
+            pinnedBuffer = GCHandle.Alloc(buffer.Array, GCHandleType.Pinned);
+            return new InputPointer(
+                pinnedBuffer.Value.AddrOfPinnedObject() + buffer.Offset,
+                buffer.Count);
+        }
+    }
+
+    public abstract class InputStreamTestsBase : IInputStreamTestsBase<InputStream>
+    {
+        // Restore the default settings at the start and end of each test
+        [SetUp]
+        [TearDown]
+        public void RestoreDefaults()
+        {
+            InputStream.ActiveAllocationChunk = InputStream.DefaultAllocationChunk;
+        }
+
+        [Test]
+        public void SkipBeyondInternalBufferSize_AllocationChunk_SeeksCorrectly()
+        {
+            InputStream.ActiveAllocationChunk = 8;
+            SkipBeyondInternalBufferSize_SeeksCorrectly();
+            InputStream.ActiveAllocationChunk = 2;
+            SkipBeyondInternalBufferSize_SeeksCorrectly();
+        }
+
+        [Test]
+        public void SkipBeyondEndOfStream_AllocationChunk_ThrowsOnNextRead()
+        {
+            InputStream.ActiveAllocationChunk = 8;
+            SkipBeyondEndOfStream_ThrowsOnNextRead();
+            InputStream.ActiveAllocationChunk = 2;
+            SkipBeyondEndOfStream_ThrowsOnNextRead();
+        }
+
+        [Test]
+        public void Stream_PositionLength_WithAllocationChunk_AsExpected()
+        {
+            InputStream.ActiveAllocationChunk = 8;
+            Stream_PositionLength_AsExpected();
+            InputStream.ActiveAllocationChunk = 2;
+            Stream_PositionLength_AsExpected();
+        }
+
+        [Test]
+        public void ReadBytes_DifferentSizesAndOffsets_WithAllocationChunk_ReadCorrectly()
+        {
+            InputStream.ActiveAllocationChunk = 8;
+            ReadBytes_DifferentSizesAndPositions_ReadCorrectly();
+            InputStream.ActiveAllocationChunk = 2;
+            ReadBytes_DifferentSizesAndPositions_ReadCorrectly();
+        }
+
+        [Test]
+        public void VarInts_With_AllocationChunk_RoundTrip()
+        {
+            InputStream.ActiveAllocationChunk = 8;
+            VarInts_RoundTrip();
+            InputStream.ActiveAllocationChunk = 2;
+            VarInts_RoundTrip();
+        }
+
+
+        [Test]
+        public void BadActiveAllocationChunkValues_Throws()
+        {
+            Assert.Throws<ArgumentOutOfRangeException>(() => InputStream.ActiveAllocationChunk = 0);
+            Assert.Throws<ArgumentOutOfRangeException>(() => InputStream.ActiveAllocationChunk = -1);
+        }
+    }
+
+    public class InputStreamTests : InputStreamTestsBase
+    {
+        protected override InputStream MakeInputStream(ArraySegment<byte> buffer)
+        {
+            var ms = new MemoryStream(
+                buffer.Array,
+                buffer.Offset,
+                buffer.Count,
+                writable: false,
+                publiclyVisible: true);
+            return new InputStream(ms, InternalBufferSize);
+        }
+    }
+
+    public class NonSeekableInputStreamTests : InputStreamTestsBase
+    {
+        protected override InputStream MakeInputStream(ArraySegment<byte> buffer)
+        {
+            var ms = new NonSeekableMemoryStream(buffer);
+            return new InputStream(ms, InternalBufferSize);
+        }
+
+        private class NonSeekableMemoryStream : MemoryStream
+        {
+            public NonSeekableMemoryStream(ArraySegment<byte> buffer)
+                : base(buffer.Array,
+                    buffer.Offset,
+                    buffer.Count,
+                    writable: false,
+                    publiclyVisible: true)
+            {
+            }
+
+            public override bool CanSeek => false;
+
+            public override long Seek(long offset, SeekOrigin loc)
+            {
+                throw new NotImplementedException(
+                    $"{nameof(NonSeekableMemoryStream)} cannot Seek.");
+            }
+        }
     }
 }