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

时间：2014-07-24 22:15:12 阅读：283 评论：0 收藏：0 [点我收藏+]

标签：des blog http java io 2014 art for

进程线程分配方式

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

1、进程线程预分配

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

2、进程线程延迟分配

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

3、前面两者结合

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

多进程与多线程比較

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

线程池

比較全面和概要的介绍能够參看：线程池的介绍及简单实现

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

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

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

线程池的实现能够參考libthreadpool，一个开源的库，sourceforge上能找到源代码

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

bubuko.com,布布扣

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调试多线程有点蛋疼，单步时非常easypthread_cond_wait把pthread_cond_signal的信号错过，出现各种错误，比方Cannot find bounds of current function等，建议大家还是多做输出，或者用日志的方式调。
set scheduler-locking off|on|step 这个命令还是非常实用的，由于用step或者continue命令调试当前被调试线程的时候，其它线程也是同一时候运行的。
详细地：off 不锁定不论什么线程，也就是全部线程都运行，这是默认值。 on 仅仅有当前被调试程序会运行。 step 在单步的时候，除了next过一个函数的情况以外，仅仅有当前线程会运行。