<commit_before>﻿using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using FilterLists.Agent.Entities;
using MediatR;

namespace FilterLists.Agent.ListArchiver
{
    public static class DownloadLists
    {
        public class Command : IRequest
        {
            public Command(IEnumerable<ListInfo> listInfo)
            {
                ListInfo = listInfo;
            }

            public IEnumerable<ListInfo> ListInfo { get; }
        }

        public class Handler : AsyncRequestHandler<Command>
        {
            private const int MaxDegreeOfParallelism = 5; //TODO: tune
            private readonly IMediator _mediator;

            public Handler(IMediator mediator)
            {
                _mediator = mediator;
            }

            protected override async Task Handle(Command request, CancellationToken cancellationToken)
            {
                var downloader = new ActionBlock<ListInfo>(
                    async l => await _mediator.Send(new DownloadList.Command(l), cancellationToken),
                    new ExecutionDataflowBlockOptions {MaxDegreeOfParallelism = MaxDegreeOfParallelism}
                );
                var orderedListInfo = ShardByHost(request.ListInfo);
                foreach (var list in orderedListInfo)
                    await downloader.SendAsync(list, cancellationToken);
                downloader.Complete();
                await downloader.Completion;
            }

            //https://keestalkstech.com/2017/10/linq-round-robin-ordering-based-segments/
            private static IEnumerable<ListInfo> ShardByHost(IEnumerable<ListInfo> listInfo)
            {
                return listInfo.GroupBy(l => l.ViewUrl.Host)
                    .SelectMany((g, gi) => g.Select((l, li) => new {Index = li, GroupIndex = gi, ListInfo = l}))
                    .OrderBy(a => a.Index)
                    .ThenBy(a => a.GroupIndex)
                    .Select(a => a.ListInfo);
            }
        }
    }
}<commit_msg>add TODO<commit_after>﻿using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using FilterLists.Agent.Entities;
using MediatR;

namespace FilterLists.Agent.ListArchiver
{
    public static class DownloadLists
    {
        public class Command : IRequest
        {
            public Command(IEnumerable<ListInfo> listInfo)
            {
                ListInfo = listInfo;
            }

            public IEnumerable<ListInfo> ListInfo { get; }
        }

        public class Handler : AsyncRequestHandler<Command>
        {
            private const int MaxDegreeOfParallelism = 5; //TODO: tune
            private readonly IMediator _mediator;

            public Handler(IMediator mediator)
            {
                _mediator = mediator;
            }

            protected override async Task Handle(Command request, CancellationToken cancellationToken)
            {
                var downloader = new ActionBlock<ListInfo>(
                    async l => await _mediator.Send(new DownloadList.Command(l), cancellationToken),
                    new ExecutionDataflowBlockOptions {MaxDegreeOfParallelism = MaxDegreeOfParallelism}
                );
                var orderedListInfo = ShardByHost(request.ListInfo);
                foreach (var list in orderedListInfo)
                    await downloader.SendAsync(list, cancellationToken);
                downloader.Complete();
                await downloader.Completion;
            }

            //https://keestalkstech.com/2017/10/linq-round-robin-ordering-based-segments/
            //TODO: space out overweighted hosts throughout rather than having them all at the end of the queue
            private static IEnumerable<ListInfo> ShardByHost(IEnumerable<ListInfo> listInfo)
            {
                return listInfo.GroupBy(l => l.ViewUrl.Host)
                    .SelectMany((g, gi) => g.Select((l, li) => new {Index = li, GroupIndex = gi, ListInfo = l}))
                    .OrderBy(a => a.Index)
                    .ThenBy(a => a.GroupIndex)
                    .Select(a => a.ListInfo);
            }
        }
    }
}
{'fraction_non_alphanumeric': 0.08211211909746452, 'fraction_numerical': 0.0032565712956501513, 'mean_word_length': 1.8457974851091992, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 29, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3949641', 'n_tokens_mistral': 1128, 'n_tokens_neox': 1075, 'n_words': 248}