第十二章 并发编程

// 基于进程的并发服务器  
#include "csapp.h"  
void echo(int connfd);  
void sigchld_handler(int sig)  
{  
    while(waitpid(-1, 0, WNOHANG) > 0) {}  
    return;  
}  
int main(int argc, char **argv)  
{  
    int listenfd, connfd;  
    socklen_t clientlen;  
    struct sockaddr_storage clientaddr;   /* Enough space for any address */  
    char client_hostname[MAXLINE], client_port[MAXLINE];  
    if(argc != 2)  
    {  
        fprintf(stderr, "usage: %s <port>\n", argv[0]);  
        exit(0);  
    }  
    Signal(SIGCHLD, sigchld_handler);  
    listenfd = Open_listenfd(argv[1]);  
    while(1)  
    {  
        clientlen = sizeof(struct sockaddr_storage);  
        connfd = Accept(listenfd, (SA *)&clientaddr, &clientlen);  
        if(Fork() == 0)  
        {  
            Close(listenfd);  /* Child closes its listening socket */  
            Getnameinfo((SA *) &clientaddr, clientlen, client_hostname, MAXLINE, client_port, MAXLINE, 0);  
            printf("Connected to (%s, %s)\n", client_hostname, client_port);  
            echo(connfd);  /* Child services client */  
            Close(connfd); /* Child closes connection with client */  
            exit(0);       /* Child exits */  
        }  
        Close(connfd);     /* Parent closes connected socket (important!) */  
    }  
    exit(0);  
}  

#include <sys/select.h>  
// 返回已准备好的非零描述符的个数，若出错则为-1  
// fdset表示读集合的描述符集合  
// n表示读集合的基数，实际上是任何描述符集合的最大基数  
// select函数会一直阻塞直到读集合中至少有一个描述符准备好可以读  
// 当前仅当一个从该描述符读取一个字节的请求不会阻塞时，描述符k就准备好可以读了  
// select会修改fdsse指向的fd_set，指明读集合的一个子集，称为准备好集合，由准备好可以读的描述符组成  
// 因此每次调用select前都要更新读集合  
int select(int n, fd_set *fdset, NULL, NULL, NULL);  
FD_ZERO(fd_set *fdset);         // Clear all bits in fdset  
FD_CLR(int fd, fd_set *fdset);  // Clear bit fd in fdset  
FD_SET(int fd, fd_set *fdset);  // Turn on bit fd in fdset  
FD_ISSET(iny fd, fd_SET *fdset); // Is bit fd in fdset on?  

// 基于I/O多路复用的并发事件驱动服务器  
#include "csapp.h"  
typedef struct   /* Represents a pool of connected descriptors */  
{  
    int maxfd;       /* Largest descriptor in read_set */  
    fd_set read_set; /* Set of all active descriptors */  
    fd_set ready_set;/* Subset of descriptors ready for reading */  
    int nready;      /* Number of ready descriptors from select */  
    int maxi;        /* High water index into client array */  
    int clientfd[FD_SETSIZE];   /* Set of active descriptors */  
    rio_t clientrio[FD_SETSIZE];/* Set of active read buffers */  
}pool;  
int byte_cnt = 0;   /* Counts total bytes received by server */  

// 初始化客户端池  
void init_pool(int listenfd, pool *p)  
{  
    /* Initially, there are no connected descriptors */  
    int i;  
    p->maxi = -1;  
    for(i = 0; i < FD_SETSIZE; ++i)  
        p->clientfd[i] = -1;  
    /* Initially, listenfd is only member of select read set */  
    p->maxfd = listenfd;  
    FD_ZERO(&p->read_set);  
    FD_SET(listenfd, &p->read_set);  
}  

// 添加一个新的客户端到活动客户端池中  
void add_client(int connfd, pool *p)  
{  
    int i;  
    --p->nready;  
    for(i = 0; i < FD_SETSIZE; ++i)   /* Find an available slot */  
    {  
        if(p->clientfd[i] < 0)  
        {  
            /* Add connected descriptor to the pool */  
            p->clientfd[i] = connfd;  
            Rio_readinitb(&p->clientrio[i], connfd);  
            /* Add the descriptor to descriptor set */  
            FD_SET(connfd, &p->read_set);  
            /* Update max descriptor and pool high water mark */  
            if(connfd > p->maxfd)  
                p->maxfd = connfd;  
            if(i > p->maxi)  
                p->maxi = i;  
            break;  
        }  
    }  
    if(i == FD_SETSIZE)  /* Couldn't find an empty slot */  
        app_error("add_client error: Too many clients");  
}  

// 回送来自每个准备好的已连接描述符的一个文本行  
void check_clients(pool *p)  
{  
    int i, connfd, n;  
    char buf[MAXLINE];  
    rio_t rio;  
    for(i = 0; (i <= p->maxi) && (p->nready > 0); ++i)  
    {  
        connfd = p->clientfd[i];  
        rio = p->clientrio[i];  
        /* If the descriptor is ready, echo a text line from it */  
        if((connfd > 0) && (FD_ISSET(connfd, &p->ready_set)))  
        {  
            --p->nready;  
            if((n = Rio_readlineb(&rio, buf, MAXLINE)) != 0)  
            {  
                byte_cnt += n;  
                printf("Server reveived %d (%d total) bytes on fd %d\n", n, byte_cnt, connfd);  
                Rio_writen(connfd, buf, n);  
            }  
            else  /* EOF detected, remove descriptor from pool */  
            {  
                Close(connfd);  
                FD_CLR(connfd, &p->read_set);  
                p->clientfd[i] = -1;  
            }  
        }  
    }  
}  

