Linux下另类内存缓冲池设计(CBufferPool)

 

        本文所说的缓冲池的构建与一般的开辟一个大的内存,需要申请内存时就直接在该大内存中再划分出小内存出来使用不同,CBufferPool使用的是队列链表,程序按出队/入队的方式对内存进行读写.

设计该类的目的是因为在需求中,我有一个写入线程和一个读取线程,读取线程会挨个读取写入线程写入的数据,于是想到使用队列,按先入先出的方式可以顺序读取已写入的数据,由于考虑到可能需要动态增加队列的大小,故采用了环形链表形式的队列.第一个问题解决,即决定了所要使用的数据结构--环形队列链表.

 

        第二个考虑的问题是,在使用时不使用什么GetReadBuffer/PutWriteBuffer之类的显式调用读写队列的方法,我只需要通过相同的方法如本设计所使用的GetOut/PutBack方法就可以取到读或写缓冲,仅需在参数中指定我需要获取到的是读还是写缓冲就行了,这样可以很好地屏蔽二者在调用上的差异.所以,我设计了UserBuffer结构体,并在其中添加属性type,用于标识该buffer是用于读的还是用于写的.同时,在使用GetOut方法获取到要使用的buffer后,UserBuffer中的id将被赋值为与缓冲池中对应的PoolBuffer的id相同的值,这样在使用PutBack放回该buffer时就能区分该buffer是否是CBufferPool所返回的缓冲,防止可能出现的非法操作,如放回不属于CBufferPool所返回的缓冲等.

 

        第三个问题是考虑到缓冲池中使用的数据结构应该和用户使用的数据结构是类似的,只是在缓冲池中需要有指向下一个缓冲的指针,并且缓冲池内部的数据结构对使用者是不开放的,使用者不能看到或直接使用该数据,故而,最终设计了用户使用类UserBuffer和缓冲池内部类PoolBuffer.从代码中可以看到两个数据结构基本上是相同的,不同的除了PoolBuffer中的next指针以外,就是UserBuffer中用于标识该缓冲的使用或请求类型的type和PoolBuffer中标识该缓冲处于读还是写状态的status.也就是,当用户标识UserBuffer的type为POOL_BUFFER_TYPE_READ时,通过方法GetOut方法缓冲池就会返回队列首部的可读的缓冲,同时该缓冲将被标记状态为POOL_BUFFER_STATUS_READING,防止该缓冲被写入线程使用;当UserBuffer的type为POOL_BUFFER_TYPE_WRITE时,通过GetOut方法可以得到缓冲池中可写的缓冲,同样地该缓冲的状态将被标记为POOL_BUFFER_STATUS_WRITING,防止读线程使用.

 

        第四个问题是,本设计在获取写缓冲时会根据请求的缓冲大小动态地开辟合适的内存空间,当所请求的大小和已经存在的缓冲大小不相同时便会重新开辟新的内存空间,但是出现的问题是当请求的缓冲大小和已经存在的缓冲大小相差不是很大,比如就在1KB以内,那如果是比较两者大小不相等就开辟的话,这个效率上就会很差了,所以,在设计中,采用了按片(Slice)开辟缓冲的方法,在比较时采用了比较两者所占用的片的个数是否相等来判断是否要重新开辟内存的方法,这在一定程度上减少了重复开辟新内存的消耗.不过本版本还未提供修改片大小的方法,加上也不麻烦,所以就不加了.^_^

 

        于是,经过一段时间的思考和实验,最终完成了该设计,并贴出源代码,供有需要的盆友参考

 

        本设计还存在许多不足,比如该版本只考虑了单写入线程/单读取线程,没有考虑多个写入线程/多读取线程的情况,只能说是满足了基本的要求,待改进部分还是很多的,有机会和需求时再做改进吧.

buffer.h:

/** 
 *  buffer.h 
 *  
 *  Copyright (C) 2011 flytreeleft([e-mail]). 
 * 
 * This file is placed under the LGPL.  
 * 
 * If you have some questions or advises, please email me! 
 *                        Thank you! 
 */  
  
  
