Child -> Parent pipe for COW info transferring.

6 files changed, 132 insertions, 3 deletions

diff --git a/src/Makefile b/src/Makefile
index 6bd8d8d66..2bf3c9347 100644
--- a/src/Makefile
+++ b/src/Makefile
@@ -128,7 +128,7 @@ endif
 
 REDIS_SERVER_NAME=redis-server
 REDIS_SENTINEL_NAME=redis-sentinel
-REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o
+REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o
 REDIS_CLI_NAME=redis-cli
 REDIS_CLI_OBJ=anet.o adlist.o redis-cli.o zmalloc.o release.o anet.o ae.o crc64.o
 REDIS_BENCHMARK_NAME=redis-benchmark
diff --git a/src/aof.c b/src/aof.c
index fc261db21..e52b7e214 100644
--- a/src/aof.c
+++ b/src/aof.c
@@ -1319,6 +1319,7 @@ int rewriteAppendOnlyFileBackground(void) {
 
     if (server.aof_child_pid != -1 || server.rdb_child_pid != -1) return C_ERR;
     if (aofCreatePipes() != C_OK) return C_ERR;
+    openChildInfoPipe();
     start = ustime();
     if ((childpid = fork()) == 0) {
         char tmpfile[256];
@@ -1335,6 +1336,9 @@ int rewriteAppendOnlyFileBackground(void) {
                     "AOF rewrite: %zu MB of memory used by copy-on-write",
                     private_dirty/(1024*1024));
             }
+
+            server.child_info_data.cow_size = private_dirty;
+            sendChildInfo(CHILD_INFO_TYPE_AOF);
             exitFromChild(0);
         } else {
             exitFromChild(1);
@@ -1345,6 +1349,7 @@ int rewriteAppendOnlyFileBackground(void) {
         server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */
         latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000);
         if (childpid == -1) {
+            closeChildInfoPipe();
             serverLog(LL_WARNING,
                 "Can't rewrite append only file in background: fork: %s",
                 strerror(errno));
diff --git a/src/childinfo.c b/src/childinfo.c
new file mode 100644
index 000000000..123c20421
--- /dev/null
+++ b/src/childinfo.c
@@ -0,0 +1,83 @@
+/*
+ * Copyright (c) 2016, Salvatore Sanfilippo <antirez at gmail dot com>
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *   * Redistributions of source code must retain the above copyright notice,
+ *     this list of conditions and the following disclaimer.
+ *   * Redistributions in binary form must reproduce the above copyright
+ *     notice, this list of conditions and the following disclaimer in the
+ *     documentation and/or other materials provided with the distribution.
+ *   * Neither the name of Redis nor the names of its contributors may be used
+ *     to endorse or promote products derived from this software without
+ *     specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#include "server.h"
+#include <unistd.h>
+
+/* Open a child-parent channel used in order to move information about the
+ * RDB / AOF saving process from the child to the parent (for instance
+ * the amount of copy on write memory used) */
+void openChildInfoPipe(void) {
+    if (pipe(server.child_info_pipe) == -1) {
+        /* On error our two file descriptors should be still set to -1,
+         * but we call anyway cloesChildInfoPipe() since can't hurt. */
+        closeChildInfoPipe();
+    } else if (anetNonBlock(NULL,server.child_info_pipe[0]) != ANET_OK) {
+        closeChildInfoPipe();
+    }
+}
+
+/* Close the pipes opened with openChildInfoPipe(). */
+void closeChildInfoPipe(void) {
+    if (server.child_info_pipe[0] != -1 ||
+        server.child_info_pipe[1] != -1)
+    {
+        close(server.child_info_pipe[0]);
+        close(server.child_info_pipe[1]);
+        server.child_info_pipe[0] = -1;
+        server.child_info_pipe[1] = -1;
+    }
+}
+
+/* Send COW data to parent. The child should call this function after populating
+ * the corresponding fields it want to sent (according to the process type). */
+void sendChildInfo(int ptype) {
+    if (server.child_info_pipe[1] == -1) return;
+    server.child_info_data.magic = CHILD_INFO_MAGIC;
+    server.child_info_data.process_type = ptype;
+    ssize_t wlen = sizeof(server.child_info_data);
+    if (write(server.child_info_pipe[1],&server.child_info_data,wlen) != wlen) {
+        /* Nothing to do on error, this will be detected by the other side. */
+    }
+}
+
+/* Receive COW data from parent. */
+void receiveChildInfo(void) {
+    if (server.child_info_pipe[0] == -1) return;
+    ssize_t wlen = sizeof(server.child_info_data);
+    if (read(server.child_info_pipe[0],&server.child_info_data,wlen) == wlen &&
+        server.child_info_data.magic == CHILD_INFO_MAGIC)
+    {
+        if (server.child_info_data.process_type == CHILD_INFO_TYPE_RDB) {
+            server.stat_rdb_cow_bytes = server.child_info_data.cow_size;
+        } else if (server.child_info_data.process_type == CHILD_INFO_TYPE_AOF) {
+            server.stat_aof_cow_bytes = server.child_info_data.cow_size;
+        }
+    }
+}
diff --git a/src/rdb.c b/src/rdb.c
index fd80fb3c9..5e03dbab5 100644
--- a/src/rdb.c
+++ b/src/rdb.c
@@ -1014,6 +1014,7 @@ int rdbSaveBackground(char *filename) {
 
     server.dirty_before_bgsave = server.dirty;
     server.lastbgsave_try = time(NULL);
+    openChildInfoPipe();
 
     start = ustime();
     if ((childpid = fork()) == 0) {
@@ -1031,6 +1032,9 @@ int rdbSaveBackground(char *filename) {
                     "RDB: %zu MB of memory used by copy-on-write",
                     private_dirty/(1024*1024));
             }
+
+            server.child_info_data.cow_size = private_dirty;
+            sendChildInfo(CHILD_INFO_TYPE_RDB);
         }
         exitFromChild((retval == C_OK) ? 0 : 1);
     } else {
@@ -1039,6 +1043,7 @@ int rdbSaveBackground(char *filename) {
         server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */
         latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000);
         if (childpid == -1) {
+            closeChildInfoPipe();
             server.lastbgsave_status = C_ERR;
             serverLog(LL_WARNING,"Can't save in background: fork: %s",
                 strerror(errno));
@@ -1744,6 +1749,7 @@ int rdbSaveToSlavesSockets(void) {
     }
 
     /* Create the child process. */
+    openChildInfoPipe();
     start = ustime();
     if ((childpid = fork()) == 0) {
         /* Child */
@@ -1769,6 +1775,9 @@ int rdbSaveToSlavesSockets(void) {
                     private_dirty/(1024*1024));
             }
 
+            server.child_info_data.cow_size = private_dirty;
+            sendChildInfo(CHILD_INFO_TYPE_RDB);
+
             /* If we are returning OK, at least one slave was served
              * with the RDB file as expected, so we need to send a report
              * to the parent via the pipe. The format of the message is:
@@ -1837,6 +1846,7 @@ int rdbSaveToSlavesSockets(void) {
             }
             close(pipefds[0]);
             close(pipefds[1]);
+            closeChildInfoPipe();
         } else {
             serverLog(LL_NOTICE,"Background RDB transfer started by pid %d",
                 childpid);
diff --git a/src/server.c b/src/server.c
index 36be973e1..77087b56d 100644
--- a/src/server.c
+++ b/src/server.c
@@ -1046,8 +1046,10 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
                     (int) server.aof_child_pid);
             } else if (pid == server.rdb_child_pid) {
                 backgroundSaveDoneHandler(exitcode,bysignal);
+                if (!bysignal && exitcode == 0) receiveChildInfo();
             } else if (pid == server.aof_child_pid) {
                 backgroundRewriteDoneHandler(exitcode,bysignal);
+                if (!bysignal && exitcode == 0) receiveChildInfo();
             } else {
                 if (!ldbRemoveChild(pid)) {
                     serverLog(LL_WARNING,
@@ -1056,6 +1058,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
                 }
             }
             updateDictResizePolicy();
+            closeChildInfoPipe();
         }
     } else {
         /* If there is not a background saving/rewrite in progress check if
@@ -1794,6 +1797,9 @@ void initServer(void) {
     server.aof_child_pid = -1;
     server.rdb_child_type = RDB_CHILD_TYPE_NONE;
     server.rdb_bgsave_scheduled = 0;
+    server.child_info_pipe[0] = -1;
+    server.child_info_pipe[1] = -1;
+    server.child_info_data.magic = 0;
     aofRewriteBufferReset();
     server.aof_buf = sdsempty();
     server.lastsave = time(NULL); /* At startup we consider the DB saved. */
@@ -1805,6 +1811,8 @@ void initServer(void) {
     /* A few stats we don't want to reset: server startup time, and peak mem. */
     server.stat_starttime = time(NULL);
     server.stat_peak_memory = 0;
+    server.stat_rdb_cow_bytes = 0;
+    server.stat_aof_cow_bytes = 0;
     server.resident_set_size = 0;
     server.lastbgsave_status = C_OK;
     server.aof_last_write_status = C_OK;
@@ -2889,13 +2897,15 @@ sds genRedisInfoString(char *section) {
             "rdb_last_bgsave_status:%s\r\n"
             "rdb_last_bgsave_time_sec:%jd\r\n"
             "rdb_current_bgsave_time_sec:%jd\r\n"
+            "rdb_last_cow_size:%zu\r\n"
             "aof_enabled:%d\r\n"
             "aof_rewrite_in_progress:%d\r\n"
             "aof_rewrite_scheduled:%d\r\n"
             "aof_last_rewrite_time_sec:%jd\r\n"
             "aof_current_rewrite_time_sec:%jd\r\n"
             "aof_last_bgrewrite_status:%s\r\n"
-            "aof_last_write_status:%s\r\n",
+            "aof_last_write_status:%s\r\n"
+            "aof_last_cow_size:%zu\r\n",
             server.loading,
             server.dirty,
             server.rdb_child_pid != -1,
@@ -2904,6 +2914,7 @@ sds genRedisInfoString(char *section) {
             (intmax_t)server.rdb_save_time_last,
             (intmax_t)((server.rdb_child_pid == -1) ?
                 -1 : time(NULL)-server.rdb_save_time_start),
+            server.stat_rdb_cow_bytes,
             server.aof_state != AOF_OFF,
             server.aof_child_pid != -1,
             server.aof_rewrite_scheduled,
@@ -2911,7 +2922,8 @@ sds genRedisInfoString(char *section) {
             (intmax_t)((server.aof_child_pid == -1) ?
                 -1 : time(NULL)-server.aof_rewrite_time_start),
             (server.aof_lastbgrewrite_status == C_OK) ? "ok" : "err",
-            (server.aof_last_write_status == C_OK) ? "ok" : "err");
+            (server.aof_last_write_status == C_OK) ? "ok" : "err",
+            server.stat_aof_cow_bytes);
 
         if (server.aof_state != AOF_OFF) {
             info = sdscatprintf(info,
diff --git a/src/server.h b/src/server.h
index 0af439cc5..8aad8f983 100644
--- a/src/server.h
+++ b/src/server.h
@@ -806,6 +806,10 @@ struct clusterState;
 #undef hz
 #endif
 
+#define CHILD_INFO_MAGIC 0xC17DDA7A12345678LL
+#define CHILD_INFO_TYPE_RDB 0
+#define CHILD_INFO_TYPE_AOF 1
+
 struct redisServer {
     /* General */
     pid_t pid;                  /* Main process pid. */
@@ -884,6 +888,8 @@ struct redisServer {
     size_t resident_set_size;       /* RSS sampled in serverCron(). */
     long long stat_net_input_bytes; /* Bytes read from network. */
     long long stat_net_output_bytes; /* Bytes written to network. */
+    size_t stat_rdb_cow_bytes;      /* Copy on write bytes during RDB saving. */
+    size_t stat_aof_cow_bytes;      /* Copy on write bytes during AOF rewrite. */
     /* The following two are used to track instantaneous metrics, like
      * number of operations per second, network traffic. */
     struct {
@@ -958,6 +964,13 @@ struct redisServer {
     int stop_writes_on_bgsave_err;  /* Don't allow writes if can't BGSAVE */
     int rdb_pipe_write_result_to_parent; /* RDB pipes used to return the state */
     int rdb_pipe_read_result_from_child; /* of each slave in diskless SYNC. */
+    /* Pipe and data structures for child -> parent info sharing. */
+    int child_info_pipe[2];         /* Pipe used to write the child_info_data. */
+    struct {
+        int process_type;           /* AOF or RDB child? */
+        size_t cow_size;            /* Copy on write size. */
+        unsigned long long magic;   /* Magic value to make sure data is valid. */
+    } child_info_data;
     /* Propagation of commands in AOF / replication */
     redisOpArray also_propagate;    /* Additional command to propagate. */
     /* Logging */
@@ -1411,6 +1424,12 @@ void aofRewriteBufferReset(void);
 unsigned long aofRewriteBufferSize(void);
 ssize_t aofReadDiffFromParent(void);
 
+/* Child info */
+void openChildInfoPipe(void);
+void closeChildInfoPipe(void);
+void sendChildInfo(int process_type);
+void receiveChildInfo(void);
+
 /* Sorted sets data type */
 
 /* Input flags. */