有关并发(3)

Apr 4, 2014
Technology

当一个程序的正确性依赖于一个线程需要在另一个线程到达Y点之前到达它控制流中的x点时，就会发生竞争。竞争是因为程序员假定线程按照某种特殊的轨线穿过执行状态空间，而忘了另一条规则规定的，多线程程序必须对任何可行的轨线都正确工作。
例子如下：

/* 
 * race.c - demonstrates a race condition
 */
/* $begin race */
#include "csapp.h"
#define N 4

void *thread(void *vargp);

int main() 
{
    pthread_t tid[N];
    int i;

    for (i = 0; i < N; i++) 
	Pthread_create(&tid[i], NULL, thread, &i); //line:conc:race:createthread
    for (i = 0; i < N; i++) 
	Pthread_join(tid[i], NULL);
    exit(0);
}

/* thread routine */
void *thread(void *vargp) 
{
    int myid = *((int *)vargp);  //line:conc:race:derefarg
    printf("Hello from thread %d\n", myid);
    return NULL;
}
/* $end race */

运行结果如下:

[Trusty@DashArch chapter12]$ ./race 
Hello from thread 3
Hello from thread 1
Hello from thread 2
Hello from thread 3

可以看到程序的执行流并非我们想象的。因而我们可以对程序做改造。改造后的代码如下：

/* 
 * norace.c - fixes the race in race.c
 */
/* $begin norace */
#include "csapp.h"
#define N 4

void *thread(void *vargp);

int main() 
{
    pthread_t tid[N];
    int i, *ptr;

    for (i = 0; i < N; i++) {
	ptr = Malloc(sizeof(int));                    //line:conc:norace:createthread1
	*ptr = i;                                     //line:conc:norace:createthread2
	Pthread_create(&tid[i], NULL, thread, ptr);   //line:conc:norace:createthread3
	printf("Now create the %d thread.\n", i);
    } //line:conc:norace:endloop
    for (i = 0; i < N; i++) 
	Pthread_join(tid[i], NULL);
    exit(0);
}

/* thread routine */
void *thread(void *vargp) 
{
    int myid = *((int *)vargp);
    Free(vargp); 
    printf("Hello from thread %d\n", myid);
    return NULL;
}
/* $end norace */

运行结果如下：

[Trusty@DashArch chapter12]$ ./norace 
Now create the 0 thread.
Hello from thread 0
Hello from thread 1
Now create the 1 thread.
Now create the 2 thread.
Hello from thread 2
Now create the 3 thread.
Hello from thread 3

可以看到，为ID分配独立块，再把指向这个块的指针传递给线程是一个很好的主意。但是注意线程需要释放这些块，以避免存储器泄漏。

Convert PDF from Markdown

Apr 3, 2014
Technology

There are 2 ways for me to convert Markdown into PDF.

pandoc Way

Simple way for converting:

pandoc a.md -o b.pdf

Convert with embedded font:

pandoc jailbreak.md -o jailbreak.pdf --latex-engine=xelatex -V mainfont="WenQuanYi Zen Hei"

But if you includes the images, then the format maybe wrong.

RStudio Way

Install rstudio in archlinux via:

yaourt -S rstudio-desktop-bin

When installing you will meet curl need ssl certification problem, simply add following line into the pkg make file.

==> Edit PKGBUILD ? [Y/n] ("A" to abort)
==> ------------------------------------
==> y

Please add $EDITOR to your environment variables
for example:
export EDITOR="vim" (in ~/.bashrc)
(replace vim with your favorite editor)

==> Edit PKGBUILD with:  vim

Add Following:   

DLAGENTS=("https::/usr/bin/curl -k -o %o %u")

After you installed rstudio, open the markdown file in the browser,and you have to install following packages:

> install.packages("knitr")
> install.packages("devtools")
> devtools::install_github("rstudio/rmarkdown")
> library(rmarkdown)
> render('test1.Rmd', pdf_document())

Makesure your pandoc is the latest version before you really install the rmarkdown package.
If the markdown is written as in octopress, you have to specify the absolute position of the file:

>  render('/home/Trusty/code/octo/heroku/Tomcat/source/dancecrf.markdown', pdf_document())

Now in the same file you will get the generated pdf.

CURL Error in Yaourt

