fixed bugs in IO proccess of master_sig.c/master_status.c

Author: ubuntu14

lib_acl/src/master/framework/master_sig.c

@@ -51,7 +51,6 @@ int   acl_master_sig_pipe[2];
 #define ACL_SIG_PIPE_READ_FD acl_master_sig_pipe[0]
 #endif
 
-ACL_VSTREAM *ACL_SIG_PIPE_WRITE_STREAM = NULL;
 ACL_VSTREAM *ACL_SIG_PIPE_READ_STREAM = NULL;
 
 int   acl_var_master_gotsigchld = 0;
@@ -126,19 +125,29 @@ static void master_sigchld(int sig acl_unused)
 /* master_sig_event - called upon return from select() */
 
 static void master_sig_event(int type acl_unused, ACL_EVENT *event acl_unused,
-	ACL_VSTREAM *stream acl_unused, void *context acl_unused)
+	ACL_VSTREAM *stream, void *context acl_unused)
 {
 	char    c[1];
 
-#if 0
-	while (read(ACL_SIG_PIPE_READ_FD, c, 1) > 0)
-		/* void */ ;
-#else
-	while (acl_vstream_read(stream, c, 1) > 0)
-	{
-		/* void */
+	if (stream->rw_timeout > 0) {
+		acl_msg_warn("%s:%d, pipe(%d)'s rw_timeout(%d) > 0",
+			__FUNCTION__, __LINE__, ACL_VSTREAM_SOCK(stream),
+			stream->rw_timeout);
+		stream->rw_timeout = 0;
 	}
-#endif
+
+	while (1) {
+		if (acl_vstream_read(stream, c, 1) == ACL_VSTREAM_EOF) {
+			if (acl_msg_verbose)
+				acl_msg_info(">>>%s:%d: %s<<<", __FUNCTION__,
+					__LINE__, acl_last_serror());
+			break;
+		} else if (acl_msg_verbose) {
+			acl_msg_info(">>>%s:%d: %s, c: %d<<<", __FUNCTION__,
+				__LINE__, acl_last_serror(), c[0]);
+		}
+	}
+
 	acl_var_master_gotsigchld = 1;
 }
 
@@ -231,14 +240,18 @@ void    acl_master_sigsetup(void)
 #ifdef USE_SIG_PIPE
 	if (pipe(acl_master_sig_pipe))
 		acl_msg_fatal("pipe: %s", strerror(errno));
+
 	acl_non_blocking(ACL_SIG_PIPE_WRITE_FD, ACL_NON_BLOCKING);
 	acl_non_blocking(ACL_SIG_PIPE_READ_FD, ACL_NON_BLOCKING);
 	acl_close_on_exec(ACL_SIG_PIPE_WRITE_FD, ACL_CLOSE_ON_EXEC);
 	acl_close_on_exec(ACL_SIG_PIPE_READ_FD, ACL_CLOSE_ON_EXEC);
 
+	/* Must set io rw_timeout to 0 to avoiding blocking read which
+	 * will blocking the main event loop.
+	 */
 	ACL_SIG_PIPE_READ_STREAM = acl_vstream_fdopen(ACL_SIG_PIPE_READ_FD,
-			O_RDONLY, acl_var_master_buf_size,
-			acl_var_master_rw_timeout, ACL_VSTREAM_TYPE_SOCK);
+		O_RDONLY, acl_var_master_buf_size, 0, ACL_VSTREAM_TYPE_SOCK);
+
 	if (ACL_SIG_PIPE_READ_STREAM == NULL)
 		acl_msg_fatal("%s(%d)->%s: acl_vstream_fdopen error=%s",
 			__FILE__, __LINE__, myname, strerror(errno));
@@ -247,6 +260,7 @@ void    acl_master_sigsetup(void)
 		acl_msg_info("%s(%d)->%s: call acl_event_enable_read, "
 			"SIG_PIPE_READ_FD = %d", __FILE__, __LINE__, myname,
 			ACL_SIG_PIPE_READ_FD);
+
 	acl_event_enable_read(acl_var_master_global_event,
 		ACL_SIG_PIPE_READ_STREAM, 0, master_sig_event, (void *) 0);
 #endif

lib_acl/src/master/framework/master_status.c

