上位机开发笔记：环形缓冲区

2023-05-16

文章目录

前言
1. 环形缓冲区工作机制
- 1.1 实现原理
- 1.2 区分缓冲区满或者空
- - 1) 总是保持一个存储单元为空
  - 2) 使用计数数据
  - 3) 镜像指示位
2. Qt实现环形缓冲区
- 2.1 QByteArray环形缓冲区
- 2.2 QSemaphore实现环形缓冲区
参考资料

前言

环形缓冲区(Ring buffer/Circular buffer)或称环形队列，是一种用于表示一个固定尺寸、头尾相连的缓冲区的数据结构，适合缓存数据流。比如对通信中接收到的数据进行处理，当数据量太大或者处理速度跟不上接收速度时，就会将之前还未处理的数据覆盖掉，即出现丢包的现象。此时，使用环形缓冲区就可以有效解决这一问题。

1. 环形缓冲区工作机制

环形缓冲区是一种先进先出(FIFO)的数据结构，本质上就是将一个顺序存储结构的访问逻辑由线性处理为环形，通常由一个固定大小的顺序容器实现，通过一定的方法对数据的读写范围进行控制。

1.1 实现原理

一般地，环形缓冲区通过读写指针控制读写范围：

读指针：指向缓冲区内下一个可读数据的位置。
写指针：指向缓冲区内下一个可写数据的位置。

初始化时，读指针和写指针都指向缓冲区的0号元素，通过移动这两个指针即可对缓冲区的数据进行读写操作。当指针移动到容器末尾时，立即返回容器头以实现首尾相连的环形结构。

在这里插入图片描述

1.2 区分缓冲区满或者空

缓冲区是满或是空，都有可能出现读指针与写指针指向同一位置，有多种策略用于检测缓冲区是满或是空：

在这里插入图片描述

1) 总是保持一个存储单元为空

缓冲区中总是有一个存储单元保持未使用状态，此时缓冲区最多存入size - 1个数据。如果读写指针指向同一位置，则缓冲区为空；如果读指针位于写指针的相邻后一个位置，则缓冲区为满。

数据结构：

typedef struct
{
    unsigned int size;
    unsigned int writeIndex;
    unsigned int readIndex;
    unsigned char *buff = nullptr;
} RING_BUFFER;

缓冲区满空判断：

bool ringBufferIsEmpty(const RING_BUFFER &rb)
{
    return rb.writeIndex == rb.readIndex; //如果读写指针指向同一位置，则缓冲区为空。
}

bool ringBufferIsFull(const RING_BUFFER &rb)
{
    //通过除余缓冲区大小实现环形索引
    return ((rb.writeIndex + 1) % rb.size) == (rb.readIndex % rb.size); //如果读指针位于写指针的相邻后一个位置，则缓冲区为满。
}

缓冲区写入数据：

一般来说，写操作禁止覆盖缓冲区的数据，采取返回一个错误码或者抛出异常的策略。但在某些情况下覆盖环形缓冲区中未被处理的数据是允许的，特别是在多媒体处理时。例如，音频的生产者可以覆盖掉声卡尚未来得及处理的音频数据。

int writeToRingBuffer(RING_BUFFER &rb, unsigned char *szData, int iLen)
{
    if (ringBufferIsFull(rb))
    {
        return -1;
    }

    //先判断可写数据数量
    int available = 0;
    if (rb.writeIndex >= rb.readIndex)
    {
        available = rb.size - (rb.writeIndex - rb.readIndex) - 1; //写指针不能写到读指针所在位置，要保持一个存储单元
    }
    else
    {
        available = rb.readIndex - rb.writeIndex - 1;
    }
    if (iLen > available)
    {
        // return -2; //抛出异常并退出
        iLen = available; //修改写入数据数量
    }

    //跨越边界要进行分段拷贝
    if ((rb.writeIndex + iLen) >= rb.size)
    {
        unsigned int len1 = rb.size - rb.writeIndex;
        unsigned int len2 = iLen - len1;
        memcpy(rb.buff + rb.writeIndex, szData, len1);
        memcpy(rb.buff, szData + len1, len2);
        rb.writeIndex = len2;
    }
    else
    {
        memcpy(rb.buff + rb.writeIndex, szData, iLen);
        rb.writeIndex += iLen;
    }
    return 0;
}