Apr 3, 2014
Technology

When downloading package, the yaourt complains:

	yaourt curl: (60) SSL certificate problem

The solution is via: Add following lines at the top of the pkg build file:

	DLAGENTS=("https::/usr/bin/curl -k -o %o %u")

Then restart the yaourt, you can pass through the building.

有关并发(2)

Apr 2, 2014
Technology

I/O多路技术可以作为并发事件驱动(event-driver)程序的基础。事件驱动中，流作为某种事件的结果前进。逻辑流可以被转化为状态机。在服务器中，我们可以用I/O多路复用，借助select函数，检测输入事件的发声。当每个已连接描述符准备好可读时，服务器为相应的状态机执行状态转移。在例子中，就是从描述符读入和写回一个文本行。 ###并发事件驱动服务器首先看代码:

	/* 
	 * echoservers.c - A concurrent echo server based on select
	 */
	/* $begin echoserversmain */
	#include "csapp.h"
	
	typedef struct { /* represents a pool of connected descriptors */ //line:conc:echoservers:beginpool
	    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;         /* highwater index into client array */
	    int clientfd[FD_SETSIZE];    /* set of active descriptors */
	    rio_t clientrio[FD_SETSIZE]; /* set of active read buffers */
	} pool; //line:conc:echoservers:endpool
	/* $end echoserversmain */
	void init_pool(int listenfd, pool *p);
	void add_client(int connfd, pool *p);
	void check_clients(pool *p);
	/* $begin echoserversmain */
	
	int byte_cnt = 0; /* counts total bytes received by server */
	
	int main(int argc, char **argv)
	{
	    int listenfd, connfd, port; 
	    socklen_t clientlen = sizeof(struct sockaddr_in);
	    struct sockaddr_in clientaddr;
	    static pool pool; 
	
	    if (argc != 2) {
		fprintf(stderr, "usage: %s <port>\n", argv[0]);
		exit(0);
	    }
	    port = atoi(argv[1]);
	
	    listenfd = Open_listenfd(port);
	    init_pool(listenfd, &pool); //line:conc:echoservers:initpool
	    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)) { //line:conc:echoservers:listenfdready
		    connfd = Accept(listenfd, (SA *)&clientaddr, &clientlen); //line:conc:echoservers:accept
		    add_client(connfd, &pool); //line:conc:echoservers:addclient
		}
		
		/* Echo a text line from each ready connected descriptor */ 
		check_clients(&pool); //line:conc:echoservers:checkclients
	    }
	}
	/* $end echoserversmain */
	
	/* $begin init_pool */
	void init_pool(int listenfd, pool *p) 
	{
	    /* Initially, there are no connected descriptors */
	    int i;
	    p->maxi = -1;                   //line:conc:echoservers:beginempty
	    for (i=0; i< FD_SETSIZE; i++)  
		p->clientfd[i] = -1;        //line:conc:echoservers:endempty
	
	    /* Initially, listenfd is only member of select read set */
	    p->maxfd = listenfd;            //line:conc:echoservers:begininit
	    FD_ZERO(&p->read_set);
	    FD_SET(listenfd, &p->read_set); //line:conc:echoservers:endinit
	}
	/* $end init_pool */
	
	/* $begin add_client */
	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;                 //line:conc:echoservers:beginaddclient
		    Rio_readinitb(&p->clientrio[i], connfd); //line:conc:echoservers:endaddclient
	
		    /* Add the descriptor to descriptor set */
		    FD_SET(connfd, &p->read_set); //line:conc:echoservers:addconnfd
	
		    /* Update max descriptor and pool highwater mark */
		    if (connfd > p->maxfd) //line:conc:echoservers:beginmaxfd
			p->maxfd = connfd; //line:conc:echoservers:endmaxfd
		    if (i > p->maxi)       //line:conc:echoservers:beginmaxi
			p->maxi = i;       //line:conc:echoservers:endmaxi
		    break;
		}
	    if (i == FD_SETSIZE) /* Couldn't find an empty slot */
		app_error("add_client error: Too many clients");
	}
	/* $end add_client */
	
	/* $begin check_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; //line:conc:echoservers:beginecho
			printf("Server received %d (%d total) bytes on fd %d\n", 
			       n, byte_cnt, connfd);
			Rio_writen(connfd, buf, n); //line:conc:echoservers:endecho
		    }
	
		    /* EOF detected, remove descriptor from pool */
		    else { 
			Close(connfd); //line:conc:echoservers:closeconnfd
			FD_CLR(connfd, &p->read_set); //line:conc:echoservers:beginremove
			p->clientfd[i] = -1;          //line:conc:echoservers:endremove
		    }
		}
	    }
	}
	/* $end check_clients */

