Process download chunks as tasks to increase concurrency

DepotDownloader/CDNClientPool.cs

@@ -17,10 +17,11 @@ namespace DepotDownloader
         private const int ServerEndpointMinimumSize = 8;
 
         private readonly Steam3Session steamSession;
+        private readonly uint appId;
 
         public CDNClient CDNClient { get; }
 
-        private readonly ConcurrentBag<CDNClient.Server> activeConnectionPool;
+        private readonly ConcurrentStack<CDNClient.Server> activeConnectionPool;
         private readonly BlockingCollection<CDNClient.Server> availableServerEndpoints;
 
         private readonly AutoResetEvent populatePoolEvent;
@@ -28,12 +29,13 @@ namespace DepotDownloader
         private readonly CancellationTokenSource shutdownToken;
         public CancellationTokenSource ExhaustedToken { get; set; }
 
-        public CDNClientPool(Steam3Session steamSession)
+        public CDNClientPool(Steam3Session steamSession, uint appId)
         {
             this.steamSession = steamSession;
+            this.appId = appId;
             CDNClient = new CDNClient(steamSession.steamClient);
 
-            activeConnectionPool = new ConcurrentBag<CDNClient.Server>();
+            activeConnectionPool = new ConcurrentStack<CDNClient.Server>();
             availableServerEndpoints = new BlockingCollection<CDNClient.Server>();
 
             populatePoolEvent = new AutoResetEvent(true);
@@ -97,12 +99,23 @@ namespace DepotDownloader
                         return;
                     }
 
-                    var weightedCdnServers = servers.Where(x => x.Type == "SteamCache" || x.Type == "CDN").Select(x =>
-                    {
-                        AccountSettingsStore.Instance.ContentServerPenalty.TryGetValue(x.Host, out var penalty);
+                    var weightedCdnServers = servers
+                        .Where(x =>
+                        {
+#if STEAMKIT_UNRELEASED
+                            var isEligibleForApp = x.AllowedAppIds == null || x.AllowedAppIds.Contains(appId);
+                            return isEligibleForApp && (x.Type == "SteamCache" || x.Type == "CDN");
+#else
+                            return x.Type == "SteamCache" || x.Type == "CDN";
+#endif
+                        })
+                        .Select(x =>
+                        {
+                            AccountSettingsStore.Instance.ContentServerPenalty.TryGetValue(x.Host, out var penalty);
 
-                        return Tuple.Create(x, penalty);
-                    }).OrderBy(x => x.Item2).ThenBy(x => x.Item1.WeightedLoad);
+                            return Tuple.Create(x, penalty);
+                        })
+                        .OrderBy(x => x.Item2).ThenBy(x => x.Item1.WeightedLoad);
 
                     foreach (var (server, weight) in weightedCdnServers)
                     {
@@ -122,7 +135,27 @@ namespace DepotDownloader
             }
         }
 
-        private async Task<string> AuthenticateConnection(uint appId, uint depotId, CDNClient.Server server)
+        private CDNClient.Server BuildConnection(CancellationToken token)
+        {
+            if (availableServerEndpoints.Count < ServerEndpointMinimumSize)
+            {
+                populatePoolEvent.Set();
+            }
+
+            return availableServerEndpoints.Take(token);
+        }
+
+        public CDNClient.Server GetConnection(CancellationToken token)
+        {
+            if (!activeConnectionPool.TryPop(out var connection))
+            {
+                connection = BuildConnection(token);
+            }
+
+            return connection;
+        }
+
+        public async Task<string> AuthenticateConnection(uint appId, uint depotId, CDNClient.Server server)
         {
             var host = steamSession.ResolveCDNTopLevelHost(server.Host);
             var cdnKey = $"{depotId:D}:{host}";
@@ -140,39 +173,14 @@ namespace DepotDownloader
             }
         }
 
-        private CDNClient.Server BuildConnection(CancellationToken token)
-        {
-            if (availableServerEndpoints.Count < ServerEndpointMinimumSize)
-            {
-                populatePoolEvent.Set();
-            }
-
-            return availableServerEndpoints.Take(token);
-        }
-
-        public async Task<Tuple<CDNClient.Server, string>> GetConnectionForDepot(uint appId, uint depotId, CancellationToken token)
-        {
-            // Take a free connection from the connection pool
-            // If there were no free connections, create a new one from the server list
-            if (!activeConnectionPool.TryTake(out var server))
-            {
-                server = BuildConnection(token);
-            }
-
-            // If we don't have a CDN token yet for this server and depot, fetch one now
-            var cdnToken = await AuthenticateConnection(appId, depotId, server);
-
-            return Tuple.Create(server, cdnToken);
-        }
-
-        public void ReturnConnection(Tuple<CDNClient.Server, string> server)
+        public void ReturnConnection(CDNClient.Server server)
         {
             if (server == null) return;
 
-            activeConnectionPool.Add(server.Item1);
+            activeConnectionPool.Push(server);
         }
 
