int
FileRepConnServer_ReceiveMessageType(
									 FileRepConsumerProcIndex_e *fileRepMessageType)
{

	char		messageType;

	int			status = STATUS_OK;

	messageType = pq_getbyte();


	switch (messageType)
	{

		case '1':
			*fileRepMessageType = FileRepMessageTypeXLog;
			break;

		case '2':
			*fileRepMessageType = FileRepMessageTypeAO01;
			break;

		case '3':
			*fileRepMessageType = FileRepMessageTypeWriter;
			break;

		case 'S':
			*fileRepMessageType = FileRepMessageTypeShutdown;
			break;

		case EOF:
			ereport(WARNING,
					(errcode_for_socket_access(),
					 errmsg("receive EOF on connection: %m")));
			status = STATUS_ERROR;
			break;
		case 'X':
			/* Close Message(sent by PQfinish()) */

			/*
			 * Client closed connection.  Client does not wait for response.
			 */
			ereport(WARNING,
					(errcode_for_socket_access(),
					 errmsg("receive close on connection: %m")));
			status = STATUS_ERROR;
			break;
		default:
			ereport(WARNING,
					(errcode_for_socket_access(),
					 errmsg("receive unexpected message type on connection: %m")));
			status = STATUS_ERROR;
			break;
	}

	return status;
}

int
FileRepConnServer_CreateConnection()
{
	int			status = STATUS_OK;

	port = (Port *) calloc(1, sizeof(Port));
	if (port == NULL)
	{
		ereport(ERROR,
				(errcode(ERRCODE_OUT_OF_MEMORY),
				 (errmsg("not enough memory to create connection"))));
		return status;
	}

	status = StreamConnection(listenSocket[0], port);
	if (status != STATUS_OK)
	{
		ereport(WARNING,
				(errcode_for_socket_access(),
				 errmsg("could not accept connection: %m"),
				 FileRep_errcontext()));

		if (port->sock >= 0)
		{
			StreamClose(port->sock);
		}
		ConnFree();
	}
	else
	{
		/*
		 * MPP-14225: On NIC failure, filerep receiver process's recv() system
		 * call will take hours to timeout, depending on the TCP timeout. Add
		 * SO_RCVTIMEO timeout to filerep receiver process's socket to avoid
		 * this.
		 */
		struct timeval tv;

		tv.tv_sec = file_rep_socket_timeout;
		tv.tv_usec = 0;			/* Not initializing this can cause strange
								 * errors */

		if (setsockopt(port->sock, SOL_SOCKET, SO_RCVTIMEO, (char *) &tv, sizeof(struct timeval)) == -1)
			ereport(WARNING,
					(errcode_for_socket_access(),
					 errmsg("could not set receive timeout on socket")));

		/* set TCP keep-alive parameters for FileRep connection */
		(void) pq_setkeepalivesidle(gp_filerep_tcp_keepalives_idle, port);
		(void) pq_setkeepalivesinterval(gp_filerep_tcp_keepalives_interval, port);
		(void) pq_setkeepalivescount(gp_filerep_tcp_keepalives_count, port);

		MyProcPort = port;
	}

	return status;
}

/*
 * Wait until we can read WAL stream, or timeout.
 *
 * Returns true if data has become available for reading, false if timed out
 * or interrupted by signal.
 *
 * This is based on pqSocketCheck.
 */
static bool
libpq_select(int timeout_ms)
{
	int			ret;

	Assert(streamConn != NULL);
	if (PQsocket(streamConn) < 0)
		ereport(ERROR,
				(errcode_for_socket_access(),
				 errmsg("socket not open")));

	/* We use poll(2) if available, otherwise select(2) */
	{
#ifdef HAVE_POLL
		struct pollfd input_fd;

		input_fd.fd = PQsocket(streamConn);
		input_fd.events = POLLIN | POLLERR;
		input_fd.revents = 0;

		ret = poll(&input_fd, 1, timeout_ms);
#else							/* !HAVE_POLL */

		fd_set		input_mask;
		struct timeval timeout;
		struct timeval *ptr_timeout;

		FD_ZERO(&input_mask);
		FD_SET(PQsocket(streamConn), &input_mask);

		if (timeout_ms < 0)
			ptr_timeout = NULL;
		else
		{
			timeout.tv_sec = timeout_ms / 1000;
			timeout.tv_usec = (timeout_ms % 1000) * 1000;
			ptr_timeout = &timeout;
		}

		ret = select(PQsocket(streamConn) + 1, &input_mask,
					 NULL, NULL, ptr_timeout);
#endif   /* HAVE_POLL */
	}

	if (ret == 0 || (ret < 0 && errno == EINTR))
		return false;
	if (ret < 0)
		ereport(ERROR,
				(errcode_for_socket_access(),
				 errmsg("select() failed: %m")));
	return true;
}

