Running multiple threads

[R3DDOT]

So I'm working on a link farmer, basicly it finds links that hold special items(hidden items that can be bought) by making an HttpWebRequest and reading the response stream.

I thought working with multiple threads might speed this up, since in theory having multiple threads searching is better than just one.
But apparently, it seems to slowdown the process.

Here's the code:

int t_count = GetSearchType(); //don't worry about this.

                for (; t_count >= 0; t_count--)
                    new System.Threading.Thread(
                        () =>
                        {

                            for (Search_CurrentID += (int)IncrementNum.Value; Search_CurrentID < Search_MaxID && Running; Search_CurrentID += (int)IncrementNum.Value)
                            {
                                string link = link_mask + Search_CurrentID.ToString();
                                HttpWebRequest req = (HttpWebRequest)WebRequest.Create(link);
                                req.Timeout = 2000;
                                req.Method = WebRequestMethods.Http.Post;
                                req.ContentType = "application/x-www-form-urlencoded";
                                req.UserAgent = "Mozilla/5.0 (Windows NT 6.1; WOW64_R3D) AppleWebKit/537.36 (KHTML, like Gecko) Safari/537.36";
                                req.ContentLength = 0;

                                using (StreamReader wStream = new StreamReader(req.GetResponse().GetResponseStream()))
                                {
                                    for (string line = wStream.ReadLine(); !wStream.EndOfStream; line = wStream.ReadLine())
                                    {
                                        if (line.Contains("title story-title"))
                                        {
                                            Links_Searched++;
                                            string title = line.Replace("<h1 class=\"title story-title\">", "").Replace("</h1>", "").Remove(0, " 					".Length - 1);
                                            loglink(link, title);
                                        }

                                    }
                                }
                            }

                            Running = false;
                            if (Search_CurrentID >= Search_MaxID && PopupOnCompleteCB.Checked)
                                notifyIcon.ShowBalloonTip(1000, "Aeria Link Farmer", "Search has been completed!", ToolTipIcon.Info);
                        }
                        ).Start();
            }

If you can figure out what's wrong with this, I'd be really appreciated.

[SteamThief]

What is the problem you are having, don't you get the other threads to run or does it just not run as fast as expected?

[R3DDOT]

What is the problem you are having, don't you get the other threads to run or does it just not run as fast as expected?

From what I've seen, it seems to run the all threads. But they run slower, and I'm not sure if that's caused because of the http request(10 requests in 1 sec ==> lag) or if I did something wrong.

[SteamThief]

What CPU do you have?
How many cores / Threads does it have?

I'll take a closer look to the at the code when i'm home.

[R3DDOT]

What CPU do you have?
How many cores / Threads does it have?

I'll take a closer look to the at the code when i'm home.

I got an i7(2.8 GHz). The problem isn't with threads apparently, it's with the requests. If I send too many at once, it lags. I don't think there's a way to get around that, unless by decreasing their amount(like to 3~5 threads).

[SteamThief]

I got an i7(2.8 GHz). The problem isn't with threads apparently, it's with the requests. If I send too many at once, it lags. I don't think there's a way to get around that, unless by decreasing their amount(like to 3~5 threads).

Like you said, try to decrease the requests, but I find it weird that it lags if you send to many at once. 10 is not that many, but lower and than go higher until it fails. You can use STopwatch to see have many milisecounds each time and than you'll see easy if it goes faster / slower.

[abuckau907]

@OP Normally multi-threading is used to split the work between multiple threads.
It looks like you're just making 10 (or..t_count) threads all do the exact same work, instead of splitting up the work. (?)

Since you're inner for() loop is going from some Start_ID to Max_ID ...you could make each thread responsible for a small range of IDs.

edit: I don't know, but I'm guessing one thread per cpu core is most efficient ...the OS will 'manage' multiple threads for you..and make sure each gets a little run-time on the cpu, but there is over-head when the OS switches between threads.
You could try more threads per core...2, 4, maybe even 8...but my guess is 1 (maybe 2?) threads per core is most efficient. Again, just a guess.

[R3DDOT]

@OP Normally multi-threading is used to split the work between multiple threads.
It looks like you're just making 10 (or..t_count) threads all do the exact same work, instead of splitting up the work. (?)

Since you're inner for() loop is going from some Start_ID to Max_ID ...you could make each thread responsible for a small range of IDs.

edit: I don't know, but I'm guessing one thread per cpu core is most efficient ...the OS will 'manage' multiple threads for you..and make sure each gets a little run-time on the cpu, but there is over-head when the OS switches between threads.
You could try more threads per core...2, 4, maybe even 8...but my guess is 1 (maybe 2?) threads per core is most efficient. Again, just a guess.

