Fetching huge datasets through bottlenecked output/endpoint using divide and conquer multi-threading. C#

Question

I have been put into the situation that i have to probe a relatively huge dataset through a chocked output which is an id based single record returning endpoint. Iterating this endpoint to scrape all the data sequentially is not obviously a practical option so I tried to create parallelized multiple processes each has its own range of ids to deal with. Here is what i managed to do which is still not working perfectly. "Refer to the questions at the bottom"

Code Preface ↴

In this demo I used http://numbersapi.com/. It is simple you put a number and the API returns a random fact about it from the facts associated with it, If there are any otherwise It returns randomized one of 4 phrases containing the facts-free number.
API Examples:
Number that has atleast one fact → http://numbersapi.com/1
Number that has no facts associated → http://numbersapi.com/123456789

Appendix

ThreadRange : Items to be processed by a single thread

Code ↴

using System;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.IO;
using System.Net;
using System.Net.Http;
using System.Net.Http.Headers;
using System.Security.Authentication;
using System.Text;
using System.Text.RegularExpressions;
using System.Threading;
using System.Threading.Tasks;

namespace CSharp
{
    class Program
    {
        private static Stopwatch CrawlerStopWatch = new Stopwatch();
        private static HttpClient HttpClientCrawler = null;
        private static HttpClientHandler HttpClientCrawlerHttpsHandler = new HttpClientHandler() { SslProtocols = SslProtocols.Tls12 | SslProtocols.Tls11 | SslProtocols.Tls };
        private static BlockingCollection<Thread> CrawlerThreads = new BlockingCollection<Thread>();
        private static BlockingCollection<String> CrawlerFailuresBlockingCollection = new BlockingCollection<String>();
        private static BlockingCollection<String> CrawlerHitsBlockingCollection = new BlockingCollection<String>();
        private static BlockingCollection<String> CrawlerMissesBlockingCollection = new BlockingCollection<String>();
        private static String BaseUrl = "http://numbersapi.com/";
        private static String FullUrl = String.Concat(BaseUrl, "{0}");
        private static long ThreadsCount = 1;
        private static long ThreadRange = 10;
        private static long Offset = 2;
        private static long ItemsToProcess = ThreadsCount * ThreadRange;
        private static int MaxUrlLength = String.Format(FullUrl, ItemsToProcess).Length;
        private static ReaderWriterLockSlim CrawlerReaderWriterLockSlim = new ReaderWriterLockSlim();
        private static String CrawlerResultantFileName = "z.CrawlerResult.txt";
        public static void Main(String[] args)
        {
            CrawlerStopWatch.Start();
            //### Managing HttpClient ###/
            ServicePointManager.DefaultConnectionLimit = 50;
            if (Regex.IsMatch(BaseUrl, @"^https.*$")) { HttpClientCrawler = new HttpClient(HttpClientCrawlerHttpsHandler); } else { HttpClientCrawler = new HttpClient(); }
            HttpClientCrawler.BaseAddress = new Uri(BaseUrl);
            HttpClientCrawler.DefaultRequestHeaders.Accept.Add(new MediaTypeWithQualityHeaderValue("text/html"));
            //### Managing HttpClient ###/
            InitiateCrawler(BaseUrl, ThreadsCount, ThreadRange, Offset);
        }
        //### Crawler Methods ###
        public static void InitiateCrawler(String BaseUrl, long ThreadsCount, long ThreadRange, long Offset = 0)
        {
            Console.WriteLine("###############################");
            Console.WriteLine($"Commenced crawling the <{BaseUrl}> endpoint and working on <{ItemsToProcess}> items by creating <{ThreadsCount}> threads which each working on <{ThreadRange}> items.");
            while (CrawlerThreads.Count < ThreadsCount)
            {
                long Start = Offset + ThreadRange * CrawlerThreads.Count + 1;
                long End = Offset + ThreadRange * (CrawlerThreads.Count + 1);
                Thread CrawlerThread = new Thread(() => Crawl(Start, End));
                CrawlerThreads.Add(CrawlerThread);
                CrawlerThread.Start();
            }
            Task FinalizeCrawlerTask = Task.Run(() => { foreach (Thread CrawlerThread in CrawlerThreads) { CrawlerThread.Join(); } FinalizeCrawler(); });
            FinalizeCrawlerTask.Wait();
        }
        public static void Crawl(long Start, long End)
        {
            long Current = Start;
            while (Current <= End)
            {
                String CurrentUrlParamters = String.Format("{0}", Current);
                String CurrentUrl = $"{HttpClientCrawler.BaseAddress.AbsoluteUri}{CurrentUrlParamters}";
                String CurrentPageContent = "";
                HttpResponseMessage HttpResponseMessage = HttpClientCrawler.GetAsync(CurrentUrlParamters).Result;
                if (HttpResponseMessage.IsSuccessStatusCode)
                {
                    CurrentPageContent = Encoding.UTF8.GetString(HttpResponseMessage.Content.ReadAsByteArrayAsync().Result);
                    if (isResultRelevant(CurrentPageContent)) { HandleCrawlerRelevantResult(CurrentUrl, CurrentPageContent); } else { HandleCrawlerIrrelevantResult(CurrentUrl, CurrentPageContent); }
                }
                else
                {
                    HandleCrawlerFailure(CurrentUrl, HttpResponseMessage);
                }
                Current++;
            }
        }
        public static void HandleCrawlerFailure(String Url, HttpResponseMessage HttpResponseMessage)
        {
            CrawlerFailuresBlockingCollection.Add(Url);
            int ProcessedItems = CrawlerHitsBlockingCollection.Count + CrawlerMissesBlockingCollection.Count + CrawlerFailuresBlockingCollection.Count;
            Console.WriteLine($"[Item #{ProcessedItems.ToString().PadRight(ItemsToProcess.ToString().Length)}] {Url.PadRight(MaxUrlLength)} returned {(int)HttpResponseMessage.StatusCode} Code | {HttpResponseMessage.ReasonPhrase}");
        }
        public static Boolean isResultRelevant(String Content)
        {
            Boolean IsRelevant = true;
            String[] RegularExpressionsArray = new string[]
            {
                @"^[\d]+ is a boring number\.$",
                @"^[\d]+ is an uninteresting number\.$",
                @"^[\d]+ is an unremarkable number\.$",
                @"^[\d]+ is a number for which we're missing a fact (submit one to numbersapi at google mail!)\.$",
            };
            foreach (String RegularExpression in RegularExpressionsArray) { if (Regex.IsMatch(Content, RegularExpression)) { IsRelevant = false; break; } }
            return IsRelevant;
        }
        public static void HandleCrawlerRelevantResult(String Url, String Content)
        {
            CrawlerResultantFileWriteLine(Url);
            CrawlerHitsBlockingCollection.Add(Url);
            int ProcessedItems = CrawlerHitsBlockingCollection.Count + CrawlerMissesBlockingCollection.Count + CrawlerFailuresBlockingCollection.Count;
            Console.WriteLine($"[Item #{ProcessedItems.ToString().PadRight(ItemsToProcess.ToString().Length)}] {Url.PadRight(MaxUrlLength)} is relevant");
        }
        public static void HandleCrawlerIrrelevantResult(String Url, String Content)
        {
            CrawlerMissesBlockingCollection.Add(Url);
            int ProcessedItems = CrawlerHitsBlockingCollection.Count + CrawlerMissesBlockingCollection.Count + CrawlerFailuresBlockingCollection.Count;
            Console.WriteLine($"[Item #{ProcessedItems.ToString().PadRight(ItemsToProcess.ToString().Length)}] {Url.PadRight(MaxUrlLength)} is irrelevant");
        }
        public static void FinalizeCrawler()
        {
            CrawlerStopWatch.Stop();
            TimeSpan TimeSpan = TimeSpan.FromMilliseconds(CrawlerStopWatch.ElapsedMilliseconds);
            String TimeLapseInformation = String.Format("{0:D2}h:{1:D2}m:{2:D2}s:{3:D3}ms",
                                    TimeSpan.Hours,
                                    TimeSpan.Minutes,
                                    TimeSpan.Seconds,
                                    TimeSpan.Milliseconds);
            Console.WriteLine($"Crawling finished in {TimeLapseInformation}.");
            Console.WriteLine($"<{CrawlerFailuresBlockingCollection.Count + CrawlerHitsBlockingCollection.Count + CrawlerMissesBlockingCollection.Count}> out of <{ItemsToProcess}> items have been crawled having <{CrawlerHitsBlockingCollection.Count}> relevant items, <{CrawlerMissesBlockingCollection.Count}> irrelevant items and <{CrawlerFailuresBlockingCollection.Count}> failures.");
            Console.WriteLine("###############################");
        }
        //### Crawler Methods ###
        //### Auxiliary Methods ###
        public static void CrawlerResultantFileWriteLine(String Line)
        {
            CrawlerReaderWriterLockSlim.EnterWriteLock();
            try { using (StreamWriter StreamWriter = File.AppendText(String.Concat(AppDomain.CurrentDomain.BaseDirectory, "\\", CrawlerResultantFileName))) { StreamWriter.WriteLine(Line); StreamWriter.Close(); } }
            finally { CrawlerReaderWriterLockSlim.ExitWriteLock(); }
        }
        //### Auxiliary Methods ###
    }
}