int 
FileRepConnServer_StartListener(
		char	*hostAddress,
		int		portLocal)
{

	int	status = STATUS_OK;
	int i;
	
	for (i=0; i < FILEREP_MAX_LISTEN; i++) {
		listenSocket[i] = -1;
	}
	
	/* NOTE check if family AF_UNIX has to be considered as well */
	status = StreamServerPort(
					AF_UNSPEC, 
					hostAddress,
					(unsigned short) portLocal,
					NULL,
					listenSocket,
					FILEREP_MAX_LISTEN);

	if (status != STATUS_OK) {
		ereport(WARNING, 
				(errcode_for_socket_access(),
				 errmsg("could not start listener, host:'%s' port:'%d': %m",
						hostAddress,
						portLocal),
				 errSendAlert(true),
				 FileRep_errcontext()));
	}
	
	return status;
}

int
FileRepConnServer_ReceiveMessageLength(uint32 *len)

{
	int32 length;
	
	if (pq_getbytes((char*) &length, 4) == EOF) {
		ereport(WARNING,
				(errcode_for_socket_access(),
				 errmsg("receive EOF on connection: %m")));
		
		return STATUS_ERROR;
	}

	length = ntohl(length);
	
	if (length < 4) {
		ereport(WARNING,
				 (errmsg("receive unexpected message length on connection")));
		return STATUS_ERROR;
	}
	
	length -= 4;
	
	*len = length;

	return STATUS_OK;
}

/*
 * set_connection_status_bad does not remove the given connection from the connection hash.
 * It simply shuts down the underlying socket. On success, it returns true.
 */
Datum
set_connection_status_bad(PG_FUNCTION_ARGS)
{
	char *nodeName = PG_GETARG_CSTRING(0);
	int32 nodePort = PG_GETARG_INT32(1);
	int socket = -1;
	int shutdownStatus = 0;
	int pqStatus PG_USED_FOR_ASSERTS_ONLY = 0;

	PGconn *connection = GetOrEstablishConnection(nodeName, nodePort);
	if (connection == NULL)
	{
		PG_RETURN_BOOL(false);
	}

	/* Prevent further reads/writes... */
	socket = PQsocket(connection);
	shutdownStatus = shutdown(socket, SHUT_RDWR);
	if (shutdownStatus != 0)
	{
		ereport(ERROR, (errcode_for_socket_access(), errmsg("shutdown failed")));
	}

	/* ... and make libpq notice by reading data. */
	pqStatus = PQconsumeInput(connection);

	Assert(pqStatus == 0); /* expect failure */

	PG_RETURN_BOOL(true);
}

/* --------------------------------
 *		pq_recvbuf - load some bytes into the input buffer
 *
 *		returns 0 if OK, EOF if trouble
 * --------------------------------
 */
static int
pq_recvbuf(void)
{
    if (PqRecvPointer > 0)
    {
        if (PqRecvLength > PqRecvPointer)
        {
            /* still some unread data, left-justify it in the buffer */
            memmove(PqRecvBuffer, PqRecvBuffer + PqRecvPointer,
                    PqRecvLength - PqRecvPointer);
            PqRecvLength -= PqRecvPointer;
            PqRecvPointer = 0;
        }
        else
            PqRecvLength = PqRecvPointer = 0;
    }

    /* Ensure that we're in blocking mode */
    pq_set_nonblocking(false);

    /* Can fill buffer from PqRecvLength and upwards */
    for (;;)
    {
        int			r;

        r = secure_read(MyProcPort, PqRecvBuffer + PqRecvLength,
                        PQ_RECV_BUFFER_SIZE - PqRecvLength);

        if (r < 0)
        {
            if (errno == EINTR)
                continue;		/* Ok if interrupted */

            /*
             * Careful: an ereport() that tries to write to the client would
             * cause recursion to here, leading to stack overflow and core
             * dump!  This message must go *only* to the postmaster log.
             */
            ereport(COMMERROR,
                    (errcode_for_socket_access(),
                     errmsg("could not receive data from client: %m")));
            return EOF;
        }
        if (r == 0)
        {
            /*
             * EOF detected.  We used to write a log message here, but it's
             * better to expect the ultimate caller to do that.
             */
            return EOF;
        }
        /* r contains number of bytes read, so just incr length */
        PqRecvLength += r;
        return 0;
    }
}