I understand what you are saying, but by incrementing its value at the end of each loop, they should be using different values. I might be wrong though.

[abuckau907]

I'm not sure sure I understand you, because we want each thread to be different...to have it's own subsection of ID's to scan.
If all threads are doing the exact same thing, there is no point; you're just doing the same work multiple times.. (?)

I mean to create 2 more variables in the outmost for() loop - each thread will get a START_ID and STOP_ID. Increment these 2 variable in the outer for loop, not inside the newly created threads.
^^Or however you want to implement "giving each thread a chunk of the IDs." But by this point, you'll have to decided maximum number of threads you'll use.

int start_id = 0
int stop_id = ****
for ( ; t_count >=0; t_count--)

{
 spawnThread(start_id,stop_id); 
 start_id = end_id + 1 '' right, lol?
 end_id   += (total_IDS / number_cores) + ****; //something similar to this
}

^^obviously a little primitive. Point is, split up the range of ID's you make each thread scan. If the dataset isn't a perfect multiple of your thread_count, some thread will have to do a couple extra/less ID's, which is ok.

In your code:

  ...
  for (Search_CurrentID += (int)IncrementNum.Value; Search_CurrentID < Search_MaxID && Running; Search_CurrentID += (int)IncrementNum.Value)
  ...

^^is the part that could be changed. Don't make each thread scan the entire range of data.


I believe each thread gets a private copy 'Search_CurrentID' ...if not, all threads are trying to update the same variable, which is bad. (can you imagine if all 10 threads executed their For() loop's increment condition (current_id would go up 10x more than it should). I'd add more debugging-output to check. I don't see anything about mutexes or locks (etc?) so I'm assuming each thread gets private copies of any variables used, but I don't know (we don't see how you declare the variables..). I'm guessing it gives each thread a local copy by default, and only share the memory of some variable if you specifically request it. Just guessing though : ( Looking forward to your reply. I might even test it myself soon (you know..have more test sets..I'm on core i3 lol).

[Jason]

This is one of the "traps" of multi-threading. Just because you run code in multiple threads doesn't automatically mean you'll experience any sort of speed up. It's important to know that multi-threading your application is NOT some magical solution to increase performance; in a lot of cases it will actually make your code slower if done incorrectly.

I'd recommend reading up on multi-threading and parallelism. It's a pretty complex topic at first but it's essential you know what you're doing when you start otherwise you're likely to end up getting shafted later on.

In the mean time, this kind of problem suits a task-based approach. If you're using .NET 4.0+ then I'd highly recommend using the new "Task" class in the System.Threading.Tasks namespace. It abstracts away a lot of the manually thread creation/synchronisation and makes it a lot easier to do concurrent work.

Next, add the following class to your project. It's mostly a direct copy from an article from MSDN with some minor adjustments on my part:

using System;
using System.Threading;
using System.Threading.Tasks;
using System.Collections.Generic;

namespace JLibrary.Net {
    public class SparseScheduler : TaskScheduler {
        [ThreadStatic]
        private static bool _currentThreadIsProcessingItems;

        private readonly LinkedList<Task> _waitingTasks = new LinkedList<Task>();
        private readonly int _maxConcurrent;
        private int _runningDelegates;

        public SparseScheduler(int maximumConcurrent) {
            if (maximumConcurrent < 1) {
                throw new ArgumentOutOfRangeException("maximumConcurrent");
            }

            _maxConcurrent = maximumConcurrent;
            _runningDelegates = 0;
        }

        protected override void QueueTask(Task task) {
            lock (_waitingTasks) {
                _waitingTasks.AddLast(task);
                if (_runningDelegates < _maxConcurrent) {
                    ++_runningDelegates;
                    NotifyThreadPool();
                }
            }
        }

        private void NotifyThreadPool() {
            ThreadPool.UnsafeQueueUserWorkItem(_ => {
                _currentThreadIsProcessingItems = true;
                try {
                    while (true) {
                        Task item;
                        lock (_waitingTasks) {
                            if (_waitingTasks.Count == 0) {
                                --_runningDelegates;
                                break;
                            }

                            item = _waitingTasks.First.Value;
                            _waitingTasks.RemoveFirst();
                        }

                        base.TryExecuteTask(item);
                    }
                }
                finally {
                    _currentThreadIsProcessingItems = false;
                }
            }, null);
        }

        protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued) {
            if (!_currentThreadIsProcessingItems) {
                return false;
            }

            return taskWasPreviouslyQueued
                ? TryDequeue(task) && base.TryExecuteTask(task)
                : base.TryExecuteTask(task);
        }

