【转】缓冲区设计--环形队列

原文链接：http://blog.csdn.net/billow_zhang/article/details/4420789

在程序的两个模块间进行通讯的时候，缓冲区成为一个经常使用的机制。

如上图，写入模块将信息写入缓冲区中，读出模块将信息读出缓冲区。这样使得：

• 将程序清晰地划分模块，建立良好的模块化架构，使得写入和读出成为高聚合，低耦合的模块。
• 对于写入和读出的处理可能产生的快慢不均匀的情况进行平衡，使得整个处理的速度趋于平滑的均匀状态，避免出现读出模块处理的慢速使得写入模块等待使得响应速度下降的状况；同样，也避免写入模块的快慢不均匀，使得读出模块忙闲不一的情况。
• 可以增加处理的并发性。由于写入和读出模块的良好设计和划分，可以使得它们彼此隔离和独立，从而，使用线程和进程产生不同的并发处理，并通过缓冲区大小的调节，使得这个处理达到良好的匹配和运行状态。例如，写入模块可以有N个线程或进程，读出模块可以有M个线程和进程，缓存冲区可以配置L的大小。N、M、L可以通过模拟试验设定适应具体应用的值。也可以建立一套自动调节的机制，当然，这样会造成设计的复杂性。

缓冲区显然不适合下面的情况：

• 数据的接收处理原本就是密切相关，难以划分模块。
• 处理中的模块间明显不存在处理不均匀的情况，或者不是主要问题。
• 需要同步响应的情况。显然，写入端只是将信息push到队列中，并不能得到读出端的处理响应信息，只能适合于异步的信息传递的情况。

缓冲区的设计：

• 缓冲区是一个先进先出队列。写入模块将信息插入队列；读出模块将信息弹出队列。
• 写入模块与读出模块需要进行信息的协调和同步。
• 对于多线程和多进程的写入或读出模块，写入模块间以及读出模块间需要进行临界区处理。

队列使用环形队列，如上图。环形队列的特点是，不需要进行动态的内存释放和分配，使用固定大小的内存空间反复使用。在实际的队列插入和弹出操作中， 是不断交叉进行的，当push操作时，head会增加；而pop操作时，tail会增加。push的速度快的时候，有可能追上 tail，这个时候说明队列已经满了，不能再进行push的操作了，需要等待 pop 操作腾出队列的空间。当 pop 的操作快，使得 tail 追上 head，这个时候说明队列已空了，不能再进行 pop 操作了，需要等待 push 进来数据。

下面列出了一个环形队列类的的数据结构的源程序 。

[cpp] view plaincopy

1. /* LoopQue.h
2. Author: ZhangTao
3. Date: July 26, 2009
4. */
5. # ifndef LoopQue_h
6. # define LoopQue_h
7. # include       <new>
8. namespace xtl   {
9. template<typename _Tp>
10. class LoopQue_impl  {
11. public:
12. static int addsize(int max_size) {
13. return max_size * sizeof(_Tp);
14. }
15. LoopQue_impl(int msize) : max_size(msize), _front(0), _rear(0), _size(0) {}
16. _Tp& front() { return data[_front]; }
17. void push(const _Tp& value)  {
18. data[_rear] = value;
19. _rear = (_rear + 1) % max_size;
20. _size++;
21. }
22. void pop()  {
23. _front = (_front + 1) % max_size;
24. _size--;
25. }
26. int check_pop(_Tp& tv)  {
27. if ( empty() )
28. return -1;
29. tv = front();
30. pop();
31. }
32. int check_push(const _Tp& value)  {
33. if ( full() )
34. return -1;
35. push(value);
36. }
37. bool full() const  { return _size == max_size; }
38. bool empty() const { return _size == 0; }
39. int  size() const { return _size; }
40. int  capacity() const { return max_size; }
41. private:
42. int32_t _front;  // front index
43. int32_t _rear;   // rear index
44. int32_t _size;   // queue data record number
45. const int32_t max_size; // queue capacity
46. _Tp  data[0];    // data record occupy symbol
47. };
48. template<typename _Tp>
49. struct LoopQue_allocate  {
50. LoopQue_impl<_Tp>& allocate(int msize)    {
51. char *p = new char[sizeof(LoopQue_impl<_Tp>) +
52. LoopQue_impl<_Tp>::addsize(msize)];
53. return *(new (p) LoopQue_impl<_Tp>(msize));
54. }
55. void deallocate(void *p)   {
56. delete [] (char *)p;
57. }
58. };
59. template< typename _Tp, typename Alloc = LoopQue_allocate<_Tp> >
60. class LoopQue   {
61. public:
62. typedef _Tp  value_type;
63. LoopQue(int msize) : impl(alloc.allocate(msize)) {}
64. ~LoopQue()  { alloc.deallocate((void *)&impl); }
65. value_type& front() { return impl.front(); }
66. const value_type& front() const { return impl.front; }
67. void push(const value_type& value)  { impl.push(value); }
68. void pop()  { impl.pop(); }
69. int check_pop(value_type& tv)  { return impl.check_pop(tv); }
70. int check_push(const value_type& value)  { return impl.check_push(value); }
71. bool full() const  { return impl.full(); }
72. bool empty() const { return impl.empty(); }
73. int  size() const { return impl.size(); }
74. private:
75. Alloc alloc;
76. LoopQue_impl<_Tp>& impl;
77. };
78. } // end of < namespace stl >
79. # endif  // end of <ifndef LoopQue_h>

