EPOLL单线程版本 基于reactor 的 httpserver文件下载 支持多个客户端同时处理

之前写了一个httpserver的问价下载服务器    如果有多个客户端请求过来只能串行处理必须得等当前的操作完成之后才会处理   

另外还存在 文件大的时候 会出错 处理不了  原因就是 sendfile是在一个while循环中处理的  

当调用send失败返回-1之后 就  结束了   而一般来讲  send的时候发送的数据超过内核中的send buffer的大小的时候  就会  失败了  

这个时候 必须 要保存下来当前文件的已发送的字节数 以及当前文件的偏移指针 等下一次 EPOLLOUT事件的时候再次 发送给客户端  

目前已经实现了这个功能 采用的是单线程版本的reactor模式  

支持 多个客户端同时下载文件 

还存在bug 但是  功能是有了  

#include 
#include 
 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
 
 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
 
#include 
 
 
 
 
 
typedef int (*READ_CB)(void *user_data);
typedef int (*WRITE_CB)(void *user_data);
typedef int (*ACCEPT_CB)(int epoll_fd,int fd,void *user_data);
 
 
 
#define READ_ONETIME	100
 
#define MAX_SESSIONS	1024
typedef struct
{
	int fd;
	int file_fd = -1;
	char write_buffer[1024];
	char read_buffer[1024];	
	int write_offset;
	int read_offset;
	int send_file_read_len = 0;
	char writeable;
	char is_dir;
	char head_has_send = 0;
	char file_path[512]={0};
	int file_size = 0;
	READ_CB read_cb;
	WRITE_CB write_cb;
	ACCEPT_CB accept_cb;
}Session;
 
 
typedef struct 
{
	int epoll_fd;
	int server_fd;
	int count;
	Session sessions[MAX_SESSIONS];
	
}Reactor;
 
 
int create_socket(bool is_tcp,bool block_mode,const char *led_ip,int port)
{
	#define LISTEN_BACKLOG 10
 
	int socket_fd ;
 
	
	const char *server_ip = led_ip;
 	struct sockaddr_in server_addr;
 
	if(is_tcp)
	{
		if(block_mode)
		{
			socket_fd = socket(AF_INET,SOCK_STREAM,0);	
		}
		else
		{
			socket_fd = socket(AF_INET,SOCK_STREAM|SOCK_NONBLOCK,0);
		}
 
	}
	else
	{
		if(block_mode)
		{
			socket_fd = socket(AF_INET,SOCK_DGRAM,0);	
		}
		else
		{
			socket_fd = socket(AF_INET,SOCK_DGRAM|SOCK_NONBLOCK,0);
		}
 
	}
	
	int opt = 1;
 
	if (socket_fd == -1) 
	{
		printf("Create socket error\n");
		goto ERROR;
	}
 
 
	setsockopt(socket_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof (opt));
 
 
 
	bzero(&server_addr,sizeof(server_addr));
 
	server_addr.sin_family = AF_INET;
	server_addr.sin_port = htons(port);
	inet_pton(AF_INET,server_ip,&server_addr.sin_addr);
 
 
	if (bind(socket_fd, (struct sockaddr *) &server_addr,sizeof(server_addr)) == -1)
	{
		printf("Bind error\n");
		goto ERROR;
	}
	
	if (listen(socket_fd, LISTEN_BACKLOG) == -1)
	{
		printf("listen error\n");
		goto ERROR;
	}
 
 
	
	return socket_fd;
 
	ERROR:
	if(socket_fd>0)
	{
		close(socket_fd);
	}
	return -1;
}
 
 
void set_nonblock(int fd)
{
	
	int opts=fcntl(fd, F_GETFL);	
	if(opts<0)	
	{	
		fprintf(stderr, "fcntl(sock,GETFL)\n");  
		return ;
	} 
	
	opts = opts|O_NONBLOCK;  
	if(fcntl(fd,F_SETFL,opts)<0)	
	{	
		fprintf(stderr, "fcntl(sock,SETFL,opts)\n");  
		return; 
	}	
 
}
 
 
 
