Merge pull request #1173 from vzabavnov/B1167-Fetch

Fix errors during fetch #1167

src/GitTfs.VsCommon/Wrappers.cs

@@ -40,8 +40,9 @@ namespace GitTfs.VsCommon
                 new ChangesetVersionSpec(changesetNumber),
                 DeletedState.NonDeleted,
                 ItemType.Any,
-                true
-                );
+                // do not load the loading info
+                false);
+
             return _bridge.Wrap<WrapperForItem, Item>(itemSet.Items);
         }
 
@@ -468,6 +469,12 @@ namespace GitTfs.VsCommon
             Retry.Do(() => DoUntilNoFailures(() => _workspace.Get(new ChangesetVersionSpec(changeset), GetOptions.Overwrite | GetOptions.GetAll)));
         }
 
+        public void GetSpecificVersion(int changesetId, IEnumerable<IItem> items)
+        {
+            var version = new ChangesetVersionSpec(changesetId);
+            GetRequests(items.Select(e => new GetRequest(new ItemSpec(e.ServerItem, RecursionType.Full), version)));
+        }
+
         public void GetSpecificVersion(IChangeset changeset)
         {
             GetSpecificVersion(changeset.ChangesetId, changeset.Changes);
@@ -475,15 +482,7 @@ namespace GitTfs.VsCommon
 
         public void GetSpecificVersion(int changesetId, IEnumerable<IChange> changes)
         {
-            Retry.Do(() =>
-            {
-                var requests = from change in changes
-                               select
-                                   new GetRequest(
-                                       new ItemSpec(change.Item.ServerItem, RecursionType.None, change.Item.DeletionId),
-                                       changesetId);
-                DoUntilNoFailures(() => _workspace.Get(requests.ToArray(), GetOptions.Overwrite));
-            });
+            GetRequests(changes.Select(change => new GetRequest(new ItemSpec(change.Item.ServerItem, RecursionType.None, change.Item.DeletionId), changesetId)));
         }
 
         public string GetLocalItemForServerItem(string serverItem)
@@ -536,6 +535,27 @@ namespace GitTfs.VsCommon
                 throw new GitTfsGatedCheckinException(gatedException.ShelvesetName, gatedException.AffectedBuildDefinitions, gatedException.CheckInTicket);
             }
         }
+
+        public void GetRequests(IEnumerable<GetRequest> source, int batchSize = 20)
+        {
+            source.ToBatch(batchSize).DoParallel(batch =>
+            {
+                var items = batch;
+                Retry.Do(() =>
+                {
+                    while (items.Length > 0)
+                    {
+                        var status = _workspace.Get(items.ToArray(), GetOptions.Overwrite | GetOptions.GetAll);
+                        if (status.NumFailures == 0)
+                        {
+                            break;
+                        }
+
+                        items = status.GetFailures().Join(items, e => e.ServerItem, e => e.ItemSpec.Item, (failure, request) => request).ToArray();
+                    }
+                });
+            });
+        }
     }
 
     public class WrapperForBranchObject : WrapperFor<BranchObject>, IBranchObject

src/GitTfs.VsFake/TfsHelper.VsFake.cs

@@ -258,6 +258,11 @@ namespace GitTfs.VsFake
                 _repositoryRoot = repositoryRoot;
             }
 
+            public void GetSpecificVersion(int changesetId, IEnumerable<IItem> items)
+            {
+                throw new NotImplementedException();
+            }
+
             public void GetSpecificVersion(IChangeset changeset)
             {
                 GetSpecificVersion(changeset.ChangesetId, changeset.Changes);

src/GitTfs/Core/Ext.cs

@@ -11,6 +11,9 @@ using StructureMap;
 
 namespace GitTfs.Core
 {
+    using System.Collections.Concurrent;
+    using System.Threading;
+
     public static class Ext
     {
         public static T Tap<T>(this T o, Action<T> block)
@@ -159,5 +162,190 @@ namespace GitTfs.Core
                 }
             }
         }