缓冲区读取数据：

int readFromRingBuffer(RING_BUFFER &rb, unsigned char *szData, int &iLen)
{
    if (ringBufferIsEmpty(rb))
    {
        return -1;
    }

    //先判断可读数据数量
    int available = 0;
    if (rb.writeIndex >= rb.readIndex)
    {
        available = rb.writeIndex - rb.readIndex; //读指针可以读到写指针的位置，此时缓冲区为空
    }
    else
    {
        available = rb.size - (rb.readIndex - rb.writeIndex);
    }
    if (iLen > available)
    {
        iLen = available;
    }

    //跨越边界要进行分段拷贝
    if ((rb.readIndex + iLen) >= rb.size)
    {
        unsigned int len1 = rb.size - rb.readIndex;
        unsigned int len2 = iLen - len1;
        memcpy(szData, rb.buff + rb.readIndex, len1);
        memcpy(szData + len1, rb.buff, len2);
        rb.readIndex = len2;
    }
    else
    {
        memcpy(szData, rb.buff + rb.readIndex, iLen);
        rb.readIndex += iLen;
    }
    return 0;
}

这种策略的优点是简单、粗暴，缺点是语义上实际可存数据量与缓冲区容量不一致，且多线程时需要并发控制。

2) 使用计数数据

这种策略需要在数据结构内添加一个变量保存缓冲区内存储数据的计数，写操作增加计数、读操作减少计数。如果存储数据数量为0，则缓冲区为空；如果存储数据数量等于缓冲区容量，则缓冲区为满。

数据结构：

typedef struct
{
    unsigned int size;
    unsigned int writeIndex;
    unsigned int readIndex;
    unsigned int buffCount; //存储数据计数
    unsigned char *buff = nullptr;
} RING_BUFFER;

缓冲区满空判断：

bool ringBufferIsEmpty(const RING_BUFFER &rb)
{
    return 0 == rb.buffCount;
}

bool ringBufferIsFull(const RING_BUFFER &rb)
{
    return rb.buffCount == rb.size;
}

缓冲区写入数据：

int writeToRingBuffer(RING_BUFFER &rb, unsigned char *szData, int iLen)
{
    if (ringBufferIsFull(rb))
    {
        return -1;
    }

    int available = rb.size - rb.buffCount;
    if (iLen > available)
    {
        // return -2;
        iLen = available;
    }

    if ((rb.writeIndex + iLen) >= rb.size)
    {
        unsigned int len1 = rb.size - rb.writeIndex;
        unsigned int len2 = iLen - len1;
        memcpy(rb.buff + rb.writeIndex, szData, len1);
        memcpy(rb.buff, szData + len1, len2);
        rb.writeIndex = len2;
    }
    else
    {
        memcpy(rb.buff + rb.writeIndex, szData, iLen);
        rb.writeIndex += iLen;
    }
    rb.buffCount += iLen;
    return 0;
}

缓冲区读取数据：

int readFromRingBuffer(RING_BUFFER &rb, unsigned char *szData, int &iLen)
{
    if (ringBufferIsEmpty(rb))
    {
        return -1;
    }

    int available = rb.buffCount;
    if (iLen > available)
    {
        iLen = available;
    }

    if ((rb.readIndex + iLen) >= rb.size)
    {
        unsigned int len1 = rb.size - rb.readIndex;
        unsigned int len2 = iLen - len1;
        memcpy(szData, rb.buff + rb.readIndex, len1);
        memcpy(szData + len1, rb.buff, len2);
        rb.readIndex = len2;
    }
    else
    {
        memcpy(szData, rb.buff + rb.readIndex, iLen);
        rb.readIndex += iLen;
    }
    rb.buffCount -= iLen;
    return 0;
}

该策略的优点是运算速度快、使用简单。缺点是读写操作都需要修改这个存储数据计数，对于多线程访问缓冲区需要并发控制。