int
FileRepConnClient_EstablishConnection(
									  char *hostAddress,
									  int port,
									  bool reportError)
{
	int			status = STATUS_OK;
	char		portbuf[11];
	char		timeoutbuf[11];
	const char *keys[5];
	const char *vals[5];

/*	FileRepConnClient_CloseConnection();*/

	snprintf(portbuf, sizeof(portbuf), "%d", port);
	snprintf(timeoutbuf, sizeof(timeoutbuf), "%d", gp_segment_connect_timeout);

	keys[0] = "host";
	vals[0] = hostAddress;
	keys[1] = "port";
	vals[1] = portbuf;
	keys[2] = "dbname";
	vals[2] = "postgres";
	keys[3] = "connect_timeout";
	vals[3] = timeoutbuf;
	keys[4] = NULL;
	vals[4] = NULL;

	filerep_conn = PQconnectdbParams(keys, vals, false);

	if (PQstatus(filerep_conn) != CONNECTION_OK)
	{
		if (reportError || Debug_filerep_print)
			ereport(WARNING,
					(errcode_for_socket_access(),
					 errmsg("could not establish connection with server, host:'%s' port:'%d' err:'%s' : %m",
							hostAddress,
							port,
							PQerrorMessage(filerep_conn)),
					 errSendAlert(true),
					 FileRep_errcontext()));

		status = STATUS_ERROR;

		if (filerep_conn)
		{
			PQfinish(filerep_conn);
			filerep_conn = NULL;
		}
	}

	/* NOTE Handle error message see ftsprobe.c */

	return status;
}

/* --------------------------------
 * pool_recvbuf - load some bytes into the input buffer
 *
 * returns 0 if OK, EOF if trouble
 * --------------------------------
 */
static int
pool_recvbuf(PoolPort *port)
{
	if (port->RecvPointer > 0)
	{
		if (port->RecvLength > port->RecvPointer)
		{
			/* still some unread data, left-justify it in the buffer */
			memmove(port->RecvBuffer, port->RecvBuffer + port->RecvPointer,
					port->RecvLength - port->RecvPointer);
			port->RecvLength -= port->RecvPointer;
			port->RecvPointer = 0;
		}
		else
			port->RecvLength = port->RecvPointer = 0;
	}

	/* Can fill buffer from PqRecvLength and upwards */
	for (;;)
	{
		int			r;

		r = recv(Socket(*port), port->RecvBuffer + port->RecvLength,
				 POOL_BUFFER_SIZE - port->RecvLength, 0);

		if (r < 0)
		{
			if (errno == EINTR)
				continue;		/* Ok if interrupted */

			/*
			 * Report broken connection
			 */
			ereport(LOG,
					(errcode_for_socket_access(),
					 errmsg("could not receive data from client: %m")));
			return EOF;
		}
		if (r == 0)
		{
			/*
			 * EOF detected.  We used to write a log message here, but it's
			 * better to expect the ultimate caller to do that.
			 */
			return EOF;
		}
		/* r contains number of bytes read, so just incr length */
		port->RecvLength += r;
		return 0;
	}
}

static int
internal_flush(void)
{
	static int	last_reported_send_errno = 0;

	char	   *bufptr = PqSendBuffer;
	char	   *bufend = PqSendBuffer + PqSendPointer;

	while (bufptr < bufend)
	{
		int			r;

		r = secure_write(MyProcPort, bufptr, bufend - bufptr);

		if (r <= 0)
		{
			if (errno == EINTR)
				continue;		/* Ok if we were interrupted */

			/*
			 * Careful: an ereport() that tries to write to the client would
			 * cause recursion to here, leading to stack overflow and core
			 * dump!  This message must go *only* to the postmaster log.
			 *
			 * If a client disconnects while we're in the midst of output, we
			 * might write quite a bit of data before we get to a safe query
			 * abort point.  So, suppress duplicate log messages.
			 */
			if (errno != last_reported_send_errno)
			{
				last_reported_send_errno = errno;
				ereport(COMMERROR,
						(errcode_for_socket_access(),
						 errmsg("could not send data to client: %m")));
			}

			/*
			 * We drop the buffered data anyway so that processing can
			 * continue, even though we'll probably quit soon.
			 */
			PqSendPointer = 0;
			return EOF;
		}

		last_reported_send_errno = 0;	/* reset after any successful send */
		bufptr += r;
	}

	PqSendPointer = 0;
	return 0;
}

