From 088f44272f4067bff5983aa76bc705193c351158 Mon Sep 17 00:00:00 2001 From: "Collin M. Barrett" Date: Thu, 27 Jun 2019 22:13:45 -0500 Subject: [PATCH] wip fix TPL Data Flow batch downloads --- .../Entities/Aggregates/ListDownload.cs | 17 ++++++++++++ .../ListArchiver/DownloadList.cs | 23 ++++++++-------- .../ListArchiver/DownloadLists.cs | 27 +++++++++++++------ .../DownloadTxt.cs | 27 +++++++++---------- 4 files changed, 60 insertions(+), 34 deletions(-) create mode 100644 src/FilterLists.Agent/Entities/Aggregates/ListDownload.cs diff --git a/src/FilterLists.Agent/Entities/Aggregates/ListDownload.cs b/src/FilterLists.Agent/Entities/Aggregates/ListDownload.cs new file mode 100644 index 000000000..717b31b7b --- /dev/null +++ b/src/FilterLists.Agent/Entities/Aggregates/ListDownload.cs @@ -0,0 +1,17 @@ +using System.Net.Http; + +namespace FilterLists.Agent.Entities.Aggregates +{ + public class ListDownload + { + public ListDownload(ListInfo listInfo, HttpResponseMessage httpResponseMessage) + { + ListInfo = listInfo; + HttpResponseMessage = httpResponseMessage; + } + + public ListInfo ListInfo { get; } + + public HttpResponseMessage HttpResponseMessage { get; } + } +} \ No newline at end of file diff --git a/src/FilterLists.Agent/ListArchiver/DownloadList.cs b/src/FilterLists.Agent/ListArchiver/DownloadList.cs index 29c840eac..a8b73cd79 100644 --- a/src/FilterLists.Agent/ListArchiver/DownloadList.cs +++ b/src/FilterLists.Agent/ListArchiver/DownloadList.cs @@ -4,6 +4,7 @@ using System.Threading; using System.Threading.Tasks; using FilterLists.Agent.Entities; +using FilterLists.Agent.Entities.Aggregates; using FilterLists.Agent.ListArchiver.DownloadRequestsByFileExtension; using MediatR; using Microsoft.Extensions.Logging; @@ -14,18 +15,18 @@ public static class DownloadList { public class Command : IRequest { - public Command(ListInfo listInfo) + public Command(ListDownload listDownload) { - ListInfo = listInfo; + ListDownload = listDownload; } - public ListInfo ListInfo { get; } + public ListDownload ListDownload { get; } } public class Handler : AsyncRequestHandler { - private static readonly Dictionary> DownloadRequestsByFileExtension - = new Dictionary> + private static readonly Dictionary> DownloadRequestsByFileExtension + = new Dictionary> { {"", l => new DownloadTxt.Command(l)}, {".7z", l => throw new NotImplementedException()}, @@ -67,18 +68,18 @@ protected override async Task Handle(Command request, CancellationToken cancella { try { - var extension = Path.GetExtension(request.ListInfo.ViewUrl.AbsolutePath); + var extension = Path.GetExtension(request.ListDownload.ListInfo.ViewUrl.AbsolutePath); if (DownloadRequestsByFileExtension.ContainsKey(extension)) { _logger.LogInformation( - $"Downloading list {request.ListInfo.Id} from {request.ListInfo.ViewUrl}..."); - await _mediator.Send(DownloadRequestsByFileExtension[extension].Invoke(request.ListInfo), + $"Downloading list {request.ListDownload.ListInfo.Id} from {request.ListDownload.ListInfo.ViewUrl}..."); + await _mediator.Send(DownloadRequestsByFileExtension[extension].Invoke(request.ListDownload), cancellationToken); } else { _logger.LogWarning( - $"File extension not recognized for list {request.ListInfo.Id} from {request.ListInfo.ViewUrl}."); + $"File extension not recognized for list {request.ListDownload.ListInfo.Id} from {request.ListDownload.ListInfo.ViewUrl}."); } //TODO: upsert into MariaDB Rules table https://stackoverflow.com/questions/15271202/mysql-load-data-infile-with-on-duplicate-key-update @@ -86,12 +87,12 @@ await _mediator.Send(DownloadRequestsByFileExtension[extension].Invoke(request.L catch (NotImplementedException) { _logger.LogWarning( - $"File extension not supported for list {request.ListInfo.Id} from {request.ListInfo.ViewUrl}."); + $"File extension not supported for list {request.ListDownload.ListInfo.Id} from {request.ListDownload.ListInfo.ViewUrl}."); } catch (ArgumentException ex) { _logger.LogError(ex, - $"Could not determine the file extension for list {request.ListInfo.Id} from {request.ListInfo.ViewUrl}."); + $"Could not determine the file extension for list {request.ListDownload.ListInfo.Id} from {request.ListDownload.ListInfo.ViewUrl}."); } } } diff --git a/src/FilterLists.Agent/ListArchiver/DownloadLists.cs b/src/FilterLists.Agent/ListArchiver/DownloadLists.cs index 107e48b26..fb83c43f8 100644 --- a/src/FilterLists.Agent/ListArchiver/DownloadLists.cs +++ b/src/FilterLists.Agent/ListArchiver/DownloadLists.cs @@ -1,8 +1,10 @@ using System.Collections.Generic; +using System.Net.Http; using System.Threading; using System.Threading.Tasks; using System.Threading.Tasks.Dataflow; using FilterLists.Agent.Entities; +using FilterLists.Agent.Entities.Aggregates; using MediatR; namespace FilterLists.Agent.ListArchiver @@ -21,28 +23,37 @@ public Command(IEnumerable lists) public class Handler : AsyncRequestHandler { - private const int MaxDegreeOfParallelism = 50; + private const int MaxDegreeOfParallelism = 5; + private readonly HttpClient _httpClient; private readonly IMediator _mediator; - public Handler(IMediator mediator) + public Handler(IMediator mediator, HttpClient httpClient) { _mediator = mediator; + _httpClient = httpClient; } // https://stackoverflow.com/a/22492731/2343739 protected override async Task Handle(Command request, CancellationToken cancellationToken) { - var downloader = new TransformBlock( - l => _mediator.Send(new DownloadList.Command(l), cancellationToken), - new ExecutionDataflowBlockOptions {MaxDegreeOfParallelism = MaxDegreeOfParallelism}); - var buffer = new BufferBlock(); + var downloader = new TransformBlock( + async l => new ListDownload(l, await _httpClient.GetAsync(l.ViewUrl, cancellationToken)), + new ExecutionDataflowBlockOptions {MaxDegreeOfParallelism = MaxDegreeOfParallelism} + ); + var buffer = new BufferBlock(); downloader.LinkTo(buffer); foreach (var list in request.Lists) await downloader.SendAsync(list, cancellationToken); downloader.Complete(); + + foreach (var _ in request.Lists) + { + var listDownload = await buffer.ReceiveAsync(cancellationToken); + await _mediator.Send(new DownloadList.Command(listDownload), cancellationToken); + listDownload.HttpResponseMessage.Dispose(); + } + await downloader.Completion; - if (buffer.TryReceiveAll(out var tasks)) - await Task.WhenAll(tasks); } } } diff --git a/src/FilterLists.Agent/ListArchiver/DownloadRequestsByFileExtension/DownloadTxt.cs b/src/FilterLists.Agent/ListArchiver/DownloadRequestsByFileExtension/DownloadTxt.cs index 7349eb50b..819ebece1 100644 --- a/src/FilterLists.Agent/ListArchiver/DownloadRequestsByFileExtension/DownloadTxt.cs +++ b/src/FilterLists.Agent/ListArchiver/DownloadRequestsByFileExtension/DownloadTxt.cs @@ -2,7 +2,7 @@ using System.Net.Http; using System.Threading; using System.Threading.Tasks; -using FilterLists.Agent.Entities; +using FilterLists.Agent.Entities.Aggregates; using MediatR; using Microsoft.Extensions.Logging; @@ -12,12 +12,12 @@ public static class DownloadTxt { public class Command : IRequest { - public Command(ListInfo listInfo) + public Command(ListDownload listDownload) { - ListInfo = listInfo; + ListDownload = listDownload; } - public ListInfo ListInfo { get; } + public ListDownload ListDownload { get; } } public class Handler : AsyncRequestHandler @@ -35,21 +35,18 @@ protected override async Task Handle(Command request, CancellationToken cancella { try { - using (var result = await _httpClient.GetAsync(request.ListInfo.ViewUrl, cancellationToken)) - { - if (result.IsSuccessStatusCode) - using (Stream output = - File.OpenWrite(Path.Combine("archives", $"{request.ListInfo.Id}.txt"))) - using (var input = await result.Content.ReadAsStreamAsync()) - { - input.CopyTo(output); - } - } + if (request.ListDownload.HttpResponseMessage.IsSuccessStatusCode) + using (Stream output = + File.OpenWrite(Path.Combine("archives", $"{request.ListDownload.ListInfo.Id}.txt"))) + using (var input = await request.ListDownload.HttpResponseMessage.Content.ReadAsStreamAsync()) + { + input.CopyTo(output); + } } catch (HttpRequestException ex) { _logger.LogError(ex, - $"Error downloading list {request.ListInfo.Id} from {request.ListInfo.ViewUrl}."); + $"Error downloading list {request.ListDownload.ListInfo.Id} from {request.ListDownload.ListInfo.ViewUrl}."); } } }