在C++ 11中等待多个条件变量的最佳方法是什么？

Question

首先是一个小小的上下文:我正在学习C++ 11中的线程,为此目的,我正在尝试构建一个actor小类,基本上(我将异常处理和传播内容留下),如下所示:

class actor {
    private: std::atomic<bool> stop;
    private: std::condition_variable interrupt;
    private: std::thread actor_thread;
    private: message_queue incoming_msgs;

    public: actor() 
    : stop(false), 
      actor_thread([&]{ run_actor(); })
    {}

    public: virtual ~actor() {
        // if the actor is destroyed, we must ensure the thread dies too
        stop = true;
        // to this end, we have to interrupt the actor thread which is most probably
        // waiting on the incoming_msgs queue:
        interrupt.notify_all();
        actor_thread.join();
    }

    private: virtual void run_actor() {
        try {
            while(!stop)
                // wait for new message and process it
                // but interrupt the waiting process if interrupt is signaled:
                process(incoming_msgs.wait_and_pop(interrupt));
        } 
        catch(interrupted_exception) {
            // ...
        }
    };

    private: virtual void process(const message&) = 0;
    // ...
};

每个actor都自己运行actor_thread,等待新的传入消息,incoming_msgs并在消息到达时处理它.

在actor_thread与共同创造actor并拥有与它同归于尽,这就是为什么我需要在某种中断机制message_queue::wait_and_pop(std::condition_variable interrupt).

基本上,我要求wait_and_pop阻止直到a)新的message到来或b)直到interrupt被解雇,在这种情况下 - 理想情况下 - interrupted_exception将被抛出.

新消息的到来message_queue目前也由以下方式建模std::condition_variable new_msg_notification:

// ...
// in class message_queue:
message wait_and_pop(std::condition_variable& interrupt) {
    std::unique_lock<std::mutex> lock(mutex);

    // How to interrupt the following, when interrupt fires??
    new_msg_notification.wait(lock,[&]{
        return !queue.empty();
    });
    auto msg(std::move(queue.front()));
    queue.pop();
    return msg;
}

简而言之,问题是:如何在触发new_msg_notification.wait(...)时中断等待新消息interrupt(不引入超时)？

或者,问题可以解读为:如何等到两个中std::condition_variable的任何一个发出信号？

一种天真的方法似乎根本就std::condition_variable不是用于中断,而只是使用原子标志std::atomic<bool> interrupted,然后忙着等待new_msg_notification很短的超时,直到新消息到达或直到true==interrupted.但是,我非常想避免忙碌的等待.

---

编辑:

从评论和pilcrow的答案看,基本上有两种可能的方法.

1. 排列一个特殊的"终止"消息,由Alan,mukunda和pilcrow提出.我决定反对这个选项,因为在我希望actor终止时我不知道队列的大小.它很可能(因为在我希望快速终止的情况下大多数情况下)在队列中有数千条消息需要处理,等待它们被处理似乎是不可接受的,直到最终终止消息得到它转.
2. 实现条件变量的自定义版本,可以通过将通知转发到第一个线程正在等待的条件变量来中断另一个线程.我选择了这种方法.

对于您感兴趣的人,我的实现如下.在我的情况下,条件变量实际上是一个semaphore(因为我更喜欢它们,因为我喜欢这样做的练习).我为这个信号量配备了一个interrupt可以从信号量获得的相关信号semaphore::get_interrupt().如果现在有一个线程阻塞semaphore::wait(),另一个线程有可能调用semaphore::interrupt::trigger()信号量的中断,导致第一个线程解除阻塞并传播interrupt_exception.

struct
interrupt_exception {};

class
semaphore {
    public: class interrupt;
    private: mutable std::mutex mutex;

    // must be declared after our mutex due to construction order!
    private: interrupt* informed_by;
    private: std::atomic<long> counter;
    private: std::condition_variable cond;

    public: 
    semaphore();

    public: 
    ~semaphore() throw();

    public: void 
    wait();

    public: interrupt&
    get_interrupt() const { return *informed_by; }

    public: void
    post() {
        std::lock_guard<std::mutex> lock(mutex);
        counter++;
        cond.notify_one(); // never throws
    }

    public: unsigned long
    load () const {
        return counter.load();
    }
};

