Как параллельно выполнять LINQ 'let'?

У меня такой код:

var list = new List<int> {1, 2, 3, 4, 5};

var result = from x in list.AsParallel()
             let a = LongRunningCalc1(x)
             let b = LongRunningCalc2(x)
             select new {a, b};

Скажем, что методы LongRunningCalc занимают 1 секунду. Приведенный выше код занимает около 2 секунд для запуска, потому что в то время как список из 5 элементов работает параллельно, два метода, вызываемые из операторов let, вызываются последовательно.

Однако эти методы можно смело вызывать параллельно. Очевидно, что им нужно объединиться для select, но до тех пор, пока он не будет запускаться параллельно - select должен их ждать.

Есть ли способ достичь этого?

Ответы

Ответ 1

Вы не сможете использовать синтаксис запроса или операцию let, но вы можете написать метод для выполнения нескольких операций для каждого элемента параллельно:

public static ParallelQuery<TFinal> SelectAll<T, TResult1, TResult2, TFinal>(
    this ParallelQuery<T> query,
    Func<T, TResult1> selector1,
    Func<T, TResult2> selector2,
    Func<TResult1, TResult2, TFinal> resultAggregator)
{
    return query.Select(item =>
    {
        var result1 = Task.Run(() => selector1(item));
        var result2 = Task.Run(() => selector2(item));
        return resultAggregator(result1.Result, result2.Result);
    });
}

Это позволит вам написать:

var query = list.AsParallel()
    .SelectAll(LongRunningCalc1, 
        LongRunningCalc2, 
        (a, b) => new {a, b})

Вы также можете добавить перегрузки для дополнительных параллельных операций:

public static ParallelQuery<TFinal> SelectAll<T, TResult1, TResult2, TResult3, TFinal>
    (this ParallelQuery<T> query,
    Func<T, TResult1> selector1,
    Func<T, TResult2> selector2,
    Func<T, TResult3> selector3,
    Func<TResult1, TResult2, TResult3, TFinal> resultAggregator)
{
    return query.Select(item =>
    {
        var result1 = Task.Run(() => selector1(item));
        var result2 = Task.Run(() => selector2(item));
        var result3 = Task.Run(() => selector3(item));
        return resultAggregator(
            result1.Result,
            result2.Result,
            result3.Result);
    });
}

Можно написать версию для обработки ряда селекторов, неизвестных во время компиляции, но для этого все они должны вычислить значение одного и того же типа:

public static ParallelQuery<IEnumerable<TResult>> SelectAll<T, TResult>(
    this ParallelQuery<T> query,
    IEnumerable<Func<T, TResult>> selectors)
{
    return query.Select(item => selectors.AsParallel()
            .Select(selector => selector(item))
            .AsEnumerable());
}
public static ParallelQuery<IEnumerable<TResult>> SelectAll<T, TResult>(
    this ParallelQuery<T> query,
    params Func<T, TResult>[] selectors)
{
    return SelectAll(query, selectors);
}

Ответ 2

Я бы сделал это с помощью Microsoft Reactive Framework ( "Rx-Main" в NuGet).

Вот он:

var result =
    from x in list.ToObservable()
    from a in Observable.Start(() => LongRunningCalc1(x))
    from b in Observable.Start(() => LongRunningCalc2(x))
    select new {a, b};

Хорошо, что вы можете получить доступ к результатам по мере их создания с помощью метода .Subscribe(...):

result.Subscribe(x => /* Do something with x.a and/or x.b */ );

Супер простой!