root/cserver/server.c

/*
 * Copyright (c) 2006, Federico Tomassini AKA efphe (effetom AT gmail DOT com)
 * All rights reserved.
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 *     * Redistributions of source code must retain the above copyright
 *       notice, this list of conditions and the following disclaimer.
 *     * 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.
 *     * Neither the name of the University of California, Berkeley nor the
 *       names of its contributors may be used to endorse or promote products
 *       derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE REGENTS 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 REGENTS AND 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 "server.h"
#include "server_utils.h"
#include "wsignals.h"
#include "wmalloc.h"

static pmt _servers_must_die_mutex=PTHREAD_MUTEX_INITIALIZER;
static int _servers_must_die=0;
static pmt _servers_working_threads_mutex=PTHREAD_MUTEX_INITIALIZER;
static pct _servers_working_threads_cond=PTHREAD_COND_INITIALIZER;
static int _servers_working_threads=0;

/*
 * Server API
 *  Handle the server struct with the following functions,
 *  never directly.
 */

void server_log_init(LogLevel level, SyslogFacility facility, int on_stderr)
{
    if (level==-1)
        level= SYSLOG_LEVEL_INFO;
    if (facility==-1)
        facility= LOG_DAEMON;
    log_init("cserver", level, facility, on_stderr);
    return;
}

void server_set_signal_handler(server *s, int signum, void(*f)(int))
{
    ssh **o,*k;
    sigset_t temp;
    void (*old_f)(int)=0;

    w_sigemptyset(&temp);
    w_sigaddset(&temp, signum);
    o=&(s->signal_handlers);

    while ((k=*o)) {
        /* Overwrite handler. */
        if (k->signum==signum) {
            old_f= k->f;
            k->f= f;
        }

        /* 
         * Ei, this handler is already used. Collect 
         * signal in temp. 
         */
        if (k->f==f) {
            w_sigaddset(&k->sa_mask, signum);
            w_sigaddset(&temp, k->signum);
        }
        o=&((*o)->next);
    }

    if (!old_f) /* This is a new signal. */
        *o=_server_new_signal_handler(signum, &temp, f);

    else { /* A signal handler was modified. Remove dependencies. */
        o=&(s->signal_handlers);
        while((k=*o)) {
            if (k->f == old_f )
                w_sigdelset(&k->sa_mask, signum);
            o=&((*o)->next);
        }
    }

    return;
}


void server_init(server *s)
{
    memset(s,0,sizeof(server));

    server_set_tps(s,DEF_THREADS_PER_SOCKET);
    server_set_family(s,AF_INET);
    server_set_type(s,SOCK_STREAM);
    server_set_backlog(s,DEF_SERVER_BACKLOG);
    server_set_signal_handler(s,SIGINT,&_server_default_sigint_handler);
    server_set_signal_handler(s,SIGTERM,&_server_default_sigterm_handler);
    server_log_init(-1,-1,1);

    return;
}

void server_set_name(server *s,char *name)
{
    int slen;

    slen=strlen(name)+1;
    if (slen>MAX_SERVER_NAME_LEN)
        fatal("Server name (%s) is too long: max is %d", 
                name, MAX_SERVER_NAME_LEN);
    strcpy(s->name,name);
}

void server_set_backlog(server *s,int backlog)
{
    if (backlog>MAX_SERVER_BACKLOG)
        fatal("Backlog value too high: max is %d.", 
                MAX_SERVER_BACKLOG);
    s->backlog=backlog;
    return;
}

void server_set_tps(server *s,int tps)
{
    if (tps>MAX_THREADS_PER_SOCKET)
        fatal("Too many threads per socket: max is %d.",
                MAX_THREADS_PER_SOCKET);
    s->threads_per_socket=tps;
    return;
}

void servers_destroy(server *s, int nserver)
{
    int i;

    for (i=0; i<nserver; i++) 
        _server_destroy(s+i);
}

