linux网络编程学习笔记之五 -----并发机制与线程池

进程线程分配方式

简述下常见的进程和线程分配方式：（好吧，我只是举几个例子作为笔记。。。并发的水太深了，不敢妄谈。。。）

1、进程线程预分配

简言之，当I/O开销大于计算开销且并发量较大时，为了节省每次都要创建和销毁进程和线程的开销。可以在请求到达前预先进行分配。

2、进程线程延迟分配

预分配节省了处理时的负担，但操作系统管理这些进程线程也会带来一定的开销。由此，有个折中的方法是，当某个处理需要花费较长时间的时候，我们创建一个并发的进程或线程来处理该请求。实现也很简单，在主线程中定时，定时到期，开新分支。

3、前面两者结合

还是举个例子，比如可以：在启动时不进行预分配，某个处理太长，则创建从进程，任务结束后不退出。

多进程与多线程比较

可以参考这篇论文：Linux下多进程和多线程性能分析  和这篇Blog：多进程or多线程  总结起来，在任务执行效率上，在任务量较大（文中单次5k以上），多进程的效率高点，反之，多线程站优势，但整体上出入不大。而在创建和销毁的效率上，线程的优势明显，约为5~6倍。然后在服务器上，并发量不大（小于几千），预先创建线程也没太大优势，因为动态管理线程的开销亦不可忽略。

线程池

比较全面和概要的介绍可以参看：线程池的介绍及简单实现

基本思路是，预先创建一定数量的线程，让它们处于阻塞状态，占用很小的内存空间。当任务到来时，选择一个空闲的线程执行，完成后线程不退出，继续阻塞。池子的创建销毁和管理有一个线程单独完成。

进一步地，动态地对线程的数量进行管理，负载较大时，增加线程数量。负载小时，减少之，设法让一段时间不活跃的线程退出，比如让线程在等待下一个请求前先启动一个定时器，若请求到达前定时器到期，则线程退出。

对于处理时间短，处理数目巨大的情况，线程池有天然优势。尤其是对性能要求高的应用或者突发性大规模请求，比如电商秒杀神马的。

线程池的实现可以参考libthreadpool，一个开源的库，sourceforge上能找到源码

我用简单的模型实现了一个，功能上基本满足，主从线程的流程如下图所示，唤醒空闲线程的时候不加以区分，即steven说的惊群，这会一定程度上的损失性能。与之对应的是有主线程采取一定的方式对空闲线程的唤醒进行调度以均衡负载和工作量。下面的代码是我的1.0版（改得太乱了，看官们勿怪），更进一步的功能，如动态地改变池子的尺寸，后续继续完善

pthread_pool.h

#ifndef _PTHREAD_POOL_H_
#define _PTHREAD_POOL_H_

#include<stdio.h>
#include<unistd.h>
#include<stdlib.h>
#include<string.h>
#include<pthread.h>

#define MAX_PTHREAD_NUM 100

typedef struct task_node{
	void * (*func)(void * p);
	void * arg;
	struct task_node* next;
}task_node;

typedef struct p_pool{
	pthread_cond_t cond;
	pthread_mutex_t mutex;
	task_node *head, *tail;
	pthread_t *p_tid;			//mark , unsigned long
	int max_num;
	int current_num;
	int working_num;
	int if_destory;
	int decrease;
}p_pool;

p_pool * pool = NULL;

void pool_init(int pthread_num);
void *pthread_process(void *arg);
int add_task(void * (*func)(void *p), void *arg);
void pool_destory();

#endif

pthread_pool.c

/*
 *   a simple thread pool
 *				Mon Jun 9 21:44:36 CST 2014
 *              by  Simon Xia
 */
#include"pthread_pool.h"

/* each worker thread's thread function to handle the task */
void *pthread_process(void *arg)
{
	task_node *tmp = NULL;

//	printf("Now in the %lu thread\n", pthread_self());
	while (1) {
		pthread_mutex_lock(&pool->mutex);
//		printf("%lu thread get lock\n", pthread_self());

		while (!pool->head && !pool->if_destory/* && !pool->decrease*/) { //While !  用是否有任务来控制
//			printf("%lu thread will wait\n", pthread_self());
			pthread_cond_wait(&pool->cond, &pool->mutex);
		}

//		printf("%lu thread: signal is coming\n", pthread_self());
		if (pool->if_destory /*|| pool->decrease*/)
			break;

		tmp = pool->head;

		pool->head = pool->head->next;
	//		pool->working_num++;

		pthread_mutex_unlock(&pool->mutex);
//		printf("%lu thread pick task from queue\n", pthread_self());
		(tmp->func)(tmp->arg);
//		printf("%lu thread finish task\n", pthread_self());
		free(tmp);
		tmp = NULL; //mark

		/*
			pthread_mutex_lock(&pool->mutex);
			pool->working_num--;
			pthread_mutex_unlock(&pool->mutex);
			*/

	}
	pthread_mutex_unlock(&pool->mutex);//先解锁！！
	printf("%lu thread will exit\n", pthread_self());
	pthread_exit(NULL);
}