@@ -46,22 +46,26 @@ static void master_status_event(int type, ACL_EVENT *event acl_unused,
 		return;
 
 	/*
-	 * We always keep the child end of the status pipe open, so an EOF read
-	 * condition means that we're seriously confused. We use non-blocking
-	 * reads so that we don't get stuck when someone sends a partial message.
-	 * Messages are short, so a partial read means someone wrote less than a
-	 * whole status message. Hopefully the next read will be in sync again...
-	 * We use a global child process status table because when a child dies
-	 * only its pid is known - we do not know what service it came from.
+	 * We always keep the child end of the status pipe open, so an EOF
+	 * read condition means that we're seriously confused. We use
+	 * non-blocking reads so that we don't get stuck when someone sends
+	 * a partial message. Messages are short, so a partial read means
+	 * someone wrote less than a whole status message. Hopefully the next
+	 * read will be in sync again... We use a global child process status
+	 * table because when a child dies only its pid is known - we do not
+	 * know what service it came from.
 	 */
 
-#if 0
-	n = read(ACL_VSTREAM_SOCK(serv->status_read_stream),
-		(char *) &stat_buf, sizeof(stat_buf));
-#else
+	if (serv->status_read_stream->rw_timeout > 0) {
+		acl_msg_warn("%s:%d, pipe(%d)'s rw_timeout(%d) > 0",
+			__FUNCTION__, __LINE__,
+			ACL_VSTREAM_SOCK(serv->status_read_stream),
+			serv->status_read_stream->rw_timeout);
+		serv->status_read_stream->rw_timeout = 0;
+	}
+
 	n = acl_vstream_read(serv->status_read_stream,
 		(char *) &stat_buf, sizeof(stat_buf));
-#endif
 
 	switch (n) {
 	case -1:
@@ -167,21 +171,24 @@ void    acl_master_status_init(ACL_MASTER_SERV *serv)
 	 */
 	if (acl_duplex_pipe(serv->status_fd) < 0)
 		acl_msg_fatal("pipe: %s", strerror(errno));
+
 	acl_non_blocking(serv->status_fd[0], ACL_BLOCKING);
 	acl_close_on_exec(serv->status_fd[0], ACL_CLOSE_ON_EXEC);
 	acl_close_on_exec(serv->status_fd[1], ACL_CLOSE_ON_EXEC);
+
+	/* Must set io rw_timeout to 0 to avoiding blocking read which
+	 * will blocking the main event loop.
+	 */
 	serv->status_read_stream = acl_vstream_fdopen(serv->status_fd[0],
-		O_RDWR, acl_var_master_buf_size,
-		acl_var_master_rw_timeout, ACL_VSTREAM_TYPE_SOCK);
+		O_RDWR, acl_var_master_buf_size, 0, ACL_VSTREAM_TYPE_SOCK);
 
 	if (acl_msg_verbose)
 		acl_msg_info("%s(%d)->%s: call acl_event_enable_read, "
 			"status_fd = %d", __FILE__, __LINE__,
 			myname, serv->status_fd[0]);
 
 	acl_event_enable_read(acl_var_master_global_event,
-		serv->status_read_stream, 0, master_status_event,
-		(void *) serv);
+		serv->status_read_stream, 0, master_status_event, serv);
 }
 
 /* acl_master_status_cleanup - stop status event processing for this service */

lib_acl/src/stdlib/iostuff/acl_read_wait.c

@@ -57,7 +57,7 @@ static void thread_epoll_init(void)
 int acl_read_wait(ACL_SOCKET fd, int timeout)
 {
 	const char *myname = "acl_read_wait";
-	int op = EPOLL_CTL_ADD, delay = timeout * 1000, *epoll_fd;
+	int op = EPOLL_CTL_ADD, delay = timeout * 1000, *epoll_fd, ret;
 	struct epoll_event ee, events[1];
 
 	acl_assert(acl_pthread_once(&epoll_once, thread_epoll_init) == 0);
@@ -78,16 +78,41 @@ int acl_read_wait(ACL_SOCKET fd, int timeout)
 	ee.events = EPOLLIN | EPOLLHUP | EPOLLERR;
 	ee.data.u64 = 0;
 	ee.data.fd = fd;
-	if (epoll_ctl(*epoll_fd, op, fd, &ee) == -1) {
+	if (epoll_ctl(*epoll_fd, op, fd, &ee) == -1
+		&& acl_last_error() != EEXIST)
+	{
 		acl_msg_error("%s(%d): epoll_ctl error: %s, fd: %d",
 			myname, __LINE__, acl_last_serror(), fd);
 		return -1;
 	}
 
-	if (epoll_wait(*epoll_fd, events, 1, delay) == -1) {
-		acl_msg_error("%s(%d): epoll_wait error: %s, fd: %d",
-			myname, __LINE__, acl_last_serror(), fd);
-		return -1;
+	for (;;) {
+		switch (epoll_wait(*epoll_fd, events, 1, delay)) {
+		case -1:
+			if (acl_last_error() == ACL_EINTR)
+			{
+				acl_msg_warn(">>>>catch EINTR, try again<<<");
+				continue;
+			}
+
+			acl_msg_error("%s(%d): epoll_wait error: %s, fd: %d",
+				myname, __LINE__, acl_last_serror(), fd);
+			ret = -1;
+			break;
+		case 0:
+			acl_set_error(ACL_ETIMEDOUT);
+			ret = -1;
+			break;
+		default:
+			if ((events[0].events & (EPOLLERR | EPOLLHUP)) != 0)
+				ret = -1;
+			else if ((events[0].events & EPOLLIN) == 0) {
+				acl_set_error(ACL_ETIMEDOUT);
+				ret = -1;
+			} else
+				ret = 0;
+			break;
+		}
 	}
 
 	ee.events = 0;
@@ -99,15 +124,7 @@ int acl_read_wait(ACL_SOCKET fd, int timeout)
 		return -1;
 	}
 
-	if ((events[0].events & (EPOLLERR | EPOLLHUP)) != 0)
-		return -1;
-
-	if ((events[0].events & EPOLLIN) == 0) {
-		acl_set_error(ACL_ETIMEDOUT);
-		return -1;
-	}
-
-	return 0;
+	return ret;
 }
 
 #elif	defined(ACL_UNIX)