MXS-1220: Create separate admin thread

When MaxScale is started, a separate thread for the administrative
interface is started. This allows the worker threads to handle client
requests while the administrative thread handles the lower priority
administrative requests.

The administrative interface responds to all request with a 200 OK HTTP
response. This allows the administrative interface itself to be tested.

server/core/CMakeLists.txt

@@ -1,4 +1,6 @@
 add_library(maxscale-common SHARED
+  admin.cc
+  adminclient.cc
   adminusers.cc
   alloc.cc
   atomic.cc

server/core/admin.cc

@@ -0,0 +1,157 @@
+/*
+ * Copyright (c) 2016 MariaDB Corporation Ab
+ *
+ * Use of this software is governed by the Business Source License included
+ * in the LICENSE.TXT file and at www.mariadb.com/bsl11.
+ *
+ * Change Date: 2019-07-01
+ *
+ * On the date above, in accordance with the Business Source License, use
+ * of this software will be governed by version 2 or later of the General
+ * Public License.
+ */
+
+#include "maxscale/admin.hh"
+
+#include <climits>
+#include <new>
+
+#include <maxscale/atomic.h>
+#include <maxscale/debug.h>
+#include <maxscale/thread.h>
+#include <maxscale/utils.h>
+
+#define DEFAULT_ADMIN_HOST "127.0.0.1"
+#define DEFAULT_ADMIN_PORT 8080
+
+static AdminListener* admin = NULL;
+static THREAD admin_thread;
+
+// TODO: Read values from the configuration
+static AdminConfig config = {DEFAULT_ADMIN_HOST, DEFAULT_ADMIN_PORT};
+
+void admin_main(void* data)
+{
+    AdminListener* admin = reinterpret_cast<AdminListener*>(data);
+    admin->start();
+}
+
+AdminConfig& mxs_admin_get_config()
+{
+    return config;
+}
+
+bool mxs_admin_init()
+{
+    ss_dassert(admin == NULL);
+    bool rval = false;
+    struct sockaddr_storage addr = {};
+    int sock = open_network_socket(MXS_SOCKET_LISTENER, &addr, config.host.c_str(), config.port);
+
+    if (sock > -1)
+    {
+        if (listen(sock, INT_MAX) == 0)
+        {
+            admin = new (std::nothrow) AdminListener(sock);
+
+            if (admin)
+            {
+                if (thread_start(&admin_thread, admin_main, admin))
+                {
+                    rval = true;
+                }
+                else
+                {
+                    delete admin;
+                    admin = NULL;
+                }
+            }
+            else
+            {
+                MXS_OOM();
+            }
+        }
+        else
+        {
+            MXS_ERROR("Failed to start listening on '[%s]:%u': %d, %s",
+                      config.host.c_str(), config.port,  errno, mxs_strerror(errno));
+            close(sock);
+        }
+    }
+
+    return rval;
+}
+
+void mxs_admin_shutdown()
+{
+    if (admin)
+    {
+        admin->stop();
+        thread_wait(admin_thread);
+        delete admin;
+        admin = NULL;
+    }
+}
+
+AdminListener::AdminListener(int sock):
+    m_socket(sock),
+    m_active(0),
+    m_timeout(10)
+{
+}
+
+AdminListener::~AdminListener()
+{
+    close(m_socket);
+}
+
+void AdminListener::handle_clients()
+{
+    AdminClient* client = accept_client();
+
+    if (client)
+    {
+        client->process();
+        delete client;
+    }
+}
+
+void AdminListener::start()
+{
+    atomic_write(&m_active, 1);
+
+    while (atomic_read(&m_active))
+    {
+        MXS_EXCEPTION_GUARD(handle_clients());
+    }
+}
+
+void AdminListener::stop()
+{
+    atomic_write(&m_active, 0);
+}
+
+AdminClient* AdminListener::accept_client()
+{
+    AdminClient* rval = NULL;
+    struct sockaddr_storage addr = {};
+    socklen_t len = sizeof (addr);
+    int fd = accept(m_socket, (struct sockaddr*) &addr, &len);
+
+    if (fd > -1)
+    {
+        setnonblocking(fd);
+        rval = new AdminClient(fd, addr, m_timeout);
+    }
+    else if (errno == EAGAIN || errno == EWOULDBLOCK)
+    {
+        // TODO: Use epoll for this
+        thread_millisleep(1);
+    }
+    else
+    {
+        MXS_ERROR("Failed to accept client: %d, %s\n", errno, mxs_strerror(errno));
+    }
+
+    return rval;
+}

server/core/adminclient.cc

@@ -0,0 +1,104 @@
+/*
+ * Copyright (c) 2016 MariaDB Corporation Ab
+ *
+ * Use of this software is governed by the Business Source License included
+ * in the LICENSE.TXT file and at www.mariadb.com/bsl11.
+ *
+ * Change Date: 2019-07-01
+ *
+ * On the date above, in accordance with the Business Source License, use
+ * of this software will be governed by version 2 or later of the General
+ * Public License.
+ */
+
+#include "maxscale/adminclient.hh"
+
+#include <string>
+
+#include <maxscale/atomic.h>
+#include <maxscale/hk_heartbeat.h>
+#include <maxscale/log_manager.h>
+
+using std::string;
+
+AdminClient::AdminClient(int fd, const struct sockaddr_storage& addr, int timeout):
+    m_fd(fd),
+    m_timeout(timeout),
+    m_addr(addr)
+{
+
+}
+
+AdminClient::~AdminClient()
+{
+    close(m_fd);
+}
+
+static bool read_request_header(int fd, int timeout, string& output)
+{
+    int64_t start = atomic_read_int64(&hkheartbeat);
+
+    while ((atomic_read_int64(&hkheartbeat) - start) / 10 < timeout)
+    {
+        char buf[1024];
+        int rc = read(fd, buf, sizeof(buf));
+
+        if (rc == -1 && errno != EAGAIN && errno != EWOULDBLOCK)
+        {
+            return false;
+        }
+        else if (rc > 0)
+        {
+            buf[rc] = '\0';
+            output += buf;
+
+            if (output.find("\r\n\r\n") != std::string::npos)
+            {
+                break;
+            }
+        }
+    }
+
+    return true;
+}
+
+static bool write_response(int fd, int timeout, string input)
+{
+    int64_t start = atomic_read_int64(&hkheartbeat);
+
+    while ((atomic_read_int64(&hkheartbeat) - start) / 10 < timeout && input.length() > 0)
+    {
+        int rc = write(fd, input.c_str(), input.length());
+
+        if (rc == -1 && errno != EAGAIN && errno != EWOULDBLOCK)
+        {
+            return false;
+        }
+        else if (rc > 0)
+        {
+            input.erase(0, rc);
+        }
+    }
+
+    return true;
+
+}
+
+void AdminClient::process()
+{
+    string request;
+
+    if (read_request_header(m_fd, m_timeout, request))
+    {
+        /** Send the Status-Line part of the response */
+        string response = "HTTP/1.1 200 OK\r\n";
+
+        response += "\r\n";
+
+        write_response(m_fd, m_timeout, response);
+    }
+    else
+    {
+        MXS_ERROR("Failed to read client request: %d, %s", errno, mxs_strerror(errno));
+    }
+}

server/core/gateway.cc

@@ -57,6 +57,7 @@
 #include "maxscale/poll.h"
 #include "maxscale/service.h"
 #include "maxscale/statistics.h"
+#include "maxscale/admin.hh"
 #include "maxscale/worker.hh"
 
 using namespace maxscale;
@@ -1991,6 +1992,14 @@ int main(int argc, char **argv)
         write_child_exit_code(daemon_pipe[1], rc);
     }
 
