/Users/maurits/Documents/studie/afstuderen/biosphere/daemon/thread.c

Go to the documentation of this file.

/*
 * Author: MA Hartman
 * Date: feb 15, 2007
 * 
 * Function:
 * Creation and cleanup of threads.
 * 
 * License information:
 * 
 * Copyright (c) 2006 Maurits Hartman
 *
 * Permission is hereby granted, free of charge, to any person
 * obtaining a copy of this software and associated documentation
 * files (the "Software"), to deal in the Software without
 * restriction, including without limitation the rights to use,
 * copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the
 * Software is furnished to do so, subject to the following
 * conditions:
 * 
 * The above copyright notice and this permission notice shall be
 * included in all copies or substantial portions of the Software.
 * 
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
 * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
 * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
 * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
 * OTHER DEALINGS IN THE SOFTWARE.
 * 
 */

#include "thread.h"
#include "http.h"
#include "module.h"
#include "mutex.h"
#include "options.h"
#include "settings.h"
#include <biosphere.h>
#include <copy.h>
#include <list.h>

#include <apr_file_io.h>
#include <apr_strings.h>
#include <apr_network_io.h>
#include <apr_thread_proc.h>
#include <apr_time.h>
#include <assert.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#include <sys/time.h>


typedef struct worker_info {
        apr_socket_t *sock;     
        apr_time_t ts;          
        apr_pool_t *pool;       
} worker_info;

typedef struct worker_thread_args {
        apr_socket_t *sock;     
        apr_pool_t *pool;       
} worker_thread_args;

#define MAX_HEADER_CHECK_ATTEMPTS 4

extern apr_pool_t *global_mp;

static apr_socket_t *listener_socket;

static bs_list *sockets;


static bs_status init_listen_socket(apr_socket_t **sock,
                apr_pool_t *mp)
{
    apr_status_t rv;
    apr_socket_t *s;
    apr_sockaddr_t *sa;
    
    /* Set the sockaddr info: */
    rv = apr_sockaddr_info_get(&sa,
                NULL, /* hostname */
                APR_INET,
                global_listen_port,
                0, /* flags */
                mp);
    if (rv != APR_SUCCESS) {
        char err[256];
        apr_strerror(rv, err, sizeof(err));
        fprintf(stderr, "Unable to create socket address: %s\n", err);
        return BS_ERROR;
    }
    
    /* Create the socket: */
    rv = apr_socket_create(&s, sa->family, SOCK_STREAM, APR_PROTO_TCP, mp);
    if (rv != APR_SUCCESS) {
        char err[256];
        apr_strerror(rv, err, sizeof(err));
        fprintf(stderr, "Unable to create socket: %s\n", err);
        return BS_ERROR;
    }

    /* It is a good idea to specify socket options explicitly.
     * in this case, we make a blocking-forever socket as the
     * listening socket: */
    apr_socket_opt_set(s, APR_SO_NONBLOCK, 0);
    apr_socket_timeout_set(s, -1);
    /* this is useful for a server(socket listening) process.
     * When the server crashes and quickly restarted, it can use
     * the same socket again: */
    apr_socket_opt_set(s, APR_SO_REUSEADDR, 1);

        /* Bind the socket to the address and start listening: */
    rv = apr_socket_bind(s, sa);
    if (rv != APR_SUCCESS) {
        char err[256];
        apr_strerror(rv, err, sizeof(err));
        fprintf(stderr, "Unable to bind to address: %s\n", err);
                apr_socket_close(s);
        return BS_ERROR;
    }
    rv = apr_socket_listen(s, SOMAXCONN);
    if (rv != APR_SUCCESS) {
        char err[256];
        apr_strerror(rv, err, sizeof(err));
        fprintf(stderr, "Unable to start listening on socket: %s\n", err);
        apr_socket_close(s);
        return BS_ERROR;
    }

    *sock = s;
    return BS_OK;
}


static void remove_from_socket_list(apr_socket_t *sock)
{
        unsigned i;
        lock_thread_mux();
        for (i = 0; i < list_size(sockets); i++) {
                worker_info *wi = list_index(sockets, i);
                if (wi->sock == sock) {
                        list_remove(sockets, i);
                        break;  
                }       
        }
        unlock_thread_mux();
}


