lokinet/llarp/queue.hpp

534 lines
12 KiB
C++

#ifndef LLARP_QUEUE_HPP
#define LLARP_QUEUE_HPP
#include <llarp/threading.hpp>
#include <queue_manager.hpp>
#include <atomic>
#include <optional>
#include <tuple>
namespace llarp
{
namespace thread
{
template < typename Type >
class QueuePushGuard;
template < typename Type >
class QueuePopGuard;
template < typename Type >
class Queue
{
// This class provides a thread-safe, lock-free, fixed-size queue.
public:
static constexpr size_t Alignment = 64;
private:
Type *m_data;
const char m_dataPadding[Alignment - sizeof(Type *)];
QueueManager m_manager;
std::atomic<std::uint32_t> m_waitingPoppers;
util::Semaphore m_popSemaphore;
const char
m_popSemaphorePadding[(2u * Alignment) - sizeof(util::Semaphore)];
std::atomic<std::uint32_t> m_waitingPushers;
util::Semaphore m_pushSemaphore;
const char
m_pushSemaphorePadding[(2u * Alignment) - sizeof(util::Semaphore)];
friend QueuePopGuard< Type >;
friend QueuePushGuard< Type >;
public:
explicit Queue(size_t capacity);
~Queue();
Queue(const Queue &) = delete;
Queue &
operator=(const Queue &) = delete;
// Push back to the queue, blocking until space is available (if
// required). Will fail if the queue is disabled (or becomes disabled
// while waiting for space on the queue).
QueueReturn
pushBack(const Type &value);
QueueReturn
pushBack(Type &&value);
// Try to push back to the queue. Return false if the queue is full or
// disabled.
QueueReturn
tryPushBack(const Type &value);
QueueReturn
tryPushBack(Type &&value);
// Remove an element from the queue. Block until an element is available
Type
popFront();
std::optional< Type >
tryPopFront();
// Remove all elements from the queue. Note this is not atomic, and if
// other threads `pushBack` onto the queue during this call, the `size` of
// the queue is not guaranteed to be 0.
void
removeAll();
// Disable the queue. All push operations will fail "fast" (including
// blocked operations). Calling this method on a disabled queue has no
// effect.
void
disable();
// Enable the queue. Calling this method on a disabled queue has no
// effect.
void
enable();
size_t
capacity() const;
size_t
size() const;
bool
enabled() const;
bool
full() const;
bool
empty() const;
};
// Provide a guard class to provide exception safety for pushing to a queue.
// On destruction, unless the `release` method has been called, will remove
// and destroy all elements from the queue, putting the queue into an empty
// state.
template < typename Type >
class QueuePushGuard
{
private:
Queue< Type > *m_queue;
uint32_t m_generation;
uint32_t m_index;
public:
QueuePushGuard(Queue< Type > &queue, uint32_t generation, uint32_t index)
: m_queue(&queue), m_generation(generation), m_index(index)
{
}
~QueuePushGuard();
void
release();
};
// Provide a guard class to provide exception safety for popping from a
// queue. On destruction, this will pop the the given element from the
// queue.
template < typename Type >
class QueuePopGuard
{
private:
Queue< Type > &m_queue;
uint32_t m_generation;
uint32_t m_index;
public:
QueuePopGuard(Queue< Type > &queue, uint32_t generation, uint32_t index)
: m_queue(queue), m_generation(generation), m_index(index)
{
}
~QueuePopGuard();
};
template < typename Type >
Queue< Type >::Queue(size_t capacity)
: m_data(nullptr)
, m_dataPadding()
, m_manager(capacity)
, m_waitingPoppers(0)
, m_popSemaphore(0)
, m_popSemaphorePadding()
, m_waitingPushers(0)
, m_pushSemaphore(0)
, m_pushSemaphorePadding()
{
m_data = static_cast< Type * >(::operator new(capacity * sizeof(Type)));
}
template < typename Type >
Queue< Type >::~Queue()
{
removeAll();
// We have already deleted the queue members above, free as (void *)
::operator delete(static_cast< void * >(m_data));
}
template < typename Type >
QueueReturn
Queue< Type >::tryPushBack(const Type &value)
{
uint32_t generation = 0;
uint32_t index = 0;
// Sync point A
//
// The next call writes with full sequential consistency to the push
// index, which guarantees that the relaxed read to the waiting poppers
// count sees any waiting poppers from Sync point B.
QueueReturn retVal = m_manager.reservePushIndex(generation, index);
if(retVal != QueueReturn::Success)
{
return retVal;
}
// Copy into the array. If the copy constructor throws, the pushGuard will
// roll the reserve back.
QueuePushGuard< Type > pushGuard(*this, generation, index);
// Construct in place.
::new(&m_data[index]) Type(value);
pushGuard.release();
m_manager.commitPushIndex(generation, index);
if(m_waitingPoppers > 0)
{
m_popSemaphore.notify();
}
return QueueReturn::Success;
}
template < typename Type >
QueueReturn
Queue< Type >::tryPushBack(Type &&value)
{
uint32_t generation = 0;
uint32_t index = 0;
// Sync point A
//
// The next call writes with full sequential consistency to the push
// index, which guarantees that the relaxed read to the waiting poppers
// count sees any waiting poppers from Sync point B.
QueueReturn retVal = m_manager.reservePushIndex(generation, index);
if(retVal != QueueReturn::Success)
{
return retVal;
}
// Copy into the array. If the copy constructor throws, the pushGuard will
// roll the reserve back.
QueuePushGuard< Type > pushGuard(*this, generation, index);
Type &dummy = value;
// Construct in place.
::new(&m_data[index]) Type(std::move(dummy));
pushGuard.release();
m_manager.commitPushIndex(generation, index);
if(m_waitingPoppers > 0)
{
m_popSemaphore.notify();
}
return QueueReturn::Success;
}
template < typename Type >
std::optional< Type >
Queue< Type >::tryPopFront()
{
uint32_t generation;
uint32_t index;
// Sync Point C.
//
// The call to reservePopIndex writes with full *sequential* consistency,
// which guarantees the relaxed read to waiting poppers is synchronized
// with Sync Point D.
QueueReturn retVal = m_manager.reservePopIndex(generation, index);
if(retVal != QueueReturn::Success)
{
return {};
}
// Pop guard will (even if the move/copy constructor throws)
// - destroy the original object
// - update the queue
// - notify any waiting pushers
QueuePopGuard popGuard(*this, generation, index);
return std::optional< Type >(std::move(m_data[index]));
}
template < typename Type >
QueueReturn
Queue< Type >::pushBack(const Type &value)
{
for(;;)
{
QueueReturn retVal = tryPushBack(value);
switch(retVal)
{
// Queue disabled.
case QueueReturn::QueueDisabled:
// We pushed the value back
case QueueReturn::Success:
return retVal;
default:
// continue on.
break;
}
m_waitingPushers.fetch_add(1, std::memory_order_relaxed);
// Sync Point B.
//
// The call to `full` below loads the push index with full *sequential*
// consistency, which gives visibility of the change above to
// waiting pushers in Synchronisation Point B.
if(full() && enabled())
{
m_pushSemaphore.wait();
}
m_waitingPushers.fetch_add(-1, std::memory_order_relaxed);
}
}
template < typename Type >
QueueReturn
Queue< Type >::pushBack(Type &&value)
{
for(;;)
{
QueueReturn retVal = tryPushBack(std::move(value));
switch(retVal)
{
// Queue disabled.
case QueueReturn::QueueDisabled:
// We pushed the value back
case QueueReturn::Success:
return retVal;
default:
// continue on.
break;
}
m_waitingPushers.fetch_add(1, std::memory_order_relaxed);
// Sync Point B.
//
// The call to `full` below loads the push index with full *sequential*
// consistency, which gives visibility of the change above to
// waiting pushers in Synchronisation Point C.
if(full() && enabled())
{
m_pushSemaphore.wait();
}
m_waitingPushers.fetch_add(-1, std::memory_order_relaxed);
}
}
template < typename Type >
Type
Queue< Type >::popFront()
{
uint32_t generation = 0;
uint32_t index = 0;
while(m_manager.reservePopIndex(generation, index)
!= QueueReturn::Success)
{
m_waitingPoppers.fetch_add(1, std::memory_order_relaxed);
if(empty())
{
m_popSemaphore.wait();
}
m_waitingPoppers.fetch_sub(1, std::memory_order_relaxed);
}
QueuePopGuard popGuard(*this, generation, index);
return Type(std::move(m_data[index]));
}
template < typename Type >
void
Queue< Type >::removeAll()
{
size_t elemCount = size();
uint32_t poppedItems = 0;
while(poppedItems++ < elemCount)
{
uint32_t generation = 0;
uint32_t index = 0;
if(m_manager.reservePopIndex(generation, index) != QueueReturn::Success)
{
break;
}
m_data[index].~Type();
m_manager.commitPopIndex(generation, index);
}
size_t wakeups = std::min(poppedItems, m_waitingPushers.load());
while(wakeups--)
{
m_pushSemaphore.notify();
}
}
template < typename Type >
void
Queue< Type >::disable()
{
m_manager.disable();
uint32_t numWaiting = m_waitingPushers;
while(numWaiting--)
{
m_pushSemaphore.notify();
}
}
template < typename Type >
void
Queue< Type >::enable()
{
m_manager.enable();
}
template < typename Type >
size_t
Queue< Type >::capacity() const
{
return m_manager.capacity();
}
template < typename Type >
size_t
Queue< Type >::size() const
{
return m_manager.size();
}
template < typename Type >
bool
Queue< Type >::enabled() const
{
return m_manager.enabled();
}
template < typename Type >
bool
Queue< Type >::full() const
{
return (capacity() <= size());
}
template < typename Type >
bool
Queue< Type >::empty() const
{
return (0 >= size());
}
template < typename Type >
QueuePushGuard< Type >::~QueuePushGuard()
{
if(m_queue)
{
// Thread currently has the cell at index/generation. Dispose of it.
uint32_t generation = 0;
uint32_t index = 0;
// We should always have at least one item to pop.
size_t poppedItems = 1;
while(m_queue->m_manager.reservePopForClear(generation, index,
m_generation, m_index))
{
m_queue->m_data[index].~Type();
poppedItems++;
m_queue->m_manager.commitPopIndex(generation, index);
}
// And release
m_queue->m_manager.abortPushIndexReservation(m_generation, m_index);
while(poppedItems--)
{
m_queue->m_pushSemaphore.notify();
}
}
}
template < typename Type >
void
QueuePushGuard< Type >::release()
{
m_queue = nullptr;
}
template < typename Type >
QueuePopGuard< Type >::~QueuePopGuard()
{
m_queue.m_data[m_index].~Type();
m_queue.m_manager.commitPopIndex(m_generation, m_index);
// Notify a pusher
if(m_queue.m_waitingPushers > 0)
{
m_queue.m_pushSemaphore.notify();
}
}
} // namespace thread
} // namespace llarp
#endif