+    if (!mxs_admin_init())
+    {
+        const char* logerr = "Failed to initialize admin interface";
+        print_log_n_stderr(true, true, logerr, logerr, 0);
+        rc = MAXSCALE_INTERNALERROR;
+        goto return_main;
+    }
+
     /*<
      * Run worker 0 in the main thread.
      */
@@ -1998,6 +2007,9 @@ int main(int argc, char **argv)
     ss_dassert(worker);
     worker->run();
 
+    /** Stop administrative interface */
+    mxs_admin_shutdown();
+
     /*<
      * Wait for the housekeeper to finish.
      */

server/core/housekeeper.cc

@@ -267,7 +267,7 @@ hkthread(void *data)
         for (i = 0; i < 10; i++)
         {
             thread_millisleep(100);
-            hkheartbeat++;
+            atomic_add_int64(&hkheartbeat, 1);
         }
         now = time(0);
         spinlock_acquire(&tasklock);

server/core/maxscale/admin.hh

@@ -0,0 +1,79 @@
+#pragma once
+/*
+ * Copyright (c) 2016 MariaDB Corporation Ab
+ *
+ * Use of this software is governed by the Business Source License included
+ * in the LICENSE.TXT file and at www.mariadb.com/bsl11.
+ *
+ * Change Date: 2019-07-01
+ *
+ * On the date above, in accordance with the Business Source License, use
+ * of this software will be governed by version 2 or later of the General
+ * Public License.
+ */
+
+#include <maxscale/cppdefs.hh>
+
+#include <string>
+
+#include <maxscale/thread.h>
+
+#include "adminclient.hh"
+
+using std::string;
+
+/** The admin interface configuration */
+struct AdminConfig
+{
+    string   host;
+    uint16_t port;
+};
+
+class AdminListener
+{
+public:
+    /**
+     * @brief Create a new admin interface instance
+     *
+     * @param sock Listener socket for the interface
+     */
+    AdminListener(int sock);
+    ~AdminListener();
+
+    /**
+     * Start the admin interface
+     */
+    void start();
+
+    /**
+     * Stop the admin listener
+     */
+    void stop();
+
+private:
+    void         handle_clients();
+    AdminClient* accept_client();
+
+    int    m_socket;  /**< The network socket we listen on */
+    int    m_active;  /**< Positive value if the admin is active */
+    int    m_timeout; /**< Network timeout in seconds */
+};
+
+/**
+ * @brief Start the administrative interface
+ *
+ * @return True if the interface was successfully started
+ */
+bool mxs_admin_init();
+
+/**
+ * @brief Shutdown the administrative interface
+ */
+void mxs_admin_shutdown();
+
+/**
+ * @brief Get the administative interface configuration
+ *
+ * @return A reference to the administrative interface configuration
+ */
+AdminConfig& mxs_admin_get_config();

server/core/maxscale/adminclient.hh

@@ -0,0 +1,44 @@
+#pragma once
+
+/*
+ * Copyright (c) 2016 MariaDB Corporation Ab
+ *
+ * Use of this software is governed by the Business Source License included
+ * in the LICENSE.TXT file and at www.mariadb.com/bsl11.
+ *
+ * Change Date: 2019-07-01
+ *
+ * On the date above, in accordance with the Business Source License, use
+ * of this software will be governed by version 2 or later of the General
+ * Public License.
+ */
+
+#include <maxscale/cppdefs.hh>
+
+#include <map>
+#include <sys/socket.h>
+
+class AdminClient
+{
+public:
+    /**
+     * @brief Create a new client connection
+     *
+     * @param fd      Client socket
+     * @param addr    Network address where @c fd is connected to
+     * @param timeout Network timeout for reads and writes
+     */
+    AdminClient(int fd, const struct sockaddr_storage& addr, int timeout);
+
+    ~AdminClient();
+
+    /**
+     * Process one request
+     */
+    void process();
+
+private:
+    int m_fd;                       /**< The client socket */
+    int m_timeout;                  /**< Network timeout for reads and writes */
+    struct sockaddr_storage m_addr; /**< Network info for the client */
+};