Wie schreibe ich eine Thread-bewusste Erweiterungsfunktion für PLINQ?

Question

Kann jemand eine Erweiterungsfunktion schreiben, die eine ParallelQuery in PLINQ zurückgibt?

Genauer gesagt habe ich das folgende Problem: Ich möchte eine Transformation innerhalb einer PLINQ-Abfrage durchführen, die eine Engine benötigt, deren Erstellung teuer ist und auf die nicht gleichzeitig zugegriffen werden kann.

konnte ich folgendes tun:

var result = source.AsParallel().Select ((i) => { var e = new Engine(); return e.Process(i); }) 

Hier wird der Motor einmal pro Element erstellt, das zu teuer ist.

Ich möchte die Engine einmal erstellt pro Thread.

Mit Aggregate, kann ich nahe kommen, was ich mit etwas wollen wie

// helper class: engine to use plus list of results obtained in thread so far 
class EngineAndResults { 
    public Engine engine = null; 
    public IEnumerable<ResultType> results; 
} 

var result = source.AsParallel().Aggregate (

    // done once per block of items (=thread), 
    // returning an empty list, but a new engine 
    () => new EngineAndList() { 
     engine = new Engine(), 
     results = Enumerable.Empty<ResultType>() 
    }, 

    // we process a new item and put it to the thread-local list, 
    // preserving the engine for further use 
    (engineAndResults, item) => new EngineAndResults() { 
     engine = engineAndResults.engine, 
     results = Enumerable.Concat (
      engineAndResults.results, 
      new ResultType [] { engineAndResults.engine.Process (item) } 
     ) 
    }, 

    // tell linq how to aggregate across threads 
    (engineAndResults1, engineAndResults2) => new EngineAndResults() { 
     engine = engineAndResults1.engine, 
     results = Enumerable.Concat (engineAndResults1.results, engineAndResults2.results) 
    }, 

    // after all aggregations, how do we come to the result? 
    engineAndResults => engineAndResults.results 
); 

Wie Sie sehen können, mißbrauchen ich den Akkumulator einen Motor pro Thread zu tragen. Das Problem hierbei ist, dass PLINQ am Ende die Ergebnisse in einem einzigen IEnumerable aggregiert, wodurch die Threads synchronisiert werden. Das ist nicht sehr schön, wenn ich danach noch eine PLINQ-Erweiterung anhängen möchte.

würde ich so etwas wie

var result = source.AsParallel() 
        .SelectWithThreadwiseInitWhichIAmLookingFor (
         () => new Engine(), 
         (engine, item) => engine.Process (item) 
      ) 

Hat jemand eine Idee schätzen, wie dies zu erreichen?

Answer 1

Sie könnten dazu ThreadLocal<T> verwenden. Etwas wie:

var engine = new ThreadLocal<Engine>(() => new Engine()); 
var result = source.AsParallel() 
        .Select(item => engine.Value.Process(item));