1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

/* badcnt.c */
/* Global shared variable */ 
volatile long cnt = 0; /* Counter */ 
int main(int argc, char **argv) { 
	long niters; 
	pthread_t tid1, tid2; 
	niters = atoi(argv[1]); 
	Pthread_create(&tid1, NULL, thread, &niters); 
	Pthread_create(&tid2, NULL, thread, &niters); 
	Pthread_join(tid1, NULL); 
	Pthread_join(tid2, NULL); 
	
	/* Check result */ 
	if (cnt != (2 * niters)) 
		printf("BOOM! cnt=%ld\n", cnt); 
	else printf("OK cnt=%ld\n", cnt); 
		exit(0); 
}

/* Thread routine */ 
void *thread(void *vargp) { 
	long i, niters = *((long *)vargp); 
	for (i = 0; i < niters; i++) cnt++; 
	return NULL; 
}

1
2
3
4

#include <semaphore.h>
int sem_init(sem_t *s, 0, unsigned int val);} /* s = val */ 
int sem_wait(sem_t *s); /* P(s) */ 
int sem_post(sem_t *s); /* V(s) */

1
2
3
4
5

for (i = 0; i < niters; i++) { 
	P(&mutex); 
	cnt++; 
	V(&mutex); 
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

#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);

/* 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 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 0 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) { 
	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; 
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

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; 
	
	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 buf */ 
		echo_cnt(connfd); /* Service client */ 
		Close(connfd); 
	} 
}

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; 
		printf("thread %d received %d (%d total) bytes on fd %d\n", (int) pthread_self(), n, byte_cnt, connfd); 
		V(&mutex); 
		Rio_writen(connfd, buf, n); 
	} 
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

int main() { 
	pthread_t tid[2]; 
	Sem_init(&mutex[0], 0, 1); /* mutex[0] = 1 */ 
	Sem_init(&mutex[1], 0, 1); /* mutex[1] = 1 */ 
	Pthread_create(&tid[0], NULL, count, (void*) 0); 
	Pthread_create(&tid[1], NULL, count, (void*) 1); 
	Pthread_join(tid[0], NULL); 
	Pthread_join(tid[1], NULL); 
	printf("cnt=%d\n", cnt); exit(0); 
} 

void *count(void *vargp) { 
	int i; 
	int id = (int) vargp; 
	for (i = 0; i < NITERS; i++) { 
		P(&mutex[id]); P(&mutex[1-id]); 
		cnt++; 
		V(&mutex[id]); V(&mutex[1-id]); 
	} 
	return NULL; 
}