ThreadSafeQueue

A C++ thread-safe queue implementation.

Overview

ThreadSafeQueue is a template class that provides a thread-safe wrapper around std::queue. It uses mutex and condition variables to ensure safe concurrent access from multiple threads.

Features

• Thread-safe operations: All operations are protected by mutex locks
• Move semantics: Uses std::unique_ptr for efficient memory management
• Blocking and non-blocking operations: Flexible API for different use cases
• Timeout support: Timed wait operations with configurable timeouts
• No copy operations: Prevents resource conflicts by disabling copy constructor and assignment

Requirements

• C++17 or later
• Standard library headers: <queue>, <mutex>, <condition_variable>, <chrono>, <thread>

API Reference

Methods

• push(std::unique_ptr<T> value) - Add an element to the queue
• try_pop(std::unique_ptr<T>& value) - Non-blocking pop, returns false if empty
• wait_and_pop(std::unique_ptr<T>& value) - Blocking pop, waits for element
• try_wait_and_pop(std::unique_ptr<T>& value, duration timeout) - Pop with timeout
• empty() - Check if queue is empty
• size() - Get number of elements in queue

Usage Examples

Basic Usage

#include "ThreadSafeQueue.hpp"
#include <iostream>
#include <memory>

// Create a queue for integers
ThreadSafeQueue<int> queue;

// Push elements
queue.push(std::make_unique<int>(42));
queue.push(std::make_unique<int>(100));

// Pop element (non-blocking)
std::unique_ptr<int> value;
if (queue.try_pop(value)) {
    std::cout << "Popped: " << *value << std::endl;
}

Producer-Consumer Pattern

#include "ThreadSafeQueue.hpp"
#include <thread>
#include <iostream>

ThreadSafeQueue<int> queue;

// Producer thread
void producer() {
    for (int i = 0; i < 10; ++i) {
        queue.push(std::make_unique<int>(i));
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
}

// Consumer thread
void consumer() {
    for (int i = 0; i < 10; ++i) {
        std::unique_ptr<int> value;
        queue.wait_and_pop(value);
        std::cout << "Consumed: " << *value << std::endl;
    }
}

int main() {
    std::thread prod(producer);
    std::thread cons(consumer);
    
    prod.join();
    cons.join();
    
    return 0;
}

Using Timeout

#include "ThreadSafeQueue.hpp"
#include <chrono>

ThreadSafeQueue<std::string> queue;

std::unique_ptr<std::string> value;
auto timeout = std::chrono::seconds(5);

if (queue.try_wait_and_pop(value, timeout)) {
    std::cout << "Got value: " << *value << std::endl;
} else {
    std::cout << "Timeout expired, no value available" << std::endl;
}

Working with Custom Types

struct Message {
    int id;
    std::string content;
    
    Message(int i, std::string c) : id(i), content(std::move(c)) {}
};

ThreadSafeQueue<Message> messageQueue;

// Push custom type
messageQueue.push(std::make_unique<Message>(1, "Hello World"));

// Pop and process
std::unique_ptr<Message> msg;
if (messageQueue.try_pop(msg)) {
    std::cout << "Message " << msg->id << ": " << msg->content << std::endl;
}

Thread Safety Guarantees

• All public methods are thread-safe and can be called concurrently from multiple threads
• The queue uses RAII and lock guards to ensure exception safety
• Condition variables are used for efficient waiting without busy-polling

Performance Considerations

• Uses std::unique_ptr to avoid unnecessary copying of large objects
• notify_one() is used instead of notify_all() for better performance when multiple consumers are waiting
• Lock contention may occur under heavy concurrent access

License

Copyright 2025 TierOne Software

Licensed under the Apache License, Version 2.0. See the source file for full license text.