PoC for writing directly into the next stream

The filters are run in-line with the rest of the system, and libgit2 is
set up to have its last filter/writestream write to the
filesystem. Instead of using an intermediate buffer/file/pipe, we can
make the writes from the user's Smude() or Clean() filter into their output
stream propagate immediately into the next filter, leading up the chain
to the filesystem, removing any intermediate buffering other than that
which is necessary for the filters themselves to work.

In the general case, we only have the one clean or smudge filter, which
means that even in the case of out-of-band storage for large files,
where a small blob can cause very large outputs, we would still write
directly to the filesystem.

by J Wyman (@whoisj): Buffered writes to next filter to minimize managed to native marshaling for performance optimizations

by J Wyman (@whoisj): Split `WriteStream` into its own file and cleaned it up (checks, style, etc)

Author: carlosmn

LibGit2Sharp/Core/WriteStream.cs

@@ -0,0 +1,64 @@
+using System;
+using System.IO;
+
+namespace LibGit2Sharp.Core
+{
+    class WriteStream : Stream
+    {
+        readonly GitWriteStream nextStream;
+        readonly IntPtr nextPtr;
+
+        public WriteStream(GitWriteStream nextStream, IntPtr nextPtr)
+        {
+            this.nextStream = nextStream;
+            this.nextPtr = nextPtr;
+        }
+
+        public override bool CanWrite { get { return true; } }
+
+        public override bool CanRead { get { return false; } }
+
+        public override bool CanSeek { get { return false; } }
+
+        public override long Position
+        {
+            get { throw new NotImplementedException(); }
+            set { throw new InvalidOperationException(); }
+        }
+
+        public override long Length { get { throw new InvalidOperationException(); } }
+
+        public override void Flush()
+        {
+        }
+
+        public override void SetLength(long value)
+        {
+            throw new InvalidOperationException();
+        }
+
+        public override int Read(byte[] buffer, int offset, int count)
+        {
+            throw new InvalidOperationException();
+        }
+
+        public override long Seek(long offset, SeekOrigin origin)
+        {
+            throw new InvalidOperationException();
+        }
+
+        public override void Write(byte[] buffer, int offset, int count)
+        {
+            unsafe
+            {
+                fixed (byte* bufferPtr = &buffer[offset])
+                {
+                    if (nextStream.write(nextPtr, (IntPtr)bufferPtr, (UIntPtr)count) < 0)
+                    {
+                        throw new LibGit2SharpException("failed to write to next buffer");
+                    }
+                }
+            }
+        }
+    }
+}

LibGit2Sharp/Filter.cs

@@ -1,6 +1,5 @@
 using System;
 using System.Collections.Generic;
-using System.Diagnostics;
 using System.IO;
 using System.Linq;
 using System.Runtime.InteropServices;
