ConcurrentQueue and Parallel.ForEach

Question

I have a ConcurrentQueue with a list of URLs that I need to get the the source of. When using the Parallel.ForEach with the ConcurrentQueue object as the input parameter, the Pop method won't work nothing (Should return a string).

I'm using Parallel with the MaxDegreeOfParallelism set to four. I really need to block the number of concurrent threads. Is using a queue with Parallelism redundant?

Thanks in advance.

// On the main class
var items = await engine.FetchPageWithNumberItems(result);
// Enqueue List of items
itemQueue.EnqueueList(items);
var crawl = Task.Run(() => { engine.CrawlItems(itemQueue); });

// On the Engine class
public void CrawlItems(ItemQueue itemQueue)
{
Parallel.ForEach(
            itemQueue,
            new ParallelOptions {MaxDegreeOfParallelism = 4},
            item =>
            {

                var worker = new Worker();
                // Pop doesn't return anything
                worker.Url = itemQueue.Pop();
                /* Some work */
             });
 }

// Item Queue
class ItemQueue : ConcurrentQueue<string>
    {
        private ConcurrentQueue<string> queue = new ConcurrentQueue<string>();

        public string Pop()
        {
            string value = String.Empty;
            if(this.queue.Count == 0)
                throw new Exception();
            this.queue.TryDequeue(out value);
            return value;
        }

        public void Push(string item)
        {
            this.queue.Enqueue(item);
        }

        public void EnqueueList(List<string> list)
        {
            list.ForEach(this.queue.Enqueue);
        }
    }

ItemQueue should not both derive from `ConcurrentQueue` and contain a `ConcurrentQueue`, pick one. — Scott Chamberlain, Jun 11 '16 at 22:47
@Zroq: Since downloading the source of a URL is an I/O-bound operation, I must state that parallelism is the wrong tool to use. Asynchronous concurrency would use far fewer resources and be just as fast. — Stephen Cleary, Jun 12 '16 at 02:17
@StephenCleary async/await with a SemaphoreSlim to force the number of active threads or there is another more efficient method? — Zroq, Jun 12 '16 at 08:44

score 6 · Answer 1 · answered Jul 10 '16 at 00:56

You don't need ConcurrentQueue<T> if all you're going to do is to first add items to it from a single thread and then iterate it in Parallel.ForEach(). A normal List<T> would be enough for that.

Also, your implementation of ItemQueue is very suspicious:

It inherits from ConcurrentQueue<string> and also contains another ConcurrentQueue<string>. That doesn't make much sense, is confusing and inefficient.
The methods on ConcurrentQueue<T> were designed very carefully to be thread-safe. Your Pop() isn't thread-safe. What could happen is that you check Count, notice it's 1, then call TryDequeue() and not get any value (i.e. value will be null), because another thread removed the item from the queue in the time between the two calls.

yonisha · Answer 2 · 2016-06-11T21:45:23.697

0

The issue is with CrawlItems method, since you shouldn't call Pop in the action provided to the ForEach method. The reason is that the action is being called on each popped item, hence the item was already popped. This is the reason that the action has an 'item' argument.

I assume that you're getting null since all of the items already popped by the other threads, by the ForEach method.

Therefore, your code should look like this:

public void CrawlItems(ItemQueue itemQueue)
{
    Parallel.ForEach(
        itemQueue,
        new ParallelOptions {MaxDegreeOfParallelism = 4},
        item =>
        {
            worker.Url = item;
            /* Some work */
         });
}

edited Jun 11 '16 at 21:45

answered Jun 11 '16 at 21:23

yonisha

2,956
2
25
32

After the itemQueue.EnqueueList(items); I've checked that the queue trully has 41 items – Zroq Jun 11 '16 at 21:26
Could you please share the implementation if ItemQueue? – yonisha Jun 11 '16 at 21:28

ConcurrentQueue and Parallel.ForEach

2 Answers2