服务器项目关键组件

2017-08-31

定时器

话说网络程序通常需要处理的三类事件:第一是IO事件,第二是信号,第三就是定时事件,比如定期检测一个客户连接的状态,定时指的是一段时间后触发某段代码的机制.
为了实现一个定时器容器,我们选择用链表来实现,可按触发时间来排序存储,每个节点都是一个定时器,储存着:

触发时间和触发事件
所处理的对象
上一个和下一个定时器的位置

再设计一个脉搏函数,每次间隔最小时间后就调用,然后将容器中所有到时的定时器触发.

时间轮

如果按上述方式实现定时器,则插入的时间为O(n),效率随着定时器增加而降低,所以我们根据哈希的思想,用空间换时间,改用时间轮的方法.
用两个轮,小轮走一圈,大轮走一格,每个槽上都有一个链表,存储时间是该轮整数倍的定时器.

时间轮

实现代码:

//实现一个时间轮定时器,按超时时间排一圈,每个刻度上都有一个链表,存储所有
#include<time.h>
#include<net_h.h>
#include<algorithm>
#define BUFFER_SIZE 64
class tw_timer;
//用户数据结构,客户端地址,socket文件描述符,读缓冲区和定时器
struct client_data{
	sockaddr_in address;
	int sockfd;
	char buf[BUFFER_SIZE];
	tw_timer* timer;
};
class tw_timer{
public:
	tw_timer():rotation(-1),time_slot(-1),prev(nullptr),next(nullptr){}
	//tw_timer(tw_timer&);
	tw_timer(int ro,int ts):rotation(ro),time_slot(ts),prev(nullptr),next(nullptr){}
private:
	//时间轮转多少圈后生效
	int rotation;
	//属于哪个槽
	int time_slot;
	void (*cb_func)(client_data*);
	client_data* user_data;
	tw_timer* prev;
	tw_timer* next;
};
class tw_wheels{
public:
	tw_wheels():curr_slot(-1){
		fill_n(this->wheel,nullptr,sizeof(wheel));
	}
	~tw_wheels();
//	void addtimer(tw_timer&);
	void addtimer(tw_timer* ,tw_timer* );
	void del_timer(tw_timer*);
	void work(int ,int);//让某一个槽上的所有定时器工作
	void set_timer(tw_timer,int time_slot,int rotation);
	void tick();
private:
	//void addtimer(tw_timer* head,);
	//释放一条链表上的定时器
	void release(tw_timer*);
	static const int K=2;//多少轮子
	static const int N=60;//每个轮子多少个槽
	static const int SI=1;//最底层轮子的槽间隔为1s
	tw_timer* wheels[K][N];
	int curr_slots[K];//当前指向的槽
};
tw_wheels::~tw_wheels(){
	for(int i=0;i<K;i++){
		for(int j=0;j<N;j++)
			this->release(wheels[i][j]);
			wheels[i][j]=nullptr;
	}
}
void tw_wheel::release(tw_timer* head){
	if(head==nullptr)return;
	while(head){
		tw_timer* temp=head->next;
		del_timer(head);
		head=temp;
	}
}
void tw_wheels::addtimer(int timeout,void(*func)(client_data*)){
	int rot=0,ts=0,i;
	for( i=K-1;i>=0;i--){
		if(timeout+curr_slots[0])%pow(N,i)==0){
			
			int time=(timeout/pow(N,i));
			rot=time/N;
			ts=(time+this->curr_slots[i])%N;
			break;
		}
	}
	tw_timer* new_timer=new tw_timer(rot,ts);
	new_timer->cb_run=func;
	addtimer(i,new_timer);
}
void tw_wheels::addtimer(int wheelsize,tw_timer* target){
	int slotpos=target->time_slot;
	if(wheels[wheelsize][slotpos]==nullptr){
		wheels[wheelsize][slotpos]=target;
	}else{
		tw_timer* p=wheels[wheelsize][slotpos];
		wheels[wheelsize][slotpos]=target;
		target->next=p;
		p->prev=target;
	}
}
void tw_wheels::del_timer(tw_timer* obj){
	if(obj==nullptr)return;
	if(obj->prev){
		obj->prev->next=obj->next;
	}
	if(obj->next){
		obj->next->prev=obj->prev;
	}
	delete obj;
}
void tw_wheels::tick(){
	int count=1;
	for(int i=0;i<K;i++){
		if(count>0){
		curr_slots[i]=(curr_slots[i]+count)%N;
		work(i,curr_slots[i]);
		}
		count=(curr_slots[i])/N;
	}
}
void tw_wheels::work(int wheelsize,int slotpos){
	tw_timer* cur=wheels[wheelsize][slotpos];
	while(cur){
		if(cur->rotation==0){
			cur->cb_func(cur->user_data);
			tw_timer* p=cur;
			cur=cur->next;
			del_timer(p);
		}else{
			cur->rotation--;
			cur=cur->next;
		}
	}
}

