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)


相关推荐

  • 此网页正试图从未经验证的来源加载脚本,怎么解决?

    此网页正试图从未经验证的来源加载脚本,怎么解决?

    2021年10月14日
  • C语言学习——位运算

    C语言学习——位运算原码反码补码介绍原码 就是前面所介绍的二进制定点表示法,即最高位为符号位,“ 0 ”表示正,“ 1 ”表示负,其余位表示数值的大小。反码 表示法规定:正数的反码与其原码相同;负数的反码是对其原码逐位取反,但符号位除外。补码 表示法规定:正数的补码与其原码相同;负数的补码是在其反码的末位加 1 。补码详细介绍补码是为了表示一个负数的二进制形式。其转化方式是,先将负数当成正数,转化成二进制…

  • python mkv转mp4,如何将mkv格式转换成mp4视频呢

    python mkv转mp4,如何将mkv格式转换成mp4视频呢在日常生活中都会使用到MKV视频文件的。MKV视频文件主要是视频文件、音频文件和字幕压制的。MKV视频一般在网上都是可以直接下载的。各种种子和磁链下载的也基本都是MKV视频。但有时可能会碰到视频播放错误。无法播放或者不支持文件播放的。一般都是可以通过转换视频格式修改的。那今天就教大家怎么将mkv格式转换成mp4格式吧。1、首先点击下方的立即下载按钮然后弹出下载迅捷视频转换器的下载框。下载打开之后,…

    2022年10月16日
  • cs与bs架构_cs架构和bs架构的应用

    cs与bs架构_cs架构和bs架构的应用CS和BS架构介绍

  • calendar 月份加减_calendar获取时间

    calendar 月份加减_calendar获取时间calendar类,日期加减Java代码public class test1 {        public static void main(String[] args) {            Calendar c = Calendar.getInstance();            int year=c.get(Calendar.YEAR);                 …

  • 360天擎卸载(2021年亲测有效)[通俗易懂]

    360天擎卸载(2021年亲测有效)[通俗易懂]360天擎卸载指南该方法面向个人电脑的,公司电脑安全模式下可能会蓝屏一、找到核心文件位置右键360天擎-文件位置->EntClient->conf->EntBase.dat[base]persistent_connetion=closeshow_tip=1net_env=1communication_interval=900[api_frequency]checkupdate=180get_client_tasks=180getconf=180s

发表回复

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

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