#ifndef __BUFFER_H_  
#define __BUFFER_H_  
  
typedef unsigned int uint32_t;  
  
typedef struct UserBuffer {  
    // the buffer pool will use the id to decide  
    // whether this buffer belongs to it or not  
    uint32_t id;  
    // using type -- to read or to write  
    int type;  
    // the size of buffer which is got  
    // or needed to set  
    uint32_t size;  
    void *start; // address of buffer  
} Buffer;  
  
#endif  

 bufferpool.h:

/** 
 *  bufferpool.h 
 * 
 *  Copyright (C) 2011 flytreeleft([e-mail]). 
 * 
 * This file is placed under the LGPL.  
 * 
 * If you have some questions or advises, please email me! 
 *                        Thank you! 
 */  
  
  
#ifndef __BUFFER_POOL_H_  
#define __BUFFER_POOL_H_  
  
#include <pthread.h>  
#include "buffer.h"  
  
#define MIN_POOL_BUFFER_COUNT   (1)  
#define POOL_BUFFER_INVALID_ID  (0)  
enum {  
    POOL_BUFFER_TYPE_UNKNOWN,  
    POOL_BUFFER_TYPE_READ,  
    POOL_BUFFER_TYPE_WRITE  
};  
enum {  
    POOL_BUFFER_STATUS_UNUSED = 0,  
    POOL_BUFFER_STATUS_READING,  
    POOL_BUFFER_STATUS_WRITING  
};  
  
class CBufferPool {  
    private:  
        typedef struct _PoolBuffer {  
            // every buffer has a different id in pool  
            uint32_t id;  
            // using status of buffer  
            int status;  
            // address of buffer  
            void *start;  
            uint32_t size;  
            struct _PoolBuffer *next;  
        } PoolBuffer;  
  
        uint32_t id_seed;  
        PoolBuffer *input;  
        PoolBuffer *output;  
        PoolBuffer *head;  
        uint32_t slice_size; // size per slice  
        pthread_mutex_t rw_mutex;  
        int size;  
    public:  
        CBufferPool(void);  
        CBufferPool(int count);  
  
        void SetSliceSize(uint32_t size);  
        int GetSize(void) { return size; }  
  
        bool Increase(int increase);  
        bool Decrease(int decrease);  
        void Create(int count);  
        void Clear(void);  
  
        bool GetOut(Buffer *buffer);  
        bool PutBack(Buffer *buffer);  
  
        bool IsFull(void);  
        bool IsEmpty(void);  
  
        void Print(void);  
  
        ~CBufferPool(void);  
    private:  
        void InitData(void);  
        bool IsClear(void);  
        bool GetReadBuffer(Buffer *reader);  
        bool GetWriteBuffer(Buffer *writer);  
        bool PutReadBuffer(Buffer *reader);  
        bool PutWriteBuffer(Buffer *writer);  
        uint32_t GenerateId(void) {  
            return ++id_seed;  
        }  
};  
  
#endif  

 bufferpool.cpp:

/** 
 *  bufferpool.cpp 
 * 
 *  Copyright (C) 2011 flytreeleft([e-mail]). 
 * 
 * This file is placed under the LGPL. 
 * 
 * If you have some questions or advises, please email me! 
 *                        Thank you! 
 */  
  
  
#include <stdio.h>  
#include <stdlib.h>  
#include <unistd.h>  
#include <string.h>  
  
#include "bufferpool.h"  
#include "zero.h"  
  
CBufferPool::CBufferPool(void) {  
    InitData();  
}  
  
CBufferPool::CBufferPool(int count) {  
    InitData();  
    Create(count);  
}  
  
void CBufferPool::InitData(void) {  
    head = NULL;  
    input = output = head;  
    size = 0;  
    slice_size = 1024; // 1KB  
  
    pthread_mutex_init(&rw_mutex, NULL);  
}  
  
