suricata 源码分析之ringbuffer（环形缓冲区）-白红宇

suricata 源码分析之ringbuffer（环形缓冲区）

阅读量：2791 次

发布时间：2019-05-13

本文共 8611 字，大约阅读时间需要 28 分钟。

环形缓冲区也就是ringbuffer,其性能相对于异步队列是非常高效的，原因： 1)无锁; 2)cpu cache 友好;3) 内存不会疯涨。

当然ringbuffer也有其缺点(也是优点，不会出现内存暴增)，就是其长度是固定的，如果满了无法再往里面放入数据，而异步队列没有这个限制，但是有可能因为并发不够，处理不过来，队列太长导致内存被大量占用，程序性能越来越差。

suricata是多线程架构的，所以线程间的数据传递是必不可少的。因此在suricata-0.8中选用的数据结构是异步队列PacketQueue，

而在suricata-2.0.8中使用的是ringbuffer数据结构。在后期的所有版本中都淘汰了异步队列PacketQueue，使用ringbuffer或者

使用函数 pthread_key_create() 用来创建线程私有数据，其性能是高于异步队列的。

为什么ringbuffer的性能高于异步队列？

ringbuffer的锁粒度是单个节点，并且在生产者的情况下才会加锁（往ringbuffer中放节点），而异步队列是整个队列，并发远远大于异步队列。

如果只有一个消费者生产者，则ringbuffer不需要加锁，性能更高。

二者在suricata中的数据包使用

2.0.8中

void TmqhPacketpoolRegister (void) {    tmqh_table[TMQH_PACKETPOOL].name = "packetpool";    tmqh_table[TMQH_PACKETPOOL].InHandler = TmqhInputPacketpool;    tmqh_table[TMQH_PACKETPOOL].OutHandler = TmqhOutputPacketpool;    ringbuffer = RingBufferInit();    if (ringbuffer == NULL) {        SCLogError(SC_ERR_FATAL, "Error registering Packet pool handler (at ring buffer init)");        exit(EXIT_FAILURE);    }}

0.8中

void TmqhPacketpoolRegister (void) {    tmqh_table[TMQH_PACKETPOOL].name = "packetpool";    tmqh_table[TMQH_PACKETPOOL].InHandler = TmqhInputPacketpool;    tmqh_table[TMQH_PACKETPOOL].OutHandler = TmqhOutputPacketpool;}

suricata中关于ringbuffer的定义与实现在util-ringbuffer.h和util-ringbuffer.c中。

以下就以一个生产者和一个消费者的代码实现

/* 该环形缓冲区的大小 256 （ unsigned char 的大小） */

#define RING_BUFFER_8_SIZE 256typedef struct RingBuffer8_ {    /* write 和 read 都是 unsigned char 类型，巧妙地利用其值溢出效果来循环, 使用的是原子变量，来实现无锁*/    SC_ATOMIC_DECLARE(unsigned char, write);  /**< idx where we put data */    SC_ATOMIC_DECLARE(unsigned char, read);   /**< idx where we read data */    uint8_t shutdown;#ifdef RINGBUFFER_MUTEX_WAIT    SCCondT wait_cond;    SCMutex wait_mutex;#endif /* RINGBUFFER_MUTEX_WAIT */    SCSpinlock spin; /**< lock protecting writes for multi writer mode*/    /* 环形缓冲区里面存储实际的数据，可以是任何类型 */    void *array[RING_BUFFER_8_SIZE];} RingBuffer8;

其使用接口：

RingBuffer8Init：初始化，创建RingBuffer8，将读写计数清零，初始化自旋锁。

RingBufferSrSw8Put：生产者，先判断缓冲区是否已满，若满，循环等待直到有空间存储然后存储。

RingBufferSrMw8Get：消费者，先判断缓冲区是否为空，若空，循环等待直到有数据然后取走。

RingBuffer8Destroy：销毁缓冲区。

下面针对实际代码进行解释：

