C环形队列

环形队列是什么

队列是一种常用的数据结构，这种结构保证了数据是按照“先进先出”的原则进行操作的，即最先进去的元素也是最先出来的元素.环形队列是一种特殊的队列结构，保证了元素也是先进先出的，但与一般队列的区别是，他们是环形的，即队列头部的上个元素是队列尾部，通常是容纳元素数固定的一个闭环。

环形队列的工作场景

一般应用于需要高效且频繁进行多线程通信传递数据的场景，例如：linux捕包、发包等等，（linux系统中对PACKET_RX_RING和PACKET_TX_RING的支持实质就是内核实现的一种环形队列）

#include <assert.h>
#include <string.h>

typedef unsigned char u_char;

#define CAN_WRITE 0x00
#define CAN_READ 0x01
#define READING 0x02
#define WRITING 0x03

typedef struct tag
{
    u_char tag_value;
}TAG;


class Ring_Queue
{
    public:
        Ring_Queue(int nmemb,int size):_nmemb(nmemb),_size(size)
                                         ,_read_now(0),_write_now(0)
    {
        if ( nmemb <= 0 || size <=0 )
        {
            assert(0);
        }
        _queue_p = NULL;
        _queue_p = new u_char[ nmemb * (sizeof(TAG) + size)];
        memset(_queue_p,0,nmemb * (sizeof(TAG) + size));

    }
        ~Ring_Queue()
        {
            if (_queue_p) delete []_queue_p;

        }
        u_char * SOLO_Read()
        {
            u_char * g_p = 0;
            TAG * tag_p = 0;
            u_char *user_data = 0;

            g_p = queue_peek_nth(_queue_p,_read_now);
            tag_p = (TAG *)g_p;
            if (tag_p->tag_value == CAN_READ)
            {
                user_data = (u_char *)g_p + sizeof(TAG);
                tag_p->tag_value = READING;
            }
            return user_data;
        }
        void SOLO_Read_Over()
        {
            u_char * g_p = 0;
            TAG * tag_p = 0;

            g_p = queue_peek_nth(_queue_p,_read_now);
            tag_p = (TAG *)g_p;
            if (tag_p->tag_value == READING)
            {
                tag_p->tag_value = CAN_WRITE;
                _read_now = (_read_now + 1)% _nmemb;
            }
        }
        u_char * SOLO_Write()
        {
            u_char * g_p = 0;
            TAG * tag_p = 0;
            u_char *user_data = 0;

            g_p = queue_peek_nth(_queue_p,_write_now);
            tag_p = (TAG *)g_p;
            if (tag_p->tag_value == CAN_WRITE)
            {  
                user_data = (u_char *)g_p + sizeof(TAG);
                tag_p->tag_value = WRITING;
            }                
            return user_data;
        }
        void SOLO_Write_Over()
        {
            u_char * g_p = 0;
            TAG * tag_p = 0;

            g_p = queue_peek_nth(_queue_p,_write_now);
            tag_p = (TAG *)g_p;
            if (tag_p->tag_value == WRITING)
            {
                tag_p->tag_value = CAN_READ;
                _write_now = (_write_now + 1)% _nmemb;
            }
        }
    private:
        u_char *queue_peek_nth(u_char *queue_p,int pos)
        {
            u_char *rst = 0;
            if (queue_p && pos < _nmemb)
            {
                rst = queue_p + pos * (sizeof(TAG) + _size);
            }
            return rst;
        }
        u_char * _queue_p;
        int _nmemb;
        int _size;
        volatile int _read_now;
        volatile int _write_now;
};

#include <stdio.h>
#include "ring_queue.h" 
#include <unistd.h>
#include <sys/time.h>

const int LOOP_SIZE = 10000;

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>


#define THREAD_NUM 1

#define DATA_TYPE int

void *customer(void *arg)
{
    Ring_Queue queue = *(Ring_Queue *)arg;

    while(1)
    {
        for(int i = 0;i < LOOP_SIZE; )
        {
            int *p = 0;
            p = (DATA_TYPE *)queue.SOLO_Read();
            if (p)
            {
                assert(i == *p);
//              printf("[%d]:%d\n",i,*p);
                queue.SOLO_Read_Over();
                i++;
            }
        }
    }
}

void *producer(void *arg)
{
    Ring_Queue queue = *(Ring_Queue *)arg;
    
    int loop = 0;
    struct timeval last_time,now_time;
    gettimeofday(&last_time,NULL);
    gettimeofday(&now_time,NULL);
    while(1)
    {
        for(int i = 0;i < LOOP_SIZE; )
        {
            int *p = 0;
            p = (DATA_TYPE *)queue.SOLO_Write();
            if (p)
            {
                *p = i;
                queue.SOLO_Write_Over();
                i++;
            }
        }
        gettimeofday(&now_time,NULL);
        if (now_time.tv_sec - last_time.tv_sec >= 5)
        {
            printf("LOOP_COUNT is %.2f=[ %d[RING_SIZE] * %.2f[RING_COUNT]] per second\n",(LOOP_SIZE*loop)/5.0,LOOP_SIZE,loop/5.0);
            last_time = now_time;
            loop = 0;
        }
        loop++;
    }
}

int main(int argc,char *argv[])
{
    pthread_t tid_customer[THREAD_NUM];
    pthread_t tid_producer[THREAD_NUM];
    Ring_Queue *queue = new Ring_Queue[THREAD_NUM](LOOP_SIZE,sizeof(DATA_TYPE));

    for (int i = 0; i < THREAD_NUM; i++)
    {
        if (pthread_create(&tid_customer[i],NULL,&customer,(void*)&queue[i]) != 0)
        {
            fprintf(stderr,"thread create failed\n");
            return -1;
        }
    }
    for (int i = 0; i < THREAD_NUM; i++)
    {
        if (pthread_create(&tid_producer[i],NULL,&producer,(void*)&queue[i]) != 0)
        {
            fprintf(stderr,"thread create failed\n");
            return -1;
        }
    }

    for (int i = 0 ;i < THREAD_NUM; i++)
        pthread_join(tid_customer[i],NULL);
    
    for (int i = 0 ;i < THREAD_NUM; i++)
        pthread_join(tid_producer[i],NULL);

}