progettoso/src/server.c

#include <unistd.h>
#include <assert.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <pthread.h>
#include <ctype.h>
#include <signal.h>
#include <sys/select.h>

#include <threadpool.h>
#include <conn.h>
#include <util.h>
#include <serverWorker.h>
#include <ini.h>
#include <serverUtil.h>
#include <fileQueue.h>


/**
 *  @struct sigHandlerArgs_t
 *  @brief struttura contenente le informazioni da passare
 *         al signal handler thread
 *
 */
typedef struct {
    sigset_t   *set;          // set dei segnali da gestire (mascherati)
    int         signal_pipe;  // descrittore di scrittura di una pipe senza nome
} sigHandler_t;



// funzione eseguita dal signal handler thread
static void *sigHandler(void *arg);

static void usage(const char *argv0) {
    // TODO change this
    fprintf(stderr, "use: %s config file location\n", argv0);
}

static void checkargs(int argc, char* argv[]) {
    // TODO change all this
    if (argc != 2) {
	usage(argv[0]);
	_exit(EXIT_FAILURE);
    }
    ini_t *config = ini_load(argv[1]);
    if ( config == NULL) {
	fprintf(stderr, "ERROR: unable to read config file\n");
	usage(argv[0]);
	_exit(EXIT_FAILURE);
    }
    ini_free(config);
}

