在 PLINQ 中绑定源线程

Question

我有一个使用 PLINQ 并行化的计算，如下所示：

• Source IEnumerable<T> source 正在提供从 文件。

• 我有一个重量级计算 HeavyComputation 我需要做 每个T，我希望这些跨线程分流，所以我 使用 PLINQ，例如：AsParallel().Select(HeavyComputation)

这里是有趣的地方：由于文件的限制 提供source 的阅读器类型，我需要source 在初始线程上枚举，而不是在并行工作者上。我需要 source 的完整评估将 绑定 到主 线。然而，似乎来源实际上是在工人身上列举的 线程。

我的问题是：有没有一种直接的方法可以修改此代码以 将source 的枚举绑定到初始线程，而 将繁重的工作分给并行工人？请记住， 只是在AsParallel() 之前做一个急切的.ToList() 不是这里的选择， 因为来自文件的数据流很大。

下面是一些示例代码，可以演示我看到的问题：

using System.Threading;
using System.Collections.Generic;
using System.Linq;
using System;

public class PlinqTest
{
        private static string FormatItems<T>(IEnumerable<T> source)
        {
                return String.Format("[{0}]", String.Join(";", source));
        }

        public static void Main()
        {
            var expectedThreadIds = new[] { Thread.CurrentThread.ManagedThreadId };

            var threadIds = Enumerable.Range(1, 1000)
                    .Select(x => Thread.CurrentThread.ManagedThreadId) // (1)
                    .AsParallel()
                    .WithDegreeOfParallelism(8)
                    .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
                    .AsOrdered()
                    .Select(x => x)                                    // (2)
                    .ToArray();

            // In the computation above, the lambda in (1) is a
            // stand in for the file-reading operation that we
            // want to be bound to the main thread, while the
            // lambda in (2) is a stand-in for the "expensive
            // computation" that we want to be farmed out to the
            // parallel worker threads.  In fact, (1) is being
            // executed on all threads, as can be seen from the
            // output.

            Console.WriteLine("Expected thread IDs: {0}",
                              FormatItems(expectedThreadIds));
            Console.WriteLine("Found thread IDs: {0}",
                              FormatItems(threadIds.Distinct()));
        }
}

我得到的示例输出是：

Expected thread IDs: [1]
Found thread IDs: [7;4;8;6;11;5;10;9]

Answer 1

如果您放弃 PLINQ 并明确使用任务并行库，这相当简单（尽管可能不那么简洁）：

// Limits the parallelism of the "expensive task"
var semaphore = new SemaphoreSlim(8);

var tasks = Enumerable.Range(1, 1000)
    .Select(x => Thread.CurrentThread.ManagedThreadId)
    .Select(async x =>
    {
        await semaphore.WaitAsync();
        var result = await Task.Run(() => Tuple.Create(x, Thread.CurrentThread.ManagedThreadId));
        semaphore.Release();

        return result;
    });

return Task.WhenAll(tasks).Result;

请注意，我使用Tuple.Create 来记录来自主线程的线程 ID 和来自衍生任务的线程 ID。根据我的测试，对于每个元组，前者总是相同的，而后者则各不相同，这是应该的。

信号量确保并行度永远不会超过 8（尽管创建元组的廉价任务无论如何都不太可能）。如果达到 8，任何新任务都将等到信号量上有可用点。

Answer 2

您可以使用下面的OffloadQueryEnumeration 方法，该方法确保源序列的枚举将发生在枚举结果IEnumerable<TResult> 的同一线程上。 querySelector 是将源序列的代理转换为ParallelQuery<T> 的委托。此查询在ThreadPool 线程上进行内部枚举，但输出值会返回到当前线程。

/// <summary>Enumerates the source sequence on the current thread, and enumerates
/// the projected query on a ThreadPool thread.</summary>
public static IEnumerable<TResult> OffloadQueryEnumeration<TSource, TResult>(
    this IEnumerable<TSource> source,
    Func<IEnumerable<TSource>, IEnumerable<TResult>> querySelector)
{
    // Arguments validation omitted
    var locker = new object();
    (TSource Value, bool HasValue) input = default; bool inputCompleted = false;
    (TResult Value, bool HasValue) output = default; bool outputCompleted = false;
    using var sourceEnumerator = source.GetEnumerator();

    IEnumerable<TSource> GetSourceProxy()
    {
        while (true)
        {
            TSource item;
            lock (locker)
            {
                if (!input.HasValue)
                {
                    if (inputCompleted) yield break;
                    Monitor.Wait(locker); continue;
                }
                item = input.Value; input = default;
                Monitor.PulseAll(locker);
            }
            yield return item;
        }
    }

    var query = querySelector(GetSourceProxy());

    var outputReaderTask = Task.Run(() =>
    {
        try
        {
            foreach (var result in query)
            {
                lock (locker)
                {
                    while (output.HasValue) Monitor.Wait(locker);
                    output = (result, true);
                    Monitor.PulseAll(locker);
                }
            }
        }
        finally
        {
            lock (locker) { outputCompleted = true; Monitor.PulseAll(locker); }
        }
    });

    List<Exception> exceptions = new();
    while (true)
    {
        TResult result;
        lock (locker)
        {
            if (output.HasValue)
            {
                result = output.Value; output = default;
                Monitor.PulseAll(locker);
                goto yieldResult;
            }
            if (outputCompleted)
            {
                if (!inputCompleted)
                {
                    inputCompleted = true; Monitor.PulseAll(locker);
                }
                break;
            }
            if (input.HasValue || inputCompleted)
            {
                Monitor.Wait(locker); continue;
            }
            try
            {
                if (sourceEnumerator.MoveNext())
                    input = (sourceEnumerator.Current, true);
                else
                    inputCompleted = true;
            }
            catch (Exception ex)
            {
                exceptions.Add(ex);
                inputCompleted = true;
            }
            Monitor.PulseAll(locker); continue;
        }
    yieldResult:
        yield return result;
    }

    // Propagate possible exceptions
    try { outputReaderTask.GetAwaiter().GetResult(); }
    catch (OperationCanceledException) { throw; }
    catch (AggregateException aex) { exceptions.AddRange(aex.InnerExceptions); }
    catch (Exception ex) { exceptions.Add(ex); } // Fallback (should never happen)

    if (exceptions.Count > 0) throw new AggregateException(exceptions);
}

此方法使用Monitor.Wait/Monitor.Pulse 机制 (tutorial)，以同步从一个线程到另一个线程的值传输。

使用示例：

int[] threadIds = Enumerable
    .Range(1, 1000)
    .Select(x => Thread.CurrentThread.ManagedThreadId)
    .OffloadQueryEnumeration(proxy => proxy
        .AsParallel()
        .WithDegreeOfParallelism(8)
        .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
        .AsOrdered()
        .Select(x => x)
    )
    .ToArray();