经典生产者-消费者问题解析

1.生产者-消费者问题

生产者和消费者问题在现实系统中是很普遍的。例如在一个多媒体系统中，生产者编码视频帧，而消费者消费（解码）视频帧，缓冲区的目的就是减少视频流的抖动。又如在图形用户接口设计中，生产者检测到鼠标和键盘事件，并将其插入到缓冲区中。消费者以某种基于优先级的方式从缓冲区中取出这些事件并显示在屏幕上。

生产者和消费者模式共享一个有n个槽位的有限缓冲区。生产者反复地生成新的item，并将它插入到缓冲区中。消费者不断从缓冲区中取出这些item，并消费它们。它有3个显著特点：

• 因为生产和消费都涉及共享变量的更新，所以必须保证对缓冲区的访问是互斥的。
• 如果缓冲区是满的，那么生产者必须等待直到有一个槽位可用。
• 如果缓冲区是空的，那么消费者必须等待直到有一个item可以。

下文将开发一个简单的生产者消费者包SBUF，它操作类型为sbuf_t的有限缓冲区。item存放在一个动态分配的容量为n的整数数组buf中。索引值front和rear分别指向数组的首尾项。三个信号量同步对缓冲区的访问。mutex信号量提供互斥访问，slots和items信号量分别记录空槽位和可用item的数量。

typedef struct{
    int *buf;		//缓冲区指针（指向一个数组）
    int n;			//最大空槽位数量（缓冲区大小）
    int front;		//指向数组第一个item，即buf[(front+1)%n]
    int rear;		//指向数组最后一个item，即buf[rear%n]
    sem_t mutex;	//缓冲区互斥锁
    sem_t slots;	//可用槽位数
    sem_t items;	//可用item数
}sbuf_t;

Posix标准定义的信号量操作函数：

#include <semaphore.h>
int sem_init(sem_t *sem, 0, unsigned int value);
int sem_wait(sem_t *s);		//等价于P(s)
int sem_post(sem_t *s);		//等价于V(s)

下文继续给出SBUF包实现的代码：

• sbuf_init函数初始化一个缓冲区，在其它任意函数前被调用；
• sbuf_deinit函数释放一个缓冲区，在其它任意函数后被调用；
• sbuf_insert函数等待一个可用槽位并添加item；
• sbuf_remove函数等待并消费一个item；

/* $begin sbufc */
#include "csapp.h"
#include "sbuf.h"

/* Create an empty, bounded, shared FIFO buffer with n slots */
/* $begin sbuf_init */
void sbuf_init(sbuf_t *sp, int n)
{
    sp->buf = Calloc(n, sizeof(int)); 
    sp->n = n;                       /* Buffer holds max of n items */
    sp->front = sp->rear = 0;        /* Empty buffer iff front == rear */
    Sem_init(&sp->mutex, 0, 1);      /* Binary semaphore for locking */
    Sem_init(&sp->slots, 0, n);      /* Initially, buf has n empty slots */
    Sem_init(&sp->items, 0, 0);      /* Initially, buf has zero data items */
}
/* $end sbuf_init */

/* Clean up buffer sp */
/* $begin sbuf_deinit */
void sbuf_deinit(sbuf_t *sp)
{
    Free(sp->buf);
}
/* $end sbuf_deinit */

/* Insert item onto the rear of shared buffer sp */
/* $begin sbuf_insert */
void sbuf_insert(sbuf_t *sp, int item)
{
    P(&sp->slots);                          /* Wait for available slot */
    P(&sp->mutex);                          /* Lock the buffer */
    sp->buf[(++sp->rear)%(sp->n)] = item;   /* Insert the item */
    V(&sp->mutex);                          /* Unlock the buffer */
    V(&sp->items);                          /* Announce available item */
}
/* $end sbuf_insert */

/* Remove and return the first item from buffer sp */
/* $begin sbuf_remove */
int sbuf_remove(sbuf_t *sp)
{
    int item;
    P(&sp->items);                          /* Wait for available item */
    P(&sp->mutex);                          /* Lock the buffer */
    item = sp->buf[(++sp->front)%(sp->n)];  /* Remove the item */
    V(&sp->mutex);                          /* Unlock the buffer */
    V(&sp->slots);                          /* Announce available slot */
    return item;
}
/* $end sbuf_remove */
/* $end sbufc */