+
+        /// <summary>
+        /// Translate the <paramref name="source"/> to the sequence of array's items
+        /// </summary>
+        /// <typeparam name="TSource">The source item type</typeparam>
+        /// <typeparam name="TResult">The output item type</typeparam>
+        /// <param name="source">The source collection</param>
+        /// <param name="selector">The delegate to use to translate</param>
+        /// <param name="batchSize">the of item in the batch array</param>
+        /// <returns>The <see cref="IEnumerable{T}"/> with arrays of items sized by <paramref name="batchSize"/></returns>
+        public static IEnumerable<TResult[]> ToBatch<TSource, TResult>(this IEnumerable<TSource> source, Func<TSource, TResult> selector, int batchSize)
+        {
+            var batch = new List<TResult>(batchSize);
+            
+            foreach (var item in source)
+            {
+                if (batch.Count >= batchSize)
+                {
+                    yield return batch.ToArray();
+                    batch.Clear();
+                }
+
+                batch.Add(selector(item));
+            }
+
+            if (batch.Count > 0)
+            {
+                yield return batch.ToArray();
+            }
+        }
+
+        /// <summary>
+        /// Translate the <paramref name="source"/> to the sequence of array's items
+        /// </summary>
+        /// <typeparam name="T">The source item type</typeparam>
+        /// <param name="source">The source collection</param>
+        /// <param name="batchSize">the of item in the batch array</param>
+        /// <returns>The <see cref="IEnumerable{T}"/> with arrays of items sized by <paramref name="batchSize"/></returns>
+        [DebuggerStepThrough]
+        public static IEnumerable<T[]> ToBatch<T>(this IEnumerable<T> source, int batchSize)
+        {
+            return ToBatch(source, e => e, batchSize);
+        }
+
+        /// <summary>
+        /// Executes the <paramref name="action"/> for each item in the <paramref name="source"/> simultaneously.
+        /// </summary>
+        /// <param name="source">The source sequence.</param>
+        /// <param name="action">The action to execute.</param>
+        /// <typeparam name="T">The source item type</typeparam>
+        public static void DoParallel<T>(this IEnumerable<T> source, Action<T> action)
+        {
+#if DEBUG && NO_PARALLEL
+            foreach (var item in source)
+            {
+                action(item);
+            }
+#else
+            (source as ParallelQuery<T> ?? source.AsParallel()).ForAll(action);
+#endif
+        }
+
+        /// <summary>
+        /// TRuns the <paramref name="action"/> on each item in the <paramref name="source"/> in parallel
+        /// </summary>
+        /// <param name="source">
+        /// The source collection of items.
+        /// </param>
+        /// <param name="action">
+        /// The action to process of the item.
+        /// </param>
+        /// <param name="retryInterval">
+        /// The delay between retries.
+        /// </param>
+        /// <typeparam name="T">The type of items in the <paramref name="source"/>.</typeparam>
+        /// <returns>
+        /// Returns <b>true</b> when all items processed successfully.</returns>
+        public static void DoParallelRetry<T>(
+            this IEnumerable<T> source,
+            Action<T> action,
+            TimeSpan retryInterval)
+        {
+            DoParallelRetry(source, action, 10, retryInterval);
+        }
+
+        /// <summary>
+        /// TRuns the <paramref name="action"/> on each item in the <paramref name="source"/> in parallel
+        /// </summary>
+        /// <param name="source">
+        /// The source collection of items.
+        /// </param>
+        /// <param name="action">
+        /// The action to process of the item.
+        /// </param>
+        /// <typeparam name="T">The type of items in the <paramref name="source"/>.</typeparam>
+        /// <returns>
+        /// Returns <b>true</b> when all items processed successfully.</returns>
+        public static void DoParallelRetry<T>(this IEnumerable<T> source, Action<T> action)
+        {
+            DoParallelRetry(source, action, 10, TimeSpan.FromSeconds(1));
+        }
+
+        /// <summary>
+        /// TRuns the <paramref name="action"/> on each item in the <paramref name="source"/> in parallel
+        /// </summary>
+        /// <param name="source">
+        /// The source collection of items.
+        /// </param>
+        /// <param name="action">
+        /// The action to process of the item.
+        /// </param>
+        /// <param name="retryCount">
+        /// TThe number of retries.
+        /// </param>
+        /// <typeparam name="T">The type of items in the <paramref name="source"/>.</typeparam>
+        /// <returns>
+        /// Returns <b>true</b> when all items processed successfully.</returns>
+        public static void DoParallelRetry<T>(this IEnumerable<T> source, Action<T> action, int retryCount)
+        {
+            DoParallelRetry(source, action, retryCount, TimeSpan.FromSeconds(1));
+        }
+
+        /// <summary>
+        /// TRuns the <paramref name="action"/> on each item in the <paramref name="source"/> in parallel
+        /// </summary>
+        /// <param name="source">
+        /// The source collection of items.
+        /// </param>
+        /// <param name="action">
+        /// The action to process of the item.
+        /// </param>
+        /// <param name="retryCount">
+        /// TThe number of retries.
+        /// </param>
+        /// <param name="retryInterval">
+        /// The delay between retries.
+        /// </param>
+        /// <typeparam name="T">The type of items in the <paramref name="source"/>.</typeparam>
+        /// <returns>
+        /// Returns <b>true</b> when all items processed successfully.</returns>
+        public static void DoParallelRetry<T>(this IEnumerable<T> source, Action<T> action, int retryCount, TimeSpan retryInterval)
+        {
+            var fails = new ConcurrentBag<Exception>();
+
+            source.DoParallel(
+                item =>
+                {
+                    List<Exception> exceptions = null;
+
+                    for (var i = 0; i < retryCount; i++)
+                    {
+                        if (i != 0)
+                        {
+                            Thread.Sleep(retryInterval);
+                        }
+
+                        try
+                        {
+                            action(item);
+                            return;
+                        }
+                        catch (Exception e)
+                        {
+                            Trace.TraceError("The action is failing: {0}", e.Message);
+
+                            if (exceptions == null)
+                            {
+                                exceptions = new List<Exception>();
+                            }
+
+                            exceptions.Add(e);
+                        }
+                    }
+
+                    if (exceptions != null)
+                    {
+                        fails.Add(new AggregateException(exceptions));
+                    }
+                });
+
+            if (fails.Count > 0)
+            {
+                throw new AggregateException(fails);
+            }
+        }
     }
 }

