linux 使用epoll实现网络通信

最近在准备一些面试的相关内容， 正好我们的项目中有个关于多个socket处理内容， 虽然项目中我们偷懒直接使用了匿名管道通信解决了这个问题， 当然好像在我们这个项目中好像也没有方法可以使用select， 或者iocp 模型， 毕竟有一部分socket是人家底层封装好了的。

我们想到了，之前在学习linux的过程中还是使用epoll 写过一个玩具程序的， 虽然是一年前写的， 不过好久没碰过linux了， 还是有些生疏的。我们将自己原先写的代码重新读了一遍， 并加上些注释放到这里， 做个备份。

说下使用epoll进行网络通信：

主要使用3个api ， epoll_create 注册epoll， epoll_ctl设置监听fd的监听事件， epoll_wait等待事件到来

epoll 有两种工作模式：

LT ：当epoll_wait检测到描述符事件发生并将此事件通知应用程序，应用程序可以不立即处理该事件。下次调用epoll_wait时，会再次响应应用程序并通知此事件。

ET： 当epoll_wait检测到描述符事件发生并将此事件通知应用程序，应用程序必须立即处理该事件。如果不处理，下次调用epoll_wait时，不会再次响应应用程序并通知此事件。

ET模式在很大程度上减少了epoll事件被重复触发的次数，因此效率要比LT模式高。epoll工作在ET模式的时候，必须使用非阻塞套接口，以避免由于一个文件句柄的阻塞读/阻塞写操作把处理多个文件描述符的任务饿死。

下面是我们1年之前写的代码

关于代码中使用了goto做一个说明， 使用goto主要是为了方便清理资源， 避免每次都要写重复的资源释放代码， 因为当时这个是个C风格的程序， 不像C++可以使用智能指针进行管理。

server.c

/*
* server.c
*
*  Created on: 2015年4月13日
*      Author: ThinkPad User
*      target: 利用epoll实现qq通信
*/

#include "general.h"
#define MAX_EVENTS 10

/*
*   function 设置非阻塞IO
*   input ： fd 表示 输入文件描述符
*/
bool setnonblocking(int fd)
{
int status = fcntl(fd, F_GETFD);
if (-1 == status)
{
MYTRACE_errno;
return false;
}

int res = fcntl(fd, F_SETFD, status | O_NONBLOCK);
if (-1 == res)
{
MYTRACE_errno;
return false;
}

return true;
}

