Unix/Linux编程：将线程用于并发(TCP)

太过爱你忘了你带给我的痛 2022-10-05 14:00 71阅读 0赞

# Linux线程概述 #

Linux线程具有如下特性：

*  **动态创建**：pthread\_create函数可以创建一个新线程。只是操作系统对并发线程数有一个上限限制，就像限制并发进程数量一样
 *  **并发执行**：所有线程就像在同一时间执行一样。在多处理器机器上，系统可以为每个线程分配一个CPU，多个线程可以并行执行
 *  **抢先**：操作系统为线程分配CPU资源。如果活动线程超过可用的CPU数(比如单处理器机器)，系统将自动在多个线程间调用CPU资源，每次只让一个线程执行一小段时间。API中有一个函数sched\_yield，线程一旦调用该函数，在其时间片用完之前就主动放弃CPU的调用
 *  **私有局部变量**：每个线程都有自己的私有堆栈。堆栈用于存放分配的局部变量和过程的活动记录(j即关于过程调用的信息)
 *  **共享全局变量**：一个进程内的所有线程共享一组全局变量
 *  **共享文件描述符**：一个进程内的所有进程共享一组文件描述符
 *  **协调和同步函数**：线程API中包括协调和同步线程执行的函数

## 优点 ##

*  更高的效率：
    
     *  效率的提高源于上下文交换的额外开销减少。
     *  上下文交换是指操作系统将CPU从一个运行线程调度到另一个线程所需执行的指令
     *  在线程间切换时，操作系统必须保存原先线程的状态(比如寄存器中的值)并读取新线程的状态
     *  同一进程中的线程共享的进程状态越多，操作系统需要改变的状态就越少。
     *  因此，同一进程中的两个线程的切换比不同进程中的两个线程快。
     *  尤其是因为同一进程中的线程共享一个存储器地址空间，进程内的线程切换就意味着操作系统不必改变虚拟存储器映射
 *  共享存储器
    
     *  并发服务器中多个服务器副本需要相互通信或者访问共享的数据，因此利用线程更容易构造并发服务器。
     *  利用线程也更容易构造监控系统。尤其是因为它们共享存储器，一个服务器中的从线程可将统计数据留在共享存储器中，从而使得监控线程可以把从线程的活动情况报告给系统管理员

## 缺点 ##

由于线程间共享存储器和进程状态，一个线程的动作可以会对同一个进程内的其他线程产生影响。比如，当两个线程试图同时访问同一个变量时，它们之间就会产生相互干扰

另一个缺点是缺乏健壮性。在单线程的服务器中，如果某一个并发服务器的副本造成服务器出错，操作系统只会终止引发故障的进程。但是在多线程服务器中，如果一个线程使服务器出错，操作系统将导致终止整个进程。

# 线程的协调和同步 #

由于各个线程按照自己的步调并发运行，线程间的同步是必须的。比如，如果一个线程在进行IO操作时被阻塞，延迟时间长度取决于操作系统和下层的硬件：无法预知在该线程的延迟期间其他线程的行为会在它们将执行哪些指令。因此，程序员希望能够协调线程的执行，从而引入了一些函数调用。Linux提供了三种同步机制：互斥mutex、信号量semaphore、条件变量conditionvarivable

## 互斥 ##

线程使用互斥可对共享数据段进行排他性访问。

通过调用pthread\_mutex\_init可以动态初始化一个互斥；程序员可以为每个需要保护的数据项都安排一个独立的互斥。

一个互斥初始化之后，线程在使用数据项之前必须调用pthread\_mutex\_lock，使用完之后在调用pthread\_mutex\_unlock。这样可以确保某一时刻只有一个线程访问该数据项。

在一个互斥中，第一个调用pthread\_mutex\_lock的线程将不受阻挡的继续执行。但在该线程使用数据项期间，后继调用pthread\_mutex\_lock的线程间被阻塞，直到第一个线程使用完数据项并调用了pthread\_mutex\_unlock之后，另一个线程才得以使用给数据项。此时，操作系统将使其中一个阻塞等待的线程回到运行状态。

