Store zlog stream in each child so it can be reused

This change results in using the same buffer for multiple
stdio events which should fix inconsistencies of handling
messages that are not ended with a new line and possibly
very long messages that are split to multiple events.

8 files changed, 92 insertions, 53 deletions

diff --git a/sapi/fpm/fpm/fpm_children.c b/sapi/fpm/fpm/fpm_children.c
index 505f71d9d5..ca2ff7dec3 100644
--- a/sapi/fpm/fpm/fpm_children.c
+++ b/sapi/fpm/fpm/fpm_children.c
@@ -57,6 +57,10 @@ static struct fpm_child_s *fpm_child_alloc() /* {{{ */
 
 static void fpm_child_free(struct fpm_child_s *child) /* {{{ */
 {
+	if (child->log_stream) {
+		zlog_stream_close(child->log_stream);
+		free(child->log_stream);
+	}
 	free(child);
 }
 /* }}} */
diff --git a/sapi/fpm/fpm/fpm_children.h b/sapi/fpm/fpm/fpm_children.h
index 1a846a8786..04c857bec7 100644
--- a/sapi/fpm/fpm/fpm_children.h
+++ b/sapi/fpm/fpm/fpm_children.h
@@ -9,6 +9,7 @@
 
 #include "fpm_worker_pool.h"
 #include "fpm_events.h"
+#include "zlog.h"
 
 int fpm_children_create_initial(struct fpm_worker_pool_s *wp);
 int fpm_children_free(struct fpm_child_s *child);
@@ -30,6 +31,7 @@ struct fpm_child_s {
 	int idle_kill;
 	pid_t pid;
 	int scoreboard_i;
+	struct zlog_stream *log_stream;
 };
 
 #endif
diff --git a/sapi/fpm/fpm/fpm_request.c b/sapi/fpm/fpm/fpm_request.c
index 9dfdb0f91e..56d2a5244f 100644
--- a/sapi/fpm/fpm/fpm_request.c
+++ b/sapi/fpm/fpm/fpm_request.c
@@ -17,6 +17,7 @@
 #include "fpm_children.h"
 #include "fpm_scoreboard.h"
 #include "fpm_status.h"
+#include "fpm_stdio.h"
 #include "fpm_request.h"
 #include "fpm_log.h"
 
@@ -200,6 +201,7 @@ void fpm_request_end(void) /* {{{ */
 #endif
 	proc->memory = memory;
 	fpm_scoreboard_proc_release(proc);
+	fpm_stdio_flush_child();
 }
 /* }}} */
 
diff --git a/sapi/fpm/fpm/fpm_stdio.c b/sapi/fpm/fpm/fpm_stdio.c
index 599333e3c4..f5768cc506 100644
--- a/sapi/fpm/fpm/fpm_stdio.c
+++ b/sapi/fpm/fpm/fpm_stdio.c
@@ -23,12 +23,6 @@
 static int fd_stdout[2];
 static int fd_stderr[2];
 
-static void fpm_stdio_cleanup(int which, void *arg) /* {{{ */
-{
-	zlog_cleanup();
-}
-/* }}} */
-
 int fpm_stdio_init_main() /* {{{ */
 {
 	int fd = open("/dev/null", O_RDWR);
@@ -37,9 +31,6 @@ int fpm_stdio_init_main() /* {{{ */
 		zlog(ZLOG_SYSERROR, "failed to init stdio: open(\"/dev/null\")");
 		return -1;
 	}
-	if (0 > fpm_cleanup_add(FPM_CLEANUP_PARENT, fpm_stdio_cleanup, 0)) {
-		return -1;
-	}
 
 	if (0 > dup2(fd, STDIN_FILENO) || 0 > dup2(fd, STDOUT_FILENO)) {
 		zlog(ZLOG_SYSERROR, "failed to init stdio: dup2()");
@@ -116,6 +107,12 @@ int fpm_stdio_init_child(struct fpm_worker_pool_s *wp) /* {{{ */
 }
 /* }}} */
 