Questions

(Q1.) Is the presented approach optimal ?

---

(Q2.) In FinalizeCrawler() function,In case of the number of the collective items to process in all threads is too big, The count of the processed items don't equal the designated number of items to process in the InitializeCrawler(UrlHandle,ThreadsCount,ThreadRange) call in main as:
(CrawlerFailuresBlockingCollection.Count + CrawlerHits.Count + CrawlerMisses.Count) has to equal(ThreadsCount * ThreadRange)
Do some items get omitted? if so, Why?

(A2.) As Joshua Robinson pointed out, I was using System.Collections.Generic.List which is unsafe collection to modify from multiple threads. I switched to System.Collections.Concurrent.BlockingCollection, tested it with big number like 10000 and it worked.

---

(Q3.) How to pinpoint the optimal ThreadRange and ThreadsCount combination regarding performance. Which one?
- (A) Min ThreadsCount with max ThreadRange
- (B) Low ThreadsCount with high ThreadRange
- (C) Median thread range with Median thread count
- (B) High ThreadsCount with low ThreadRange
- (E) Max ThreadsCount with min ThreadRange
To be more elaborate, If I wanted to process 10000 Items from this endpoint these are all the possible combinations satisfying the necessary predicate ThreadsCount*ThreadRange=10000 by that I have choose only from the them.
Which is the most optimal? And why?

