5
\$\begingroup\$

I am working on multi-threading application to learn multithreading better.
My application has a queue with 400 unique links, 4 threads and each thread is pulling link from the queue and sending a Get request to this queue.

My application seems to run, and it takes 18 - 22 seconds for it to finish on 400 links.
I am using ThreadPool.QueueUserWorkItem to do the multithreading.

I wanted someone to review my code and tell me if I use ThreadPool in the correct way.
I need it to be efficient as it can be.

I created a similar application in Python and it takes 12 seconds to run 400 requests. Therefore, I think I am not using the multithreading ability of C# correctly.

The code:

using System;
using System.Collections.Concurrent;
using System.IO;
using System.Net;
using System.Threading;
using System.Threading.Tasks;

namespace PlayGround
{
    class Program
    {
        static void Main(string[] args)
        {
            var watch = System.Diagnostics.Stopwatch.StartNew();
            RunThreadPool();
            watch.Stop();
            var elapsedMs = watch.ElapsedMilliseconds;
            Console.WriteLine("[*] Elapsed time: {0}", elapsedMs / 1000);
            Console.ReadLine();
        }

        static BlockingCollection<string> inputQueue = new BlockingCollection<string>();

        // https://stackoverflow.com/questions/6529659/wait-for-queueuserworkitem-to-complete
        private static ManualResetEvent resetEvent = new ManualResetEvent(false);

        private static void RunThreadPool()
        {
            string url = @"https://google.com";
            for (int i = 0; i < 4; i++)
            {
                var tempWorkerId = i;
                ThreadPool.QueueUserWorkItem(state => Worker(tempWorkerId));
            }

            for (int i = 0; i < 400; ++i)
            {
                //Console.WriteLine("Queueing work item {0}", url + "/" + i);
                inputQueue.Add(url + "/" + i);
                //Thread.Sleep(50);
            }

            Console.WriteLine("Stopping adding.");
            inputQueue.CompleteAdding();

            resetEvent.WaitOne();
            Console.WriteLine("Done.");
        }

        // https://stackoverflow.com/a/22689821/2153777
        static void Worker(int workerId)
        {
            Console.WriteLine("Worker {0} is starting.", workerId);

            foreach (var workItem in inputQueue.GetConsumingEnumerable())
            {
                string res = "";
                //Console.WriteLine("Worker {0} is processing item {1}", workerId, workItem);
                try
                {
                    res = Get(workItem);
                }
                catch (Exception)
                {
                    res = "404";
                }
                //Console.WriteLine("Worker {0} is processing item {1} with result {2}", workerId, workItem, res);
                //Thread.Sleep(100);          // Simulate work.
            }
            resetEvent.Set();
            Console.WriteLine("Worker {0} is stopping.", workerId);
        }

        // https://stackoverflow.com/a/27108442/2153777
        public static string Get(string uri)
        {
            HttpStatusCode status;
            HttpWebRequest request = (HttpWebRequest)WebRequest.Create(uri);
            request.AutomaticDecompression = DecompressionMethods.GZip | DecompressionMethods.Deflate;

            using (HttpWebResponse response = (HttpWebResponse)request.GetResponse())
            {
                status = response.StatusCode;
            }

            //Thread.Sleep(2000);
            return status.ToString() + "; Thread: " + Thread.CurrentThread.ManagedThreadId.ToString();
        }
    }

}

EDIT: Using async\await was very slow (44 seconds) (4.12.2018)
Following the comments about using async\await, I used the link from Microsoft to create a version with async\await and it seems to work faster than anything else I tried.

Here is the code:

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

namespace AsyncAwait
{
    class Program
    {
        static void Main(string[] args)
        {
            SumPageSizesAsync();
            Console.ReadLine();
        }

        private static async Task SumPageSizesAsync()
        {
            var watch = System.Diagnostics.Stopwatch.StartNew();
            // Make a list of web addresses.  
            List<string> urlList = SetUpURLList();

            // Create a query.   
            IEnumerable<Task<HttpStatusCode>> downloadTasksQuery =
                from url in urlList select GetURLStatusCodeAsync(url);

            // Use ToArray to execute the query and start the download tasks.  
            Task<HttpStatusCode>[] downloadTasks = downloadTasksQuery.ToArray();

            // You can do other work here before awaiting.  

            // Await the completion of all the running tasks.  
            HttpStatusCode[] lengths = await Task.WhenAll(downloadTasks);
            watch.Stop();
            var elapsedMs = watch.ElapsedMilliseconds;
            Console.WriteLine("[*] Elapsed time: {0}", elapsedMs / 1000);
        }

        private static async Task<HttpStatusCode> GetURLStatusCodeAsync(string url)
        {
            HttpStatusCode status;

            try
            {
                var request = (HttpWebRequest)WebRequest.Create(url);
                request.AutomaticDecompression = DecompressionMethods.GZip | DecompressionMethods.Deflate;

                // Send the request to the Internet resource and wait for  
                // the response.  
                using (HttpWebResponse response = (HttpWebResponse)await request.GetResponseAsync())
                {
                    status = response.StatusCode;
                }

            }
            catch (Exception)
            {

                status = HttpStatusCode.NotFound;
            }

            //Console.WriteLine("[*] Status code: {0}, Thread: {1}, Url: {2}", status, Thread.CurrentThread.ManagedThreadId, url);

            return status;
        }

