path: root/fluent-bit/lib/monkey/mk_server/mk_server.c

/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */

/*  Monkey HTTP Server
 *  ==================
 *  Copyright 2001-2017 Eduardo Silva <eduardo@monkey.io>
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

#include <monkey/mk_info.h>
#include <monkey/monkey.h>
#include <monkey/mk_config.h>
#include <monkey/mk_scheduler.h>
#include <monkey/mk_plugin.h>
#include <monkey/mk_utils.h>
#include <monkey/mk_server.h>
#include <monkey/mk_server_tls.h>
#include <monkey/mk_scheduler.h>
#include <monkey/mk_core.h>
#include <monkey/mk_fifo.h>
#include <monkey/mk_http_thread.h>

#ifdef _WIN32
#include <winsock2.h>
#else
#include <sys/socket.h>
#include <netinet/in.h>
#endif

#ifndef _WIN32
#include <sys/time.h>
#include <sys/resource.h>
#endif

pthread_key_t mk_server_fifo_key;

static int mk_server_lib_notify_event_loop_break(struct mk_sched_worker *sched);

/* Return the number of clients that can be attended  */
unsigned int mk_server_capacity(struct mk_server *server)
{
    int ret;
    int cur;

#ifndef _WIN32
    struct rlimit lim;

    /* Limit by system */
    getrlimit(RLIMIT_NOFILE, &lim);
    cur = lim.rlim_cur;

    if (server->fd_limit > cur) {
        lim.rlim_cur = server->fd_limit;
        lim.rlim_max = server->fd_limit;

        ret = setrlimit(RLIMIT_NOFILE, &lim);
        if (ret == -1) {
            mk_warn("Could not increase FDLimit to %i.", server->fd_limit);
        }
        else {
            cur = server->fd_limit;
        }
    }
    else if (server->fd_limit > 0) {
        cur = server->fd_limit;
    }

#else
    ret = 0;
    cur = INT_MAX; 

    /* This is not the right way to plug this, according to raymond chen the only limit
     * to fd count is free memory in their winsock provider and there are no other limits
     * that I know of but I should still look for a more elegant solution. (even if it
     * was just ignoring the server_capacity limit in scheduler.c: _next_target)
    */
#endif

    return cur;
}

static inline
struct mk_sched_conn *mk_server_listen_handler(struct mk_sched_worker *sched,
                                               void *data,
                                               struct mk_server *server)
{
    int ret;
    int client_fd = -1;
    struct mk_sched_conn *conn;
    struct mk_server_listen *listener = data;

    client_fd = mk_socket_accept(listener->server_fd);
    if (mk_unlikely(client_fd == -1)) {
        MK_TRACE("[server] Accept connection failed: %s", strerror(errno));
        goto error;
    }

    conn = mk_sched_add_connection(client_fd, listener, sched, server);
    if (mk_unlikely(!conn)) {
        goto error;
    }

    ret = mk_event_add(sched->loop, client_fd,
                       MK_EVENT_CONNECTION, MK_EVENT_READ, conn);
    if (mk_unlikely(ret != 0)) {
        mk_err("[server] Error registering file descriptor: %s",
               strerror(errno));
        goto error;
    }

    sched->accepted_connections++;
    MK_TRACE("[server] New connection arrived: FD %i", client_fd);
    return conn;

error:
    if (client_fd != -1) {
        listener->network->network->close(listener->network, client_fd);
    }

    return NULL;
}

void mk_server_listen_free()
{
    struct mk_list *list;
    struct mk_list *tmp;
    struct mk_list *head;
    struct mk_server_listen *listener;

    list = MK_TLS_GET(mk_tls_server_listen);
    mk_list_foreach_safe(head, tmp, list) {
        listener = mk_list_entry(head, struct mk_server_listen, _head);
        mk_list_del(&listener->_head);
        mk_mem_free(listener);
    }
}