## 信号量 ##

信号量是一种同步机制，它用于系统中有N个资源可用的情况，是对互斥机制的一种推广。信号量运行N个线程同时执行，而不是像互斥一样在某一个时刻只允许一个线程执行通过临时区

使用函数sem\_init动态初始化一个信号量：它带有一个参数N标识可用的资源。

初始化一个信号量之后，一个线程在使用一个资源前必须调用sem\_wait，并在用完之后调用sem\_post返还资源。N个线程都可在调用sem\_wait后不被影响------每个线程都可以在调用后继续执行。但是，如果再有其他线程调用sem\_wait，它们就将被阻塞。这些线程将一直出于阻塞状态，直到某个运行线程调用了信号量sem\_post后，其中一个阻塞的线程才得以继续运行。

## 条件变量 ##

实质上，只有一种情况需要条件变量

*  一组进程使用互斥对同一个资源提供排他性访问
 *  一旦某个线程获得资源，它需要等待一个特定的条件发生

如果没有条件变量，面对这种情况，程序必须使用一种忙等待(busy waiting)的形式：每个线程要重复的获得一个互斥、测试条件是否满足，然后释放互斥。条件变量允许线程原子的完整这两个动作，从而令等待更为高效。在被条件变量阻塞前，线程获得了一个互斥。在线程调用pthread\_cond\_wait以便等待某个条件变量时，线程同时指定了条件变量和所拥有的互斥。操作系统将同时释放线程拥有的互斥并阻塞线程。

线程执行pthread\_cond\_wait后将处于阻塞状态，直到其他线程给此变量发信号时才被唤醒(在一个条件变量上等待的线程可能在收到信号后解除阻塞。因此在实际应用中，程序员会设计一个循环来反复调用pthread\_cond\_wait，只有当线程等待的条件满足时才能退出循环)

给条件变量发送信号由两种形式，差别在于多个等待线程被处理的方式不同。即使有多个线程等待同一条件变量，函数pthread\_cond\_sigal只允许一个线程继续执行；pthread\_cond\_broadcast让所有被阻塞的线程都可以继续执行。在操作系统允许某个线程继续执行前，它将在线程被解除阻塞的同时再次获得阻塞前

# 使用线程的ECHO服务器 #

初始化之后，主线程执行主程序进入一个循环，在循环的accept调用处线程将被阻塞，等待一个TCP连接的到来。在连接到达时，主线程继续执行，调用pthread\_create创建一个新线程来处理程序。然后主程序继续循环，再次被阻塞而等待新连接的到来。新创建的线程执行TCPecho过程，该过程中包括一个循环，新线程将反复从TCP连接读数据，然后将数据返回发送者：

