C++11 —— 单生产者/单消费者 的 FIFO 无锁队列

  发现 zeromq 的 yqueue_t 模板类，其数据存储理念设计得非常妙。借这一理念，按照 STL 的泛型类 queue 的接口标准，我设计了一个线程安全的 单生产者/单消费者（单线程push/单线程pop） FIFO 队列，以此满足更为广泛的应用。

1. 数据存储理念的结构图

• 队列的整体结构上，使用链表的方式，将多个固定长度的 chunk 串联起来；
• 每个 chunk 则可用于存储队列所需要的元素；
• 增加一个可交换的 chunk 单元，利于内存复用；
• 队列使用时，支持 单个线程的 push（生产） 和 单个线程 pop（消费）的并发操作（内部并未加锁）。

2. 源码 （xqueue.h）

/**
 * @file xqueue.h
 * Copyright (c) 2021 Gaaagaa. All rights reserved.
 * 
 * @author  : Gaaagaa
 * @date    : 2019-11-29
 * @version : 1.0.0.0
 * @brief   : 实现双线程安全的 单生产者/单消费者 FIFO 队列。
 */

/**
 * The MIT License (MIT)
 * Copyright (c) 2019, Gaaagaa All rights reserved.
 * 
 * Permission is hereby granted, free of charge, to any person obtaining a copy of
 * this software and associated documentation files (the "Software"), to deal in
 * the Software without restriction, including without limitation the rights to
 * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
 * of the Software, and to permit persons to whom the Software is furnished to do
 * so, subject to the following conditions:
 * 
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 * 
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
 * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
 * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
 * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */

#ifndef __XQUEUE_H__
#define __XQUEUE_H__

#include <memory>
#include <atomic>
#include <cassert>

////////////////////////////////////////////////////////////////////////////////
// xqueue_t : single producer/single consumer FIFO queue

/**
 * @class xqueue_t
 * @brief 双线程安全的 单生产者/单消费者 FIFO队列。
 * 
 * @tparam __elem_t  : 队列存储的节点元素类型。
 * @tparam __csize_v : 队列中的存储分块可容纳节点元素的数量。
 * @tparam __alloc_t : 元素对象分配器。
 */
template< typename __elem_t,
          size_t   __csize_v = 16,
          typename __alloc_t = std::allocator< __elem_t > >
class xqueue_t : protected __alloc_t
{
    static_assert(__csize_v >= 4,
                  "__csize_v size value must be greater than or equal to 4!");

    // common data types
public:
    using value_type      = __elem_t;
    using reference       = __elem_t &;
    using const_reference = const __elem_t &;
    using size_type       = size_t;

    static constexpr const size_type xchunk_size = __csize_v;

private:
    /** 前置声明 存储分块 的类型 */
    struct x_chunk_t;

    using x_chkptr_t  = struct x_chunk_t *;
    using x_chkpos_t  = size_t;
    using x_array_t   = value_type[xchunk_size];
    using x_swapchk_t = std::atomic< x_chkptr_t >;
    using x_quesize_t = std::atomic< size_type >;

    using x_alvalue_t = typename std::allocator_traits<
                            __alloc_t >::template rebind_alloc< __elem_t >;
    using x_alchunk_t = typename std::allocator_traits<
                            __alloc_t >::template rebind_alloc< x_chunk_t >;

    /**
     * @struct x_chunk_t
     * @brief  存储元素节点的连续内存块结构体。
     */
    typedef struct x_chunk_t
    {
        x_chkptr_t xchk_next;  ///< 指向后一内存块节点
        x_array_t  xchk_elem;  ///< 当前内存块中的元素节点数组
    } x_chunk_t;

    // constructor/destructor
public:
    explicit xqueue_t(void)
        : m_chk_swap(nullptr)
        , m_que_size(0)
        , m_que_front({ nullptr, 0 })
        , m_que_back ({ nullptr, 0 })
    {
        m_que_front.m_chk_vptr = alloc_chunk();
        m_que_back.m_chk_vptr  = m_que_front.m_chk_vptr;
    }

