title: My thread-safe queue code appears to work, any possible race conditions, deadlocks, or other design problems?
body: <p>I am new to using <code>condition_variable</code>s and <code>unique_lock</code>s in C++. I am working on creating an event loop that polls two custom event-queues and a &quot;boolean&quot; (see integer acting as boolean), which can be acted upon by multiple sources.</p>
  <p>I have a demo (below) that appears to work, which I would greatly appreciate if you can review and confirm if it follows the best practices for using <code>unique_lock</code> and <code>condition_variable</code>s and any problems you foresee happening (race conditions, thread blocking, etc).</p>
  <ol>
  <li><p>In <code>ThreadSafeQueue::enqueue(...):</code> are we unlocking twice by calling notify and having the unique_lock go out of scope?</p>
  </li>
  <li><p>In the method <code>TheadSafeQueue::dequeueAll():</code> We assume it is being called by a method that has been notified (cond.notify), and therefore has been locked. Is there a better way to encapsulate this to keep the caller cleaner?</p>
  </li>
  <li><p>Do we need to make our class members volatile <a href="https://stackoverflow.com/questions/539824/problem-with-thread-safe-queue">similar to this</a>?</p>
  </li>
  <li><p>Is there a better way to mockup our situation that allows us to test if we've correctly implemented the locks? Perhaps without the sleep statements and automating the checking process?</p>
  </li>
  </ol>
  <p>ThreadSafeQueue.h:</p>
  <pre><code>#include &lt;condition_variable&gt;
  #include &lt;cstdint&gt;
  #include &lt;iostream&gt;
  #include &lt;mutex&gt;
  #include &lt;vector&gt;
  
  template &lt;class T&gt;
  class ThreadSafeQueue {
   public:
    ThreadSafeQueue(std::condition_variable* cond, std::mutex* unvrsl_m)
        : ThreadSafeQueue(cond, unvrsl_m, 1) {}
    ThreadSafeQueue(std::condition_variable* cond, std::mutex* unvrsl_m,
                    uint32_t capacity)
        : cond(cond),
          m(unvrsl_m),
          head(0),
          tail(0),
          capacity(capacity),
          buffer((T*)malloc(get_size() * sizeof(T))),
          scratch_space((T*)malloc(get_size() * sizeof(T))) {}
  
    std::condition_variable* cond;
  
    ~ThreadSafeQueue() {
      free(scratch_space);
      free(buffer);
    }
  
    void resize(uint32_t new_cap) {
      std::unique_lock&lt;std::mutex&gt; lock(*m);
      check_params_resize(new_cap);
  
      free(scratch_space);
      scratch_space = buffer;
      buffer = (T*)malloc(sizeof(T) * new_cap);
      copy_cyclical_queue();
      free(scratch_space);
      scratch_space = (T*)malloc(new_cap * sizeof(T));
  
      tail = get_size();
      head = 0;
      capacity = new_cap;
    }
    void enqueue(const T&amp; value) {
      std::unique_lock&lt;std::mutex&gt; lock(*m);
      resize();
      buffer[tail++] = value;
      if (tail == get_capacity()) {
        tail = 0;
      } else if (tail &gt; get_capacity())
        throw(&quot;Something went horribly wrong TSQ: 75&quot;);
      cond-&gt;notify_one();
    }
  
  // Assuming m has already been locked by the caller...
    void dequeueAll(std::vector&lt;T&gt;* vOut) {
      if (get_size() == 0) return;
      scratch_space = buffer;
      copy_cyclical_queue();
      vOut-&gt;insert(vOut-&gt;end(), buffer, buffer + get_size());
      head = tail = 0;
    }
  
    // Const functions because they shouldn't be modifying the internal variables
    // of the object
    bool is_empty() const { return get_size() == 0; }
    uint32_t get_size() const {
      if (head == tail)
        return 0;
      else if (head &lt; tail) {
        // 1 2 3
        // 0 1 2
        // 1
        // 0
        return tail - head;
      } else {
        // 3 _ 1 2
        // 0 1 2 3
        // capacity-head + tail+1 = 4-2+0+1 = 2 + 1
        return get_capacity() - head + tail + 1;
      }
    }
    uint32_t get_capacity() const { return capacity; }
    //---------------------------------------------------------------------------
   private:
    std::mutex* m;
    uint32_t head;
    uint32_t tail;
    uint32_t capacity;
    T* buffer;
    T* scratch_space;
    uint32_t get_next_empty_spot();
    void copy_cyclical_queue() {
      uint32_t size = get_size();
      uint32_t cap = get_capacity();
      if (size == 0) {
        return;  // because we have nothing to copy
      }
      if (head + size &lt;= cap) {
        // _ 1 2 3 ... index = 1, size = 3, 1+3 = 4 = capacity... only need 1 copy
        memcpy(buffer, scratch_space + head, sizeof(T) * size);
      } else {
        // 5 1 2 3 4 ... index = 1, size = 5, 1+5 = 6 = capacity... need to copy
        // 1-4 then 0-1
  
        // copy number of bytes: front = 1, to (5-1 = 4 elements)
        memcpy(buffer, scratch_space + head, sizeof(T) * (cap - head));
        // just copy the bytes from the front up to the first element in the old
        // array
        memcpy(buffer + (cap - head), scratch_space, sizeof(T) * tail);
      }
    }
    void check_params_resize(uint32_t new_cap) {
      if (new_cap &lt; get_size()) {
        std::cerr &lt;&lt; &quot;ThreadSafeQueue: check_params_resize: size(&quot; &lt;&lt; get_size()
                  &lt;&lt; &quot;) &gt; new_cap(&quot; &lt;&lt; new_cap
                  &lt;&lt; &quot;)... data &quot;
                     &quot;loss will occur if this happens. Prevented.&quot;
                  &lt;&lt; std::endl;
      }
    }
    void resize() {
      uint32_t new_cap;
      uint32_t size = get_size();
      uint32_t cap = get_capacity();
      if (size + 1 &gt;= cap - 1) {
        std::cout &lt;&lt; &quot;RESIZE CALLED --- BAD&quot; &lt;&lt; std::endl;
        new_cap = 2 * cap;
  
        check_params_resize(new_cap);
  
        free(scratch_space);     // free existing (too small) scratch space
        scratch_space = buffer;  // transfer pointer over
        buffer = (T*)malloc(sizeof(T) * new_cap);  // allocate a bigger buffer
        copy_cyclical_queue();
        // move over everything with memcpy from scratch_space to buffer
        free(scratch_space);  // free what used to be the too-small buffer
        scratch_space =
            (T*)malloc(sizeof(T) * new_cap);  // recreate scratch space
  
        tail = size;
        head = 0;
        // since we're done with the old array... delete for memory management-&gt;
  
        capacity = new_cap;
      }
    }
  };
  // Event Types
  // keyboard/mouse
  // network
  // dirty flag
  </code></pre>
  <p>Main.cpp:</p>
  <pre><code>
  #include &lt;unistd.h&gt;
  
  #include &lt;cstdint&gt;
  #include &lt;iostream&gt;
  #include &lt;mutex&gt;
  #include &lt;queue&gt;
  #include &lt;sstream&gt;
  #include &lt;thread&gt;
  
  #include &quot;ThreadSafeQueue.h&quot;
  using namespace std;
  
  void write_to_threadsafe_queue(ThreadSafeQueue&lt;uint32_t&gt; *q,
                                 uint32_t startVal) {
    uint32_t count = startVal;
    while (true) {
      q-&gt;enqueue(count);
      cout &lt;&lt; &quot;Successfully enqueued: &quot; &lt;&lt; count &lt;&lt; endl;
      count += 2;
      sleep(count);
    }
  }
  
  void sleep_and_set_redraw(int *redraw, condition_variable *cond) {
    while (true) {
      sleep(3);
      __sync_fetch_and_or(redraw, 1);
      cond-&gt;notify_one();
    }
  }
  
  void process_events(vector&lt;uint32_t&gt; *qOut, condition_variable *cond,
                      ThreadSafeQueue&lt;uint32_t&gt; *q1,
                      ThreadSafeQueue&lt;uint32_t&gt; *q2, int *redraw, mutex *m) {
    while (true) {
      unique_lock&lt;mutex&gt; lck(*m);
      cond-&gt;wait(lck);
      q1-&gt;dequeueAll(qOut);
      q2-&gt;dequeueAll(qOut);
      if (__sync_fetch_and_and(redraw, 0)) {
        cout &lt;&lt; &quot;FLAG SET&quot; &lt;&lt; endl;
        qOut-&gt;push_back(0);
      }
      for (auto a : *qOut) cout &lt;&lt; a &lt;&lt; &quot;\t&quot;;
      cout &lt;&lt; endl;
      cout &lt;&lt; &quot;PROCESSING: &quot; &lt;&lt; qOut-&gt;size() &lt;&lt; endl;
      qOut-&gt;clear();
    }
  }
  
  void test_2_queues_and_bool() {
    try {
      condition_variable cond;
      mutex m;
      ThreadSafeQueue&lt;uint32_t&gt; q1(&amp;cond, &amp;m, 1024);
      ThreadSafeQueue&lt;uint32_t&gt; q2(&amp;cond, &amp;m, 1024);
      int redraw = 0;
      vector&lt;uint32_t&gt; qOut;
      thread t1(write_to_threadsafe_queue, &amp;q1, 2);
      thread t2(write_to_threadsafe_queue, &amp;q2, 1);
      thread t3(sleep_and_set_redraw, &amp;redraw, &amp;cond);
      thread t4(process_events, &amp;qOut, &amp;cond, &amp;q1, &amp;q2, &amp;redraw, &amp;m);
      t1.join();
      t2.join();
      t3.join();
      t4.join();
    } catch (system_error &amp;e) {
      cout &lt;&lt; &quot;MAIN TEST CRASHED&quot; &lt;&lt; e.what();
    }
  }
  
  int main() { test_2_queues_and_bool(); }
  </code></pre>