int server_add_hostname(server *s,char *hname)
{
    int n,len;
    char *shname;

    n=s->hostnames_n;
    if (n>= MAX_SERVER_HOSTNAMES) 
        fatal("Too many server hostnames: max is %d.",
                MAX_SERVER_HOSTNAMES);

    len=strlen(hname);
    s->hostnames[n]= (char*)w_malloc(len+1);
    shname= s->hostnames[n];
    strcpy(shname, hname);
    s->hostnames_n++;
    return 0;
}

void server_set_type(server *s, int type)
{
    s->type= type;
    return;
}
    
void server_set_family(server *s, int family)
{
    s->family= family;
    return;
}


void server_set_port(server *s,uint16_t port)
{
    s->port=port;
    return;
}

void server_set_handler(server *s, 
        void (*f)(int,struct sockaddr*,socklen_t*))
{
    s->handler=f;
    return;
}


void servers_serve(server *servers,int nserver)
{
    int i;
    server *s;

    for (i=0; i<nserver; i ++) {
        s=servers+i;
        _server_serve(s);
    }

    while (pause()) 
        if (_servers_are_stopped()) 
            break;

    servers_destroy(servers, nserver);

    return;
}

/*
 * do setsockopt for each socket on server.
 * Returns the sum of each return value of setsockopt,
 * so, the number of error is the module of the
 * returned value
 */
int server_setsockopt(server *s,
                      int level, 
                      int optname, 
                      const void *optval, 
                      socklen_t optlen)
{
    int i,res=0;

    for (i=0; i<s->sockets_n; i++) 
        res+=setsockopt(s->sockets[i], level, optname, 
                optval, optlen);
    return res;
}

void servers_stop(void)
{
    w_pthread_mutex_lock(&_servers_must_die_mutex);
    _servers_must_die=1;
    w_pthread_mutex_unlock(&_servers_must_die_mutex);
    return;
}

void servers_slow_stop(void)
{
    /* Waiting threads end serving */
    servers_stop();

    w_pthread_mutex_lock(&_servers_working_threads_mutex);

    /* If there are active connections, wait */
    if (_servers_working_threads)
        w_pthread_cond_wait_unlock(&_servers_working_threads_cond,
                                   &_servers_working_threads_mutex);

    /* Waiting for active connections, then stop. */
    return;
}

/*
 * Internal functions. Internal use.
 */

ssh* _server_new_signal_handler(int signum, sigset_t *sa_mask, void(*f)(int))
{
    ssh *sh;

    sh=(ssh*)w_malloc(SSH_SIZE);

    sh->signum= signum;
    sh->f= f;
    sh->next= 0;
    w_sigemptyset(&sh->sa_mask);

    if (sa_mask) 
        memcpy(&sh->sa_mask,sa_mask,sizeof(sigset_t));
    else
        w_sigaddset(&sh->sa_mask,signum);

    return sh;
}

ssh* _server_has_signal_handler(server *s, int signum)
{
    ssh* o;

    o=s->signal_handlers;

    while(o) {
        if (o->signum==signum)
            return o;
        else o= o-> next;
    }

    return NULL;
}

int _servers_are_stopped()
{
    int res;

    w_pthread_mutex_lock(&_servers_must_die_mutex);
    res=_servers_must_die;
    w_pthread_mutex_unlock(&_servers_must_die_mutex);
    return res;
}

void _server_destroy(server *s)
{
    int n_hname,i;
    ssh *sh,*temp;

    n_hname=s->hostnames_n;

    sh=s->signal_handlers;
    while(sh) {
        temp=sh->next;
        w_free(sh);
        sh=temp;
    }

    if (n_hname)
        for (i=0;i<n_hname;i++)
            w_free(s->hostnames[i]);
    return;
}
    
