C++ Advanced · Lesson 08 of 23

Thread-safe queues

A blocking queue with a mutex and condition variable for passing work between threads.

Estimated time · 60–90 min

Learning goals

What you will understand

Explain thread-safe queues in your own words
Use it in a small robotics example
Verify the result instead of assuming it worked

Understand the idea first

The mental model

A thread-safe queue lets a producer thread push items and a consumer thread pop them, with the mutex hidden inside the class. A std::condition_variable lets the consumer sleep until an item arrives instead of burning CPU in a loop.

Always wait with a predicate (cv.wait(lock, [] { return !q.empty(); })) because threads can wake spuriously. Add a close() flag so consumers can finish cleanly at shutdown. ROS 2 executors use queues of ready callbacks the same way.

Guided lesson

Work through these steps

1.
Read the thread-safe queues example and identify what each line owns or changes.
2.
Create a small source file, compile with -std=c++17 and warnings enabled, and run it.
3.
Change one input, predict the result before running, then compare the output.
4.
Record one compiler or runtime error and explain the correction in your own words.

Worked example

Read it before you run it

template <typename T>
// A queue for any item type.
class SafeQueue {
// Thread-safe from the outside; the mutex is an implementation detail.
public:
// Public interface.
  void push(T v) {
// Called by producer threads.
    { std::lock_guard<std::mutex> lock{m_}; q_.push(std::move(v)); }
// Lock only long enough to add the item.
    cv_.notify_one();
// Wake one waiting consumer.
  }
// End of push.
  T pop() {
// Called by a consumer; blocks until an item exists.
    std::unique_lock<std::mutex> lock{m_};
// unique_lock is needed because wait() unlocks and relocks.
    cv_.wait(lock, [this] { return !q_.empty(); });
// Sleep until the queue is not empty (checked again after each wake-up).
    T v = std::move(q_.front()); q_.pop(); return v;
// Take the oldest item.
  }
// End of pop.
private:
// Hidden state.
  std::mutex m_; std::condition_variable cv_; std::queue<T> q_;
// The lock, the wake-up signal and the storage.
};
// End of the class.

What the example is doing

1.Read declarations from right to left: identify each value's type, name, and initial value.
2.Follow the data through each expression before considering syntax shortcuts.
3.Compile with warnings enabled; a clean build is part of the result, not an optional polish step.

Watch it run · Messages between nodes

See what this code does

The terminal types this lesson's practice solution and runs it. The scene shows the kind of robot behaviour that code drives. It plays by itself and loops.

Practice code

Now type it yourself

Copy the starter into practice.cpp and replace every TODO with your own code. Build and run it: g++ -std=c++17 -Wall -Wextra -Wpedantic practice.cpp -o practice -pthread && ./practice

#include <condition_variable>
#include <iostream>
#include <mutex>
#include <queue>
#include <thread>

template <typename T>
class SafeQueue {
 public:
  void push(T v) {
    {
      std::lock_guard<std::mutex> lock{m_};
      q_.push(std::move(v));
    }
    cv_.notify_one();
  }
  T pop() {
    // TODO: lock with unique_lock, wait until the queue is not empty, then return and remove the front item
  }

 private:
  std::mutex m_;
  std::condition_variable cv_;
  std::queue<T> q_;
};

int main() {
  SafeQueue<int> queue;
  std::thread producer{[&] {
    for (int i = 1; i <= 100; ++i) queue.push(i);
  }};
  int sum = 0;
  for (int i = 0; i < 100; ++i) sum += queue.pop();
  producer.join();
  std::cout << sum << '\n';
  return 0;
}

Expected output

Stuck? Show the full solution (try it yourself first)

#include <condition_variable>
#include <iostream>
#include <mutex>
#include <queue>
#include <thread>

template <typename T>
class SafeQueue {
 public:
  void push(T v) {
    {
      std::lock_guard<std::mutex> lock{m_};
      q_.push(std::move(v));
    }
    cv_.notify_one();
  }
  T pop() {
    std::unique_lock<std::mutex> lock{m_};
    cv_.wait(lock, [this] { return !q_.empty(); });
    T v = std::move(q_.front());
    q_.pop();
    return v;
  }

 private:
  std::mutex m_;
  std::condition_variable cv_;
  std::queue<T> q_;
};

int main() {
  SafeQueue<int> queue;
  std::thread producer{[&] {
    for (int i = 1; i <= 100; ++i) queue.push(i);
  }};
  int sum = 0;
  for (int i = 0; i < 100; ++i) sum += queue.pop();
  producer.join();
  std::cout << sum << '\n';
  return 0;
}

Expected result

How you know it worked

After adding the shown fragment to the guided exercise with its listed headers and surrounding program, the finished program builds with -Wall -Wextra -Wpedantic.
Changing the stated input produces the result you predicted.

Practical exercise

Apply the lesson

Use SafeQueue<int> with one producer thread pushing 1..100 and one consumer summing them (the result should be 5050). Add a close() method so the consumer exits, then run the program under ThreadSanitizer.

Before you continue

The program builds with -std=c++17 and no warnings
You can explain the key rule without reading the example
The normal and edge cases both have recorded results

If something goes wrong

The compiler reports that a name or type is unknown.

Start at the first error. Check spelling, required #include lines, and whether the declaration appears before use.

The program runs but the value is wrong.

Print the inputs and intermediate values, then check conversions, units, and collection order.