void add_client_socket(int server_socket, int epoll) {
    struct sockaddr_in cliaddr;
    socklen_t addr_size = sizeof cliaddr;

    int client_socket = accept(server_socket, (struct sockaddr*) &cliaddr, &addr_size);
    if (client_socket < 0) {
        perror("accept");
        return;
    }

    setnonblocking(client_socket);
    struct epoll_event connev;
    connev.data.fd = client_socket;
    connev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLHUP;
    if (!epoll_ctl(epoll, EPOLL_CTL_ADD, client_socket, &connev) < 0) {
        perror("Epoll fd add");
        close(client_socket);
        return;
    }
}

void accept_cb(EV_P_ ev_io *w, int revents)
{
    struct listen_ctx *listener = (struct listen_ctx *)w;
    int serverfd = accept(listener->fd, NULL, NULL);
    if (serverfd == -1) {
        ERROR("accept");
        return;
    }
    setnonblocking(serverfd);
    int opt = 1;
    setsockopt(serverfd, SOL_TCP, TCP_NODELAY, &opt, sizeof(opt));
#ifdef SO_NOSIGPIPE
    setsockopt(serverfd, SOL_SOCKET, SO_NOSIGPIPE, &opt, sizeof(opt));
#endif

    struct server *server = new_server(serverfd, listener->method);
    server->listener = listener;

    ev_io_start(EV_A_ & server->recv_ctx->io);
}

int 
pxy_start_listen()
{
	struct sockaddr_in addr1;

	master->listen_fd = socket(AF_INET,SOCK_STREAM,0);
	if (master->listen_fd < 0) {
		D("create listen fd error");
		return -1;
	}

	int reuse = 1;
	setsockopt(master->listen_fd,
		   SOL_SOCKET,
		   SO_REUSEADDR,
		   &reuse,
		   sizeof(int));
	
	if (setnonblocking(master->listen_fd) < 0) {
		D("set nonblocling error");
		return -1;
	}

	addr1.sin_family = AF_INET;
	addr1.sin_port = htons(setting.client_port);
	inet_aton(setting.ip, &addr1.sin_addr);
	
	if (bind(master->listen_fd, (struct sockaddr*)&addr1,
		sizeof(addr1)) < 0) {
		E("bind address %s:%d error", setting.ip,
		  setting.client_port);
		return -1;
	}

	if (listen(master->listen_fd,1000) < 0) {
		E("listen error");
		return -1;
	}
    
	return 0;
}

int accept_agent_side( agent_t *agent, event_info_t *event_info) 
{
   socklen_t sin_size; 
   struct sockaddr_storage their_addr; 
   sin_size = sizeof(their_addr); 
   int index = find_empty_agent_sock(agent); 
	struct epoll_event event; 

   if(index < 0) 
   { 
      printf("agent sock table full!!!\n"); 
      exit(1); 
   } 

   if(( agent->agent_fd_pool[index] = 
		accept(agent->listen_fds.agent_listen_sock[event_info->fd], 
   	(struct sockaddr *) &their_addr, &sin_size)) == -1)
	{
		perror(""); 
		printf("%s %d\n", __FILE__, __LINE__); 
		exit(1); 	
   }

   setnonblocking(agent->agent_fd_pool[index]); 

	event.events = EPOLLOUT; 
	event.data.ptr =  &agent->agent_fd_pool_event[index]; 
	agent->agent_fd_pool_event[index].type =	AGENT_CONNECTED_UUID;  
	agent->agent_fd_pool_event[index].fd = index; 	

	if( epoll_ctl(agent->event_pool, EPOLL_CTL_ADD, 
	   agent->agent_fd_pool[index], &event)) 
	{ 
	   perror("");
	   printf("%s %d\n", __FILE__, __LINE__); 
	   exit(1); 
	}


   return EXIT_SUCCESS; 
}

