Persiste variabile sul thread in C # Parallel.ForEach

Question

Sto cercando di parallelizzare l'elaborazione di un'attività che dipende da un object ( State), che non è thread-safe e la cui costruzione è costosa in termini di tempo .

Per questo motivo, stavo esaminando le variabili locali della partizione , ma o lo sto facendo male o sto cercando qualcos'altro. Questo più o meno rappresenta la mia attuale implementazione:

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

Tuttavia, ho aggiunto un Console.WriteLinenel mio Stateinizializzatore e vedo che per ogni iterazione del ciclo Stateviene chiamato il costruttore, con un conseguente notevole calo delle prestazioni. Vorrei che l' istanza di Statein un thread venisse passata alla successiva iterazione sullo stesso thread.

Come posso ottenerlo?

Answer 1

Hai un paio di opzioni. Il più semplice è creare un singolo Stateoggetto e sincronizzare l'accesso ad esso utilizzando 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);
});

Presumo che questo non sia fattibile perché il DoSomethingmetodo richiede tempo e la sincronizzazione eliminerà il parallelismo.

Un'altra opzione è usare un file ThreadLocal. Questa classe fornisce un'archiviazione locale dei dati, quindi il numero di Stateoggetti creati sarà uguale al numero di thread utilizzati da 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);
});

Questo probabilmente creerà meno Stateoggetti del Parallel.ForEach<TSource, TLocal>sovraccarico, ma comunque non uguale a quello configurato MaxDegreeOfParallelism. Gli Parallel.ForEachusi discussioni da ThreadPool, ed è possibile che si utilizzerà tutti loro durante il calcolo, a condizione che l'elenco dei foldersè sufficientemente lungo. E hai poco controllo sulle dimensioni del file ThreadPool. Quindi anche questa non è una soluzione particolarmente allettante.

La terza e ultima opzione a cui posso pensare è creare un pool di Stateoggetti e Rent/ Returnuno in ogni ciclo:

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

In questo modo il numero degli Stateoggetti istanziati sarà uguale al massimo grado di parallelismo.

L'unico problema è che non esiste una ObjectPoolclasse nella piattaforma .NET (c'è solo una ArrayPoolclasse), quindi dovrai trovarne una. Ecco una semplice implementazione basata su 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();
}