title: What is the best way to wait on multiple condition variables in C++11?
body: <p>First a little <strong>context</strong>: I'm in the process of learning about threading in C++11 and for this purpose, I'm trying to build a small <code>actor</code> class, essentially (I left the exception handling and propagation stuff out) like so:</p>
  
  <pre><code>class actor {
      private: std::atomic&lt;bool&gt; stop;
      private: std::condition_variable interrupt;
      private: std::thread actor_thread;
      private: message_queue incoming_msgs;
  
      public: actor() 
      : stop(false), 
        actor_thread([&amp;]{ 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&amp;) = 0;
      // ...
  };
  </code></pre>
  
  <p>Every actor runs in its own <code>actor_thread</code>, waits on a new incoming message on <code>incoming_msgs</code> and -- when a message arrives -- processes it. </p>
  
  <p>The <code>actor_thread</code> is created together with the <code>actor</code> and has to die together with it, which is why I need some kind of interrupt mechanism in the <code>message_queue::wait_and_pop(std::condition_variable interrupt)</code>. </p>
  
  <p>Essentially, I require that <code>wait_and_pop</code> blocks until either 
  a) a new <code>message</code> arrives or 
  b) until the <code>interrupt</code> is fired, in which case -- ideally -- an <code>interrupted_exception</code> is to be thrown.</p>
  
  <p>The arrival of a new message in the <code>message_queue</code> is presently modeled also by a <code>std::condition_variable new_msg_notification</code>:</p>
  
  <pre><code>// ...
  // in class message_queue:
  message wait_and_pop(std::condition_variable&amp; interrupt) {
      std::unique_lock&lt;std::mutex&gt; lock(mutex);
  
      // How to interrupt the following, when interrupt fires??
      new_msg_notification.wait(lock,[&amp;]{
          return !queue.empty();
      });
      auto msg(std::move(queue.front()));
      queue.pop();
      return msg;
  }
  </code></pre>
  
  <p>To cut the long story short, the <strong>question</strong> is this: How do I interrupt the waiting for a new message in <code>new_msg_notification.wait(...)</code> when the <code>interrupt</code> is triggered (without introducing a time-out)?</p>
  
  <p>Alternatively, the question may be read as: How do I wait until any one of two <code>std::condition_variable</code>s are signaled? </p>
  
  <p>One naive approach seems to be not to use <code>std::condition_variable</code> at all for the interrupt and instead just use an atomic flag <code>std::atomic&lt;bool&gt; interrupted</code> and then busy wait on <code>new_msg_notification</code> with a very small time-out until either a new message has arrived or until <code>true==interrupted</code>. However, I would very much like to avoid busy waiting.</p>
  
  <hr>
  
  <h2>EDIT:</h2>
  
  <p>From the comments and the answer by pilcrow, it looks like there are basically two approaches possible.</p>
  
  <ol>
  <li>Enqueue a special "Terminate" message, as proposed by Alan, mukunda and pilcrow. I decided against this option because I have no idea about the size of the queue at the time I want the actor to terminate. It may very well be (as it is mostly the case when I want something to quickly terminate) that there are thousands of messages left to process in the queue and it seems unacceptable to wait for them to be processed until finally the terminate message gets its turn.</li>
  <li>Implement a custom version of a condition variable, that may be interrupted by another thread by forwarding the notification to the condition variable that the first thread is waiting on. I opted for this approach.</li>
  </ol>
  
  <p>For those of you interested, my implementation goes as follows. The condition variable in my case is actually a <code>semaphore</code> (because I like them more and because I liked the exercise of doing so). I equipped this semaphore with an associated <code>interrupt</code> which can be obtained from the semaphore via <code>semaphore::get_interrupt()</code>. If now one thread blocks in <code>semaphore::wait()</code>, another thread has the possibility to call <code>semaphore::interrupt::trigger()</code> on the interrupt of the semaphore, causing the first thread to unblock and propagate an <code>interrupt_exception</code>.</p>
  
  <pre><code>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&lt;long&gt; counter;
      private: std::condition_variable cond;
  
      public: 
      semaphore();
  
      public: 
      ~semaphore() throw();
  
      public: void 
      wait();
  
      public: interrupt&amp;
      get_interrupt() const { return *informed_by; }
  
      public: void
      post() {
          std::lock_guard&lt;std::mutex&gt; 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&lt;bool&gt; triggered;
  
      public:
      interrupt(semaphore *forward_posts_to) : triggered(false), forward_posts_to(forward_posts_to) {
          assert(forward_posts_to);
          std::lock_guard&lt;std::mutex&gt; lock(forward_posts_to-&gt;mutex);
          forward_posts_to-&gt;informed_by = this;
      }
  
      public: void
      trigger() {
          assert(forward_posts_to);
          std::lock_guard&lt;std::mutex&gt;(forward_posts_to-&gt;mutex);
  
          triggered = true;
          forward_posts_to-&gt;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&lt;std::mutex&gt; lock(mutex);
      if(0L==counter) {
          cond.wait(lock,[&amp;]{
              if(informed_by-&gt;is_triggered())
                  throw interrupt_exception();
              return counter&gt;0;
          });
      }
      counter--;
  }
  </code></pre>
  
  <p>Using this <code>semaphore</code>, my message queue implementation now looks like this (using the semaphore instead of the <code>std::condition_variable</code> I could get rid of the <code>std::mutex</code>:</p>
  
  <pre><code>class
  message_queue {    
      private: std::queue&lt;message&gt; queue;
      private: semaphore new_msg_notification;
  
      public: void
      push(message&amp;&amp; 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&amp;
      get_interrupt() const { return new_msg_notification.get_interrupt(); }
  };
  </code></pre>
  
  <p>My <code>actor</code>, is now able to interrupt its thread with very low latency in its thread. The implementation presently like this:</p>
  
  <pre><code>class
  actor {
      private: message_queue
      incoming_msgs;
  
      /// must be declared after incoming_msgs due to construction order!
      private: semaphore::interrupt&amp;
      interrupt;
  
      private: std::thread
      my_thread;
  
      private: std::exception_ptr
      exception;
  
      public:
      actor()
      : interrupt(incoming_msgs.get_interrupt()), my_thread(
          [&amp;]{
              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&amp;) = 0;
  
      public: void
      notify(message&amp;&amp; msg_in) {
          incoming_msgs.push(std::forward&lt;message&gt;(msg_in));
      }
  
      public: virtual
      ~actor() throw (interrupt_exception) {
          interrupt.trigger();
          my_thread.join();
          if(exception)
              std::rethrow_exception(exception);
      }
  };
  </code></pre>