<1>     ThreadsCount * <10000> ThreadRange
<2>     ThreadsCount * <5000>  ThreadRange
<4>     ThreadsCount * <2500>  ThreadRange
<5>     ThreadsCount * <2000>  ThreadRange
<8>     ThreadsCount * <1250>  ThreadRange
<10>    ThreadsCount * <1000>  ThreadRange
<16>    ThreadsCount * <625>   ThreadRange
<20>    ThreadsCount * <500>   ThreadRange
<25>    ThreadsCount * <400>   ThreadRange
<40>    ThreadsCount * <250>   ThreadRange
<50>    ThreadsCount * <200>   ThreadRange
<80>    ThreadsCount * <125>   ThreadRange
<100>   ThreadsCount * <100>   ThreadRange // Exact Median → square root of items to process if integer
<125>   ThreadsCount * <80>    ThreadRange
<200>   ThreadsCount * <50>    ThreadRange
<250>   ThreadsCount * <40>    ThreadRange
<400>   ThreadsCount * <25>    ThreadRange
<500>   ThreadsCount * <20>    ThreadRange
<625>   ThreadsCount * <16>    ThreadRange
<1000>  ThreadsCount * <10>    ThreadRange
<1250>  ThreadsCount * <8>     ThreadRange
<2000>  ThreadsCount * <5>     ThreadRange
<2500>  ThreadsCount * <4>     ThreadRange
<5000>  ThreadsCount * <2>     ThreadRange
<10000> ThreadsCount * <1>     ThreadRange

Answer 1

(Q1.) Is the presented approach optimal ?

I have a suggestion how to enhance current solution. Currently all crawling threads process their http responses by themself. Moreover when they process their http responses they try to acquire an exclusive lock (ReaderWriterLockSlim.EnterWriteLock) to write results to the file.