void CBufferPool::SetSliceSize(uint32_t size) {  
    slice_size = (size > 0 ? size : 1);  
}  
  
/** 
 * create additional buffers 
 */  
bool CBufferPool::Increase(int increase) {  
    PoolBuffer *tmp = NULL;  
    bool succ = true;  
  
    pthread_mutex_lock(&rw_mutex);  
  
    if (!IsClear()) {  
        while (increase > 0) {  
            tmp = SAFE_MALLOC(PoolBuffer,   
                                sizeof(PoolBuffer));  
  
            if (tmp != NULL) {  
                ZERO(tmp);  
                tmp->id = GenerateId();  
  
                size++;  
                increase--;  
                  
                tmp->next = input->next;  
                input->next = tmp;  
            } else {  
                printf("no enough space,"\  
                        "when increasing pool:"\  
                        "last size is %d\n",  
                            size);  
                break;  
            }  
        }  
    } else {  
        printf("Please create pool first!\n");  
    }  
  
    pthread_mutex_unlock(&rw_mutex);  
  
    return succ;  
}  
  
bool CBufferPool::Decrease(int decrease) {  
    PoolBuffer *next = NULL;  
    PoolBuffer *tmp = NULL;  
    bool succ = true;  
  
    pthread_mutex_lock(&rw_mutex);  
  
    if (!IsClear()) {  
        next = input->next;  
  
        while (next != output   
                    && decrease > 0   
                && size > MIN_POOL_BUFFER_COUNT) {  
            tmp = next->next;  
  
            SAFE_FREE(next->start);  
            SAFE_FREE(next);  
  
            next = tmp;  
  
            size--;  
            decrease--;  
        }  
  
        input->next = next;  
    } else {  
        printf("There is no buffer in pool!\n");  
    }  
  
    pthread_mutex_unlock(&rw_mutex);  
  
    return succ;  
}  
  
/** 
 * get out buffer to read or write, according to buffer's type 
 */  
bool CBufferPool::GetOut(Buffer *buffer) {  
    bool succ = false;  
      
    pthread_mutex_lock(&rw_mutex);  
  
    if (buffer != NULL) {  
        switch (buffer->type) {  
            case POOL_BUFFER_TYPE_READ:  
                succ = GetReadBuffer(buffer);  
                break;  
            case POOL_BUFFER_TYPE_WRITE:  
                succ = GetWriteBuffer(buffer);  
                break;  
        }  
    }  
  
    pthread_mutex_unlock(&rw_mutex);  
  
    return succ;  
}  
  
bool CBufferPool::PutBack(Buffer *buffer) {  
    bool succ = false;  
      
    pthread_mutex_lock(&rw_mutex);  
  
    if (buffer != NULL) {  
        switch (buffer->type) {  
            case POOL_BUFFER_TYPE_READ:  
                succ = PutReadBuffer(buffer);  
                break;  
            case POOL_BUFFER_TYPE_WRITE:  
                succ = PutWriteBuffer(buffer);  
                break;  
        }  
    }  
  
    buffer->id = POOL_BUFFER_INVALID_ID;  
    buffer->size = 0;  
    buffer->start = NULL;  
  
    pthread_mutex_unlock(&rw_mutex);  
  
    return succ;  
}  
  
bool CBufferPool::GetReadBuffer(Buffer *reader) {  
    bool succ = false;  
  
    if (!IsClear()  
            && !IsEmpty()   
            && output->status != POOL_BUFFER_STATUS_WRITING) {  
        reader->id = output->id;  
        reader->size = output->size;  
        reader->start = output->start;  
  
        output->status = POOL_BUFFER_STATUS_READING;  
        succ = true;  
    }  
  
    return succ;  
}  
  
