如何在等待时保持消息泵送？

Question

我有一个基于消息泵线程池架构的应用程序。 每当有一个操作可能阻塞时，它就会被实现为“回调完成/触发 evnet”操作，因此它不会停止执行线程。

虽然这种技术适用于大多数情况，但在某些情况下它会变得非常不方便并且使代码过于复杂。

我希望能够做的是，在等待时以透明的方式继续处理事件，而不将函数分解为前/后等待部分。

我该怎么做？

我想到了两个选择：

1. 在等待时从执行函数内运行消息循环。
2. 在等待时创建一个新的工作线程，并在恢复时终止它（以适当的方式）。

这两个选项都有其缺陷，仅举几例：

对于 1：

• 可能会导致堆栈溢出。
• 可能最终会陷入僵局。
• 如果内部消息导致等待第二个事件完成，而外部事件同时完成，则外部函数在第二个事件完成之前无法继续，这种情况可能会扩大。

选项 2 最终可能会创建越来越多的线程。

当然，可能还有其他我没有想到的选择。

编辑：语言是C++，因此不能以简单（可移植？）的方式退出和进入函数。 平台是Windows（API），尽管我认为它不相关。

Answer 1

对于可移植的 C++，这是行不通的，但既然您提到您的平台是 Windows，为什么不使用 MsgWaitForMultipleObjects？ 它的目的是让你完全按照你的问题所说的去做——在等待的时候继续发送消息。

Answer 2

编辑：您提到不想“将功能分解为前/后等待部分”。

你用什么语言开发？ 如果它有延续（C# 中的yield return），那么这就提供了一种编写代码的方法，该代码看起来是过程性的，但可以轻松地暂停，直到阻塞操作进行完成回调为止。

这是一篇关于这个想法的文章： http://msdn.microsoft.com/en -us/magazine/cc546608.aspx

更新：

不幸的是，语言是 C++

这将成为一个很棒的 T 恤标语。

好的，您可能会发现将顺序代码构造为状态机很有帮助，因此它具有中断/恢复功能。

例如，您的痛苦是需要编写两个函数，一个启动函数，另一个充当完成事件的处理程序：

void send_greeting(const std::string &msg)
{
    std::cout << "Sending the greeting" << std::endl;
    begin_sending_string_somehow(msg, greeting_sent_okay);
}

void greeting_sent_okay()
{
    std::cout << "Greeting has been sent successfully." << std::endl;
}

您的想法是等待：

void send_greeting(const std::string &msg)
{
    std::cout << "Sending the greeting" << std::endl;

    waiter w;
    begin_sending_string_somehow(msg, w);
    w.wait_for_completion();

    std::cout << "Greeting has been sent successfully." << std::endl;
}

在该示例中，waiter 重载了operator()，因此它可以充当回调，并且 wait_for_completion 会以某种方式挂起，直到它看到operator() 已被调用。

我假设 begin_sending_string_somehow 的第二个参数是一个模板参数，它可以是任何不接受参数的可调用类型。

但正如你所说，这有缺点。 每当一个线程像这样等待时，您就添加了另一个潜在的死锁，并且您还消耗了整个线程及其堆栈的“资源”，这意味着必须在其他地方创建更多线程才能完成工作，这与线程池的整体观点相矛盾。

因此，请编写一个类：

class send_greeting
{
    int state_;
    std::string msg_;

public:
    send_greeting(const std::string &msg)
        : state_(0), msg_(msg) {}

    void operator()
    {
        switch (state_++)
        {
            case 0:
                std::cout << "Sending the greeting" << std::endl;
                begin_sending_string_somehow(msg, *this);
                break;

            case 1:
                std::cout << "Greeting has been sent successfully." 
                          << std::endl;
                break;
        }
    }
};

该类实现函数调用运算符 ()。 每次调用它时，它都会执行逻辑中的下一步。 （当然，作为一个如此简单的示例，现在这主要是状态管理噪音，但在具有四个或五个状态的更复杂的示例中，它可能有助于阐明代码的顺序性质）。

问题：