I suggest to make crawling threads only send http requests and immediately add http responses into some collection. And another thread (lets call it processing thread) will process these http responses (extract data, parse it, write to file and etc). Such approach will give us next benefits:

1. Crawling threads will not waste their time to process http responses. They will only execute the most expensive operations: send http requests and receive http responses.
2. An exclusive lock will not be needed to process http responses. Processing thread will be able to process http responses without setting any locks.

If in your real program processing of the http responses is more time consumable than in the provided sample then these benefits will decrease the overall execution time of the program.

Here are changes that you need to make in your program to implement this suggestion:

// Now we can declare this collections as simple Lists, not BlockingCollections.
private static List<Thread> CrawlerThreads = new List<Thread>();
private static List<String> CrawlerFailuresBlockingCollection = new List<String>();
private static List<String> CrawlerHitsBlockingCollection = new List<String>();
private static List<String> CrawlerMissesBlockingCollection = new List<String>();

// I found out that for my environment this values are optimal and give better performance.
private static long ThreadsCount = 200;
private static long ThreadRange = 50;

// Add this collection. Crawling threads will add http responses into it.
// And processing thread will process them.
private static BlockingCollection<HttpResponseMessage> ResponsesToProcess = new BlockingCollection<HttpResponseMessage>();

public static void Main(String[] args)
{
   ...
   // If we use 200 crawling threads then we should set
   // DefaultConnectionLimit=200 to make possible for all
   // crawling threads to make http responses simultaneously
   // without waiting for available connections.
   ServicePointManager.DefaultConnectionLimit = 200;
   ...
}

public static void InitiateCrawler(String BaseUrl, long ThreadsCount, long ThreadRange, long Offset = 0)
{
    Console.WriteLine("###############################");
    Console.WriteLine($"Commenced crawling the <{BaseUrl}> endpoint and working on <{ItemsToProcess}> items by creating <{ThreadsCount}> threads which each working on <{ThreadRange}> items.");

    while (CrawlerThreads.Count < ThreadsCount)
    {
        long Start = Offset + ThreadRange * CrawlerThreads.Count + 1;
        long End = Offset + ThreadRange * (CrawlerThreads.Count + 1);

        Thread CrawlerThread = new Thread(() => Crawl(Start, End));
        CrawlerThreads.Add(CrawlerThread);
        CrawlerThread.Start();
    }

    Task FinalizeCrawlerTask = Task.Run(() =>
    { 
        foreach (Thread CrawlerThread in CrawlerThreads) 
        { 
            CrawlerThread.Join(); 
        }
        // Notify processing thread that there is no more
        // http responses to process.
        ResponsesToProcess.CompleteAdding();
    });

    // Processing thread.
    Task ProcessResponsesThread = Task.Run(() =>
    {
        foreach (var HttpResponseMessage in ResponsesToProcess.GetConsumingEnumerable())
        {
            string CurrentUrl = HttpResponseMessage.RequestMessage.RequestUri.AbsoluteUri;

            if (HttpResponseMessage.IsSuccessStatusCode)
            {
                string CurrentPageContent = Encoding.UTF8.GetString(HttpResponseMessage.Content.ReadAsByteArrayAsync().Result);
                if (isResultRelevant(CurrentPageContent))
                {
                    HandleCrawlerRelevantResult(CurrentUrl, CurrentPageContent);
                }
                else
                {
                    HandleCrawlerIrrelevantResult(CurrentUrl, CurrentPageContent);
                }
            }
            else
            {
                HandleCrawlerFailure(CurrentUrl, HttpResponseMessage);
            }
        }
    });

    FinalizeCrawlerTask.Wait();
    ProcessResponsesThread.Wait();

    // Now we print the results of the program here because we must
    // ensure that finalizer and processing threads have finished.
    FinalizeCrawler();
}