int main(int arg, char * args[])
{
BOOL bRet = FALSE;
int epollfd;
int sockfd;
int sockcli = -1;
do
{
/****************************************************************************/
/*         正常创建socket的流程           socket, bind, listen, setnonblocking  */
/****************************************************************************/
//int port = 8000;
if (arg < 2)
{
MYTRACE("Missing port information\n");
return bRet;
}
int port = atoi(args[1]);

sockfd = socket(AF_INET, SOCK_STREAM, 0);
if (-1 == sockfd)
{
MYTRACE_errno;
break;
}

struct sockaddr_in sock_serv;
sock_serv.sin_family = AF_INET;
sock_serv.sin_port = htons(port);
sock_serv.sin_addr.s_addr = htonl(INADDR_ANY);

int res = bind(sockfd, (struct sockaddr *)&sock_serv, sizeof(struct sockaddr));
if (-1 == res)
{
MYTRACE_errno;
break;
}

res = listen(sockfd, 5);
if (-1 == res)
{
MYTRACE_errno;
break;
}

res = setnonblocking(sockfd);
if (false == res)
{
break;
}

/****************************************************************************/
/*         epoll 处理                                                       */
/****************************************************************************/
epollfd = epoll_create(MAX_EVENTS);
if (-1 == epollfd)
{
MYTRACE_errno;
break;
}

struct epoll_event events[MAX_EVENTS];
struct epoll_event ev;

// 设置监听 内容为数据输入
ev.events = EPOLLIN | EPOLLERR | EPOLLHUP;
ev.data.fd = sockfd;
res = epoll_ctl(epollfd, EPOLL_CTL_ADD, sockfd, &ev);
if (-1 == res)
{
MYTRACE_errno;
break;
}

while(TRUE)
{
// 等待数据到来， res 为有数据的socket的个数
res = epoll_wait(epollfd, events, MAX_EVENTS, -1);
if (-1 == res)
{
MYTRACE_errno;
goto END;
}

int i = 0;
for (; i < res; ++i)
{
// 如果是 sockfd 得到数据， 表明有客户端连接过来了
if (sockfd == events[i].data.fd)
{
struct sockaddr_in addr_cli;
socklen_t len = sizeof(struct sockaddr);
sockcli = accept(sockfd, (struct sockaddr *)&addr_cli, &len);
if (-1 == sockcli)
{
MYTRACE_errno;
goto END;
}

printf("%s connected\n", inet_ntoa(addr_cli.sin_addr));

res = setnonblocking(sockcli);
if (false == res)
{
close(sockcli);
goto END;
}

// 将客户端 的sockcli 加入epoll池中， 监听他的数据收发消息
ev.events = EPOLLIN | EPOLLOUT;
ev.data.fd = sockcli;
res = epoll_ctl(epollfd, EPOLL_CTL_ADD, sockcli, &ev);
if (-1 == res)
{
MYTRACE_errno;
goto END;
}
}

else
{
/*
if (events[i].events & EPOLLHUP)
{
MYTRACE("error");
events[i].data.fd = -1;
}

if (events[i].events & EPOLLERR)
{
MYTRACE("error");
events[i].data.fd = -1;
}
*/

// 数据接收
if (events[i].events & EPOLLIN)
{
char buf[MAX_BYTE] = {0};
res = recv(sockcli, buf, MAX_BYTE, 0);
if (res == -1)
{
MYTRACE_noerrno;
goto END;
}

printf("recv: %s\n", buf);
memset(buf, 0, MAX_BYTE);
}

// 数据发送
if (events[i].events & EPOLLOUT)
{
char buf[MAX_BYTE] = {0};
res = read(STDIN_FILENO, buf, MAX_BYTE);
if (-1 == res)
{
MYTRACE_errno;
goto END;
}

res = send(sockcli, buf, res, 0);
if (res == -1)
{
MYTRACE_noerrno;
goto END;
}

printf("send: %d byte\n", res);
memset(buf, 0, MAX_BYTE);
}
}

}
}

bRet = TRUE;
}while (FALSE);

END:
if (epollfd)
close(epollfd);
if (sockfd)
close(sockfd);
return bRet;
}

client.c

/*
* tcp_client.c
*
*  Created on: 2015年4月8日
*      Author: ThinkPad User
*/

#include "general.h"

pthread_mutex_t g_mutex;

typedef struct _packet
{
int sockfd;
pthread_t * pthr;
}packet;

// 发送数据
void * sock_send(void * args)
{
int sock = *(int *)args;
do
{
char szbuf[MAX_BYTE] = {0};
while(TRUE)
{
int res = read(STDIN_FILENO, szbuf, MAX_BYTE);
if (-1 == res)
{
MYTRACE_errno;
break;
}

res = send(sock, szbuf, strlen(szbuf), 0);
if (-1 == res)
{
MYTRACE_errno;
break;
}

printf("send : %d byte\n", res);
memset(szbuf, 0, MAX_BYTE);
}

}while (FALSE);

return NULL;
}

// 接收数据
void * sock_recv(void * args)
{
packet ps = *(packet *)args;
int sock = ps.sockfd;
pthread_t thr = *ps.pthr;
do
{
char szbuf[MAX_BYTE] = {0};
while(TRUE)
{
int res = recv(sock, szbuf, MAX_BYTE, 0);
if (-1 == res)
{
MYTRACE_noerrno;
break;
}

res = write(STDOUT_FILENO, szbuf, strlen(szbuf));
if (-1 == res)
{
MYTRACE_errno;
break;
}

memset(szbuf, 0, MAX_BYTE);
}

}while (FALSE);

pthread_cancel(thr);

return NULL;
}