+int fpm_stdio_flush_child() /* {{{ */
+{
+	return write(STDERR_FILENO, "\0", 1);
+}
+/* }}} */
+
 static void fpm_stdio_child_said(struct fpm_event_s *ev, short which, void *arg) /* {{{ */
 {
 	static const int max_buf_size = 1024;
@@ -126,9 +123,9 @@ static void fpm_stdio_child_said(struct fpm_event_s *ev, short which, void *arg)
 	struct fpm_event_s *event;
 	int fifo_in = 1, fifo_out = 1;
 	int in_buf = 0;
-	int read_fail = 0;
+	int read_fail = 0, finish_log_stream = 0;
 	int res;
-	struct zlog_stream stream;
+	struct zlog_stream *log_stream;
 
 	if (!arg) {
 		return;
@@ -142,12 +139,17 @@ static void fpm_stdio_child_said(struct fpm_event_s *ev, short which, void *arg)
 		event = &child->ev_stderr;
 	}
 
-	zlog_stream_init_ex(&stream, ZLOG_WARNING, STDERR_FILENO);
-	zlog_stream_set_decorating(&stream, child->wp->config->decorate_workers_output);
-	zlog_stream_set_wrapping(&stream, ZLOG_TRUE);
-	zlog_stream_set_msg_prefix(&stream, "[pool %s] child %d said into %s: ",
-			child->wp->config->name, (int) child->pid, is_stdout ? "stdout" : "stderr");
-	zlog_stream_set_msg_quoting(&stream, ZLOG_TRUE);
+	if (!child->log_stream) {
+		log_stream = child->log_stream = malloc(sizeof(struct zlog_stream));
+		zlog_stream_init_ex(log_stream, ZLOG_WARNING, STDERR_FILENO);
+		zlog_stream_set_decorating(log_stream, child->wp->config->decorate_workers_output);
+		zlog_stream_set_wrapping(log_stream, ZLOG_TRUE);
+		zlog_stream_set_msg_prefix(log_stream, "[pool %s] child %d said into %s: ",
+				child->wp->config->name, (int) child->pid, is_stdout ? "stdout" : "stderr");
+		zlog_stream_set_msg_quoting(log_stream, ZLOG_TRUE);
+	} else {
+		log_stream = child->log_stream;
+	}
 
 	while (fifo_in || fifo_out) {
 		if (fifo_in) {
@@ -160,6 +162,11 @@ static void fpm_stdio_child_said(struct fpm_event_s *ev, short which, void *arg)
 				}
 			} else {
 				in_buf += res;
+				/* if buffer ends with \0, then the stream will be finished */
+				if (!buf[in_buf - 1]) {
+					finish_log_stream = 1;
+					in_buf--;
+				}
 			}
 		}
 
@@ -173,8 +180,8 @@ static void fpm_stdio_child_said(struct fpm_event_s *ev, short which, void *arg)
 				if (nl) {
 					/* we should print each new line int the new message */
 					int out_len = nl - buf;
-					zlog_stream_str(&stream, buf, out_len);
-					zlog_stream_finish(&stream);
+					zlog_stream_str(log_stream, buf, out_len);
+					zlog_stream_finish(log_stream);
 					/* skip new line */
 					out_len++;
 					/* move data in the buffer */
@@ -182,7 +189,7 @@ static void fpm_stdio_child_said(struct fpm_event_s *ev, short which, void *arg)
 					in_buf -= out_len;
 				} else if (in_buf == max_buf_size - 1 || !fifo_in) {
 					/* we should print if no more space in the buffer or no more data to come */
-					zlog_stream_str(&stream, buf, in_buf);
+					zlog_stream_str(log_stream, buf, in_buf);
 					in_buf = 0;
 				}
 			}
@@ -190,8 +197,8 @@ static void fpm_stdio_child_said(struct fpm_event_s *ev, short which, void *arg)
 	}
 
 	if (read_fail) {
-		zlog_stream_set_msg_suffix(&stream, NULL, ", pipe is closed");
-		zlog_stream_close(&stream);
+		zlog_stream_set_msg_suffix(log_stream, NULL, ", pipe is closed");
+		zlog_stream_finish(log_stream);
 		if (read_fail < 0) {
 			zlog(ZLOG_SYSERROR, "unable to read what child say");
 		}
@@ -205,8 +212,8 @@ static void fpm_stdio_child_said(struct fpm_event_s *ev, short which, void *arg)
 			close(child->fd_stderr);
 			child->fd_stderr = -1;
 		}
-	} else {
-		zlog_stream_close(&stream);
+	} else if (finish_log_stream) {
+		zlog_stream_finish(log_stream);
 	}
 }
 /* }}} */
diff --git a/sapi/fpm/fpm/fpm_stdio.h b/sapi/fpm/fpm/fpm_stdio.h
index d9c6db24f8..1e16ec0956 100644
--- a/sapi/fpm/fpm/fpm_stdio.h
+++ b/sapi/fpm/fpm/fpm_stdio.h
@@ -9,6 +9,7 @@
 int fpm_stdio_init_main();
 int fpm_stdio_init_final();
 int fpm_stdio_init_child(struct fpm_worker_pool_s *wp);
+int fpm_stdio_flush_child();
 int fpm_stdio_prepare_pipes(struct fpm_child_s *child);
 void fpm_stdio_child_use_pipes(struct fpm_child_s *child);
 int fpm_stdio_parent_use_pipes(struct fpm_child_s *child);
diff --git a/sapi/fpm/fpm/zlog.c b/sapi/fpm/fpm/zlog.c
index e7e07cbc9c..c507e931e5 100644
--- a/sapi/fpm/fpm/zlog.c
+++ b/sapi/fpm/fpm/zlog.c
@@ -29,7 +29,6 @@ static int zlog_fd = -1;
 static int zlog_level = ZLOG_NOTICE;
 static int zlog_limit = ZLOG_DEFAULT_LIMIT;
 static zlog_bool zlog_buffering = ZLOG_DEFAULT_BUFFERING;
-static struct zlog_stream_buffer zlog_buffer = {NULL, 0};
 static int launched = 0;
 static void (*external_logger)(int, char *, size_t) = NULL;
 
@@ -128,16 +127,6 @@ int zlog_set_buffering(zlog_bool buffering) /* {{{ */
 }
 /* }}} */
 
-void zlog_cleanup() /* {{{ */
-{
-	if (zlog_buffer.data) {
-		free(zlog_buffer.data);
-		zlog_buffer.data = NULL;
-		zlog_buffer.size = 0;
-	}
-}
-/* }}} */
-
 static inline size_t zlog_truncate_buf(char *buf, size_t buf_size, size_t space_left) /* {{{ */
 {
 	memcpy(buf + buf_size - sizeof("...") + 1 - space_left, "...", sizeof("...") - 1);
@@ -304,16 +293,7 @@ static zlog_bool zlog_stream_buf_alloc_ex(struct zlog_stream *stream, size_t nee
 		buf = realloc(stream->buf.data, size);
 	} else {
 		size = MIN(zlog_limit, MAX(size, needed));
-		if (stream->shared_buffer && zlog_buffer.data) {
-			if (zlog_buffer.size < size) {
-				buf = realloc(stream->buf.data, size);
-			} else {
-				buf = zlog_buffer.data;
-				size = zlog_buffer.size;
-			}
-		} else {
-			buf = malloc(size);
-		}
+		buf = malloc(size);
 	}
 
 	if (buf == NULL) {
@@ -322,10 +302,6 @@ static zlog_bool zlog_stream_buf_alloc_ex(struct zlog_stream *stream, size_t nee
 
 	stream->buf.data = buf;
 	stream->buf.size = size;
-	if (stream->shared_buffer) {
-		zlog_buffer.data = buf;
-		zlog_buffer.size = size;
-	}
 
 	return 1;
 }
@@ -554,7 +530,6 @@ static inline void zlog_stream_init_internal(
 	stream->prefix_buffer = (flags & ZLOG_LEVEL_MASK) >= zlog_level &&
 			(stream->use_fd || stream->use_stderr || stream->use_syslog);
 	stream->fd = fd > -1 ? fd : STDERR_FILENO;
-	stream->shared_buffer = external_logger == NULL;
 }
 /* }}} */
 
@@ -858,7 +833,7 @@ zlog_bool zlog_stream_finish(struct zlog_stream *stream) /* {{{ */
 
 void zlog_stream_destroy(struct zlog_stream *stream) /* {{{ */
 {
-	if (!stream->shared_buffer && stream->buf.data != NULL) {
+	if (stream->buf.data != NULL) {
 		free(stream->buf.data);
 	}
 	if (stream->msg_prefix != NULL) {
diff --git a/sapi/fpm/fpm/zlog.h b/sapi/fpm/fpm/zlog.h
index 3869523a3b..4b9f732cdf 100644
--- a/sapi/fpm/fpm/zlog.h
+++ b/sapi/fpm/fpm/zlog.h
@@ -24,7 +24,6 @@ int zlog_set_limit(int new_value);
 int zlog_set_buffering(zlog_bool buffering);
 const char *zlog_get_level_name(int log_level);
 void zlog_set_launched(void);
-void zlog_cleanup();
 
 size_t zlog_print_time(struct timeval *tv, char *timebuf, size_t timebuf_len);
 
@@ -75,7 +74,6 @@ struct zlog_stream {
 	unsigned int wrap:1;
 	unsigned int msg_quote:1;
 	unsigned int decorate:1;
-	unsigned int shared_buffer:1;
 	int fd;
 	int line;
 	const char *function;
diff --git a/sapi/fpm/tests/log-bwd-multiple-msgs.phpt b/sapi/fpm/tests/log-bwd-multiple-msgs.phpt
new file mode 100644
index 0000000000..a21e4722b3
--- /dev/null
+++ b/sapi/fpm/tests/log-bwd-multiple-msgs.phpt
@@ -0,0 +1,50 @@
+--TEST--
+FPM: Buffered worker output decorated log with multiple continuous messages
+--SKIPIF--
+<?php include "skipif.inc"; ?>
+--FILE--
+<?php
+
+require_once "tester.inc";
+
+$cfg = <<<EOT
+[global]
+error_log = {{FILE:LOG}}
+[unconfined]
+listen = {{ADDR}}
+pm = dynamic
+pm.max_children = 5
+pm.start_servers = 1
+pm.min_spare_servers = 1
+pm.max_spare_servers = 3
+catch_workers_output = yes
+EOT;
+
+$code = <<<EOT
+<?php
+file_put_contents('php://stderr', "msg 1 - ");
+usleep(1);
+file_put_contents('php://stderr', "msg 2 - ");
+usleep(1);
+file_put_contents('php://stderr', "msg 3");
+EOT;
+
+$tester = new FPM\Tester($cfg, $code);
+$tester->start();
+$tester->expectLogStartNotices();
+$tester->request()->expectEmptyBody();
+$tester->request()->expectEmptyBody();
+$tester->terminate();
+$tester->expectLogLine('msg 1 - msg 2 - msg 3');
+$tester->expectLogLine('msg 1 - msg 2 - msg 3');
+$tester->close();
+
+?>
+Done
+--EXPECT--
+Done
+--CLEAN--
+<?php
+require_once "tester.inc";
+FPM\Tester::clean();
+?>
\ No newline at end of file