int build_server() {
    struct addrinfo hints, *res;
    int s;

    memset(&hints, 0, sizeof hints);
    hints.ai_family = AF_UNSPEC;
    hints.ai_socktype = SOCK_STREAM;

    s = getaddrinfo("0.0.0.0", setting.faddr_port, &hints, &res);
    if (s != 0) {
        perror("getaddrinfo error");
        return -1;
    }

    serv_sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);

    if(serv_sock < 0) {
        perror("socket error");
        exit(1);
    }

    int opt = 1;
    setsockopt(serv_sock, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
    setsockopt(serv_sock, IPPROTO_TCP, TCP_NODELAY, &opt, sizeof(opt));
#ifdef SO_NOSIGPIPE
    setsockopt(serv_sock, SOL_SOCKET, SO_NOSIGPIPE, &opt, sizeof(opt));
#endif


    if ( bind(serv_sock, res->ai_addr, res->ai_addrlen) == -1) {
        perror("bind error");
        exit(1);
    }

    if( listen(serv_sock, SOMAXCONN) == -1 ) {
        perror("listen error");
        exit(1);
    }

    setnonblocking(serv_sock);
}

/* Creates a new unix domain socket. */
static int
create_socket(char *name)
{
	int s;
	struct sockaddr_un sockun;

	if (strlen(name) > sizeof(sockun.sun_path) - 1)
	{
		errno = ENAMETOOLONG;
		return -1;
	}

	s = socket(PF_UNIX, SOCK_STREAM, 0);
	if (s < 0)
		return -1;
	sockun.sun_family = AF_UNIX;
	strcpy(sockun.sun_path, name);
	if (bind(s, (struct sockaddr*)&sockun, sizeof(sockun)) < 0)
	{
		close(s);
		return -1;
	}
	if (listen(s, 128) < 0)
	{
		close(s);
		return -1;
	}
	if (setnonblocking(s) < 0)
	{
		close(s);
		return -1;
	}
	/* chmod it to prevent any suprises */
	if (chmod(name, 0600) < 0)
	{
		close(s);
		return -1;
	}
	return s;
}

static remote_t *create_remote(listen_ctx_t *listener,
                               struct sockaddr *addr)
{
    struct sockaddr *remote_addr;

    int index = rand() % listener->remote_num;
    if (addr == NULL) {
        remote_addr = listener->remote_addr[index];
    } else {
        remote_addr = addr;
    }

    int remotefd = socket(remote_addr->sa_family, SOCK_STREAM, IPPROTO_TCP);

    if (remotefd < 0) {
        ERROR("socket");
        return NULL;
    }

    int opt = 1;
    setsockopt(remotefd, SOL_TCP, TCP_NODELAY, &opt, sizeof(opt));
#ifdef SO_NOSIGPIPE
    setsockopt(remotefd, SOL_SOCKET, SO_NOSIGPIPE, &opt, sizeof(opt));
#endif

    // Setup
    setnonblocking(remotefd);
#ifdef SET_INTERFACE
    if (listener->iface) {
        setinterface(remotefd, listener->iface);
    }
#endif

    remote_t *remote = new_remote(remotefd, listener->timeout);
    remote->addr_len = get_sockaddr_len(remote_addr);
    memcpy(&(remote->addr), remote_addr, remote->addr_len);
    remote->remote_index = index;

    return remote;
}

void setUpServer(int argc, char *argv[]) //Sets up the server
{
    timeout.tv_sec  = TIMEOUT_SEC;
    timeout.tv_usec = TIMEOUT_MS;
    
    
    if (argc < MIN_ARGS) 
    {
        fprintf(stderr,"ERROR, no port provided\n");
        exit(1);
    }
    //set up the socket. we are using TCP
    sockfd = socket(AF_INET, SOCK_STREAM, 0);
    
    if (sockfd < 0)  //make sure we can open the socket
    {
        error("ERROR opening socket");
    }
    
    setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &reuse_addr, sizeof(reuse_addr)); //allow reuse of address
    
    setnonblocking(sockfd); //nonblocking
    
    //port stuff
    bzero((char *) &serv_addr, sizeof(serv_addr));
    
    portno = atoi(argv[PORT_ARG_SERVER]);
    serv_addr.sin_family = AF_INET;
    serv_addr.sin_addr.s_addr = INADDR_ANY;
    serv_addr.sin_port = htons(portno);
    
    if (bind(sockfd, (struct sockaddr *) &serv_addr, sizeof(serv_addr)) < 0) //check it we can bind it
    {
        error("ERROR on binding");
    }
    
    listen(sockfd, LISTENING_QUEUE); //set up a listening queue
    
    clilen = sizeof(cli_addr); //set the length of the address
}