/*
 * Read result from specified connection.
 * Return 0 at success or EOF at error.
 */
int
pool_recvres(PoolPort *port)
{
	int			r;
	int			res = 0;
	uint		n32;
	char		buf[SEND_RES_BUFFER_SIZE];

	r = recv(Socket(*port), &buf, SEND_RES_BUFFER_SIZE, 0);
	if (r < 0)
	{
		/*
		 * Report broken connection
		 */
		ereport(ERROR,
				(errcode_for_socket_access(),
				 errmsg("could not receive data from client: %m")));
		goto failure;
	}
	else if (r == 0)
	{
		goto failure;
	}
	else if (r != SEND_RES_BUFFER_SIZE)
	{
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg("incomplete message from client")));
		goto failure;
	}

	/* Verify response */
	if (buf[0] != 's')
	{
		ereport(ERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg("unexpected message code")));
		goto failure;
	}

	memcpy(&n32, buf + 1, 4);
	n32 = ntohl(n32);
	if (n32 != 0)
		return EOF;

	return res;

failure:
	return EOF;
}

 int
 FileRepConnServer_ReceiveMessageData(
			char		*data,
			uint32		length)
 {

	 if (pq_getbytes(data, length) == EOF) 
	 {
		 ereport(WARNING,
				 (errcode_for_socket_access(),
				  errmsg("receive EOF on connection: %m")));

		 return STATUS_ERROR;
	 }

	 return STATUS_OK;
 }

/* --------------------------------
 *		pq_getbyte_if_available - get a single byte from connection,
 *			if available
 *
 * The received byte is stored in *c. Returns 1 if a byte was read,
 * 0 if no data was available, or EOF if trouble.
 * --------------------------------
 */
int
pq_getbyte_if_available(unsigned char *c)
{
    int			r;

    if (PqRecvPointer < PqRecvLength)
    {
        *c = PqRecvBuffer[PqRecvPointer++];
        return 1;
    }

    /* Put the socket into non-blocking mode */
    pq_set_nonblocking(true);

    r = secure_read(MyProcPort, c, 1);
    if (r < 0)
    {
        /*
         * Ok if no data available without blocking or interrupted (though
         * EINTR really shouldn't happen with a non-blocking socket).
         * Report other errors.
         */
        if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR)
            r = 0;
        else
        {
            /*
             * Careful: an ereport() that tries to write to the client
             * would cause recursion to here, leading to stack overflow
             * and core dump!  This message must go *only* to the
             * postmaster log.
             */
            ereport(COMMERROR,
                    (errcode_for_socket_access(),
                     errmsg("could not receive data from client: %m")));
            r = EOF;
        }
    }
    else if (r == 0)
    {
        /* EOF detected */
        r = EOF;
    }

    return r;
}

/*
 * action can be one of EPOLL_CTL_ADD | EPOLL_CTL_MOD | EPOLL_CTL_DEL
 */
static void
WaitEventAdjustEpoll(WaitEventSet *set, WaitEvent *event, int action)
{
	struct epoll_event epoll_ev;
	int			rc;

	/* pointer to our event, returned by epoll_wait */
	epoll_ev.data.ptr = event;
	/* always wait for errors */
	epoll_ev.events = EPOLLERR | EPOLLHUP;

	/* prepare pollfd entry once */
	if (event->events == WL_LATCH_SET)
	{
		Assert(set->latch != NULL);
		epoll_ev.events |= EPOLLIN;
	}
	else if (event->events == WL_POSTMASTER_DEATH)
	{
		epoll_ev.events |= EPOLLIN;
	}
	else
	{
		Assert(event->fd != PGINVALID_SOCKET);
		Assert(event->events & (WL_SOCKET_READABLE | WL_SOCKET_WRITEABLE));

		if (event->events & WL_SOCKET_READABLE)
			epoll_ev.events |= EPOLLIN;
		if (event->events & WL_SOCKET_WRITEABLE)
			epoll_ev.events |= EPOLLOUT;
	}

	/*
	 * Even though unused, we also pass epoll_ev as the data argument if
	 * EPOLL_CTL_DEL is passed as action.  There used to be an epoll bug
	 * requiring that, and actually it makes the code simpler...
	 */
	rc = epoll_ctl(set->epoll_fd, action, event->fd, &epoll_ev);

	if (rc < 0)
		ereport(ERROR,
				(errcode_for_socket_access(),
				 errmsg("epoll_ctl() failed: %m")));
}