int main(int argc, char *argv[]) {
    // TODO read config file
    checkargs(argc, argv);
    ini_t *config = ini_load(argv[1]);
    int threadsInPool; CONFGETINT(threadsInPool, config, "threadpool", "quantity", NULL, 10);
    int pendingSize; CONFGETINT(pendingSize, config, "threadpool", "pending", NULL, 10);
    int maxFiles; CONFGETINT(maxFiles, config, "files", "MaxFiles", NULL, 10);
    int maxSize; CONFGETINT(maxSize, config, "files", "MaxSize", NULL, 10);
    ini_free(config);

    sigset_t mask;
    sigemptyset(&mask);
    sigaddset(&mask, SIGINT);
    sigaddset(&mask, SIGQUIT);
    sigaddset(&mask, SIGTERM);

    if (pthread_sigmask(SIG_BLOCK, &mask, NULL) != 0) {
	fprintf(stderr, "ERROR setting mask\n");
	goto _cleanup;
    }

    // ignoro SIGPIPE per evitare di essere terminato da una scrittura su un socket
    struct sigaction s;
    memset(&s, 0, sizeof(s));
    s.sa_handler = SIG_IGN;
    if ( (sigaction(SIGPIPE,&s,NULL)) == -1 ) {
	perror("sigaction");
	goto _cleanup;
    }

    printf("File Server ready.");
    fflush(stdout);

    int signal_pipe[2];
    if (pipe(signal_pipe) == -1) {
	perror("pipe");
	goto _cleanup;
    }

    // todo logging + statistiche



    // thread per la gestione delle interruzioni
    pthread_t sighandler_thread;
    sigHandler_t handlerArgs = { &mask, signal_pipe[1] };
    if (pthread_create(&sighandler_thread, NULL, sigHandler, &handlerArgs) != 0) {
	fprintf(stderr, "ERROR creating signal handler thread\n");
	goto _cleanup;
    }

    int listenfd;
    if ((listenfd = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) {
	perror("socket");
	goto _cleanup;
    }

    // thread per i segnali
    struct sockaddr_un serv_addr;
    memset(&serv_addr, '0', sizeof(serv_addr));
    serv_addr.sun_family = AF_UNIX;
    strncpy(serv_addr.sun_path, SOCKNAME, strlen(SOCKNAME)+1);

    if (bind(listenfd, (struct sockaddr*) &serv_addr, sizeof(serv_addr)) == -1) {
	perror("bind");
	goto _cleanup;
    }
    if (listen(listenfd, MAXBACKLOG) == -1) {
	perror("listen");
	goto _cleanup;
    }

    // creo la queue
    queueT *queue = createQueue(maxFiles, maxSize);

    threadpool_t *pool = NULL;

    pool = createThreadPool(threadsInPool, pendingSize);
    if (!pool) {
	fprintf(stderr, "ERROR creating thread pool\n");
	goto _cleanup;
    }

    fd_set set, tmpset;
    FD_ZERO(&set);
    FD_ZERO(&tmpset);

    FD_SET(listenfd, &set);        // aggiungo il listener fd al master set
    FD_SET(signal_pipe[0], &set);  // aggiungo il descrittore di lettura della signal_pipe

    // tengo traccia del file descriptor con id piu' grande
    int fdmax = (listenfd > signal_pipe[0]) ? listenfd : signal_pipe[0];



    volatile int quit = 0;
    sig_atomic_t stopNewConnections = 0; // true to stop new connections
    sig_atomic_t numberOfConnections = 0;

    while(!quit) {
	// copio il set nella variabile temporanea per la select
	tmpset = set;
	if (select(fdmax+1, &tmpset, NULL, NULL, NULL) == -1) {
	    perror("select");
	    goto _cleanup;
	}
	// cerchiamo di capire da quale fd abbiamo ricevuto una richiesta
	for(int i=0; i <= fdmax; ++i) {
	    if (FD_ISSET(i, &tmpset)) {
		long* connfd = malloc(sizeof(long));
		if (!connfd) {
		    // TODO logging utility
		    perror("ERROR FATAL malloc");
		    goto _cleanup;
		}
		if (i == listenfd) { // e' una nuova richiesta di connessione
		    if(stopNewConnections) { // non vogliamo nuove connessioni
			// TODO log
			FD_CLR(i, &set);
			close(i);
			continue;
		    }

		    // accetto la connessione nuova
		    if ((*connfd = accept(listenfd, (struct sockaddr*)NULL ,NULL)) == -1) {
			perror("accept");
			goto _cleanup;
		    }

		    // creo gli argomenti da passare al thread
		    void* args = NULL;

		    // aggiungo al threadpool
		    int r = addToThreadPool(pool, threadF, args);
		    if (r == 0)
			numberOfConnections++;
			continue; // aggiunto con successo
		    if (r < 0) // errore interno
			fprintf(stderr, "ERROR FATAL adding to the thread pool\n");
		    else // coda dei pendenti piena
			fprintf(stderr, "ERROR SERVER TOO BUSY\n");
		    free(args);
		    close(*connfd);
		    free(connfd);
		    continue;
		}
		if (i == signal_pipe[0]) { // controllo se devo terminare
		    int code;
		    if (readn(signal_pipe[0], &code, sizeof(int)) == -1) {
			perror("readn");
			break;
		    }
		    switch (code) {
		    case 0: { // stop alle connessioni
			stopNewConnections = 1;
			// TODO log
			if (numberOfConnections == 0) {
			    quit = 1;
			    // pthread_cancel(st);	// termino il signalThread
			}
			break;
		    }
		    case 1: { // stop immediato
			quit = 1;
			break;
		    }
		    default:
			perror("ERROR codice inviato dal sigThread invalido.\n");
			break;
		    }
		    break;
		}
		else { // richiesta di un client già connesso
		    FD_CLR(i, &set);

		    fdmax = (i>fdmax)?i:fdmax;

		    // creo gli argomenti da passare al thread

		    /*
		    int r = addToThreadPool(pool, serverThread, (void*) t);
		    if (r == 0)
			continue;
		    if (r < 0) // errore interno
			fprintf(stderr, "ERROR FATAL adding to the thread pool\n");
		    else // coda dei pendenti piena
			fprintf(stderr, "ERROR SERVER TOO BUSY\n");
		    */

		    continue;
		}
	    }
	}
    }

    destroyThreadPool(pool, 0); // notifico che i thread dovranno uscire

    // TODO print statistiche

    if (printQueue(queue) == -1) {
	perror("printQueue");
	return 1;
    }
    destroyQueue(queue);

    // aspetto la terminazione de signal handler thread
    pthread_join(sighandler_thread, NULL);

    unlink(SOCKNAME);

    printf("File Storage Server terminato.\n");
    fflush(stdout);
    return 0;


_cleanup:
    unlink(SOCKNAME);
    return -1;
}

// funzione eseguita dal signal handler thread
static void *sigHandler(void *arg) {
    sigset_t *set = ((sigHandler_t*)arg)->set;
    int fd_pipe   = ((sigHandler_t*)arg)->signal_pipe;

    if(pthread_sigmask(SIG_SETMASK, &set, NULL)) {
	fprintf(stderr, "ERROR setting mask\n");
	return (void*) 1;
    }

    while (1) {
	int sig;
	int code;
	int r = sigwait(&set, &sig);

	if (r != 0) {
	    errno = r;
	    perror("FATAL ERROR 'sigwait'");
	    return NULL;
	}

	switch (sig) {
	case SIGHUP:
	    code = 0;
	    // notifico il thread manager di smettere di accettare nuove connessioni in entrata
	    if (writen(fd_pipe, &code, sizeof(int)) == -1) {
		perror("writen");
	    }
	    break;
	case SIGINT:
	case SIGQUIT:
	    code = 1;
	    // notifico il thread manager di terminare il server il prima possibile
	    if (writen(fd_pipe, &code, sizeof(int)) == -1) {
		perror("writen");
	    }
	    return NULL;
	default:
	    break;
	}
    }
    return NULL;
}