123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558 |
- /*
- * Copyright (C) 2019 Weber Yann
- *
- * This file is part of PyFCGI.
- *
- * PyFCGI is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * any later version.
- *
- * PyFCGI is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with PyFCGI. If not, see <http://www.gnu.org/licenses/>.
- */
- #include "responder.h"
-
- /**@brief Exit after closing all stuff like semaphores
- * @ingroup work_master_proc */
- static void clean_exit(int status)
- {
- pyfcgi_IPC_close(IPC_WSTATE | IPC_WREQS | IPC_SEMST | IPC_SHMST);
- pyfcgi_IPC_destroy(IPC_WSTATE | IPC_WREQS | IPC_SEMST | IPC_SHMST);
- exit(status);
- }
-
- pid_t spawn_pool_handler()
- {
- pid_t res;
- struct sigaction act;
-
- act.sa_handler = pyfcgi_sighandler_drop;
- sigemptyset(&act.sa_mask);
- act.sa_flags = 0;
- act.sa_restorer = NULL;
-
- res = fork();
- if(res < 0)
- {
- pyfcgi_log(LOG_ALERT, "Failed to fork pool_handler : %s",
- strerror(errno));
- return -1;
- }
- if(!res)
- {
- if(sigaction(SIGINT, &act, NULL))
- {
- pyfcgi_log(LOG_WARNING,
- "Unable to sigaction SIGINT handler : %s",
- strerror(errno));
- }
- responder_loop();
- exit((unsigned char)-1);
- }
- return res;
- }
-
- void init_context()
- {
- PyFCGI_conf.context.pid = getpid();
- PyFCGI_conf.context.ppid = getppid();
-
- if(pyfcgi_IPC_create(IPC_WSTATE | IPC_WREQS | IPC_SEMST | IPC_SHMST) < 0)
- {
- pyfcgi_log(LOG_ALERT, "Pool handler process is unable to init IPC components");
- sleep(1);
- clean_exit(PYFCGI_FATAL);
- }
- if(sem_post(PyFCGI_SEM(SEM_STATS).sem) < 0)
- {
- pyfcgi_log(LOG_ALERT, "Unable to POST stat semaphore : %s",
- strerror(errno));
- clean_exit(PYFCGI_FATAL);
- }
- }
-
- int responder_loop()
- {
- unsigned int n_wrk, wanted_n, n;
- pid_t *wrk_pids;
- int err;
- int status;
- pid_t ret;
- /**@brief poll timeout */
- struct timespec timeout;
- /**@brief watchdog timeout */
- struct timespec pool_timeout;
- time_t idle_start, busy_start;
- short idle, busy;
- struct sigaction act;
- char *statusstr;
-
- act.sa_handler = pool_sighandler;
- sigemptyset(&act.sa_mask);
- sigaddset(&act.sa_mask, SIGTERM);
- act.sa_flags = 0;
- act.sa_restorer = NULL;
-
- if(sigaction(SIGTERM, &act, NULL))
- {
- pyfcgi_log(LOG_ALERT,
- "Sigaction error for SIGTERM pool process : %s",
- strerror(errno));
- exit(PYFCGI_FATAL);
- }
-
- timeout.tv_sec = 0;
- timeout.tv_nsec = 100000000;
- idle = busy = 0;
-
- pyfcgi_logger_set_ident("Workpool");
-
- if(PyFCGI_conf.pool_timeout)
- {
- pool_timeout.tv_nsec = 0;
- pool_timeout.tv_sec = PyFCGI_conf.pool_timeout;
- pyfcgi_wd_init(pool_wd_sighandler, &pool_timeout);
- }
-
- pyfcgi_log(LOG_INFO, "Preparing workers");
-
- init_context();
-
- pyfcgi_wd_arm();
-
- PyFCGI_conf.context.wrk_pids = &wrk_pids;
- PyFCGI_conf.context.n_wrk = 0;
- wrk_pids = malloc(sizeof(int) * PyFCGI_conf.max_wrk);
- if(!wrk_pids)
- {
- err = errno;
- pyfcgi_log( LOG_ALERT,
- "Unable to allocate memory for childs PID : %s",
- strerror(err));
- clean_exit(err);
- }
- bzero(wrk_pids, sizeof(int) * PyFCGI_conf.max_wrk);
-
- wanted_n = PyFCGI_conf.min_wrk;
- n_wrk = 0;
- // prespawning minimum worker count
- for(n_wrk=0; n_wrk < wanted_n; n_wrk++)
- {
- wrk_pids[n_wrk] = spawn(n_wrk);
- PyFCGI_conf.context.n_wrk = n_wrk;
- }
- //Wait at least for a process to be ready
- while(!pyfcgi_pool_idle(&timeout));
-
- // main loop, taking care to restart terminated workers,
- // spawn new one if needed, etc.
- while(1)
- {
- pyfcgi_wd_arm();
- PyFCGI_conf.context.n_wrk = n_wrk;
- pyfcgi_pool_shm_update(n_wrk);
- if( (ret = waitpid(0, &status, WNOHANG)) )
- {
- if(ret < 0)
- {
- //TODO : error
- }
- for(n=0; n<n_wrk; n++)
- {
- if(wrk_pids[n] == ret)
- {
- break;
- }
- }
- if(n == n_wrk)
- {
- pyfcgi_log(LOG_WARNING,
- "Child %d stopped but was notregistered",
- ret);
- continue;
- }
- if(WIFSIGNALED(status))
- {
- if(WTERMSIG(status) == 11)
- {
- pyfcgi_log(LOG_ALERT,
- "Worker[%d] segfault !",
- n);
- }
- else
- {
- pyfcgi_log(LOG_ALERT,
- "Worker[%d] terminated by signal %s(%d)",
- n, strsignal(WTERMSIG(status)),
- WTERMSIG(status));
- }
- }
- if(WEXITSTATUS(status))
- {
- statusstr = status2str(WEXITSTATUS(status));
- pyfcgi_log((WEXITSTATUS(status)&PYFCGI_FATAL)?
- LOG_ALERT:LOG_WARNING,
- "Worker[%d] exited with status %s",
- n, statusstr);
- free(statusstr);
- }
- if(!status)
- {
- pyfcgi_log(LOG_INFO,
- "Worker[%d] PID %d exited normally",
- n, wrk_pids[n]);
- }
-
- // respawn on same slot
- pyfcgi_log(LOG_DEBUG, "respawning worker #%d", n);
- wrk_pids[n] = spawn(n);
-
- continue;
- }
-
- // Check if the pool is idle or busy
- if(pyfcgi_pool_idle(&timeout))
- {
- // workers idle
- busy = 0;
- if(!idle)
- {
- idle = 1;
- idle_start = time(NULL);
- }
- else if((time(NULL) - idle_start) > PyFCGI_conf.worker_gc_timeout &&
- wanted_n > PyFCGI_conf.min_wrk
- && n_wrk - wanted_n < 2)
- {
- wanted_n--;
- idle = 0;
- }
- }
- else
- {
- idle = 0;
- if(!busy)
- {
- busy = 1;
- busy_start = time(NULL);
- }
- else if(time(NULL) - busy_start > 0 &&
- wanted_n < PyFCGI_conf.max_wrk)
- {
- pyfcgi_log( LOG_DEBUG,
- "All workers busy, spawning a new one");
- n = n_wrk;
- n_wrk++;
- wanted_n = n_wrk;
- wrk_pids[n] = spawn(n);
- if(!PyFCGI_conf.worker_fast_spawn)
- {
- busy_start = time(NULL);
- }
- }
- }
-
- // Stopping & deleting useless childs
- if(wanted_n < n_wrk && idle)
- { // need to shift the list and dec n_wrk
- busy = 0;
- n_wrk--;
- kill(wrk_pids[n_wrk], SIGTERM);
- nanosleep(&timeout, NULL);
- if( (ret = waitpid(wrk_pids[n_wrk], &status, WNOHANG)) < 0 )
- {
- pyfcgi_log(LOG_ERR, "Pool idle since %ds but unable to kill child %d (PID %d)",
- PyFCGI_conf.worker_gc_timeout,
- n_wrk, wrk_pids[n_wrk]);
- kill(wrk_pids[n_wrk], SIGKILL);
- }
- else
- {
- pyfcgi_log(LOG_INFO, "Pool idle since %ds : worker[%d](%d) killed",
- PyFCGI_conf.worker_gc_timeout,
- n_wrk, wrk_pids[n_wrk]);
- }
- idle = 0;
- continue;
- }
-
-
- nanosleep(&timeout, NULL);
- }
-
- pyfcgi_wd_arm();
- //Debug wait & exit
- for(; n_wrk != 0; n_wrk--)
- {
- waitpid(wrk_pids[n_wrk], &status, 0);
- pyfcgi_log(LOG_DEBUG, "Child %d stopped with status %d",
- wrk_pids[n_wrk], status);
- PyFCGI_conf.context.n_wrk = n_wrk;
- }
- //printf("Content-Type: text/html\r\n\r\nHello world !\n");
- pyfcgi_wd_stop();
- pyfcgi_log(LOG_INFO,"Child workers stoped, stopping responder");
- exit(0);
- }
-
- pid_t spawn(int wrk_id)
- {
- pid_t res;
- struct timespec wd_timeout;
- struct sigaction act;
- char ident[128];
-
- act.sa_handler = worker_sighandler;
- sigemptyset(&act.sa_mask);
- act.sa_flags = 0;
- act.sa_restorer = NULL;
-
-
- res = fork();
- if(res == -1)
- {
- pyfcgi_log(LOG_ERR, "Fork fails for worker #%d : %s",
- wrk_id, strerror(errno));
- return -1;
- }
- else if(!res)
- {
- // Child process
- PyFCGI_conf.context.ppid = PyFCGI_conf.context.pid;
- PyFCGI_conf.context.pid = getpid();
- snprintf(ident, 128, "Worker%2d", wrk_id);
- pyfcgi_logger_set_ident(ident);
- // Init IPC components
- if(pyfcgi_IPC_init(IPC_WSTATE | IPC_WREQS) < 0)
- {
- pyfcgi_log(LOG_ALERT, "Unable to initialize semaphore when spawning process...");
- exit(PYFCGI_FATAL);
- }
- // Set handler for SIGINT & SIGTERM
- /*
- if(sigaction(SIGINT, &(PyFCGI_conf.context.master_old_sigint),
- NULL))
- {
- pyfcgi_log(LOG_ALERT,
- "Sigaction error for worker process when restoring SIGINT handler: %s",
- strerror(errno));
- exit(PYFCGI_FATAL);
- }
- */
- if(sigaction(SIGTERM, &act, NULL))
- {
- pyfcgi_log(LOG_ALERT,
- "Sigaction error for worker process : %s",
- strerror(errno));
- exit(PYFCGI_FATAL);
- }
- // Set watchdog
- if(PyFCGI_conf.worker_timeout)
- {
- wd_timeout.tv_nsec = 0;
- wd_timeout.tv_sec = PyFCGI_conf.worker_timeout;
- pyfcgi_wd_init(worker_sigalrmhandler, &wd_timeout);
- }
-
- if(PyFCGI_conf.pep333)
- {
- exit(work333(wrk_id));
- }
- else
- {
- exit(work(wrk_id));
- }
- }
- pyfcgi_IPC_init(IPC_WSTATE | IPC_WREQS | IPC_SEMST);
- // Sleep to avoid spawning like hell thinking all workers are
- // busy. Let some time to this one to go up...
- // TODO: find a better way to avoid spawning to max_wrk
- //nanosleep(&timeout, NULL);
- pyfcgi_log( LOG_INFO,
- "Worker #%d spawned with PID %d", wrk_id, res);
- return res;
- }
-
- int pyfcgi_pool_state()
- {
- int err, res;
- if(sem_getvalue(PyFCGI_SEM(SEM_WSTATE).sem, &res) < 0)
- {
- err = errno;
- pyfcgi_log(LOG_ALERT, "Unable to read WSTATE semaphore value : %s",
- strerror(err));
- clean_exit(PYFCGI_FATAL);
- }
- return res;
- }
-
- int pyfcgi_pool_idle(const struct timespec *timeout)
- {
- int err;
- struct timespec abs_timeout;
-
- if(clock_gettime(CLOCK_REALTIME_COARSE, &abs_timeout) < 0)
- {
- //clock error
- pyfcgi_log(LOG_WARNING, "Unable to fetch asbtime for WSTATE sem_timedwait : %s",
- strerror(errno));
- }
- abs_timeout.tv_sec += timeout->tv_sec;
- if(abs_timeout.tv_nsec + timeout->tv_nsec > 999999999)
- {
- abs_timeout.tv_nsec = abs_timeout.tv_nsec + timeout->tv_nsec - 999999999;
- abs_timeout.tv_sec +=1;
- }
- else
- {
- abs_timeout.tv_nsec = timeout->tv_nsec;
- }
-
- if(sem_timedwait(PyFCGI_SEM(SEM_WSTATE).sem, &abs_timeout) < 0)
- {
- err = errno;
- switch(err)
- {
- case ETIMEDOUT:
- case EAGAIN:
- return 0; //busy
- case EINVAL:
- sleep(1);
- return 1;
- default:
- pyfcgi_log(LOG_ALERT, "Unable to wait WSTATE sem : %s",
- strerror(err));
- clean_exit(PYFCGI_FATAL);
- }
- }
- sem_post(PyFCGI_SEM(SEM_WSTATE).sem); //Hope no worker fails to set busy...
- return 1; //idle
- }
-
- void pool_sighandler(int signum)
- {
- unsigned int i, retry;
- int status, ret;
- struct timespec req;
-
- pyfcgi_log(LOG_NOTICE, "Received signal %s, cleaning & exiting...",
- strsignal(signum));
- if(PyFCGI_conf.context.n_wrk < 1) { clean_exit(0); }
-
- for(i=0; i<PyFCGI_conf.context.n_wrk; i++)
- {
- pyfcgi_log(LOG_INFO, "Sending SIGTERM to child #%d (pid %d)",
- i,(*PyFCGI_conf.context.wrk_pids)[i]);
- kill((*PyFCGI_conf.context.wrk_pids)[i], SIGTERM);
- }
- retry = i = 0;
- while(i<PyFCGI_conf.context.n_wrk)
- {
- ret = waitpid((*PyFCGI_conf.context.wrk_pids)[i], &status,
- WNOHANG);
- if(ret <= 0 && retry < 3)
- {
- retry++;
- req.tv_sec = 0;
- req.tv_nsec = 100000000; //0.1s
- nanosleep(&req, NULL);
- }
- else
- {
- if(retry < 3)
- {
- (*PyFCGI_conf.context.wrk_pids)[i] = 0;
- }
- retry = 0;
- i++;
- }
- }
- for(i=0; i<PyFCGI_conf.context.n_wrk; i++)
- {
- if((*PyFCGI_conf.context.wrk_pids)[i])
- {
- pyfcgi_log(LOG_INFO, "Sending SIGKILL to child %d", i);
- kill((*PyFCGI_conf.context.wrk_pids)[i], SIGKILL);
- }
-
- }
- clean_exit(0);
- }
-
- void pool_wd_sighandler(int signum)
- {
- unsigned int i;
- pyfcgi_log(LOG_ALERT, "Worker pool timeout ! Attempt to kill all childs");
- for(i=0; i<PyFCGI_conf.context.n_wrk; i++)
- {
- pyfcgi_log(LOG_ALERT, "Child[%d] PID %d", i, (*PyFCGI_conf.context.wrk_pids)[i]);
- kill((*PyFCGI_conf.context.wrk_pids)[i], SIGALRM);
- }
- while(PyFCGI_conf.context.n_wrk)
- {
- kill((*PyFCGI_conf.context.wrk_pids)[PyFCGI_conf.context.n_wrk], SIGALRM);
- PyFCGI_conf.context.n_wrk--;
- }
- pyfcgi_wd_stop();
- kill(PyFCGI_conf.context.pid, SIGTERM);
- clean_exit(PYFCGI_TIMEOUT);
- exit(PYFCGI_TIMEOUT);
- }
-
-
- void pyfcgi_pool_shm_update(int nworker)
- {
- short retry;
- int err;
- pyfcgi_stats_shm_t *data;
- struct timespec req;
-
- req.tv_sec = 0;
- req.tv_nsec = 10000000; //0.01s
-
- retry = 0;
- while(1)
- {
- if(sem_trywait(PyFCGI_SEM(SEM_STATS).sem) < 0)
- {
- err = errno;
- if(err == EAGAIN)
- {
- if(retry >= 5)
- {
- pyfcgi_log(LOG_ALERT,
- "Deadlock on SEM_STATS");
- clean_exit(PYFCGI_FATAL);
- }
- nanosleep(&req, NULL);
- continue;
- }
- pyfcgi_log(LOG_ALERT,
- "Unable to wait stats semaphore : %s",
- strerror(err));
- clean_exit(PYFCGI_FATAL);
- }
- break;
- }
-
- data = (pyfcgi_stats_shm_t*)PyFCGI_conf.shm.ptr;
- data->nworker = nworker;
- if(sem_getvalue(PyFCGI_SEM(SEM_WSTATE).sem, &(data->pool_load)) < 0)
- {
- data->pool_load = -1;
- }
-
- if(sem_post(PyFCGI_SEM(SEM_STATS).sem) < 0)
- {
- pyfcgi_log(LOG_ALERT, "Unable to post sem at shm update : %s",
- strerror(errno));
- clean_exit(PYFCGI_FATAL);
- }
- }
|