-
Notifications
You must be signed in to change notification settings - Fork 2
/
Copy pathbounded_buffer.hpp
143 lines (124 loc) · 3.25 KB
/
bounded_buffer.hpp
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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
#ifndef __BOUNDED_BUFFER_H__
#define __BOUNDED_BUFFER_H__
#include <deque>
#include <mutex>
#include <condition_variable>
#include <boost/circular_buffer.hpp>
template <typename T>
class BoundedBuffer {
public:
using base_container_type = boost::circular_buffer<T>;
using size_type = typename base_container_type::size_type;
BoundedBuffer(size_type max_capacity)
: buffer_(max_capacity)
{}
void clear() {
{
std::unique_lock<std::mutex> lock(mutex_);
buffer_.clear();
}
}
void push(T v) {
{
std::unique_lock<std::mutex> lock(mutex_);
cv_not_full_.wait(lock, [this]() { return this->buffer_.size() < buffer_.capacity(); });
buffer_.push_front(v);
}
cv_not_empty_.notify_one();
}
void try_push(T v) {
{
std::unique_lock<std::mutex> lock(mutex_);
if (buffer_.full()) {
return false;
}
buffer_.push_front(v);
}
cv_not_empty_.notify_one();
}
template <typename Duration>
bool try_push_for(T v, Duration&& d, T& item) {
{
std::unique_lock<std::mutex> lock(mutex_);
if(!cv_not_full_.wait_for(lock, d, [this]() { return this->buffer_.size() < buffer_.capacity(); })) {
return false;
}
buffer_.push_front(v);
}
cv_not_empty_.notify_one();
}
template <typename Timepoint>
bool try_push_untul(T v, Timepoint&& tp, T& item) {
{
std::unique_lock<std::mutex> lock(mutex_);
if(!cv_not_full_.wait_until(lock, tp, [this]() { return this->buffer_.size() < buffer_.capacity(); })) {
return false;
}
buffer_.push_front(v);
}
cv_not_empty_.notify_one();
}
T pop() {
std::unique_lock<std::mutex> lock(mutex_);
cv_not_empty_.wait(lock, [this]() { return this->buffer_.size() > 0; });
T v = buffer_.back();
buffer_.pop_back();
lock.unlock();
cv_not_full_.notify_one();
return v;
}
bool try_pop(T& item) {
std::unique_lock<std::mutex> lock(mutex_);
if (buffer_.empty()) {
return false;
}
item = buffer_.back();
buffer_.pop();
return true;
}
template <typename Duration>
bool try_pop_for(Duration&& d, T& item) {
std::unique_lock<std::mutex> lock(mutex_);
if (!cv_not_empty_.wait_for(lock, d, [this]() { return this->buffer_.size() > 0; })) {
return false;
}
item = buffer_.back();
buffer_.pop_back();
lock.unlock();
cv_not_full_.notify_one();
return true;
}
template <typename Timepoint>
bool try_pop_until(Timepoint&& tp, T& item) {
std::unique_lock<std::mutex> lock(mutex_);
if (!cv_not_empty_.wait_until(lock, tp, [this]() { return this->buffer_.size() > 0; })) {
return false;
}
item = buffer_.back();
buffer_.pop_back();
lock.unlock();
cv_not_full_.notify_one();
return true;
}
bool empty() const {
return buffer_.empty();
}
size_type size() {
return buffer_.size();
}
size_type capacity() {
return buffer_.capacity();
}
base_container_type get_queue() {
std::unique_lock<std::mutex> lock(mutex_);
return buffer_;
}
private:
// for locking
std::mutex mutex_;
std::condition_variable cv_not_empty_;
std::condition_variable cv_not_full_;
// queue
base_container_type buffer_;
} ;
#endif // __BOUNDED_BUFFER_H__