ClickHouse/dbms/include/DB/Common/ConcurrentBoundedQueue.h

123 lines
2.5 KiB
C++
Raw Normal View History

2013-05-03 02:25:50 +00:00
#pragma once
#include <queue>
#include <Poco/Mutex.h>
#include <Poco/Semaphore.h>
#include <DB/Core/Types.h>
2013-05-03 02:25:50 +00:00
/** Очень простая thread-safe очередь ограниченной длины.
* Если пытаться вынуть элемент из пустой очереди, то поток блокируется, пока очередь не станет непустой.
* Если пытаться вставить элемент в переполненную очередь, то поток блокируется, пока в очереди не появится элемент.
*/
template <typename T>
class ConcurrentBoundedQueue
{
private:
size_t max_fill;
std::queue<T> queue;
Poco::FastMutex mutex;
2013-05-03 02:25:50 +00:00
Poco::Semaphore fill_count;
Poco::Semaphore empty_count;
public:
ConcurrentBoundedQueue(size_t max_fill)
: fill_count(0, max_fill), empty_count(max_fill, max_fill) {}
void push(const T & x)
{
empty_count.wait();
{
Poco::ScopedLock<Poco::FastMutex> lock(mutex);
2013-05-03 02:25:50 +00:00
queue.push(x);
}
fill_count.set();
}
2016-04-22 19:57:40 +00:00
template <class ... Args>
void emplace(Args && ... args)
{
empty_count.wait();
{
Poco::ScopedLock<Poco::Mutex> lock(mutex);
queue.emplace(std::forward<Args>(args)...);
}
fill_count.set();
}
2013-05-03 02:25:50 +00:00
void pop(T & x)
{
fill_count.wait();
{
Poco::ScopedLock<Poco::FastMutex> lock(mutex);
x = std::move(queue.front());
2013-05-03 02:25:50 +00:00
queue.pop();
}
empty_count.set();
}
bool tryPush(const T & x, DB::UInt64 milliseconds = 0)
2013-05-03 02:25:50 +00:00
{
if (empty_count.tryWait(milliseconds))
2013-05-03 02:25:50 +00:00
{
{
Poco::ScopedLock<Poco::FastMutex> lock(mutex);
2013-05-03 02:25:50 +00:00
queue.push(x);
}
fill_count.set();
return true;
}
return false;
}
template <class ... Args>
bool tryEmplace(DB::UInt64 milliseconds, Args && ... args)
{
if (empty_count.tryWait(milliseconds))
{
{
Poco::ScopedLock<Poco::Mutex> lock(mutex);
queue.emplace(std::forward<Args>(args)...);
}
fill_count.set();
return true;
}
return false;
}
bool tryPop(T & x, DB::UInt64 milliseconds = 0)
2013-05-03 02:25:50 +00:00
{
if (fill_count.tryWait(milliseconds))
2013-05-03 02:25:50 +00:00
{
{
Poco::ScopedLock<Poco::FastMutex> lock(mutex);
x = std::move(queue.front());
2013-05-03 02:25:50 +00:00
queue.pop();
}
empty_count.set();
return true;
}
return false;
}
size_t size()
{
Poco::ScopedLock<Poco::FastMutex> lock(mutex);
return queue.size();
}
2013-05-03 02:25:50 +00:00
void clear()
{
while (fill_count.tryWait(0))
{
{
Poco::ScopedLock<Poco::FastMutex> lock(mutex);
2013-05-03 02:25:50 +00:00
queue.pop();
}
empty_count.set();
}
}
};