void mk_server_listen_exit(struct mk_list *list)
{
    struct mk_list *tmp;
    struct mk_list *head;
    struct mk_server_listen *listen;

    if (!list) {
        return;
    }

    mk_list_foreach_safe(head, tmp, list) {
        listen = mk_list_entry(head, struct mk_server_listen, _head);
        mk_event_closesocket(listen->server_fd);
        mk_list_del(&listen->_head);
        mk_mem_free(listen);
    }

    mk_mem_free(list);
}

struct mk_list *mk_server_listen_init(struct mk_server *server)
{
    int server_fd;
    int reuse_port = MK_FALSE;
    struct mk_list *head;
    struct mk_list *listeners;
    struct mk_event *event;
    struct mk_server_listen *listener;
    struct mk_sched_handler *protocol;
    struct mk_plugin *plugin;
    struct mk_config_listener *listen;

    if (!server) {
        goto error;
    }

    listeners = mk_mem_alloc(sizeof(struct mk_list));
    mk_list_init(listeners);

    if (server->scheduler_mode == MK_SCHEDULER_REUSEPORT) {
        reuse_port = MK_TRUE;
    }

    mk_list_foreach(head, &server->listeners) {
        listen = mk_list_entry(head, struct mk_config_listener, _head);

        server_fd = mk_socket_server(listen->port,
                                     listen->address,
                                     reuse_port,
                                     server);
        if (server_fd >= 0) {
            if (mk_socket_set_tcp_defer_accept(server_fd) != 0) {
#if defined (__linux__)
                mk_warn("[server] Could not set TCP_DEFER_ACCEPT");
#endif
            }

            listener = mk_mem_alloc_z(sizeof(struct mk_server_listen));

            /* configure the internal event_state */
            event = &listener->event;
            event->fd   = server_fd;
            event->type = MK_EVENT_LISTENER;
            event->mask = MK_EVENT_EMPTY;
            event->status = MK_EVENT_NONE;

            /* continue with listener setup and linking */
            listener->server_fd = server_fd;
            listener->listen    = listen;

            if (listen->flags & MK_CAP_HTTP) {
                protocol = mk_sched_handler_cap(MK_CAP_HTTP);
                if (!protocol) {
                    mk_err("HTTP protocol not supported");
                    exit(EXIT_FAILURE);
                }
                listener->protocol = protocol;
            }

#ifdef MK_HAVE_HTTP2
            if (listen->flags & MK_CAP_HTTP2) {
                protocol = mk_sched_handler_cap(MK_CAP_HTTP2);
                if (!protocol) {
                    mk_err("HTTP2 protocol not supported");
                    exit(EXIT_FAILURE);
                }
                listener->protocol = protocol;
            }
#endif
            listener->network = mk_plugin_cap(MK_CAP_SOCK_PLAIN, server);

            if (listen->flags & MK_CAP_SOCK_TLS) {
                plugin = mk_plugin_cap(MK_CAP_SOCK_TLS, server);
                if (!plugin) {
                    mk_err("SSL/TLS not supported");
                    exit(EXIT_FAILURE);
                }
                listener->network = plugin;
            }

            mk_list_add(&listener->_head, listeners);
        }
        else {
            mk_err("[server] Failed to bind server socket to %s:%s.",
                   listen->address,
                   listen->port);
            return NULL;
        }
    }

    if (reuse_port == MK_TRUE) {
        MK_TLS_SET(mk_tls_server_listen, listeners);
    }

    return listeners;

error:
    return NULL;
}

/* Here we launch the worker threads to attend clients */
void mk_server_launch_workers(struct mk_server *server)
{
    int i;
    pthread_t skip;

    /* Launch workers */
    for (i = 0; i < server->workers; i++) {
        /* Spawn the thread */
        mk_sched_launch_thread(server, &skip);
    }
}

/*
 * When using the FIFO interface, this function get's the FIFO worker
 * context and register the pipe file descriptor into the main event
 * loop.
 *
 * note: this function is invoked by each worker thread.
 */