例子解析如下，程序首先调用init_poll初始化一个用于记录活动客户端的池(poll)。初始化完池以后，服务器进入一个无限循环。每次循环中，服务器将调用select函数用于检测两个事件，一个是来自一个新客户端的链接请求到大，另一个是已经存在的客户端的已连接描述符准备好可读。如果连接请求到达，服务器将调用add_client()将此客户端的连接描述符加入到poll中。最后，服务器将调用check_client()函数，把来自每个准备好的已连接描述符的一个文本行回送回去。

add_client和check_clients是非常精妙的设计。这构成了一个有限状态模式，select函数用于检测输入事件，add_client函数创建一个新的逻辑流。check_clients函数通过回送输入行来执行状态转移，客户端完成文本行传输时，还需要删除这个状态机。

局限在于，我们在这里能监听到的最大客户端是有限的。编码是较为复杂的。而且需要的代码是基于进程的服务器的3倍，并且，随着并发性粒度的减小，其复杂性还会上升。粒度是指的每个逻辑流每个时间片执行的指令数目。在示例并发服务器中，并发粒度就是读一个完整的文本行所需要的指令数目。如果某个逻辑流忙于读一个文本行，那么其他逻辑流就只能等着。这使得"只发送部分文本行就停止"的攻击方式得以可能。 ###基于线程的并发编程首先写代码:

	/* 
	 * echoservert.c - A concurrent echo server using threads
	 */
	/* $begin echoservertmain */
	#include "csapp.h"
	
	void echo(int connfd);
	void *thread(void *vargp);
	
	int main(int argc, char **argv) 
	{
	    int listenfd, *connfdp, port;
	    socklen_t clientlen=sizeof(struct sockaddr_in);
	    struct sockaddr_in clientaddr;
	    pthread_t tid; 
	
	    if (argc != 2) {
		fprintf(stderr, "usage: %s <port>\n", argv[0]);
		exit(0);
	    }
	    port = atoi(argv[1]);
	
	    listenfd = Open_listenfd(port);
	    while (1) {
		connfdp = Malloc(sizeof(int)); //line:conc:echoservert:beginmalloc
		*connfdp = Accept(listenfd, (SA *) &clientaddr, &clientlen); //line:conc:echoservert:endmalloc
		Pthread_create(&tid, NULL, thread, connfdp);
	    }
	}
	
	/* thread routine */
	void *thread(void *vargp) 
	{  
	    int connfd = *((int *)vargp);
	    Pthread_detach(pthread_self()); //line:conc:echoservert:detach
	    Free(vargp);                    //line:conc:echoservert:free
	    echo(connfd);
	    Close(connfd);
	    return NULL;
	}
	
	void echo(int connfd)
	{
		size_t n;
		char buf[MAXLINE];
		rio_t rio;
	
		Rio_readinitb(&rio, connfd);
		while((n = Rio_readlineb(&rio, buf, MAXLINE)) != 0) {
			printf("server received %d bytes\n", n);
			Rio_writen(connfd, buf, n);
		}
	}
	
	
	/* $end echoservertmain */

这个代码其实和我当初写的DNS Server的代码类似，来一个请求则启动一个线程处理，处理完线程自动退出。因而我们可以看到在thread()函数中实现了对一个连接的完整处理，而主线程只负责监听新的连接请求，来一个请求则建立一个线程，线程创建完毕以后自己detach掉。这里有一个需要注意的，就是传递给子线程的connfd的值可能会被更改，所以我们用到了malloc()和free()用以保护。 ###多线程程序中的共享变量代码如下：

/* $begin sharing */
#include "csapp.h"
#define N 2
void *thread(void *vargp);

char **ptr;  /* global variable */ //line:conc:sharing:ptrdec