2. 读者-写者问题

读者-写者问题在现实系统中也比较常见。例如，一个在线影院座位预定系统中，允许有无限多个客户同时查看（读者）座位分配，但是正在预定的客户必须拥有对数据库的独占访问（写者）。读者写者问题又分为以下几种情况：

• 读者优先，即除非有写者正在写，否则不能让读者等；
• 写者优先，即只要写者准备好写，就尽快完成写。在写者发出写请求后到达的读者，必须等待；

下文给出读者优先的一个示例：

/*全局变量*/
int readcnt;	//统计当前在临界区中读者的数量
sem_t mutex;	//保护对readcnt的访问
sem_t w;		//控制对访问共享对象的临界区的访问

void reader(void)
{
    while(1){
        P(&mutex);
        readcnt++;
        if(readcnt == 1)	//first in
            P(&w);
        V(&mutex);
        
        /* 临界区操作语句 */
        /* 读语句 */
        
        P(&mutex);
        readcnt--;
        if(readcnt == 0)	//last out
            V(&w);
        V(&mutex);
    }
}

void writer(void)
{
    while(1){
        P(&w);
        
        /* 临界区操作语句 */
        /* 写语句 */
        
        V(&w);
    }
}

• 为了保证任意时刻临界区中只有一个写者，每当一个写者进入临界区时，它对互斥锁w加锁，每当它离开临界区时，对w解锁；

• 为了保证只要还有一个读者占用互斥锁w，那么无限多的读者就可以无障碍的进入临界区读，只有第一个进入临界区的读者对w加锁，只有最后一个离开的读者对w解锁。

  此法可能导致饥饿（starvation）:如果有读者不断到达，写者就无限期等待。

  3. 基于预线程化的并发服务器

  前文叙述了如何使用信号量来访问共享变量和调度对共享资源的访问，现在可以动手实现一个基于预线程化的技术（prethreading）的并发服务器开发。

如图所示，服务器是由一个主线程和一组工作者线程构成的。主线程不断接受来自客户端的连接请求，并将得到的连接描述符放在一个缓冲区中。每一个工作者线程反复地从共享缓冲区中取出描述符为客户端服务，然后等待下一个描述符。

下面给出具体代码：

#include "csapp.h"
#include "sbuf.h"
#define NTHREADS  4
#define SBUFSIZE  16

void echo_cnt(int connfd);
void *thread(void *vargp);

sbuf_t sbuf; /* Shared buffer of connected descriptors */