static int mk_server_fifo_worker_setup(struct mk_event_loop *evl)
{
    int ret;
    struct mk_fifo_worker *fw;

    fw = pthread_getspecific(mk_server_fifo_key);
    if (!fw) {
        return -1;
    }

    ret = mk_event_add(evl, fw->channel[0],
                       MK_EVENT_FIFO, MK_EVENT_READ,
                       fw);
    if (ret != 0) {
        mk_err("[server] Error registering fifo worker channel: %s",
               strerror(errno));
        return -1;
    }

    return 0;
}

/*
 * The loop_balancer() runs in the main process context and is considered
 * the old-fashion way to handle connections. It have an event queue waiting
 * for connections, once one arrives, it decides which worker (thread) may
 * handle it registering the accept(2)ed file descriptor on the worker
 * event monitored queue.
 */
void mk_server_loop_balancer(struct mk_server *server)
{
    size_t bytes;
    uint64_t val;
    int operation_flag;
    struct mk_list *head;
    struct mk_list *listeners;
    struct mk_server_listen *listener;
    struct mk_event *event;
    struct mk_event_loop *evl;
    struct mk_sched_worker *sched;
    struct mk_event management_event;

    /* Init the listeners */
    listeners = mk_server_listen_init(server);
    if (!listeners) {
        mk_err("Failed to initialize listen sockets.");
        return;
    }

    /* Create an event loop context */
    evl = mk_event_loop_create(MK_EVENT_QUEUE_SIZE);
    if (!evl) {
        mk_err("Could not initialize event loop");
        exit(EXIT_FAILURE);
    }

    /* Register the listeners */
    mk_list_foreach(head, listeners) {
        listener = mk_list_entry(head, struct mk_server_listen, _head);
        mk_event_add(evl, listener->server_fd,
                     MK_EVENT_LISTENER, MK_EVENT_READ,
                     listener);
    }

    memset(&management_event, 0, sizeof(struct mk_event));

    mk_event_add(evl,
                 server->lib_ch_manager[0],
                 MK_EVENT_NOTIFICATION,
                 MK_EVENT_READ,
                 &management_event);

    operation_flag = MK_TRUE;
    while (operation_flag) {
        mk_event_wait(evl);
        mk_event_foreach(event, evl) {
            if (event->mask & MK_EVENT_READ) {
                /* This signal is sent by mk_stop and both this and
                 * mk_lib_worker are expecting it.
                 */
                if (server->lib_ch_manager[0] == event->fd) {
#ifdef _WIN32
        bytes = recv(event->fd, &val, sizeof(uint64_t), MSG_WAITALL);
#else
        bytes = read(event->fd, &val, sizeof(uint64_t));
#endif

                    if (bytes <= 0) {
                        return;
                    }

                    if (val == MK_SERVER_SIGNAL_STOP) {
                        operation_flag = MK_FALSE;

                        break;
                    }

                    continue;
                }

                /*
                 * Accept connection: determinate which thread may work on this
                 * new connection.
                 */
                sched = mk_sched_next_target(server);
                if (sched != NULL) {
                    mk_server_listen_handler(sched, event, server);

                    mk_server_lib_notify_event_loop_break(sched);

#ifdef MK_HAVE_TRACE
                    int i;
                    struct mk_sched_ctx *ctx = server->sched_ctx;

                    for (i = 0; i < server->workers; i++) {
                        MK_TRACE("Worker Status");
                        MK_TRACE(" WID %i / conx = %llu",
                                 ctx->workers[i].idx,
                                 ctx->workers[i].accepted_connections -
                                 ctx->workers[i].closed_connections);
                    }
#endif
                }
                else {
                    mk_warn("[server] Over capacity.");
                }
            }
            else if (event->mask & MK_EVENT_CLOSE) {
                mk_err("[server] Error on socket %d: %s",
                       event->fd, strerror(errno));
            }
        }
    }
    mk_event_loop_destroy(evl);
    mk_server_listen_exit(listeners);
}

/*
 * This function is called when the scheduler is running in the REUSEPORT
 * mode. That means that each worker is listening on shared TCP ports.
 *
 * When using shared TCP ports the Kernel decides to which worker the
 * connection will be assigned.
 */
