使用epoll来进行异步的 TCP 连接

长话短说了 。。TCP 连接成功后， 连接的 socket 变成可写状态，那么我们就监听  EPOLLOUT 事件，来判断是不是连接成功了。但是我们需要先调用 connect 但是调用 connect 的时候会阻塞，好吧 还需要先把 socket 设置成非阻塞，这样调用完 connect 里面在线程中进入对 epoll 事件的查询。对于 connect 失败的情况，返回的事件是 EPOLLERR .简单的测试代码：

#include <iostream>
#include <sys/types.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/stat.h>
#include <sys/fcntl.h>
#include <unistd.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <pthread.h>
#include <sys/ioctl.h>
#include <errno.h>
#include <string.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

using namespace std;

typedef void (* PF_CALLBACK)(int fd);

class CEpoll
{
public:
   CEpoll(PF_CALLBACK pfOnRead,PF_CALLBACK pfOnWrite,PF_CALLBACK pfOnError)
   {
       m_epoll = epoll_create(100);//
       if(m_epoll == -1)
       {
           throw "error create epoll ~";
       }

       m_OnRead = pfOnRead;
       m_OnWrite = pfOnWrite;
	   m_OnError = pfOnError;
   }

   ~CEpoll()
   {
       if(m_epoll >= 0)
       {
           close(m_epoll);
       }
   }

public:

	bool AddFdOnWrite(int fd)
	{
	   struct epoll_event ev;
       ev.data.fd = fd;
       //ev.data.ptr = "i am sincoder";//
       ev.events = EPOLLOUT|EPOLLET;

       int ret = epoll_ctl(m_epoll,EPOLL_CTL_ADD,fd,&ev);
       if(ret == -1)
       {
           return false;
       }
       return true;
	}

   bool AddFdOnRead(int fd)
   {

       struct epoll_event ev;
       ev.data.fd = fd;
       //ev.data.ptr = "i am sincoder";//
       ev.events = EPOLLIN|EPOLLET;

       int ret =epoll_ctl(m_epoll,EPOLL_CTL_ADD,fd,&ev);
       if(ret == -1)
       {
           return false;
       }
       return true;
   }

   bool ProcessMsg()
   {
       int i;
       int nr_events = 0;
       int ret;
       char buff[1024];
       struct epoll_event events[64];

           printf("enter wait !\n");
           nr_events = epoll_wait (m_epoll,events,64,-1);
           printf("out wait%d\n",nr_events);

           if (nr_events < 0)
           {
               perror("epoll_wait");
               return false;
           }

           for (i = 0; i < nr_events; i++)
           {
               printf("event=%ld on fd=%d \n",events[i].events, events[i].data.fd);
			   if(events[i].events & EPOLLIN )
			   {
				   m_OnRead(events[i].data.fd);
			   }
			   if(events[i].events & EPOLLOUT )
			   {
				   m_OnWrite(events[i].data.fd);
			   }
			   if(events[i].events & EPOLLERR )
			   {
				   m_OnError(events[i].data.fd);
			   }
           }
       return true;
   }

private:
   int m_epoll;
   PF_CALLBACK m_OnRead;
   PF_CALLBACK m_OnWrite;
   PF_CALLBACK m_OnError;
};

void OnRead(int fd);
int client_socket;
CEpoll *g_epoll = NULL;

void OnRead(int fd)
{
   printf("enter call back ~\n");
   if(fd == client_socket){

   }else{
       char buff[1024];
       int ret = recv(fd,buff,sizeof(buff),0);
       if(ret <= 0)
       {
           printf("close fd %d \n",fd);
           close(fd);
       }
       else{
           printf("recv %d bytes from fd %d \n",ret,fd);
       }
   }
}

void OnWrite(int fd)
{
	printf("enter OnWrite \n");
	//看看 socket 是不是链接成功了 
	int result;
	socklen_t result_len = sizeof(result);
	if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &result, &result_len) < 0) {
		// error, fail somehow, close socket
		close(fd);
		return ;
	}

	if (result != 0) {
    // connection failed; error code is in ‘result‘

		printf("connect failed ~!! \n");
		close(fd);
		return ;
	}

	printf("connect OK ~!! \n");

	//将 fd 设置成非阻塞 并重新加入到 epoll  中
}

void OnError(int fd)
{
	if(fd == client_socket)
	{
		//connect error ~!
		printf("connect Error ~! \n");
		close(fd);
		return;
	}
}

int main(int argc,char **argv) {
   if(argc < 2 )
   {
       printf("usage : <ip> <port> \n");
       return 0;
   }

   struct sockaddr_in serv_addr;
   client_socket = socket(AF_INET,SOCK_STREAM,IPPROTO_TCP);
   if(client_socket == -1)
   {
       perror("socket():");
       return -1;
   }

   try {
       g_epoll = new CEpoll(OnRead,OnWrite ,OnError);
   }catch(...)
   {
       close(client_socket);
       return -1;
   }

   printf("get socket fd %d", client_socket);
   bzero((char *) &serv_addr, sizeof(serv_addr));
   serv_addr.sin_family = AF_INET;
   serv_addr.sin_addr.s_addr = inet_addr(argv[1]) ;
   serv_addr.sin_port = htons(atoi(argv[2]));

   //设置 connect 的 socket 为非阻塞

   long on = 1L;
   if (ioctl(client_socket, (int)FIONBIO, (char *)&on))
   {
       printf("ioctl FIONBIO call failed\n");
   }
   
   g_epoll->AddFdOnWrite(client_socket);

   if (connect(client_socket, (struct sockaddr *) &serv_addr, sizeof(serv_addr)) < 0) {
	   if(errno != EINPROGRESS)
	   {
			printf("%s", "ERROR connecting");
	   }
       //close(client_socket);
       //throw "can not connect to remote";
   }

   while(g_epoll->ProcessMsg());
   return 0;
}