1

I've created a simple method which is supposed to take a list of, say, 100 items, and process them asynchronously (up to MAX_CONCURRENT elements at a time), and return only once all elements have been processed:

/// <summary>Generic method to perform an action or set of actions
///          in parallel on each item in a collection of items, returning
///          only when all actions have been completed.</summary>
/// <typeparam name="T">The element type</typeparam>
/// <param name="elements">A collection of elements, each of which to
///                        perform the action on.</param>
/// <param name="action">The action to perform on each element. The
///                      action should of course be thread safe.</param>
/// <param name="MAX_CONCURRENT">The maximum number of concurrent actions.</param>

public static void PerformActionsInParallel<T>(IEnumerable<T> elements, Action<T> action)
{
    // Semaphore limiting the number of parallel requests
    Semaphore limit = new Semaphore(MAX_CONCURRENT, MAX_CONCURRENT);
    // Count of the number of remaining threads to be completed
    int remaining = 0;
    // Signal to notify the main thread when a worker is done
    AutoResetEvent onComplete = new AutoResetEvent(false);

    foreach (T element in elements)
    {
        Interlocked.Increment(ref remaining);
        limit.WaitOne();
        new Thread(() =>
        {
            try
            {
                action(element);
            }
            catch (Exception ex)
            {
                Console.WriteLine("Error performing concurrent action: " + ex);
            }
            finally
            {
                Interlocked.Decrement(ref remaining);
                limit.Release();
                onComplete.Set();
            }
        }).Start();
    }
    // Wait for all requests to complete
    while (remaining > 0)
        onComplete.WaitOne(10); // Slightly better than Thread.Sleep(10)
}

    /* We include a timeout on the `WaitOne()` before checking `remaining` again
        * to protect against the rare case where the last outstanding thread
        * decrements 'remaining' and then signals completion *between* the main thread
        * checking 'remaining' and waiting for the next completion signal, which would
        * otherwise result in the main thread missing the last signal and locking forever. */

Most of the time, this code behaves exactly as expected, but in rare occasions, I discover that the method returns (i.e. breaks out of that final while loop) before every element in the list is finished processing. It always seems to happen when there's only a few elements left - e.g. I'll have processed 97 elements, and then the method returns, and then elements 98-100 complete.

Is there anything I'm doing wrong that's maybe resulting in the remaining count hitting 0 before all elements have actually been processed?

10
  • 2
    It is formally wrong, the int is not used in a thread-safe way. Also quite possible that the while-loop never completes since the int is not declared volatile. Use the CountDownEvent class instead, get rid of onComplete. Nov 16, 2015 at 18:49
  • Note that while I agree with Hans' observations and suggestion, you could fix the volatile issue by using Volatile.Read() to access the value at the end (that's where the lack of thread safety is a problem). I still wouldn't recommend doing it that way; polling a variable like that is wasteful and unnecessary. Use an appropriate synchronization mechanism that works. Frankly, it's not clear to me why you poll the variable anyway; just have the last thread (i.e. the one that decrements remaining to 0) set the event, and wait on the event. Or use CountdownEvent, which is made for this Nov 16, 2015 at 19:01
  • Also note that you seem to be reinventing Parallel.ForEach(). If this is an academic exercise, fine...but if not, consider using the built-in features instead of writing them again. :) Nov 16, 2015 at 19:03
  • @HansPassant What isn't thread safe? I use Interlocked.Decrement and I thought reading a 32 bit int was atomic?
    – Alain
    Nov 16, 2015 at 20:14
  • @PeterDuniho I tried that but I can't have just the thread that decrements remaining to 0 because if the for loop hasn't executed its second loop yet, remaining will still be set to 1 when it is decremented back to 0 and complete will be set too early,
    – Alain
    Nov 16, 2015 at 20:16

1 Answer 1

1

Here's a modified solution that makes use of a CountdownEvent signal to avoid the use of the remaining integer and avoid the busy waiting involved in polling it with the unreliable AutoResetEvent onComplete:

public static void PerformActionsInParallel<T>(IEnumerable<T> elements, Action<T> action)
{
    int threads = MaxConcurrent ?? DefaultMaxConcurrentRequests;
    // Ensure elements is only enumerated once.
    elements = elements as T[] ?? elements.ToArray();
    // Semaphore limiting the number of parallel requests
    Semaphore limit = new Semaphore(MAX_CONCURRENT, MAX_CONCURRENT);
    // Count of the number of remaining threads to be completed
    CountdownEvent remaining = new CountdownEvent(elements.Count());

    foreach (T element in elements)
    {
        limit.WaitOne();
        new Thread(() =>
        {
            try
            {
                action(element);
            }
            catch (Exception ex)
            {
                Console.WriteLine("Error performing concurrent action: " + ex);
            }
            finally
            {
                remaining.Signal();
                limit.Release();
            }
        }).Start();
    }
    // Wait for all requests to complete
    remaining.Wait();
}

Testing underway to see if it solves the issue.

Your Answer

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

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