bool CBufferPool::GetWriteBuffer(Buffer *writer) {  
    bool succ = false;  
    uint32_t o, n;  
  
    if (!IsClear()  
            && !IsFull()   
            && input->status != POOL_BUFFER_STATUS_READING   
            && writer->size > 0) {  
        o = (input->size / slice_size)   
                + (input->size % slice_size > 0 ? 1 : 0);  
        n = (writer->size / slice_size)   
                + (writer->size % slice_size > 0 ? 1 : 0);  
  
        if (o != n) {  
            SAFE_FREE(input->start);  
            input->start = SAFE_MALLOC(void, n * slice_size);  
        }  
        input->size = writer->size;  
  
        writer->start = input->start;  
        writer->id = input->id;  
  
        input->status = POOL_BUFFER_STATUS_WRITING;  
        succ = true;  
    }  
  
    return succ;  
}  
  
bool CBufferPool::PutReadBuffer(Buffer *reader) {  
    bool succ = false;  
      
    if (!IsClear()   
            && reader->id == output->id   
            && output->status == POOL_BUFFER_STATUS_READING) {  
        output->status = POOL_BUFFER_STATUS_UNUSED;  
        output = output->next;  
        succ = true;  
    }  
  
    return succ;  
}  
  
bool CBufferPool::PutWriteBuffer(Buffer *writer) {  
    bool succ = false;  
  
    if (!IsClear()   
            && writer->id == input->id   
            && input->status == POOL_BUFFER_STATUS_WRITING) {  
        input->status = POOL_BUFFER_STATUS_UNUSED;  
        input = input->next;  
        succ = true;  
    }  
  
    return succ;  
}  
  
/** 
 * when input pointer's next and output are equal, 
 * the pool is full 
 */  
bool CBufferPool::IsFull(void) {  
    return (input != NULL ? input->next == output : false);  
}  
  
/** 
 * when output pointer and input pointer are equal, 
 * the pool is empty 
 */  
bool CBufferPool::IsEmpty(void) {  
    return (output == input);  
}  
  
/** 
 * create buffers in pool. 
 */  
void CBufferPool::Create(int count) {  
    PoolBuffer *tmp = NULL;  
      
    pthread_mutex_lock(&rw_mutex);  
  
    if (count < MIN_POOL_BUFFER_COUNT) {  
        count = MIN_POOL_BUFFER_COUNT;  
    }  
  
    if (!IsClear()) {  
        printf("Please clear the pool first!\n");  
        return;  
    }  
  
    while (count >= 0) { // i need to create one more buffer  
        tmp = SAFE_MALLOC(PoolBuffer, sizeof(PoolBuffer));  
              
        if (tmp != NULL) {  
            ZERO(tmp);  
            tmp->id = GenerateId();  
  
            size++; count--;  
  
            if (head != NULL) {  
                tmp->next = head->next;  
                head->next = tmp;  
            } else {  
                head = tmp;  
                head->next = tmp;  
            }  
        } else {  
            printf("no enough space,"\  
                    "when creating buffer pool:"\  
                    "last size is %d\n", size);  
            break;  
        }  
    }  
    // hide one buffer  
    size = (size > 0 ? size - 1 : 0);  
    input = output = head;  
  
    pthread_mutex_unlock(&rw_mutex);  
}  
  
/** 
 * free buffer from head to tail, one by one 
 */  
void CBufferPool::Clear(void) {  
    PoolBuffer *next = head;  
    PoolBuffer *tmp = NULL;  
  
    pthread_mutex_lock(&rw_mutex);  
  
    if (!IsClear()) {  
        do {  
            tmp = next->next;  
  
            SAFE_FREE(next->start);  
            SAFE_FREE(next);  
  
            next = tmp;  
        } while(next != head);  
    }  
  
    size = 0;  
    head = NULL;  
    input = output = head;  
  
    pthread_mutex_unlock(&rw_mutex);  
}  
  
/** 
 * print all buffers' information 
 */  
void CBufferPool::Print(void) {  
    PoolBuffer *h = head;  
      
    if (!IsClear()) {  
        do {  
            printf("----------------------\n");  
            printf("| id = %3d,size = %3d|\n", h->id, h->size);  
            printf("----------------------\n");  
            h = h->next;  
        } while(h != head);  
    } else {  
        printf("there is no buffer in pool...\n");  
    }  
}  
  
