title: C++11 lockfree single producer single consumer: how to avoid busy wait
body: <p>I'm trying to implement a class that uses two threads: one for the producer and one for the consumer. The current implementation does not use locks:</p>
  
  <pre><code>#include &lt;boost/lockfree/spsc_queue.hpp&gt;
  #include &lt;atomic&gt;
  #include &lt;thread&gt;
  
  using Queue =
          boost::lockfree::spsc_queue&lt;
              int,
              boost::lockfree::capacity&lt;1024&gt;&gt;;
  
  class Worker
  {
  public:
      Worker() : working_(false), done_(false) {}
      ~Worker() {
          done_ = true;    // exit even if the work has not been completed
          worker_.join();
      }
  
      void enqueue(int value) {
          queue_.push(value);
          if (!working_) {
              working_ = true;
              worker_ = std::thread([this]{ work(); });
          }
      }
  
      void work() {
          int value;
          while (!done_ &amp;&amp; queue_.pop(value)) {
              std::cout &lt;&lt; value &lt;&lt; std::endl;
          }
          working_ = false;
      }
  
  private:
      std::atomic&lt;bool&gt; working_;
      std::atomic&lt;bool&gt; done_;
      Queue queue_;
      std::thread worker_;
  };
  </code></pre>
  
  <p>The application needs to enqueue work items for a certain amount of time and then sleep waiting for an event. This is a minimal main that simulates the behavior:</p>
  
  <pre><code>int main()
  {
      Worker w;
      for (int i = 0; i &lt; 1000; ++i)
          w.enqueue(i);
      std::this_thread::sleep_for(std::chrono::seconds(1));
      for (int i = 0; i &lt; 1000; ++i)
          w.enqueue(i);
      std::this_thread::sleep_for(std::chrono::seconds(1));
  }
  </code></pre>
  
  <p>I'm pretty sure that my implementation is bugged: what if the worker thread completes and before executing <code>working_ = false</code>, another <code>enqueue</code> comes? Is it possible to make my code thread safe without using locks?</p>
  
  <p><strong>The solution requires:</strong></p>
  
  <ul>
  <li>a fast enqueue</li>
  <li>the destructor has to quit even if the queue is not empty</li>
  <li>no busy wait, because there are long period of time in which the worker thread is idle</li>
  <li>no locks if possible</li>
  </ul>
  
  <h1>Edit</h1>
  
  <p>I did another implementation of the <code>Worker</code> class, based on your suggestions. Here is my second attempt:</p>
  
  <pre><code>class Worker
  {
  public:
      Worker()
          : working_(ATOMIC_FLAG_INIT), done_(false) { } 
  
      ~Worker() {
          // exit even if the work has not been completed
          done_ = true;
          if (worker_.joinable())
              worker_.join();
      }
  
      bool enqueue(int value) {
          bool enqueued = queue_.push(value);
          if (!working_.test_and_set()) {
              if (worker_.joinable())
                  worker_.join();
              worker_ = std::thread([this]{ work(); });
          }
          return enqueued;
      }
  
      void work() {
          int value;
          while (!done_ &amp;&amp; queue_.pop(value)) {
              std::cout &lt;&lt; value &lt;&lt; std::endl;
          }
          working_.clear();
          while (!done_ &amp;&amp; queue_.pop(value)) {
              std::cout &lt;&lt; value &lt;&lt; std::endl;
          }
      }
  
  private:
      std::atomic_flag working_;
      std::atomic&lt;bool&gt; done_;
      Queue queue_;
      std::thread worker_;
  };
  </code></pre>
  
  <p>I introduced the <code>worker_.join()</code> inside the <code>enqueue</code> method. This can impact the performances, but in very rare cases (when the queue gets empty and before the thread exits, another <code>enqueue</code> comes). The <code>working_</code> variable is now an <code>atomic_flag</code> that is set in <code>enqueue</code> and cleared in <code>work</code>. The Additional <code>while</code> after <code>working_.clear()</code> is needed because if another value is pushed, before the <code>clear</code>, but after the <code>while</code>, the value is not processed.</p>
  
  <p>Is this implementation correct?</p>
  
  <p>I did some tests and the implementation seems to work.</p>
  
  <p>OT: Is it better to put this as an edit, or an answer?</p>