123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614 |
- /*
- * 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 "pyworker.h"
-
-
- /**@brief Indicate that a worker is idle
- * @param int semid */
- static inline void worker_set_idle(int);
- /**@brief Indicate that a worker is busy
- * @param int semid */
- static inline void worker_set_busy(int);
-
- /**@brief Process results from a pep333 worker
- * @param FCGX_Stream* out stream from libFCGI
- * @param PyObject* application function returned value
- */
- static inline int work333_send_result(FCGX_Stream*, PyObject* ret);
-
-
- static int worker_piper_sigrcv = 0;
-
- int work333(int wrk_id, int semid)
- {
- PyObject *entry_fun, *pyflush[2], *py_osmod, *entry_ret, *environ,
- *start_response, *args;
- FCGX_Stream *in_stream, *out_stream, *err_stream;
- char **envp;
- int count, pipe_out[2], pipe_err[2];
- int max_reqs;
- char ident[128];
- struct timeval start, stop;
-
- max_reqs = PyFCGI_conf.max_reqs;
-
- snprintf(ident, 128, "Worker[%d]", wrk_id);
- pyfcgi_logger_set_ident(ident);
-
- pyfcgi_log(LOG_INFO, "Worker started with PEP333 App");
-
- pyinit();
- pyfcgi_log(LOG_DEBUG, "Python started");
- update_python_fd(pipe_out, pipe_err);
- fetch_pyflush(&(pyflush[0]), &(pyflush[1]));
-
- //importing os
- py_osmod = python_osmod();
- // loading module
- entry_fun = import_entrypoint();
-
- pyfcgi_log(LOG_INFO, "Waiting request with %s.%s()",
- PyFCGI_conf.py_entrymod, PyFCGI_conf.py_entryfun);
-
- worker_set_idle(semid); //before failing on import
-
- if(!entry_fun) //but exit if import failed
- {
- pyfcgi_log(LOG_ALERT, "Unable to import entrypoint");
- exit(PYFCGI_FATAL);
-
- }
-
- start_response = get_start_response();
-
- // requests accepting loop
- count = 0;
- while ((!count || count != max_reqs) &&
- FCGX_Accept(&in_stream, &out_stream, &err_stream, &envp) >= 0)
- {
- gettimeofday(&start, NULL);
- worker_set_busy(semid);
- count++;
- environ = update_pyenv(py_osmod, envp);
-
- libpyfcgi_clean_response();
- libpyfcgi.out = out_stream;
- libpyfcgi.in = in_stream;
- args = Py_BuildValue("OO", environ, start_response);
- // Call application function
- entry_ret = PyObject_CallObject(entry_fun, args);
- if(entry_ret && entry_ret != Py_None )
- {
- Py_INCREF(entry_ret);
- }
-
- if(PyErr_Occurred())
- {
- pyfcgi_log(LOG_ERR, "PEP333 entrypoint function triggered an exception");
- log_expt(LOG_ERR);
- }
- // able to process returned value
- // Simulate python call of libpyfcgi.write_body()
- if(entry_ret && entry_ret != Py_None)
- {
- _pyfcgi_write_body(entry_ret);
- if(PyErr_Occurred())
- {
- log_expt(LOG_ERR);
- }
- Py_DECREF(entry_ret);
- }
-
- // clean stuffs
- Py_DECREF(args);
- Py_DECREF(environ);
- // flush & logs pystdout & pystderr
- worker_log_pipes(pipe_out[0], pipe_err[0], pyflush);
-
- FCGX_FClose(out_stream);
- FCGX_FClose(in_stream);
- FCGX_FClose(err_stream);
- FCGI_Finish();
- worker_set_idle(semid);
- gettimeofday(&stop, NULL);
- stop.tv_sec = stop.tv_sec - start.tv_sec;
- stop.tv_usec = stop.tv_usec - start.tv_usec;
- if(stop.tv_usec < 0)
- {
- stop.tv_usec += 1000000;
- stop.tv_sec -= 1;
- }
- pyfcgi_log(LOG_DEBUG, "Worker[%d] request %d END [OK] %lu bytes in %ld.%06lds",
- wrk_id, count, libpyfcgi.rep_sz, stop.tv_sec, stop.tv_usec);
- }
- return 0;
- }
-
- static inline int work333_send_result(FCGX_Stream *out, PyObject* ret)
- {
- return 0;
- }
-
- int work(int wrk_id, int semid)
- {
- PyObject *entry_fun, *pystdout_flush, *pystderr_flush, *py_osmod,
- *environ;
- FCGX_Stream *in_stream, *out_stream, *err_stream;
- char **envp;
- int count, pipe_out[2], pipe_err[2], pipe_ctl[2], err, piper_status;
- int max_reqs;
- struct sigaction act;
- struct timeval start, stop;
- sigset_t emptyset;
- char buf[PIPE_BUF];
- size_t rep_sz;
- piper_args_t piper_args;
- char *piper_stack;
-
- char ident[128];
- snprintf(ident, 128, "Worker[%d]", wrk_id);
- pyfcgi_logger_set_ident(ident);
-
- max_reqs = PyFCGI_conf.max_reqs;
- piper_args.wrk_id = wrk_id;
- piper_args.act = &act;
-
- // preparing sigaction for piper
- if(sigemptyset(&emptyset))
- {
- err = errno;
- pyfcgi_log( LOG_ALERT, "sigemptyset fails in piper : %s",
- strerror(err));
- exit(err);
- }
- act.sa_handler = worker_piper_sighandler;
- act.sa_mask = emptyset;
- //act.sa_flags = SA_RESTART;
- act.sa_flags = 0;
- act.sa_restorer = NULL;
-
-
- if( !(piper_stack = malloc(PIPER_STACK_SZ)) )
- {
- err = errno;
- pyfcgi_log(LOG_ALERT, "Error while allocating piper stack : %s",
- strerror(err));
- exit(err);
- }
-
- pyfcgi_log(LOG_INFO, "Worker %d started", wrk_id);
-
- update_python_path(); // add cwd to python path
- Py_Initialize(); // "start" python
- update_python_fd(pipe_out, pipe_err);
- piper_args.pystdout = pipe_out[0];
- piper_args.pystderr = pipe_err[0];
-
- fetch_pyflush(&pystdout_flush, &pystderr_flush);
- pyfcgi_log( LOG_INFO,
- "Worker[%d] Python started", wrk_id);
-
- //importing os
- py_osmod = python_osmod();
- // loading module
- entry_fun = import_entrypoint();
-
- pyfcgi_log( LOG_INFO,
- "Worker[%d] Waiting request with %s.%s()", wrk_id,
- PyFCGI_conf.py_entrymod, PyFCGI_conf.py_entryfun);
-
-
- worker_set_idle(semid); //before failing on import
-
- if(!entry_fun) //but exit if import failed
- {
- pyfcgi_log(LOG_ALERT, "Unable to import entrypoint");
- exit(PYFCGI_FATAL);
-
- }
-
- if(pipe(pipe_ctl) == -1)
- {
- err = errno;
- pyfcgi_log(LOG_ALERT,
- "Worker[%d] Pipe fails for piper ctl : %s",
- wrk_id, strerror(err));
- exit(err);
- }
- piper_args.ctl_pipe = pipe_ctl[1];
-
- count = 0;
- while ((!count || count != max_reqs) && FCGX_Accept(&in_stream, &out_stream, &err_stream, &envp) >= 0)
- {
- gettimeofday(&start, NULL);
- worker_set_busy(semid);
-
- count++;
- piper_args.out = out_stream;
- //piper_args.req_id = count;
- worker_piper_sigrcv = 0;
- // if(!PyFCGI_conf.pep333)
- /**@todo avoid clone on each request... (using named pipes ?) */
- pid_t pid = clone(worker_piper, piper_stack + PIPER_STACK_SZ - 1,
- CLONE_CHILD_CLEARTID | CLONE_CHILD_SETTID | \
- SIGCHLD | \
- CLONE_FILES | CLONE_FS | CLONE_IO | CLONE_VM,
- &piper_args);
- if(pid < 0)
- {
- err = errno;
- pyfcgi_log(LOG_ALERT,
- "Worker[%d] req #%d Fork failed for piper : %s",
- wrk_id, count, strerror(err));
- exit(err);
- }
- environ = update_pyenv(py_osmod, envp);
- Py_DECREF(environ);
- //close(pipe_ctl[1]);
- PyObject_CallObject(entry_fun, NULL);
- if(PyErr_Occurred())
- {
- log_expt(LOG_ERR);
- }
- else
- {
- /*
- pyfcgi_log(LOG_DEBUG, "Worker[%d] request %d funcall [OK]",
- wrk_id, count);
- */
- }
- PyObject_CallObject(pystdout_flush, NULL);
- PyObject_CallObject(pystderr_flush, NULL);
- read(pipe_ctl[0], &buf, 1); // unblock when child ready
- FCGX_FClose(in_stream);
- FCGX_FClose(err_stream);
- //close(pipe_ctl[0]);
-
- kill(pid, WPIPER_SIG); //indicate child python call ended
- waitpid(pid, &piper_status, 0);
- if(WIFSIGNALED(piper_status))
- {
- pyfcgi_log(LOG_ERR,
- "Woker[%d] req #%d piper terminated by sig %d",
- wrk_id, count, WTERMSIG(piper_status));
- exit(40);
- }
- else if(WEXITSTATUS(piper_status))
- {
- pyfcgi_log(LOG_ERR,
- "Woker[%d] req #%d piper exited with error status %d",
- wrk_id, count, WEXITSTATUS(piper_status));
- exit(41);
- }
-
- FCGI_Finish();
- if(ctl_get_rep_sz(pipe_ctl[0], &rep_sz))
- {
- rep_sz = 0;
- }
- //Increase sem showing the worker is idle
- worker_set_idle(semid);
- gettimeofday(&stop, NULL);
- stop.tv_sec = stop.tv_sec - start.tv_sec;
- stop.tv_usec = stop.tv_usec - start.tv_usec;
- if(stop.tv_usec < 0)
- {
- stop.tv_usec += 1000000;
- stop.tv_sec -= 1;
- }
- pyfcgi_log(LOG_DEBUG, "Worker[%d] request %d END [OK] %lu bytes in %ld.%06lds",
- wrk_id, count, rep_sz, stop.tv_sec, stop.tv_usec);
- }
- free(piper_stack);
- Py_Exit(count == max_reqs ?0:EXIT_PYERR);
- }
-
- /*
- void worker_piper(int wrk_id, int req_id, int pystdout, int pystderr,
- int ctl_pipe, FCGX_Stream* out)
- */
- int worker_piper(void *ptr)
- {
- piper_args_t *args;
- struct pollfd fds[2];
- int err, ret, cont, poll_ret;
- short revents, closed, nfds;
- char buf[PIPE_BUF];
- size_t out_sz;
-
- args = ptr;
- int wrk_id, pystdout, pystderr, ctl_pipe;
- //int req_id;
- FCGX_Stream *out;
- struct sigaction *act;
- wrk_id = args->wrk_id;
- //req_id = args->req_id;
- pystdout = args->pystdout;
- pystderr = args->pystderr;
- ctl_pipe = args->ctl_pipe;
- out = args->out;
- act = args->act;
-
- if(sigaction(WPIPER_SIG, act, NULL))
- {
- err = errno;
- pyfcgi_log( LOG_ALERT, "Unable to sigaction in piper : %s",
- strerror(err));
- exit(err);
- }
- write(ctl_pipe, " ", 1);
-
-
- //pyfcgi_log(LOG_DEBUG, "Worker[%d] req #%d piper", wrk_id, req_id);
-
- fds[0].fd = pystderr;
- fds[1].fd = pystdout;
- fds[0].events = fds[1].events = POLLIN;
- fds[0].revents = fds[1].revents = 0;
- nfds = 2;
-
-
- closed = out_sz = 0;
- cont = 2;
- while(cont)
- {
- poll_ret = poll(fds, nfds, 0);
- //pyfcgi_log(LOG_DEBUG, "Worker[%d] req #%d poll_ret = %d", wrk_id, req_id, poll_ret);
- if(poll_ret < 0)
- {
- err = errno;
- if(err == EINTR)
- {
- continue;
- }
- pyfcgi_log( LOG_WARNING, "Poll error : %s",
- strerror(err));
- fds[0].revents = fds[1].revents = 0;
- continue;
- }
- if(!poll_ret && worker_piper_sigrcv)
- {
- cont--;
- continue;
- }
- if(poll_ret && (revents = fds[1].revents))
- {
- /*
- pyfcgi_log(LOG_DEBUG, "Worker[%d] req #%d STDOUT evt !",
- wrk_id, req_id);
- */
- poll_ret--;
- if(revents & POLLIN)
- {
- /*
- pyfcgi_log(LOG_DEBUG, "Worker[%d] req #%d POLLIN STDOUT !",
- wrk_id, req_id);
- */
- ret = read(pystdout, buf, PIPE_BUF);
- /*
- pyfcgi_log(LOG_DEBUG, "Worker[%d] req #%d read(stdout) ret %d",
- wrk_id, req_id, ret);
- */
- if(ret < 0)
- {
- err = errno;
- if(err == EINTR)
- {
- continue;
- }
- pyfcgi_log( LOG_ERR,
- "Error reading python stdout : %s",
- strerror(err));
- exit(err);
- }
- //buf[ret] = '\0';
- ret = FCGX_PutStr(buf, ret, out);
- out_sz += ret;
- if(ret < PIPE_BUF && worker_piper_sigrcv)
- {
- FCGX_FClose(out);
- FCGI_Finish();
- write(ctl_pipe, &out_sz, sizeof(size_t));
- closed = 1;
- nfds = 1;
- }
- }
- //TODO handle other poll events
- }
- if(poll_ret && (revents = fds[0].revents))
- {
- /*
- pyfcgi_log(LOG_DEBUG, "Worker[%d] req #%d STDERR evt !",
- wrk_id, req_id);
- */
- if(revents & POLLIN)
- {
- /*
- pyfcgi_log(LOG_DEBUG, "Worker[%d] req #%d POLLIN STDERR !",
- wrk_id, req_id);
- */
- while(1)
- {
- ret = read(pystderr, buf, PIPE_BUF);
- if(ret < 0)
- {
- err = errno;
- if(err == EINTR)
- {
- continue;
- }
- pyfcgi_log( LOG_ERR,
- "Error reading python stdout : %s",
- strerror(err));
- exit(err);
- }
- buf[ret] = '\0';
- pyfcgi_log(LOG_ERR,
- "Worker[%d] Python error : %s",
- wrk_id, buf);
- break;
- }
- }
- //TODO handle other poll events
- }
- fds[0].revents = fds[1].revents = 0;
- }
- /*
- pyfcgi_log(LOG_DEBUG, "Worker[%d] req #%d piper exiting...",
- wrk_id, req_id);
- */
- if(!closed)
- {
- FCGX_FClose(out);
- FCGI_Finish();
- write(ctl_pipe, &out_sz, sizeof(size_t));
- }
- return 0;
- }
-
- void worker_log_pipes(int pipe_std, int pipe_err, PyObject *flush[2])
- {
- size_t ret;
- char buff[4096];
- int i, poll_ret;
- int loglevels[2] = {LOG_INFO, LOG_WARNING};
- struct pollfd fds[2];
- short isflush[2] = {0,0};
- char *outnames[2] = {"sys.stdout", "sys.stderr"};
-
- fds[0].fd = pipe_std;
- fds[1].fd = pipe_err;
- fds[0].events = fds[1].events = POLLIN;
- fds[0].revents = fds[1].revents = 0;
-
- for(i=0; i<2; i++)
- {
- do
- {
- poll_ret = poll(&(fds[i]), 1, 0);
- if(poll_ret)
- {
- ret = read(pipe_std, buff, 4096);
- if(ret == -1)
- {
- pyfcgi_log(LOG_ERR, "Error reading python %s pipe : %s",
- outnames[i], strerror(errno));
- exit(PYFCGI_WORKER_FAIL);
- }
- pyfcgi_log(loglevels[i], "Pyton %s : '%s'",
- outnames[i], buff);
- }
- if(!isflush[i] && (!poll_ret || ret < 4096))
- {
- isflush[i] = 1;
- PyObject_CallObject(flush[i], NULL);
- if(PyErr_Occurred())
- {
- pyfcgi_log(LOG_ERR, "Exception while flushing %s",
- outnames[i]);
- log_expt(LOG_ERR);
- exit(PYFCGI_WORKER_FAIL);
- }
- }
- }while(poll_ret && !isflush[i]);
- }
- }
-
- void worker_piper_sighandler(int signum)
- {
- worker_piper_sigrcv = 1;
- //pyfcgi_log(LOG_DEBUG, "SIG");
- }
-
- int ctl_get_rep_sz(int ctl_pipe, size_t* rep_sz)
- {
- struct pollfd fds;
- int ret, err;
-
- fds.fd = ctl_pipe;
- fds.events = POLLIN;
- fds.revents = 0;
-
- if( (ret = poll(&fds, 1, 0)) < 0)
- {
- err = errno;
- pyfcgi_log(LOG_ERR, "Failed to poll ctl pipe : %s",
- strerror(err));
- return -1;
- }
- if(!ret)
- {
- pyfcgi_log(LOG_ERR, "No data in ctl pipe for rep_sz...");
- return -1;
- }
- ret = read(ctl_pipe, rep_sz, sizeof(size_t));
- if(ret < 0)
- {
- pyfcgi_log(LOG_ERR, "Error reading ctl pipe : %s",
- strerror(errno));
- return -1;
- }
- if(ret < sizeof(size_t))
- {
- pyfcgi_log(LOG_ERR, "Incomplete read from ctl pipe, no rep_sz...");
- return -1;
- }
- return 0;
- }
-
- static void worker_set_idle(int semid)
- {
- int err;
- struct sembuf sop;
-
- sop.sem_num = 0;
- sop.sem_flg = 0;
- sop.sem_op = 1;
-
- if(semop(semid, &sop, 1) < 0)
- {
- err = errno;
- pyfcgi_log(LOG_ERR, "error incrementing the semaphore : %s",
- strerror(err));
- }
- }
-
- static void worker_set_busy(int semid)
- {
- int err;
- struct sembuf sop;
-
- sop.sem_num = 0;
- sop.sem_flg = 0;
- sop.sem_op = -1;
-
- if(semop(semid, &sop, 1) < 0)
- {
- err = errno;
- pyfcgi_log(LOG_ERR, "error incrementing the semaphore : %s",
- strerror(err));
- }
- }
-
- void worker_sighandler(int signum)
- {
- pyfcgi_log(LOG_INFO, "%s signal received, exiting...", strsignal(signum));
- exit(0);
- }
|