class
semaphore::interrupt {
    private: semaphore *forward_posts_to;
    private: std::atomic<bool> triggered;

    public:
    interrupt(semaphore *forward_posts_to) : triggered(false), forward_posts_to(forward_posts_to) {
        assert(forward_posts_to);
        std::lock_guard<std::mutex> lock(forward_posts_to->mutex);
        forward_posts_to->informed_by = this;
    }

    public: void
    trigger() {
        assert(forward_posts_to);
        std::lock_guard<std::mutex>(forward_posts_to->mutex);

        triggered = true;
        forward_posts_to->cond.notify_one(); // never throws
    }

    public: bool
    is_triggered () const throw() {
        return triggered.load();
    }

    public: void
    reset () throw() {
        return triggered.store(false);
    }
};

semaphore::semaphore()  : counter(0L), informed_by(new interrupt(this)) {}

// must be declared here because otherwise semaphore::interrupt is an incomplete type
semaphore::~semaphore() throw()  {
    delete informed_by;
}

void
semaphore::wait() {
    std::unique_lock<std::mutex> lock(mutex);
    if(0L==counter) {
        cond.wait(lock,[&]{
            if(informed_by->is_triggered())
                throw interrupt_exception();
            return counter>0;
        });
    }
    counter--;
}

使用这个semaphore,我的消息队列实现现在看起来像这样(使用信号量而不是std::condition_variable我可以摆脱std::mutex:

class
message_queue {    
    private: std::queue<message> queue;
    private: semaphore new_msg_notification;

    public: void
    push(message&& msg) {
        queue.push(std::move(msg));
        new_msg_notification.post();
    }

    public: const message
    wait_and_pop() {
        new_msg_notification.wait();
        auto msg(std::move(queue.front()));
        queue.pop();
        return msg;
    }

    public: semaphore::interrupt&
    get_interrupt() const { return new_msg_notification.get_interrupt(); }
};

我的actor,现在能够在其线程中以非常低的延迟中断其线程.目前的实施如下:

class
actor {
    private: message_queue
    incoming_msgs;

    /// must be declared after incoming_msgs due to construction order!
    private: semaphore::interrupt&
    interrupt;

    private: std::thread
    my_thread;

    private: std::exception_ptr
    exception;

    public:
    actor()
    : interrupt(incoming_msgs.get_interrupt()), my_thread(
        [&]{
            try {
                run_actor();
            }
            catch(...) {
                exception = std::current_exception();
            }
        })
    {}

    private: virtual void
    run_actor() {
        while(!interrupt.is_triggered())
            process(incoming_msgs.wait_and_pop());
    };

    private: virtual void
    process(const message&) = 0;

    public: void
    notify(message&& msg_in) {
        incoming_msgs.push(std::forward<message>(msg_in));
    }

    public: virtual
    ~actor() throw (interrupt_exception) {
        interrupt.trigger();
        my_thread.join();
        if(exception)
            std::rethrow_exception(exception);
    }
};

Answer 1

你问,

在C++ 11中等待多个条件变量的最佳方法是什么？

你不能,也必须重新设计.一个线程一次只能等待一个条件变量(及其关联的互斥锁).在这方面,用于同步的Windows工具比"POSIX风格"同步原语系列的工具更丰富.

使用线程安全队列的典型方法是将一个特殊的"全部完成!"排队.消息,或设计"可破解"(或"可关闭")队列.在后一种情况下,队列的内部条件变量然后保护复杂谓词:项目可用或队列已被破坏.

在评论中你观察到了这一点

如果没有人在等待,notify_all()将无效

这是真的,但可能不相关. wait()条件变量也意味着检查谓词,并在实际阻止通知之前检查它.因此,工作线程忙于处理"未命中"a的队列项notify_all()将在下次检查队列条件时看到谓词(新项目可用,或队列全部完成)已更改.