int _server_add_socket(server *s,int skt)
{
    int n;

    n=s->sockets_n;
    if (n>= MAX_SERVER_SOCKETS) 
        fatal("Too many socket for one server. Max is %d.",
                MAX_SERVER_SOCKETS);

    s->sockets[n]=skt;
    s->sockets_n++;
    return 0;
}
int _server_prepare_bind_sockets(server *s)
{
    char *hname,strport[6];
    int i,res,n_hname,count,skt,family;
    struct addrinfo iaddr,*riaddr,*temp;

    count=0;
    n_hname= s->hostnames_n;
    memset(&iaddr,0,sizeof(iaddr));

    family=s->family;
    iaddr.ai_family=family;
    iaddr.ai_socktype=s->type;
    snprintf(strport,6,"%d",s->port);

    if (n_hname) {
        for (i=0; i<n_hname; i++) {
            hname= s->hostnames[i];
            
            res=getaddrinfo(hname,strport,&iaddr,&riaddr);
            if (res) 
                continue;

            for (temp=riaddr; temp; temp=temp->ai_next) {
                skt=w_socket(temp->ai_family,temp->ai_socktype,0,0);
                if (skt<0) 
                    continue;

                //XXX utilizzare server_setsockopt()
                w_setsockopt_reuseaddr(skt);

                res= bind(skt,temp->ai_addr,temp->ai_addrlen);
                if (res==-1) {
                    close(skt);
                    continue;
                }

                _server_add_socket(s,skt);
                count++;
                //break;
            }
            freeaddrinfo(riaddr);
        }
    } else {

        iaddr.ai_flags |= AI_PASSIVE;

        res=getaddrinfo(0,strport,&iaddr,&riaddr);
        if (res) fatal("Fatal getaddrinfo: %s.",gai_strerror(errno));

        for (temp=riaddr; temp; temp=temp->ai_next) {

            skt=w_socket(family,s->type,0,0);
            if (skt<0) 
                continue;

            //XXX utilizzare server_setsockopt()
            w_setsockopt_reuseaddr(skt);

            res= bind(skt,temp->ai_addr,temp->ai_addrlen);
            if (res==-1) {
                close(skt);
                continue;
            }

            _server_add_socket(s,skt);
            count++;
            break;
        }
    }
    return count;
}

void _server_listen(server *s)
{
    int i,b;
    int *skts;

    /* if SOCK_DGRAM, no listen. */
    if (s->type==SOCK_DGRAM)
        return;

    skts=s->sockets;
    b=s->backlog;

    for (i=0; i<s->sockets_n; i++)
        w_listen(skts[i],b);

    return;
}

void _servers_thread_begin_work(void)
{
    w_pthread_mutex_lock(&_servers_working_threads_mutex);
    _servers_working_threads++;
    w_pthread_mutex_unlock(&_servers_working_threads_mutex);
    return;
}

void _server_thread_ends_work(void)
{
    w_pthread_mutex_lock(&_servers_working_threads_mutex);

    _servers_working_threads--;
    if (!_servers_working_threads)
        w_pthread_cond_signal(&_servers_working_threads_cond);

    w_pthread_mutex_unlock(&_servers_working_threads_mutex);
    return;
}

int _server_thread_begins(pmt *m)
{
    w_pthread_mutex_lock(m);

    if (_servers_are_stopped())
        return 0;

   return 1;
}

