Tue, 10 Nov 2015 21:11:06 +0100
refactored IO system
/* * 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; listener->ssl = NULL; if(conf->ssl) { listener->ssl = malloc(sizeof(HttpSSL)); SSL_CTX *ctx = SSL_CTX_new( SSLv23_server_method()); SSL_CTX_set_options(ctx, SSL_OP_SINGLE_DH_USE); sstr_t file = sstrdup(conf->certfile); int ret = SSL_CTX_use_certificate_file(ctx, file.ptr, SSL_FILETYPE_PEM); free(file.ptr); if(!ret) { // TODO: cleanup return NULL; } file = sstrdup(conf->privkeyfile); ret = SSL_CTX_use_PrivateKey_file(ctx, file.ptr, SSL_FILETYPE_PEM); free(file.ptr); if(!ret) { // TODO: cleanup return NULL; } // TODO: chain listener->ssl->sslctx = ctx; } 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; if(ls->ssl) { SSL *ssl = SSL_new(ls->ssl->sslctx); SSL_set_fd(ssl, clientfd); if(SSL_accept(ssl) <= 0) { free(conn); conn = NULL; } else { conn->ssl = ssl; conn->read = connection_ssl_read; conn->write = connection_ssl_write; conn->close = connection_ssl_close; } } else { conn->read = connection_read; conn->write = connection_write; conn->close = connection_close; } if(conn) { 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; }