124 lines
3.1 KiB
C++
124 lines
3.1 KiB
C++
// SPDX-FileCopyrightText: 2024 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
|
|
// SPDX-License-Identifier: GPL-3.0-only
|
|
|
|
#pragma once
|
|
|
|
#include <queue>
|
|
#include <mutex>
|
|
#include <condition_variable>
|
|
#include <set>
|
|
|
|
template <class T>
|
|
class ThreadSafeFIFO {
|
|
std::queue<T> queue;
|
|
std::condition_variable c_empty, c_full;
|
|
mutable std::mutex m;
|
|
const size_t max_size;
|
|
size_t max_utilization;
|
|
size_t utilization;
|
|
public:
|
|
explicit ThreadSafeFIFO(size_t in_max_size = UINT32_MAX) : max_size(in_max_size), max_utilization(0), utilization(0) {}
|
|
|
|
void Clear() {
|
|
std::unique_lock ul(m);
|
|
queue = {};
|
|
utilization = 0;
|
|
max_utilization = 0;
|
|
}
|
|
|
|
bool Put(T val) {
|
|
std::unique_lock ul(m);
|
|
if (queue.size() < max_size) {
|
|
queue.push(val);
|
|
c_empty.notify_one();
|
|
utilization++;
|
|
if (utilization > max_utilization)
|
|
max_utilization = utilization;
|
|
return true;
|
|
} else
|
|
return false;
|
|
};
|
|
|
|
void PutBlocking(T val) {
|
|
std::unique_lock ul(m);
|
|
c_full.wait(ul, [&]{return queue.size() < max_size;});
|
|
queue.push(val);
|
|
utilization++;
|
|
if (utilization > max_utilization)
|
|
max_utilization = utilization;
|
|
c_empty.notify_one();
|
|
};
|
|
|
|
bool PutTimeout(T val, std::chrono::milliseconds timeout) {
|
|
std::unique_lock ul(m);
|
|
if (!c_full.wait_for(ul, timeout, [&]{ return queue.size() < max_size; }))
|
|
return false;
|
|
queue.push(val);
|
|
utilization++;
|
|
if (utilization > max_utilization)
|
|
max_utilization = utilization;
|
|
c_empty.notify_one();
|
|
return true;
|
|
}
|
|
|
|
int Get(T &val) {
|
|
std::unique_lock ul(m);
|
|
|
|
if (queue.empty())
|
|
return 0;
|
|
else {
|
|
val = queue.front();
|
|
queue.pop();
|
|
c_full.notify_one();
|
|
utilization--;
|
|
return 1;
|
|
}
|
|
}
|
|
|
|
T GetBlocking() {
|
|
std::unique_lock ul(m);
|
|
c_empty.wait(ul, [&]{return !queue.empty();});
|
|
T tmp = queue.front();
|
|
queue.pop();
|
|
c_full.notify_one();
|
|
utilization--;
|
|
return tmp;
|
|
};
|
|
|
|
int GetTimeout(T &val, std::chrono::microseconds timeout) {
|
|
std::unique_lock ul(m);
|
|
if (queue.empty())
|
|
c_empty.wait_for(ul, timeout, [&]{return !queue.empty();});
|
|
|
|
if (queue.empty())
|
|
return 0;
|
|
else {
|
|
val = queue.front();
|
|
queue.pop();
|
|
c_full.notify_one();
|
|
utilization--;
|
|
return 1;
|
|
}
|
|
}
|
|
|
|
[[nodiscard]] size_t Size() const {
|
|
std::unique_lock ul(m);
|
|
return queue.size();
|
|
}
|
|
|
|
void ClearMaxUtilization() {
|
|
std::unique_lock ul(m);
|
|
max_utilization = utilization;
|
|
}
|
|
|
|
[[nodiscard]] size_t GetMaxUtilization() const {
|
|
std::unique_lock ul(m);
|
|
return max_utilization;
|
|
}
|
|
|
|
[[nodiscard]] size_t GetCurrentUtilization() const {
|
|
std::unique_lock ul(m);
|
|
return utilization;
|
|
}
|
|
};
|