public static void Crawl(long Start, long End)
{
    long Current = Start;
    while (Current <= End)
    {
        String CurrentUrlParamters = String.Format("{0}", Current);
        String CurrentUrl = $"{HttpClientCrawler.BaseAddress.AbsoluteUri}{CurrentUrlParamters}";
        String CurrentPageContent = "";
        HttpResponseMessage HttpResponseMessage = HttpClientCrawler.GetAsync(CurrentUrlParamters).Result;
        // Now crawling thread only adds http response into collection
        // of http responses to process. It doesn't process http responses.
        ResponsesToProcess.Add(HttpResponseMessage);
        Current++;
    }
}

public static void CrawlerResultantFileWriteLine(String Line)
{
    // Now we don't need a lock. We can delete it.
    using (StreamWriter StreamWriter = File.AppendText(String.Concat(AppDomain.CurrentDomain.BaseDirectory, "\\", CrawlerResultantFileName))) 
    {
        StreamWriter.WriteLine(Line); 
        StreamWriter.Close(); 
    } 
}

---

(Q3.) How to pinpoint the optimal ThreadRange and ThreadsCount combination regarding performance. Which one?

I tested your sample and found out that there was no a rule of thumb how to choose ThreadsCount. Here are results of the tests on my environment (PC):

• ThreadsCount = 20, ExecutionTime = 07m:40s
• ThreadsCount = 25, ExecutionTime = 07m:00s
• ThreadsCount = 50, ExecutionTime = 03m:30s
• ThreadsCount = 100, ExecutionTime = 01m:30s
• ThreadsCount = 200, ExecutionTime = 01m:10s
• Further increasing of ThreadsCount does not improve ExecutionTime

I think that to find an optimal value of ThreadsCount you should test different values of this parameter on your environment and choose the best.

The only thing I want to highlight is that it is important to set ServicePointManager.DefaultConnectionLimit equal or larger than ThreadsCount. If ServicePointManager.DefaultConnectionLimit is less than ThreadsCount then some of the threads will wait for available connections without doing usefull work.

Answer 2

This is what I would have written to scrape a bunch of requests. It's plenty efficient. Compared to what you wrote, I also think this approach is much simpler. It relies on Tasks to take advantage of CPU parallelism (this example is not CPU bound, it is bound by the responsiveness of the remote server), and you can manually set the number of simultaneous connections to use. (I coded this with 6 which is a typical number of connections a modern web browser would allow to each domain.)

using System;
using System.Collections.Generic;
using System.Linq;
using System.Net.Http;
using System.Text;
using System.Threading.Tasks;

namespace ScrapeExample
{
    class Program
    {
        private const int NumberOfSimultaneousConnections = 6;
        private const int MaxResponseContentBufferSize = 4096;
        private const string uriTemplate = "http://numbersapi.com/{0}";
        private const int FirstIndexToRequest = 1;
        private const int LastIndexToRequest = 100;

        static void Main(string[] args)
        {
            HttpClientHandler hch = new HttpClientHandler() { Proxy = null, UseProxy = false };
            HttpClient[] clients = Enumerable.Range(0, NumberOfSimultaneousConnections).Select(i =>
                new HttpClient(hch) { MaxResponseContentBufferSize = MaxResponseContentBufferSize }
            ).ToArray();

            List<Task<string>> tasks = new List<Task<string>>();
            for (int i = FirstIndexToRequest; i <= LastIndexToRequest; ++i) {
                string uri = string.Format(uriTemplate, i);
                tasks.Add(ProcessURLAsync(uri, clients[i % NumberOfSimultaneousConnections]));
            }
            Task.WaitAll(tasks.ToArray());
            string[] results = tasks.Select(t => t.Result).ToArray();
            Console.WriteLine(string.Join(Environment.NewLine, results));
        }

        private async static Task<string> ProcessURLAsync(string uri, HttpClient client)
        {
            HttpResponseMessage response = await client.GetAsync(uri);
            response.EnsureSuccessStatusCode();
            byte[] content = await response.Content.ReadAsByteArrayAsync();
            return Encoding.UTF8.GetString(content);
        }
    }
}

IMPORTANT:

In practice, I find that that numbersapi.com will easily get overloaded after a few hundred requests and it will start returning 502 error codes (Bad Gateway / Service Temporarily Overloaded). So this design is missing something that would try each item again in the event of an error. I just wrote EnsureSuccessStatusCode which throws an exception on error. But this is already outperforming the remote API that you are using as an example.