Implement netdata command server and cli tool (#7325)

* Checkpoint commit (POC)

* Implemented command server in the daemon

* Add netdatacli implementation

* Added prints in command server setup functions

* Make libuv version 1 a hard dependency for the agent

* Additional documentation

* Improved accuracy of names and documentation

* Fixed documentation

* Fixed buffer overflow

* Added support for exit status in cli. Added prefixes for exit code, stdout and stderr. Fixed parsers.

* Fix compilation errors

* Fix compile errors

* Fix compile errors

* Fix compile error

* Fix linker error for muslc

21 files changed, 940 insertions, 23 deletions

diff --git a/.gitignore b/.gitignore
index 0d8ba70f0f..31021152e9 100644
--- a/.gitignore
+++ b/.gitignore
@@ -37,6 +37,7 @@ sha256sums.txt
 
 # netdata binaries
 netdata
+netdatacli
 !netdata/
 upload/
 artifacts/
diff --git a/CMakeLists.txt b/CMakeLists.txt
index f3989b4c47..9eaa015838 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -628,6 +628,8 @@ set(DAEMON_FILES
         daemon/main.h
         daemon/signals.c
         daemon/signals.h
+        daemon/commands.c
+        daemon/commands.h
         daemon/unit_test.c
         daemon/unit_test.h
         )
@@ -648,6 +650,12 @@ set(NETDATA_FILES
         ${WEB_PLUGIN_FILES}
         )
 
+set(NETDATACLI_FILES
+        daemon/commands.h
+        cli/cli.c
+        cli/cli.h
+        )
+
 include_directories(AFTER .)
 
 add_definitions(
@@ -785,6 +793,15 @@ ENDIF()
 
 
 # -----------------------------------------------------------------------------
+# netdatacli
+
+add_executable(netdatacli config.h ${NETDATACLI_FILES})
+target_link_libraries (netdatacli libnetdata ${NETDATA_COMMON_LIBRARIES})
+target_include_directories(netdatacli PUBLIC ${NETDATA_COMMON_INCLUDE_DIRS})
+target_compile_options(netdatacli PUBLIC ${NETDATA_COMMON_CFLAGS})
+
+
+# -----------------------------------------------------------------------------
 # apps.plugin
 
 IF(ENABLE_PLUGIN_APPS)
diff --git a/Makefile.am b/Makefile.am
index 3ecdb14064..accf2ac691 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -482,6 +482,8 @@ DAEMON_FILES = \
     daemon/main.h \
     daemon/signals.c \
     daemon/signals.h \
+    daemon/commands.c \
+    daemon/commands.h \
     daemon/unit_test.c \
     daemon/unit_test.h \
     $(NULL)
@@ -537,6 +539,13 @@ NETDATA_COMMON_LIBS = \
     $(OPTIONAL_JSONC_LIBS) \
     $(NULL)
 
+NETDATACLI_FILES = \
+    daemon/commands.h \
+    $(LIBNETDATA_FILES) \
+    cli/cli.c \
+    cli/cli.h \
+    $(NULL)
+
 sbin_PROGRAMS += netdata
 netdata_SOURCES = $(NETDATA_FILES)
 netdata_LDADD = \
@@ -548,6 +557,17 @@ else
     netdata_LINK = $(CCLD) $(CFLAGS) $(LDFLAGS) -o $@
 endif
 
+sbin_PROGRAMS += netdatacli
+netdatacli_SOURCES = $(NETDATACLI_FILES)
+netdatacli_LDADD = \
+    $(NETDATA_COMMON_LIBS) \
+    $(NULL)
+if ENABLE_CXX_LINKER
+    netdatacli_LINK = $(CXXLD) $(CXXFLAGS) $(LDFLAGS) -o $@
+else
+    netdatacli_LINK = $(CCLD) $(CFLAGS) $(LDFLAGS) -o $@
+endif
+
 if ENABLE_PLUGIN_APPS
     plugins_PROGRAMS += apps.plugin
     apps_plugin_SOURCES = $(APPS_PLUGIN_FILES)
diff --git a/cli/Makefile.am b/cli/Makefile.am
new file mode 100644
index 0000000000..161784b8f6
--- /dev/null
+++ b/cli/Makefile.am
@@ -0,0 +1,8 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+AUTOMAKE_OPTIONS = subdir-objects
+MAINTAINERCLEANFILES = $(srcdir)/Makefile.in
+
+dist_noinst_DATA = \
+    README.md \
+    $(NULL)
diff --git a/cli/README.md b/cli/README.md
new file mode 100644
index 0000000000..3872dac1c1
--- /dev/null
+++ b/cli/README.md
@@ -0,0 +1,26 @@
+# Netdata cli
+
+You can see the commands netdatacli supports by executing it with `netdatacli` and entering `help` in
+standard input. All commands are given as standard input to `netdatacli`.
+
+The commands that a running netdata agent can execute are the following:
+
+```sh
+The commands are (arguments are in brackets):
+help
+    Show this help menu.
+reload-health
+    Reload health configuration.
+save-database
+    Save internal DB to disk for for memory mode save.
+reopen-logs
+    Close and reopen log files.
+shutdown-agent
+    Cleanup and exit the netdata agent.
+fatal-agent
+    Log the state and halt the netdata agent.
+```
+
+Those commands are the same that can be sent to netdata via [signals](../daemon#command-line-options).
+
+[![analytics](https://www.google-analytics.com/collect?v=1&aip=1&t=pageview&_s=1&ds=github&dr=https%3A%2F%2Fgithub.com%2Fnetdata%2Fnetdata&dl=https%3A%2F%2Fmy-netdata.io%2Fgithub%2Fcli%2FREADME&_u=MAC~&cid=5792dfd7-8dc4-476b-af31-da2fdb9f93d2&tid=UA-64295674-3)](<>)
diff --git a/cli/cli.c b/cli/cli.c
new file mode 100644
index 0000000000..4df0201787
--- /dev/null
+++ b/cli/cli.c
@@ -0,0 +1,201 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include "cli.h"
+#include "../libnetdata/required_dummies.h"
+
+static uv_pipe_t client_pipe;
+static uv_write_t write_req;
+static uv_shutdown_t shutdown_req;
+
+static char command_string[MAX_COMMAND_LENGTH];
+static unsigned command_string_size;
+
+static char response_string[MAX_COMMAND_LENGTH];
+static unsigned response_string_size;
+
+static int exit_status;
+
+struct command_context {
+    uv_work_t work;
+    uv_stream_t *client;
+    cmd_t idx;
+    char *args;
+    char *message;
+    cmd_status_t status;
+};
+
+static void parse_command_reply(void)
+{
+    FILE *stream = NULL;
+    char *pos;
+    int syntax_error = 0;
+
+    for (pos = response_string ;
+         pos < response_string + response_string_size  && !syntax_error ;
+         ++pos) {
+        /* Skip white-space characters */
+        for ( ; isspace(*pos) && ('\0' != *pos); ++pos) {;}
+
+        if ('\0' == *pos)
+            continue;
+
+        switch (*pos) {
+        case CMD_PREFIX_EXIT_CODE:
+            exit_status = atoi(++pos);
+            break;
+        case CMD_PREFIX_INFO:
+            stream = stdout;
+            break;
+        case CMD_PREFIX_ERROR:
+            stream = stderr;
+            break;
+        default:
+            syntax_error = 1;
+            fprintf(stderr, "Syntax error, failed to parse command response.\n");
+            break;
+        }
+        if (stream) {
+            fprintf(stream, "%s\n", ++pos);
+            pos += strlen(pos);
+            stream = NULL;
+        }
+    }
+}
+
+static void pipe_read_cb(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf)
+{
+    if (0 == nread) {
+        fprintf(stderr, "%s: Zero bytes read by command pipe.\n", __func__);
+    } else if (UV_EOF == nread) {
+//      fprintf(stderr, "EOF found in command pipe.\n");
+        parse_command_reply();
+    } else if (nread < 0) {
+        fprintf(stderr, "%s: %s\n", __func__, uv_strerror(nread));
+    }
+
+    if (nread < 0) { /* stop stream due to EOF or error */
+        (void)uv_read_stop((uv_stream_t *)client);
+    } else if (nread) {
+        size_t to_copy;
+
+        to_copy = MIN(nread, MAX_COMMAND_LENGTH - 1 - response_string_size);
+        memcpy(response_string + response_string_size, buf->base, to_copy);
+        response_string_size += to_copy;
+        response_string[response_string_size] = '\0';
+    }
+    if (buf && buf->len) {
+        free(buf->base);
+    }
+}
+
+static void alloc_cb(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf)
+{
+    (void)handle;
+
+    buf->base = malloc(suggested_size);
+    buf->len = suggested_size;
+}
+
+static void shutdown_cb(uv_shutdown_t* req, int status)
+{
+    int ret;
+
+    (void)req;
+    (void)status;
+
+    /* receive reply */
+    response_string_size = 0;
+    response_string[0] = '\0';
+
+    ret = uv_read_start((uv_stream_t *)&client_pipe, alloc_cb, pipe_read_cb);
+    if (ret) {
+        fprintf(stderr, "uv_read_start(): %s\n", uv_strerror(ret));
+        uv_close((uv_handle_t *)&client_pipe, NULL);
+        return;
+    }
+
+}
+
+static void pipe_write_cb(uv_write_t* req, int status)
+{
+    int ret;
+
+    (void)req;
+    (void)status;
+
+    ret = uv_shutdown(&shutdown_req, (uv_stream_t *)&client_pipe, shutdown_cb);
+    if (ret) {
+        fprintf(stderr, "uv_shutdown(): %s\n", uv_strerror(ret));
+        uv_close((uv_handle_t *)&client_pipe, NULL);
+        return;
+    }
+}
+
+static void connect_cb(uv_connect_t* req, int status)
+{
+    int ret;
+    uv_buf_t write_buf;
+    char *s;
+
+    (void)req;
+    if (status) {
+        fprintf(stderr, "uv_pipe_connect(): %s\n", uv_strerror(status));
+        fprintf(stderr, "Make sure the netdata service is running.\n");
+        exit(-1);
+    }
+    if (0 == command_string_size) {
+        s = fgets(command_string, MAX_COMMAND_LENGTH, stdin);
+    }
+    (void)s; /* We don't need input to communicate with the server */
+    command_string_size = strlen(command_string);
+
+    write_req.data = &client_pipe;
+    write_buf.base = command_string;
+    write_buf.len = command_string_size;
+    ret = uv_write(&write_req, (uv_stream_t *)&client_pipe, &write_buf, 1, pipe_write_cb);
+    if (ret) {
+        fprintf(stderr, "uv_write(): %s\n", uv_strerror(ret));
+    }
+//  fprintf(stderr, "COMMAND: Sending command: \"%s\"\n", command_string);
+}
+
+int main(int argc, char **argv)
+{
+    int ret, i;
+    static uv_loop_t* loop;
+    uv_connect_t req;
+
+    exit_status = -1; /* default status for when there is no command response from server */
+
+    loop = uv_default_loop();
+
+    ret = uv_pipe_init(loop, &client_pipe, 1);
+    if (ret) {
+        fprintf(stderr, "uv_pipe_init(): %s\n", uv_strerror(ret));
+        return exit_status;
+    }
+
+    command_string_size = 0;
+    command_string[0] = '\0';
+    for (i = 1 ; i < argc ; ++i) {
+        size_t to_copy;
+
+        to_copy = MIN(strlen(argv[i]), MAX_COMMAND_LENGTH - 1 - command_string_size);
+        strncpyz(command_string + command_string_size, argv[i], to_copy);
+        command_string_size += to_copy;
+
+        if (command_string_size < MAX_COMMAND_LENGTH - 1) {
+            command_string[command_string_size++] = ' ';
+        } else {
+            break;
+        }
+    }
+
+    uv_pipe_connect(&req, &client_pipe, PIPENAME, connect_cb);
+
+    uv_run(loop, UV_RUN_DEFAULT);
+
+    uv_close((uv_handle_t *)&client_pipe, NULL);
+
+    return exit_status;
+}
\ No newline at end of file
diff --git a/cli/cli.h b/cli/cli.h
new file mode 100644
index 0000000000..9e730a3019
--- /dev/null
+++ b/cli/cli.h
@@ -0,0 +1,8 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#ifndef NETDATA_CLI_H
+#define NETDATA_CLI_H 1
+
+#include "../daemon/common.h"
+
+#endif //NETDATA_CLI_H
diff --git a/configure.ac b/configure.ac
index 35147b0e92..4ae08bffc9 100644
--- a/configure.ac
+++ b/configure.ac
@@ -279,6 +279,10 @@ AC_CHECK_LIB(
     [uv_fs_scandir_next],
     [UV_LIBS="-luv"]
 )
+test -z "${UV_LIBS}" && \
+    AC_MSG_ERROR([libuv required but not found. Try installing 'libuv1-dev' or 'libuv-devel'.])
+OPTIONAL_UV_CFLAGS="${UV_CFLAGS}"
+OPTIONAL_UV_LIBS="${UV_LIBS}"
 
 
 # -----------------------------------------------------------------------------
@@ -363,9 +367,6 @@ OPTIONAL_JSONC_LIBS="${JSONC_LIBS}"
 
 # -----------------------------------------------------------------------------
 # DB engine and HTTPS
-test "${enable_dbengine}" = "yes" -a -z "${UV_LIBS}" && \
-    AC_MSG_ERROR([libuv required but not found. Try installing 'libuv1-dev' or 'libuv-devel'.])
-
 test "${enable_dbengine}" = "yes" -a -z "${LZ4_LIBS}" && \
     AC_MSG_ERROR([liblz4 required but not found. Try installing 'liblz4-dev' or 'lz4-devel'.])
 
@@ -382,8 +383,6 @@ AC_MSG_CHECKING([if netdata dbengine should be used])
 if test "${enable_dbengine}" != "no" -a "${UV_LIBS}" -a "${LZ4_LIBS}" -a "${JUDY_LIBS}" -a "${SSL_LIBS}"; then
     enable_dbengine="yes"
     AC_DEFINE([ENABLE_DBENGINE], [1], [netdata dbengine usability])
-    OPTIONAL_UV_CFLAGS="${UV_CFLAGS}"
-    OPTIONAL_UV_LIBS="${UV_LIBS}"
     OPTIONAL_LZ4_CFLAGS="${LZ4_CFLAGS}"
     OPTIONAL_LZ4_LIBS="${LZ4_LIBS}"
     OPTIONAL_JUDY_CFLAGS="${JUDY_CFLAGS}"
diff --git a/daemon/README.md b/daemon/README.md
index 0d4b0cdbbc..246586addd 100644
--- a/daemon/README.md
+++ b/daemon/README.md
@@ -185,6 +185,8 @@ The command line options of the Netdata 1.10.0 version are the following:
   - USR2                   Reload health configuration.
 ```
 
+You can send commands during runtime via [netdatacli](../cli).
+
 ## Log files
 
 Netdata uses 3 log files:
diff --git a/daemon/commands.c b/daemon/commands.c
new file mode 100644
index 0000000000..fc68ed59a7
--- /dev/null
+++ b/daemon/commands.c
@@ -0,0 +1,554 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include "common.h"
+#include "../database/engine/rrdenginelib.h"
+
+static uv_thread_t thread;
+static uv_loop_t* loop;
+static uv_async_t async;
+static struct completion completion;
+static uv_pipe_t server_pipe;
+
+char cmd_prefix_by_status[] = {
+        CMD_PREFIX_INFO,
+        CMD_PREFIX_ERROR,
+        CMD_PREFIX_ERROR
+};
+
+static int command_thread_error;
+static int command_thread_shutdown;
+static unsigned clients = 0;
+static char command_string[MAX_COMMAND_LENGTH];
+static unsigned command_string_size;
+
+struct command_context {
+    uv_work_t work;
+    uv_stream_t *client;
+    cmd_t idx;
+    char *args;
+    char *message;
+    cmd_status_t status;
+};
+
+/* Forward declarations */
+static cmd_status_t cmd_help_execute(char *args, char **message);
+static cmd_status_t cmd_reload_health_execute(char *args, char **message);
+static cmd_status_t cmd_save_database_execute(char *args, char **message);
+static cmd_status_t cmd_reopen_logs_execute(char *args, char **message);
+static cmd_status_t cmd_exit_execute(char *args, char **message);
+static cmd_status_t cmd_fatal_execute(char *args, char **message);
+
+static command_info_t command_info_array[] = {
+        {"help", cmd_help_execute, CMD_TYPE_HIGH_PRIORITY},                  // show help menu
+        {"reload-health", cmd_reload_health_execute, CMD_TYPE_ORTHOGONAL},   // reload health configuration
+        {"save-database", cmd_save_database_execute, CMD_TYPE_ORTHOGONAL},   // save database for memory mode save
+        {"reopen-logs", cmd_reopen_logs_execute, CMD_TYPE_ORTHOGONAL},       // Close and reopen log files
+        {"shutdown-agent", cmd_exit_execute, CMD_TYPE_EXCLUSIVE},            // exit cleanly
+        {"fatal-agent", cmd_fatal_execute, CMD_TYPE_HIGH_PRIORITY},          // exit with fatal error
+};
+
+/* Mutexes for commands of type CMD_TYPE_ORTHOGONAL */
+static uv_mutex_t command_lock_array[CMD_TOTAL_COMMANDS];
+/* Commands of type CMD_TYPE_EXCLUSIVE are writers */
+static uv_rwlock_t exclusive_rwlock;
+/*
+ * Locking order:
+ * 1. exclusive_rwlock
+ * 2. command_lock_array[]
+ */
+
+/* Forward declarations */
+static void cmd_lock_exclusive(unsigned index);
+static void cmd_lock_orthogonal(unsigned index);
+static void cmd_lock_idempotent(unsigned index);
+static void cmd_lock_high_priority(unsigned index);
+
+static command_lock_t *cmd_lock_by_type[] = {
+        cmd_lock_exclusive,
+        cmd_lock_orthogonal,
+        cmd_lock_idempotent,
+        cmd_lock_high_priority
+};
+
+/* Forward declarations */
+static void cmd_unlock_exclusive(unsigned index);
+static void cmd_unlock_orthogonal(unsigned index);
+static void cmd_unlock_idempotent(unsigned index);
+static void cmd_unlock_high_priority(unsigned index);
+
+static command_lock_t *cmd_unlock_by_type[] = {
+        cmd_unlock_exclusive,
+        cmd_unlock_orthogonal,
+        cmd_unlock_idempotent,
+        cmd_unlock_high_priority
+};
+
+static cmd_status_t cmd_help_execute(char *args, char **message)
+{
+    (void)args;
+
+    *message = mallocz(MAX_COMMAND_LENGTH);
+    strncpyz(*message,
+             "\nThe commands are (arguments are in brackets):\n"
+             "help\n"
+             "    Show this help menu.\n"
+             "reload-health\n"
+             "    Reload health configuration.\n"
+             "save-database\n"
+             "    Save internal DB to disk for memory mode save.\n"
+             "reopen-logs\n"
+             "    Close and reopen log files.\n"
+             "shutdown-agent\n"
+             "    Cleanup and exit the netdata agent.\n"
+             "fatal-agent\n"
+             "    Log the state and halt the netdata agent.\n",
+             MAX_COMMAND_LENGTH - 1);
+    return CMD_STATUS_SUCCESS;
+}
+
+static cmd_status_t cmd_reload_health_execute(char *args, char **message)
+{
+    (void)args;
+    (void)message;
+
+    error_log_limit_unlimited();
+    info("COMMAND: Reloading HEALTH configuration.");
+    health_reload();
+    error_log_limit_reset();
+
+    return CMD_STATUS_SUCCESS;
+}
+
+static cmd_status_t cmd_save_database_execute(char *args, char **message)
+{
+    (void)args;
+    (void)message;
+
+    error_log_limit_unlimited();
+    info("COMMAND: Saving databases.");
+    rrdhost_save_all();
+    info("COMMAND: Databases saved.");
+    error_log_limit_reset();
+
+    return CMD_STATUS_SUCCESS;
+}
+
+static cmd_status_t cmd_reopen_logs_execute(char *args, char **message)
+{
+    (void)args;
+    (void)message;
+
+    error_log_limit_unlimited();
+    info("COMMAND: Reopening all log files.");
+    reopen_all_log_files();
+    error_log_limit_reset();
+
+    return CMD_STATUS_SUCCESS;
+}
+
+static cmd_status_t cmd_exit_execute(char *args, char **message)
+{
+    (void)args;
+    (void)message;
+
+    error_log_limit_unlimited();
+    info("COMMAND: Cleaning up to exit.");
+    netdata_cleanup_and_exit(0);
+    exit(0);
+
+    return CMD_STATUS_SUCCESS;
+}
+
+static cmd_status_t cmd_fatal_execute(char *args, char **message)
+{
+    (void)args;
+    (void)message;
+
+    fatal("COMMAND: netdata now exits.");
+
+    return CMD_STATUS_SUCCESS;
+}
+
+static void cmd_lock_exclusive(unsigned index)
+{
+    (void)index;
+
+    uv_rwlock_wrlock(&exclusive_rwlock);
+}
+
+static void cmd_lock_orthogonal(unsigned index)
+{
+    uv_rwlock_rdlock(&exclusive_rwlock);
+    uv_mutex_lock(&command_lock_array[index]);
+}
+
+static void cmd_lock_idempotent(unsigned index)
+{
+    (void)index;
+
+    uv_rwlock_rdlock(&exclusive_rwlock);
+}
+
+static void cmd_lock_high_priority(unsigned index)
+{
+    (void)index;
+}
+
+static void cmd_unlock_exclusive(unsigned index)
+{
+    (void)index;
+
+    uv_rwlock_wrunlock(&exclusive_rwlock);
+}
+
+static void cmd_unlock_orthogonal(unsigned index)
+{
+    uv_rwlock_rdunlock(&exclusive_rwlock);
+    uv_mutex_unlock(&command_lock_array[index]);
+}
+
+static void cmd_unlock_idempotent(unsigned index)
+{
+    (void)index;
+
+    uv_rwlock_rdunlock(&exclusive_rwlock);
+}
+
+static void cmd_unlock_high_priority(unsigned index)
+{
+    (void)index;
+}
+
+static void pipe_write_cb(uv_write_t* req, int status)
+{
+    (void)status;
+    uv_pipe_t *client = req->data;
+
+    uv_close((uv_handle_t *)client, NULL);
+    freez(client);
+    --clients;
+    info("Command Clients = %u\n", clients);
+}
+
+static inline void add_char_to_command_reply(char *reply_string, unsigned *reply_string_size, char character)
+{
+    reply_string[(*reply_string_size)++] = character;
+}
+
+static inline void add_string_to_command_reply(char *reply_string, unsigned *reply_string_size, char *str)
+{
+    unsigned len;
+
+    len = strlen(str);
+    strncpyz(reply_string + *reply_string_size, str, len);
+    *reply_string_size += len;
+}
+
+static void send_command_reply(uv_stream_t *client, cmd_status_t status, char *message)
+{
+    int ret;
+    char reply_string[MAX_COMMAND_LENGTH] = {'\0', };
+    char exit_status_string[MAX_EXIT_STATUS_LENGTH + 1] = {'\0', };
+    unsigned reply_string_size = 0;
+    uv_buf_t write_buf;
+    uv_write_t write_req;
+
+    snprintfz(exit_status_string, MAX_EXIT_STATUS_LENGTH, "%u", status);
+    add_char_to_command_reply(reply_string, &reply_string_size, CMD_PREFIX_EXIT_CODE);
+    add_string_to_command_reply(reply_string, &reply_string_size, exit_status_string);
+    add_char_to_command_reply(reply_string, &reply_string_size, '\0');
+
+    if (message) {
+        add_char_to_command_reply(reply_string, &reply_string_size, cmd_prefix_by_status[status]);
+        add_string_to_command_reply(reply_string, &reply_string_size, message);
+    }
+
+    write_req.data = client;
+    write_buf.base = reply_string;
+    write_buf.len = reply_string_size;
+    ret = uv_write(&write_req, (uv_stream_t *)client, &write_buf, 1, pipe_write_cb);
+    if (ret) {
+        error("uv_write(): %s", uv_strerror(ret));
+    }
+    info("COMMAND: Sending reply: \"%s\"", reply_string);
+}
+
+cmd_status_t execute_command(cmd_t idx, char *args, char **message)
+{
+    cmd_status_t status;
+    cmd_type_t type = command_info_array[idx].type;
+
+    cmd_lock_by_type[type](idx);
+    status = command_info_array[idx].func(args, message);
+    cmd_unlock_by_type[type](idx);
+
+    return status;
+}
+
+static void after_schedule_command(uv_work_t *req, int status)
+{
+    struct command_context *cmd_ctx = req->data;
+
+    (void)status;
+
+    send_command_reply(cmd_ctx->client, cmd_ctx->status, cmd_ctx->message);
+    if (cmd_ctx->message)
+        freez(cmd_ctx->message);
+}
+
+static void schedule_command(uv_work_t *req)
+{
+    struct command_context *cmd_ctx = req->data;
+
+    cmd_ctx->status = execute_command(cmd_ctx->idx, cmd_ctx->args, &cmd_ctx->message);
+}
+
+static void parse_commands(uv_stream_t *client)
+{
+    char *message = NULL, *pos;
+    cmd_t i;
+    cmd_status_t status;
+    struct command_context *cmd_ctx;
+
+    status = CMD_STATUS_FAILURE;
+
+    /* Skip white-space characters */
+    for (pos = command_string ; isspace(*pos) && ('\0' != *pos) ; ++pos) {;}
+    for (i = 0 ; i < CMD_TOTAL_COMMANDS ; ++i) {
+        if (!strncmp(pos, command_info_array[i].cmd_str, strlen(command_info_array[i].cmd_str))) {
+            cmd_ctx = mallocz(sizeof(*cmd_ctx));
+            cmd_ctx->work.data = cmd_ctx;
+            cmd_ctx->client = client;
+            cmd_ctx->idx = i;
+            cmd_ctx->args = pos + strlen(command_info_array[i].cmd_str);
+            cmd_ctx->message = NULL;
+
+            assert(0 == uv_queue_work(loop, &cmd_ctx->work, schedule_command, after_schedule_command));
+            break;
+        }
+    }
+    if (CMD_TOTAL_COMMANDS == i) {
+        /* no command found */
+        message = strdupz("Illegal command. Please type \"help\" for instructions.");
+        send_command_reply(client, status, message);
+        freez(message);
+    }
+}
+
+static void pipe_read_cb(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf)
+{
+    if (0 == nread) {
+        info("%s: Zero bytes read by command pipe.", __func__);
+    } else if (UV_EOF == nread) {
+        info("EOF found in command pipe.");
+        parse_commands(client);
+    } else if (nread < 0) {
+        error("%s: %s", __func__, uv_strerror(nread));
+    }
+
+    if (nread < 0) { /* stop stream due to EOF or error */
+        (void)uv_read_stop((uv_stream_t *)client);
+    } else if (nread) {
+        size_t to_copy;
+
+        to_copy = MIN(nread, MAX_COMMAND_LENGTH - 1 - command_string_size);
+        strncpyz(command_string + command_string_size, buf->base, to_copy);
+        command_string_size += to_copy;
+    }
+    if (buf && buf->len) {
+        freez(buf->base);
+    }
+
+    if (nread < 0 && UV_EOF != nread) {
+        uv_close((uv_handle_t *)client, NULL);
+        freez(client);
+        --clients;
+        info("Command Clients = %u\n", clients);
+    }
+}
+
+static void alloc_cb(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf)
+{
+    (void)handle;
+
+    buf->base = mallocz(suggested_size);
+    buf->len = suggested_size;
+}
+
+static void connection_cb(uv_stream_t *server, int status)
+{
+    int ret;
+    uv_pipe_t *client;
+    assert(status == 0);
+
+    client = mallocz(sizeof(*client));
+    ret = uv_pipe_init(server->loop, client, 1);
+    if (ret) {
+        error("uv_pipe_init(): %s", uv_strerror(ret));
+        freez(client);
+        return;
+    }
+    ret = uv_accept(server, (uv_stream_t *)client);
+    if (ret) {
+        error("uv_accept(): %s", uv_strerror(ret));
+        uv_close((uv_handle_t *)client, NULL);
+        freez(client);
+        return;
+    }
+
+    ++clients;
+    info("Command Clients = %u\n", clients);
+    /* Start parsing a new command */
+    command_string_size = 0;
+    command_string[0] = '\0';
+
+    ret = uv_read_start((uv_stream_t*)client, alloc_cb, pipe_read_cb);
+    if (ret) {
+        error("uv_read_start(): %s", uv_strerror(ret));
+        uv_close((uv_handle_t *)client, NULL);
+        freez(client);
+        --clients;
+        info("Command Clients = %u\n", clients);
+        return;
+    }
+}
+
+static void async_cb(uv_async_t *handle)
+{
+    uv_stop(handle->loop);
+}
+
+static void command_thread(void *arg)
+{
+    int ret;
+    uv_fs_t req;
+
+    (void) arg;
+    loop = mallocz(sizeof(uv_loop_t));
+    ret = uv_loop_init(loop);
+    if (ret) {
+        error("uv_loop_init(): %s", uv_strerror(ret));
+        command_thread_error = ret;
+        goto error_after_loop_init;
+    }
+    loop->data = NULL;
+
+    ret = uv_async_init(loop, &async, async_cb);
+    if (ret) {
+        error("uv_async_init(): %s", uv_strerror(ret));
+        command_thread_error = ret;
+        goto error_after_async_init;
+    }
+    async.data = NULL;
+
+    ret = uv_pipe_init(loop, &server_pipe, 1);
+    if (ret) {
+        error("uv_pipe_init(): %s", uv_strerror(ret));
+        command_thread_error = ret;
+        goto error_after_pipe_init;
+    }
+    (void)uv_fs_unlink(loop, &req, PIPENAME, NULL);
+    uv_fs_req_cleanup(&req);
+    ret = uv_pipe_bind(&server_pipe, PIPENAME);
+    if (ret) {
+        error("uv_pipe_bind(): %s", uv_strerror(ret));
+        command_thread_error = ret;
+        goto error_after_pipe_bind;
+    }
+    if ((ret = uv_listen((uv_stream_t *)&server_pipe, SOMAXCONN, connection_cb))) {
+        error("uv_listen(): %s", uv_strerror(ret));
+        command_thread_error = ret;
+        goto error_after_uv_listen;
+    }
+
+    command_thread_error = 0;
+    command_thread_shutdown = 0;
+    /* wake up initialization thread */
+    complete(&completion);
+
+    while (command_thread_shutdown == 0) {
+        uv_run(loop, UV_RUN_DEFAULT);
+    }
+    /* cleanup operations of the event loop */
+    info("Shutting down command event loop.");
+    uv_close((uv_handle_t *)&async, NULL);
+    uv_close((uv_handle_t*)&server_pipe, NULL);
+    uv_run(loop, UV_RUN_DEFAULT);
+
+    info("Shutting down command loop complete.");
+    assert(0 == uv_loop_close(loop));
+    freez(loop);
+
+    return;
+
+error_after_uv_listen:
+error_after_pipe_bind:
+    uv_close((uv_handle_t*)&server_pipe, NULL);
+error_after_pipe_init:
+    uv_close((uv_handle_t *)&async, NULL);
+error_after_async_init:
+    assert(0 == uv_loop_close(loop));
+error_after_loop_init:
+    freez(loop);
+
+    /* wake up initialization thread */
+    complete(&completion);
+}
+
+static void sanity_check(void)
+{
+    /* The size of command_info_array must be CMD_TOTAL_COMMANDS elements */
+    BUILD_BUG_ON(CMD_TOTAL_COMMANDS != sizeof(command_info_array) / sizeof(command_info_array[0]));
+}
+
+void commands_init(void)
+{
+    cmd_t i;
+    int error;
+
+    sanity_check();
+    info("Initializing command server.");
+    for (i = 0 ; i < CMD_TOTAL_COMMANDS ; ++i) {
+        uv_mutex_init(&command_lock_array[i]);
+    }
+    assert(0 == uv_rwlock_init(&exclusive_rwlock));
+
+    init_completion(&completion);
+    error = u