使用 CCR 有序处理事件的有效方法是什么？

Question

我正在尝试使用 CCR 迭代器作为一项任务的解决方案，该任务需要并行处理大量数据源，其中每个源的数据都需要按顺序处理。所有提要都不相互依赖，因此可以按提要并行进行按顺序处理。

下面是一个带有一个整数馈送的快速而肮脏的模型，它只是以大约 1.5K/秒的速率将整数推入端口，然后使用 CCR 迭代器将它们拉出以保持按顺序处理保证。

class Program
{
    static Dispatcher dispatcher = new Dispatcher();
    static DispatcherQueue dispatcherQueue = 
       new DispatcherQueue("DefaultDispatcherQueue", dispatcher);
    static Port<int> intPort = new Port<int>();

    static void Main(string[] args)
    {
        Arbiter.Activate(
            dispatcherQueue,
            Arbiter.FromIteratorHandler(new IteratorHandler(ProcessInts)));

        int counter = 0;
        Timer t = new Timer( (x) => 
            { for(int i = 0; i < 1500; ++i) intPort.Post(counter++);}
              , null, 0, 1000);

        Console.ReadKey();
    }

    public static IEnumerator<ITask> ProcessInts()
    {
        while (true)
        {
            yield return intPort.Receive();
            int currentValue;
            if( (currentValue = intPort) % 1000 == 0)
            {
                Console.WriteLine("{0}, Current Items In Queue:{1}", 
                  currentValue, intPort.ItemCount);
            }
        }
    }
}

令我大吃一惊的是，CCR 无法跟上 Corei7 机器的速度，队列大小无限制地增长。在另一项测试中，在约 100 Post/sec 的负载下测量从 Post() 到 Receive() 的延迟，每批中的第一个 Post() 和 Receive() 之间的延迟约为 1 毫秒。

我的模型有问题吗？如果是这样，使用 CCR 执行此操作的更好方法是什么？

Answer 1

是的，我同意，这确实看起来很奇怪。您的代码最初似乎执行得很顺利，但是在几千个项目之后，处理器使用率上升到性能确实低迷的地步。这让我感到不安，并表明框架中存在问题。在玩过你的代码后，我无法真正确定为什么会出现这种情况。我建议将此问题带到 Microsoft 机器人论坛 并看看你能否让 George Chrysanthakopoulos（或其他 CCR 大脑之一）告诉你问题所在。不过，我可以推测您现在的代码效率非常低。

你处理从港口“弹出”的物品的方式是非常低效的。本质上，每当端口中有一条消息时，迭代器就会被唤醒，并且它只处理一条消息（尽管事实上端口中可能还有数百条消息），然后挂在yield上，而控制权被传递回框架。当产生的接收者导致迭代器的另一次“唤醒”时，许多消息已经填充了端口。从调度程序中拉出一个线程来仅处理单个项目（当许多项目同时堆积时）几乎肯定不是获得良好吞吐量的最佳方法。

我已经修改了您的代码，以便在屈服之后，我们检查端口以查看是否有任何其他消息排队并处理它们，从而在我们屈服返回框架之前完全清空端口。我还对您的代码进行了一些重构，以使用 CcrServiceBase ，这简化了您正在执行的某些任务的语法：

internal class Test:CcrServiceBase
{
    private readonly Port<int> intPort = new Port<int>();
    private Timer timer;
    public Test() : base(new DispatcherQueue("DefaultDispatcherQueue",
                                             new Dispatcher(0,
                                                            "dispatcher")))
    {

    }

    public void StartTest() {
        SpawnIterator(ProcessInts);
        var counter = 0;
        timer = new Timer(x =>
                          {
                              for (var i = 0; i < 1500; ++i)
                                  intPort.Post(counter++);
                          }
                          ,
                          null,
                          0,
                          1000);
    }

    public IEnumerator<ITask> ProcessInts()
    {
        while (true)
        {
            yield return intPort.Receive();
            int currentValue = intPort;
            ReportCurrent(currentValue);
            while(intPort.Test(out currentValue))
            {
                ReportCurrent(currentValue);
            }
        }
    }

    private void ReportCurrent(int currentValue)
    {
        if (currentValue % 1000 == 0)
        {
            Console.WriteLine("{0}, Current Items In Queue:{1}",
                              currentValue,
                              intPort.ItemCount);
        }
    }
}

或者，您可以完全取消迭代器，因为它在您的应用程序中并没有得到很好的使用。示例（尽管我不完全确定这对处理顺序有什么影响）：

internal class Test : CcrServiceBase
{
    private readonly Port<int> intPort = new Port<int>();
    private Timer timer;

    public Test() : base(new DispatcherQueue("DefaultDispatcherQueue",
                                             new Dispatcher(0,
                                                            "dispatcher")))
    {

    }

    public void StartTest()
    {
        Activate(
            Arbiter.Receive(true,
                            intPort,
                            i =>
                            {
                                ReportCurrent(i);
                                int currentValue;
                                while (intPort.Test(out currentValue))
                                {
                                    ReportCurrent(currentValue);
                                }
                            }));
        var counter = 0;
        timer = new Timer(x =>
                          {
                              for (var i = 0; i < 500000; ++i)
                              {
                                  intPort.Post(counter++);
                              }
                          }
                          ,
                          null,
                          0,
                          1000);
    }



    private void ReportCurrent(int currentValue)
    {
        if (currentValue % 1000000 == 0)
        {
            Console.WriteLine("{0}, Current Items In Queue:{1}",
                              currentValue,
                              intPort.ItemCount);
        }
    }
}

这两个示例都将吞吐量显着提高了几个数量级。希望这有帮助。