环形缓冲区的有效边界检查

effective bounds checking on ring buffer

本文关键字：边界 检查 有效 缓冲区      更新时间：2023-10-16

正如一些人可能注意到的那样，我正在尝试实现一个环形缓冲区。我希望在数据结构中有一定的安全措施，同时不会损失太多的效率。

我当前的解决方案实现了两种类型的计数器，一种是索引，它是缓冲区内存的直接偏移量，另一种是序列号，它只是size_t类型的计数器。迭代器使用序列号来访问环形缓冲区。因此，环形缓冲区必须在每次缓冲区访问时从序列号转换为缓冲区索引。这通常相当有效：

size_t offset = seqNum - m_tailSeq;
size_t index = (m_tailIdx + offset) % m_size;

其中，seqNum是要转换的序列号，m_tailSeq是缓冲器中最旧元素的序列号、m_tailIdx是缓冲器中最大元素的缓冲器索引，m_size是缓冲存储器的大小。

但是，如果我在缓冲区中添加元素的时间足够长，序列号就会溢出。所以我必须检查一下。当我这样做的时候，我短暂而甜蜜的转变变成了这个怪物：

size_type getIndex(size_type seqNum) const
{
    size_type headSeq = m_tailSeq + m_numElements;
    // sequence does not wrap around
    if (m_tailSeq < headSeq)
    {
        // bounds check
        if(m_tailSeq <= seqNum && seqNum < headSeq) {
            size_type offset = seqNum - m_tailSeq;
            return (m_tailIdx + offset) % m_size;
        } else {
            throw BaseException("RingBuffer: access out of bounds", __FILE__, __LINE__);
        }
    }
    // sequence does wrap around
    else if (headSeq < m_tailSeq)
    {
        //bounds check (inverted from above)
        if(seqNum < headSeq) {
            size_type offset = (SIZE_TYPE_MAX - m_tailSeq) + seqNum;
            return (m_tailIdx + offset) % m_size;
        } else if (seqNum >= m_tailSeq) {
            size_type offset = seqNum - m_tailSeq;
            return (m_tailIdx + offset) % m_size;
        } else {
            throw BaseException("RingBuffer: access out of bounds", __FILE__, __LINE__);
        }
    }
    else if (isEmpty()) {
        throw BaseException("RingBufferIterator: accessing empty buffer", __FILE__, __LINE__);
    }
}

这相当于两个整数加法、一个整数减法、三个整数比较和一个模运算，在最佳情况下对单个缓冲区访问。不用说，在缓冲区上迭代会变得非常昂贵。然而，由于我想在高性能场景中使用这个缓冲区（即软实时应用程序中的事件队列），我希望这个数据结构尽可能高效。

当前的用例将作为事件缓冲区。一个（或可能不止一个）系统会将事件写入缓冲区，而其他系统（不止一个系统）则会按照自己的速度处理这些事件，而不删除它们。当缓冲区已满时，旧事件将被简单地覆盖。这样，我总是有最后几百个事件的记录，不同的系统可以按照各自的更新率对它们进行检查，并挑选出与它们相关的事件。不同的系统将保留一个指向环形缓冲区的迭代器，这样他们就知道上次在哪里停止以及在哪里恢复。当系统开始处理事件时，它需要确定其迭代器是否仍然有效或是否已被覆盖。事件很可能一次处理成大块，因此增量和取消引用应该很快。因此，基本上我们看到的是潜在的多线程上下文中的MPMC环形缓冲区。

我能自己想出的唯一解决方案是将错误检查的负担转移到缓冲区的用户身上。也就是说，用户必须首先（通过某种方式）检查其在缓冲区中的迭代器是否有效，确保缓冲区的某一段保持有效，然后在不进行任何进一步检查的情况下对该段进行迭代。然而，这似乎很容易出错，因为我必须在程序的多个部分而不是一个地方检查访问的安全性，如果我决定让缓冲区线程安全，那就会很麻烦。

我是不是错过了什么？这还能做得更好吗？我是不是犯了一些初学者的错误？

正如我在评论中提到的，无符号整数溢出是一种定义良好的操作。在C++中实现高效的序列号是关键。所以我们可以简单地减去两个无符号整数来得到距离。然后只向前距离函数，该函数通过边界检查实现索引访问。当所有可能的索引都低于序列号最大值的一半时，它将一如既往地工作。 
#include <array>
#include <climits>
#include <iostream>
unsigned int const SEQUENCE_NUMBER_FIRST = UINT_MAX-10;
class RingBuffer
{
public:
    void PushBack( char c )
    {
        GetBySeqNumber(m_tailSeq++) = c;
        if( Size() == m_buffer.size()+1 )
            PopFront();
    }
    void PopFront()
    {
        ++m_headSeq;
        if( ++m_offset % m_buffer.size() == 0 )
            m_offset = 0;
    }
    char& GetByIndex( size_t n )
    {
        if( n >= Size() )
            throw std::out_of_range("Hello, world!");
        return m_buffer[ (n+m_offset) % m_buffer.size() ];
    }
    char& GetBySeqNumber( unsigned int n )
    {
        // It is well defined operation in C++,
        // but if you try to use signed integer
        // it will become undefined behavior
        return GetByIndex( n-m_headSeq );
    }
    size_t Size() const
    {
        return m_tailSeq - m_headSeq;
    }
private:
    size_t m_offset = 0;
    unsigned int m_headSeq = SEQUENCE_NUMBER_FIRST;
    unsigned int m_tailSeq = SEQUENCE_NUMBER_FIRST;
    std::array<char,26> m_buffer;
};
int main()
{
    // initialize
    RingBuffer buf;
    for( char i=0; i<26; ++i )
        buf.PushBack( 'a'+i );
    // access trough sequence numbers
    // add or subtract one to get out of range exception
    for( unsigned int i=0; i<buf.Size(); ++i )
        std::cout << buf.GetBySeqNumber( SEQUENCE_NUMBER_FIRST+i );
    std::cout << std::endl;
    // push some more to overwrite first 10 values
    for( char i=0; i<10; ++i )
        buf.PushBack( '0'+i );
    // access trough indexes
    // add or subtract one to get out of range exception
    for( size_t i=0; i<buf.Size(); ++i )
        std::cout << buf.GetByIndex(i);
    std::cout << std::endl;
    return 0;
}
相关文章：