#define _XOPEN_SOURCE 700 #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include /** * @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 \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); queueT *queue = NULL; taglia_t *taglia = NULL; 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_beforesigthread; } // 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_beforesigthread; } // remove("mysock"); maybe necessary??? // creo la struttura per il log taglia = taglia_init(logFile, 0); free(logFile); // free del nome del file if(taglia==NULL) { perror("taglia_init"); goto _cleanup_beforesigthread; } // 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_beforesigthread; } if (pipe(request_pipe) == -1) { perror("pipe"); goto _cleanup_beforesigthread; } // 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 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); free(connfd); 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++; free(connfd); 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"); close(*connfd); free(connfd); continue; } if (i == request_pipe[0]) { // un worker ha finito di servire un client long pdr; // ottengo il descrittore della pipe if (readn(request_pipe[0], &pdr, sizeof(long)) == -1) { perror("readn"); free(connfd); break; } switch (pdr) { case -1: // client disconnected --numberOfConnections; n = snprintf(buf, sizeof(buf), "Client %ld disconnected.\n", pdr); if( n<0 ) { perror("snprintf"); free(connfd); goto _cleanup; } if( taglia_log(taglia, buf) < 0) { free(connfd); goto _cleanup; } if (stopNewConnections && numberOfConnections <= 0) { quit = 1; // termino il signalThread 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]) { // 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; } 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; // aggiungo al threadpool int r = addToThreadPool(pool, threadF, args); if (r == 0) { free(connfd); 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"); 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: 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; } // 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; }