Concurrent Queue

Queue

FIFO data structure
Stores elements in the order they were inserted
- Elements are "pushed" onto the back of the queue
- Elements are "popped" off the front

`std::queue` Limitations

Implemented as a "memory location"
If pop() is called on an empty container, the behaviour is undefined
Removing an element involves two operations
- front() returns a reference to the element at the front
- pop() removes the element at the front, without returning anything
Intended to provide exception safety
- But removes thread safety
- Race condition between front() and pop()

Concurrent Queue

We will write a wrapper class
- std::queue object is class member
- std::mutex is class member
Same interfact as std::queue
- Will omit "uninteresting" member functions
Each member function locks the mutex
- Then calls the corresponding member function of std::queue
- Arguments will be forwarded

Concurrent Queue Special Member Functions

No requirement to copy or move objects of this class
We will delete the copy and move operators

Concurrent Queue Member Functions

push()
- Lock the mutex
- Call std::queue's push() with its argument
- Unlock the mutex
pop()
- Lock the mutex
- Call std::queue's front()
- Copy the returned value into its argument
- Call std::queue's pop()
- Unlock the mutex
- Do something sensible if the queue is empty

Makefile


CXX = g++
CFLAGS = -Wall

all: queue_main

queue_main: queue_main.o
	$(CXX) -o queue_main queue_main.o

queue_main.o: queue_main.cpp concurrent_queue.h
	$(CXX) $(CFLAGS) -c queue_main.cpp

`concurrent_queue.h`


#pragma once

#include <mutex>
#include <queue>
#include <stdexcept>

class concurrent_queue_empty : public std::runtime_error {
public:
  concurrent_queue_empty() : std::runtime_error("Queue is empty") {}
  concurrent_queue_empty(const char *s) : std::runtime_error(s) {}
};

class concurrent_queue_full : public std::runtime_error {
public:
  concurrent_queue_full() : std::runtime_error("Queue is full") {}
  concurrent_queue_full(const char *s) : std::runtime_error(s) {}
};

template <class T> class concurrent_queue {
  std::mutex mut;
  std::queue<T> que;
  size_t max{50};

public:
  concurrent_queue() = default;
  concurrent_queue(size_t max) : max(max) {};

  concurrent_queue(concurrent_queue &&) = delete;
  concurrent_queue(const concurrent_queue &) = delete;
  concurrent_queue &operator=(concurrent_queue &&) = delete;
  concurrent_queue &operator=(const concurrent_queue &) = delete;

  void push(T value) {
    std::lock_guard<std::mutex> lck_guard(mut);

    if (que.size() > max) {
      throw concurrent_queue_full();
    }

    que.push(value);
  }

  void pop(T &value) {
    std::unique_lock<std::mutex> lck_guard(mut);

    if (que.empty()) {
      throw concurrent_queue_empty();
    }

    value = que.front();
    que.pop();
  }
};

`queue_main.cpp`


#include "concurrent_queue.h"
#include <exception>
#include <future>
#include <iostream>
#include <string>
#include <thread>

// Shared queue object
concurrent_queue<std::string> conc_que;

// Waiting thread
void reader() {
  using namespace std::chrono_literals;
  std::this_thread::sleep_for(2s); // Pretend to be busy
  std::string sdata;

  // Pop some elements from the queue
  std::cout << "Reader calling pop...\n";
  for (int i = 0; i < 50; ++i) {
    conc_que.pop(sdata);
    std::cout << "Reader received data: " << sdata << '\n';
  }
}

// Modifying thread
void writer() {
  // Push the data onto the queue
  for (int i = 0; i < 50; ++i) {
    std::string sdata = "Item " + std::to_string(i);
    conc_que.push(sdata);
  }
  std::cout << "Writer returned from push...\n";
}

int main() {
  auto write_fut = std::async(std::launch::async, writer);
  auto read_fut = std::async(std::launch::async, reader);

  // Wait for them to complete
  try {
    read_fut.get();
  } catch (std::exception &e) {
    std::cout << "Exception caught: " << e.what() << '\n';
  }

  try {
    write_fut.get();
  } catch (std::exception &e) {
    std::cout << "Exception caught: " << e.what() << '\n';
  }
}

Debugging Grimoire