int main(int argc, char **argv) 
{
    int i, listenfd, connfd;
    socklen_t clientlen;
    struct sockaddr_storage clientaddr;
    pthread_t tid; 

    if (argc != 2) {
	fprintf(stderr, "usage: %s <port>
", argv[0]);
	exit(0);
    }
    listenfd = Open_listenfd(argv[1]);

    sbuf_init(&sbuf, SBUFSIZE); //line:conc:pre:initsbuf
    for (i = 0; i < NTHREADS; i++)  /* Create worker threads */ 
	Pthread_create(&tid, NULL, thread, NULL);               
    while (1) { 
        clientlen = sizeof(struct sockaddr_storage);
        connfd = Accept(listenfd, (SA *) &clientaddr, &clientlen);
        sbuf_insert(&sbuf, connfd); /* Insert connfd in buffer */
    }
}

void *thread(void *vargp) 
{  
    Pthread_detach(pthread_self()); 
    while (1) { 
	int connfd = sbuf_remove(&sbuf); /* Remove connfd from buffer */ 
	echo_cnt(connfd);                /* Service client */
	Close(connfd);
    }
}

• 首先初始化缓冲区sbuf（line 24）后，主线程创建了一组工作者线程（line 25~26）。
• 之后进入无限循环，接受连接请求，并将得到的已连接描述符插入缓冲区sbuf中。
• 每个工作者线程的行为非常简单，它等待直到能从缓冲区中取出一个已连接描述符（line 39），然后调用echo_cnt函数回送客户端的输入。

下面给出echo_cnt函数的代码，它向你展示了一个从线程例程调用的初始化程序包的一般技术。其中全局变量byte_cnt中记录了从所有客户端接受到的累计字节数。

#define RIO_BUFSIZE 8192
typedef struct {
    int rio_fd;                /* Descriptor for this internal buf */
    int rio_cnt;               /* Unread bytes in internal buf */
    char *rio_bufptr;          /* Next unread byte in internal buf */
    char rio_buf[RIO_BUFSIZE]; /* Internal buffer */
} rio_t;

static int byte_cnt;  /* Byte counter */
static sem_t mutex;   /* and the mutex that protects it */

static void init_echo_cnt(void)
{
    Sem_init(&mutex, 0, 1);
    byte_cnt = 0;
}

void echo_cnt(int connfd) 
{
    int n; 
    char buf[MAXLINE]; 
    rio_t rio;
    static pthread_once_t once = PTHREAD_ONCE_INIT;

    Pthread_once(&once, init_echo_cnt); 
    Rio_readinitb(&rio, connfd);        
    while((n = Rio_readlineb(&rio, buf, MAXLINE)) != 0) {
        P(&mutex);
        byte_cnt += n; //line:conc:pre:cntaccess1
        printf("server received %d (%d total) bytes on fd %d
", 
               n, byte_cnt, connfd); 
        V(&mutex);
        Rio_writen(connfd, buf, n);
    }
}

void rio_readinitb(rio_t *rp, int fd) 
{
    rp->rio_fd = fd;  
    rp->rio_cnt = 0;  
    rp->rio_bufptr = rp->rio_buf;
}

ssize_t rio_readlineb(rio_t *rp, void *usrbuf, size_t maxlen) 
{
    int n, rc;
    char c, *bufp = usrbuf;
    for (n = 1; n < maxlen; n++) { 
        if ((rc = rio_read(rp, &c, 1)) == 1) {
            *bufp++ = c;
            if (c == '
') {
                n++;
                break;
            }
        } else if (rc == 0) {
	    	if (n == 1)
				return 0; /* EOF, no data read */
	   		else
				break;    /* EOF, some data was read */
		} else
	    	return -1;	  /* Error */
    }
    *bufp = 0;
    return n-1;
}

ssize_t rio_writen(int fd, void *usrbuf, size_t n) 
{
    size_t nleft = n;
    ssize_t nwritten;
 	char *bufp = usrbuf;
    while (nleft > 0) {
        if ((nwritten = write(fd, bufp, nleft)) <= 0) {
            if (errno == EINTR)  /* Interrupted by sig handler return */
                nwritten = 0;    /* and call write() again */
            else
                return -1;       /* errno set by write() */
        }
        nleft -= nwritten;
        bufp += nwritten;
    }
    return n;
}

• 首先初始化byte_cnt计数器和mutex信号量；
• 一种是显式地调用一个初始化函数，一种是上文所采取的利用pthread_once函数。即当第一次有某个线程调用echo_cnt函数时，使用pthread_once函数去调用初始化函数。这个方法的优点是使程序包的使用更加容易，缺点使每一次调用echo_cnt函数都会导致调用pthread_once函数，而除了第一次，它没有做什么有用的事。
  首先初始化byte_cnt计数器和mutex信号量；
• 一种是显式地调用一个初始化函数，一种是上文所采取的利用pthread_once函数。即当第一次有某个线程调用echo_cnt函数时，使用pthread_once函数去调用初始化函数。这个方法的优点是使程序包的使用更加容易，缺点使每一次调用echo_cnt函数都会导致调用pthread_once函数，而除了第一次，它没有做什么有用的事。
• 一旦程序包被初始化，echo_cnt函数会初始化RIO带缓冲区的I/O包（line 20），然后回送从客户端接收到的每一个文本行。

---

获取更多知识，请点击关注：
嵌入式Linux&ARM
CSDN博客
简书博客
知乎专栏

---