/* 生产者 */int RingBufferSrSw8Put(RingBuffer8 *rb, void *ptr){    /* buffer is full, wait... */    /* 判断该缓冲区是否已满，如果是，循环等待。这里判断的依据就是 read write的值会不停的从0-255循环 */    while ((unsigned char)(SC_ATOMIC_GET(rb->write) + 1) == SC_ATOMIC_GET(rb->read)) {        /* break out if the engine wants to shutdown */        if (rb->shutdown != 0)            return -1;        RingBuffer8DoWait(rb);    }    /* 预先将空间给生产者使用，无锁 */    rb->array[SC_ATOMIC_GET(rb->write)] = ptr;    (void) SC_ATOMIC_ADD(rb->write, 1);#ifdef RINGBUFFER_MUTEX_WAIT    SCCondSignal(&rb->wait_cond);#endif    return 0;}

/* 消费者 */ void *RingBufferSrMw8Get(RingBuffer8 *rb){    void *ptr = NULL;    /* buffer is empty, wait... */    /* 判断该缓冲区是否为空，如果是，循环等待。这里判断的依据就是 read write的值会不停的从0-255循环 */    while (SC_ATOMIC_GET(rb->write) == SC_ATOMIC_GET(rb->read)) {        /* break out if the engine wants to shutdown */        if (rb->shutdown != 0)            return NULL;        RingBuffer8DoWait(rb);    }    /* 如果有数据可读， 直接读取，然后移动写指针 */    ptr = rb->array[SC_ATOMIC_GET(rb->read)];    (void) SC_ATOMIC_ADD(rb->read, 1);#ifdef RINGBUFFER_MUTEX_WAIT    SCCondSignal(&rb->wait_cond);#endif    return ptr;}

以上代码是256大小的环形缓冲区，，没有加锁的地方，由此可见，其使用过程中，如果缓冲区未满，在多线程并发访问下（读写线程只有一个），不会发生阻塞，效率高

下面再介绍0.8版本的PacketQueue packet_q：

typedef struct PacketQueue_ {    Packet *top;    Packet *bot;    uint16_t len;    /* 队列锁*/    SCMutex mutex_q;    SCCondT cond_q;#ifdef DBG_PERF    uint16_t dbg_maxlen;#endif /* DBG_PERF */} PacketQueue;

初始化过程在main函数中，在程序启动的时候预先分配MAX_PENDING个数据包

int i = 0;    for (i = 0; i < MAX_PENDING; i++) {        /* XXX pkt alloc function */        Packet *p = malloc(sizeof(Packet));        if (p == NULL) {            printf("ERROR: malloc failed: %s\n", strerror(errno));            exit(EXIT_FAILURE);        }        memset(p, 0, sizeof(Packet));        SCMutexInit(&p->mutex_rtv_cnt, NULL);        PacketEnqueue(&packet_q,p);    }

/* 从 packet_q队列中取出数据包*/Packet *SetupPkt (void){    Packet *p = NULL;    int r = 0;    /* 此处对packet_q整个队列加锁，其他地方不能访问*/    r = SCMutexLock(&packet_q.mutex_q);    p = PacketDequeue(&packet_q);    r = SCMutexUnlock(&packet_q.mutex_q);    if (p == NULL) {        TmqDebugList();        p = malloc(sizeof(Packet));        if (p == NULL) {            printf("ERROR: malloc failed: %s\n", strerror(errno));            exit(EXIT_FAILURE);        }        memset(p, 0, sizeof(Packet));        r = SCMutexInit(&p->mutex_rtv_cnt, NULL);        SCLogDebug("allocated a new packet...");    }    /* reset the packet csum fields */    RESET_PACKET_CSUMS(p);    return p;}

/*生产者在调用SetupPkt 将对packet_q整个异步队列进行加锁*/

Packet *TmqhInputPacketpool(ThreadVars *t){    /* XXX */    Packet *p = SetupPkt();    SCMutexLock(&mutex_pending);    pending++;    //printf("PcapFileCallback: pending %" PRIu32 "\n", pending);#ifdef DBG_PERF    if (pending > dbg_maxpending)        dbg_maxpending = pending;#endif /* DBG_PERF */    SCMutexUnlock(&mutex_pending);/* * Disabled because it can enter a 'wait' state, while * keeping the nfq queue locked thus making it impossble * to free packets, the exact condition we are waiting * for. VJ 09-01-16 *    SCMutexLock(&mutex_pending);    if (pending > MAX_PENDING) {        SCondWait(&cond_pending, &mutex_pending);    }    SCMutexUnlock(&mutex_pending);*/    return p;}

/* 消费者 */

void TmqhOutputPacketpool(ThreadVars *t, Packet *p){    PacketQueue *q = &packet_q;    char proot = 0;    if (p == NULL)        return;    if (IS_TUNNEL_PKT(p)) {        //printf("TmqhOutputPacketpool: tunnel packet: %p %s\n", p,p->root ? "upper layer":"root");        /* get a lock */        SCMutex *m = p->root ? &p->root->mutex_rtv_cnt : &p->mutex_rtv_cnt;        SCMutexLock(m);        if (IS_TUNNEL_ROOT_PKT(p)) {            //printf("TmqhOutputPacketpool: IS_TUNNEL_ROOT_PKT\n");            if (TUNNEL_PKT_TPR(p) == 0) {                //printf("TmqhOutputPacketpool: TUNNEL_PKT_TPR(p) == 0\n");                /* if this packet is the root and there are no                 * more tunnel packets, enqueue it */                /* fall through */            } else {                //printf("TmqhOutputPacketpool: TUNNEL_PKT_TPR(p) > 0\n");                /* if this is the root and there are more tunnel                 * packets, don't add this. It's still referenced                 * by the tunnel packets, and we will enqueue it                 * when we handle them */                p->tunnel_verdicted = 1;                SCMutexUnlock(m);                return;            }        } else {            //printf("TmqhOutputPacketpool: NOT IS_TUNNEL_ROOT_PKT\n");            if (p->root->tunnel_verdicted == 1 && TUNNEL_PKT_TPR(p) == 1) {                //printf("TmqhOutputPacketpool: p->root->tunnel_verdicted == 1 && TUNNEL_PKT_TPR(p) == 1\n");                /* the root is ready and we are the last tunnel packet,                 * lets enqueue them both. */                TUNNEL_DECR_PKT_TPR_NOLOCK(p);                /* handle the root */                //printf("TmqhOutputPacketpool: calling PacketEnqueue for root pkt, p->root %p (%p)\n", p->root, p);                proot = 1;                /* fall through */            } else {                //printf("TmqhOutputPacketpool: NOT p->root->tunnel_verdicted == 1 && TUNNEL_PKT_TPR(p) == 1 (%" PRIu32 ")\n", TUNNEL_PKT_TPR(p));                TUNNEL_DECR_PKT_TPR_NOLOCK(p);                 /* fall through */            }        }        SCMutexUnlock(m);        //printf("TmqhOutputPacketpool: tunnel stuff done, move on\n");    }    FlowDecrUsecnt(t,p);    if (proot && p->root != NULL) {        CLEAR_PACKET(p->root);        SCMutexLock(&q->mutex_q);        PacketEnqueue(q, p->root);        SCMutexUnlock(&q->mutex_q);    }    CLEAR_PACKET(p);

/* 对整个异步队列加锁 导致其他线程不能操作，并发下降*/    SCMutexLock(&q->mutex_q);    PacketEnqueue(q, p);    SCMutexUnlock(&q->mutex_q);    SCMutexLock(&mutex_pending);    //printf("TmqhOutputPacketpool: pending %" PRIu32 "\n", pending);    if (pending > 0) {        pending--;        if (proot) {            if (pending > 0) {                pending--;            } else {                printf("TmqhOutputPacketpool: warning, trying to subtract from 0 pending counter (tunnel root).\n");            }        }    } else {        printf("TmqhOutputPacketpool: warning, trying to subtract from 0 pending counter.\n");    }    if (pending <= MAX_PENDING)        SCCondSignal(&cond_pending);    SCMutexUnlock(&mutex_pending);}

suricata 使用 unsigned char 和unsigned short 的大小来定义环形缓冲区的大小，非常巧妙。

以上就是本人对环形缓冲区与异步队列的理解、性能分析、对比，并未对环形缓冲区原理进行介绍。

其定义可以参考以下链接：

转载地址：http://pvtmd.baihongyu.com/

你可能感兴趣的文章

Xcode5&4.6.3 + iOS 7&6免证书开发+真机调试+生成ipa全攻略