是否有 PLINQ 的异步版本？

Question

我想在以一定的并行度并行处理项目的同时对数据流执行查询。通常，我会为此使用 PLINQ，但我的工作项不受 CPU 限制，而是 IO 限制。我想使用异步 IO。 PLINQ 不支持异步工作。

运行 PLINQ 样式查询但使用异步工作项的最聪明的方法是什么？

---

这里有一个更详细的问题说明：

我的目标是以以下查询在逻辑上描述的方式处理可能无限的“项目”流：

var items = new int[10]; //simulate data

var results =
 from x in items.AsParallel().WithDegreeOfParallelism(100)
 where Predicate(x)
 select ComputeSomeValue(x);

foreach (var result in results)
 PerformSomeAction(result);

这个查询只是真实查询的草图。现在我希望每个占位符函数都是异步（返回一个Task 并且在内部基于异步IO）。

请注意，内存中存储的项目可能远远多于可存储的项目。我还必须控制并行度，以最大限度地利用底层网络和磁盘硬件。

这个问题与多核无关。它完全适用于只有一个 CPU 内核的机器，因为 IO 仍然可以从并行性中受益。想想缓慢的网络服务调用等。

Answer 1

这听起来像是微软响应式框架的工作。

我从这个代码开始作为我的初始变量：

var items = Enumerable.Range(0, 10).ToArray();

Func<int, bool> Predicate = x => x % 2 == 0;

Func<int, int> ComputeSomeValue = x =>
{
    Thread.Sleep(10000);
    return x * 3;
};

现在，我使用常规 LINQ 查询作为基线：

var results =
    from x in items
    where Predicate(x)
    select ComputeSomeValue(x);

计算以下结果需要 50 秒：

然后我切换到可观察（反应式框架）查询：

var results =
    from x in items.ToObservable()
    where Predicate(x)
    from y in Observable.Start(() => ComputeSomeValue(x))
    select y;

这需要 10 秒才能得到：

显然是并行计算。

但是，结果是乱序的。所以我把查询改成这样：

var query =
    from x in items.ToObservable()
    where Predicate(x)
    from y in Observable.Start(() => ComputeSomeValue(x))
    select new { x, y };

var results =
    query
        .ToEnumerable()
        .OrderBy(z => z.x)
        .Select(z => z.y);

这仍然需要 10 秒，但我以正确的顺序返回结果。

现在，这里唯一的问题是WithDegreeOfParallelism。这里有很多东西可以尝试。

首先，我更改了代码以生成 10,000 个值，计算时间为 10 毫秒。我的标准 LINQ 查询仍然需要 50 秒。但反应式查询耗时 6.3 秒。如果它可以同时执行所有计算，它应该花费更少的时间。这表明它正在最大化异步管道。

第二点是响应式框架使用调度器进行所有工作调度。如果内置的调度程序不能满足您的要求，您可以尝试响应式框架附带的各种调度程序来寻找替代方案。或者您甚至可以编写自己的调度程序来执行您喜欢的任何调度。

---

这是一个并行计算谓词的查询版本。

var results =
    from x in items.ToObservable()
    from p in Observable.Start(() => Predicate(x))
    where p
    from y in Observable.Start(() => ComputeSomeValue(x))
    select new { x, y };

Answer 2

如here 所述，PLINQ 用于在多核/多处理器 系统上并行运行LINQ 查询。对于具有大量磁盘单元和超级网络功能的酷系统来说，这并没有太多的事情要做。 AFAIK，它用于在更多内核上运行可执行代码，而不是用于同时向操作系统分派多个 I/O 请求。

也许您的 Predicate(x) 受 CPU 限制，因此您可以使用 PLINQ 执行过滤操作。但是您不能以相同的方式应用 I/O 要求高的操作（ComputeSomeValue、PerformSomeAction）。

您可以做的是为每个项目定义一个操作链（在您的情况下为两个）（请参阅continuation tasks）并调度该链（顺序（？））。

此外，您还提到了关于“无限项流”。这听起来有点像生产者-消费者问题 - 如果这些项目也是 I/O 生成的。

也许你的问题不是多核友好... 可能只是 I/O 要求高，仅此而已...