int main() 
{
    int i;  
    pthread_t tid;
    char *msgs[N] = {
	"Hello from foo",  
	"Hello from bar"   
    };

    ptr = msgs; 
    for (i = 0; i < N; i++)  
        Pthread_create(&tid, NULL, thread, (void *)i); 
    Pthread_exit(NULL); 
}

void *thread(void *vargp) 
{
    int myid = (int)vargp;
    static int cnt = 0; //line:conc:sharing:cntdec
    printf("[%d]: %s (cnt=%d)\n", myid, ptr[myid], ++cnt); //line:conc:sharing:stack
    return NULL;
}
/* $end sharing */

运行结果如下：

	[Trusty@XXXyyy chapter12]$ ./sharing 
	[0]: Hello from foo (cnt=1)
	[1]: Hello from bar (cnt=2)

这里需要明了的概念是：寄存器是从不共享的，但是虚拟存储器总是共享的。线程1改写了一个存储器位置，那么在线程2中是可以看到这个变化的。本地静态变量是定义在函数内部并且具备static属性的变量。虚拟存储器的读/写区域里之包含在程序中声明的每个本地静态变量的一个实例，因而在cnt中，运行时由于只有一个cnt的实例，因而每个对等线程都将读/写到这个实例。 ###用信号量同步线程代码如下:

	/* 
	 * badcnt.c - An improperly synchronized counter program 
	 */
	/* $begin badcnt */
	#include "csapp.h"
	
	void *thread(void *vargp);  /* Thread routine prototype */
	
	/* Global shared variable */
	volatile int cnt = 0; /* Counter */
	
	int main(int argc, char **argv) 
	{
	    int niters;
	    pthread_t tid1, tid2;
	
	    /* Check input argument */
	    if (argc != 2) { 
		printf("usage: %s <niters>\n", argv[0]);
		exit(0);
	    }
	    niters = atoi(argv[1]);
	
	    /* Create threads and wait for them to finish */
	    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=%d\n", cnt);
	    else
		printf("OK cnt=%d\n", cnt);
	    exit(0);
	}
	
	/* Thread routine */
	void *thread(void *vargp) 
	{
	    int i, niters = *((int *)vargp);
	
	    for (i = 0; i < niters; i++) //line:conc:badcnt:beginloop
		cnt++;                   //line:conc:badcnt:endloop
	
	    return NULL;
	}
	/* $end badcnt */

如果我们给出一个很大的数，运行将报错，但是如果我们给出的值足够小，那么，没问题，它每次都会运行成功:

	[Trusty@XXXyyy chapter12]$ ./badcnt 1000000
	BOOM! cnt=1842364
	[Trusty@XXXyyy chapter12]$ ./badcnt 1000000
	BOOM! cnt=1101701
	[Trusty@XXXyyy chapter12]$ ./badcnt 1000
	OK cnt=2000

原因只是在于当数值够大的时候，两个线程的临界区会交替执行。因此计算的最终结果，是小于2cnt。我们可以用信号量来避免这种问题的产生。 ###利用信号量访问共享变量修改过的代码如下：

	/* 
	 * badcnt.c - An improperly synchronized counter program 
	 */
	/* $begin badcnt */
	#include "csapp.h"
	
	void *thread(void *vargp);  /* Thread routine prototype */
	
	/* Global shared variable */
	volatile int cnt = 0; /* Counter */
	static sem_t mutex;
	
	int main(int argc, char **argv) 
	{
	    int niters;
	    pthread_t tid1, tid2;
	
	    //sem_t mutex;
	
	    /* Check input argument */
	    if (argc != 2) { 
		printf("usage: %s <niters>\n", argv[0]);
		exit(0);
	    }
	
	    /* Initialize the semaphore */
	    sem_init(&mutex, 0, 1);
	
	    niters = atoi(argv[1]);
	
	    /* Create threads and wait for them to finish */
	    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=%d\n", cnt);
	    else
		printf("OK cnt=%d\n", cnt);
	    exit(0);
	}
	
	/* Thread routine */
	void *thread(void *vargp) 
	{
	    int i, niters = *((int *)vargp);
	
	    P(&mutex);
	    for (i = 0; i < niters; i++) //line:conc:badcnt:beginloop
	    {
		cnt++;                   //line:conc:badcnt:endloop
	    }
	    V(&mutex);
	
	    return NULL;
	}
	/* $end badcnt */

