src/server/daemon/httplistener.c

/*
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
 *
 * Copyright 2013 Olaf Wintermann. All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 *   1. Redistributions of source code must retain the above copyright
 *      notice, this list of conditions and the following disclaimer.
 *
 *   2. 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.
 *
 * 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 HOLDER 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 "../public/nsapi.h"

#include <stdio.h>
#include <stdlib.h>
#include <fcntl.h>
#include <sys/shm.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/socket.h>
#include <sys/file.h>
#include <netinet/in.h>
#include <netdb.h>
#include <stdio.h>
#include <stdlib.h>
#include <fcntl.h>
#include <unistd.h>
#include <strings.h>
#include <stdbool.h>
#include <pthread.h>

#include <ucx/map.h>

#include "../util/atomic.h"
#include "httplistener.h"

#include "session.h"
#include "configmanager.h"
#include "log.h"

UcxMap *listener_map = NULL;

int start_all_listener() {
    ServerConfiguration *conf = cfgmgr_get_server_config();
    UcxList *ls = conf->listeners;
    while(ls) {
        HttpListener *listener = ls->data;
        http_listener_start(listener);
        ls = ls->next;
    }

    return 0;
}

HttpListener* http_listener_create(ListenerConfig *conf) {
    if(listener_map == NULL) {
        listener_map = ucx_map_new(16);
    }

    HttpListener *fl = ucx_map_sstr_get(listener_map, conf->name);
    if(fl == NULL) {
        return http_listener_new(conf);
    }
    
    HttpListener* newls = malloc(sizeof(HttpListener));
    if(newls == NULL) {
        // TODO: error
    }
    
    newls->name = conf->name;
    newls->cfg = conf->cfg;
    newls->nacceptors = conf->nacceptors;
    newls->default_vs.vs_name = conf->vs.ptr;  
    newls->port = fl->port;
    newls->server_socket = fl->server_socket;
    newls->running = 1;
    newls->threadpool = NULL;
    newls->ref = 2; // 1 reference is fl->next
    
    newls->session_handler = fl->session_handler; // TODO
    
    // the listener threadpool might be changed
    if(conf->threadpool.ptr != NULL) {
        newls->threadpool = get_threadpool(conf->threadpool);
    }
    if(newls->threadpool == NULL) {
        newls->threadpool = get_default_threadpool();
    }
    
    // create acceptor threads
    newls->acceptors = calloc(newls->nacceptors, sizeof(void*));
    for (int i=0;i<newls->nacceptors;i++) {
        newls->acceptors[i] = acceptor_new(newls);
    }
    
    // fl hold one reference of newls
    fl->next = newls;
    
    
    ucx_map_sstr_put(listener_map, newls->name, newls);
    
    for (int i=0;i<newls->nacceptors;i++) {
        acceptor_start(newls->acceptors[i]);
    }
    
    // check if a restart is required to apply all changes
    
    if(newls->port != conf->port) {
        // TODO: log
    }
    
    return newls;
}

HttpListener* http_listener_new(ListenerConfig *conf) {
    // TODO: remove
    if(listener_map == NULL) {
        listener_map = ucx_map_new(16);
    }

    HttpListener *fl = ucx_map_sstr_get(listener_map, conf->name);
    if(fl != NULL) {
        return fl;
    }
    // end remove

    HttpListener *listener = malloc(sizeof(HttpListener));
    listener->running = 0;
    listener->cfg = conf->cfg;
    listener->name = conf->name;
    listener->default_vs.vs_name = conf->vs.ptr;
    listener->threadpool = NULL;
    if(conf->threadpool.ptr != NULL) {
        listener->threadpool = get_threadpool(conf->threadpool);
    }
    if(listener->threadpool == NULL) {
        listener->threadpool = get_default_threadpool();
    }
    //listener->session_handler = create_basic_session_handler();
    listener->session_handler = create_event_session_handler();
    listener->nacceptors = conf->nacceptors;
    listener->port = conf->port;
    listener->ref = 1;
    listener->next = NULL;
    ucx_map_sstr_put(listener_map, listener->name, listener);

    struct sockaddr_in servaddr;   /* server address */

    /* init address structure */
    memset(&servaddr, 0, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    servaddr.sin_port = htons(conf->port);

    /* create socket */
    if((listener->server_socket = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
        perror("Error: http_listener_new: socket");
        return NULL;
    }

    int o = 1;
    setsockopt(
            listener->server_socket,
            SOL_SOCKET, SO_REUSEADDR,
            &o,
            sizeof(int));

    /* bind server socket to address */
    if(bind(listener->server_socket, (struct sockaddr*)&servaddr, sizeof(servaddr))){
        perror("Error: http_listener_new: bind");
        printf("port: %d\n", conf->port);
        return NULL;
    }

    /* create acceptors */
    listener->acceptors = calloc(listener->nacceptors, sizeof(void*));
    for (int i=0;i<listener->nacceptors;i++) {
        listener->acceptors[i] = acceptor_new(listener);
    }

    return listener;
}

int http_listener_start(HttpListener *listener) {
    if(listener->running) {
        return 0;
    }
    printf("INFO: start listener\n");
    log_ereport(LOG_INFORM, "start listener on port %d", listener->port);

    if (listen(listener->server_socket, 256) == -1) {
        perror("Error: http_listener_start: listen");
        return -1;
    }

    /* start acceptor threads */
    for (int i=0;i<listener->nacceptors;i++) {
        acceptor_start(listener->acceptors[i]);
    }

    return 0;
}

void http_listener_ref(HttpListener *listener) {
    ws_atomic_inc32(&listener->ref);
}

void http_listener_unref(HttpListener *listener) {
    uint32_t ref = ws_atomic_dec32(&listener->ref);
    if(ref == 0) {
        free(listener->acceptors);
        // TODO: unref cfg
        // TODO: unref session handler
        free(listener);
    }
}


Acceptor* acceptor_new(HttpListener *listener) {
    Acceptor *acceptor = malloc(sizeof(Acceptor));
    acceptor->listener = listener;
    return acceptor;
}

void acceptor_start(Acceptor *a) {
    if(pthread_create(
            &a->tid,
            NULL,
            (void*(*)(void*))acceptor_thread,
            a) != 0)
    {
        perror("Error: acceptor_start: pthread_create");
    }
}

void* acceptor_thread(Acceptor *acceptor) {
    WS_ASSERT(acceptor);
    WS_ASSERT(acceptor->listener);
    WS_ASSERT(acceptor->listener->session_handler);
    WS_ASSERT(acceptor->listener->session_handler->enqueue_connection);
    
    HttpListener *listener = acceptor->listener;

    for (;;) {
        /* accept connections */
        struct sockaddr_in ca;
        socklen_t length = sizeof(ca);
        int clientfd;

        /* accept a connection */
        clientfd = accept(
                listener->server_socket,
                (struct sockaddr*)&ca,
                &length);
        if (clientfd == -1) {
            perror("Error: acceptor_thread: accept");
            continue;
        }
        
        // check listener
        HttpListener *ls = listener;
        int acceptor_exit = 0;
        while(ls->next) {
            ls = ls->next;
            acceptor_exit = 1;
        }
        
        /* create Connection object */
        Connection *conn = malloc(sizeof(Connection));
        conn->address = ca;
        conn->fd = clientfd;
        conn->listener = ls;
        
        cfg_ref(ls->cfg);

        /* enqueue the connection */
        ls->session_handler->enqueue_connection(
                ls->session_handler,
                conn);

        /* ready for new connection */
        
        if(acceptor_exit) {
            // this acceptor is outdated
            break;
        }
    }
    
    http_listener_unref(listener->next);
    http_listener_unref(listener);
    
    return NULL;
}