程序里定义了两个类 LoopQue_impl及LoopQueue。前者定义了环形队列的基本数据结构和实现，后者又进行了一次内存分配包装。
21行的LoopQue_impl的构造函数是以队列的空间大小作为参数创建这个类的。也就是说，在类创建的时候，就决定了队列的大小。

63行中定义的队列的数组空间为 _Tp data[0]。这似乎是一个奇怪的事情。事实上，这个空间大小应该是max_size个数组空间。 但由于max_size是在类创建的时候确定的，在这里，data[0]只起到一个占位符的作用。所以，LoopQue_impl这个类是不能直接使用的， 需要正确的分配好内存大小，才能使用，这也是需要里另外设计一个类LoopQueue的重要原因之一。也许您会奇怪，为什么要这样使用呢？如果定义一个指针， 例如：_Tp *data，然后在构造函数里面使用 data = new _Tp[max_size]，不是很容易吗？但是，不要忘记了，我们这个环形队列类有可能会是一个进程间共享类。 例如，一个进程push操作，另一个进程pop操作。这样，这个类是需要建立在共享内存中的。而共享内存中的类的成员，如果包含有指针或者引用这样的类型， 将给内存分配带来很大的麻烦。而我们这样以这个占位符的方式设计这个类，将减少这种麻烦和复杂性。

17行的addsize的类函数确定了LoopQue_impl需要的另外的内存空间的大小。

LoopQueue显示了怎样使用LoopQue_impl，解决内存分配问题的。从79行的模版参数中，我们看到，除了缓冲区数据存放类型_Tp的参数外，还有一个Alloc类型。 这便是用于分配LoopQue_impl内存空间使用的模版类。

在LoopQueue的成员中，定义了LoopQue_impl的一个引用 impl（102行）。这个引用便是指向使用Alloc分配空间得来的LoopQue_impl的空间。

Alloc模版参数有一个缺省的定义值 LoopQue_allocate。从这个缺省的分配内存的类里，我们可以看到一个分配LoopQue_impl的实现样例，见69行：

　　 char *p = new char[sizeof(LoopQue_impl<_Tp>) + LoopQue_impl<_Tp>::addsize(msize)];
       return *(new (p) LoopQue_impl<_Tp>(msize));

这里，先根据msize分配好了靠虑到了data[msize]的足够的内存，然后，再使用定位的new操作，将LoopQue_impl创建在这个内存区中。这样，LoopQue_impl类就可以使用它其中的 _Tp data[0]的成员。实际上，这个成员已经有 _Tp data[msize]这样的空间了。 这里，如果我们设计另外一个分配内存的类，例如，LoopQue_ShmAlloc。这个类是使用共享内存，并在其中建立LoopQue_impl类。这样我们就可以使用：
LoopQue<_Tp, LoopQue_ShmAlloc>

来创建一个可以在进程间共享而进行通讯的环形队列类了。

至此，我们可以总结一下：

• 环形队列的数据结构和基本操作都是相当简单的。主要的操作就是push和get。
• 环形队列可能需要线程或者进程间共享操作。尤其是进程间的共享，使得内存分配成为了一个相对复杂的问题。对于这个问题，我们将基础的数据结构定义为无指针和引用成员，适合于内存管理的类，然后又设计了一个代理的进行二次包装的类，将内存分配的功能作为模版可定制的参数。

这里，我们设计并实现了环形队列的数据结构，并也为这个结构能够定制存放到共享内存设计好了方针策略。但下面的工作，将更富于挑战性：

• 对于push的操作，需要操作前等待队列已经有了空间，也就是说队列没有满的状态。等到这个状态出现了，才继续进行push的操作，否则，push操作挂起。
• 对于get 的操作，需要操作前等待队列有了数据，也就是说队列不为空的状态。等到这个状态出现了，才继续进行get的操作，否则，get操作挂起。
• 上面的push操作/get操作一般是不同的进程和线程。同时，也有可能有多个push的操作和get操作的进程和线程。

这些工作我们将在随后的设计中进行。且听下回分解。

附件: LoopQue的测试程序:

[cpp] view plaincopy

1. /* tst-loopque.cpp
2. test program for <LoopQue> class
3. Author: ZhangTao
4. Date: July 27, 2009
5. */
6. # include       <iostream>
7. # include       <cstdlib>       // for <atol> function
8. # include       "xtl/LoopQue.h"
9. int
10. main(int argc, char **argv)
11. {
12. int  qsize = 0;
13. if ( argc > 1 )
14. qsize = atol(argv[1]);
15. if ( qsize < 1 )
16. qsize = 5;
17. xtl::LoopQue<int>  queue(qsize);
18. for ( int i = 0; i < (qsize - 1); i++ )       {
19. queue.push(i);
20. std::cout << "Loop push:" << i << "/n";
21. }
22. queue.check_push(1000);
23. std::cout << "Full:" << queue.full() << " Size:"
24. << queue.size() << "/n/n";
25. for ( int i = 0; i < qsize; i++ )   {
26. int val = queue.front();
27. std::cout << "Loop pop:" << val << "/n";
28. queue.pop();
29. }
30. std::cout << "/nEmpty:" << queue.empty() << " Size:"
31. << queue.size() << "/n";
32. return 0;
33. }