int accept_host_side(agent_t *agent, client_t *new_client) 
{
	socklen_t sin_size; 
	struct sockaddr_in their_addr; 
	struct epoll_event event; 

	sin_size = sizeof(their_addr); 


	if(( new_client->host_sock = accept(agent->listen_fds.host_listen_sock, 
		(struct sockaddr *) &their_addr, &sin_size)) == -1)
	{
      perror(""); 
	   printf("%s %d\n", __FILE__, __LINE__); 
      exit(1); 
	}
   
	
	strcpy(new_client->source_ip, inet_ntoa(their_addr.sin_addr)); 
	new_client->source_port = htons(their_addr.sin_port); 
	

   new_client->host_fd_poll = OFF; 
	setnonblocking(new_client->host_sock); 
	event.events = EPOLLOUT; 
   event.data.ptr =  &new_client->host_side_event_info; 
   new_client->host_side_event_info.type = HOST_CONNECTED;  
	new_client->host_side_event_info.client = new_client; 

	if( epoll_ctl(agent->event_pool, EPOLL_CTL_ADD, 
	   new_client->host_sock, &event)) 
	{ 
	   perror("");
	   printf("%s %d\n", __FILE__, __LINE__); 
	   exit(1); 
	}


	return EXIT_SUCCESS; 
}

CAcceptor::CAcceptor(CInetAddr& sockAddr, 
	int backlog, int timeout) : CSock(AF_INET, SOCK_STREAM)
{
	int res;
	if (setsockopt(m_fd, SOL_SOCKET, SO_REUSEADDR, &res, sizeof(int)) < 0)
	{
		KL_SYS_ERRORLOG("file: "__FILE__", line: %d, " \
			"set SO_REUSEADDR failed, err: %s", \
			__LINE__, strerror(errno));
		throw errno;
	}

	if(bind(m_fd, sockAddr.getsockaddr(), sizeof(struct sockaddr)) == -1)
	{
			KL_SYS_ERRORLOG("file: "__FILE__", line: %d, " \
				"call bind failed, err: %s", \
				__LINE__, strerror(errno));
			throw errno;
	}
	listen(m_fd, backlog);

	if(timeout > 0) //server option
	{
		if((res = setserveropt(timeout)) != 0)
		{
			KL_SYS_ERRORLOG("file: "__FILE__", line: %d, " \
				"acceptor set server option failed, err: %s", \
				__LINE__, strerror(res));
			throw res;
		}
	}

	if((res = setnonblocking()) != 0)
	{
		KL_SYS_ERRORLOG("file: "__FILE__", line: %d, " \
			"acceptor set nonblocking failed, err: %s", \
			__LINE__, strerror(res));
		throw res;
	}
}

int
init_socket(int efd, uint16_t sock_type, struct sockaddr *srv, uint16_t srvlen, size_t simultaneous_connection)
{
  struct epoll_event ev;
  static int optval = 1;
  int fd;

  if ((fd = socket(sock_type, SOCK_STREAM,  0)) == -1)
    return -1;

  if (!setnonblocking(fd))
    goto error;

  memset(&ev, 0, sizeof(struct epoll_event));

  ev.data.fd = fd;
  ev.events  = EPOLLIN | EPOLLET;

  if (epoll_ctl(efd, EPOLL_CTL_ADD, fd, &ev) == -1)
    goto error;

  setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval));

  int on = 1;
  if (srv->sa_family == AF_INET6)
    setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &on, sizeof(on));

  if (bind(fd, srv, srvlen) == -1)
    goto error;

  if (listen(fd, simultaneous_connection) == -1)
    goto error;

  return fd;
