Um método robusto de rastrear trabalhadores fracassados ​​com Threadpool

Question

Estou procurando um bom método de rastreamento (contando) que os trabalhadores falharam quando fizeram fila com um threadpool e usando waithandle.waitall () para que todos os threads concluam.

Interligar um balcão é uma boa técnica ou existe uma estratégia mais robusta?

Answer 1

Ok, aqui está uma abordagem que você pode adotar. Eu encapsulei os dados que queremos rastrear em uma classe TrackedWorkers. Existe um construtor nesta classe que permite definir quantos trabalhadores estarão funcionando. Então, os trabalhadores são lançados usando LaunchWorkers que requer um delegado que come um object e retorna a bool. o object representa a entrada para o trabalhador e o bool representa sucesso ou fracasso, dependendo de true ou false sendo o valor de retorno, respectivamente.

Então, basicamente, o que fazemos, temos uma matriz para rastrear o estado dos trabalhadores. Lançamos os trabalhadores e estabelecemos o status correspondente a esse trabalhador, dependendo do valor de retorno do trabalhador. Quando o trabalhador retornar, definimos um AutoResetEvent e WaitHandle.WaitAll para tudo AutoResetEvents a ser definido.

Observe que existe uma classe aninhada para rastrear o trabalho (o delegado) que o trabalhador deve fazer, a contribuição para esse trabalho e um ID usado para definir o status AutoResetEvent correspondente a esse tópico.

Observe com muito cuidado que, uma vez feito o trabalho, não estamos mantendo uma referência ao delegado de trabalho func nem para o input. Isso é importante para não impedir que as coisas sejam coletadas de lixo.

Existem métodos para obter o status de um determinado trabalhador, bem como todos os índices dos trabalhadores que tiveram sucesso e todos os índices dos trabalhadores que falharam.

Uma última nota: não considero este código de produção pronto. É apenas um esboço da abordagem que eu adotaria. Você precisa tomar cuidado para adicionar testes, manipulação de exceções e outros detalhes.

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;
            }
        }
    }
}

Uso de amostra:

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);

    }
}

O programa acima lançará sessenta e quatro tópicos. o iTh Thread tem a tarefa de adicionar os números i e 2 * i e imprimir o resultado no console. No entanto, adicionei uma quantidade aleatória de sono (menos de um segundo) para simular a ocupação e viro uma moeda para determinar o sucesso ou a falha do encadeamento. Aqueles que sucedem imprimem a soma com que foram encarregados e retornam true. Aqueles que falham não imprimem e retornam false.

Aqui eu usei

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;
    }
}