除了时间轮之外,还有的实现方法为时间堆.

信号量

信号量是用来解决多进程之间的同步问题,总共有三个系统调用:semget,semop,semctl

1 2	#include<sys/sem.h> int semget(key_t key,int num_sems,int sem_flags);

key参数用来全局唯一表示一个信号量集,要通过信号量通信的进程需要使用相同的键值来创建/获取该信号量.
num_sems指定要创建/获取的信号量集中信号量的数目,获取信号量可设置0\
sem_flags指定一组标志
semget成功返回一个正整数值,他是信号量集的标示符,创建信号量集内核中关联的数据结构体semid_ds将被创建:

#include<sys/sem.h>
struct ipc_perm{
	key_t key;//键值
	uid_t uid;//所有者的有效用户id
	gid_t gid;//所有者的有效组id
	uid_t cuid;
	gid_t cgid;
	mode_t mode;//访问权限
};
struct semid_ds{
	struct ipc_perm sem_perm;//信号量的操作权限
	unsigned long int sem_nsems;//信号量数目
	time_t sem_otime;//最后一次调用semop的时间
	time_t sem_ctime;//最后一次调用semctl的时间
};

semop系统调用改变信号量的值,主要改变内核中如下变量:
unsigned short semval;//信号量的值
unsigned short semzcnt; //等待信号量值变为0的进程数量
unsigned short semncnt;//等待信号量
pid_t sempid;//最后一次执行semop操作的进程ID

1 2	#include<sys/sem.h> int semop(int sem_id,struct sembuf* sem_ops,size_t num_sem_ops);

进程池

实现一个进程池,池中所有的子进程都运行着相同的代码,并具有相同的属性,比如优先级,PGID,因为进程池在服务器启动之初就创建好了
主进程选择子进程算法有随机选取和轮流选取算法
主进程管理所有监听socket,如果有新的连接,通知子进程调用accept接受,无需传递连接socket
我们用一个全局管道来统一事件源,为了让进程收到的信号不中断进程,用管道接受信号,然后注册到epoll内核事件表一起监听.
每个子进程都有一个管道用来与父进程通信,主要是父进程通过管道通知子进程有新的连接到来.

