i2pd/Queue.h

127 lines
2.3 KiB
C
Raw Normal View History

2013-12-10 13:00:13 +00:00
#ifndef QUEUE_H__
#define QUEUE_H__
#include <queue>
#include <mutex>
#include <thread>
#include <condition_variable>
namespace i2p
{
namespace util
{
template<typename Element>
class Queue
{
public:
void Put (Element * e)
{
std::unique_lock<std::mutex> l(m_QueueMutex);
m_Queue.push (e);
m_NonEmpty.notify_one ();
}
Element * GetNext ()
{
std::unique_lock<std::mutex> l(m_QueueMutex);
Element * el = GetNonThreadSafe ();
if (!el)
{
m_NonEmpty.wait (l);
el = GetNonThreadSafe ();
}
return el;
}
Element * GetNextWithTimeout (int usec)
{
std::unique_lock<std::mutex> l(m_QueueMutex);
Element * el = GetNonThreadSafe ();
if (!el)
{
m_NonEmpty.wait_for (l, std::chrono::milliseconds (usec));
el = GetNonThreadSafe ();
}
return el;
}
2014-01-11 01:21:38 +00:00
bool Wait (int sec, int usec)
{
std::unique_lock<std::mutex> l(m_QueueMutex);
return m_NonEmpty.wait_for (l, std::chrono::seconds (sec) + std::chrono::milliseconds (usec)) != std::cv_status::timeout;
}
bool IsEmpty ()
{
std::unique_lock<std::mutex> l(m_QueueMutex);
return m_Queue.empty ();
}
2013-12-10 13:00:13 +00:00
void WakeUp () { m_NonEmpty.notify_one (); };
Element * Get ()
{
std::unique_lock<std::mutex> l(m_QueueMutex);
return GetNonThreadSafe ();
}
2014-01-11 01:21:38 +00:00
Element * Peek ()
{
std::unique_lock<std::mutex> l(m_QueueMutex);
return GetNonThreadSafe (true);
}
2013-12-10 13:00:13 +00:00
private:
2014-01-11 01:21:38 +00:00
Element * GetNonThreadSafe (bool peek = false)
2013-12-10 13:00:13 +00:00
{
if (!m_Queue.empty ())
{
Element * el = m_Queue.front ();
2014-01-11 01:21:38 +00:00
if (!peek)
m_Queue.pop ();
2013-12-10 13:00:13 +00:00
return el;
}
return nullptr;
}
private:
std::queue<Element *> m_Queue;
std::mutex m_QueueMutex;
std::condition_variable m_NonEmpty;
};
template<class Msg>
class MsgQueue: public Queue<Msg>
{
public:
2014-02-08 02:15:08 +00:00
MsgQueue (): m_Thread (std::bind (&MsgQueue<Msg>::Run, this)) , running(1) {};
2014-02-04 23:15:42 +00:00
void Stop()
{
2014-02-08 02:15:08 +00:00
running = 0;
m_Thread.join();
2014-02-04 23:15:42 +00:00
}
2013-12-10 13:00:13 +00:00
private:
void Run ()
{
Msg * msg = nullptr;
2014-02-08 02:15:08 +00:00
while ((msg = Queue<Msg>::GetNext ()) != nullptr && running)
2013-12-10 13:00:13 +00:00
{
msg->Process ();
delete msg;
}
}
private:
std::thread m_Thread;
2014-02-08 21:06:01 +00:00
volatile int running;
2013-12-10 13:00:13 +00:00
};
}
}
#endif