Persistir la variable en el hilo en C # Parallel.ForEach

Question

Estoy buscando paralelizar el procesamiento de una tarea que depende de un objeto ( State), que no es seguro para subprocesos y cuya construcción es costosa en tiempo .

Por esta razón, estaba buscando variables locales de partición , pero lo estaba haciendo mal o estaba buscando algo más. Esto más o menos representa mi implementación actual:

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) => { });

Sin embargo, agregué un Console.WriteLineen mi Stateinicializador, y veo que para cada iteración del ciclo, Statese llama al constructor, lo que resulta en un gran impacto en el rendimiento. Me gustaría que la instancia de Stateen un hilo se pase a la iteración posterior en ese mismo hilo.

¿Cómo puedo lograrlo?

Answer 1

Tienes unas cuantas opciones. La más sencilla es crear un único Stateobjeto y sincronizar el acceso a él mediante 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);
});

Supongo que esto no es viable porque el DoSomethingmétodo lleva mucho tiempo y sincronizarlo acabará con el paralelismo.

Otra opción es utilizar un ThreadLocal. Esta clase proporciona un almacenamiento de datos local de subprocesos, por lo que el número de Stateobjetos creados será igual al número de subprocesos empleados por 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);
});

Esto probablemente creará menos Stateobjetos que la Parallel.ForEach<TSource, TLocal>sobrecarga, pero aún así no será igual a los configurados MaxDegreeOfParallelism. El Parallel.ForEachusa subprocesos de ThreadPool, y es muy posible que los use todos durante el cálculo, siempre que la lista de folderssea ​​lo suficientemente larga. Y tienes poco control sobre el tamaño del ThreadPool. Así que esta tampoco es una solución particularmente atractiva.

La tercera y última opción en la que puedo pensar es crear un grupo de Stateobjetos y Rent/ Returnuno en cada bucle:

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 esta forma, el número de Stateobjetos instanciados será igual al grado máximo de paralelismo.

El único problema es que no hay una ObjectPoolclase en la plataforma .NET (solo hay una ArrayPoolclase), por lo que tendrá que encontrar una. Aquí hay una implementación simple basada en 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();
}