#ifndef PROC_POOL_H
#define PROC_POOL_H
#include"./server_head.h"
//描述一个子进程的类
class process{
public:
	process():m_pid(-1){}
	~process(){}
private:
	pid_t m_pid;//自己的进程id
	int m_pipefd[2];//用来和父进程通信的管道
};
template<typename T>
class processpool{
private:
	processpool(int listenfd,int process_number=8);//单例模式,构造函数私有
public:
	static processpool<T>* create(int listenfd,int process_number){
		if(m_instance==nullptr){
			m_instance=new processpool<T>(listenfd,process_number);
		}
		return m_instance;
	}
	~processpool(){
		delete []this->m_sub_proc;
	void run();//启动进程池
	void setup_sig_pipe();
	void run_child();
	void run_parent();
pirvate:
	static const int MAX_PROCESS_NUMBER=16;//进程池最大子进程数量
	static const int USER_PER_PROCESS=65536;//每个子进程最多能处理的客户数量
	static const int MAX_EVENT_NUMBER=10000;//epoll最多能处理的事件数
	int m_process_number;//进程池中的进程总数
	int m_idx;//子进程在池中的序号
	int m_epollfd;//每个进程都有一个内核事件表,
	int m_listenfd;//监听socket
	bool m_stop;//是否运行
	process* m_sub_proc;
	static processpool<T>* m_instance;
}:
template<typename T>
processpool<T>::m_instance=nullptr;//静态成员函数初始化
static int sig_pipefd[2];//用于处理进程收到信号的管道(和从父进程收到的信号分开),以实现统一事件源,信号管道
//一些功能函数
//设置fd为非阻塞
 static int setnonblocking(int fd){
	 int old_option=fcntl(fd,F_GETFL);
	 fcntl(fd,F_SETFL,old_option|O_NONBLOCK);
	 return old_option;
 }
static void addfd(int m_epollfd,int fd){
	epoll_event event;
	event.data.fd=fd;
	event.event=EPOLLIN|EPOLLET;
	epoll_ctl(m_epollfd,EPOLL_CTL_ADD,fd,&event);
	setnonblocking(fd);
}
static void removefd(int epollfd,int fd){
	epoll_ctl(epollfd,EPOLL_CTL_DEL,fd,0);
	close(fd);
}
static void sig_add(int sig,void(*handler)(int),bool restart=true){
	struct sigaction si;
	memset(&si,0,sizeof(si));
	si.sa_handler=handler;
	si.sa_flags|=restart?SA_RESTART:0;
	sigfillset(&si.sa_mask);
}	
static void sig_handler(int sig){
	int save_errno=errno;
	send(sig_pipefd[0],static_cast<char*>&sig,1,0);//?为什么只传1个字节
	errno=save_errno;
}
template<typename T>
void processpool<T>::processpool(int listenfd,int process_number):
	m_listenfd(listenfd),m_process_number(process_number),
	m_idx(-1),(m_stop(false){
		assert(process_number>0 && process_number<=MAX_PROCESS_NUMBER);
		m_sub_proc=new process[process_number];
		for(int i=0;i<process_number;i++){
			sockpair(PF_UNIX,SOCK_STREAM,0,m_sub_proc[i].m_pipefd);
			m_sub_proc[i].m_pid=fork();
			if(m_sub_proc[i].m_pid==0){
				m_idx=i;
				close(m_sub_proc[i].m_pipefd[0]);//关闭写端
				break;
			}else{
				
				close(m_sub_proc[i].m_pipefd[1])
			}
		}		
}
template<typename T>
processpool<T>::setup_sig_pipe(){
	this->epollfd=epoll_create(5);
	assert(this->epollfd!=-1);
	assert(sockpair(PF_UNIX,SOCK_STREAM,0,sig_pipefd));
	setnonblocking(sig_pipefd[1]);
	addfd(m_epollfd,sig_pipefd[0]);
	sig_add(SIGCHLD,sig_handler);
	sig_add(SIGTERM,sig_handler);
	sig_add(SIGINT,sig_handler);
	sig_add(SIGPIPE,SIG_IGN);
template<typename T>
void processpool<T>::run(){
	if(this->m_idx!=0){
		run_child();
	}else{
		run_parent();
	}
}
//子进程调用epoll_wait,监听到的事件分类处理.
template<typename T>
void processpool::run_child(){
	this->setup_sig_pipe();//第一步统一事件源
	epoll_event events[MAX_EVENT_NUMBER];
	T* user=new T[USER_PER_PROCESS];
	int pipefd=this->m_sub_proc[m_idx].m_pipefd[0];
	while(!this->m_stop){
		int number=epoll_wait(this->m_epollfd,events,MAX_EVENT_NUMBER,-1);
		if(number<0&&(errno!=EINTER){
			printf("epoll failure\n");
			break;
		}
		for(int i=0;i<number;i++){
			int sockfd=events[i].data.fd;
			if((sockfd==pipefd)&&(events[i].events &EPOLLIN){//管道全双工,我们处理读事件
				int client=0;
				recv(sockfd,(char*)&client,sizeof(client),0);
				struct sockaddr_in client_address;
				socklen_t client_addrlength=sizeof(client_address);
				int connfd=accept(this->m_listenfd,(struct sockaddr*)&client_address,&client_addrlength);
				addfd(this->m_epollfd,connfd);
				user[connfd].init(m_epollfd,connfd,client_address);//具体处理方法的初始化
			}else if((sockfd==sig_pipefd[1])&&(events[i].event & EPOLLIN)){
				//处理信号,可能一次收到多个信号
				int sig;
				char signals[1024];
				int ret=recv(sockfd,signals,sizeof(signals),0);
				for(int i=0;i<ret;i++){
					switch(signals[i]){//信号占一个字节
						case SIGCHLD://子进程终止信号
						{
							pid_t pid;
							int stat;
							while((pid=waitpid(-1,&stat,WNOHANG) )>0){
								continue;
							}
							break;
						}
						case SIGTERM:
						case SIGINT://终止信号
						{
							this->m_stop=true;
							break;
						}
						default:
						{
							break;
						}
					}
				}
			}
			else if(events[i].events & EPOLLIN){//其他类型描述符,肯定是已连接描述符上请求到来
				user[sockfd].process();
			}else{
				continue;
			}
		}
	}
	delete[] user;		
	user=nullptr;
	close(pipefd);
	close(m_epollfd);
}
template<typname T>
void processpool<T>::run_parent(){
	this->setup_sig_pipe();
	addfd(this->m_epollfd,this->m_listenfd);
	epoll_event events[USER_PER_PROCESS];
	int new_connfd=1;
	int sub_process_counter=0;
	while(!m_stop){
		int number=epoll_wait(this->m_epollfd,events,MAX_EVENT_NUMBER,-1);
		if((number<0)&&(errno!=EINTER)){
			printf("epoll failure\n");
			break;
		}
		for(int i=0;i<number;i++){
			int sockfd=events[i].data.fd;
			if((sockfd==this->m_listenfd){
				//接下来通过随机选取算法选择一个子进程通知.
				int i=sub_process_counter;
				do{
					if(m_sub_process[i].m_pid!=-1){
						break;
					}
					i=(i+1)%m_process_number;
				}while(i!=sub_process_counter);
				if(m_sub_process[i].m_pid==-1){
					m_stop=true;
					break;
				}
				sub_process_counter=(i+1)%m_process_number;
				send(m_sub_process[i].m_pipefd[0],(char*)&new_coon,sizeof(new_conn),0);//写入一个数值表示有新连接
				printf("send requeset to child %d\n",i);
			}
			else if((sockfd=sig_pipefd[1])&&(events[i].event&EPOLLIN){//处理自己收到的信号
				int sig;
				char signals[1024];
				ret=recv(sig_pipefd[0],signals,sizeof(signals),0);
				if(ret<=0){
					continue;
				}else{
					for(int i=0;i<ret;i++){
					switch(signals[i]){
						case SIGCHLD:
						{
							pid_t pid;
							int stat;
							while((pid=waitpid(-1,&stat,WNOHANG) )>0){
								for(int i=0;i<m_process_number;i++){//关闭通信管道的写端
									if(this->m_sub_process[i].m_pid==pid){
										close(this->m_sub_process.m_pipefd[0]);
										m_sub_process[i].m_pid=-1;
									}
								}
							}
							//如果没有子进程存在,则父进程也退出
							m_stop=true;
							for(int i=0;i<process_number;i++){
								if(m_sub_process[i].m_pid!=-1){
									m_stop=false;
									break;
								}
							}
							break;
						}
						case SIGTERM:
						case SIGINT:
						{
							printf("kill all child now\n");
							for(int i=0;i<m_process_number;i++){
								int pid=m_sub_process[i].m_pid;
								if(pid!=-1){
									kill(pid,SIGTERM);
								}
							}
							break;
						}
						default:
							break;
					}
					}
				}
			}
			else{
				continue;
			}
		}
	}
	close(m_epollfd);
}
#endif

线程同步

为了使用方便,我们将posix线程标准下用来同步线程的三种锁进行封装,信号量,互斥量,条件变量

#ifndef LOCKER_H
#define LOCKER_H
#include"./server_head.h"
class cond;
class sem{
public:
	sem(){
		if(sem_init(&m_sem,0,0)!=0){
			throw std::exception();
		}
	}
	~sem(){
		sem_destroy(&m_sem);
	}
	bool wait(){
		return sem_wait(&m_sem)==0;
	}
	bool post(){
		return sem_post(&m_sem)==0;
	}
private:
	sem_t m_sem;
};
class locker{
public:
	friend class cond;
	locker(){
		init();
	}
	~locker(){
		release();
	}
	void init(){
		if(pthread_mutex_init(&m_mutex,NULL)!=0)
			throw std::exception();
	}
	void release(){
		pthread_mutex_destroy(&m_mutex);
	}
	bool lock(){
		return pthread_mutex_lock(&m_mutex)==0;
	}
	bool unlock(){
		return pthread_mutex_lock(&m_mutex)==0;
	}
private:
	pthread_mutex_t m_mutex;
	pid_t m_pid;
};
class cond{
public:
	cond(){
		m_lock.init();
		if(!init()){
			m_lock.release();
		}
	}
	~cond(){
		m_lock.release();
		this->release();
	}
	bool init(){
		return pthread_cond_init(&m_cond,NULL)==0;
	}
	void release(){
		pthread_cond_destroy(&m_cond);
	}
	bool signal(){
		return pthread_cond_signal(&m_cond)==0;
	}
	bool broadcast(){
		return pthread_cond_broadcast(&m_cond)==0;
	}
	bool wait(){
		int ret=0;
		m_lock.lock();
		ret=pthread_cond_wait(&m_cond,&m_lock.m_mutex);
		m_lock.unlock();
		return ret==0;
	}
private:
	locker m_lock;
	pthread_cond_t m_cond;
};
#endif

mhcoderwl

服务器项目关键组件

定时器

时间轮

信号量

进程池

线程同步

线程池