代码的改动只在于加入了mutex的概念，并调用了我们经过封装的P()和V()，这两个函数分别封装了sem_wait()和sem_post(), P/V源自于荷兰语，分别代表Proberen(测试)和Verhogen（增加)的意思。我自己改写的第一版的goodcnt.c中，把P/V加载了for循环中，这将大量增加CPU时间。第二版中在for意外加/减锁。 ###基于预线程化的并发服务器代码如下：

	/* 
	 * echoservert_pre.c - A prethreaded concurrent echo server
	 */
	/* $begin echoservertpremain */
	#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, port;
	    socklen_t clientlen=sizeof(struct sockaddr_in);
	    struct sockaddr_in clientaddr;
	    pthread_t tid; 
	
	    if (argc != 2) {
		fprintf(stderr, "usage: %s <port>\n", argv[0]);
		exit(0);
	    }
	    port = atoi(argv[1]);
	    sbuf_init(&sbuf, SBUFSIZE); //line:conc:pre:initsbuf
	    listenfd = Open_listenfd(port);
	
	    for (i = 0; i < NTHREADS; i++)  /* Create worker threads */ //line:conc:pre:begincreate
		Pthread_create(&tid, NULL, thread, NULL);               //line:conc:pre:endcreate
	
	    while (1) { 
		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 */ //line:conc:pre:removeconnfd
		echo_cnt(connfd);                /* Service client */
		Close(connfd);
	    }
	}
	/* $end echoservertpremain */

这个例子建立了主线程和工作线程。一个主线程用于接受连接并将得到的连接描述符放在一个共享缓冲区中。而每一个工作线程则反复的从共享缓冲区中取出描述符并为客户端服务，然后等待下一个描述符。这种情况适合于短连接。创建的工作线程创建完毕后其实是detach运行的，然后每个线程将调用sbuf_remove()从sbuf中取出connfd(连接字)，而后调用echo_cnt用于回写行并在终端打印出收到的字节数。运行结果:

	thread -1909590272 received 6 (258 total) bytes on fd 9
	thread -1909590272 received 5 (263 total) bytes on fd 9
	thread -1909590272 received 5 (268 total) bytes on fd 9
	thread -1892804864 received 6 (274 total) bytes on fd 6
	thread -1917982976 received 7 (281 total) bytes on fd 7
	thread -1901197568 received 5 (286 total) bytes on fd 8

需要初始化byte_cnt计数器和mutex信号量。pthread_once()函数调用初始化函数在这里是需要注意的。这使得函数包的调用更加容易。

Reading Digests for APP2E

Apr 1, 2014
Technology

###Preparation Download the files from the student’s website of app2e via:

	wget http://csapp.cs.cmu.edu/public/ics2/code.tar
	tar xvf code.tar

Start building the static libs and copy it to system library:

	[Trusty@XXXyyy lib]$ pwd
	/home/Trusty/code/app2e/practise/lib
	[Trusty@XXXyyy lib]$ cp /home/Trusty/code/app2e/code/src/csapp.c  .
	[Trusty@XXXyyy lib]$ cp /home/Trusty/code/app2e/code/include/csapp.h  .
	[Trusty@XXXyyy lib]$ gcc -c -o csapp.o csapp.c 
	[Trusty@XXXyyy lib]$ ar rcs libcsapp.a csapp.o
	[Trusty@XXXyyy lib]$ sudo cp libcsapp.a  /usr/lib/
	[Trusty@XXXyyy lib]$ sudo cp csapp.h  /usr/include/

Now you can directly use libcssapp.a in your own files:

#include <csapp.h>

int main(int argc, char **argv)
{
	int n;
	rio_t rio;
	char buf[MAXLINE];
	return 0;
}

Compile the file via:

	gcc -o test test.c

###Rio

#include <csapp.h>

int main(int argc, char **argv)
{
	int n;
	rio_t rio;
	char buf[MAXLINE];

	Rio_readinitb(&rio, STDIN_FILENO);
	while((n = Rio_readlineb(&rio, buf, MAXLINE)) != 0)
	{
		Rio_writen(STDOUT_FILENO, buf, n);
	}
	return 0;
}