ConcurrentQueue.h 2.31 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
/*
 * Copyright 2015 Institut Laue-Langevin
 *
 * Licensed under the EUPL, Version 1.1 only (the "License");
 * You may not use this work except in compliance with the Licence.
 * You may obtain a copy of the Licence at:
 *
 * http://joinup.ec.europa.eu/software/page/eupl
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the Licence is distributed on an "AS IS" basis,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the Licence for the specific language governing permissions and
 * limitations under the Licence.
 */

#ifndef CAMEO_CONCURRENTQUEUE_H_
#define CAMEO_CONCURRENTQUEUE_H_

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

namespace cameo {

/**
 * Class implementing a concurrent queue. This is a modified version of the implementation:
 * https://juanchopanzacpp.wordpress.com/2013/02/26/concurrent-queue-c11/
 * Supports only pointer types.
 */
template<typename Type>
class ConcurrentQueue {

public:
	/**
	 * Destructor. Deletes all the items.
	 */
	~ConcurrentQueue() {

		std::unique_lock<std::mutex> lock(m_mutex);

		while (!m_queue.empty()) {
			delete m_queue.front();
			m_queue.pop();
		}
	}

	/**
	 * Gets the front item if the queue is not empty. Returns a null pointer otherwise.
	 */
	std::unique_ptr<Type> poll() {

		std::unique_lock<std::mutex> lock(m_mutex);

		if (m_queue.empty()) {
			return std::unique_ptr<Type>();
		}
		auto item = m_queue.front();
		m_queue.pop();

		return std::unique_ptr<Type>(item);
	}

	/**
	 * Gets the front item. Blocking call until there is an item.
	 */
	std::unique_ptr<Type> pop() {

		std::unique_lock<std::mutex> lock(m_mutex);

		while (m_queue.empty()) {
			m_condition.wait(lock);
		}
		auto item = m_queue.front();
		m_queue.pop();

		return std::unique_ptr<Type>(item);
	}

	/**
	 * Pushes an item.
	 */
	void push(std::unique_ptr<Type> & item) {

		std::unique_lock<std::mutex> lock(m_mutex);
		m_queue.push(item.release());
		lock.unlock();
		m_condition.notify_one();
	}

	/**
	 * Returns the size of the queue.
	 */
	typename std::queue<Type *>::size_type size() {

		std::unique_lock<std::mutex> lock(m_mutex);
		return m_queue.size();
	}

private:
	std::queue<Type *> m_queue;
	std::mutex m_mutex;
	std::condition_variable m_condition;
};

}

#endif