New features and bug fixes:

- 'gvfs prefetch' can now hydrate files as it downloads their blobs
- Added a new 'gvfs service' verb for interacting with the GVFS.Service to mount/unmount all repos
- Reduced the perf overhead of GvFlt notifications for handle creation
- Treat 'git stage' the same as 'git add'
- Gather object and packfile counts in 'gvfs diagnose' to help with perf investigations

Author: sanoursa

GVFS/FastFetch/CheckoutFetchHelper.cs

@@ -68,7 +68,7 @@ public override void FastFetch(string branchOrCommit, bool isBranch)
             // Checkout diff output => FindMissingBlobs => BatchDownload => IndexPack => Checkout available blobs
             CheckoutJob checkout = new CheckoutJob(this.checkoutThreadCount, this.FolderList, commitToFetch, this.Tracer, this.Enlistment);
             FindMissingBlobsJob blobFinder = new FindMissingBlobsJob(this.SearchThreadCount, checkout.RequiredBlobs, checkout.AvailableBlobShas, this.Tracer, this.Enlistment);
-            BatchObjectDownloadJob downloader = new BatchObjectDownloadJob(this.DownloadThreadCount, this.ChunkSize, blobFinder.DownloadQueue, checkout.AvailableBlobShas, this.Tracer, this.Enlistment, this.ObjectRequestor, this.GitObjects);
+            BatchObjectDownloadJob downloader = new BatchObjectDownloadJob(this.DownloadThreadCount, this.ChunkSize, blobFinder.MissingBlobs, checkout.AvailableBlobShas, this.Tracer, this.Enlistment, this.ObjectRequestor, this.GitObjects);
             IndexPackJob packIndexer = new IndexPackJob(this.IndexThreadCount, downloader.AvailablePacks, checkout.AvailableBlobShas, this.Tracer, this.GitObjects);
 
             // Start pipeline

GVFS/FastFetch/FastFetch.csproj

@@ -74,6 +74,7 @@
     <Compile Include="Git\FastFetchLibGit2Repo.cs" />
     <Compile Include="Git\GitIndexGenerator.cs" />
     <Compile Include="HashingStream.cs" />
+    <Compile Include="Jobs\ReadFilesJob.cs" />
     <Compile Include="WorkingTree.cs" />
     <Compile Include="GitEnlistment.cs" />
     <Compile Include="Git\DiffHelper.cs" />

GVFS/FastFetch/FetchHelper.cs

@@ -10,6 +10,7 @@
 using System.Collections.Generic;
 using System.IO;
 using System.Linq;