还有一种实现方法是用2个变量分别保存写入、读出缓冲区的数据数量，其差值就是缓冲区中尚未被处理的有效数据的数量。如果写入数据数量与读出数据数量相等，则缓冲区为空；如果写入数据数量与读出数据数量的差值等于缓冲区容量，则缓冲区为满。

数据结构：

这种策略甚至可以省略读写指针，通过读写数据数量%缓冲区容量就可以实现索引访问。

typedef struct
{
    unsigned int size;
    unsigned long writeNum;
    unsigned long readNum;
    unsigned char *buff = nullptr;
} RING_BUFFER;

缓冲区满空判断：

bool ringBufferIsEmpty(const RING_BUFFER &rb)
{
    return rb.writeNum == rb.readNum;
}

bool ringBufferIsFull(const RING_BUFFER &rb)
{
    return rb.writeNum == (rb.readNum + rb.size);
}

缓冲区写入数据：

int writeToRingBuffer(RING_BUFFER &rb, unsigned char *szData, int iLen)
{
    if (ringBufferIsFull(rb))
    {
        return -1;
    }

    int available = rb.size - (rb.writeNum - rb.readNum);
    if (iLen > available)
    {
        //return -2;
        iLen = available;
    }

    unsigned int writeIndex = rb.writeNum % rb.size;
    if ((writeIndex + iLen) >= rb.size)
    {
        unsigned int len1 = rb.size - writeIndex;
        unsigned int len2 = iLen - len1;
        memcpy(rb.buff + writeIndex, szData, len1);
        memcpy(rb.buff, szData + len1, len2);
    }
    else
    {
        memcpy(rb.buff + writeIndex, szData, iLen);
    }
    rb.writeNum += iLen;
    return 0;
}

缓冲区读取数据：

int readFromRingBuffer(RING_BUFFER &rb, unsigned char *szData, int &iLen)
{
    if (ringBufferIsEmpty(rb))
    {
        return -1;
    }

    int available = rb.writeNum - rb.readNum;
    if (iLen > available)
    {
        iLen = available;
    }

    unsigned int readIndex = rb.readNum % rb.size;
    if ((readIndex + iLen) >= rb.size)
    {
        unsigned int len1 = rb.size - readIndex;
        unsigned int len2 = iLen - len1;
        memcpy(szData, rb.buff + readIndex, len1);
        memcpy(szData + len1, rb.buff, len2);
    }
    else
    {
        memcpy(szData, rb.buff + readIndex, iLen);
    }
    rb.readNum += iLen;   
    return 0;
}

3) 镜像指示位

一般情况下，环形缓冲区的长度如果是n，则读写指针可访问的范围为0～n-1。该策略规定读写指针的可访问范围为0～2n-1，其中0～n-1对应于常规的逻辑地址空间，n至2n-1为镜像逻辑地址空间。

在这里插入图片描述

使用一位表示写指针或读指针是否进入了虚拟的镜像存储区：置位表示进入，不置位表示没进入。在读写指针的值相同情况下，如果二者的指示位相同，说明缓冲区为空；如果二者的指示位不同，说明缓冲区为满。

相关代码可查看：维基百科-环形缓冲区

这种方法优点是判断缓冲区满/空很简单、不需要做取余数操作，读写线程可以分别设计专用算法策略，能实现精致的并发控制。缺点是读写指针各需要额外的一位作为指示位。

2. Qt实现环形缓冲区

2.1 QByteArray环形缓冲区

Qt中专门封装了字节数组QByteArray类，使得我们可以灵活地分配字节数组的内存，不用每次都在堆内存上创建，杜绝了内存泄漏的风险。此外，QByteArray几乎可以说是为通信编程量身定制，不同于QString里面存的是Unicode编码的字符串，QByteArray内存储的是没有经过编码的原始数据。对于上述的环形缓冲区就可以用QByteArray进行改写，这里采用读/写计数的方式：

数据结构：

typedef struct
{
    unsigned int size;
    unsigned int writeNum;
    unsigned int readNum;
    QByteArray buff;
} RING_BUFFER;

初始化：