error:
  close(fd);
  return -1;
}

/* Create the bound socket in the parent process */
static int create_main_socket() {
    struct addrinfo *ai, hints;
    memset(&hints, 0, sizeof hints);
    hints.ai_family = AF_UNSPEC;
    hints.ai_socktype = SOCK_STREAM;
    hints.ai_flags = AI_PASSIVE | AI_ADDRCONFIG;
    const int gai_err = getaddrinfo(OPTIONS.FRONT_IP, OPTIONS.FRONT_PORT,
                                    &hints, &ai);
    if (gai_err != 0) {
        ERR("{getaddrinfo}: [%s]\n", gai_strerror(gai_err));
        exit(1);
    }

    int s = socket(ai->ai_family, SOCK_STREAM, IPPROTO_TCP);
    int t = 1;
    setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &t, sizeof(int));
#ifdef SO_REUSEPORT
    setsockopt(s, SOL_SOCKET, SO_REUSEPORT, &t, sizeof(int));
#endif
    setnonblocking(s);

    if (bind(s, ai->ai_addr, ai->ai_addrlen)) {
        fail("{bind-socket}");
    }

#if TCP_DEFER_ACCEPT
    int timeout = 1; 
    setsockopt(s, IPPROTO_TCP, TCP_DEFER_ACCEPT, &timeout, sizeof(int) );
#endif /* TCP_DEFER_ACCEPT */

    prepare_proxy_line(ai->ai_addr);

    freeaddrinfo(ai);
    listen(s, OPTIONS.BACKLOG);

    return s;
}

/*
*	初始化网络
*/
int inet_init()
{
	socket_t s;
	int buf_size = 8*1024;
	int reuse = 1;
	
	if ((listening.s = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
		log_error(errno, "socket()");
		exit(1);
	}
	
	if (setsockopt(listening.s, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)) == -1 || 
		setsockopt(listening.s, SOL_SOCKET, SO_REUSEADDR, &buf_size, sizeof(buf_size)) == -1) {
		log_error(errno, "setsockopt()");
		exit(1);
	}
	
	bzero(&listening.ls_addr, sizeof(listening.ls_addr));
	listening.ls_addr.sin_family = AF_INET;
	listening.ls_addr.sin_port = htons(LISTEN_PORT);
	listening.ls_addr.sin_addr.s_addr = htonl(INADDR_ANY);
	
	if (bind(listening.s, (struct sockaddr *)&listening.ls_addr, sizeof(listening.ls_addr)) == -1) {
		log_error(errno, "bind()");
		exit(1);
	}
	
	if (listen(listening.s, LISTEN_BACKLOG) == -1) {
		log_error(errno, "listen()");
		exit(1);
	}
	
	setnonblocking(listening.s);
	
	//注册一个accept()后的处理钩子
	//listening.handler = http_connection_init;
}

bool CLoveTcp::CreateSocket()
{
	if((m_nSocket = socket(AF_INET,SOCK_STREAM,0))==-1)  
	{
		LogErrorMsg("create socket error");
		return false;
	}
	if (setnonblocking(m_nSocket) == -1)
	{
		LogErrorMsg("setnonblocking error");
	}
	int nSock = 0;
	if (setsockopt(m_nSocket,SOL_SOCKET,SO_REUSEADDR,&nSock,sizeof(nSock)) < 0)
	{
		/* code */
		LogErrorMsg("setsockopt failed");
	}
	m_nEpoll = epoll_create1(EPOLL_CLOEXEC);
	if (EpollAdd(m_nSocket) == false)
	{
		return false;
	};
	return true;
}