        private static List<string> SetUpURLList()
        {
            string url = @"https://google.com";
            List<string> urls = new List<string>();
            for (int i = 0; i < 400; ++i)
            {
                urls.Add(url + "/" + i);
            }

            return urls;
        }
    }
}
\$\endgroup\$
4
  • 1
    \$\begingroup\$ Have you tried the same with async/await or do just want to use threads? \$\endgroup\$
    – t3chb0t
    Dec 1, 2018 at 11:43
  • \$\begingroup\$ I didn't try it. I don't care too much. I am trying to find the fastest and correct way to use multithreading. You think it should improve the performance? \$\endgroup\$
    – E235
    Dec 1, 2018 at 12:24
  • \$\begingroup\$ async-await example docs.microsoft.com/en-us/dotnet/csharp/programming-guide/… \$\endgroup\$
    – Nkosi
    Dec 1, 2018 at 21:28
  • \$\begingroup\$ @t3chb0t @Nkosi, I used async\await (see my last edit from 4.12.2018) and it took between 36 - 44 seconds. Very slow compared to my first method. \$\endgroup\$
    – E235
    Dec 4, 2018 at 7:38

1 Answer 1

6
\$\begingroup\$

I think your use of ManualResetEvent is incorrect in this scenario, because as I read it, you intend to wait for all threads to complete here:

        resetEvent.WaitOne();
        Console.WriteLine("Done.");

But you actually only wait for the first thread to call resetEvent.Set().

Instead you'll have to provide a ManualResetEvent (or another waithandle type - for instance ManualResetEventSlim) for each thread in order to wait for them all to end. This could be done like this:

private static void RunThreadPool()
{
  string url = @"https://google.com";
  List<ManualResetEvent> waitHandles = new List<ManualResetEvent>();

  for (int i = 0; i < 4; i++)
  {
    ManualResetEvent waitHandle = new ManualResetEvent(false);
    waitHandles.Add(waitHandle);
    int workerId = i;
    ThreadPool.QueueUserWorkItem(state => Worker(workerId, waitHandle));
  }

  Console.WriteLine("Starting adding.");
  for (int i = 0; i < 400; ++i)
  {
    inputQueue.Add(url); // + "/" + i);
  }

  Console.WriteLine("Stopping adding.");
  inputQueue.CompleteAdding();

  WaitHandle.WaitAll(waitHandles.ToArray());

  waitHandles.ForEach(wh => wh.Dispose());
  //resetEvent.WaitOne();
  Console.WriteLine("Done.");
}

static void Worker(int workerId, ManualResetEvent waitHandle)
{
  Console.WriteLine("Worker {0} is starting.", workerId);

  foreach (var workItem in inputQueue.GetConsumingEnumerable())
  {
    string res = "";
    try
    {
      res = Get(workItem);
      Console.WriteLine($"{res} - {workerId}");
    }
    catch (Exception)
    {
      res = "404";
    }
  }
  waitHandle.Set();
  Console.WriteLine("Worker {0} is stopping.", workerId);
}

Besides that I think your use of ThreadPool.QueueUserWorkItem() is OK.


It is not quite clear what kind of concurrency you are exercising but if it's just about executing an action on each element in an existing collection in a multi threaded/parallel manner then, you can experiment with Parallel.ForEach(..):

private static void RunParallel()
{
  string url = @"https://google.com";

  List<string> inputQueue = new List<string>();

  for (int i = 0; i < 400; ++i)
  {
    inputQueue.Add(url); // + "/" + i);
  }

  int workerId = 0;
  ParallelOptions options = new ParallelOptions();
  options.MaxDegreeOfParallelism = 8;
  Parallel.ForEach(inputQueue, options, (uri) =>
  {
    Worker(workerId++, uri);
  });

  Console.WriteLine("Done.");
}

static void Worker(int workerId, string uri)
{
  Console.WriteLine("Worker {0} is starting.", workerId);

  string res = "";
  try
  {
    res = Get(uri);
    Console.WriteLine(res);
  }
  catch (Exception)
  {
    res = "404";
  }

  Console.WriteLine("Worker {0} is stopping.", workerId);
}

If you experiment with different values of options.MaxDegreeOfParallelism you can investigate how the Parallel react to that in the use of threads and how it may influence on performance.

\$\endgroup\$
2
  • \$\begingroup\$ I'd use semaphore instead (workers call Release, main calls WaitOne four times). +1 Anyway \$\endgroup\$
    – user52292
    Dec 2, 2018 at 16:17
  • \$\begingroup\$ @Henrik Hansen, thanks. The last method with the Parallel.ForEach(..) was the fastet (14 seconds). Btw, I also tried async\await (my last edit from 4.12.2018) but it was much slower compared to your suggestions. \$\endgroup\$
    – E235
    Dec 4, 2018 at 7:41

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.