Persistez la variable sur le thread en C # Parallel.ForEach

Question

Je cherche à paralléliser le traitement d'une tâche qui dépend d'un objet ( State), qui n'est pas thread-safe, et dont la construction prend du temps .

Pour cette raison, je recherchais des variables locales de partition , mais je le fais mal ou je cherche autre chose. Cela représente plus ou moins ma mise en œuvre actuelle:

Parallel.ForEach<string, State>(folders, config, () => new State(), (source, loopState, index, threadState) => {
    var content = File.ReadAllText(source);        // read file
    var result = threadState.doSomething(content); // do something
    File.WriteAllText(outputFile, result);         // write output
    return threadState;
}, (threadState) => { });

Cependant, j'ai ajouté un Console.WriteLinedans mon Stateinitialiseur, et je vois que pour chaque itération de la boucle, le Stateconstructeur est appelé, ce qui entraîne un gros problème de performances. Je voudrais que l' instance de Statedans un thread soit passée à l'itération suivante sur ce même thread.

Comment puis-je y parvenir?

Answer 1

Vous avez plusieurs options. Le plus simple est de créer un seul Stateobjet, et de synchroniser l'accès à celui-ci en utilisant un lock:

var state = new State();

Parallel.ForEach(folders, config, source =>
{
    var content = File.ReadAllText(source);
    string result;
    lock (state) { result = state.DoSomething(content); }
    File.WriteAllText(outputFile, result);
});

Je suppose que ce n'est pas viable car la DoSomethingméthode prend du temps, et sa synchronisation tuera le parallélisme.

Une autre option consiste à utiliser un fichier ThreadLocal. Cette classe fournit un stockage de données local au thread, de sorte que le nombre d' Stateobjets créés sera égal au nombre de threads utilisés par le Parallel.ForEach.

var threadLocalState = new ThreadLocal<State>(() => new State());

Parallel.ForEach(folders, config, source =>
{
    var content = File.ReadAllText(source);
    var result = threadLocalState.Value.DoSomething(content);
    File.WriteAllText(outputFile, result);
});

Cela créera probablement moins d' Stateobjets que la Parallel.ForEach<TSource, TLocal>surcharge, mais toujours pas égal à la configuration MaxDegreeOfParallelism. Il Parallel.ForEachutilise les threads du ThreadPool, et il est fort possible qu'il les utilise tous lors du calcul, à condition que la liste des folders. Et vous avez peu de contrôle sur la taille du fichier ThreadPool. Ce n'est donc pas non plus une solution particulièrement séduisante.

La troisième et dernière option à laquelle je peux penser est de créer un pool d' Stateobjets, et Rent/ Returnun dans chaque boucle:

var statePool = new ObjectPool<State>(() => new State());

Parallel.ForEach(folders, config, source =>
{
    var state = statePool.Rent();
    var content = File.ReadAllText(source);
    var result = state.DoSomething(content);
    File.WriteAllText(outputFile, result);
    statePool.Return(state);
});

De cette façon, le nombre d' Stateobjets instanciés sera égal au degré maximum de parallélisme.

Le seul problème est qu'il n'y a pas de ObjectPoolclasse dans la plateforme .NET (il n'y a qu'une ArrayPoolclasse), vous devrez donc en trouver une. Voici une implémentation simple basée sur un ConcurrentBag:

public class ObjectPool<T> : IEnumerable<T> where T : new()
{
    private readonly ConcurrentBag<T> _bag = new ConcurrentBag<T>();
    private readonly Func<T> _factory;

    public ObjectPool(Func<T> factory = null) => _factory = factory;

    public T Rent()
    {
        if (_bag.TryTake(out var obj)) return obj;
        return _factory != null ? _factory() : new T();
    }

    public void Return(T obj) => _bag.Add(obj);

    public IEnumerator<T> GetEnumerator() => _bag.GetEnumerator();
    IEnumerator IEnumerable.GetEnumerator() => this.GetEnumerator();
}