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>

typedef struct {
    sigset_t  *set;         // set of signals to handle (masked)
    int       signal_pipe;  // unnamed pipe's descriptor
} sigHandler_t;

// signal handler thread function
static void *sigHandler(void *arg);

static void usage(const char *argv0) {
    fprintf(stderr, "Uso: %s <path al file di config>\n", argv0);
}

static void checkargs(int argc, char* argv[]) {
    if (argc != 2) {
	usage(argv[0]);
	_exit(EXIT_FAILURE);
    }
    ini_t *config = ini_load(argv[1]);
    if (config == NULL) {
	fprintf(stderr, "Error: reading 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);

    queueT *queue = NULL;
    taglia_t *taglia = NULL;

    sigset_t mask;
    sigfillset(&mask);
    sigdelset(&mask, SIGPIPE); // only sigpipe is excluded

    if (pthread_sigmask(SIG_SETMASK, &mask, NULL) != 0) {
	fprintf(stderr, "Error: setting mask.\n");
	goto _cleanup_beforesigthread;
    }
    struct sigaction s;
    memset(&s, 0, sizeof(s));
    s.sa_handler = SIG_IGN;
    if ( (sigaction(SIGPIPE,&s,NULL)) == -1 ) {
	perror("main: sigaction");
	goto _cleanup_beforesigthread;
    }

    (void) unlink(socketName);

    // create structure for logging
    taglia = taglia_init(logFile, 0);
    free(logFile); // free the name of the file
    if(taglia==NULL) {
	perror("main: taglia_init");
	goto _cleanup_beforesigthread;
    }
    // buffer for loggin
    char buf[2048];
    int n;

    // pipes for comunication between main thread, sigHandler thread
    // and worker threads
    int signal_pipe[2];
    int request_pipe[2];
    if (pipe(signal_pipe) == -1) {
	perror("main: pipe, signal_pipe");
	goto _cleanup_beforesigthread;
    }
    if (pipe(request_pipe) == -1) {
	perror("main: pipe, request_pipe");
	goto _cleanup_beforesigthread;
    }


    // thread for handling interruptions
    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;
    }

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


    // lock for operations on files
    pthread_mutex_t lock;
    if (pthread_mutex_init(&lock, NULL) != 0) {
	perror("main: pthread_mutex_init");
	goto _cleanup;
    }

    // create socket
    int listenfd;
    if ((listenfd = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) {
	perror("main: 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("main: bind");
	goto _cleanup;
    }
    if (listen(listenfd, maxBacklog) == -1) {
	perror("main: listen");
	goto _cleanup;
    }

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


    // create queue
    queue = createQueue(maxFiles, maxSize);

    // create queue for clients waiting on a lock
    waiting_t *waiting = NULL;

    // create threadpool
    threadpool_t *pool = NULL;

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

    // write to logfile
    n = snprintf(buf, sizeof(buf), "Creato threadpool di dimensione %d e pending size %d\n", threadsInPool, pendingSize);
    if(n<0) {
	perror("main: snprintf");
	goto _cleanup;
    }
    if(taglia_log(taglia, buf) < 0)
	goto _cleanup;


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

    // add fd for socket, signal pipe and request pipe.
    FD_SET(listenfd, &set);
    FD_SET(signal_pipe[0], &set);
    FD_SET(request_pipe[0], &set);

    // get max file descriptor
    int fdmax = (listenfd > signal_pipe[0]) ? listenfd : signal_pipe[0];
    fdmax = (fdmax > request_pipe[0]) ? fdmax : request_pipe[0];


    // write to logfile
    n = snprintf(buf, sizeof(buf), "File Server ready.\n\tMaxFiles: %d\n\tMaxSize: %d\n", maxFiles, maxSize);
    if( n<0 ) {
	perror("main: 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;
    volatile int numberOfConnections = 0;

    while(!quit) {
	// copy the set in the tmp variable for the select
	tmpset = set;
	if (select(fdmax+1, &tmpset, NULL, NULL, NULL) == -1) {
	    perror("main: select");
	    goto _cleanup;
	}
	// search for the ready fd
	for(long i=0; i <= fdmax; ++i) {
	    if (FD_ISSET(i, &tmpset)) {
		long* connfd = malloc(sizeof(long));
		if (!connfd) {
		    perror("main: malloc");
		    goto _cleanup;
		}
		if (i == listenfd) { // new request for connection
		    if(stopNewConnections) { // no new connections allowed
			// write to logfile
			if( taglia_log(taglia, "Nuova connessione rifiutata, server in terminazione\n") < 0) {
			    free(connfd);
			    goto _cleanup;
			}

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

		    // accept new connection
		    if ((*connfd = accept(listenfd, (struct sockaddr*)NULL ,NULL)) == -1) {
			perror("main: accept");
			free(connfd);
			goto _cleanup;
		    }

		    // write to logfile
		    n = snprintf(buf, sizeof(buf), "Nuovo client: %ld\n", *connfd);
		    if( n<0 ) {
			perror("main: snprintf");
			free(connfd);
			goto _cleanup;
		    }
		    if(taglia_log(taglia, buf) < 0) {
			free(connfd);
			goto _cleanup;
		    }

		    // create args to pass to the worker
		    threadT* args = calloc(1, sizeof(threadT));
		    if(!args) {
			perror("main: 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;

		    // add to threadpool
		    int r = addToThreadPool(pool, threadF, args);
		    if (r == 0) { // no errors
			numberOfConnections++;
			free(connfd);
			continue;
		    }
		    if (r < 0) // internal error
			fprintf(stderr, "Error: adding to the thread pool.\n");
		    else // pending queue full
			fprintf(stderr, "Error: server too busy.\n");
		    close(*connfd);
		    free(connfd);
		    continue;
		}
		if (i == request_pipe[0]) { // worker finished task
		    long pdr; // get pipe descriptor
		    if (readn(request_pipe[0], &pdr, sizeof(long)) == -1) {
			perror("main: readn");
			free(connfd);
			break;
		    }
		    switch (pdr) {
		    case -1: // client disconnected
			--numberOfConnections;
			if (stopNewConnections && numberOfConnections <= 0) {
			    // quitting and terminating signalThread
			    quit = 1;
			    pthread_cancel(sighandler_thread);
			    break;
			}
			free(connfd);
			continue;
		    default: // client served but not disconnected
			FD_SET(pdr, &set);
			if (pdr > fdmax) {
			    fdmax = pdr;
			}
			break;
		    }
		    free(connfd);
		    continue;
		}
		if (i == signal_pipe[0]) { // check if we need to terminate
		    int code;
		    if (readn(signal_pipe[0], &code, sizeof(int)) == -1) {
			perror("main: readn");
			free(connfd);
			break;
		    }
		    switch (code) {
		    case 0: { // stop to new connections
			stopNewConnections = 1;

			// write to logfile
			if(taglia_log(taglia, "Stop new connections\n") < 0){
			    free(connfd);
			    goto _cleanup;
			}
			if(numberOfConnections == 0) {
			    quit = 1;
			    // stop signalThread
			    pthread_cancel(sighandler_thread);
			}
			break;
		    }
		    case 1: { // immediate stop
			quit = 1;
			// write to logfile
			if( taglia_log(taglia, "Immediate quit\n") < 0) {
			    free(connfd);
			    goto _cleanup;
			}
			break;
		    }
		    default:
			perror(NULL);
			fprintf(stderr, "Error: invalid code recived from sigThread.\n");
			break;
		    }
		    free(connfd);
		    break;
		}
		else { // request from an already connected client
		    FD_CLR(i, &set);

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

		    // create args for worker thread
		    threadT* args = calloc(1, sizeof(threadT));
		    if(!args) {
			perror("main: calloc");
			free(connfd);
			goto _cleanup;
		    }
		    args->connfd = i;
		    args->quit = &quit;
		    args->request_pipe = request_pipe[1];
		    args->q = queue;
		    args->taglia = taglia;
		    args->pool = pool;
		    args->lock = &lock;
		    args->waiting = &waiting;

		   // add to the threadpool
		    int r = addToThreadPool(pool, threadF, args);
		    if (r == 0) {
			free(connfd);
			continue;
		    }
		    if (r < 0) // internal error
			fprintf(stderr, "Error: adding to the thread pool.\n");
		    else // pending queue full
			fprintf(stderr, "Error: server too busy.\n");
		    close(*connfd);
		    free(connfd);
		    continue;
		}
	    }
	}
    }
    fprintf(stdout, "\n");

    destroyThreadPool(pool, 0); // notify the threads that need to stop
    clearWaiting(&waiting); // destroy waiting list
    taglia_stats(taglia, stdout); // print stats

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

    // wait for sigHandler thread
    pthread_join(sighandler_thread, NULL);

    // free log file structure
    taglia_del(taglia);

    // unlink socket
    unlink(socketName);

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

_cleanup:
    pthread_cancel(sighandler_thread);
    pthread_join(sighandler_thread, NULL);
_cleanup_beforesigthread:
    if(queue) {
	destroyQueue(queue);
    }
    if(taglia) {
	taglia_del(taglia);
    }
    unlink(socketName);
    free(socketName);
    return -1;
}

// signal handler thread function
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("sigHandler: sigwait");
	    return NULL;
	}

	switch (sig) {
	case SIGHUP:
	    code = 0;
	    // notify main thread to stop new connections
	    if (writen(fd_pipe, &code, sizeof(int)) == -1) {
		perror("sigHandler: writen");
	    }
	    break;
	case SIGINT:
	case SIGQUIT:
	    code = 1;
	    // notify main thread to stop immediatly
	    if (writen(fd_pipe, &code, sizeof(int)) == -1) {
		perror("sigHandler: writen");
	    }
	    return NULL;
	default:
	    break;
	}
    }
    return NULL;
}