15833572503

Committed 23 Jun 2025 07:42PM UTC coverage: 42.236% (+0.1%) from 42.099%

Build # 15833572503

Build Type

push

github

Committed by

HebaruSan

Commit Message

Merge #4398 Exception handling revamp, parallel multi-host inflation

Run Details

3881 of 9479 branches covered (40.94%)

Branch coverage included in aggregate %.

48 of 137 new or added lines in 30 files covered. (35.04%)

12 existing lines in 6 files now uncovered.

8334 of 19442 relevant lines covered (42.87%)

3.51 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.73

/Core/Net/NetAsyncModulesDownloader.cs

using System;
using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Threading.Tasks;
using System.IO;
using System.Linq;
using System.Threading;
using System.Security.Cryptography;
using System.Runtime.ExceptionServices;

using log4net;
using Autofac;

using CKAN.Configuration;
using CKAN.Extensions;

namespace CKAN
{
    /// <summary>
    /// Download lots of files at once!
    /// </summary>
    public class NetAsyncModulesDownloader : IDownloader
    {
        public event Action<CkanModule, long, long>? DownloadProgress;
        public event Action<ByteRateCounter>?        OverallDownloadProgress;
        public event Action<CkanModule, long, long>? StoreProgress;
        public event Action<CkanModule>?             OneComplete;
        public event Action?                         AllComplete;

        /// <summary>
        /// Returns a perfectly boring NetAsyncModulesDownloader.
        /// </summary>
        public NetAsyncModulesDownloader(IUser             user,
                                         NetModuleCache    cache,
                                         string?           userAgent   = null,
                                         CancellationToken cancelToken = default)
        {
            modules    = new List<CkanModule>();
            downloader = new NetAsyncDownloader(user, SHA256.Create, userAgent, cancelToken);
            // Schedule us to process each module on completion.
            downloader.onOneCompleted += ModuleDownloadComplete;
            downloader.TargetProgress += (target, remaining, total) =>
            {
                if (targetModules?[target].First() is CkanModule mod)
                {
                    DownloadProgress?.Invoke(mod, remaining, total);
                }
            };
            downloader.OverallProgress += brc => OverallDownloadProgress?.Invoke(brc);
            this.cache = cache;
            this.cancelToken = cancelToken;
        }

        internal NetAsyncDownloader.DownloadTarget TargetFromModuleGroup(
                HashSet<CkanModule> group,
                string?[]           preferredHosts)
            => TargetFromModuleGroup(group, group.OrderBy(m => m.identifier).First(), preferredHosts);

        private NetAsyncDownloader.DownloadTargetFile TargetFromModuleGroup(
                HashSet<CkanModule> group,
                CkanModule          first,
                string?[]           preferredHosts)
            => new NetAsyncDownloader.DownloadTargetFile(
                group.SelectMany(mod => mod.download ?? Enumerable.Empty<Uri>())
                     .Concat(group.Select(mod => mod.InternetArchiveDownload)
                                  .OfType<Uri>()
                                  .OrderBy(uri => uri.ToString()))
                     .Distinct()
                     .OrderBy(u => u,
                              new PreferredHostUriComparer(preferredHosts))
                     .ToList(),
                cache.GetInProgressFileName(first)?.FullName,
                first.download_size,
                string.IsNullOrEmpty(first.download_content_type)
                    ? defaultMimeType
                    : $"{first.download_content_type};q=1.0,{defaultMimeType};q=0.9");

        /// <summary>
        /// <see cref="IDownloader.DownloadModules"/>
        /// </summary>
        public void DownloadModules(IEnumerable<CkanModule> modules)
        {
            var activeURLs = this.modules.SelectMany(m => m.download ?? Enumerable.Empty<Uri>())
                                         .OfType<Uri>()
                                         .ToHashSet();
            var moduleGroups = CkanModule.GroupByDownloads(modules);
            // Make sure we have enough space to download and cache
            cache.CheckFreeSpace(moduleGroups.Sum(grp => grp.First().download_size));
            // Add all the requested modules
            this.modules.AddRange(moduleGroups.SelectMany(grp => grp));

            var preferredHosts = ServiceLocator.Container.Resolve<IConfiguration>().PreferredHosts;
            targetModules = moduleGroups
                // Skip any group that already has a URL in progress
                .Where(grp => grp.All(mod => mod.download?.All(dlUri => !activeURLs.Contains(dlUri)) ?? false))
                // Each group gets one target containing all the URLs
                .ToDictionary(grp => TargetFromModuleGroup(grp, preferredHosts),
                              grp => grp.ToArray());
            try
            {
                // Start the downloads!
                downloader.DownloadAndWait(targetModules.Keys);
                this.modules.Clear();
                targetModules.Clear();
                AllComplete?.Invoke();
            }
            catch (DownloadErrorsKraken kraken)
            {
                // Associate the errors with the affected modules
                var exc = new ModuleDownloadErrorsKraken(
                    kraken.Exceptions
                          .SelectMany(kvp => targetModules[kvp.Key]
                                             .Select(m => new KeyValuePair<CkanModule, Exception>(
                                                              m, kvp.Value.GetBaseException() ?? kvp.Value)))
                          .ToList());
                // Clear this.modules because we're done with these
                this.modules.Clear();
                targetModules.Clear();
                throw exc;
            }
        }