@@ -17,6 +16,8 @@ public abstract class Filter : IEquatable<Filter>
     {
         private static readonly LambdaEqualityHelper<Filter> equalityHelper =
         new LambdaEqualityHelper<Filter>(x => x.Name, x => x.Attributes);
+        // 64K is optimal buffer size per https://technet.microsoft.com/en-us/library/cc938632.aspx
+        private const int BufferSize = 64 * 1024;
 
         private readonly string name;
         private readonly IEnumerable<FilterAttributeEntry> attributes;
@@ -51,6 +52,7 @@ protected Filter(string name, IEnumerable<FilterAttributeEntry> attributes)
         private IntPtr thisPtr;
         private IntPtr nextPtr;
         private FilterSource filterSource;
+        private Stream output;
 
         /// <summary>
         /// The name that this filter was registered with
@@ -80,7 +82,7 @@ internal GitFilter GitFilter
         /// Complete callback on filter
         /// 
         /// This optional callback will be invoked when the upstream filter is
-        /// closed. Gives the filter a change to perform any final actions or
+        /// closed. Gives the filter a chance to perform any final actions or
         /// necissary clean up.
         /// </summary>
         /// <param name="path">The path of the file being filtered</param>
@@ -224,6 +226,7 @@ int StreamCreateCallback(out IntPtr git_writestream_out, GitFilter self, IntPtr
                 nextStream = new GitWriteStream();
                 Marshal.PtrToStructure(nextPtr, nextStream);
                 filterSource = FilterSource.FromNativePtr(filterSourcePtr);
+                output = new WriteStream(nextStream, nextPtr);
             }
             catch (Exception exception)
             {
@@ -254,20 +257,10 @@ int StreamCloseCallback(IntPtr stream)
                 Ensure.ArgumentNotZeroIntPtr(stream, "stream");
                 Ensure.ArgumentIsExpectedIntPtr(stream, thisPtr, "stream");
 
-                string tempFileName = Path.GetTempFileName();
-                // Setup a file system backed write-to stream to work with this gives the  runtime
-                // somewhere to put bits if the amount of data could cause an OOM scenario.
-                using (FileStream output = File.Open(tempFileName, FileMode.Open, FileAccess.Write, FileShare.ReadWrite))
-                // Setup a septerate read-from stream on the same file system backing
-                // a seperate stream helps avoid a flush to disk when reading the written content
-                using (FileStream reader = File.Open(tempFileName, FileMode.Open, FileAccess.Read, FileShare.ReadWrite))
+                using (BufferedStream outputBuffer = new BufferedStream(output, BufferSize))
                 {
-                    Complete(filterSource.Path, filterSource.Root, output);
-                    output.Flush();
-                    WriteToNextFilter(reader);
+                    Complete(filterSource.Path, filterSource.Root, outputBuffer);
                 }
-                // clean up after outselves
-                File.Delete(tempFileName);
             }
             catch (Exception exception)
             {
@@ -310,53 +303,22 @@ unsafe int StreamWriteCallback(IntPtr stream, IntPtr buffer, UIntPtr len)
 
                 string tempFileName = Path.GetTempFileName();
                 using (UnmanagedMemoryStream input = new UnmanagedMemoryStream((byte*)buffer.ToPointer(), (long)len))
-                // Setup a file system backed write-to stream to work with this gives the  runtime
-                // somewhere to put bits if the amount of data could cause an OOM scenario.
-                using (FileStream output = File.Open(tempFileName, FileMode.Open, FileAccess.Write, FileShare.ReadWrite))
-                // Setup a septerate read-from stream on the same file system backing
-                // a seperate stream helps avoid a flush to disk when reading the written content
-                using (FileStream reader = File.Open(tempFileName, FileMode.Open, FileAccess.Read, FileShare.ReadWrite))
+                using (BufferedStream outputBuffer = new BufferedStream(output, BufferSize))
                 {
                     switch (filterSource.SourceMode)
                     {
                         case FilterMode.Clean:
-                            try
-                            {
-                                Clean(filterSource.Path, filterSource.Root, input, output);
-                            }
-                            catch (Exception exception)
-                            {
-                                Log.Write(LogLevel.Error, "Filter.StreamWriteCallback exception");
-                                Log.Write(LogLevel.Error, exception.ToString());
-                                Proxy.giterr_set_str(GitErrorCategory.Filter, exception);
-                                result = (int)GitErrorCode.Error;
-                            }
+                            Clean(filterSource.Path, filterSource.Root, input, outputBuffer);
                             break;
 
                         case FilterMode.Smudge:
-                            try
-                            {
-                                Smudge(filterSource.Path, filterSource.Root, input, output);
-                            }
-                            catch (Exception exception)
-                            {
-                                Log.Write(LogLevel.Error, "Filter.StreamWriteCallback exception");
-                                Log.Write(LogLevel.Error, exception.ToString());
-                                Proxy.giterr_set_str(GitErrorCategory.Filter, exception);
-                                result = (int)GitErrorCode.Error;
-                            }
+                            Smudge(filterSource.Path, filterSource.Root, input, outputBuffer);
                             break;
+
                         default:
                             Proxy.giterr_set_str(GitErrorCategory.Filter, "Unexpected filter mode.");
                             return (int)GitErrorCode.Ambiguous;
                     }
-
-                    if (result == (int)GitErrorCode.Ok)
-                    {
-                        // have to flush the write-to stream to enable the read stream to get access to the bits
-                        output.Flush();
-                        result = WriteToNextFilter(reader);
-                    }
                 }
 
                 // clean up after outselves
@@ -372,37 +334,5 @@ unsafe int StreamWriteCallback(IntPtr stream, IntPtr buffer, UIntPtr len)
 
             return result;
         }
-
-        private unsafe int WriteToNextFilter(Stream output)
-        {
-            // 64K is optimal buffer size per https://technet.microsoft.com/en-us/library/cc938632.aspx
-            const int BufferSize = 64 * 1024;
-
-            Debug.Assert(output != null, "output parameter is null");
-            Debug.Assert(output.CanRead, "output.CanRead parameter equals false");
-
-            int result = 0;
-            byte[] bytes = new byte[BufferSize];
-            IntPtr bytesPtr = Marshal.AllocHGlobal(BufferSize);
-            try
-            {
-                int read = 0;
-                while ((read = output.Read(bytes, 0, bytes.Length)) > 0)
-                {
-                    Marshal.Copy(bytes, 0, bytesPtr, read);
-                    if ((result = nextStream.write(nextPtr, bytesPtr, (UIntPtr)read)) != (int)GitErrorCode.Ok)
-                    {
-                        Proxy.giterr_set_str(GitErrorCategory.Filter, "Filter write to next stream failed");
-                        break;
-                    }
-                }
-            }
-            finally
-            {
-                Marshal.FreeHGlobal(bytesPtr);
-            }
-
-            return result;
-        }
     }
 }

LibGit2Sharp/LibGit2Sharp.csproj

@@ -75,6 +75,7 @@
     <Compile Include="Core\Platform.cs" />
     <Compile Include="Core\Handles\ConflictIteratorSafeHandle.cs" />
     <Compile Include="Core\GitWriteStream.cs" />
+    <Compile Include="Core\WriteStream.cs" />
     <Compile Include="DescribeOptions.cs" />
     <Compile Include="DescribeStrategy.cs" />
     <Compile Include="Core\GitDescribeFormatOptions.cs" />