int main(int arg, char * args[])
{
BOOL bRet = FALSE;
if (arg < 3)
{
MYTRACE("missing port and IP information\n");
return bRet;
}

do
{
int sock_serv = socket(AF_INET, SOCK_STREAM, 0);
if (-1 == sock_serv)
{
MYTRACE_errno;
break;
}

int port = atoi(args[2]);
char * IP = args[1];

struct sockaddr_in sock_addr;
sock_addr.sin_family = AF_INET;
sock_addr.sin_port = htons(port);
sock_addr.sin_addr.s_addr = inet_addr(IP);

int res = connect(sock_serv, (struct sockaddr *)&sock_addr, sizeof(struct sockaddr));
if (-1 == res)
{
MYTRACE_errno;
break;
}

// 使用 pthread_create 创建子线程
pthread_t thr_send, thr_recv;
int errnum = pthread_create(&thr_send, NULL, sock_send, (void *)&sock_serv);
if (0 != errnum)
{
MYTRACE_noerrno;
break;
}

packet ps;
ps.sockfd = sock_serv;
ps.pthr = &thr_send;
errnum = pthread_create(&thr_recv, NULL, sock_recv, (void *)&ps);
if (0 != errnum)
{
MYTRACE_noerrno;
break;
}

errnum = pthread_join(thr_send, NULL);
if (0 != errnum)
{
MYTRACE_noerrno;
break;
}

errnum = pthread_join(thr_recv, NULL);
if (0 != errnum)
{
MYTRACE_noerrno;
break;
}

bRet = TRUE;
}while (FALSE);

return bRet;
}

general.h

/*
* general.h
*
*  Created on: 2015年4月1日
*      Author: ThinkPad User
*      鉴于每次写linux程序的时候需要调用各种头文件，查找起来非常的不爽，这里将
*      常用的几个头文件罗列在这里,以后直接调用这个头文件就可以了
*/

#ifndef GENERAL_H_
#define GENERAL_H_

#ifdef __cplusplus
extern "C"
{
#endif
// common headers
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/wait.h>
#include <time.h>
#include <signal.h>
#include <sys/klog.h>

// 线程
#include <pthread.h>

// internet
#include <sys/socket.h>
#include <arpa/inet.h>

// epoll
#include <sys/epoll.h>

// support bool operations
#define TRUE 1
#define FALSE 0
typedef int BOOL;

#define true 1
#define false 0
typedef int bool;

// define buffer size
#define MAX_BYTE 0xFF
#define MAX_BUFFER 64*1024

// trace for debug
#define MYTRACE(condition)  printf("%s %s: [%s](%d)<%s> %s error\n", __DATE__, __TIME__, __FILE__, __LINE__, __FUNCTION__, condition)
#define MYTRACE_errno   printf("%s %s: [%s](%d)<%s> error: %s\n", __DATE__, __TIME__, __FILE__, __LINE__, __FUNCTION__, strerror(errno))
#define MYTRACE_noerrno     printf("%s %s: [%s](%d)<%s> error\n", __DATE__, __TIME__, __FILE__, __LINE__, __FUNCTION__)

// create my signal1 in replace of signal
int signal1(int signum, void(*func)(int))
{
struct sigaction act, oact;
act.sa_handler = func;
sigemptyset(&act.sa_mask);
act.sa_flags = 0;
return sigaction(signum, &act, &oact);
}

#ifdef __cplusplus
}
#endif
#endif /* GENERAL_H_ */

makefile

.SUFFIXES:.c .o

CC = gcc
SRC = client.c
OBJ = $(SRC:.c=.o)
EXEC = client

SRC1 = server.c
OBJ1 = $(SRC1:.c=.o)
EXEC1 = server

start:$(OBJ) $(OBJ1)
$(CC) -o $(EXEC) $(OBJ) -pthread
$(CC) -o $(EXEC1) $(OBJ1)
@echo '--------------------done-----------------------'

.c.o:
$(CC) -Wall -g -o $@ -c $<

clean:
rm -f $(OBJ)
rm -f $(OBJ1)
rm -f core*

参考资源：

IO多路复用之epoll总结