src/GitTfs/Core/ITfsWorkspace.cs

@@ -34,6 +34,12 @@ namespace GitTfs.Core
         /// Populates the workspace with a snapshot, as of the given changeset.
         /// </summary>
         void Get(int changesetId);
+
+		/// <summary>
+        /// Populates the workspace with specified items, as of the given changeset.
+        /// </summary>
+        void Get(int changesetId, IEnumerable<IItem> items);
+
         /// <summary>
         /// Gets the files changed in a given changeset.
         /// </summary>

src/GitTfs/Core/TfsChangeset.cs

@@ -120,7 +120,7 @@ namespace GitTfs.Core
             }
             else
             {
-                workspace.Get(_changeset.ChangesetId);
+                workspace.Get(_changeset.ChangesetId, tfsTreeEntries.Select(e => e.Item));
                 foreach (var entry in tfsTreeEntries)
                 {
                     Add(entry.Item, entry.FullName, treeBuilder, workspace);

src/GitTfs/Core/TfsInterop/IWorkspace.cs

@@ -14,6 +14,7 @@ namespace GitTfs.Core.TfsInterop
         int PendRename(string pathFrom, string pathTo);
         void ForceGetFile(string path, int changeset);
         void GetSpecificVersion(int changeset);
+        void GetSpecificVersion(int changeset, IEnumerable<IItem> items);
         void GetSpecificVersion(IChangeset changeset);
         void GetSpecificVersion(int changeset, IEnumerable<IChange> changes);
         string GetLocalItemForServerItem(string serverItem);

src/GitTfs/Core/TfsWorkspace.cs

@@ -235,6 +235,11 @@ namespace GitTfs.Core
             _workspace.GetSpecificVersion(changesetId);
         }
 
+        public void Get(int changesetId, IEnumerable<IItem> items)
+        {
+            _workspace.GetSpecificVersion(changesetId, items);
+        }
+
         public void Get(IChangeset changeset)
         {
             _workspace.GetSpecificVersion(changeset);