/* main thread function to manage the thread pool */
/*
void *pthread_main(void *arg)
{
	printf("This is main thread\n");
	int i;
	while (1)
	{
		usleep(50000);
		pthread_mutex_lock(&pool->mutex);
		if (pool->if_destory)
			break;
		if (pool->working_num == pool->current_num) {
			for (i = pool->current_num; i < 2 * pool->current_num; i++)
				pthread_create(&pool->p_tid[i], NULL, pthread_process, NULL);
			pool->current_num *= 2;
			printf("The number of thread has been enlarged to %d\n", pool->current_num);
		}
		else if (pool->working_num <= pool->current_num / 4){
			pool->decrease = 1;
			pthread_mutex_unlock(&pool->mutex);
			for (i = 0; i < pool->current_num / 2; i++)
				pthread_cond_signal(&pool->cond);
			pool->current_num /= 2;
			pool->decrease = 0;
			printf("The number of thread has been decrease to %d\n", pool->current_num);
		}
		pthread_mutex_unlock(&pool->mutex);
	}
	pthread_exit(NULL);
}
*/

/* Initialize the thread pool
 * Input: number of worker thread
 */
void pool_init(int pthread_num)
{
	int i = 0;

	pool = (p_pool*)malloc(sizeof(p_pool));
	pthread_mutex_init(&pool->mutex, NULL);
	pthread_cond_init(&pool->cond, NULL);
	pool->head = pool->tail = NULL;
	pool->max_num = MAX_PTHREAD_NUM;
	pool->current_num = pthread_num;
	pool->working_num = 0;
	pool->if_destory = 0;
	pool->decrease = 0;
	pool->p_tid = (pthread_t*)malloc(pthread_num * sizeof(pthread_t));

//	pthread_create(&pool->p_tid[i], NULL, pthread_main, NULL);

	for (i = 0; i < pthread_num; i++)
		pthread_create(&pool->p_tid[i], NULL, pthread_process, NULL);

}

/* add task into task queue */
int add_task(void * (*func)(void *p), void *arg)
{
	task_node *tmp = (task_node*)malloc(sizeof(task_node));
	tmp->func = *func; //Mark
	tmp->arg = arg;
	tmp->next = NULL;

	pthread_mutex_lock(&pool->mutex);
	if (pool->head) {
		pool->tail = pool->tail->next = tmp;
	}
	else {
		pool->tail = pool->head = tmp;
	}

	pthread_mutex_unlock(&pool->mutex);
	//不加不行？
	//printf("Add task %d success!\n",*(int*)tmp->arg);
	//sleep(1);
	pthread_cond_signal(&pool->cond);

	tmp = NULL; //can't free
	return 0;
}

/* destory the pool after all work */
void pool_destory()
{
	int i;

//	pthread_mutex_lock(&pool->mutex);
	pool->if_destory = 1;
//	pthread_mutex_unlock(&pool->mutex);

	pthread_cond_broadcast(&pool->cond);

	for (i = 0; i < pool->current_num; i++)
	{
		if (!pthread_join(pool->p_tid[i], NULL))
			printf("Success to collect thread %lu\n", pool->p_tid[i]);
		else
			printf("Fail to collect thread %lu\n", pool->p_tid[i]);
	}

	free(pool->p_tid);
	free(pool->head);
	free(pool->tail);

	pthread_cond_destroy(&pool->cond);
	pthread_mutex_destroy(&pool->mutex);

	free(pool);
	pool = NULL;
}

基于这个线程池的服务端程序：

#include"simon_socket.h"

#define SERV_PORT 12345
#define THREAD_CNT 10

extern void pool_init(int );
extern int add_task(void* (*) (void*), void*);
extern void pool_destory();

typedef struct client_info{
	int fd;
	struct sockaddr_in addr;
	struct client_info *next;
}client_info;

void *process(void *arg)
{
	process_client(((client_info*)arg)->fd, &((client_info*)arg)->addr);
	return NULL;
}

void sig_int(int signo)
{
	pool_destory();
	exit(0);
}

int main()
{
	int sockfd, acfd;
	size_t sin_len;
	struct sockaddr_in client_addr;
	client_info *info_tmp, *info_head = NULL, *info_tail = NULL;

	signal(SIGINT, sig_int);

	sin_len = sizeof(struct sockaddr);
	sockfd = init_tcp_psock(SERV_PORT);
	pool_init(THREAD_CNT);

	for ( ; ; )
	{
		if ((acfd = accept(sockfd, (struct sockaddr *)&client_addr, &sin_len)) == -1)
		{
			perror("Accept request failed: ");
			return 1;
		}
		else
			printf("Get a connection from %s:%d !\n", inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port));

		info_tmp = (client_info *)malloc(sizeof(client_info));
		memset(info_tmp, 0, sizeof(client_info));
		info_tmp->fd = acfd;
		info_tmp->addr = client_addr;
		info_tmp->next = NULL;

		if (info_head) {
			info_tail = info_tail -> next = info_tmp;
		}
		else{
			info_head = info_tail = info_tmp;
		}
		add_task(process, (void*)info_tmp);
	}

	for (info_tmp = info_head; info_tmp; free(info_tmp))
		info_head = info_head -> next;

	return 0;
}

吐槽下多线程的调试。。。gdb调试多线程有点蛋疼，单步时很容易pthread_cond_wait把pthread_cond_signal的信号错过，出现各种错误，比如Cannot find bounds of current function等，建议大家还是多做输出，或者用日志的方式调。

set scheduler-locking off|on|step  这个命令还是很有用的，因为用step或者continue命令调试当前被调试线程的时候，其他线程也是同时执行的。

具体地：off 不锁定任何线程，也就是所有线程都执行，这是默认值。 on 只有当前被调试程序会执行。 step 在单步的时候，除了next过一个函数的情况以外，只有当前线程会执行。

反正我最后是老老实实输出。。各位路过大牛有什么好方法，求指点 ~~

linux网络编程学习笔记之五 -----并发机制与线程池