        protected override bool TryDequeue(Task task) {
            lock (_waitingTasks) {
                return _waitingTasks.Remove(task);
            }
        }

        protected override IEnumerable<Task> GetScheduledTasks() {
            bool lockTaken = false;
            try {
                Monitor.TryEnter(_waitingTasks, ref lockTaken);
                if (lockTaken) return _waitingTasks;
                else throw new NotSupportedException();
            }
            finally {
                if (lockTaken) {
                    Monitor.Exit(_waitingTasks);
                }
            }
        }
    }
}

This is what's known as a TaskScheduler. Essentially you feed it small bodies of work (tasks) and it will handle all of the scheduling/threading for you. It's important to note that there's no guarantee on which thread your task will be run on, so care must be taken when writing your tasks.

This particular scheduler is known as a SparseScheduler. I originally had need for it for precisely the situation you're in now: I wanted to run multiple web requests in parallel, but didn't want to spam all of them out at once as this bloats the network unnecessarily and often causes huge amounts of lag. The SparseScheduler only allows a certain amount of jobs to be running at one time. This means I can generate a task for all of my web requests, send them to the scheduler and know that only X will be running at any one time.

Next, a really simple wrapper class around the SparseScheduler to make it specifically for WebRequests (add this in its own class file)

using System;
using System.Net;
using System.Threading;
using System.Threading.Tasks;
using JLibrary.Threading;

namespace JLibrary.Net {
    // Weak wrapper around a Task
    public class WebRequestToken {
        private readonly Task _task;
        internal Task Task { get { return _task; } }

        internal WebRequestToken(Task task) {
            _task = task;
        }

        private WebRequestToken() {
            // Hide the empty constructor
        }
    };

    public class WebRequestFactory {
        private readonly TaskFactory _factory;
        private readonly SparseScheduler _scheduler;
        private readonly CancellationTokenSource _cts;

        public WebRequestFactory(int maxConcurrentRequests) {
            _scheduler = new SparseScheduler(maxConcurrentRequests);
            _factory = new TaskFactory(_scheduler);
            _cts = new CancellationTokenSource();
        }

        public WebRequestToken SendRequest(WebRequest request, Action<Exception,WebResponse> handler) {
            if (request == null)
                throw new ArgumentNullException("request");
            if (handler == null)
                throw new ArgumentNullException("handler");

            var task = _factory.StartNew(() => {
                try {
                    using (var response = request.GetResponse())
                        handler.Invoke(null, response);
                }
                catch (Exception e) {
                    handler.Invoke(e, null);
                }
            }, _cts.Token);

            return new WebRequestToken(task);
        }

        public void WaitAll(params WebRequestToken[] tokens) {
            Task.WaitAll(Array.ConvertAll(tokens, t => t.Task), _cts.Token);
        }

        public void WaitAll(WebRequestToken[] tokens, int millisecondsToWait) {
            Task.WaitAll(Array.ConvertAll(tokens, t => t.Task), millisecondsToWait, _cts.Token);
        }
    }
}

Finally, onto your actual code. Add the reference to the correct namespace to your code:

using JLibrary.Net;

And now your code becomes very simple. First define a function to process a response:

private void HandleResponse(Exception e, WebResponse response) {
    if (e != null) {
        // do something with the exception
        throw e;
    }

    // Store a local "searched" counter to avoid locking the mutex every time a link is found
    int searched = 0;
    using(var sreader = new StreamReader(response.GetResponseStream())) {
        for(string line = sreader.ReadLine(); line != null; line = sreader.ReadLine()) {
            if (line.Contains("title story-title")) {
                ++searched;
                var title = line.Replace("<h1 class=\"title story-title\">", "").Replace("</h1>", "").Trim();
                loglink(response.ResponseUri.ToString(), title);
            }
        }
    }

    lock(_mutex) {
        Links_Searched += searched;
    }
}

NOTE: The use of _mutex here. This is just an arbitrary variable to allow for a lock to be entered when incrementing the "Links_Searched" count. Otherwise you'll likely get race conditions.

private readonly object _mutex = new object();

Will suffice.

Onto your actual code. I've cleaned it up somewhat but as you only included a small snippet I wasn't actually able to make it as nice as I would have liked.

int maxConcurrent = 8; // Maximum 8 web requests running at any time.
int increment = (int)IncrementNum.Value;

var factory = new WebRequestFactory(maxConcurrent);
var tokens = new List<WebRequestToken>();

