Rewrite streamwrapper to fix bugs (I forgot which)

ReFuel.Gltf/GltfDocument.cs

@@ -133,7 +133,7 @@ namespace ReFuel.Gltf
             throw new Exception("File does not have a valid GLB header.");
         }
 
-        private static GltfDocument OpenBinary(StreamWrapper str, in GlbHeader header)
+        private static GltfDocument OpenBinary(StreamWrapper fileStream, in GlbHeader header)
         {
             if (header.Version < 2)
             {
@@ -142,24 +142,23 @@ namespace ReFuel.Gltf
 
             JsonDocument? document = null;
             List<StreamWrapper> binaryStreams = new List<StreamWrapper>();
-
+            while (GlbChunk.Read(fileStream, out GlbChunk chunk))
-            while (GlbChunk.Read(str, out GlbChunk chunk))
             {
-                using StreamWrapper substream = str.ForkAbs(str.Position, str.Position + chunk.Size, true);
+                using StreamWrapper chunkStream = fileStream.ForkAbs(fileStream.Position, fileStream.Position + chunk.Size, true);
-
                 if (chunk.Type == GlbChunkType.Json)
                 {
-                    document = JsonDocument.Parse(substream.Fork(keepOpen:true));
+                    document = JsonDocument.Parse(chunkStream);
                 }
                 else
                 {
                     Stream copy = new MemoryStream();
-
+                    //TODO: is this reasonable?
-                    substream.CopyTo(copy);
+                    chunkStream.CopyTo(copy);
                     copy.Position = 0;
 
                     binaryStreams.Add(new StreamWrapper(copy));
                 }
+                fileStream.Seek(chunk.Size, SeekOrigin.Current);
             }
 
             if (document == null)

ReFuel.Gltf/IO/StreamWrapper.cs

@@ -4,213 +4,327 @@ namespace ReFuel.Gltf.IO
 {
     public class StreamWrapper : Stream
     {
-        /// <summary>
+        private readonly Stream _baseStream;
-        /// The private synchronized proxy for the stream.
-        /// </summary>
-        private readonly Stream _proxy;
 
-        /// <summary>
+        // Shared across forks so that concurrent access to the same base stream is serialized.
-        /// Head pointer for stream r/w operations.
+        private readonly object _sync;
-        /// </summary>
-        public long Head { get; private init; } = 0;
 
-        /// <summary>
+        // Logical position relative to Head. This is the fix for the "Position lies" problem.
-        /// Tail pointer for stream r/w operations. -1 signifies unbound.
+        private long _position;
-        /// </summary>
-        public long Tail { get; private init; } = -1;
-
-        /// <summary>
-        /// The number of bytes in the stream if there is a limit set.
-        /// </summary>
-        public long Limit
-        {
-            get
-            {
-                if (Tail <= 0)
-                {
-                    return -1;
-                }
-
-                if (CanWrite)
-                {
-                    return Tail - Head;
-                }
-
-                // Return the limit size or however many bytes are available.
-                return Math.Min(Tail - Head, _proxy.Length - Head);
-            }
-        }
 
+        public long Head { get; private init; }
+        public long Tail { get; private set; } = -1; // -1 = unbound
         /// <summary>
         /// The base stream for this wrapper.
         /// </summary>
-        public Stream BaseStream { get; }
+        public Stream BaseStream => _baseStream;
-
         /// <summary>
         /// True if the stream has been set to be read only.
         /// </summary>
-        public bool IsReadOnly { get; private init; } = false;
+        public bool IsReadOnly { get; private init; }
-
         /// <summary>
         /// True if the base stream should be kept open.
         /// </summary>
         public bool KeepOpen { get; }
 
-        public override bool CanRead => _proxy.CanRead;
+        public long Limit
-        public override bool CanSeek => _proxy.CanSeek;
+        {
-        public override bool CanWrite => !IsReadOnly || _proxy.CanWrite;
+            get
-        public override long Length => Tail < 0 ? _proxy.Length - Head : Math.Min(Tail - Head, _proxy.Length - Head);
+            {
+                if (Tail < 0)
+                {
+                    return -1;
+                }
+
+                long end = Math.Min(Tail, _baseStream.Length);
+                return Math.Max(0, end - Head);
+            }
+        }
+
+        public override bool CanRead => _baseStream.CanRead;
+        public override bool CanSeek => _baseStream.CanSeek;
+        public override bool CanWrite => !IsReadOnly && _baseStream.CanWrite;
+
+        public override long Length
+        {
+            get
+            {
+                long end = Tail >= 0 ? Math.Min(Tail, _baseStream.Length) : _baseStream.Length;
+                return Math.Max(0, end - Head);
+            }
+        }
 
         public override long Position
         {
-            get => ToRelOffset(_proxy.Position);
+            get { lock (_sync)
-            set => _proxy.Position = ToAbsOffset(value, true);
+                {
+                    return _position;
+                }
+            }
+            set { lock (_sync)
+                {
+                    Seek(value, SeekOrigin.Begin);
+                }
+            }
         }
 
         public StreamWrapper(Stream baseStream, bool keepOpen = false, bool isReadOnly = false)
         {
-            BaseStream = baseStream;
+            _baseStream = baseStream ?? throw new ArgumentNullException(nameof(baseStream));
             KeepOpen = keepOpen;
             IsReadOnly = isReadOnly;
-
+            _sync = new object();
-            _proxy = Synchronized(baseStream);
+            Head = 0;
+            Tail = -1;
+            _position = 0;
         }
 
-        public override void Flush() => _proxy.Flush();
+        private StreamWrapper(Stream baseStream, object sync,
+                              long head, long tail,
+                              bool keepOpen, bool isReadOnly)
+        {
+            _baseStream = baseStream;
+            _sync = sync;
+            Head = head;
+            Tail = tail;
+            KeepOpen = keepOpen;
+            IsReadOnly = isReadOnly;
+            _position = 0;
+        }
+
+        public override void Flush()
+        {
+            lock (_sync)
+            {
+                _baseStream.Flush();
+            }
+        }
 
         public override int Read(byte[] buffer, int offset, int count)
         {
-            int clamped = ClampReadLength();
+            ValidateBufferArgs(buffer, offset, count);
-            int rc =_proxy.Read(buffer, offset, clamped);
-            Debug.Assert(Tail < 0 || _proxy.Position <= Tail);
-            return rc;
 
-            int ClampReadLength()
+            lock (_sync)
             {
-                if (Tail > 0)
+                int clamped = ClampReadLength(count);
+                if (clamped == 0)
                 {
-                    return Math.Min(count, (int)Math.Min(int.MaxValue, Math.Max(0, Tail - _proxy.Position)));
+                    return 0;
-                }
-                else
-                {
-                    return count;
                 }
+
+                _baseStream.Seek(Head + _position, SeekOrigin.Begin);
+                int rc = _baseStream.Read(buffer, offset, clamped);
+                _position += rc;
+                return rc;
             }
         }
 
+        public override int Read(Span<byte> buffer)
+        {
+            lock (_sync)
+            {
+                int clamped = ClampReadLength(buffer.Length);
+                if (clamped == 0)
+                {
+                    return 0;
+                }
+
+                _baseStream.Seek(Head + _position, SeekOrigin.Begin);
+                int rc = _baseStream.Read(buffer.Slice(0, clamped));
+                _position += rc;
+                return rc;
+            }
+        }
+
+        public override int ReadByte()
+        {
+            lock (_sync)
+            {
+                if (ClampReadLength(1) == 0)
+                {
+                    return -1;
+                }
+
+                _baseStream.Seek(Head + _position, SeekOrigin.Begin);
+                int b = _baseStream.ReadByte();
+                if (b >= 0)
+                {
+                    _position++;
+                }
+
+                return b;
+            }
+        }
+
+        private int ClampReadLength(int count)
+        {
+            if (Tail < 0)
+            {
+                return count;
+            }
+
+            long remaining = Tail - (Head + _position);
+            if (remaining <= 0)
+            {
+                return 0;
+            }
+
+            return (int)Math.Min(count, remaining);
+        }
 
         public override long Seek(long offset, SeekOrigin origin)
         {
-            long absOffset = 0;
+            lock (_sync)
-            switch (origin)
             {
-                case SeekOrigin.Begin:
+                long newRel;
-                    absOffset = ToAbsOffset(offset, true);
+                switch (origin)
-                    break;
+                {
-                case SeekOrigin.End:
+                    case SeekOrigin.Begin:
-                    if (Tail < 0)
+                        newRel = offset;
-                    {
+                        break;
-                        try
-                        {
-                            return ToRelOffset(_proxy.Seek(offset, origin), true);
-                        }
-                        catch (ArgumentOutOfRangeException)
-                        {
-                            // Set the stream back to a known state.
-                            _proxy.Seek(Tail, SeekOrigin.Begin);
-                            throw;
-                        }
-                    }
 
-                    absOffset = Math.Max(Head, Math.Min(Tail, Tail + offset));
+                    case SeekOrigin.Current:
-                    break;
+                        newRel = _position + offset;
-                case SeekOrigin.Current:
+                        break;
-                    absOffset = Math.Max(Head, Math.Min(_proxy.Position + offset, Tail));
+
-                    break;
+                    case SeekOrigin.End:
+                        // End of our window, not necessarily end of the underlying stream.
+                        long windowEndRel = Tail >= 0
+                            ? Tail - Head
+                            : _baseStream.Length - Head;
+                        newRel = windowEndRel + offset;
+                        break;
+
+                    default:
+                        throw new ArgumentOutOfRangeException(nameof(origin));
+                }
+
+                if (newRel < 0)
+                {
+                    throw new IOException("Cannot seek before the beginning of the stream.");
+                }
+
+                long max = Length;
+                if (Tail >= 0 && newRel > max)
+                {
+                    throw new IOException("Cannot seek past the end of the bounded stream.");
+                }
+
+                _position = newRel;
+                return _position;
+            }
+        }
+
+        public override void Write(byte[] buffer, int offset, int count)
+        {
+            ValidateBufferArgs(buffer, offset, count);
+
+            if (IsReadOnly)
+            {
+                throw new NotSupportedException("This stream is read only.");
+            }
+
+            lock (_sync)
+            {
+                if (Tail >= 0 && Head + _position + count > Tail)
+                {
+                    throw new NotSupportedException("Write would overrun the stream limit.");
+                }
+
+                _baseStream.Seek(Head + _position, SeekOrigin.Begin);
+                _baseStream.Write(buffer, offset, count);
+                _position += count;
+            }
+        }
+
+        public override void Write(ReadOnlySpan<byte> buffer)
+        {
+            if (IsReadOnly)
+            {
+                throw new NotSupportedException("This stream is read only.");
+            }
+
+            lock (_sync)
+            {
+                if (Tail >= 0 && Head + _position + buffer.Length > Tail)
+                {
+                    throw new NotSupportedException("Write would overrun the stream limit.");
+                }
+
+                _baseStream.Seek(Head + _position, SeekOrigin.Begin);
+                _baseStream.Write(buffer);
+                _position += buffer.Length;
             }
-            return ToRelOffset(_proxy.Seek(absOffset, SeekOrigin.Begin));
         }
 
         public override void SetLength(long value)
         {
             if (IsReadOnly)
-                throw new Exception("This stream is read only.");
-
-            if (Tail < 0)
-                throw new NotSupportedException("Cannot expand the stream at this point.");
-
-            if (Tail != _proxy.Length)
-                throw new NotSupportedException("Cannot resize the stream at this point.");
-
-            _proxy.SetLength(Head + value);
-        }
-
-        public override void Write(byte[] buffer, int offset, int count)
-        {
-            if (IsReadOnly)
-                throw new Exception("This stream is read only.");
-
-            if (Tail > 0 && _proxy.Position + count >= Tail)
-                throw new Exception("Source array will overrun the current limit.");
-
-            _proxy.Write(buffer, offset, count);
-        }
-
-        public StreamWrapper Fork(long head = 0, long tail = -1, bool keepOpen = false, bool isReadOnly = false)
-        {
-            return ForkAbs(ToAbsOffset(head, true), tail < 0 ? Tail : ToAbsOffset(tail, true), keepOpen, isReadOnly);
-        }
-
-        public StreamWrapper ForkAbs(long head = 0, long tail = -1, bool keepOpen = false, bool isReadOnly = false)
-        {
-            if (head < 0 || head > BaseStream.Length)
-                throw new ArgumentOutOfRangeException(nameof(head), "Head pointer out of range.");
-
-            if (tail >= 0 && tail > BaseStream.Length)
-                throw new ArgumentOutOfRangeException(nameof(tail), "Tail pointer is too large for the given offset.");
-
-            return new StreamWrapper(BaseStream, keepOpen)
             {
-                Head = head,
+                throw new NotSupportedException("This stream is read only.");
-                Tail = tail,
-                IsReadOnly = IsReadOnly || isReadOnly,
-            };
-        }
-
-        private long ToAbsOffset(long relOffset, bool safe = false)
-        {
-            if (relOffset < 0)
-                throw new ArgumentOutOfRangeException(nameof(relOffset), relOffset, "The relative offset cannot be less than zero.");
-
-            if (Limit >= 0 && relOffset > Limit)
-                throw new ArgumentOutOfRangeException(nameof(relOffset), relOffset, "The relative offset cannot be greater than the limit.");
-
-            long value = relOffset + Head;
-
-            if (safe)
-            {
-                if ((Tail > 0 && value >= Tail) || value < Head)
-                {
-                    throw new ArgumentOutOfRangeException(nameof(relOffset));
-                }
             }
 
-            return value;
+            ArgumentOutOfRangeException.ThrowIfNegative(value);
-        }
 
-        private long ToRelOffset(long absOffset, bool safe = false)
+            lock (_sync)
-        {
-            long value = absOffset - Head;
-
-            if (safe)
             {
-                if ((Tail > 0 && value > Length) || value < 0)
+                long newEnd = Head + value;
+
+                if (Tail >= 0 && newEnd > Tail)
                 {
-                    throw new ArgumentOutOfRangeException(nameof(absOffset));
+                    throw new NotSupportedException("Cannot grow the stream past the tail bound.");
+                }
+
+                _baseStream.SetLength(newEnd);
+
+                // Update our window end to match what we just did.
+                Tail = Tail >= 0 ? Tail : -1;
+
+                if (_position > value)
+                {
+                    _position = value;
                 }
             }
+        }
 
-            return value;
+        public StreamWrapper Fork(long head = 0, long tail = -1,
+                                  bool keepOpen = false, bool isReadOnly = false)
+        {
+            long absHead = Head + head;
+            long absTail = tail < 0 ? Tail : Head + tail;
+            return ForkAbs(absHead, absTail, keepOpen, isReadOnly);
+        }
+
+        public StreamWrapper ForkAbs(long head = 0, long tail = -1,
+                                     bool keepOpen = false, bool isReadOnly = false)
+        {
+            if (head < 0 || head > _baseStream.Length)
+            {
+                throw new ArgumentOutOfRangeException(nameof(head));
+            }
+
+            if (tail >= 0 && tail > _baseStream.Length)
+            {
+                throw new ArgumentOutOfRangeException(nameof(tail));
+            }
+
+            if (tail >= 0 && tail < head)
+            {
+                throw new ArgumentOutOfRangeException(nameof(tail));
+            }
+
+            return new StreamWrapper(_baseStream, _sync, head, tail,
+                                     keepOpen, IsReadOnly || isReadOnly);
+        }
+
+        // ---------- Helpers ----------
+
+        private static void ValidateBufferArgs(byte[] buffer, int offset, int count)
+        {
+            ArgumentNullException.ThrowIfNull(buffer);
+            ArgumentOutOfRangeException.ThrowIfNegative(offset);
+            ArgumentOutOfRangeException.ThrowIfNegative(count);
+            ArgumentOutOfRangeException.ThrowIfGreaterThan(count, buffer.Length - offset);
         }
 
         protected override void Dispose(bool disposing)
@@ -218,10 +332,19 @@ namespace ReFuel.Gltf.IO
             base.Dispose(disposing);
 
             if (!disposing)
+            {
                 return;
+            }
 
-            if (!KeepOpen)
+            if (KeepOpen)
-                _proxy.Dispose();
+            {
+                return;
+            }
+
+            lock (_sync)
+            {
+                _baseStream.Dispose();
+            }
         }
     }
 }