        public IEnumerable<CkanModule> ModulesAsTheyFinish(ICollection<CkanModule> cached,
                                                           ICollection<CkanModule> toDownload)
        {
            var (dlTask, blockingQueue) = DownloadsCollection(toDownload);
            return ModulesAsTheyFinish(cached, dlTask, blockingQueue);
        }

        private static IEnumerable<CkanModule> ModulesAsTheyFinish(ICollection<CkanModule>        cached,
                                                                   Task                           dlTask,
                                                                   BlockingCollection<CkanModule> blockingQueue)
        {
            foreach (var m in cached)
            {
                yield return m;
            }
            foreach (var m in blockingQueue.GetConsumingEnumerable())
            {
                yield return m;
            }
            blockingQueue.Dispose();
            try
            {
                dlTask.Wait();
                if (dlTask.Exception is AggregateException { InnerException: Exception exc })
                {
                    ExceptionDispatchInfo.Capture(exc).Throw();
                }
            }
            catch (AggregateException agExc) when (agExc is { InnerException: Exception exc })
            {
                ExceptionDispatchInfo.Capture(exc).Throw();
            }
        }

        private (Task dlTask, BlockingCollection<CkanModule> blockingQueue) DownloadsCollection(ICollection<CkanModule> toDownload)
        {
            var blockingQueue = new BlockingCollection<CkanModule>(new ConcurrentQueue<CkanModule>());
            Action<CkanModule> oneComplete = m => blockingQueue.Add(m);
            OneComplete += oneComplete;
            return (Task.Run(() => DownloadModules(toDownload))
                        .ContinueWith(t =>
                                      {
                                          blockingQueue.CompleteAdding();
                                          OneComplete -= oneComplete;
                                          if (t.Exception is AggregateException { InnerException: Exception exc })
                                          {
                                              ExceptionDispatchInfo.Capture(exc).Throw();
                                          }
                                      }),
                    blockingQueue);
        }

        private void ModuleDownloadComplete(NetAsyncDownloader.DownloadTarget target,
                                            Exception?                        error,
                                            string?                           etag,
                                            string?                           sha256)
        {
            if (target is NetAsyncDownloader.DownloadTargetFile fileTarget && targetModules != null)
            {
                var url      = fileTarget.urls.First();
                var filename = fileTarget.filename;

                log.DebugFormat("Received download completion: {0}, {1}, {2}",
                                url, filename, error?.Message);
                if (error != null)
                {
                    // If there was an error in DOWNLOADING, keep the file so we can retry it later
                    log.Info(error.Message);
                }
                else
                {
                    // Cache if this download succeeded
                    CkanModule? module = null;
                    try
                    {
                        var completedMods = targetModules[fileTarget];
                        module = completedMods.First();

                        // Check hash if defined in module
                        if (module.download_hash?.sha256 != null
                            && sha256 != module.download_hash.sha256)
                        {
                            throw new InvalidModuleFileKraken(
                                module, filename,
                                string.Format(Properties.Resources.NetModuleCacheMismatchSHA256,
                                              module, filename,
                                              sha256, module.download_hash.sha256));
                        }

                        User.RaiseMessage(Properties.Resources.NetAsyncDownloaderValidating,
                                          module.name);
                        var fileSize = new FileInfo(filename).Length;
                        cache.Store(module, filename,
                                    new ProgressImmediate<long>(bytes => StoreProgress?.Invoke(module,
                                                                                               fileSize - bytes,
                                                                                               fileSize)),
                                    module.StandardName(),
                                    false,
                                    cancelToken);
                        File.Delete(filename);
                        foreach (var m in completedMods)
                        {
                            OneComplete?.Invoke(m);
                        }
                    }
                    catch (InvalidModuleFileKraken)
                    {
                        if (module != null)
                        {
                            // Finish out the progress bar
                            StoreProgress?.Invoke(module, 0, 100);
                        }
                        // If there was an error in STORING, delete the file so we can try it from scratch later
                        File.Delete(filename);

                        // Tell downloader there is a problem with this file
                        throw;
                    }
                    catch (OperationCanceledException exc)
                    {
                        log.WarnFormat("Cancellation token threw, validation incomplete: {0}", filename);
                        User.RaiseMessage("{0}", exc.Message);
                        if (module != null)
                        {
                            // Finish out the progress bar
                            StoreProgress?.Invoke(module, 0, 100);
                        }
                        // Don't delete because there might be nothing wrong
                    }
                    catch (FileNotFoundException e)
                    {
                        log.WarnFormat("cache.Store(): FileNotFoundException: {0}", e.Message);
                    }
                    catch (InvalidOperationException)
                    {
                        log.WarnFormat("No module found for completed URL: {0}", url);
                    }
                }
            }
        }