void mk_server_worker_loop(struct mk_server *server)
{
    int ret = -1;
    int timeout_fd;
    uint64_t val;
    struct mk_event *event;
    struct mk_event_loop *evl;
    struct mk_list *list;
    struct mk_list *head;
    struct mk_sched_conn *conn;
    struct mk_sched_worker *sched;
    struct mk_server_listen *listener;
    struct mk_server_timeout *server_timeout;

    /* Get thread conf */
    sched = mk_sched_get_thread_conf();
    evl = sched->loop;

    /*
     * The worker will NOT process any connection until the master
     * process through mk_server_loop() send us the green light
     * signal MK_SERVER_SIGNAL_START.
     */
    mk_event_wait(evl);
    mk_event_foreach(event, evl) {
        if ((event->mask & MK_EVENT_READ) &&
            event->type == MK_EVENT_NOTIFICATION) {
            if (event->fd == sched->signal_channel_r) {
        /* When using libevent _mk_event_channel_create creates a unix socket
         * instead of a pipe and windows doesn't us calling read / write on a
         * socket instead of recv / send
         */
#ifdef _WIN32
                ret = recv(event->fd, &val, sizeof(val), MSG_WAITALL);
#else
                ret = read(event->fd, &val, sizeof(val));
#endif
                if (ret < 0) {
                    mk_libc_error("read");
                    continue;
                }
                if (val == MK_SERVER_SIGNAL_START) {
                    MK_TRACE("Worker %i started (SIGNAL_START)", sched->idx);
                    break;
                }
            }
        }
    }

    if (server->scheduler_mode == MK_SCHEDULER_REUSEPORT) {
        /* Register listeners */
        list = MK_TLS_GET(mk_tls_server_listen);
        mk_list_foreach(head, list) {
            listener = mk_list_entry(head, struct mk_server_listen, _head);
            mk_event_add(sched->loop, listener->server_fd,
                         MK_EVENT_LISTENER, MK_EVENT_READ,
                         listener);
        }
    }

    /*
     * If running in library mode, register the FIFO pipe file descriptors
     * into the main event loop.
     */
    if (server->lib_mode == MK_TRUE) {
        mk_server_fifo_worker_setup(evl);
    }

    /* create a new timeout file descriptor */
    server_timeout = mk_mem_alloc_z(sizeof(struct mk_server_timeout));
    MK_TLS_SET(mk_tls_server_timeout, server_timeout);
    timeout_fd = mk_event_timeout_create(evl, server->timeout, 0, server_timeout);

    while (1) {
        mk_event_wait(evl);
        mk_event_foreach(event, evl) {
            ret = 0;
            if (event->type & MK_EVENT_IDLE) {
                continue;
            }

            if (event->type == MK_EVENT_CONNECTION) {
                conn = (struct mk_sched_conn *) event;

                if (event->mask & MK_EVENT_WRITE) {
                    MK_TRACE("[FD %i] Event WRITE", event->fd);
                    ret = mk_sched_event_write(conn, sched, server);
                }

                if (event->mask & MK_EVENT_READ) {
                    MK_TRACE("[FD %i] Event READ", event->fd);
                    ret = mk_sched_event_read(conn, sched, server);
                }


                if (event->mask & MK_EVENT_CLOSE && ret != -1) {
                    MK_TRACE("[FD %i] Event CLOSE", event->fd);
                    ret = -1;
                }

                if (ret < 0 && conn->status != MK_SCHED_CONN_CLOSED) {
                    MK_TRACE("[FD %i] Event FORCE CLOSE | ret = %i",
                             event->fd, ret);
                    mk_sched_event_close(conn, sched, MK_EP_SOCKET_CLOSED,
                                         server);
                }
            }
            else if (event->type == MK_EVENT_LISTENER) {
                /*
                 * A new connection have been accepted..or failed, despite
                 * the result, we let the loop continue processing the other
                 * events triggered.
                 */
                conn = mk_server_listen_handler(sched, event, server);
                if (conn) {
                    //conn->event.mask = MK_EVENT_READ
                    //goto speed;
                }
                continue;
            }
            else if (event->type == MK_EVENT_CUSTOM) {
                /*
                 * We got an event associated to a custom interface, that
                 * means a plugin registered some file descriptor on this
                 * event loop and an event was triggered. We pass the control
                 * to the defined event handler.
                 */
                event->handler(event);
            }
            else if (event->type == MK_EVENT_NOTIFICATION) {
#ifdef _WIN32
                ret = recv(event->fd, &val, sizeof(val), MSG_WAITALL);
#else
                ret = read(event->fd, &val, sizeof(val));
#endif
                if (ret < 0) {
                    mk_libc_error("read");
                    continue;
                }

                if (event->fd == sched->signal_channel_r) {
                    if (val == MK_SCHED_SIGNAL_DEADBEEF) {
                        //FIXME:mk_sched_sync_counters();
                        continue;
                    }
                    else if (val == MK_SCHED_SIGNAL_FREE_ALL) {
                        if (timeout_fd > 0) {
                            mk_event_timeout_destroy(evl, server_timeout);
                        }
                        mk_mem_free(MK_TLS_GET(mk_tls_server_timeout));
                        mk_server_listen_exit(sched->listeners);
                        mk_event_loop_destroy(evl);
                        mk_sched_worker_free(server);
                        return;
                    }
                    else if (val == MK_SCHED_SIGNAL_EVENT_LOOP_BREAK) {
                        /* NOTE: This is just a notification that's sent to break out
                         *       of the libevent loop in windows after accepting a new
                         *       client
                        */
                        MK_TRACE("New client accepted, awesome!");
                    }
                }
                else if (event->fd == timeout_fd) {
                    mk_sched_check_timeouts(sched, server);
                }
                continue;
            }
            else if (event->type == MK_EVENT_THREAD) {
                mk_http_thread_event(event);
                continue;
            }
            else if (event->type == MK_EVENT_FIFO) {
                mk_fifo_worker_read(event);
                continue;
            }
        }
        mk_sched_threads_purge(sched);
        mk_sched_event_free_all(sched);
    }
}