for(Search_CurrentID+=increment; Search_CurrentID < Search_MaxID; Search_CurrentID+=increment) {
    var link = string.Format("{0}{1}", link_mask, Search_CurrentID);
    var req = (HttpWebRequest)WebRequest.Create(link);
    req.Timeout = 2000;
    req.Method = WebRequestMethods.Http.Post;
    req.ContentType = "application/x-www-form-urlencoded";
    req.UserAgent = "Mozilla/5.0 (Windows NT 6.1; WOW64_R3D) AppleWebKit/537.36 (KHTML, like Gecko) Safari/537.36";
    req.ContentLength = 0;

    tokens.Add(factory.SendRequest(req, HandleResponse));
}

// Now wait for the requests to complete
factory.WaitAll(tokens.ToArray());

// Done.

I didn't explain much here so if you have any questions feel free to ask

[R3DDOT]

This is one of the "traps" of multi-threading. Just because you run code in multiple threads doesn't automatically mean you'll experience any sort of speed up. It's important to know that multi-threading your application is NOT some magical solution to increase performance; in a lot of cases it will actually make your code slower if done incorrectly.

I'd recommend reading up on multi-threading and parallelism. It's a pretty complex topic at first but it's essential you know what you're doing when you start otherwise you're likely to end up getting shafted later on.

In the mean time, this kind of problem suits a task-based approach. If you're using .NET 4.0+ then I'd highly recommend using the new "Task" class in the System.Threading.Tasks namespace. It abstracts away a lot of the manually thread creation/synchronisation and makes it a lot easier to do concurrent work.

Next, add the following class to your project. It's mostly a direct copy from an article from MSDN with some minor adjustments on my part:

using System;
using System.Threading;
using System.Threading.Tasks;
using System.Collections.Generic;

namespace JLibrary.Net {
    public class SparseScheduler : TaskScheduler {
        [ThreadStatic]
        private static bool _currentThreadIsProcessingItems;

        private readonly LinkedList<Task> _waitingTasks = new LinkedList<Task>();
        private readonly int _maxConcurrent;
        private int _runningDelegates;

        public SparseScheduler(int maximumConcurrent) {
            if (maximumConcurrent < 1) {
                throw new ArgumentOutOfRangeException("maximumConcurrent");
            }

            _maxConcurrent = maximumConcurrent;
            _runningDelegates = 0;
        }

        protected override void QueueTask(Task task) {
            lock (_waitingTasks) {
                _waitingTasks.AddLast(task);
                if (_runningDelegates < _maxConcurrent) {
                    ++_runningDelegates;
                    NotifyThreadPool();
                }
            }
        }

        private void NotifyThreadPool() {
            ThreadPool.UnsafeQueueUserWorkItem(_ => {
                _currentThreadIsProcessingItems = true;
                try {
                    while (true) {
                        Task item;
                        lock (_waitingTasks) {
                            if (_waitingTasks.Count == 0) {
                                --_runningDelegates;
                                break;
                            }

                            item = _waitingTasks.First.Value;
                            _waitingTasks.RemoveFirst();
                        }

                        base.TryExecuteTask(item);
                    }
                }
                finally {
                    _currentThreadIsProcessingItems = false;
                }
            }, null);
        }

        protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued) {
            if (!_currentThreadIsProcessingItems) {
                return false;
            }

            return taskWasPreviouslyQueued
                ? TryDequeue(task) && base.TryExecuteTask(task)
                : base.TryExecuteTask(task);
        }

        protected override bool TryDequeue(Task task) {
            lock (_waitingTasks) {
                return _waitingTasks.Remove(task);
            }
        }

        protected override IEnumerable<Task> GetScheduledTasks() {
            bool lockTaken = false;
            try {
                Monitor.TryEnter(_waitingTasks, ref lockTaken);
                if (lockTaken) return _waitingTasks;
                else throw new NotSupportedException();
            }
            finally {
                if (lockTaken) {
                    Monitor.Exit(_waitingTasks);
                }
            }
        }
    }
}

This is what's known as a TaskScheduler. Essentially you feed it small bodies of work (tasks) and it will handle all of the scheduling/threading for you. It's important to note that there's no guarantee on which thread your task will be run on, so care must be taken when writing your tasks.

This particular scheduler is known as a SparseScheduler. I originally had need for it for precisely the situation you're in now: I wanted to run multiple web requests in parallel, but didn't want to spam all of them out at once as this bloats the network unnecessarily and often causes huge amounts of lag. The SparseScheduler only allows a certain amount of jobs to be running at one time. This means I can generate a task for all of my web requests, send them to the scheduler and know that only X will be running at any one time.

Next, a really simple wrapper class around the SparseScheduler to make it specifically for WebRequests (add this in its own class file)