int handle_accept(int fd)
{
	struct epoll_event ev;
	struct echo_data* pecho_data;

	int newsock = 0;
	do{//edge-trigger
		newsock = accept(fd,NULL,NULL);
		//if( newsock <=0 ){
		if( newsock == -1 ){ //error occur
			if( (errno == EAGAIN) || (errno == EWOULDBLOCK) ) break;
			else{
				perror("accept error");
				return -1;
			}
		} else {
			printf("Client accept with descriptor #%d\n" ,newsock);
			pecho_data = alloc_echo_data(newsock, MAX_ECHO_BUF);
			if( NULL == pecho_data ){
				perror("malloc error");
				return -1;
			}

			setnonblocking(newsock); //set nonblocking 
			ev.data.ptr = pecho_data;

			ev.events = EPOLLIN|EPOLLPRI|EPOLLET; //edge-trigger
			if( -1 == epoll_ctl(epfd, EPOLL_CTL_ADD, newsock, &ev) ){
				perror("epoll_ctl error!");
				return -1;
			}
		}
	}while(1);

	return 0;
}

void handle_new_connection() 
{
	int listnum;	    //Current item in fd
	int connection;     // Socket file descriptor for incoming connections 
    
	connection = accept(sockfd, (struct sockaddr *) &cli_addr, &clilen);
    
	if (connection < 0) 
    {
		perror("accept");
		exit(EXIT_FAILURE);
	}
    
	setnonblocking(connection);
    
	for (listnum = 0; (listnum < MAX_USERS) && (connection != -1); listnum ++)
		if (fd[listnum] == 0) 
        {
			printf("\nConnection accepted:   FD=%d; Slot=%d\n", connection, listnum);
			fd[listnum] = connection;
			connection = -1;
		}
    
	if (connection != -1) 
    {
		// No room left in the queue!
		printf("\nNo room left for new client.\n");
    }
    else
    {
        struct user newUser;
        newUser.address.sin_addr.s_addr = cli_addr.sin_addr.s_addr; //store the ip
        printf("USER IP: %u at %d\n", newUser.address.sin_addr.s_addr, listnum - 1);
        users[fd[listnum - 1]] = newUser;
    }
}

/**
 * HTTTP daemon listener entry point
 *
 * @param	listener	The Listener DCB
 * @param	config		Configuration (ip:port)
 */