#include <sys/types.h>
    #include <sys/signal.h>
    #include <sys/socket.h>
    #include <sys/time.h>
    #include <sys/resource.h>
    #include <sys/wait.h>
    #include <sys/errno.h>
    #include <netinet/in.h>
    
    #define	QLEN		  32	/* maximum connection queue length	*/
    #define	BUFSIZE		4096
    
    #define	INTERVAL	5	/* secs */
    
    struct {
        
    	pthread_mutex_t	st_mutex;
    	unsigned int	st_concount;
    	unsigned int	st_contotal;
    	unsigned long	st_contime;
    	unsigned long	st_bytecount;
    } stats;
    
    void	prstats(void);
    int	TCPechod(int fd);
    int	errexit(const char *format, ...);
    int	passiveTCP(const char *service, int qlen);
    
    /*------------------------------------------------------------------------
     * main - Concurrent TCP server for ECHO service
     *------------------------------------------------------------------------
     */
    int
    main(int argc, char *argv[])
    {
        
    	pthread_t	th;
    	pthread_attr_t	ta;
    	char	*service = "echo";	/* service name or port number	*/
    	struct	sockaddr_in fsin;	/* the address of a client	*/
    	unsigned int	alen;		/* length of client's address	*/
    	int	msock;			/* master server socket		*/
    	int	ssock;			/* slave server socket		*/
    
    	switch (argc) {
        
    	case	1:
    		break;
    	case	2:
    		service = argv[1];
    		break;
    	default:
    		errexit("usage: TCPechod [port]\n");
    	}
    
    	msock = passiveTCP(service, QLEN);
    
    	(void) pthread_attr_init(&ta);
    	(void) pthread_attr_setdetachstate(&ta, PTHREAD_CREATE_DETACHED);
    	(void) pthread_mutex_init(&stats.st_mutex, 0);
    
    	if (pthread_create(&th, &ta, (void * (*)(void *))prstats, 0) < 0)
    		errexit("pthread_create(prstats): %s\n", strerror(errno));
    
    	while (1) {
        
    		alen = sizeof(fsin);
    		ssock = accept(msock, (struct sockaddr *)&fsin, &alen);
    		if (ssock < 0) {
        
    			if (errno == EINTR)
    				continue;
    			errexit("accept: %s\n", strerror(errno));
    		}
    		if (pthread_create(&th, &ta, (void * (*)(void *))TCPechod,
    		    (void *)ssock) < 0)
    			errexit("pthread_create: %s\n", strerror(errno));
    	}
    }
    
    /*------------------------------------------------------------------------
     * TCPechod - echo data until end of file
     *------------------------------------------------------------------------
     */
    int
    TCPechod(int fd)
    {
        
    	time_t	start;
    	char	buf[BUFSIZ];
    	int	cc;
    
    	start = time(0);
    	(void) pthread_mutex_lock(&stats.st_mutex);
    	stats.st_concount++;
    	(void) pthread_mutex_unlock(&stats.st_mutex);
    	while (cc = read(fd, buf, sizeof buf)) {
        
    		if (cc < 0)
    			errexit("echo read: %s\n", strerror(errno));
    		if (write(fd, buf, cc) < 0)
    			errexit("echo write: %s\n", strerror(errno));
    		(void) pthread_mutex_lock(&stats.st_mutex);
    		stats.st_bytecount += cc;
    		(void) pthread_mutex_unlock(&stats.st_mutex);
    	}
    	(void) close(fd);
    	(void) pthread_mutex_lock(&stats.st_mutex);
    	stats.st_contime += time(0) - start;
    	stats.st_concount--;
    	stats.st_contotal++;
    	(void) pthread_mutex_unlock(&stats.st_mutex);
    	return 0;
    }
    
    /*------------------------------------------------------------------------
     * prstats - print server statistical data
     *------------------------------------------------------------------------
     */
    void
    prstats(void)
    {
        
    	time_t	now;
    
    	while (1) {
        
    		(void) sleep(INTERVAL);
    
    		(void) pthread_mutex_lock(&stats.st_mutex);
    		now = time(0);
    		(void) printf("--- %s", ctime(&now));
    		(void) printf("%-32s: %u\n", "Current connections",
    			stats.st_concount);
    		(void) printf("%-32s: %u\n", "Completed connections",
    			stats.st_contotal);
    		if (stats.st_contotal) {
        
    			(void) printf("%-32s: %.2f (secs)\n",
    				"Average complete connection time",
    				(float)stats.st_contime /
    				(float)stats.st_contotal);
    			(void) printf("%-32s: %.2f\n",
    				"Average byte count",
    				(float)stats.st_bytecount /
    				(float)(stats.st_contotal +
    				stats.st_concount));
    		}
    		(void) printf("%-32s: %lu\n\n", "Total byte count",
    			stats.st_bytecount);
    		(void) pthread_mutex_unlock(&stats.st_mutex);
    
    	}
    }

# 监控 #

此服务器实例中实现了一个监控机制。监控程序使用一个执行prstats过程的独立线程实现，prstats中包含一个循环，在每一次循环中，监控程序打印连接的相关统计信息，然后睡眠INTERVAL秒。

监控线程和从线程使用一个共享的全局数据结构------stats相互通信。每个从线程给将各自连接的有关信息加入到stats结构中，而监控线程而INTERVAL秒提取一次信息。为确保任意时刻只能由一个线程访问共享结构，用到了锁。