Loading...
Searching...
No Matches
thread_queue.h
Go to the documentation of this file.
/*******************************************************************************
* Copyright (c) 2017-2022 Frank Pagliughi <fpagliughi@mindspring.com>
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
* and Eclipse Distribution License v1.0 which accompany this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v20.html
* and the Eclipse Distribution License is available at
* http://www.eclipse.org/org/documents/edl-v10.php.
*
* Contributors:
* Frank Pagliughi - initial implementation and documentation
*******************************************************************************/
#ifndef __mqtt_thread_queue_h
#define __mqtt_thread_queue_h
#include <algorithm>
#include <condition_variable>
#include <deque>
#include <limits>
#include <mutex>
#include <queue>
#include <thread>
namespace mqtt {
class queue_closed : public std::runtime_error
{
public:
queue_closed() : std::runtime_error("queue is closed") {}
};
template <typename T, class Container = std::deque<T>>
class thread_queue
{
public:
using container_type = Container;
using value_type = T;
using size_type = typename Container::size_type;
static constexpr size_type MAX_CAPACITY = std::numeric_limits<size_type>::max();
private:
mutable std::mutex lock_;
std::condition_variable notEmptyCond_;
std::condition_variable notFullCond_;
size_type cap_{MAX_CAPACITY};
bool closed_{false};
std::queue<T, Container> que_;
using guard = std::lock_guard<std::mutex>;
using unique_guard = std::unique_lock<std::mutex>;
bool is_done() const { return closed_ && que_.empty(); }
public:
thread_queue() {}
explicit thread_queue(size_t cap) : cap_(std::max<size_type>(cap, 1)) {}
bool empty() const {
guard g{lock_};
return que_.empty();
}
size_type capacity() const {
guard g{lock_};
return cap_;
}
void capacity(size_type cap) {
guard g{lock_};
cap_ = std::max<size_type>(cap, 1);
if (cap_ > que_.size())
notFullCond_.notify_all();
}
size_type size() const {
guard g{lock_};
return que_.size();
}
void close() {
guard g{lock_};
closed_ = true;
notFullCond_.notify_all();
notEmptyCond_.notify_all();
}
bool closed() const {
guard g{lock_};
return closed_;
}
bool done() const {
guard g{lock_};
return is_done();
}
void clear() {
guard g{lock_};
while (!que_.empty()) que_.pop();
notFullCond_.notify_all();
}
void put(value_type val) {
unique_guard g{lock_};
notFullCond_.wait(g, [this] { return que_.size() < cap_ || closed_; });
if (closed_)
throw queue_closed{};
que_.emplace(std::move(val));
notEmptyCond_.notify_one();
}
bool try_put(value_type val) {
guard g{lock_};
if (que_.size() >= cap_ || closed_)
return false;
que_.emplace(std::move(val));
notEmptyCond_.notify_one();
return true;
}
template <typename Rep, class Period>
bool try_put_for(value_type val, const std::chrono::duration<Rep, Period>& relTime) {
unique_guard g{lock_};
bool to = !notFullCond_.wait_for(g, relTime, [this] {
return que_.size() < cap_ || closed_;
});
if (to || closed_)
return false;
que_.emplace(std::move(val));
notEmptyCond_.notify_one();
return true;
}
template <class Clock, class Duration>
bool try_put_until(
value_type val, const std::chrono::time_point<Clock, Duration>& absTime
) {
unique_guard g{lock_};
bool to = !notFullCond_.wait_until(g, absTime, [this] {
return que_.size() < cap_ || closed_;
});
if (to || closed_)
return false;
que_.emplace(std::move(val));
notEmptyCond_.notify_one();
return true;
}
bool get(value_type* val) {
if (!val)
return false;
unique_guard g{lock_};
notEmptyCond_.wait(g, [this] { return !que_.empty() || closed_; });
if (que_.empty()) // We must be done
return false;
*val = std::move(que_.front());
que_.pop();
notFullCond_.notify_one();
return true;
}
value_type get() {
unique_guard g{lock_};
notEmptyCond_.wait(g, [this] { return !que_.empty() || closed_; });
if (que_.empty()) // We must be done
throw queue_closed{};
value_type val = std::move(que_.front());
que_.pop();
notFullCond_.notify_one();
return val;
}
bool try_get(value_type* val) {
if (!val)
return false;
guard g{lock_};
if (que_.empty())
return false;
*val = std::move(que_.front());
que_.pop();
notFullCond_.notify_one();
return true;
}
template <typename Rep, class Period>
bool try_get_for(value_type* val, const std::chrono::duration<Rep, Period>& relTime) {
if (!val)
return false;
unique_guard g{lock_};
notEmptyCond_.wait_for(g, relTime, [this] { return !que_.empty() || closed_; });
if (que_.empty())
return false;
*val = std::move(que_.front());
que_.pop();
notFullCond_.notify_one();
return true;
}
template <class Clock, class Duration>
bool try_get_until(
value_type* val, const std::chrono::time_point<Clock, Duration>& absTime
) {
if (!val)
return false;
unique_guard g{lock_};
notEmptyCond_.wait_until(g, absTime, [this] { return !que_.empty() || closed_; });
if (que_.empty())
return false;
*val = std::move(que_.front());
que_.pop();
notFullCond_.notify_one();
return true;
}
};
} // namespace mqtt
#endif // __mqtt_thread_queue_h
mqtt::queue_closed
Definition thread_queue.h:43
mqtt::queue_closed::queue_closed
queue_closed()
Definition thread_queue.h:45
mqtt::thread_queue::size_type
typename Container::size_type size_type
Definition thread_queue.h:93
mqtt::thread_queue::value_type
T value_type
Definition thread_queue.h:91
mqtt::thread_queue::size
size_type size() const
Definition thread_queue.h:167
mqtt::thread_queue::capacity
void capacity(size_type cap)
Definition thread_queue.h:157
mqtt::thread_queue::done
bool done() const
Definition thread_queue.h:200
mqtt::thread_queue::container_type
Container container_type
Definition thread_queue.h:89
mqtt::thread_queue::try_put
bool try_put(value_type val)
Definition thread_queue.h:234
mqtt::thread_queue::try_get_for
bool try_get_for(value_type *val, const std::chrono::duration< Rep, Period > &relTime)
Definition thread_queue.h:360
mqtt::thread_queue::try_put_for
bool try_put_for(value_type val, const std::chrono::duration< Rep, Period > &relTime)
Definition thread_queue.h:253
mqtt::thread_queue::MAX_CAPACITY
static constexpr size_type MAX_CAPACITY
Definition thread_queue.h:96
mqtt::thread_queue::try_get
bool try_get(value_type *val)
Definition thread_queue.h:336
mqtt::thread_queue::thread_queue
thread_queue()
Definition thread_queue.h:126
mqtt::thread_queue::try_put_until
bool try_put_until(value_type val, const std::chrono::time_point< Clock, Duration > &absTime)
Definition thread_queue.h:276
mqtt::thread_queue::clear
void clear()
Definition thread_queue.h:208
mqtt::thread_queue::closed
bool closed() const
Definition thread_queue.h:190
mqtt::thread_queue::get
bool get(value_type *val)
Definition thread_queue.h:297
mqtt::thread_queue::empty
bool empty() const
Definition thread_queue.h:139
mqtt::thread_queue::close
void close()
Definition thread_queue.h:177
mqtt::thread_queue::capacity
size_type capacity() const
Definition thread_queue.h:147
mqtt::thread_queue::thread_queue
thread_queue(size_t cap)
Definition thread_queue.h:133
mqtt::thread_queue::put
void put(value_type val)
Definition thread_queue.h:219
mqtt::thread_queue::get
value_type get()
Definition thread_queue.h:317
mqtt::thread_queue::try_get_until
bool try_get_until(value_type *val, const std::chrono::time_point< Clock, Duration > &absTime)
Definition thread_queue.h:386
mqtt
Definition async_client.h:60