int main(int argc, char **argv)  
{  
    int listenfd, connfd;  
    socklen_t clientlen;  
    struct sockaddr_storage clientaddr;  
    static pool pool;  
    if(argc != 2)  
    {  
        fprintf(stderr, "usage: %s <port>\n", argv[0]);  
        exit(0);  
    }  
    listenfd = Open_listenfd(argv[1]);  
    init_pool(listenfd, &pool);  
    while(1)  
    {  
        /* Wait for listening/connected descriptor(s) to become ready */  
        pool.ready_set = pool.read_set;  
        pool.nready = Select(pool.maxfd + 1, &pool.ready_set, NULL, NULL, NULL);  
        /* If listening descriptor ready, add new client to pool */  
        if(FD_ISSET(listenfd, &pool.ready_set))  
        {  
            clientlen = sizeof(struct sockaddr_storage);  
            connfd = Accept(listenfd, (SA *)&clientaddr, &clientlen);  
            add_client(connfd, &pool);  
        }  
        /* Echo a text line from each ready connected descriptor */  
        check_clients(&pool);  
    }  
    exit(0);  
}  

#include <pthread.h>  
typedef void *(func)(void *);  
// 成功则返回0，出错返回非零  
int pthread_create(pthread_t *tid, pthread_attr_t *attr, func *f, void *argp);  
// 返回调用者的线程ID  
pthread_t pthread_self();  

#include <pthread.h>  
void pthread_exit(void *thread_return);  
// 成功返回0， 出错返回非零  
int pthread_cancel(pthread_t tid);  

#include <pthread.h>  
// 成功返回0， 出错返回非零  
int pthread_join(pthread_t tid, void **thread_return);  

#include <pthread.h>  
// 分离可结合线程tid  
// 线程通过以pthread_self()为参数调用来分离自己  
int pthread_detach(pthread_t tid);  

#include <pthread.h>  
// once_control是一个全局或者静态变量，总是被初始化为PTHREAD_ONCE_INIT  
pthread_once_t once_control = PTHREAD_ONCE_INIT;  
// 第一次以once_control调用pthread_once时，调用init_routine  
// 接下来再调用时将不做任何事  
int pthread_once(pthread_once_t *once_control, void (*init_routine)(void));  

// 基于线程的并发服务器  
#include "csapp.h"  
void echo(int connfd);  
void *thread(void *vargp);  
int main(int argc, char **argv)  
{  
    int listenfd, *connfdp;  
    socklen_t clientlen;  
    struct sockaddr_storage clientaddr;  
    pthread_t tid;  
    if(argc != 2)  
    {  
        fprintf(stderr, "usage: %s <port>\n", argv[0]);  
        exit(0);  
    }  
    listenfd = Open_listenfd(argv[1]);  

    while(1)  
    {  
        clientlen = sizeof(struct sockaddr_storage);  
        connfdp = Malloc(sizeof(int));  
        *connfdp = Accept(listenfd, (SA *)&clientaddr, &clientlen);  
        Pthread_create(&tid, NULL, thread, connfdp);  
    }  
}  

/* Thread routine */  
void *thread(void *vargp)  
{  
    int connfd = *((int *)vargp);  
    Pthread_detach(pthread_self());  
    Free(vargp);  
    echo(connfd);  
    Close(connfd);  
    return NULL;  
}  

#include <semaphore.h>  
// 将信号量sem初始化为value，每个信号量在使用之前必须初始化  
int sem_init(sem_t *sem, 0, unsigned int value);  
int sem_wait(set_t *s);   /* P(s) */  
int sem_post(sem_t *s);   /* V(s) */  

#include "csapp.h"  
void P(sem_t *s);   // Wrapper function for sem_wait  
void V(sem_t *s);   // Wrapper function for sem_post  

// sbuf.h  
#include "csapp.h"  
typedef struct  
{  
    int *buf;    /* Buffer array */  
    int n;       /* Maximum number of slots */  
    int front;   /* buf[(front + 1) % n] is first item */  
    int rear;    /* buf[rear % n] is last item */  
    sem_t mutex; /* Protects accesses to buf */  
    sem_t slots; /* Counts available slots */  
    sem_t items; /* Counts available items */  
}sbuf_t;  
void sbuf_init(sbuf_t *sp, int n);  
void sbuf_deinit(sbuf_t *sp);  
void sbuf_insert(sbuf_t *sp, int item);  
int sbuf_remove(sbuf_t *sp);  

#include "csapp.h"  
#include "sbuf.h"  
/* Create an empty, bounded, shared FIFO buffer with n slots */  
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 if 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); /* Initiallt, buf has zero data items */  
}  

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

/* Insert item onto the rear of shared buffer sp */  
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 */  
}  

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

// Global variables  
int readcnt;   // Initially = 0  
sem_t mutex, w;  // Both initially = 1  

void reader()  
{  
    P(&mutex);  
    ++readcnt;  
    if(readcnt == 1)   // First in  
        P(&w);  
    V(&mutex);  
    // Reading happens  
    P(&mutex)  
    --readcnt;  
    if(raedcnt == 0)   // Last out  
        V(&w);  
    V(&mutex);  
}  

void writer()  
{  
    P(&w);  
    // Writing happens ...  
    V(&w);  
}  

// 预线程化的并发服务器  
#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>\n", argv[0]);  
        exit(0);  
    }  
    listenfd = Open_listenfd(argv[1]);  

    sbuf_init(&sbuf, SBUFSIZE);  
    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);  
    }  
}  

// echo-cnt.c  
#include "csapp.h"  
static int byte_cnt;   /* Byte counter */  
static sem_t mutex;    /* and the mutex that protects it */  

static void init_echo_cnt()  
{  
    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;  
        printf("server received %d (%d total bytes) on fd %d\n", n, byte_cnt, connfd);  
        V(&mutex);  
        Rio_writen(connfd, buf, n);  
    }  
}