双向蛇形：探索循环数组和队列

• 下载源代码 - 12.6 KB

引言

所以我想到了一个绝妙的主意，想为计算机科学的经典理论贡献一些新东西。我想，这就是傲慢的意思。

我曾认为可以创建一个类，它能实现双向链表的功能，但基于数组，从而获得惊人的性能。它将有两个“标记”，指向数组中的某个位置，以跟踪列表的头部和尾部，并且您可以轻松且低成本地在任一端添加或删除元素。我称这种数据结构为 Snake，因为它有头部和尾部，并且可以绕自身一圈。

我听说过循环数组，我想它们被称为环形缓冲区，但我不知何故忘记了这一点。唉。

于是我写下了我的新颖想法。编码过程非常有趣！我编写了单元测试，修复了所有 bug，并对我的创造感到非常满意。然后我开始测量性能。但在此之前，先谈谈误解……

误解

我的第一个误解是 .NET 的 Queue 类基于 LinkedList 或某种内部链表。事实并非如此。根据文档，Queue 使用循环数组，这比 Snake 更简单，但它们拥有快速添加元素到一端并从另一端移除所需的所有技术，而无需像链表那样进行大量内存分配。Queue 类支持各种操作和访问器；这是一个很棒的类。

我还有几个误解，包括 C++ 的 std::queue 类是基于双向链表 std::list 类，以及 std::deque 类效率低下。C++ 的 queue 类默认是基于 std::deque 构建的。而 std::deque 如您在下文的性能对比部分将看到的，效率相当高。

我最后的误解——可能不是我将要有的最后一个，只是在这里的最后一个——是 .NET 的 ConcurrentQueue 类性能很差。

所有这些说法都是错误的！

循环数组描述

循环数组使用一个普通数组来存储一系列元素。与普通数组不同，循环数组允许在序列的开头添加元素而不必将现有元素向下移动。它通过标记来实现一个概念上的列表：一个标记表示头部，另一个表示尾部。当您在头部添加元素时，数组索引值会递减，当您到达数组的开头（索引 0）时，标记会绕到末尾（索引大小 - 1），然后您继续从那里添加元素，索引值继续递减，直到您碰到尾部标记。此时，您会扩展数组并重置标记，然后从头开始，在末尾留出空间以容纳更多元素。

向末尾添加元素是更好的选择。读写使用前向数据访问，这能更好地利用处理器缓存。向尾部添加时，您会一直添加直到到达数组末尾或头部标记，然后绕到数组的前面。您会一直添加，直到碰到头部标记，那时您会扩展数组并重新开始。

这听起来像是需要大量的代码来跟踪自身并进行大量扩展。这种数据结构最有意义的场景是当一个代码部分从头部读取数据，而另一部分代码向尾部写入数据。这时头部会追赶尾部读取数据。这就形成了一个队列。

循环数组似乎非常适合实现队列数据结构。

Here is what a circular buffer looks like in memory with three elements 
and free spaces around the head and tail:
[ ][ ][ ][1][2][3][ ][ ]
       ^              ^
       head marker    tail marker

Add element 4:
[ ][ ][ ][1][2][3][4][ ]
       ^              ^
       head marker    tail marker

Add 5:
[ ][ ][ ][1][2][3][4][5]
 ^     ^                 
 tail  head marker 
      
Add 6:
[6][ ][ ][1][2][3][4][5]
    ^  ^                 
 tail  head marker

Try to add 7, tail marker hits head marker, array is grown, element added, 
plenty of room to grow
[ ][1][2][3][4][5][6][7][ ][ ][ ][ ][ ][ ]...
 ^                       ^
 head                    tail

At this point, or perhaps while all that adding was going on, 
you could remove forward from the head until you reach the tail.

The head chases the tail forward in memory.

Snake 代码

我首先用 C# 编写了 Snake，因为它比 C++ 开发更快、更易于掌握。一旦我对我的 C# 类满意，移植到 C++ 就变得容易了。我有测试来验证它，并且是逐行移植，没有意外。

这是 Snake 的 C++ 代码

#pragma once

#pragma warning(push)
#pragma warning(disable : 4365)
#include <exception>
#include <string>
#include <sstream>
#pragma warning(pop)

template <typename T>
class Snake
{
public:
    Snake()
        : m_data(new T[2]) // this must always have two array spots 
                           // for the two end markers
        , m_capacity(2)    // how big m_data is
        , m_size(0)        // size is never the length of the array, 
                           // always at least two less
        , m_startIdx(0)    // point the end markers...
        , m_endIdx(1)      // ...into the initial array
    {
    }

    ~Snake()
    {
        delete[] m_data;
        m_data = nullptr;
    }

    size_t size() const { return m_size; }
    
    size_t capacity() const { return m_capacity; }

    T front() const 
    { 
        if (m_size == 0)
            throw std::exception("No first element to get");
        else
            return m_data[(m_startIdx + 1) % m_capacity];
    }
    
    T back() const 
    { 
        if (m_size == 0)
            throw std::exception("No last element to get");
        else
            return m_data[(m_startIdx + m_size) % m_capacity];
    }

