progettoso/src/server.c

#define _XOPEN_SOURCE 700
#include <signal.h>
#include <unistd.h>
#include <assert.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <pthread.h>
#include <ctype.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>
#include <taglialegna.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[]) {
    // 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);
    const char *buff;
    char *logFile=NULL; CONFGETSTR(logFile, config, "log", "logFile", buff);
    char *socketName=NULL; CONFGETSTR(socketName, config, "socket", "name", buff);
    int maxBacklog; CONFGETINT(maxBacklog, config, "socket", "backlog", NULL, 10);
    ini_free(config);

    sigset_t mask;
    sigfillset(&mask);
    sigdelset(&mask, SIGPIPE); // tolgo soltanto la sigpipe

    if (pthread_sigmask(SIG_SETMASK, &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;
    }

    // remove("mysock"); maybe necessary???


    // creo la struttura per il log
    taglia_t *taglia = taglia_init(logFile, 0);
    free(logFile); // free del nome del file
    if(taglia==NULL) {
	perror("taglia_init");
	goto _cleanup;
    }
    // risorse per il logfile usate nel main
    char buf[2048];
    int n;


    // pipes di comunicazione fra thread main e sigHandler thread/threadF threads
    int signal_pipe[2];
    int request_pipe[2];
    if (pipe(signal_pipe) == -1) {
	perror("pipe");
	goto _cleanup;
    }
    if (pipe(request_pipe) == -1) {
	perror("pipe");
	goto _cleanup;
    }



    // 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;
    }

    // scrivo sul log
    n = snprintf(buf, sizeof(buf), "Creato signal thread con id: %ld\n", (long) sighandler_thread);
    if( n<0 ) {
	perror("snprintf");
	goto _cleanup;
    }
    if( taglia_log(taglia, buf) < 0)
	goto _cleanup;


    // creo lock per operazioni su file
    pthread_mutex_t lock;
    if (pthread_mutex_init(&lock, NULL) != 0) {
	perror("pthread_mutex_init lock");
	goto _cleanup;
    }

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

    struct sockaddr_un serv_addr;
    memset(&serv_addr, '0', sizeof(serv_addr));
    serv_addr.sun_family = AF_UNIX;
    strncpy(serv_addr.sun_path, socketName, strlen(socketName)+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;
    }

    // scrivo sul log
    n = snprintf(buf, sizeof(buf), "Creato socket: %s\n", socketName);
    if( n<0 ) {
	perror("snprintf");
	goto _cleanup;
    }
    if( taglia_log(taglia, buf) < 0)
	goto _cleanup;


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

    // creo la lista dei client in attesa a una lock
    waiting_t *waiting = NULL;

    // creo la threadpool
    threadpool_t *pool = NULL;

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

    // scrivo sul log
    n = snprintf(buf, sizeof(buf), "Creato threadpool di dimensione %d e pending size %d\n", threadsInPool, pendingSize);
    if( n<0 ) {
	perror("snprintf");
	goto _cleanup;
    }
    if( taglia_log(taglia, buf) < 0)
	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
    FD_SET(request_pipe[0], &set); // aggiungo il descrittore di lettura della request_pipe

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


    // scrivo sul log
    n = snprintf(buf, sizeof(buf), "File Server ready.\n\tMaxFiles: %d\n\tMaxSize: %d\n", maxFiles, maxSize);
    if( n<0 ) {
	perror("snprintf");
	goto _cleanup;
    }
    if( taglia_log(taglia, buf) < 0)
	goto _cleanup;

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



    volatile int quit = 0;
    volatile int stopNewConnections = 0; // true to stop new connections
    volatile int 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(long i=0; i <= fdmax; ++i) {
	    if (FD_ISSET(i, &tmpset)) {
		long* connfd = malloc(sizeof(long));
		if (!connfd) {
		    perror("ERROR FATAL malloc");
		    goto _cleanup;
		}
		if (i == listenfd) { // e' una nuova richiesta di connessione
		    if(stopNewConnections) { // non vogliamo nuove connessioni
			// scrivo sul log
			if( taglia_log(taglia, "Nuova connessione rifiutata, server in terminazione\n") < 0) {
			    free(connfd);
			    goto _cleanup;
			}

			FD_CLR(i, &set);
			close(i);
			continue;
		    }

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


		    // scrivo sul log
		    n = snprintf(buf, sizeof(buf), "New client: %ld\n", *connfd);
		    if( n<0 ) {
			perror("snprintf");
			free(connfd);
			goto _cleanup;
		    }
		    if( taglia_log(taglia, buf) < 0) {
			free(connfd);
			goto _cleanup;
		    }

		    // creo gli argomenti da passare al thread
		    threadT* args = calloc(1, sizeof(threadT));
		    if(!args) {
			perror("ERROR FATAL calloc");
			free(connfd);
			goto _cleanup;
		    }
		    args->connfd = connfd;
		    args->quit = &quit;
		    args->request_pipe = request_pipe[1];
		    args->q = queue;
		    args->taglia = taglia;
		    args->pool = pool;
		    args->lock = &lock;
		    args->waiting = &waiting;

		    // 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 == request_pipe[0]) { // un worker ha finito di servire un client
		    int pdr; // ottengo il descrittore della pipe
		    if (readn(request_pipe[0], &pdr, sizeof(int)) == -1) {
			perror("readn");
			free(connfd);
			break;
		    }

		    switch (pdr) {
		    case -1: // client disconnected
			--numberOfConnections;
			if (stopNewConnections && numberOfConnections <= 0) {
			    quit = 1;
			    // termino il signalThread
			    pthread_cancel(sighandler_thread);
			    break;
			}
			continue;
		    default: // client served but not disconnected
			FD_SET(pdr, &set);
			if (pdr > fdmax) {
			    fdmax = pdr;
			}
			break;
		    }
		    continue;
		}
		if (i == signal_pipe[0]) { // controllo se devo terminare
		    int code;
		    if (readn(signal_pipe[0], &code, sizeof(int)) == -1) {
			perror("readn");
			free(connfd);
			break;
		    }
		    switch (code) {
		    case 0: { // stop alle connessioni
			stopNewConnections = 1;

			// scrivo sul log
			if( taglia_log(taglia, "Stop new connections\n") < 0){
			    free(connfd);
			    goto _cleanup;
			}
			if (numberOfConnections == 0) {
			    quit = 1;
			    // termino il signalThread
			    pthread_cancel(sighandler_thread);
			}
			break;
		    }
		    case 1: { // stop immediato
			quit = 1;
			// scrivo sul log
			if( taglia_log(taglia, "Immediate quit\n") < 0) {
			    free(connfd);
			    goto _cleanup;
			}
			break;
		    }
		    default:
			perror("ERROR codice inviato dal sigThread invalido.\n");
			break;
		    }
		    free(connfd);
		    break;
		}
		else { // richiesta di un client già connesso
		    FD_CLR(i, &set);

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

		    // creo gli argomenti da passare al thread
		    threadT* args = calloc(1, sizeof(threadT));
		    if(!args) {
			perror("ERROR FATAL calloc");
			free(connfd);
			goto _cleanup;
		    }
		    *connfd = i;
		    args->connfd = connfd;
		    args->quit = &quit;
		    args->request_pipe = request_pipe[1];
		    args->q = queue;
		    args->taglia = taglia;
		    args->pool = pool;
		    args->lock = &lock;
		    args->waiting = &waiting;

		   // 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;
		}
	    }
	}
    }
    fprintf(stdout, "\n");

    destroyThreadPool(pool, 0); // notifico che i thread dovranno uscire
    clearWaiting(&waiting); // destroy waiting list
    taglia_stats(taglia, stdout); // print stats

    // print all files in storage during shutdown
    if (printQueue(stdout, queue) == -1) {
	perror("printQueue");
	return 1;
    }
    destroyQueue(queue);

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

    taglia_del(taglia);

    unlink(socketName);

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

_cleanup:
    unlink(socketName);
    free(socketName);
    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;
}