    ~xqueue_t(void)
    {
        while (size() > 0)
            pop();

        assert(m_que_front.m_chk_vptr == m_que_back.m_chk_vptr);
        recyc_chunk(m_que_front.m_chk_vptr);
        recyc_chunk(nullptr);
    }

    xqueue_t(xqueue_t && xobject) = delete;
    xqueue_t & operator = (xqueue_t && xobject) = delete;
    xqueue_t(const xqueue_t & xobject) = delete;
    xqueue_t & operator = (const xqueue_t & xobject) = delete;

    // public interfaces
public:
    /**********************************************************/
    /**
     * @brief 当前队列中的元素数量。
     */
    inline size_type size(void) const
    {
        return m_que_size;
    }

    /**********************************************************/
    /**
     * @brief 判断队列是否为空。
     */
    inline bool empty(void) const
    {
        return (0 == size());
    }

    /**********************************************************/
    /**
     * @brief 向队列后端压入一个元素。
     */
    void push(const value_type & xelem_value)
    {
        if (size() > 0)
        {
            back_forward();
        }

        x_alvalue_t::construct(
            &m_que_back.m_chk_vptr->xchk_elem[m_que_back.m_chk_npos],
            xelem_value);
        m_que_size.fetch_add(1);
    }

    /**********************************************************/
    /**
     * @brief 以右值引用方式，向队列后端压入一个元素。
     */
    void push(value_type && xelem_value)
    {
        if (size() > 0)
        {
            back_forward();
        }

        x_alvalue_t::construct(
            &m_que_back.m_chk_vptr->xchk_elem[m_que_back.m_chk_npos],
            std::forward< value_type >(xelem_value));
        m_que_size.fetch_add(1);
    }

    /**********************************************************/
    /**
     * @brief 向队列后端压入一个元素。
     */
    template< typename... __args_t >
    decltype(auto) emplace(__args_t &&... xargs)
    {
        if (size() > 0)
        {
            back_forward();
        }

        x_alvalue_t::construct(
            &m_que_back.m_chk_vptr->xchk_elem[m_que_back.m_chk_npos],
            std::forward< __args_t >(xargs)...);
        m_que_size.fetch_add(1);

        return back();
    }

    /**********************************************************/
    /**
     * @brief 从队列前端弹出一个元素。
     */
    void pop(void)
    {
        assert(size() > 0);

        x_alvalue_t::destroy(
            &m_que_front.m_chk_vptr->xchk_elem[m_que_front.m_chk_npos]);
        if (m_que_size.fetch_sub(1) > 1)
        {
            front_forward();
        }
    }

    /**********************************************************/
    /**
     * @brief 返回队列前端元素。
     */
    inline reference front(void)
    {
        assert(!empty());
        return m_que_front.m_chk_vptr->xchk_elem[m_que_front.m_chk_npos];
    }

    /**********************************************************/
    /**
     * @brief 返回队列前端元素。
     */
    inline const_reference front(void) const
    {
        assert(!empty());
        return m_que_front.m_chk_vptr->xchk_elem[m_que_front.m_chk_npos];
    }

    /**********************************************************/
    /**
     * @brief 返回队列后端元素。
     */
    inline reference back(void)
    {
        assert(!empty());
        return m_que_back.m_chk_vptr->xchk_elem[m_que_back.m_chk_npos];
    }

    /**********************************************************/
    /**
     * @brief 返回队列后端元素。
     */
    inline const_reference back(void) const
    {
        assert(!empty());
        return m_que_back.m_chk_vptr->xchk_elem[m_que_back.m_chk_npos];
    }

    // internal invoking
private:
    /**********************************************************/
    /**
     * @brief 申请分块。
     */
    inline x_chkptr_t alloc_chunk(void)
    {
        x_chkptr_t xchunk_ptr = m_chk_swap.exchange(nullptr);
        if (nullptr == xchunk_ptr)
        {
            x_alchunk_t xalloc_chunk;
            xchunk_ptr = xalloc_chunk.allocate(1);
            assert(nullptr != xchunk_ptr);
        }

        xchunk_ptr->xchk_next = nullptr;
        return xchunk_ptr;
    }

