c# - 如何在查询完成之前按顺序开始处理并行查询的结果？

Question

我有一个字符串集合，我需要对其执行两个操作。

其中第一个可以按任何顺序安全地独​​立处理 (yay)，但随后必须按原始顺序顺序处理 (boo) 输出。

以下 Plinq 帮助我完成了大部分工作:

myStrings.AsParallel().AsOrdered()
         .Select( str => Operation1(str) )
         .AsSequential()
         .Select( str => Operation2(str) );
//immagine Operation2() maintains some sort of state and must take the outputs from Operation1 in the original order    

这让我了解了大部分内容，但问题是由于 AsOrdered()，Operation1 首先在每个字符串上执行，然后结果元素被排序回它们的原始顺序，最后 Operation2 开始执行。

理想情况下，一旦 Operation1 调用返回第一个字符串(即 myStrings[0]，而不是第一个返回的字符串)，我希望 Operation2 开始工作。

所以这是我尝试解决一般性问题:

public static class ParallelHelper
{
    public static IEnumerable<U> SelectAsOrdered<T, U>(this ParallelQuery<T> query, Func<T, U> func)
    {
        var completedTasks = new Dictionary<int, U>();
        var queryWithIndexes = query.Select((x, y) => new { Input = x, Index = y })
                                    .AsParallel()
                                    .Select(t => new { Value = func(t.Input), Index = t.Index })
                                    .WithMergeOptions(ParallelMergeOptions.NotBuffered);

        int i = 0;
        foreach (var task in queryWithIndexes)
        {
            if (i==task.Index)
            {
                Console.WriteLine("immediately yielding task: {0}", i);
                i++;
                yield return task.Value;

                U previouslyCompletedTask;
                while (completedTasks.TryGetValue(i, out previouslyCompletedTask))
                {
                    completedTasks.Remove(i);
                    Console.WriteLine("delayed yielding task: {0}", i);
                    yield return previouslyCompletedTask;
                    i++;
                }
            }
            else
            {
                completedTasks.Add(task.Index, task.Value);
            }
        }
        yield break;
    }
}

然后我可以将我的原始代码块重写为:

myStrings.AsParallel()
         .SelectAsOrdered( str => Operation1(str) )
         .Select(str => Operation2(str));

一旦 myStrings[0] 从 Operation1 中出来，Operation2 就会启动。

我想知道的是:

1. 这是并行化中一个相当普遍的问题/模式，我是否错过了在 .Net 框架中执行此操作的开箱即用的东西？或者有更简单的方法吗？
2. 虽然上述扩展方法似乎可以完成工作，但如何对其进行改进？代码中有什么地方看起来像是个坏主意吗？

谢谢!

安迪

以防万一你感兴趣:

• 如果不调用 .WithMergeOptions(ParallelMergeOptions.NotBuffered)，Operation2 直到所有 Operation1 调用都已启动后才会开始工作(这比等待所有调用完成的原始代码要好)。

• 现实生活中的问题:
  Operation1 正在大量文本中搜索法律引用和引用文献(例如:“children act 1989”)。
  这些引用文献通常是独立的，但有时抄本会包含类似“前面提到的法案的第 6 节”之类的内容。 Operation2 依赖于 Operation1 的捕获来获取这些部分引用。

Answer 1

如果你需要速度，你可以并行化所有过程(加载数据、准备数据、处理数据和聚合数据)，我认为最好使用生产者/消费者模式。

但是，如果您使用“Linq”，则无法生成(以一种很好的方式来执行完整的并行工作流)并行数据(但是可以:准备、处理和恢复)。

另一方面，我认为尝试将“Linq”用作“并行(A) + 顺序(B)”是错误的(你可以，是的)，你的过程(我认为)是

B = f(A)

那么，B必须等待A。

为什么不简单地做“并行(A/B)”？

你可以做一个帮助程序(扩展)，但我认为它通常没有用。

在您的真实情况下，只需使用 Semaphore 来防止过早访问“文章 ID”。

并行准备、处理和恢复(不生成)的完整代码是:

class Text {
    public static Regex rx = new Regex(@" (PREVID|ACTID\=([0-9]+)) ");

    private Text prv; // previous article
    private string ot; // original text
    private int id; // act id on text
    private Semaphore isComputed = new Semaphore(0, 1);

    public int ActID {
        get {
            isComputed.WaitOne();
            int _id = id;
            isComputed.Release();
            return _id;
        }
    }

    public bool ProcessText() {
        var mx = rx.Match(ot);
        var prev = mx.Groups [1].Value == "PREVID";
        if(prev)
            id = prv == null ? 0 : prv.ActID;
        else
            if(!int.TryParse(mx.Groups [2].Value, out id))
                throw new Exception(string.Format(@"Incorrect article id ""{0}""", mx.Groups [0].Value));
        isComputed.Release();
        return !prev;
    }

    public Text(string original_text, Text previous) {
        prv = previous;
        ot = original_text;
    }

}

public static void Main(String [] args) {

    // same random stream (for debugging)
    var rnd = new Random(1);

    var noise = @"These references are usually independent, but occasionally";

    // some noise text
    var bit = new Func<string>(() =>
        noise.Substring(0, rnd.Next(noise.Length)));

    // random article
    var text = new Func<string>(() =>
        string.Format(@"{0}{1}{2}", bit(),
            rnd.Next() % 2 == 0 ? " PREVID "
                                : string.Format(@" ACTID={0} ", rnd.Next()), bit()));

    // random data input
    var data = new List<Text>();
    Text prv = null;
    for(var n = 0; n < 1000000; n++)
        // producer / consumer is better to parallelize load data step
        data.Add(prv = new Text(text(), prv));

    Console.Write("Press key to start...");
    Console.ReadKey();

    // parallel processing
    Console.WriteLine("{0} unique ID's", data.AsParallel().Where(n => n.ProcessText()).Count());

    Console.WriteLine("Process completed.");
}

如您所见，ProcessText 并行处理所有文章。只有 PREVID 文章会等到它们的前一篇文章计算它们自己的 ID。

抽象此行为的真正问题(我认为)是项目关系(一个项目依赖于另一个项目)，在 Linq 中，自然的方式是“无项目关系”(您必须使用“分组依据”来执行它).

我建议你使用生产者/消费者模式。