    void push_front(const T& elem)
    {
        size_t new_start_idx, data_idx;
        if (m_startIdx == 0)
        {
            new_start_idx = m_capacity - 1;
            data_idx = 0;
        }
        else
        {
            new_start_idx = m_startIdx - 1;
            data_idx = m_startIdx;
        }

        if (new_start_idx == m_endIdx)
        {
            grow();
            push_front(elem);
            return;
        }

        m_startIdx = new_start_idx;
        m_data[data_idx] = elem;
        ++m_size;
    }

    void push_back(const T& elem)
    {
        size_t new_end_idx, data_idx;
        if (m_endIdx == m_capacity - 1)
        {
            new_end_idx = 0;
            data_idx = m_capacity - 1;
        }
        else
        {
            data_idx = m_endIdx;
            new_end_idx = m_endIdx + 1;
        }

        if (new_end_idx == m_startIdx)
        {
            grow();
            push_back(elem);
            return;
        }

        m_endIdx = new_end_idx;
        m_data[data_idx] = elem;
        ++m_size;
    }

    void pop_front()
    {
        size_t proposed_idx = (m_startIdx + 1) % m_capacity;
        if (proposed_idx == m_endIdx)
            throw std::exception("No first element to remove");
        else
            m_startIdx = proposed_idx;
        --m_size;
    }

    void pop_back()
    {
        size_t proposed_idx = m_endIdx == 0 ? m_capacity - 1 : m_endIdx - 1;
        if (proposed_idx == m_startIdx)
            throw std::exception("No last element to remove");
        else
            m_endIdx = proposed_idx;
        --m_size;
    }

    T operator[](size_t idx) const
    {
        if (idx >= m_size)
            throw std::exception("The index is out of range");
        else
            return m_data[(m_startIdx + idx + 1) % m_capacity];
    }

    std::string diagnostic_dump()
    {
        std::stringstream sb;
        
        sb
        << "size: " << m_size
        << " - "
        << "start: " << m_startIdx
        << " - "
        << "end: " << m_endIdx
        << "\n";

        for (size_t idx = 0; idx < m_capacity; ++idx)
        {
            sb << "[" << idx << "]";

            if (idx == m_startIdx)
                sb << " (start)";
            if (idx == m_endIdx)
                sb << " (end)";

            sb << " " << m_data[idx] << "\n";
        }

        return sb.str();
    }

private:
    void grow()
    {
        // create the new larger array
        size_t new_capacity = m_capacity * 2;
        T* new_data = new T[new_capacity];

        // copy data from the original array into the front of the new array, 
        // sort of compacting it
        // start idx at 1 so it is after the start marker
#pragma warning(push)
#pragma warning(disable : 6386)
        for (size_t idx = 1; idx <= m_size; ++idx)
            new_data[idx] = m_data[(m_startIdx + idx) % m_capacity];
#pragma warning(pop)

        // reset state
        delete[] m_data;
        m_data = new_data;
        m_capacity = new_capacity;
        m_startIdx = 0;
        m_endIdx = m_size + 1;
    }

private:
    // The raw array and the number of elements in it
    T* m_data;
    size_t m_capacity;
    size_t m_size;

    // These indexes point to the head and tail markers in the array
    size_t m_startIdx;
    size_t m_endIdx;
};

附件代码布局

附件的 ZIP 文件包含源代码

• snakenet - 包含 Snake 类和 BufferQueue（一个线程安全的包装队列类）的 C# 类库
• snaketest - C# 单元测试
• snakeperf - C# 命令行应用程序，用于比较各种队列方法的性能
• snakeplus - 包含 Snake 类、单元测试和性能测试驱动程序（全部集成在一起）的 C++ 项目

性能对比

现在是大家期待已久的环节，让我们将它们进行对比，看看结果如何。测试涉及入队和出队一千万个 int。

这是 .NET 的测试结果

Array (.NET List): 87 ms // no FIFO on this, just LIFO to measure 
                         // general performance of working with the 10M elements

// these "Easy" tests simply add the elements, then remove them all
Linked List Easy: 1279 ms
Snake Easy: 380 ms
Queue Easy: 140 ms       // wow, Queue is much faster than Snake!

// these tests create a Task for dequeuing, then enqueue the elements, 
// then wait on the Task
Linked List: 846 ms
Snake: 845 ms
BufferQueue: 744 ms      // a thread-safe class I created based on Snake
Queue: 675 ms

// Oh my, the clear winner!  It's not even close.
ConcurrentQueue: 162 ms 

这是 C++ 的测试结果

std::vector: 46 ms       // not FIFO, LIFO: push_back the ints, pop_back until empty
Snake Easy: 204 ms
std::list (Linked List) Easy: 673 ms
std::queue (std::deque) Easy: 267 ms 

结论和关注点

• 编写集合类型并使其正确高效是件有趣的事情。
• 只是不要指望“改变世界”：阳光下很少有真正全新的东西。
• .NET 的 Queue 类非常高效，在不关心线程安全的情况下是一个绝佳的选择。
• .NET 的 ConcurrentQueue 类在所有测试类型中提供了迄今为止最快的线程间排队性能。
• C++ 的数据结构可以比其逐行对应的 C# 版本快 2 倍。
• C++ 的 std::queue 是基于 std::deque 的，而不是 std::list。
• C++ 的 std::deque 非常高效。
• 在所有此处进行的测试中，应该很清楚，使用链表进行排队是没有充分理由的。
• 没有令人信服的理由仅为了提高排队代码的性能而转向使用 C++。

历史

• 2024 年 3 月 7 日：初始版本