I implemented, somewhat closely based on this implementation, a blocking queue with emphasis on filling up first. So if one producer and one consumer thread are using the queue, the producing queue gets prioritized.
// BQueue.hpp
//#pragma once
#ifndef _BQUEUE_HPP
#define _BQUEUE_HPP
#include <condition_variable>
#include <mutex>
#include <queue>
template <class T> class BQueue {
public:
BQueue(size_t size);
void push(T item);
bool pop(T &item);
/* ALTERNATIVE 1
void push(std::unique_ptr<T> item);
bool pop(std::unique_ptr<T> &item);
*/
private:
std::mutex _mutex;
//std::queue<std::unique_ptr<T>> _queue; // ALTERNATIVE 1
std::queue<T> _queue;
size_t _size;
std::condition_variable _condition_full;
std::condition_variable _condition_empty;
};
#include "BQueue.hxx"
#endif // _BQUEUE_HPP
implementation
// BQueue.hxx
#include "BQueue.hpp"
#include <condition_variable>
#include <cstdlib>
#include <iostream>
#include <mutex>
#include <queue>
template <class T> BQueue<T>::BQueue(size_t size) : _size(size) {}
// ALTERNATIVE 1
// template <class T> void BQueue<T>::push(std::unique_ptr<T> item) {
template <class T> void BQueue<T>::push(T item) {
std::unique_lock<std::mutex> lock(_mutex);
while (_queue.size() >= _size) {
_condition_full.wait(lock, [&]() { return (_queue.size() < _size); });
}
_queue.push(std::move(item));
// _queue.push(item); // ALTERNATIVE 1
// if queue is full, notify consumation part first
if (_queue.size() >= _size) {
_condition_empty.notify_one();
}
_condition_full.notify_one();
}
// template <class T> bool BQueue<T>::pop(T &item) {
template <class T> bool BQueue<T>::pop(T &item) {
std::unique_lock<std::mutex> lock(_mutex);
while (_queue.empty()) {
if (!_condition_empty.wait_for(lock, std::chrono::seconds(1),
[&]() { return !_queue.empty(); })) {
// waited too long for input.
return false;
}
}
item = std::move(_queue.front());
// item = _queue.fron(); // ALTERNATIVE
_queue.pop();
/* THIS FOLLOWING CODE MAY BE NEEDED;
* KEEP IT IN CASE
* _condition_empty.notify_one();
* // if queue is empty, notify production
* if (_queue.empty()) { //*/
_condition_full.notify_one();
// assert(_queue.size() < _size); //*/
return true;
}
main
// alternatively: class Resource
struct Resource {
int x;
};
int main() {
BQueue<std::unique_ptr<struct Resource>> q{40};
//BQueue<struct Resource> q_alternative1[40};
std::unique_ptr<struct Resource> res1{new struct Resource};
res1->x = 42;
q.push(std::move(res1));
q.push(std::move(std::unique_ptr<struct Resource>{new struct Resource}));
for (size_t i = 0; i < 30; i++) {
std::unique_ptr<struct Resource> res{new struct Resource};
res->x = i;
q.push(std::move(res));
}
for (size_t i = 0; i < 15; i++)
q.pop(res1);
return 0;
}
Now, I came across the question of resource handling. What does make more sense? Using the shown code or implement the whole thing using std::unique_ptr where necessary (see ALTERNATIVE 1)? std::move even works with non-resource wrapped types, apparently.
And, is the main function leak-free or did I miss anything? Does it work for shared_ptr types as well, or do I have to specialize because of pop(T item &) being a reference?