static void add_to_socket_list(apr_socket_t *sock, apr_pool_t *mp)
{
        worker_info *wi = (worker_info *) apr_palloc(mp, sizeof(worker_info));
        wi->sock = sock;
        wi->ts = apr_time_now();
        wi->pool = mp;
        
        lock_thread_mux();
        list_append(sockets, wi);
        unlock_thread_mux();
}


static void* APR_THREAD_FUNC worker_thread_main(apr_thread_t *thd,
                void *data)
{
        apr_status_t rv;
        worker_thread_args *args;
        apr_socket_t *sock;
        apr_pool_t *mp; /* Thread's private, standalone pool */
        struct timeval t;
        
        char *headerbuf = NULL; /* Buffer for the HTTP header data */
        apr_size_t headerlen = 0;
        int attempts = 0;
        
        args = data;
        sock = args->sock;
        mp = args->pool;
        add_to_socket_list(sock, mp);
        
        /* Start reading header info from the socket until full headers are
         * received: */
        while (!received_full_headers(headerbuf)) {
                char tmpbuf[HTTP_BUFFER_SIZE];
                apr_size_t tmpbuflen = sizeof(tmpbuf) - 1;
                memset(tmpbuf, '\0', tmpbuflen + 1); /* Null-terminated */
                
                rv = apr_socket_recv(sock, tmpbuf, &tmpbuflen);
                if (rv == APR_EOF || tmpbuflen == 0) {
                        apr_socket_close(sock);
                        remove_from_socket_list(sock);
                        apr_pool_destroy(mp);
                        return NULL;
                }
                
                if (headerbuf == NULL)
                        headerbuf = apr_pstrndup(mp, tmpbuf, tmpbuflen);
                else
                        headerbuf = apr_pstrcat(mp, headerbuf, tmpbuf, NULL);
                headerlen = strlen(headerbuf);
                
                /* If http header collection takes too long, exit: */
                attempts++;
                if (attempts > MAX_HEADER_CHECK_ATTEMPTS) {
                        send_http_error(sock, "HTTP/1.0 406 Not Acceptable", mp);
                        fprintf(stderr, "Too many connections\n");
                        apr_socket_close(sock);
                        remove_from_socket_list(sock);
                        apr_pool_destroy(mp);
                        return NULL;    
                }
        }
        
        /* Determine the kind of request: */
        if (headerbuf[0] == 'P')
                handle_http_post(sock, headerbuf, strlen(headerbuf), mp);
        else if (headerbuf[0] == 'G')
                handle_http_get(sock, mp);
        else send_http_error(sock, "HTTP/1.0 404 Not Found", mp);
        
        gettimeofday(&t, 0);
        printf("%ld.%06ld\n", t.tv_sec, t.tv_usec);
        
        /* Wrapup and exit worker thread: */
        apr_socket_close(sock);
        remove_from_socket_list(sock);
        apr_pool_destroy(mp);
        return NULL;    
}


static bs_status setup_thread(apr_socket_t *sock)
{
        apr_status_t rv;
        apr_thread_t *thd;
        apr_threadattr_t *thd_attr;
        apr_pool_t *thread_pool;
        
        /* Set the attributes to create a default detachable thread: */
        rv = apr_pool_create(&thread_pool, NULL);
        if (rv != APR_SUCCESS) return BS_ERROR;
        apr_threadattr_create(&thd_attr, thread_pool);
        if (rv != APR_SUCCESS) return BS_ERROR;

        /* Create the arguments for the thread: */
        worker_thread_args *args =
                        (worker_thread_args *) apr_palloc(thread_pool,
                        sizeof(worker_thread_args));
        args->pool = thread_pool;
        args->sock = sock;

        rv = apr_thread_create(&thd,
                        thd_attr,
                        worker_thread_main,
                        args,
                        thread_pool);
        if (rv != APR_SUCCESS) return BS_ERROR;
        
        return BS_OK;
}


bs_status init_threadsystem(apr_pool_t *mp)
{
        sockets = new_list(mp);
        return init_listen_socket(&listener_socket, mp);
}


