I require hundreds of equal, small sized buffers in my current project. The bulk of the computation in the program operates on data stored directly in those containers, so high performance is imperative. As the number of the buffers stays constant at runtime, my current approach is to use a std::vector<std::deque<MyObj>>
as the buffer, but I do not like the low cache locality (since the actual storage of each deque is spread on the heap) and possibility for memory fragmentation that this approach entails.
Thus I would like to replace my current container with something like a std::vector<somecontainer<MyObj>>
, where somecontainer has at least a part of the contained objects stored locally, within the container object itself, and thus in contigous storage in the vector.
My attempt to implement such a class is presented below. The container Ring
is implemented as a circular buffer, permitting fast insertion and deletion at either end, and supports indexed access to the contents. The size is fixed at compile time, and is restricted to powers of two in order to effectively utilize binary modular integer arithmetic. It is still incomplete, lacking mainly iterators and an assignment operator capable of handling Ring
objects to others with different capacities. I do not plan to add exception support to the class or bounds checking to the access functions.
Are there any serious pitfalls in the design, or any significant improvements to be made? Will the memory alignment of the elements be correct, and will move constructors be used when inserting elements if possible? Could the container even be adapted to work as a thread safe, single producer - single consumer queue by substituting the current indexes of type uintN_t with indexes of type std::atomic_uintN_t along with other small modifications?
Usage example:
Ring<int, 8> buf(4, 0);
buf.popFront();
buf.pushBack(1);
buf.pushFront(-1);
while(!buf.isFull())
buf.pushBack(2);
auto buf2(buf);
buf.clear();
for(size_t n = 0; n < buf2.getSize(); n++)
std::cout << buf2[n] << ", ";
//-1, 0, 0, 0, 1, 2, 2, 2,
Implementation:
#include <cstdint>
template<int N> struct UintByBits{ };
template<> struct UintByBits<8> { using type = uint8_t; };
template<> struct UintByBits<16> { using type = uint16_t; };
template<> struct UintByBits<32> { using type = uint32_t; };
template<> struct UintByBits<64> { using type = uint64_t; };
constexpr getReqdBitCount(uint64_t val)
{
return (val <= 0xFF) ? 8 : (
(val <= 0xFFFF) ? 16 : (
(val <= 0xFFFFFFFF) ? 32 : 64));
}
//MinUint is an alias for the smallest unsigned integer type where MaxVal fits
template<uint64_t MaxVal>
using MinUint = typename UintByBits<getReqdBitCount(MaxVal)>::type;
constexpr bool isPowerOfTwo(uint64_t val)
{
return val != 0 && (val & (val - 1)) == 0;
}
//A constant-sized dual-ended queue implemented as a ring buffer. The size of
//..the container may only be a power of two (this is a requirement of the
//..modular arithmetic performed internally)
template<typename T, size_t BufSize>
class Ring
{
static_assert(isPowerOfTwo(BufSize),
"'Ring' buffer size (template param 'BufSize') is not a power of two");
//Aligned raw memory array for constructing objects of type T into
using ElemT = typename std::aligned_storage<sizeof(T), alignof(T)>::type;
ElemT ring_[BufSize];
//Indexes to the raw memory pointig to the head and tail end of the ring.
//..The indexes are each guaranteed to have one extra "parity bit" of storage,
//..which is used for distinguishing between an empty ring and a full ring.
MinUint<BufSize> ringFront_;
MinUint<BufSize> ringBack_;
//Bit masks used in bitwise operations: INDEX_MASK is used to remove all bits
//..from an index that do not express the actual element position, PARITY_BIT
//..is used for masking and comparing to the parity bit.
static constexpr MinUint<BufSize> INDEX_MASK = BufSize - 1;
static constexpr MinUint<BufSize> PARITY_BIT = BufSize;
public:
//When the Ring object is first constructed, ringFront_ points to the first
//..element and ringBack_ to the last element in the container. Pushing one
//..element to either end causes the index of the affected end to roll around,
//..resulting in both indexes pointing to the same element, at which point the
//..element can also be popped from either end.
Ring() : ringFront_(0), ringBack_(BufSize - 1)
{ }
//Fill constructor: copy construct n elements
Ring(size_t n, const T &val) : ringFront_(0), ringBack_(BufSize - 1)
{
while(n-- > 0)
pushBack(val);
}
//Fill constructor: construct n elements using default constructor
explicit Ring(size_t n) : ringFront_(0), ringBack_(BufSize - 1)
{
while(n-- > 0)
pushBack();
}
//The assignment operator destroys all elements copy constructs new ones.
//..Perhaps a templated version with a bounds check could accept a different
//..sized Ring of the same value type?
Ring& operator=(const Ring &other)
{
if(this != &other)
{
clear();
size_t n = other.getSize();
while(n-- > 0)
pushFront(other[n]);
}
return *this;
}
//Copy constructor
Ring(const Ring &other) : ringFront_(0), ringBack_(BufSize - 1)
{
size_t n = other.getSize();
while(n-- > 0)
pushFront(other[n]);
}
~Ring()
{
clear();
}
//Get the element at the position 'position' relative to the front of the
//..ring (that is, the element returned by getFront()). Pushing/popping at
//..the front will thus change which element a given index points to, while
//..pushing/popping at the back will not. Accessing at indexes equal getSize()
//..NOT BOUNDS CHECKED
T &operator[] (size_t position)
{
return *static_cast<T*>(static_cast<void*>(
ring_ + ((ringFront_ + position) & INDEX_MASK))
);
}
const T &operator[] (size_t position) const
{
return *static_cast<const T*>(static_cast<const void*>(
ring_ + ((ringFront_ + position) & INDEX_MASK))
);
}
//Construct an element to the front. NOT BOUNDS CHECKED
template<typename ... Args>
void pushFront(Args&& ... args)
{
--ringFront_;
new(ring_ + (ringFront_ & INDEX_MASK)) T(std::forward<Args>(args)...);
}
//Destroy the element at the front. NOT BOUNDS CHECKED
void popFront()
{
getFront().~T();
++ringFront_;
}
//Get a reference to the front element. If size == 1 returns the same element
//..as getBack(). Undefined for an empty container.
T &getFront()
{
return *static_cast<T*>(static_cast<void*>(
ring_ + (ringFront_ & INDEX_MASK))
);
}
const T &getFront() const
{
return *static_cast<const T*>(static_cast<const void*>(
ring_ + (ringFront_ & INDEX_MASK))
);
}
//Construct an element to the back. NOT BOUNDS CHECKED
template<typename ... Args>
void pushBack(Args&& ... args)
{
++ringBack_;
new(ring_ + (ringBack_ & INDEX_MASK)) T(std::forward<Args>(args)...);
}
//Destroy the element at the front. NOT BOUNDS CHECKED
void popBack()
{
getBack().~T();
--ringBack_;
}
//Get a reference to the back element. If size == 1 returns the same element
//..as getFront(). Undefined for an empty container.
T &getBack()
{
return *static_cast<T*>(static_cast<void*>(
ring_ + (ringBack_ & INDEX_MASK))
);
}
const T &getBack() const
{
return *static_cast<const T*>(static_cast<const void*>(
ring_ + (ringBack_ & INDEX_MASK))
);
}
//Destroy all elements in the container.
void clear()
{
while(!isEmpty())
popFront();
}
bool isEmpty() const
{
//if the indexes are otherwise equal but the parity bits differ, the ring is empty
return (((ringBack_ + 1) ^ ringFront_) & (PARITY_BIT | INDEX_MASK)) == PARITY_BIT;
}
bool isFull() const
{
//if the indexes are equal including the parity bits, the ring is empty
return (((ringBack_ + 1) ^ ringFront_) & (PARITY_BIT | INDEX_MASK)) == 0;
}
size_t getSize() const
{
//distinguish between a full and an empty ring, otherwise it would never return 0
return isEmpty() ? 0 : ((ringBack_ - ringFront_) & INDEX_MASK) + 1;
}
constexpr size_t getCapacity() const
{
return BufSize;
}
};