+using System.Threading;
 
 namespace FastFetch
 {
@@ -30,7 +31,6 @@ public class FetchHelper
         protected readonly bool SkipConfigUpdate;
 
         private const string AreaPath = nameof(FetchHelper);
-        private const int CommitDepth = 1;
 
         public FetchHelper(
             ITracer tracer,
@@ -132,15 +132,24 @@ public static bool TryLoadFileList(Enlistment enlistment, string filesInput, Lis
         /// <param name="branchOrCommit">A specific branch to filter for, or null for all branches returned from info/refs</param>
         public virtual void FastFetch(string branchOrCommit, bool isBranch)
         {
-            int matchedBlobs;
-            int downloadedBlobs;
-            this.FastFetchWithStats(branchOrCommit, isBranch, out matchedBlobs, out downloadedBlobs);
+            int matchedBlobCount;
+            int downloadedBlobCount;
+            int readFileCount;
+            
+            this.FastFetchWithStats(branchOrCommit, isBranch, false, out matchedBlobCount, out downloadedBlobCount, out readFileCount);
         }
 
-        public void FastFetchWithStats(string branchOrCommit, bool isBranch, out int matchedBlobs, out int downloadedBlobs)
+        public void FastFetchWithStats(
+            string branchOrCommit,
+            bool isBranch,
+            bool readFilesAfterDownload,
+            out int matchedBlobCount,
+            out int downloadedBlobCount,
+            out int readFileCount)
         {
-            matchedBlobs = 0;
-            downloadedBlobs = 0;
+            matchedBlobCount = 0;
+            downloadedBlobCount = 0;
+            readFileCount = 0;
 
             if (string.IsNullOrWhiteSpace(branchOrCommit))
             {
@@ -169,16 +178,13 @@ public void FastFetchWithStats(string branchOrCommit, bool isBranch, out int mat
             }
 
             this.DownloadMissingCommit(commitToFetch, this.GitObjects);
-            
-            // Dummy output queue since we don't need to checkout available blobs
-            BlockingCollection<string> availableBlobs = new BlockingCollection<string>();
 
             // Configure pipeline
             // LsTreeHelper output => FindMissingBlobs => BatchDownload => IndexPack
             string shallowFile = Path.Combine(this.Enlistment.WorkingDirectoryRoot, GVFSConstants.DotGit.Shallow);
 
             string previousCommit = null;
-            
+
             // Use the shallow file to find a recent commit to diff against to try and reduce the number of SHAs to check
             DiffHelper blobEnumerator = new DiffHelper(this.Tracer, this.Enlistment, this.FileList, this.FolderList);
             if (File.Exists(shallowFile))
@@ -192,21 +198,32 @@ public void FastFetchWithStats(string branchOrCommit, bool isBranch, out int mat
                 }
             }
 
-            blobEnumerator.PerformDiff(previousCommit, commitToFetch);
-            this.HasFailures |= blobEnumerator.HasFailures;
+            new Thread(
+                () =>
+                {
+                    blobEnumerator.PerformDiff(previousCommit, commitToFetch);
+                    this.HasFailures |= blobEnumerator.HasFailures;
+                }).Start();
+
+            BlockingCollection<string> availableBlobs = new BlockingCollection<string>();
 
             FindMissingBlobsJob blobFinder = new FindMissingBlobsJob(this.SearchThreadCount, blobEnumerator.RequiredBlobs, availableBlobs, this.Tracer, this.Enlistment);
-            BatchObjectDownloadJob downloader = new BatchObjectDownloadJob(this.DownloadThreadCount, this.ChunkSize, blobFinder.DownloadQueue, availableBlobs, this.Tracer, this.Enlistment, this.ObjectRequestor, this.GitObjects);
+            BatchObjectDownloadJob downloader = new BatchObjectDownloadJob(this.DownloadThreadCount, this.ChunkSize, blobFinder.MissingBlobs, availableBlobs, this.Tracer, this.Enlistment, this.ObjectRequestor, this.GitObjects);
             IndexPackJob packIndexer = new IndexPackJob(this.IndexThreadCount, downloader.AvailablePacks, availableBlobs, this.Tracer, this.GitObjects);
+            ReadFilesJob readFiles = new ReadFilesJob(Environment.ProcessorCount * 2, blobEnumerator.FileAddOperations, availableBlobs, this.Tracer);
 
             blobFinder.Start();
             downloader.Start();
-            
+
+            if (readFilesAfterDownload)
+            {
+                readFiles.Start();
+            }
+
             // If indexing happens during searching, searching progressively gets slower, so wait on searching before indexing.
             blobFinder.WaitForCompletion();
             this.HasFailures |= blobFinder.HasFailures;
 
-            // Index regardless of failures, it'll shorten the next fetch.
             packIndexer.Start();
 
             downloader.WaitForCompletion();
@@ -215,8 +232,17 @@ public void FastFetchWithStats(string branchOrCommit, bool isBranch, out int mat
             packIndexer.WaitForCompletion();
             this.HasFailures |= packIndexer.HasFailures;
 
-            matchedBlobs = blobFinder.AvailableBlobCount + blobFinder.MissingBlobCount;
-            downloadedBlobs = blobFinder.MissingBlobCount;
+            availableBlobs.CompleteAdding();
+
+            if (readFilesAfterDownload)
+            {
+                readFiles.WaitForCompletion();
+                this.HasFailures |= readFiles.HasFailures;
+            }
+
+            matchedBlobCount = blobFinder.AvailableBlobCount + blobFinder.MissingBlobCount;
+            downloadedBlobCount = blobFinder.MissingBlobCount;
+            readFileCount = readFiles.ReadFileCount;
 
             if (!this.SkipConfigUpdate && !this.HasFailures)
             {
@@ -313,15 +339,14 @@ protected void DownloadMissingCommit(string commitSha, GitObjects gitObjects)
         {
             EventMetadata startMetadata = new EventMetadata();
             startMetadata.Add("CommitSha", commitSha);
-            startMetadata.Add("CommitDepth", CommitDepth);
 
             using (ITracer activity = this.Tracer.StartActivity("DownloadTrees", EventLevel.Informational, Keywords.Telemetry, startMetadata))
             {
                 using (FastFetchLibGit2Repo repo = new FastFetchLibGit2Repo(this.Tracer, this.Enlistment.WorkingDirectoryRoot))
                 {
                     if (!repo.ObjectExists(commitSha))
                     {
-                        if (!gitObjects.TryEnsureCommitIsLocal(commitSha, commitDepth: CommitDepth))
+                        if (!gitObjects.TryDownloadCommit(commitSha))
                         {
                             EventMetadata metadata = new EventMetadata();
                             metadata.Add("ObjectsEndpointUrl", this.ObjectRequestor.CacheServer.ObjectsEndpointUrl);

GVFS/FastFetch/Git/GitIndexGenerator.cs

@@ -108,7 +108,7 @@ private void WriteAllEntries(uint version, HashSet<string> sparseCheckoutEntries
 
                     uint lastStringLength = 0;
                     LsTreeEntry entry;
-                    while (this.entryQueue.TryTake(out entry, millisecondsTimeout: -1))
+                    while (this.entryQueue.TryTake(out entry, Timeout.Infinite))
                     {
                         bool skipWorkTree = 
                             sparseCheckoutEntries != null && 

GVFS/FastFetch/Jobs/BatchObjectDownloadJob.cs

@@ -24,7 +24,7 @@ public class BatchObjectDownloadJob : Job
 
         private static readonly TimeSpan HeartBeatPeriod = TimeSpan.FromSeconds(20);
 
-        private readonly BlockingAggregator<string, BlobDownloadRequest> inputQueue;
+        private readonly DownloadRequestAggregator downloadRequests;
 
         private int activeDownloadCount;
 
@@ -39,7 +39,7 @@ public class BatchObjectDownloadJob : Job
         public BatchObjectDownloadJob(
             int maxParallel,
             int chunkSize,
-            BlockingCollection<string> inputQueue,
+            BlockingCollection<string> missingBlobs,
             BlockingCollection<string> availableBlobs,
             ITracer tracer,
             Enlistment enlistment,
@@ -49,7 +49,7 @@ public BatchObjectDownloadJob(
         {
             this.tracer = tracer.StartActivity(AreaPath, EventLevel.Informational, Keywords.Telemetry, metadata: null);
 
-            this.inputQueue = new BlockingAggregator<string, BlobDownloadRequest>(inputQueue, chunkSize, objectIds => new BlobDownloadRequest(objectIds));
+            this.downloadRequests = new DownloadRequestAggregator(missingBlobs, chunkSize);
 
             this.enlistment = enlistment;
             this.objectRequestor = objectRequestor;
@@ -73,7 +73,7 @@ protected override void DoBeforeWork()
         protected override void DoWork()
         {
             BlobDownloadRequest request;
-            while (this.inputQueue.TryTake(out request))
+            while (this.downloadRequests.TryTake(out request))
             {
                 Interlocked.Increment(ref this.activeDownloadCount);
 
@@ -89,7 +89,6 @@ protected override void DoWork()
                         HashSet<string> successfulDownloads = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
                         RetryWrapper<GitObjectsHttpRequestor.GitObjectTaskResult>.InvocationResult result = this.objectRequestor.TryDownloadObjects(
                                 () => request.ObjectIds.Except(successfulDownloads),
-                                commitDepth: 1,
                                 onSuccess: (tryCount, response) => this.WriteObjectOrPack(request, tryCount, response, successfulDownloads),
                                 onFailure: RetryWrapper<GitObjectsHttpRequestor.GitObjectTaskResult>.StandardErrorHandler(activity, request.PackId, DownloadAreaPath),
                                 preferBatchedLooseObjects: true);
@@ -199,42 +198,45 @@ private void EmitHeartbeat(object state)
             this.tracer.RelatedEvent(EventLevel.Verbose, "DownloadHeartbeat", metadata);
         }
 
-        private class BlockingAggregator<InputType, OutputType>
+        private class DownloadRequestAggregator
         {
-            private BlockingCollection<InputType> inputQueue;
+            private BlockingCollection<string> missingBlobs;
             private int chunkSize;
-            private Func<List<InputType>, OutputType> factory;
 
-            public BlockingAggregator(BlockingCollection<InputType> input, int chunkSize, Func<List<InputType>, OutputType> factory)
+            public DownloadRequestAggregator(BlockingCollection<string> missingBlobs, int chunkSize)
             {
-                this.inputQueue = input;
+                this.missingBlobs = missingBlobs;
                 this.chunkSize = chunkSize;
-                this.factory = factory;
             }
 
-            public bool TryTake(out OutputType output)
+            public bool TryTake(out BlobDownloadRequest request)
             {
-                List<InputType> intermediary = new List<InputType>();
+                List<string> blobsInChunk = new List<string>();
+
                 for (int i = 0; i < this.chunkSize; ++i)
                 {
-                    InputType data;
-                    if (this.inputQueue.TryTake(out data, millisecondsTimeout: -1))
+                    // Only wait a short while for new work to show up, otherwise go ahead and download what we have accumulated so far
+                    const int TimeoutMs = 100;
+
+                    string blobId;
+                    if (this.missingBlobs.TryTake(out blobId, TimeoutMs))
                     {
-                        intermediary.Add(data);
+                        blobsInChunk.Add(blobId);
                     }
-                    else
+                    else if (blobsInChunk.Count > 0 ||
+                        this.missingBlobs.IsAddingCompleted)
                     {
                         break;
                     }
                 }
 
-                if (intermediary.Any())
+                if (blobsInChunk.Count > 0)
                 {
-                    output = this.factory(intermediary);
+                    request = new BlobDownloadRequest(blobsInChunk);
                     return true;
                 }
 
-                output = default(OutputType);
+                request = null;
                 return false;
             }
         }

GVFS/FastFetch/Jobs/CheckoutJob.cs

@@ -282,7 +282,7 @@ private void HandleAllFileAddOperations()
             using (FastFetchLibGit2Repo repo = new FastFetchLibGit2Repo(this.tracer, this.enlistment.WorkingDirectoryRoot))
             {
                 string availableBlob;
-                while (this.AvailableBlobShas.TryTake(out availableBlob, millisecondsTimeout: -1))
+                while (this.AvailableBlobShas.TryTake(out availableBlob, Timeout.Infinite))
                 {
                     if (this.HasFailures)
                     {

GVFS/FastFetch/Jobs/Data/BlobDownloadRequest.cs

@@ -1,5 +1,4 @@
 using System.Collections.Generic;
-using System.Linq;
 using System.Threading;
 
 namespace FastFetch.Jobs.Data

GVFS/FastFetch/Jobs/Data/IndexPackRequest.cs

@@ -1,10 +1,4 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace FastFetch.Jobs.Data
+namespace FastFetch.Jobs.Data
 {
     public class IndexPackRequest
     {

GVFS/FastFetch/Jobs/FindMissingBlobsJob.cs

@@ -20,28 +20,28 @@ public class FindMissingBlobsJob : Job
         private int missingBlobCount;
         private int availableBlobCount;
 
-        private BlockingCollection<string> inputQueue;
+        private BlockingCollection<string> requiredBlobs;
 
         private ConcurrentHashSet<string> alreadyFoundBlobIds;
 
         public FindMissingBlobsJob(
             int maxParallel,
-            BlockingCollection<string> inputQueue,
+            BlockingCollection<string> requiredBlobs,
             BlockingCollection<string> availableBlobs,
             ITracer tracer,
             Enlistment enlistment)
             : base(maxParallel)
         {
             this.tracer = tracer.StartActivity(AreaPath, EventLevel.Informational, Keywords.Telemetry, metadata: null);
-            this.inputQueue = inputQueue;
+            this.requiredBlobs = requiredBlobs;
             this.enlistment = enlistment;
             this.alreadyFoundBlobIds = new ConcurrentHashSet<string>();
 
-            this.DownloadQueue = new BlockingCollection<string>();
+            this.MissingBlobs = new BlockingCollection<string>();
             this.AvailableBlobs = availableBlobs;
         }
 
-        public BlockingCollection<string> DownloadQueue { get; }
+        public BlockingCollection<string> MissingBlobs { get; }
         public BlockingCollection<string> AvailableBlobs { get; }
 
         public int MissingBlobCount
@@ -59,14 +59,14 @@ protected override void DoWork()
             string blobId;
             using (FastFetchLibGit2Repo repo = new FastFetchLibGit2Repo(this.tracer, this.enlistment.WorkingDirectoryRoot))
             {
-                while (this.inputQueue.TryTake(out blobId, Timeout.Infinite))
+                while (this.requiredBlobs.TryTake(out blobId, Timeout.Infinite))
                 {
                     if (this.alreadyFoundBlobIds.Add(blobId))
                     {
                         if (!repo.ObjectExists(blobId))
                         {
                             Interlocked.Increment(ref this.missingBlobCount);
-                            this.DownloadQueue.Add(blobId);
+                            this.MissingBlobs.Add(blobId);
                         }
                         else
                         {
@@ -80,7 +80,7 @@ protected override void DoWork()
 
         protected override void DoAfterWork()
         {
-            this.DownloadQueue.CompleteAdding();
+            this.MissingBlobs.CompleteAdding();
 
             EventMetadata metadata = new EventMetadata();
             metadata.Add("TotalMissingObjects", this.missingBlobCount);

GVFS/FastFetch/Jobs/IndexPackJob.cs

@@ -12,7 +12,7 @@ public class IndexPackJob : Job
         private const string AreaPath = "IndexPackJob";
         private const string IndexPackAreaPath = "IndexPack";
 
-        private readonly BlockingCollection<IndexPackRequest> inputQueue;
+        private readonly BlockingCollection<IndexPackRequest> availablePacks;
 
         private ITracer tracer;
         private GitObjects gitObjects;
@@ -21,14 +21,14 @@ public class IndexPackJob : Job
 
         public IndexPackJob(
             int maxParallel,
-            BlockingCollection<IndexPackRequest> inputQueue,
+            BlockingCollection<IndexPackRequest> availablePacks,
             BlockingCollection<string> availableBlobs,
             ITracer tracer,
             GitObjects gitObjects)
             : base(maxParallel)
         {
             this.tracer = tracer.StartActivity(AreaPath, EventLevel.Informational, Keywords.Telemetry, metadata: null);
-            this.inputQueue = inputQueue;
+            this.availablePacks = availablePacks;
             this.gitObjects = gitObjects;
             this.AvailableBlobs = availableBlobs;
         }
@@ -38,7 +38,7 @@ public IndexPackJob(
         protected override void DoWork()
         {
             IndexPackRequest request;
-            while (this.inputQueue.TryTake(out request, millisecondsTimeout: -1))
+            while (this.availablePacks.TryTake(out request, Timeout.Infinite))
             {
                 EventMetadata metadata = new EventMetadata();
                 metadata.Add("PackId", request.DownloadRequest.PackId);