-        public void ReturnBrokenConnection(Tuple<CDNClient.Server, string> server)
+        public void ReturnBrokenConnection(CDNClient.Server server)
         {
             if (server == null) return;
 

DepotDownloader/Program.cs

@@ -108,7 +108,7 @@ namespace DepotDownloader
 
             ContentDownloader.Config.VerifyAll = HasParameter( args, "-verify-all" ) || HasParameter( args, "-verify_all" ) || HasParameter( args, "-validate" );
             ContentDownloader.Config.MaxServers = GetParameter<int>( args, "-max-servers", 20 );
-            ContentDownloader.Config.MaxDownloads = GetParameter<int>( args, "-max-downloads", 4 );
+            ContentDownloader.Config.MaxDownloads = GetParameter<int>( args, "-max-downloads", 8 );
             ContentDownloader.Config.MaxServers = Math.Max( ContentDownloader.Config.MaxServers, ContentDownloader.Config.MaxDownloads );
             ContentDownloader.Config.LoginID = HasParameter( args, "-loginid" ) ? (uint?)GetParameter<uint>( args, "-loginid" ) : null;
 
@@ -341,8 +341,8 @@ namespace DepotDownloader
             Console.WriteLine();
             Console.WriteLine( "\t-manifest-only\t\t\t- downloads a human readable manifest for any depots that would be downloaded." );
             Console.WriteLine( "\t-cellid <#>\t\t\t\t- the overridden CellID of the content server to download from." );
-            Console.WriteLine( "\t-max-servers <#>\t\t- maximum number of content servers to use. (default: 8)." );
-            Console.WriteLine( "\t-max-downloads <#>\t\t- maximum number of chunks to download concurrently. (default: 4)." );
+            Console.WriteLine( "\t-max-servers <#>\t\t- maximum number of content servers to use. (default: 20)." );
+            Console.WriteLine( "\t-max-downloads <#>\t\t- maximum number of chunks to download concurrently. (default: 8)." );
             Console.WriteLine( "\t-loginid <#>\t\t- a unique 32-bit integer Steam LogonID in decimal, required if running multiple instances of DepotDownloader concurrently." );
         }
     }

DepotDownloader/Steam3Session.cs

@@ -129,16 +129,24 @@ namespace DepotDownloader
         }
 
         public delegate bool WaitCondition();
+        private object steamLock = new object();
+
         public bool WaitUntilCallback( Action submitter, WaitCondition waiter )
         {
             while ( !bAborted && !waiter() )
             {
-                submitter();
+                lock (steamLock)
+                {
+                    submitter();
+                }
 
                 int seq = this.seq;
                 do
                 {
-                    WaitForCallbacks();
+                    lock (steamLock)
+                    {
+                        WaitForCallbacks();
+                    }
                 }
                 while ( !bAborted && this.seq == seq && !waiter() );
             }
@@ -473,7 +481,7 @@ namespace DepotDownloader
 
         public void TryWaitForLoginKey()
         {
-            if ( logonDetails.Username == null || !ContentDownloader.Config.RememberPassword ) return;
+            if ( logonDetails.Username == null || !credentials.LoggedOn || !ContentDownloader.Config.RememberPassword ) return;
 
             var totalWaitPeriod = DateTime.Now.AddSeconds( 3 );
 
@@ -584,7 +592,7 @@ namespace DepotDownloader
                     }
                     else
                     {
-                        Console.WriteLine( "Login key was expired. Please enter your password: " );
+                        Console.Write( "Login key was expired. Please enter your password: " );
                         logonDetails.Password = Util.ReadPassword();
                     }
                 }

DepotDownloader/Util.cs

@@ -5,6 +5,7 @@ using System.Linq;
 using System.Runtime.InteropServices;
 using System.Security.Cryptography;
 using System.Text;
+using System.Threading.Tasks;
 
 namespace DepotDownloader
 {
@@ -132,5 +133,38 @@ namespace DepotDownloader
                 ( sb, v ) => sb.Append( v.ToString( "x2" ) )
                 ).ToString();
         }
+        
+        public static async Task InvokeAsync(IEnumerable<Func<Task>> taskFactories, int maxDegreeOfParallelism)
+        {
+            if (taskFactories == null) throw new ArgumentNullException(nameof(taskFactories));
+            if (maxDegreeOfParallelism <= 0) throw new ArgumentException(nameof(maxDegreeOfParallelism));
+
+            Func<Task>[] queue = taskFactories.ToArray();
+
+            if (queue.Length == 0)
+            {
+                return;
+            }
+
+            List<Task> tasksInFlight = new List<Task>(maxDegreeOfParallelism);
+            int index = 0;
+
+            do
+            {
+                while (tasksInFlight.Count < maxDegreeOfParallelism && index < queue.Length)
+                {
+                    Func<Task> taskFactory = queue[index++];
+
+                    tasksInFlight.Add(taskFactory());
+                }
+
+                Task completedTask = await Task.WhenAny(tasksInFlight).ConfigureAwait(false);
+
+                await completedTask.ConfigureAwait(false);
+
+                tasksInFlight.Remove(completedTask);
+            }
+            while (index < queue.Length || tasksInFlight.Count != 0);
+        }
     }
 }

README.md

@@ -50,6 +50,6 @@ Parameter | Description
 -validate				| Include checksum verification of files already downloaded
 -manifest-only			| downloads a human readable manifest for any depots that would be downloaded.
 -cellid \<#>			| the overridden CellID of the content server to download from.
--max-servers \<#>		| maximum number of content servers to use. (default: 8).
--max-downloads \<#>		| maximum number of chunks to download concurrently. (default: 4).
+-max-servers \<#>		| maximum number of content servers to use. (default: 20).
+-max-downloads \<#>		| maximum number of chunks to download concurrently. (default: 8).
 -loginid \<#>			| a unique 32-bit integer Steam LogonID in decimal, required if running multiple instances of DepotDownloader concurrently.