进程间传递文件描述符
unix一个传统的服务器模型就是一连接一进程模型。进程对地址空间的保护作用是显而易见的,为某个连接服务的进程出现错误导致崩溃,其它的进程还能正常的运行。但是,当连接数大时,
创建过多的进程显然会影响效率。那么启动一系列的进程,每个进程都利用epoll为多个连接服务,不是就可以避免创建过多的进程,同时也利用到了进程地址空间保护的优势了吗?
但是还有一个问题,我们不能为每一个进程都创建一个监听套接口。
unix系统还有一个特性,就是可以将一个文件描述符传递给其它进程,方法有很多,unix域套接字是其中一种办法。
利用unix域套接字传递文件描述符,就可以用一个进程监听套接口,将到来的连接传递给其它进程,由其它进程提供后续的服务。
本文介绍的模型主要结构是,一个listener进程,在启动的时候创建数个data_worker进程,data_worker进程利用epoll为多个连接服务.listerner阻塞在listen调用上,每当一个新的连
接到来,就将accept到的套接字传递给一个data_worker,由data_worker为连接做后续服务.
因为data_worker是利用了epoll为多个连接服务,所以避免了单进程单连接的确点。同时,一个data_worker的崩溃并不会影响到其它正常的data_worker.只要主控进程重新启动一个
data_worker便可以了。
epoll.h
// C/C++ header file
// Author: root
// File: Epoll.h
// Created: 23:34:29 2008-05-05
// Modified: 05:14:35 2008-10-28
// Brief:
#ifndef _EPOLL_H
#define _EPOLL_H
#include <sys/epoll.h>
#include <error.h>
#include <stdio.h>
#include "noncopyable.h"
#include "singleton.h"
#define MAX_EPOLL 4096
class Epoll : private noncopyable
{
public:
Epoll():current_Size(0){}
void initEpoll(unsigned int _max)
{
this->_max = _max;
epfd = TEMP_FAILURE_RETRY(epoll_create(_max));
}
Epoll(unsigned int _max):_max(_max)
{
epfd = TEMP_FAILURE_RETRY(epoll_create(_max));
}
~Epoll()
{
TEMP_FAILURE_RETRY(close(epfd));
}
bool isFull()
{
return current_Size >= _max;
}
bool addEpoll(int fd,int eEvent)
{
if(current_Size >= _max)
return false;
epoll_event ev;
ev.data.fd = fd;
ev.events = eEvent;//EPOLLIN | EPOLLET;
int ret;
TEMP_FAILURE_RETRY(ret = epoll_ctl(epfd,EPOLL_CTL_ADD,fd,&ev));
++current_Size;
return true;
}
void delEpoll(int fd)
{
epoll_event ev;
ev.data.fd = fd;
TEMP_FAILURE_RETRY(epoll_ctl(epfd,EPOLL_CTL_DEL,fd,&ev));
--current_Size;
}
/* * para maxWait: -1:无限等待,0:马上返回,>0:等待一定时间 */
int wait(epoll_event *events,int maxWait = -1)
{
return TEMP_FAILURE_RETRY(epoll_wait(epfd,events,_max,maxWait));
}
private:
int epfd;
unsigned int current_Size;
unsigned int _max;
};
#endif
listener.cpp
复制代码
#include "SocketWrapper.h"
int main(int argc, char ** argv)
{
if(argc < 2)
{
printf("server port/n");
return 0;
}
//创建子进程,创建域套接字
int unsock = create_un_execl("./data_worker","data_worker");
struct sockaddr_in servaddr;
int fd;
if((fd = Socket_Bind_Listen(Inet,Stream,Tcp,NULL,atoi(argv[1]),servaddr,5)) > 0)
{
for( ; ;)
{
struct sockaddr_in cliaddr;
socklen_t len;
int connfd = Accept(fd,(struct sockaddr*)NULL,NULL);
if(connfd > 0)
{
char address[128];
unsigned short port;
getRemoteAddrPort(connfd,&cliaddr,&len,address,port);
//将套接字传递给data_worker
write_fd(unsock,(void*)"",1,connfd);
printf("ip:%s,port:%d/n",address,port);
//listener 可以关闭那个套接字了
close(connfd);
}
}
}
return 0;
}
data_worker.cpp
#include "SocketWrapper.h" #include "Epoll.h" int main(int arcv, char **argv) { int ufd=atoi(argv[1]);//unix domain 描述字 Epoll _epoll(100); _epoll.addEpoll(ufd,EPOLLIN); for( ; ; ) { epoll_event events[4096]; int nfds = _epoll.wait(events,-1); for( int i = 0; i < nfds; ++i) { if(events[i].events & EPOLLIN) { if(events[i].data.fd == ufd) { //listener传来一个连接,将这个连接添加到epoll中 int fd; char c; if(read_fd(ufd,&c,1,&fd) >= 0) { _epoll.addEpoll(fd,EPOLLIN); } } else { char buf[1024]; bzero(buf,sizeof(buf)); int fd = events[i].data.fd; int ret = read(fd,buf,1024); if(ret > 0) { struct sockaddr_in cliaddr; socklen_t len; char address[128]; unsigned short port; getRemoteAddrPort(fd,&cliaddr,&len,address,port); printf("from %s:%d,%s/n",address,port,buf); } else if(ret == 0) { _epoll.delEpoll(fd); close(fd); } } } } } }