int
FileRepConnServer_Select(void)
{
	struct timeval timeout;
	fd_set		rfds;
	int			retval;

	timeout.tv_sec = 0;
	timeout.tv_usec = 100 * 1000L;

	FD_ZERO(&rfds);

	FD_SET(listenSocket[0], &rfds);

	retval = select(listenSocket[0] + 1, &rfds, NULL, NULL, &timeout);

	/*
	 * check and process any signals received The routine returns TRUE if the
	 * received signal requests process shutdown.
	 */

	if (retval)
	{
		if (!FD_ISSET(listenSocket[0], &rfds))
		{
			retval = -1;
		}
	}

	if (retval == -1)
	{
		ereport(WARNING,
				(errcode_for_socket_access(),
				 errmsg("receive failure on connection: %m"),
				 FileRep_errcontext()));
	}

	return retval;
}

//.........这里部分代码省略.........
			if (wakeEvents & WL_SOCKET_READABLE)
				pfds[0].events |= POLLIN;
			if (wakeEvents & WL_SOCKET_WRITEABLE)
				pfds[0].events |= POLLOUT;
			pfds[0].revents = 0;
			nfds++;
		}

		pfds[nfds].fd = selfpipe_readfd;
		pfds[nfds].events = POLLIN;
		pfds[nfds].revents = 0;
		nfds++;

		if (wakeEvents & WL_POSTMASTER_DEATH)
		{
			/* postmaster fd, if used, is always in pfds[nfds - 1] */
			pfds[nfds].fd = postmaster_alive_fds[POSTMASTER_FD_WATCH];
			pfds[nfds].events = POLLIN;
			pfds[nfds].revents = 0;
			nfds++;
		}

		/* Sleep */
		rc = poll(pfds, nfds, (int) cur_timeout);

		/* Check return code */
		if (rc < 0)
		{
			/* EINTR is okay, otherwise complain */
			if (errno != EINTR)
			{
				waiting = false;
				ereport(ERROR,
						(errcode_for_socket_access(),
						 errmsg("poll() failed: %m")));
			}
		}
		else if (rc == 0)
		{
			/* timeout exceeded */
			if (wakeEvents & WL_TIMEOUT)
				result |= WL_TIMEOUT;
		}
		else
		{
			/* at least one event occurred, so check revents values */
			if ((wakeEvents & WL_SOCKET_READABLE) &&
				(pfds[0].revents & (POLLIN | POLLHUP | POLLERR | POLLNVAL)))
			{
				/* data available in socket, or EOF/error condition */
				result |= WL_SOCKET_READABLE;
			}
			if ((wakeEvents & WL_SOCKET_WRITEABLE) &&
				(pfds[0].revents & POLLOUT))
			{
				result |= WL_SOCKET_WRITEABLE;
			}

			/*
			 * We expect a POLLHUP when the remote end is closed, but because
			 * we don't expect the pipe to become readable or to have any
			 * errors either, treat those cases as postmaster death, too.
			 */
			if ((wakeEvents & WL_POSTMASTER_DEATH) &&
				(pfds[nfds - 1].revents & (POLLHUP | POLLIN | POLLERR | POLLNVAL)))
			{

/*
 *	Attempt to negotiate SSL connection.
 */
static int
open_server_SSL(Port *port)
{
	int			r;
	int			err;

	Assert(!port->ssl);
	Assert(!port->peer);

	if (!(port->ssl = SSL_new(SSL_context)))
	{
		ereport(COMMERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg("could not initialize SSL connection: %s",
						SSLerrmessage())));
		close_SSL(port);
		return -1;
	}
	if (!my_SSL_set_fd(port->ssl, port->sock))
	{
		ereport(COMMERROR,
				(errcode(ERRCODE_PROTOCOL_VIOLATION),
				 errmsg("could not set SSL socket: %s",
						SSLerrmessage())));
		close_SSL(port);
		return -1;
	}

aloop:
	r = SSL_accept(port->ssl);
	if (r <= 0)
	{
		err = SSL_get_error(port->ssl, r);
		switch (err)
		{
			case SSL_ERROR_WANT_READ:
			case SSL_ERROR_WANT_WRITE:
#ifdef WIN32
				pgwin32_waitforsinglesocket(SSL_get_fd(port->ssl),
											(err == SSL_ERROR_WANT_READ) ?
					   FD_READ | FD_CLOSE | FD_ACCEPT : FD_WRITE | FD_CLOSE,
											INFINITE);
#endif
				goto aloop;
			case SSL_ERROR_SYSCALL:
				if (r < 0)
					ereport(COMMERROR,
							(errcode_for_socket_access(),
							 errmsg("could not accept SSL connection: %m")));
				else
					ereport(COMMERROR,
							(errcode(ERRCODE_PROTOCOL_VIOLATION),
					errmsg("could not accept SSL connection: EOF detected")));
				break;
			case SSL_ERROR_SSL:
				ereport(COMMERROR,
						(errcode(ERRCODE_PROTOCOL_VIOLATION),
						 errmsg("could not accept SSL connection: %s",
								SSLerrmessage())));
				break;
			case SSL_ERROR_ZERO_RETURN:
				ereport(COMMERROR,
						(errcode(ERRCODE_PROTOCOL_VIOLATION),
				   errmsg("could not accept SSL connection: EOF detected")));
				break;
			default:
				ereport(COMMERROR,
						(errcode(ERRCODE_PROTOCOL_VIOLATION),
						 errmsg("unrecognized SSL error code: %d",
								err)));
				break;
		}
		close_SSL(port);
		return -1;
	}

	port->count = 0;

	/* Get client certificate, if available. */
	port->peer = SSL_get_peer_certificate(port->ssl);

	/* and extract the Common Name from it. */
	port->peer_cn = NULL;
	if (port->peer != NULL)
	{
		int len;

		len = X509_NAME_get_text_by_NID(X509_get_subject_name(port->peer),
						NID_commonName, NULL, 0);

		if (len != -1)
		{
			char *peer_cn;

			peer_cn = MemoryContextAlloc(TopMemoryContext, len + 1);
			r = X509_NAME_get_text_by_NID(X509_get_subject_name(port->peer),
						      NID_commonName, peer_cn, len+1);
//.........这里部分代码省略.........

/*
 * Wait using linux's epoll_wait(2).
 *
 * This is the preferrable wait method, as several readiness notifications are
 * delivered, without having to iterate through all of set->events. The return
 * epoll_event struct contain a pointer to our events, making association
 * easy.
 */
static inline int
WaitEventSetWaitBlock(WaitEventSet *set, int cur_timeout,
					  WaitEvent *occurred_events, int nevents)
{
	int			returned_events = 0;
	int			rc;
	WaitEvent  *cur_event;
	struct epoll_event *cur_epoll_event;

	/* Sleep */
	rc = epoll_wait(set->epoll_fd, set->epoll_ret_events,
					nevents, cur_timeout);

	/* Check return code */
	if (rc < 0)
	{
		/* EINTR is okay, otherwise complain */
		if (errno != EINTR)
		{
			waiting = false;
			ereport(ERROR,
					(errcode_for_socket_access(),
					 errmsg("epoll_wait() failed: %m")));
		}
		return 0;
	}
	else if (rc == 0)
	{
		/* timeout exceeded */
		return -1;
	}

	/*
	 * At least one event occurred, iterate over the returned epoll events
	 * until they're either all processed, or we've returned all the events
	 * the caller desired.
	 */
	for (cur_epoll_event = set->epoll_ret_events;
		 cur_epoll_event < (set->epoll_ret_events + rc) &&
		 returned_events < nevents;
		 cur_epoll_event++)
	{
		/* epoll's data pointer is set to the associated WaitEvent */
		cur_event = (WaitEvent *) cur_epoll_event->data.ptr;

		occurred_events->pos = cur_event->pos;
		occurred_events->user_data = cur_event->user_data;
		occurred_events->events = 0;

		if (cur_event->events == WL_LATCH_SET &&
			cur_epoll_event->events & (EPOLLIN | EPOLLERR | EPOLLHUP))
		{
			/* There's data in the self-pipe, clear it. */
			drainSelfPipe();

			if (set->latch->is_set)
			{
				occurred_events->fd = PGINVALID_SOCKET;
				occurred_events->events = WL_LATCH_SET;
				occurred_events++;
				returned_events++;
			}
		}
		else if (cur_event->events == WL_POSTMASTER_DEATH &&
				 cur_epoll_event->events & (EPOLLIN | EPOLLERR | EPOLLHUP))
		{
			/*
			 * We expect an EPOLLHUP when the remote end is closed, but
			 * because we don't expect the pipe to become readable or to have
			 * any errors either, treat those cases as postmaster death, too.
			 *
			 * As explained in the WAIT_USE_SELECT implementation, select(2)
			 * may spuriously return. Be paranoid about that here too, a
			 * spurious WL_POSTMASTER_DEATH would be painful.
			 */
			if (!PostmasterIsAlive())
			{
				occurred_events->fd = PGINVALID_SOCKET;
				occurred_events->events = WL_POSTMASTER_DEATH;
				occurred_events++;
				returned_events++;
			}
		}
		else if (cur_event->events & (WL_SOCKET_READABLE | WL_SOCKET_WRITEABLE))
		{
			Assert(cur_event->fd != PGINVALID_SOCKET);

			if ((cur_event->events & WL_SOCKET_READABLE) &&
				(cur_epoll_event->events & (EPOLLIN | EPOLLERR | EPOLLHUP)))
			{
				/* data available in socket, or EOF */
				occurred_events->events |= WL_SOCKET_READABLE;
//.........这里部分代码省略.........

/*
 * Client authentication starts here.  If there is an error, this
 * function does not return and the backend process is terminated.
 */
void
ClientAuthentication(Port *port)
{
	int			status = STATUS_ERROR;

	/*
	 * Get the authentication method to use for this frontend/database
	 * combination.  Note: a failure return indicates a problem with the
	 * hba config file, not with the request.  hba.c should have dropped
	 * an error message into the postmaster logfile if it failed.
	 */
	if (hba_getauthmethod(port) != STATUS_OK)
		ereport(FATAL,
				(errcode(ERRCODE_CONFIG_FILE_ERROR),
				 errmsg("missing or erroneous pg_hba.conf file"),
				 errhint("See server log for details.")));

	switch (port->auth_method)
	{
		case uaReject:

			/*
			 * This could have come from an explicit "reject" entry in
			 * pg_hba.conf, but more likely it means there was no matching
			 * entry.  Take pity on the poor user and issue a helpful
			 * error message.  NOTE: this is not a security breach,
			 * because all the info reported here is known at the frontend
			 * and must be assumed known to bad guys. We're merely helping
			 * out the less clueful good guys.
			 */
			{
				char		hostinfo[NI_MAXHOST];

				getnameinfo_all(&port->raddr.addr, port->raddr.salen,
								hostinfo, sizeof(hostinfo),
								NULL, 0,
								NI_NUMERICHOST);

#ifdef USE_SSL
				ereport(FATAL,
				   (errcode(ERRCODE_INVALID_AUTHORIZATION_SPECIFICATION),
					errmsg("no pg_hba.conf entry for host \"%s\", user \"%s\", database \"%s\", %s",
						   hostinfo, port->user_name, port->database_name,
				   port->ssl ? gettext("SSL on") : gettext("SSL off"))));
#else
				ereport(FATAL,
				   (errcode(ERRCODE_INVALID_AUTHORIZATION_SPECIFICATION),
					errmsg("no pg_hba.conf entry for host \"%s\", user \"%s\", database \"%s\"",
					   hostinfo, port->user_name, port->database_name)));
#endif
				break;
			}

		case uaKrb4:
			/* Kerberos 4 only seems to work with AF_INET. */
			if (port->raddr.addr.ss_family != AF_INET
				|| port->laddr.addr.ss_family != AF_INET)
				ereport(FATAL,
						(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				   errmsg("Kerberos 4 only supports IPv4 connections")));
			sendAuthRequest(port, AUTH_REQ_KRB4);
			status = pg_krb4_recvauth(port);
			break;

		case uaKrb5:
			sendAuthRequest(port, AUTH_REQ_KRB5);
			status = pg_krb5_recvauth(port);
			break;

		case uaIdent:
			/*
			 * If we are doing ident on unix-domain sockets, use SCM_CREDS
			 * only if it is defined and SO_PEERCRED isn't.
			 */
#if !defined(HAVE_GETPEEREID) && !defined(SO_PEERCRED) && \
	(defined(HAVE_STRUCT_CMSGCRED) || defined(HAVE_STRUCT_FCRED) || \
	 (defined(HAVE_STRUCT_SOCKCRED) && defined(LOCAL_CREDS)))
			if (port->raddr.addr.ss_family == AF_UNIX)
			{
#if defined(HAVE_STRUCT_FCRED) || defined(HAVE_STRUCT_SOCKCRED)
				/*
				 * Receive credentials on next message receipt, BSD/OS,
				 * NetBSD. We need to set this before the client sends the
				 * next packet.
				 */
				int			on = 1;

				if (setsockopt(port->sock, 0, LOCAL_CREDS, &on, sizeof(on)) < 0)
					ereport(FATAL,
							(errcode_for_socket_access(),
					 errmsg("could not enable credential reception: %m")));
#endif

				sendAuthRequest(port, AUTH_REQ_SCM_CREDS);
			}
#endif
//.........这里部分代码省略.........

/*
 * StreamConnection -- create a new connection with client using
 *		server port.  Set port->sock to the FD of the new connection.
 *
 * ASSUME: that this doesn't need to be non-blocking because
 *		the Postmaster uses select() to tell when the server master
 *		socket is ready for accept().
 *
 * RETURNS: STATUS_OK or STATUS_ERROR
 */
int
StreamConnection(pgsocket server_fd, Port *port)
{
	/* accept connection and fill in the client (remote) address */
	port->raddr.salen = sizeof(port->raddr.addr);
	if ((port->sock = accept(server_fd,
							 (struct sockaddr *) & port->raddr.addr,
							 &port->raddr.salen)) < 0)
	{
		ereport(LOG,
				(errcode_for_socket_access(),
				 errmsg("could not accept new connection: %m")));

		/*
		 * If accept() fails then postmaster.c will still see the server
		 * socket as read-ready, and will immediately try again.  To avoid
		 * uselessly sucking lots of CPU, delay a bit before trying again.
		 * (The most likely reason for failure is being out of kernel file
		 * table slots; we can do little except hope some will get freed up.)
		 */
		pg_usleep(100000L);		/* wait 0.1 sec */
		return STATUS_ERROR;
	}

#ifdef SCO_ACCEPT_BUG

	/*
	 * UnixWare 7+ and OpenServer 5.0.4 are known to have this bug, but it
	 * shouldn't hurt to catch it for all versions of those platforms.
	 */
	if (port->raddr.addr.ss_family == 0)
		port->raddr.addr.ss_family = AF_UNIX;
#endif

	/* fill in the server (local) address */
	port->laddr.salen = sizeof(port->laddr.addr);
	if (getsockname(port->sock,
					(struct sockaddr *) & port->laddr.addr,
					&port->laddr.salen) < 0)
	{
		elog(LOG, "getsockname() failed: %m");
		return STATUS_ERROR;
	}

	/* select NODELAY and KEEPALIVE options if it's a TCP connection */
	if (!IS_AF_UNIX(port->laddr.addr.ss_family))
	{
		int			on;

#ifdef	TCP_NODELAY
		on = 1;
		if (setsockopt(port->sock, IPPROTO_TCP, TCP_NODELAY,
					   (char *) &on, sizeof(on)) < 0)
		{
			elog(LOG, "setsockopt(TCP_NODELAY) failed: %m");
			return STATUS_ERROR;
		}
#endif
		on = 1;
		if (setsockopt(port->sock, SOL_SOCKET, SO_KEEPALIVE,
					   (char *) &on, sizeof(on)) < 0)
		{
			elog(LOG, "setsockopt(SO_KEEPALIVE) failed: %m");
			return STATUS_ERROR;
		}

#ifdef WIN32

		/*
		 * This is a Win32 socket optimization.  The ideal size is 32k.
		 * http://support.microsoft.com/kb/823764/EN-US/
		 */
		on = PQ_BUFFER_SIZE * 4;
		if (setsockopt(port->sock, SOL_SOCKET, SO_SNDBUF, (char *) &on,
					   sizeof(on)) < 0)
		{
			elog(LOG, "setsockopt(SO_SNDBUF) failed: %m");
			return STATUS_ERROR;
		}
#endif

		/*
		 * Also apply the current keepalive parameters.  If we fail to set a
		 * parameter, don't error out, because these aren't universally
		 * supported.  (Note: you might think we need to reset the GUC
		 * variables to 0 in such a case, but it's not necessary because the
		 * show hooks for these variables report the truth anyway.)
		 */
		(void) pq_setkeepalivesidle(tcp_keepalives_idle, port);
		(void) pq_setkeepalivesinterval(tcp_keepalives_interval, port);
//.........这里部分代码省略.........