static int
httpd_listen(DCB *listener, char *config)
{
struct sockaddr_in	addr;
char			*port;
int			one = 1;
short			pnum;
int                     rc;

	memcpy(&listener->func, &MyObject, sizeof(GWPROTOCOL));

	port = strrchr(config, ':');
	if (port)
		port++;
	else
		port = "6442";

	addr.sin_family = AF_INET;
	addr.sin_addr.s_addr = htonl(INADDR_ANY);
	pnum = atoi(port);
	addr.sin_port = htons(pnum);

	if ((listener->fd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
	{
		return 0;
	}

        /* socket options */
	setsockopt(listener->fd,
                   SOL_SOCKET,
                   SO_REUSEADDR,
                   (char *)&one,
                   sizeof(one));

        /* set NONBLOCKING mode */
        setnonblocking(listener->fd);

        /* bind address and port */
        if (bind(listener->fd, (struct sockaddr *)&addr, sizeof(addr)) < 0)
	{
        	return 0;
	}

        rc = listen(listener->fd, SOMAXCONN);
        
        if (rc == 0) {
            fprintf(stderr,
                    "Listening http connections at %s\n",
                    config);
        } else {
            int eno = errno;
            errno = 0;
            fprintf(stderr,
                    "\n* Failed to start listening http due error %d, %s\n\n",
                    eno,
                    strerror(eno));
            return 0;
        }

        
        if (poll_add_dcb(listener) == -1)
	{
		return 0;
	}
	return 1;
}

//.........这里部分代码省略.........
    if (nofile > 1024) {
        if (verbose) {
            LOGI("setting NOFILE to %d", nofile);
        }
        set_nofile(nofile);
    }
#endif

    if (local_addr == NULL) {
        local_addr = "127.0.0.1";
    }

    if (pid_flags) {
        USE_SYSLOG(argv[0]);
        daemonize(pid_path);
    }

    if (ipv6first) {
        LOGI("resolving hostname to IPv6 address first");
    }

    if (auth) {
        LOGI("onetime authentication enabled");
    }

    // ignore SIGPIPE
    signal(SIGPIPE, SIG_IGN);
    signal(SIGABRT, SIG_IGN);
    signal(SIGINT, signal_cb);
    signal(SIGTERM, signal_cb);

    // Setup keys
    LOGI("initializing ciphers... %s", method);
    int m = enc_init(password, method);

    // Setup proxy context
    listen_ctx_t listen_ctx;
    listen_ctx.remote_num  = remote_num;
    listen_ctx.remote_addr = ss_malloc(sizeof(struct sockaddr *) * remote_num);
    for (int i = 0; i < remote_num; i++) {
        char *host = remote_addr[i].host;
        char *port = remote_addr[i].port == NULL ? remote_port :
                     remote_addr[i].port;
        struct sockaddr_storage *storage = ss_malloc(sizeof(struct sockaddr_storage));
        memset(storage, 0, sizeof(struct sockaddr_storage));
        if (get_sockaddr(host, port, storage, 1, ipv6first) == -1) {
            FATAL("failed to resolve the provided hostname");
        }
        listen_ctx.remote_addr[i] = (struct sockaddr *)storage;
    }
    listen_ctx.timeout = atoi(timeout);
    listen_ctx.method  = m;
    listen_ctx.mptcp   = mptcp;

    struct ev_loop *loop = EV_DEFAULT;

    if (mode != UDP_ONLY) {
        // Setup socket
        int listenfd;
        listenfd = create_and_bind(local_addr, local_port);
        if (listenfd == -1) {
            FATAL("bind() error");
        }
        if (listen(listenfd, SOMAXCONN) == -1) {
            FATAL("listen() error");
        }
        setnonblocking(listenfd);

        listen_ctx.fd = listenfd;

        ev_io_init(&listen_ctx.io, accept_cb, listenfd, EV_READ);
        ev_io_start(loop, &listen_ctx.io);
    }

    // Setup UDP
    if (mode != TCP_ONLY) {
        LOGI("UDP relay enabled");
        init_udprelay(local_addr, local_port, listen_ctx.remote_addr[0],
                      get_sockaddr_len(listen_ctx.remote_addr[0]), mtu, m, auth, listen_ctx.timeout, NULL);
    }

    if (mode == UDP_ONLY) {
        LOGI("TCP relay disabled");
    }

    LOGI("listening at %s:%s", local_addr, local_port);

    // setuid
    if (user != NULL && ! run_as(user)) {
        FATAL("failed to switch user");
    }

    if (geteuid() == 0){
        LOGI("running from root user");
    }

    ev_run(loop, 0);

    return 0;
}

int unblock_connect( const char* ip, int port, int time )
{
    int ret = 0;
    struct sockaddr_in address;
    bzero( &address, sizeof( address ) );
    address.sin_family = AF_INET;
    inet_pton( AF_INET, ip, &address.sin_addr );
    address.sin_port = htons( port );

    int sockfd = socket( PF_INET, SOCK_STREAM, 0 );
    int fdopt = setnonblocking( sockfd );
    ret = connect( sockfd, ( struct sockaddr* )&address, sizeof( address ) );
    if ( ret == 0 )
    {
        printf( "connect with server immediately\n" );
        fcntl( sockfd, F_SETFL, fdopt );
        return sockfd;
    }
    else if ( errno != EINPROGRESS )
    {
        printf( "unblock connect not support\n" );
        return -1;
    }

    fd_set readfds;
    fd_set writefds;
    struct timeval timeout;

    FD_ZERO( &readfds );
    FD_SET( sockfd, &writefds );

    timeout.tv_sec = time;
    timeout.tv_usec = 0;

    ret = select( sockfd + 1, NULL, &writefds, NULL, &timeout );
    if ( ret <= 0 )
    {
        printf( "connection time out\n" );
        close( sockfd );
        return -1;
    }

    if ( ! FD_ISSET( sockfd, &writefds  ) )
    {
        printf( "no events on sockfd found\n" );
        close( sockfd );
        return -1;
    }

    int error = 0;
    socklen_t length = sizeof( error );
    if( getsockopt( sockfd, SOL_SOCKET, SO_ERROR, &error, &length ) < 0 )
    {
        printf( "get socket option failed\n" );
        close( sockfd );
        return -1;
    }

    if( error != 0 )
    {
        printf( "connection failed after select with the error: %d \n", error );
        close( sockfd );
        return -1;
    }
    
    printf( "connection ready after select with the socket: %d \n", sockfd );
    fcntl( sockfd, F_SETFL, fdopt );
    return sockfd;
}

static void
accept_cb(EV_P_ ev_io *w, int revents)
{
    listen_ctx_t *listener = (listen_ctx_t *)w;
    struct sockaddr_storage destaddr;
    int err;

    int serverfd = accept(listener->fd, NULL, NULL);
    if (serverfd == -1) {
        ERROR("accept");
        return;
    }

    err = getdestaddr(serverfd, &destaddr);
    if (err) {
        ERROR("getdestaddr");
        return;
    }

    setnonblocking(serverfd);
    int opt = 1;
    setsockopt(serverfd, SOL_TCP, TCP_NODELAY, &opt, sizeof(opt));
#ifdef SO_NOSIGPIPE
    setsockopt(serverfd, SOL_SOCKET, SO_NOSIGPIPE, &opt, sizeof(opt));
#endif

    int index                    = rand() % listener->remote_num;
    struct sockaddr *remote_addr = listener->remote_addr[index];

    int remotefd = socket(remote_addr->sa_family, SOCK_STREAM, IPPROTO_TCP);
    if (remotefd == -1) {
        ERROR("socket");
        return;
    }

    // Set flags
    setsockopt(remotefd, SOL_TCP, TCP_NODELAY, &opt, sizeof(opt));
#ifdef SO_NOSIGPIPE
    setsockopt(remotefd, SOL_SOCKET, SO_NOSIGPIPE, &opt, sizeof(opt));
#endif

    // Enable TCP keepalive feature
    int keepAlive    = 1;
    int keepIdle     = 40;
    int keepInterval = 20;
    int keepCount    = 5;
    setsockopt(remotefd, SOL_SOCKET, SO_KEEPALIVE, (void *)&keepAlive, sizeof(keepAlive));
    setsockopt(remotefd, SOL_TCP, TCP_KEEPIDLE, (void *)&keepIdle, sizeof(keepIdle));
    setsockopt(remotefd, SOL_TCP, TCP_KEEPINTVL, (void *)&keepInterval, sizeof(keepInterval));
    setsockopt(remotefd, SOL_TCP, TCP_KEEPCNT, (void *)&keepCount, sizeof(keepCount));

    // Set non blocking
    setnonblocking(remotefd);

    // Enable MPTCP
    if (listener->mptcp == 1) {
        int err = setsockopt(remotefd, SOL_TCP, MPTCP_ENABLED, &opt, sizeof(opt));
        if (err == -1) {
            ERROR("failed to enable multipath TCP");
        }
    }

    server_t *server = new_server(serverfd, listener->method);
    remote_t *remote = new_remote(remotefd, listener->timeout);
    server->remote   = remote;
    remote->server   = server;
    server->destaddr = destaddr;

    int r = connect(remotefd, remote_addr, get_sockaddr_len(remote_addr));

    if (r == -1 && errno != CONNECT_IN_PROGRESS) {
        ERROR("connect");
        close_and_free_remote(EV_A_ remote);
        close_and_free_server(EV_A_ server);
        return;
    }

    // listen to remote connected event
    ev_io_start(EV_A_ & remote->send_ctx->io);
    ev_timer_start(EV_A_ & remote->send_ctx->watcher);
    ev_io_start(EV_A_ & server->recv_ctx->io);
}