缓冲区设计—进程间通讯（POSIX 共享内存）

在《缓冲区设计-线程间通讯 》一文中，完成了BufQueue的模版类的设计。这个模版类定义如下：

template< typename _Tp, typename Alloc = BufQueue_allocate<_Tp> >
class BufQueue;

 

其中的模版参数Alloc可以订制为不同的环冲内存形式。在线程间的通讯中，这个参数当然是最简单的，它只要简单的分配空间就可以了，因为同一进程中的线程是共享进程的内存空间的。

 

但是，如果要在进程间使用这个环形缓冲区，则需要使用共享内存的机制了。只所以将BufQueue设计成有Alloc这样一个模版参数，也正是为了这样的目的的。这一篇里，我们将实现这个目标，也就是要写一个使用共享内存的Alloc。

 

这里使用POSIX共享内存的接口来实现。

POSIX共享内存主要接口为：使用shm_open打开或创建一个共享内存区，然后使用mmap将这个共享内存区影射到进程的内存空间中。

这两个接口的定义如下：

int shm_open(const char *name, int oflag, mode_t mode);
void *mmap(void *start, size_t length, int prot, int flags, int fd, off_t offset);

 

详细的POSIX共享内存的调用和接口说明，请参考有关的资料。

 

以下是源程序：

  /* BufQuePShm.h Author: ZhangTao Date: Aug 1, 2009 */   # ifndef BufQuePShm_h # define BufQueShm_h   # include <sys/types.h> # include <sys/mman.h> # include <sys/stat.h> # include <fcntl.h>   # include "xtl/except.h" # include "xtl/BufQueue.h"   namespace xtl {   const char * const ShmName = "/BufQueue";   template<typename _Tp> struct BufQuePosixShmAlloc { typedef BufQueue_Impl<_Tp> BufQueImpl;   BufQueImpl& allocate(int msize) { asize = sizeof(BufQueImpl) + LoopQue_impl<_Tp>::addsize(msize); bool creator = true;   errno = 0; int fd = shm_open(ShmName, O_RDWR | O_CREAT | O_EXCL, 0666);   if ( errno == EEXIST ) { // <ShmName> exist creator = false; fd = shm_open(ShmName, O_RDWR, 0); }   if ( fd < 0 ) throw_fmtException(__LINE__, __FUNCTION__, "shm_open (%s)", creator? "Crt" : "Use");   if ( creator ) ftruncate(fd, asize); else { struct stat sbuf; fstat(fd, &sbuf); if ( sbuf.st_size != asize ) throw_fmtException(__LINE__, __FUNCTION__, "shmsize expect %d but %d", asize, sbuf.st_size); }   void *paddr = mmap(0, asize, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);   close(fd);   if ( paddr == (void *)-1 ) throw_fmtException(__LINE__, __FUNCTION__, "mmap size(%d)", asize);   if ( creator ) return *(new (paddr) BufQueImpl(msize, true)); else return *(BufQueImpl *)paddr; }   void deallocate(void *p) { munmap(0, asize); shm_unlink(ShmName); }   int asize; };   } // end of <namespace xtl>   # endif /* end of <# ifndef BufQuePShm_h> */  

 

 

下面的程序是测试程序，也可称为使用示例程序。

  # include <stdio.h> # include <iostream> # include <sys/wait.h>   # define NBUFF 10 # define DEFAULT_ITEMS 20   typedef struct { pid_t id; int seq; } Data;   # ifdef PSHM   // Posix share memory # include "xtl/BufQuePShm.h" typedef xtl::BufQueue< Data, xtl::BufQuePosixShmAlloc<Data> > Queue;   # else   // System V share memory # include "xtl/BufQueShm.h" typedef xtl::BufQueue< Data, xtl::BufQueShmAlloc<Data> > Queue;   # endif   int main(int argc, char **argv) { Data data; int status; int nitems;   if ( argc < 2 ) nitems = DEFAULT_ITEMS; else nitems = atol(argv[1]);   pid_t pid; Queue *pque;   try { pque = new Queue(NBUFF); } catch ( xtl::utilException& ex ) { std::cerr << ex << "/n"; return -1; }   if ( (pid = fork()) == 0 ) { // child process int seq = 0;; int cnt;   printf("create child process %ld/n", (long)getpid());   cnt = 0; while( pque->get(data) ) { printf("%d %lu %d/n", seq++, data.id, data.seq); cnt++; }   printf("Child process end cnt %d/n", cnt);   return 0;   } else if ( pid < 0 ) { fprintf(stderr, "can not create process/n"); return -1; }   data.id = getpid();   for ( data.seq = 0; data.seq < nitems; data.seq++ ) pque->put(data);   pque->stop();   pid = wait(&status);   printf("child<%ld> exit with code <%d>/n", (long)pid, status);   return 0; }

 

 

程序的14行到19行对应的是使用本篇文章介绍的使用POSIX共享内存的缓冲区类型。而21行到26行使用的是Alloc模版参数为 xtl::BufQueShmAlloc，这是使用System V的共享内存实现的模版类（将在下一篇中给出实现的代码）。为了使用我们本篇的POSIX共享内存类，我们需要在上面的程序的16行之前增加一行：

# define PSHM

 

51行到64行之间为子进程，不断地读出队列数据。

73行到74行的循环里，是父进程向缓冲队列内写数据。