int reactor_init(Reactor &rt,ACCEPT_CB accept_cb,READ_CB read_cb,WRITE_CB write_cb)
{
	rt.epoll_fd = epoll_create(10); 
 
	if(rt.epoll_fd == -1)
	{
		perror("epoll_create failed");
		return -1;
	}
 
	rt.server_fd = create_socket(true, true, "0,0,0,0", 1234);
	
	if(rt.server_fd == -1)
	{
		perror("create_socket failed");
		close(rt.epoll_fd);
		return -1;
	}
 
	struct epoll_event event;
 
	event.data.fd = rt.server_fd;
	event.events = EPOLLIN|EPOLLET|EPOLLOUT;
	int ret = epoll_ctl(rt.epoll_fd,EPOLL_CTL_ADD ,rt.server_fd,&event);
	if(ret == -1)
	{
		perror("epoll_ctl failed");
		close(rt.epoll_fd);
		close(rt.server_fd);		
		return -1;
	}
 
	for(int i = 0;i
	{
		rt.sessions[i].accept_cb = accept_cb;
		rt.sessions[i].read_cb = read_cb;		
		rt.sessions[i].write_cb = write_cb;				
	}
	
	rt.count = 0;
 
	printf("Reactor init success epollfd = %d serverfd = %d\n",rt.epoll_fd,rt.server_fd);
	return 0;
}
 
int reactor_run(Reactor &rt)
{
 
	struct epoll_event events[100];
	
	while(true)	
	{
		int ready_count = epoll_wait(rt.epoll_fd, events, 100, -1);
 
		//printf("ready_count = %d\n",ready_count);
		for(int i = 0;i
		{
			int index = events[i].data.fd;
 
			//printf("index = %d epollfd = %d events[i].data.fd = %d events=%08X\n",index,rt.epoll_fd,events[i].data.fd,events[i].events);			
			Session * session = &rt.sessions[index];
			
			if(events[i].data.fd == rt.server_fd)
			{
				printf("index = %d epollfd = %d cfd = %d\n",index,rt.epoll_fd,events[i].data.fd);
				session->accept_cb(rt.epoll_fd,events[i].data.fd,&rt);
			}
			else
			{
				if(events[i].events & EPOLLIN)
				{
					session->read_cb(session);
				}
				
				if(events[i].events & EPOLLOUT)
				{
					session->write_cb(session);
				}				
			}
		}
	}
}
 
int reactor_deinit(Reactor &rt)
{
	if(rt.epoll_fd >0)
	{
		close(rt.epoll_fd);
	}
	return 0;
}
 
 
 
 
int Accept_cb(int epoll_fd,int fd,void *user_data)
{
	if(fd > 0 && epoll_fd >0)
	{
 
		int cfd = accept(fd,NULL,NULL);
		if(cfd == -1)
		{
			perror("accept failed");
			return -1;
		}
 
	
		set_nonblock(cfd);
 
		printf("Accept_cb epollfd = %d cfd = %d\n",epoll_fd,cfd);
 
 
		struct epoll_event ev = {0};
		ev.data.fd = cfd;
		ev.events = EPOLLIN|EPOLLOUT|EPOLLET;
		
		int ret = epoll_ctl(epoll_fd,EPOLL_CTL_ADD,cfd,&ev);
		if(ret == -1)
		{
			perror("epoll_ctrl failed");
			return -1;
		}
 
		Reactor *reactor = (Reactor*)user_data;
		reactor->sessions[cfd].fd = cfd;
		//session->fd = cfd;
		return 0;
	}
 
	return -1;
}
 
 
void http_request(Session *session)
{
	char method[12]={0},path[512]={0},protocol[20]={0},headers[512]={0};
	printf("buf len[%d] content[%s]\n",session->read_offset,session->read_buffer);
 
	char *p = strstr(session->read_buffer,"\r\n\r\n");
 
 
	
	int ret = sscanf(session->read_buffer,"%[^ ] %[^ ] %[^ \r\n]%[^\r\n]",method,path,protocol,headers);
	printf("sscanf ret is %d headers is %s\n",ret,headers);
	if(ret !=3)
	{
		printf("Wait a whole http header\n");
		session->writeable = 0;
		return ;
	}
	else
	{
		printf("This is a whole http packet\n");
	}
 
	session->writeable = 1;
 
	session->read_offset = 0;
 
	if(strcasecmp(method,"get") == 0)
	{
		if(strcmp(path,"/") == 0)
		{	
		  strcpy(session->file_path ,"./");
		}
		else
		{
		  strcpy(session->file_path ,path+1);
		}
	 
		struct stat st;
	 
		int ret = stat(session->file_path,&st);
		if(ret == -1)
		{
		  printf("file doest not exist\n");
		  //SendHead(event,404,"Not found",GetFileType(".html"),-1);
		  //SendFile(event,"404.html");
		  session->is_dir = -1;
		  return ;
		}
	 
		if(S_ISDIR(st.st_mode))
		{
		  printf("Directory\n");
		  //SendHead(event,200,"OK",GetFileType(".html"),-1);
		  //SendDir(event,file);
		  session->is_dir = 1;
		}
		else
		{
		  printf("File\n");
		  session->file_size = st.st_size;
		  //SendHead(event,200,"OK",GetFileType(file),st.st_size);
		  //SendFile(event,file);
  		  session->is_dir = 0;
		}
 
	}
 
}
 
 
 
#define BURSIZE 1024
int hex2dec(char c)
{
	if ('0' <= c && c <= '9') {
		return c - '0';
	} else if ('a' <= c && c <= 'f') {
		return c - 'a' + 10;
	} else if ('A' <= c && c <= 'F') {
		return c - 'A' + 10;
	} else {
		return -1;
	}
}
 
char dec2hex(short int c)
{
	if (0 <= c && c <= 9) {
		return c + '0';
	} else if (10 <= c && c <= 15) {
		return c + 'A' - 10;
	} else {
		return -1;
	}
}
 
 
/*
 * 编码一个url
 */
void urlencode(char url[])
{
	int i = 0;
	int len = strlen(url);
	int res_len = 0;
	char res[BURSIZE];
	for (i = 0; i < len; ++i) {
		char c = url[i];
		if (('0' <= c && c <= '9') ||
				('a' <= c && c <= 'z') ||
				('A' <= c && c <= 'Z') || c == '/' || c == '.') {
			res[res_len++] = c;
		} else {
			int j = (short int)c;
			if (j < 0)
				j += 256;
			int i1, i0;
			i1 = j / 16;
			i0 = j - i1 * 16;
			res[res_len++] = '%';
			res[res_len++] = dec2hex(i1);
			res[res_len++] = dec2hex(i0);
		}
	}
	res[res_len] = '\0';
	strcpy(url, res);
}
 
/*
 * 解码url
 */
void urldecode(char url[])
{
    
	int i = 0;
	int len = strlen(url);
	int res_len = 0;
	char res[BURSIZE];
	for (i = 0; i < len; ++i) {
		char c = url[i];
		if (c != '%') {
			res[res_len++] = c;
		} else {
			char c1 = url[++i];
			char c0 = url[++i];
			int num = 0;
			num = hex2dec(c1) * 16 + hex2dec(c0);
			res[res_len++] = num;
		}
	}
	res[res_len] = '\0';
	strcpy(url, res);
}
 
const char *GetFileType(const char *filename)
{
    const char *dot = strrchr(filename,'.');
    if(dot == NULL)
    {
        return "text/plain; charset=utf-8";
    }
    if(strcmp(dot,".jpg") == 0 ||strcmp(dot,".jpeg") == 0)
    {
        return "image/jpg";
    }
    if(strcmp(dot,".html") == 0 ||strcmp(dot,".htm") == 0)
    {
        return "text/html; charset=utf-8";
    }    
    if(strcmp(dot,".png") == 0)
    {
        return "image/png";
    }    
    if(strcmp(dot,".bmp") == 0)
    {
        return "image/bmp";
    }        
    if(strcmp(dot,".gif") == 0)
    {
        return "image/gif";
    }            
    if(strcmp(dot,".css") == 0)
    {
        return "text/css";
    }           
    if(strcmp(dot,".mp3") == 0)
    {
        return "audio/mpeg";
    }               
 
    return "text/plain; charset=utf-8";
}
 
 
int SendHead(int cfd,int status ,const char *desc,const char *type,int size)
{
    char buf[4096] = {0};
    sprintf(buf,"http/1.1 %d %s\r\n",status,desc);
    sprintf(buf+strlen(buf),"content-type: %s\r\n",type);
    sprintf(buf+strlen(buf),"content-length: %d\r\n\r\n",size);    
 
    printf("SendHead buf[%s]\n",buf);
 
    return send(cfd,buf,strlen(buf),0);
}
 
 
int SendDir(Session *session,const char *dirname)
{
    char buf[4096] = {0};
 
    sprintf(buf,"",dirname);
 
    printf("SendDir dirname=[%s]\n",dirname);
    struct dirent **namelist;
    int count = scandir(dirname,&namelist,NULL,alphasort);
    printf("SendDir count=[%d]\n",count);
    for(int i = 0;i< count;i++)
    {
        char *name = namelist[i]->d_name;
        struct stat st;
        char sub_path[1024]={0};
        sprintf(sub_path,"%s/%s",dirname,name);
        stat(sub_path,&st);
        if(S_ISDIR(st.st_mode))
        {
            sprintf(buf+strlen(buf),
                "
",name,name,st.st_size);
        }
        else
        {
            sprintf(buf+strlen(buf),
                "
",name,name,st.st_size);
        }
 
        //printf("cfd:%d Sendbuf[%s]\n",cfd,buf);
        send(session->fd,buf,strlen(buf),0);
        memset(buf,0,sizeof(buf));
        free(namelist[i]);
    }
 
    sprintf(buf,"
%s
%ld
%s
%ld
");
 
    //printf("cfd:%d Sendbuf[%s]\n",cfd,buf);
 
    send(session->fd,buf,strlen(buf),0);
    free(namelist);
 
    return 0;
}
 
int SendFile(Session *session,const char* filename)
{
 
	if(session->file_fd == -1)
	{
		session->file_fd = open(filename,O_RDONLY); 		
	}
    
    if(session->file_fd >0)
    {
        #if 1
        while(1)
        {
            char buf[1024];
            int len = read(session->file_fd,buf,sizeof (buf));
            if(len >0)
            {
           		session->send_file_read_len+=len;
              	int ret = send(session->fd,buf,len,0);
				if(ret >0)
				{
					session->write_offset += ret;
					//printf("This time send [%d] total send [%d] bytes\n",ret,session->write_offset);
				}
				else if(ret ==0)
				{
					printf("Send file return 0 close socket this time len = %d total len = %d \n",len,session->send_file_read_len);
					close(session->file_fd);				
					close(session->fd);
				}
				else
				{
					int seek_ret = lseek(session->file_fd,session->write_offset,SEEK_SET);
					//printf("Seekret = %d session->writeoffset = %d\n",seek_ret,session->write_offset);
					if(seek_ret == -1)
					{
						perror("lseek failed");
					}
					session->send_file_read_len-=len;
					//printf("Send file return -1 wait next send this time len = %d total len = %d\n",len,session->send_file_read_len);
					return -1;
				}
            }
            else if(len == 0)
            {
                printf("Read file end this time len = %d total len = %d\n",len,session->send_file_read_len);
				close(session->file_fd);				
				close(session->fd);
 
				session->write_offset = 0;
				session->send_file_read_len = 0;
 
				session->fd = 0;
				session->file_fd = -1;
				session->writeable = 0;
				return 0;
                break;
            }
            else
            {
            	close(session->file_fd);				
				close(session->fd);
                perror("read error");
            }
        }
        #else
        off_t offset = 0;
        int file_size = lseek(fd,0,SEEK_END);
        lseek(fd,0,SEEK_SET);
 
        while(offset 
        {
            int send_len = sendfile(cfd,fd,&offset,file_size-offset);
            
            if(send_len == -1)
            {
                if(errno == EAGAIN)
                {
                    //perror("sendfile no data send");
                }
                else
                {
                    perror("sendfile ret -1");
                }
                
            }
            else
            {
                printf("Send len:%d\n",send_len);
            }
        }
        
        #endif
    }
    else
    {
        perror("open file failed");
    }
    //close(fd);
    return 0;
}
 
 
 
 
void http_response(Session *session)
{
 
	//printf("session->writeable = %d\n",session->writeable);
 
	if(session->writeable == 0)
	{
		printf("Not writable\n");
		return ;
	}
 
 
	if(session->is_dir == -1)
	{
		if(session->head_has_send == 0)
		{
			SendHead(session->fd,404,"Not found",GetFileType(".html"),-1);			
			session->head_has_send = 1;
		}
 
		SendFile(session,"404.html");	
		session->writeable = 0;
 
	}
	else if(session->is_dir == 1)
	{
		if(session->head_has_send == 0)
		{
			SendHead(session->fd,200,"OK",GetFileType(".html"),-1);			
			session->head_has_send = 1;
		}
		SendDir(session,session->file_path);			
	}
	else if(session->is_dir == 0)
	{
		if(session->head_has_send == 0)
		{
			SendHead(session->fd,200,"OK",GetFileType(session->file_path),session->file_size);			
			session->head_has_send = 1;
		}
 
		SendFile(session,session->file_path);			
	}
 
	
}
 
 
int Read_cb(void *user_data)
{
	int nread,offset = 0;  
 
	if(user_data == NULL) return -1;
	
	Session *sesion = (Session *)(user_data);
 
	printf("Enter readcb1111 sesion->fd = %d\n",sesion->fd);	
	if(sesion)
	{
		while ((nread = read(sesion->fd, sesion->read_buffer+sesion->read_offset, 1024-1)) > 0) {  
		    sesion->read_offset += nread; 
			
			http_request(sesion);
		}  
 
		printf("nread = %d\n",nread);
		if (nread == -1 && errno != EAGAIN) {  
		    perror("read error");  
		} 
		
		//conn->recv_size = offset;
	}
	
 
	
	return 0;
}
 
int Write_cb(void *user_data)
{
	if(user_data == NULL) return -1;
	
	Session *session = (Session *)(user_data);
 
	http_response(session);
	return 0;
}
 
 
int main(int argc ,char *argv[])
{
	printf("Reactor\n");
 
	signal(SIGPIPE, SIG_IGN);
 
	Reactor reactor;
 
	reactor_init(reactor,Accept_cb,Read_cb,Write_cb);
	reactor_run(reactor);
	reactor_deinit(reactor);
 
	
	return 0;
}