void initRingBuffer(RING_BUFFER &rb, unsigned int sz)
{
    rb.size = sz;
    rb.readNum = 0;
    rb.writeNum = 0;
    rb.buff.resize(sz); //使用resize()设置数组的大小，每个元素初始化为0
}

缓冲区满空判断与上文相同。

缓冲区写入数据：Qt中绝大部分的通信API都是以char*为基本数据格式的，可以直接将char*作为参数写入缓冲区。

int writeToRingBuffer(RING_BUFFER &rb, char *szData, int iLen)
{
    if (ringBufferIsFull(rb))
    {
        return -1;
    }

    mutex.lock();
    int available = rb.size - (rb.writeNum - rb.readNum);
    mutex.unlock();
    if (iLen > available)
    {
        //return -2;
        iLen = available;
    }

    QByteArray arr(szData, iLen); //通过QByteArray的构造函数将char*转为QByteArray
    for(auto it = arr.cbegin(); it != arr.cend(); ++it)
    {
        rb.buff[rb.writeNum % rb.size] = *it;
        ++rb.writeNum;
    }

    return 0;
}

缓冲区读取数据：

int readFromRingBuffer(RING_BUFFER &rb, QByteArray &arr, int &iLen)
{
    if (ringBufferIsEmpty(rb))
    {
        return -1;
    }

    mutex.lock();
    int available = rb.writeNum - rb.readNum;
    mutex.unlock();
    if (iLen > available)
    {
        iLen = available;
    }

    for (int i = 0; i < iLen; ++i)
    {
        arr.push_back(rb.buff.at(rb.readNum % rb.size));
        ++rb.readNum;
    }
    return 0;
}

2.2 QSemaphore实现环形缓冲区

在实际使用中，对环形缓冲区的读/写操作通常都是由不同线程进行的，不同线程对同一资源进行操作，使用信号量处理更加合适。Qt中使用QSemaphore类实现信号量，用2个信号量分别表示可读/可写资源数量，通过读写计数实现索引。

数据结构：

/*---------------------------global.h----------------------------*/
extern const quint32 BUFFERSIZE;
extern QByteArray g_szBuffer;
extern QSemaphore freeBytes;
extern QSemaphore usedBytes;
extern quint32 writeNum;
extern quint32 readNum;
extern QMutex g_mutex;

/*---------------------------global.cpp----------------------------*/
#include "global.h"

const quint32 BUFFERSIZE = 10;
QByteArray g_szBuffer(BUFFERSIZE, '\0');
QSemaphore freeBytes(BUFFERSIZE);
QSemaphore usedBytes;
quint32 writeNum = 0;
quint32 readNum = 0;
QMutex g_mutex;

缓冲区写入数据：

void writeToRingBuffer(char *szData, int iLen)
{
    if(szData == nullptr || iLen == 0)
    {
        return;
    }
    
	if(freeBytes.available() - iLen < 0)
    {
    	iLen = freeBytes.available();
    }
    freeBytes.acquire(iLen);

    g_mutex.lock();
    QByteArray arr(szData, iLen);
    for(auto it = arr.cbegin(); it != arr.cend(); ++it)
    {
        g_szBuffer[writeNum % BUFFERSIZE] = *it;
        ++writeNum;
    }
	usedBytes.release(iLen);
	g_mutex.unlock();
}

缓冲区读取数据：

void readFromRingBuffer(QByteArray &arr, int &iLen)
{
    if(usedBytes.available() - iLen < 0)
    {
    	iLen = (usedBytes.available() == 0)? 1 : usedBytes.available(); //保证acquire的个数不为0
    }
    usedBytes.acquire(iLen);
    
    mutex.lock();
    for (int i = 0; i < iLen; ++i)
    {
        arr.push_back(g_szBuffer.at(readNum % BUFFERSIZE));
        ++readNum;
    }
    freeBytes.release(iLen);
    mutex.unlock();
}

调用信号量void QSemaphore::acquire(int n = 1)函数，如果n > available()，则线程将一直阻塞直到该信号量获得足够资源。

Qt的信号量使用详见我的这篇博客：Qt学习笔记：多线程的使用

参考资料

Wiki - Circular buffer

环形缓冲区C语言实现

STM32进阶之串口环形缓冲区实现

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)