        private static readonly ILog log = LogManager.GetLogger(typeof(NetAsyncModulesDownloader));

        private const    string                  defaultMimeType = "application/octet-stream";

        private readonly List<CkanModule>         modules;
        private          IDictionary<NetAsyncDownloader.DownloadTarget, CkanModule[]>? targetModules;
        private readonly NetAsyncDownloader       downloader;
        private          IUser                    User => downloader.User;
        private readonly NetModuleCache           cache;
        private readonly CancellationToken        cancelToken;
    }
}

1	using System;
2	using System.Collections.Generic;
3	using System.Collections.Concurrent;
4	using System.Threading.Tasks;
5	using System.IO;
6	using System.Linq;
7	using System.Threading;
8	using System.Security.Cryptography;
9	using System.Runtime.ExceptionServices;
10
11	using log4net;
12	using Autofac;
13
14	using CKAN.Configuration;
15	using CKAN.Extensions;
16
17	namespace CKAN
18	{
19	/// <summary>
20	/// Download lots of files at once!
21	/// </summary>
22	public class NetAsyncModulesDownloader : IDownloader
23	{
24	public event Action<CkanModule, long, long>? DownloadProgress;
25	public event Action<ByteRateCounter>? OverallDownloadProgress;
26	public event Action<CkanModule, long, long>? StoreProgress;
27	public event Action<CkanModule>? OneComplete;
28	public event Action? AllComplete;
29
30	/// <summary>
31	/// Returns a perfectly boring NetAsyncModulesDownloader.
32	/// </summary>
33	public NetAsyncModulesDownloader(IUser user,	8✔
34	NetModuleCache cache,
35	string? userAgent = null,
36	CancellationToken cancelToken = default)
37	{	8✔
38	modules = new List<CkanModule>();	8✔
39	downloader = new NetAsyncDownloader(user, SHA256.Create, userAgent, cancelToken);	8✔
40	// Schedule us to process each module on completion.
41	downloader.onOneCompleted += ModuleDownloadComplete;	8✔
42	downloader.TargetProgress += (target, remaining, total) =>	8✔
43	{	8✔
44	if (targetModules?[target].First() is CkanModule mod)	8!
45	{	8✔
46	DownloadProgress?.Invoke(mod, remaining, total);	8!
47	}	8✔
48	};	8✔
49	downloader.OverallProgress += brc => OverallDownloadProgress?.Invoke(brc);	8✔
50	this.cache = cache;	8✔
51	this.cancelToken = cancelToken;	8✔
52	}	8✔
53
54	internal NetAsyncDownloader.DownloadTarget TargetFromModuleGroup(
55	HashSet<CkanModule> group,
56	string?[] preferredHosts)
57	=> TargetFromModuleGroup(group, group.OrderBy(m => m.identifier).First(), preferredHosts);	8✔
58
59	private NetAsyncDownloader.DownloadTargetFile TargetFromModuleGroup(
60	HashSet<CkanModule> group,
61	CkanModule first,
62	string?[] preferredHosts)
63	=> new NetAsyncDownloader.DownloadTargetFile(	8✔
64	group.SelectMany(mod => mod.download ?? Enumerable.Empty<Uri>())	8!
65	.Concat(group.Select(mod => mod.InternetArchiveDownload)	8✔
66	.OfType<Uri>()
67	.OrderBy(uri => uri.ToString()))	8✔
68	.Distinct()
69	.OrderBy(u => u,	8✔
70	new PreferredHostUriComparer(preferredHosts))
71	.ToList(),
72	cache.GetInProgressFileName(first)?.FullName,
73	first.download_size,
74	string.IsNullOrEmpty(first.download_content_type)
75	? defaultMimeType
76	: $"{first.download_content_type};q=1.0,{defaultMimeType};q=0.9");
77
78	/// <summary>
79	/// <see cref="IDownloader.DownloadModules"/>
80	/// </summary>
81	public void DownloadModules(IEnumerable<CkanModule> modules)
82	{	8✔
83	var activeURLs = this.modules.SelectMany(m => m.download ?? Enumerable.Empty<Uri>())	4!
84	.OfType<Uri>()
85	.ToHashSet();
86	var moduleGroups = CkanModule.GroupByDownloads(modules);	8✔
87	// Make sure we have enough space to download and cache
88	cache.CheckFreeSpace(moduleGroups.Sum(grp => grp.First().download_size));	8✔
89	// Add all the requested modules
90	this.modules.AddRange(moduleGroups.SelectMany(grp => grp));	8✔
91
92	var preferredHosts = ServiceLocator.Container.Resolve<IConfiguration>().PreferredHosts;	8✔
93	targetModules = moduleGroups	8✔
94	// Skip any group that already has a URL in progress
95	.Where(grp => grp.All(mod => mod.download?.All(dlUri => !activeURLs.Contains(dlUri)) ?? false))	8!
96	// Each group gets one target containing all the URLs
97	.ToDictionary(grp => TargetFromModuleGroup(grp, preferredHosts),	8✔
98	grp => grp.ToArray());	8✔
99	try
100	{	8✔
101	// Start the downloads!
102	downloader.DownloadAndWait(targetModules.Keys);	8✔
103	this.modules.Clear();	8✔
104	targetModules.Clear();	8✔
105	AllComplete?.Invoke();	8!
106	}	8✔
107	catch (DownloadErrorsKraken kraken)	8✔
108	{	8✔
109	// Associate the errors with the affected modules
110	var exc = new ModuleDownloadErrorsKraken(	8✔
111	kraken.Exceptions
112	.SelectMany(kvp => targetModules[kvp.Key]	8✔
113	.Select(m => new KeyValuePair<CkanModule, Exception>(	8✔
114	m, kvp.Value.GetBaseException() ?? kvp.Value)))
115	.ToList());
116	// Clear this.modules because we're done with these
117	this.modules.Clear();	8✔
118	targetModules.Clear();	8✔
119	throw exc;	8✔
120	}
121	}	8✔
122
123	public IEnumerable<CkanModule> ModulesAsTheyFinish(ICollection<CkanModule> cached,
124	ICollection<CkanModule> toDownload)
125	{	8✔
126	var (dlTask, blockingQueue) = DownloadsCollection(toDownload);	8✔
127	return ModulesAsTheyFinish(cached, dlTask, blockingQueue);	8✔
128	}	8✔
129
130	private static IEnumerable<CkanModule> ModulesAsTheyFinish(ICollection<CkanModule> cached,
131	Task dlTask,
132	BlockingCollection<CkanModule> blockingQueue)
133	{	8✔
134	foreach (var m in cached)	16!
135	{	×
136	yield return m;	×
137	}	×
138	foreach (var m in blockingQueue.GetConsumingEnumerable())	20✔
139	{	8✔
140	yield return m;	8✔
141	}	8✔
142	blockingQueue.Dispose();	8✔
143	try
144	{	8✔
145	dlTask.Wait();	8✔
146	if (dlTask.Exception is AggregateException { InnerException: Exception exc })	8✔
147	{	×
148	ExceptionDispatchInfo.Capture(exc).Throw();	×
149	}	×
150	}	8✔
151	catch (AggregateException agExc) when (agExc is { InnerException: Exception exc })	8✔
152	{	8✔
153	ExceptionDispatchInfo.Capture(exc).Throw();	8✔
154	}	×
155	}	8✔
156
157	private (Task dlTask, BlockingCollection<CkanModule> blockingQueue) DownloadsCollection(ICollection<CkanModule> toDownload)
158	{	8✔
159	var blockingQueue = new BlockingCollection<CkanModule>(new ConcurrentQueue<CkanModule>());	8✔
160	Action<CkanModule> oneComplete = m => blockingQueue.Add(m);	8✔
161	OneComplete += oneComplete;	8✔
162	return (Task.Run(() => DownloadModules(toDownload))	8✔
163	.ContinueWith(t =>
164	{	8✔
165	blockingQueue.CompleteAdding();	8✔
166	OneComplete -= oneComplete;	8✔
167	if (t.Exception is AggregateException { InnerException: Exception exc })	8✔
168	{	8✔
169	ExceptionDispatchInfo.Capture(exc).Throw();	8✔
NEW 170	}	×
171	}),	8✔
172	blockingQueue);
173	}	8✔
174
175	private void ModuleDownloadComplete(NetAsyncDownloader.DownloadTarget target,
176	Exception? error,
177	string? etag,
178	string? sha256)
179	{	8✔
180	if (target is NetAsyncDownloader.DownloadTargetFile fileTarget && targetModules != null)	8!
181	{	8✔
182	var url = fileTarget.urls.First();	8✔
183	var filename = fileTarget.filename;	8✔
184
185	log.DebugFormat("Received download completion: {0}, {1}, {2}",	8✔
186	url, filename, error?.Message);
187	if (error != null)	8✔
188	{	8✔
189	// If there was an error in DOWNLOADING, keep the file so we can retry it later
190	log.Info(error.Message);	8✔
191	}	8✔
192	else
193	{	8✔
194	// Cache if this download succeeded
195	CkanModule? module = null;	8✔
196	try
197	{	8✔
198	var completedMods = targetModules[fileTarget];	8✔
199	module = completedMods.First();	8✔
200
201	// Check hash if defined in module
202	if (module.download_hash?.sha256 != null	8!
203	&& sha256 != module.download_hash.sha256)
204	{	×
205	throw new InvalidModuleFileKraken(	×
206	module, filename,
207	string.Format(Properties.Resources.NetModuleCacheMismatchSHA256,
208	module, filename,
209	sha256, module.download_hash.sha256));
210	}
211
212	User.RaiseMessage(Properties.Resources.NetAsyncDownloaderValidating,	8✔
213	module.name);
214	var fileSize = new FileInfo(filename).Length;	8✔
215	cache.Store(module, filename,	8✔
216	new ProgressImmediate<long>(bytes => StoreProgress?.Invoke(module,	8✔
217	fileSize - bytes,
218	fileSize)),
219	module.StandardName(),
220	false,
221	cancelToken);
222	File.Delete(filename);	8✔
223	foreach (var m in completedMods)	20✔
224	{	8✔
225	OneComplete?.Invoke(m);	8!
226	}	8✔
227	}	8✔
228	catch (InvalidModuleFileKraken)	8✔
229	{	8✔
230	if (module != null)	8!
231	{	8✔
232	// Finish out the progress bar
233	StoreProgress?.Invoke(module, 0, 100);	8!
234	}	8✔
235	// If there was an error in STORING, delete the file so we can try it from scratch later
236	File.Delete(filename);	8✔
237
238	// Tell downloader there is a problem with this file
239	throw;	8✔
240	}
241	catch (OperationCanceledException exc)	×
242	{	×
243	log.WarnFormat("Cancellation token threw, validation incomplete: {0}", filename);	×
244	User.RaiseMessage("{0}", exc.Message);	×
245	if (module != null)	×
246	{	×
247	// Finish out the progress bar
248	StoreProgress?.Invoke(module, 0, 100);	×
249	}	×
250	// Don't delete because there might be nothing wrong
251	}	×
252	catch (FileNotFoundException e)	×
253	{	×
254	log.WarnFormat("cache.Store(): FileNotFoundException: {0}", e.Message);	×
255	}	×
256	catch (InvalidOperationException)	×
257	{	×
258	log.WarnFormat("No module found for completed URL: {0}", url);	×
259	}	×
260	}	8✔
261	}	8✔
262	}	8✔
263
264	private static readonly ILog log = LogManager.GetLogger(typeof(NetAsyncModulesDownloader));	8✔
265
266	private const string defaultMimeType = "application/octet-stream";
267
268	private readonly List<CkanModule> modules;
269	private IDictionary<NetAsyncDownloader.DownloadTarget, CkanModule[]>? targetModules;
270	private readonly NetAsyncDownloader downloader;
271	private IUser User => downloader.User;	8✔
272	private readonly NetModuleCache cache;
273	private readonly CancellationToken cancelToken;
274	}
275	}

KSP-CKAN / CKAN / 15833572503

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous