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.

Author: bukka

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);
 }
 /* }}} */

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

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();
 }
 /* }}} */
 

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,25 +180,25 @@ 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 */
 					memmove(buf, buf + out_len, in_buf - out_len);
 					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;
 				}
 			}
 		}
 	}
 
 	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);
 	}
 }
 /* }}} */

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);

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) {

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;

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();
+?>