/** 
 * is there no buffer in pool? 
 */  
bool CBufferPool::IsClear(void) {  
    bool clear = true;  
  
    if (size >= MIN_POOL_BUFFER_COUNT  
            && head != NULL   
            && input != NULL   
            && output != NULL) {  
        clear = false;  
    }  
  
    return clear;  
}  
  
CBufferPool::~CBufferPool(void) {  
    Clear();  
}  

 zero.h:

 

/** 
 *  zero.h 
 * 
 *  Copyright (C) 2011 flytreeleft([e-mail]). 
 * 
 * This file is placed under the LGPL.  
 * 
 * If you have some questions or advises, please email me! 
 *                        Thank you! 
 */  
  
  
#ifndef __ZERO_H_  
#define __ZERO_H_  
  
#include <stdlib.h>  
#include <string.h>  
  
#define ZERO(obj) memset(obj, 0, sizeof(*obj))  
#define ZERO_ARRAY(array,count) memset(array, 0, (count) * sizeof(array[0]))  
#define SAFE_MALLOC(type,size) (type *)safe_malloc(size)  
#define SAFE_FREE(p) safe_free(p)  
  
static inline void *safe_malloc(int size) {  
    return size > 0 ? malloc(size) : NULL;  
}  
  
static inline void safe_free(void *p) {  
    if (p != NULL) {  
        free(p);  
    }  
}  
  
#endif 

 

       加上测试代码:

pooltest.cpp:

 

/** 
 *  pooltest.cpp 
 * 
 *  Copyright (C) 2011 flytreeleft([e-mail]). 
 * 
 * This file is placed under the LGPL.  
 * 
 * If you have some questions or advises, please email me! 
 *                        Thank you! 
 */  
  
  
#include <stdio.h>  
#include <stdlib.h>  
#include <string.h>  
  
#include "bufferpool.h"  
  
#define POOL_SIZE   (1)  
#define TEXT_SIZE   (10)  
  
int main(void) {  
    Buffer buffer;  
    CBufferPool pool;  
    int i;  
    char text[TEXT_SIZE];  
  
    // increase of decrease befor create  
    pool.Increase(2);  
    pool.Decrease(2);  
    // create new  
    pool.Create(POOL_SIZE);  
    pool.Print();  
    // create befor clear  
    pool.Create(POOL_SIZE);  
  
    printf("writing...\n");  
    // write buffer in pool  
    for (i = 0; i < pool.GetSize(); i++) {  
        // initialize  
        buffer.type = POOL_BUFFER_TYPE_WRITE;  
        buffer.size = TEXT_SIZE;  
  
        // get input buffer from pool  
        if (pool.GetOut(&buffer)) {  
            printf("input : id %d\n", buffer.id);  
            // write buffer  
            snprintf(text, TEXT_SIZE, "test %d", i);  
            memcpy(buffer.start, text, TEXT_SIZE);  
            printf("\ttext : %s\n", text);  
            // put buffer to pool  
            pool.PutBack(&buffer);  
        }  
    }  
  
    pool.Print();  
  
    printf("reading...\n");  
    // read buffer in pool  
    for (i = 0; i < pool.GetSize(); i++) {  
        // initialize  
        buffer.type = POOL_BUFFER_TYPE_READ;  
  
        // get output buffer from pool  
        if (pool.GetOut(&buffer)) {  
            printf("output : id %d, size %d\n", buffer.id, buffer.size);  
            // read buffer  
            printf("\ttext : %s\n", (char *)buffer.start);  
            // put buffer to pool  
            pool.PutBack(&buffer);  
        }  
    }  
  
    printf("increase buffer...\n");  
    pool.Increase(2);  
    pool.Print();  
    printf("decrease buffer...\n");  
    pool.Decrease(2);  
    pool.Print();  
  
    return 0;  
}  

 
        编译代码:

 

$ gcc -o pooltest pooltest.cpp bufferpool.cpp