linux网络编程学习笔记之五 —–并发机制与线程�

linux网络编程学习笔记之五 —–并发机制与线程�

大家好,又见面了,我是全栈君,祝每个程序员都可以多学几门语言。

进程线程分配方式

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

1、进程线程预分配

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

2、进程线程延迟分配

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

3、前面两者结合

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

多进程与多线程比較

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

线程池

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

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

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

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

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

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

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

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过一个函数的情况以外,仅仅有当前线程会运行。

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

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/118877.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)
blank

相关推荐

  • MySQL常用命令总结

    MySQL常用命令总结一.连接MySQL格式:mysql-h主机地址-u用户名-p用户密码或者:mysql-u用户名-p//回车后要求输入密码,密码不可见1、连接到本机上的MYSQL首先打开DOS窗口,然后进入目录mysql\bin,再键入命令mysql-uroot-p,回车后提示你输密码.注意用户名前可以有空格也可以没有空格,但是如果-p后带有用

  • k1145次列车经过站点_最小生成树和最短路径的区别

    k1145次列车经过站点_最小生成树和最短路径的区别北极的某区域共有 n 座村庄,每座村庄的坐标用一对整数 (x,y) 表示。为了加强联系,决定在村庄之间建立通讯网络,使每两座村庄之间都可以直接或间接通讯。通讯工具可以是无线电收发机,也可以是卫星设备。无线电收发机有多种不同型号,不同型号的无线电收发机有一个不同的参数 d,两座村庄之间的距离如果不超过 d,就可以用该型号的无线电收发机直接通讯,d 值越大的型号价格越贵。现在要先选择某一种型号的无线电收发机,然后统一给所有村庄配备,数量不限,但型号都是 相同的。配备卫星设备的两座村庄无论相距多远都可以直

  • Python线程指南[通俗易懂]

    Python线程指南[通俗易懂]本文介绍了Python对于线程的支持,包括“学会”多线程编程需要掌握的基础以及Python两个线程标准库的完整介绍及使用示例。注意:本文基于Python2.4完成;如果看到不明白的词汇请记得百度谷

  • JavaScript中location.hash详解「建议收藏」

    JavaScript中location.hash详解「建议收藏」原文地址:https://www.cnblogs.com/yeer/archive/2013/01/21/2869827.html去年9月,twitter改版。一个显著变化,就是URL加入了”#!”符号。比如,改版前的用户主页网址为  http://twitter.com/username改版后,就变成了  http://twitter.com/#!

  • 史记 乔布斯列传

    史记 乔布斯列传

  • 基于Vue的电商后台管理系统「建议收藏」

    基于Vue的电商后台管理系统「建议收藏」前端项目是基于Vue的SPA项目,前端技术栈采用Vue+VueRouter+Element-UI+Axios+Echarts。目前完成进度:登陆界面、登录和退出功能、导航守卫功能

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号