static int mk_server_lib_notify_event_loop_break(struct mk_sched_worker *sched)
{
    uint64_t val;

    /* Check the channel is valid (enabled by library mode) */
    if (sched->signal_channel_w <= 0) {
        return -1;
    }

    val = MK_SCHED_SIGNAL_EVENT_LOOP_BREAK;

#ifdef _WIN32
    return send(sched->signal_channel_w, &val, sizeof(uint64_t), 0);
#else
    return write(sched->signal_channel_w, &val, sizeof(uint64_t));
#endif
}

static int mk_server_lib_notify_started(struct mk_server *server)
{
    uint64_t val;

    /* Check the channel is valid (enabled by library mode) */
    if (server->lib_ch_start[1] <= 0) {
        return -1;
    }

    val = MK_SERVER_SIGNAL_START;

#ifdef _WIN32
    return send(server->lib_ch_start[1], &val, sizeof(uint64_t), 0);
#else
    return write(server->lib_ch_start[1], &val, sizeof(uint64_t));
#endif
}

void mk_server_loop(struct mk_server *server)
{
    uint64_t val;

    /* Rename worker */
    mk_utils_worker_rename("monkey: server");

    if (server->lib_mode == MK_FALSE) {
        mk_info("HTTP Server started");
    }

    /* Wake up workers */
    val = MK_SERVER_SIGNAL_START;
    mk_sched_broadcast_signal(server, val);

    /* Signal lib caller (if any) */
    mk_server_lib_notify_started(server);

    /*
     * When using REUSEPORT mode on the Scheduler, we need to signal
     * them so they can start processing connections.
     */
    if (server->scheduler_mode == MK_SCHEDULER_REUSEPORT) {
        /* do thing :) */
    }
    else {
        /* FIXME!: this old mode needs some checks on library mode */
        mk_server_loop_balancer(server);
    }
}