A robust method of tracking failed workers with ThreadPool

Question

I'm looking for a good method of tracking (counting) which workers have failed when queued with a Threadpool and using WaitHandle.WaitAll() for all threads to finish.

Is Interlocking a counter a good technique or is there a more robust strategy?

Answer 1

Okay, here's an approach that you could take. I've encapsulated the data that we want to track into a class TrackedWorkers. There is a constructor on this class that enables you to set how many workers will be working. Then, the workers are launched using LaunchWorkers which requires a delegate that eats an object and returns a bool. The object represents the input to the worker and the bool represents success or failure depending on true or false being the return value, respectively.

So basically what we do we have an array to track worker state. We launch the workers and set the status corresponding to that worker depending on the return value from the worker. When the worker returns, we set an AutoResetEvent and WaitHandle.WaitAll for all the AutoResetEvents to be set.

Note that there is an nested class to track the work (the delegate) the worker is supposed to do, the input to that work, and an ID used to set the status AutoResetEvent corresponding to that thread.

Note very carefully that once the work is done we are not holding a reference to the work delegate func nor to the input. This is important so that we don't accidentally prevent stuff from being garbage collected.

There are methods for getting the status of a particular worker, as well as all the indexes of the workers that succeeded and all the indexes of the workers that failed.

One last note: I do not consider this code production ready. It is merely a sketch of the approach that I would take. You need to take care to add testing, exception handling and other such details.

class TrackedWorkers {
    class WorkerState {
        public object Input { get; private set; }
        public int ID { get; private set; }
        public Func<object, bool> Func { get; private set; }
        public WorkerState(Func<object, bool> func, object input, int id) {
            Func = func;
            Input = input;
            ID = id;
        }
    }

    AutoResetEvent[] events;
    bool[] statuses;
    bool _workComplete;
    int _number;

    public TrackedWorkers(int number) {
        if (number <= 0 || number > 64) {
            throw new ArgumentOutOfRangeException(
                "number",
                "number must be positive and at most 64"
            );
        }
        this._number = number;
        events = new AutoResetEvent[number];
        statuses = new bool[number];
        _workComplete = false;
    }

    void Initialize() {
        _workComplete = false;
        for (int i = 0; i < _number; i++) {
            events[i] = new AutoResetEvent(false);
            statuses[i] = true;
        }
    }

    void DoWork(object state) {
        WorkerState ws = (WorkerState)state;
        statuses[ws.ID] = ws.Func(ws.Input);
        events[ws.ID].Set();
    }

    public void LaunchWorkers(Func<object, bool> func, object[] inputs) {
        Initialize();
        for (int i = 0; i < _number; i++) {
            WorkerState ws = new WorkerState(func, inputs[i], i);
            ThreadPool.QueueUserWorkItem(this.DoWork, ws);
        }
        WaitHandle.WaitAll(events);
        _workComplete = true;
    }

    void ThrowIfWorkIsNotDone() {
        if (!_workComplete) {
            throw new InvalidOperationException("work not complete");
        }
    }

    public bool GetWorkerStatus(int i) {
        ThrowIfWorkIsNotDone();
        return statuses[i];
    }

    public IEnumerable<int> SuccessfulWorkers {
        get {
            return WorkersWhere(b => b);
        }
    }

    public IEnumerable<int> FailedWorkers {
        get {
            return WorkersWhere(b => !b);
        }
    }

    IEnumerable<int> WorkersWhere(Predicate<bool> predicate) {
        ThrowIfWorkIsNotDone();
        for (int i = 0; i < _number; i++) {
            if (predicate(statuses[i])) {
                yield return i;
            }
        }
    }
}

Sample usage:

class Program {
    static Random rg = new Random();
    static object lockObject = new object();
    static void Main(string[] args) {
        int count = 64;
        Pair[] pairs = new Pair[count];
        for(int i = 0; i < count; i++) {
            pairs[i] = new Pair(i, 2 * i);
        }
        TrackedWorkers workers = new TrackedWorkers(count);
        workers.LaunchWorkers(SleepAndAdd, pairs.Cast<object>().ToArray());
        Console.WriteLine(
            "Number successful: {0}",
            workers.SuccessfulWorkers.Count()
        );
        Console.WriteLine(
            "Number failed: {0}",
            workers.FailedWorkers.Count()
        );
    }
    static bool SleepAndAdd(object o) {
        Pair pair = (Pair)o;
        int timeout;
        double d;
        lock (lockObject) {
            timeout = rg.Next(1000);
            d = rg.NextDouble();
        }
        Thread.Sleep(timeout);
        bool success = d < 0.5;
        if (success) {
            Console.WriteLine(pair.First + pair.Second);
        }
        return (success);

    }
}

The above program is going to launch sixty-four threads. The ith thread has the task of adding the numbers i and 2 * i and printing the result to the console. However, I have added a random amount of sleep (less than one second) to simulate busyness and I flip a coin to determine success or failure of the thread. Those that succeed print the sum they were tasked with and return true. Those that fail print nothing and return false.

Here I have used

struct Pair {
    public int First { get; private set; }
    public int Second { get; private set; }
    public Pair(int first, int second) : this() {
        this.First = first;
        this.Second = second;
    }
}