    /**********************************************************/
    /**
     * @brief 回收分块。
     */
    inline void recyc_chunk(x_chkptr_t xchunk_ptr)
    {
        x_chkptr_t xchk_rptr = m_chk_swap.exchange(xchunk_ptr);
        if (nullptr != xchk_rptr)
        {
            x_alchunk_t xalloc_chunk;
            xalloc_chunk.deallocate(xchk_rptr, 1);
        }
    }

    /**********************************************************/
    /**
     * @brief 将前端位置向后移（该接口仅由 pop() 接口调用）。
     */
    inline void front_forward(void)
    {
        if (++m_que_front.m_chk_npos == xchunk_size)
        {
            assert(nullptr != m_que_front.m_chk_vptr);
            assert(nullptr != m_que_front.m_chk_vptr->xchk_next);

            x_chkptr_t xchunk_ptr = m_que_front.m_chk_vptr;
            m_que_front.m_chk_vptr = xchunk_ptr->xchk_next;
            m_que_front.m_chk_npos = 0;

            recyc_chunk(xchunk_ptr);
        }
    }

    /**********************************************************/
    /**
     * @brief 将后端位置向后移（该接口仅由 push() 接口调用）。
     */
    inline void back_forward(void)
    {
        if (++m_que_back.m_chk_npos == xchunk_size)
        {
            assert(nullptr != m_que_back.m_chk_vptr);
            assert(nullptr == m_que_back.m_chk_vptr->xchk_next);

            x_chkptr_t xchunk_ptr = alloc_chunk();
            m_que_back.m_chk_vptr->xchk_next = xchunk_ptr;

            m_que_back.m_chk_vptr = xchunk_ptr;
            m_que_back.m_chk_npos = 0;
        }
    }

    // data members
protected:
    x_swapchk_t     m_chk_swap;   ///< 用于保存交换内存块（备用缓存块）
    x_quesize_t     m_que_size;   ///< 队列中的有效对象数量

    /** 指向队列前端的分块存储信息 */
    struct
    {
        x_chkptr_t  m_chk_vptr;   ///< 存储分块
        x_chkpos_t  m_chk_npos;   ///< 当前存储节点的索引号
    } m_que_front;

    /** 指向队列后端的分块存储信息 */
    struct
    {
        x_chkptr_t  m_chk_vptr;   ///< 存储分块
        x_chkpos_t  m_chk_npos;   ///< 当前存储节点的索引号
    } m_que_back;
};

////////////////////////////////////////////////////////////////////////////////

#endif // __XQUEUE_H__

3. 使用示例

#include "xqueue.h"
#include <iostream>
#include <thread>
#include <chrono>

////////////////////////////////////////////////////////////////////////////////

int main(int argc, char * argv[])
{
    using x_int_queue_t = xqueue_t< int, 8 >;

    x_int_queue_t spsc;

    std::cout << "sizeof(x_int_queue_t) : " << sizeof(x_int_queue_t) << std::endl;

    bool b_push_finished = false;
    std::thread xthread_in([&spsc, &b_push_finished](void) -> void
    {
        for (int i = 0; i < 1000; ++i)
        {
            spsc.push(i);
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
        }

        b_push_finished = true;
    });

    std::thread xthread_out([&spsc, &b_push_finished](void) -> void
    {
        int i = 0;
        while (true)
        {
            if (!spsc.empty())
            {
                std::cout << "[" << ++i   << "] "
                          << spsc.size()  << ", "
                          << spsc.front() << std::endl;
                spsc.pop();
                std::this_thread::sleep_for(std::chrono::milliseconds(1));
            }
            else if (b_push_finished)
            {
                break;
            }
        }
    });

    if (xthread_in.joinable())
    {
        xthread_in.join();
    }

    if (xthread_out.joinable())
    {
        xthread_out.join();
    }

    return 0;
}

////////////////////////////////////////////////////////////////////////////////