static void remove_oldest_thread(void)
{
        unsigned i;
        apr_time_t oldest = apr_time_now();
        apr_socket_t *socket_to_kill = NULL;
        apr_pool_t *pool_to_destroy = NULL;
        
        apr_sleep(10000); /* Wait for thread to finish */
        
        /* Forcefully remove the oldest */
        for (i = 0; i < list_size(sockets); i++) {
                worker_info *wi = (worker_info *) list_index(sockets, i);
                if (wi->ts < oldest) {
                        oldest = wi->ts;
                        socket_to_kill = wi->sock;
                        pool_to_destroy = wi->pool;
                }
        }
        
        assert(socket_to_kill && pool_to_destroy);
        fprintf(stderr, "Had to forcefully kill worker thread.\n");
        remove_from_socket_list(socket_to_kill);
        apr_pool_destroy(pool_to_destroy);      
}


void lets_go_threaded(void)
{
        apr_pool_t *mp;
        apr_status_t rv;
        struct timeval t;
        
        rv = apr_pool_create(&mp, global_mp);
        if (rv != APR_SUCCESS) return;
        
        /* The main loop waiting for incoming connections: */
        while (1) {
                bs_status rv2;
                apr_socket_t *ns;
                
                /* If the list is full, wait for the oldest thread: */
                lock_thread_mux();
        if (list_size(sockets) >= BIOSPHERED_MAX_CONNECTIONS)
                remove_oldest_thread();
        unlock_thread_mux();
                
                /* Wait for an incoming connection and create a thread for it:
                 * specify socket options explicitly: non-blocking, no time-out:
                 */
        rv = apr_socket_accept(&ns, listener_socket, mp);
        if (rv != APR_SUCCESS) break;
        apr_socket_opt_set(ns, APR_SO_NONBLOCK, 0);
        apr_socket_timeout_set(ns, -1);
        
        gettimeofday(&t, 0);
                printf("%ld.%06ld,", t.tv_sec, t.tv_usec);
        
        /* Setup the worker thread: */
        rv2 = setup_thread(ns);
        if (rv2 != BS_OK) return;
        }
}


typedef struct shutdown_info {
        bs_uint32 wait_time;            
        apr_pool_t *pool;       
} shutdown_info;


static void* APR_THREAD_FUNC shutdown_thread_main(apr_thread_t *thd,
                void *sinfo)
{
        unsigned i;
        int exit_code = BS_OK;
        shutdown_info *si = sinfo;
        
        /* Wait the specified amount of seconds; */
        printf("Shutdown process started, waiting %d microseconds\n", si->wait_time);
        apr_sleep(si->wait_time);
        apr_socket_close(listener_socket);
        
        /* Wait 10 seconds before cleaning all the worker threads: */
        printf("Allowing remaining threads 10 seconds to finish\n");
        apr_sleep(10*1000*1000);
        printf("Killing remaining worker threads\n");
        
        lock_thread_mux();
        for (i = 0; i < list_size(sockets); i++) {
                apr_socket_t *sock = list_index(sockets, i);
                apr_socket_close(sock);
                exit_code += BS_ERROR;
        }
        unlock_thread_mux();
        
        exit_code += unload_all_modules();
        
        if (i > 0)
                fprintf(stderr, "There were %d remaining zombie worker threads to kill.\n", i);
        
        apr_pool_destroy(si->pool);
        exit(exit_code);
}


bs_status stop_daemon(bs_uint32 num_usecs)
{
        apr_status_t rv;
        apr_thread_t *thd;
        apr_threadattr_t *thd_attr;
        apr_pool_t *thread_pool;
        shutdown_info *sinfo;
        
        /* Set the attributes to create a default detachable thread: */
        rv = apr_pool_create(&thread_pool, NULL);
        if (rv != APR_SUCCESS) return BS_ERROR;
        apr_threadattr_create(&thd_attr, thread_pool);
        if (rv != APR_SUCCESS) return BS_ERROR;
        
        sinfo = (shutdown_info *) apr_palloc(thread_pool, sizeof(shutdown_info));
        sinfo->wait_time = num_usecs;
        sinfo->pool = thread_pool;

        rv = apr_thread_create(&thd,
                        thd_attr,
                        shutdown_thread_main,
                        sinfo,
                        thread_pool);
        if (rv != APR_SUCCESS) return BS_ERROR;
        return BS_OK;   
}

---