• 如果事件回调函数签名具有特殊参数，则需要添加另一个 operator() 重载，将参数存储在额外字段中，然后调用无参数重载。 然后它开始变得混乱，因为这些字段将在初始状态下的编译时访问，即使它们在该状态下的运行时没有意义。

• 如何构造和删除该类的对象？ 对象必须存活直到操作完成或被放弃......这是 C++ 的主要陷阱。 我建议实施一个通用方案来管理它。 创建一个“需要删除的东西”的列表，并确保在某些安全点自动发生，即尝试尽可能接近 GC。 距离越远，泄漏的内存就越多。

Answer 3

如果不了解有关您的特定应用程序的更多信息（即处理消息需要多长时间等），就会有很多犹豫：

• 这是托管还是非托管 C++？

• 您使用哪个线程池？

  • 队列用户工作项？
  • 您通过 CreateIoCompletionPort 拥有自己的池？
  • 还是 Vista 的 SubmitThreadpoolWork？

我认为平台有些相关，因为线程池的性质很重要。

例如：

如果您使用 (完成端口 ）为您的线程池（即CreateIoCompletionPort）。 您可以控制并发运行的线程数量（从而控制最终创建的线程总数）。 如果您将最大并发线程数设置为 4。Windows 将尝试仅允许 4 个线程同时运行。 如果所有 4 个线程都忙于处理，并且您将第 5 个项目排队，那么 Windows 将不允许该项目运行，直到 4 个线程之一完成（重用线程）。 唯一违反此规则的情况是当线程被阻塞（即等待 I/O）时，然后允许更多线程运行。

这是了解完成端口以及平台为何相关的重要内容。 在不涉及内核的情况下实现这样的事情是非常困难的。 了解繁忙线程和阻塞线程之间的区别需要访问线程状态。 就进入内核的上下文切换数量而言，完成端口也非常高效。

回到您的问题：

您似乎应该有一个线程来处理/分派消息，并且消息处理都是通过将工作线程推送到线程池来处理的。 让完成端口处理负载平衡和并发。 您的消息处理循环永远不会阻塞，并且可以继续处理消息。

如果传入消息的速率远远超出您的处理能力，那么您可能需要注意队列大小，并在队列变得太大时进行阻塞。

Answer 4

看来你的问题是根本性的，与 C++ 无关。 其他语言在隐藏堆栈使用方面可能更好，但只要您没有从 Foo() 返回，您就需要 Foo() 的调用堆栈。 如果您还执行 Bar()，那么它也需要一个调用堆栈。

线程是一个很好的方法，因为每个线程都有自己的调用堆栈。 延续是保存调用堆栈的一种聪明但复杂的方法，因此在可用的情况下，这也是一种选择。 但如果您不想要这些，则必须使用一个调用堆栈。

使用一个调用堆栈需要解决重入问题。 在这里，对于什么是可能的，没有通用的答案。 一般来说，您将拥有一组消息 M1..Mx，它们由函数 F1...Fy 处理，并具有一些特定于应用程序和可能与状态相关的映射。 对于可重入消息循环，您可能在收到 Mj 时正在执行 Fi。 现在的问题是该怎么办。 并非所有函数 F1...Fn 都可以调用； 特别是 Fi 本身可能不可调用。 然而，一些其他功能也可能不可用，例如因为它们共享资源。 这取决于应用程序。

如果 Mj 的处理需要任何这些不可用的函数，则必须推迟它。 您可以接受队列中的下一条消息吗？ 同样，这取决于实现，甚至可能与消息类型和内容有关。 如果消息足够独立，则可以乱序执行它们。 这很快就会变得相当复杂 - 为了确定是否可以接受队列中的第 N 条消息，您必须检查它是否可以相对于前面的 N-1 条消息乱序执行。

语言可以通过不隐藏依赖关系来帮助您，但最终您必须做出明确的决定。 没有灵丹妙药。

Answer 5

您的问题是同步线程对吧？ 如果这是您的问题，为什么不使用互斥锁呢？ 它可以用一个接口包裹起来。 事实上，您可以使用 PIMPL 习惯用法使互斥体可移植。

http://msdn.microsoft.com /en-us/library/system.threading.mutex(VS.71).aspx