void* _server_accept_thread(void *arg)
{
    server *s;
    int thread_skt,client_skt,type,res,skt_pos;
    struct sockaddr sa;
    socklen_t salen;
    pmt *skt_mutex,*proc_sock;
    pct  *c;
    struct pollfd p;
    void (*f)(int,struct sockaddr*,socklen_t*);

    s=(server*)arg;
    c=&(s->threads_counter_cond);
    proc_sock=&(s->processing_socket_mutex);
    salen= sizeof(struct sockaddr);

    /* 
     * Getting my socket, decrementing the threads counter.
     * If i'm the last thread for my socket, signal 
     * the condition to server_serve()
     */
    w_pthread_mutex_lock(proc_sock);

    skt_pos= s->processing_socket;

    s->threads_counter--;
    if (!(s->threads_counter)) {
        w_pthread_mutex_unlock(proc_sock);
        w_pthread_cond_signal(c);
    } else
        w_pthread_mutex_unlock(proc_sock);

    /* Now it's time to process data */
    thread_skt= s->sockets[skt_pos];
    skt_mutex= s->sockets_mutex + skt_pos;
    f= s->handler;
    type= s->type;

    if (type==SOCK_DGRAM) {
        p.events=POLLIN;
        p.fd=thread_skt;
    }

    /* Wait my signal to start */
    w_pthread_mutex_lock_unlock(&(s->start));

    switch (type) {
            case SOCK_STREAM:
                for (;;) {

                    if (!_server_thread_begins(skt_mutex))
                        break;

                    client_skt= w_accept(thread_skt,&sa,&salen);
                    w_pthread_mutex_unlock(skt_mutex);
   
                    _servers_thread_begin_work();

                    if (client_skt!=-1) {
                        f(client_skt,&sa,&salen);
                        close(client_skt);
                    }

                    _server_thread_ends_work();
                }
            case SOCK_DGRAM:
                for (;;) {

                    if (!_server_thread_begins(skt_mutex))
                        break;

                    res= w_poll(&p,1,-1);
                    w_pthread_mutex_unlock(skt_mutex);

                    _servers_thread_begin_work();

                    if (p.revents & POLLIN) {
                        f(thread_skt,0,0);
                    }

                    _server_thread_ends_work();
                }
            default:
                fatal("Socket type not supported.");
                }

    /* Servers are stopped. */
    w_pthread_mutex_unlock(skt_mutex);
    return NULL;
}

void _server_init_signals(server *s)
{
    ssh *sh;

    sh=s->signal_handlers;

    for (sh=s->signal_handlers; sh; sh= sh->next) 
        w_sigaction(sh->signum, &sh->sa_mask, sh->f);

    return;
}


void _server_init_mutexes(server *s)
{
    int n_socks,i;
    pmt *t;

    n_socks=s->sockets_n;

    t=s->sockets_mutex;
    for (i=0; i<n_socks; i++) 
        w_pthread_mutex_init(t+i,0);

    w_pthread_mutex_init(&(s->processing_socket_mutex),0);
    w_pthread_mutex_init(&(s->threads_counter_mutex),0);
    w_pthread_mutex_init(&(s->start),0);
    w_pthread_cond_init(&(s->threads_counter_cond),0);
    return;
}

void _server_default_sigint_handler(int signum)
{
    servers_stop();
    return;
}

void _server_default_sigterm_handler(int signum)
{
    servers_slow_stop();
    return;
}

void _server_serve(server *s)
{
    int n_socks,i,j,tps;
    pthread_t  t;
    pthread_attr_t ta;
    pmt *m,*start;
    pct  *c;

    /* Init services */
    _server_prepare_bind_sockets(s);
    _server_listen(s);

    n_socks=s->sockets_n;
    if (!n_socks)
        fatal("No socket available. Exit.");

    m=&(s->threads_counter_mutex);
    c=&(s->threads_counter_cond);
    tps=s->threads_per_socket;
    start=&(s->start);

    w_pthread_attr_init(&ta);
    w_pthread_attr_setdetachstate(&ta, PTHREAD_CREATE_DETACHED);

    _server_init_mutexes(s);
    _server_init_signals(s);

    /* Threads: wait my signal before serving ... */
    w_pthread_mutex_lock(start);

    for(i=0; i<n_socks; i++) {
        w_pthread_mutex_lock(m);

        s->processing_socket=i;
        s->threads_counter=tps;

        for (j=0; j<tps; j++) 
            w_pthread_create(&t,&ta,&_server_accept_thread,(void*)s);

        /* Wait the cascade ... */
        w_pthread_cond_wait_unlock(c,m);
    }

    w_pthread_attr_destroy(&ta);

    /* Go my children, go. */
    w_pthread_mutex_unlock(start);
    return; 
}