using System;
using System.Net;
using System.Threading;
using System.Threading.Tasks;
using JLibrary.Threading;

namespace JLibrary.Net {
    // Weak wrapper around a Task
    public class WebRequestToken {
        private readonly Task _task;
        internal Task Task { get { return _task; } }

        internal WebRequestToken(Task task) {
            _task = task;
        }

        private WebRequestToken() {
            // Hide the empty constructor
        }
    };

    public class WebRequestFactory {
        private readonly TaskFactory _factory;
        private readonly SparseScheduler _scheduler;
        private readonly CancellationTokenSource _cts;

        public WebRequestFactory(int maxConcurrentRequests) {
            _scheduler = new SparseScheduler(maxConcurrentRequests);
            _factory = new TaskFactory(_scheduler);
            _cts = new CancellationTokenSource();
        }

        public WebRequestToken SendRequest(WebRequest request, Action<Exception,WebResponse> handler) {
            if (request == null)
                throw new ArgumentNullException("request");
            if (handler == null)
                throw new ArgumentNullException("handler");

            var task = _factory.StartNew(() => {
                try {
                    using (var response = request.GetResponse())
                        handler.Invoke(null, response);
                }
                catch (Exception e) {
                    handler.Invoke(e, null);
                }
            }, _cts.Token);

            return new WebRequestToken(task);
        }

        public void WaitAll(params WebRequestToken[] tokens) {
            Task.WaitAll(Array.ConvertAll(tokens, t => t.Task), _cts.Token);
        }

        public void WaitAll(WebRequestToken[] tokens, int millisecondsToWait) {
            Task.WaitAll(Array.ConvertAll(tokens, t => t.Task), millisecondsToWait, _cts.Token);
        }
    }
}

Finally, onto your actual code. Add the reference to the correct namespace to your code:

using JLibrary.Net;

And now your code becomes very simple. First define a function to process a response:

private void HandleResponse(Exception e, WebResponse response) {
    if (e != null) {
        // do something with the exception
        throw e;
    }

    // Store a local "searched" counter to avoid locking the mutex every time a link is found
    int searched = 0;
    using(var sreader = new StreamReader(response.GetResponseStream())) {
        for(string line = sreader.ReadLine(); line != null; line = sreader.ReadLine()) {
            if (line.Contains("title story-title")) {
                ++searched;
                var title = line.Replace("<h1 class=\"title story-title\">", "").Replace("</h1>", "").Trim();
                loglink(response.ResponseUri.ToString(), title);
            }
        }
    }

    lock(_mutex) {
        Links_Searched += searched;
    }
}

NOTE: The use of _mutex here. This is just an arbitrary variable to allow for a lock to be entered when incrementing the "Links_Searched" count. Otherwise you'll likely get race conditions.

private readonly object _mutex = new object();

Will suffice.

Onto your actual code. I've cleaned it up somewhat but as you only included a small snippet I wasn't actually able to make it as nice as I would have liked.

int maxConcurrent = 8; // Maximum 8 web requests running at any time.
int increment = (int)IncrementNum.Value;

var factory = new WebRequestFactory(maxConcurrent);
var tokens = new List<WebRequestToken>();

for(Search_CurrentID+=increment; Search_CurrentID < Search_MaxID; Search_CurrentID+=increment) {
    var link = string.Format("{0}{1}", link_mask, Search_CurrentID);
    var req = (HttpWebRequest)WebRequest.Create(link);
    req.Timeout = 2000;
    req.Method = WebRequestMethods.Http.Post;
    req.ContentType = "application/x-www-form-urlencoded";
    req.UserAgent = "Mozilla/5.0 (Windows NT 6.1; WOW64_R3D) AppleWebKit/537.36 (KHTML, like Gecko) Safari/537.36";
    req.ContentLength = 0;

    tokens.Add(factory.SendRequest(req, HandleResponse));
}

// Now wait for the requests to complete
factory.WaitAll(tokens.ToArray());

// Done.

I didn't explain much here so if you have any questions feel free to ask

Actually I've never heard of that class, TaskScheduler.
I'll definitely check it out later, I'm kinda busy atm. Thanks for the explanation, really helpful.

[[MPGH]Mayion]

Actually I've never heard of that class, TaskScheduler.
I'll definitely check it out later, I'm kinda busy atm. Thanks for the explanation, really helpful.

Do I close the thread and mark it as solved?

[R3DDOT]

Do I close the thread and mark it as solved?

Yeah, you can mark as solved. Thanks.

[[MPGH]Mayion]

Glad you found a solution.
Closed.