From 36c8a8c6750d5e4a58fb6a9e170d73dbe103eaef Mon Sep 17 00:00:00 2001 From: "roberto@mrspurr" Date: Sun, 3 Oct 2010 22:49:22 +0200 Subject: [PATCH] initial threading mode support --- async.c | 37 +++++++++------------ loop.c | 30 +++++++++++++++++ master.c | 25 ++++++-------- utils.c | 8 +++++ uwsgi.c | 84 ++++++++++++++++++++++++++++++++++-------------- uwsgi.h | 19 +++++++++-- uwsgi_pymodule.c | 4 +-- uwsgiconfig.py | 2 +- wsgi_handlers.c | 36 +++++++++------------ 9 files changed, 156 insertions(+), 89 deletions(-) create mode 100644 loop.c diff --git a/async.c b/async.c index 30baed8a..4e3774d6 100644 --- a/async.c +++ b/async.c @@ -267,19 +267,10 @@ int async_del(int queuefd, int fd, int etype) { #endif -inline struct wsgi_request *next_wsgi_req(struct wsgi_request *wsgi_req) { - - uint8_t *ptr = (uint8_t *) wsgi_req ; - - ptr += sizeof(struct wsgi_request) ; - - return (struct wsgi_request *) ptr ; -} - int async_get_timeout() { - struct wsgi_request* wsgi_req = uwsgi.wsgi_requests ; + struct wsgi_request* wsgi_req = uwsgi.wsgi_requests[0] ; int i ; time_t curtime, tdelta = 0 ; int ret = 0 ; @@ -297,7 +288,7 @@ int async_get_timeout() { } } } - wsgi_req = next_wsgi_req(wsgi_req) ; + wsgi_req = uwsgi.wsgi_requests[i+1]; } curtime = time(NULL); @@ -312,7 +303,7 @@ int async_get_timeout() { void async_expire_timeouts() { - struct wsgi_request* wsgi_req = uwsgi.wsgi_requests ; + struct wsgi_request* wsgi_req = uwsgi.wsgi_requests[0] ; int i ; time_t deadline = time(NULL); @@ -324,20 +315,20 @@ void async_expire_timeouts() { wsgi_req->async_timeout_expired = 1 ; } } - wsgi_req = next_wsgi_req(wsgi_req) ; + wsgi_req = uwsgi.wsgi_requests[i+1]; } } struct wsgi_request *find_first_available_wsgi_req() { - struct wsgi_request* wsgi_req = uwsgi.wsgi_requests ; + struct wsgi_request* wsgi_req = uwsgi.wsgi_requests[0] ; int i ; for(i=0;iasync_status == UWSGI_OK) { return wsgi_req ; } - wsgi_req = next_wsgi_req(wsgi_req) ; + wsgi_req = uwsgi.wsgi_requests[i+1]; } return NULL ; @@ -354,7 +345,7 @@ struct wsgi_request *find_wsgi_req_by_id(int async_id) { struct wsgi_request *find_wsgi_req_by_fd(int fd, int etype) { - struct wsgi_request* wsgi_req = uwsgi.wsgi_requests ; + struct wsgi_request* wsgi_req = uwsgi.wsgi_requests[0] ; int i ; if (etype != -1) { @@ -362,7 +353,7 @@ struct wsgi_request *find_wsgi_req_by_fd(int fd, int etype) { if (wsgi_req->async_waiting_fd == fd && wsgi_req->async_waiting_fd_type == etype) { return wsgi_req ; } - wsgi_req = next_wsgi_req(wsgi_req) ; + wsgi_req = uwsgi.wsgi_requests[i+1]; } } else { @@ -370,7 +361,7 @@ struct wsgi_request *find_wsgi_req_by_fd(int fd, int etype) { if (wsgi_req->async_waiting_fd == fd) { return wsgi_req ; } - wsgi_req = next_wsgi_req(wsgi_req) ; + wsgi_req = uwsgi.wsgi_requests[i+1] ; } } @@ -388,7 +379,7 @@ void async_set_timeout(struct wsgi_request *wsgi_req, time_t timeout) { void async_write_all(char *data, size_t len) { - struct wsgi_request *wsgi_req = uwsgi.wsgi_requests ; + struct wsgi_request *wsgi_req = uwsgi.wsgi_requests[0] ; int i; ssize_t rlen ; @@ -402,18 +393,20 @@ void async_write_all(char *data, size_t len) { wsgi_req->response_size += rlen ; } } + wsgi_req = uwsgi.wsgi_requests[i+1] ; } } void async_unpause_all() { - struct wsgi_request *wsgi_req = uwsgi.wsgi_requests ; + struct wsgi_request *wsgi_req = uwsgi.wsgi_requests[0] ; int i; for(i=0;iasync_status == UWSGI_PAUSED) { wsgi_req->async_status = UWSGI_AGAIN; } + wsgi_req = uwsgi.wsgi_requests[i+1] ; } } @@ -423,7 +416,7 @@ struct wsgi_request * async_loop() { int i ; uwsgi.async_running = -1 ; - wsgi_req = uwsgi.wsgi_requests ; + wsgi_req = uwsgi.wsgi_requests[0] ; for(i=0;i= uwsgi.numproc && uwsgi.to_hell) { @@ -196,15 +200,7 @@ void master_loop(char **argv, char **environ) { } if (diedpid == 0) { - /* PLEASE, do not run python threads in the master process, you can potentially destroy the world, - we support this for hyperultramegagodprogrammer and systems - */ -#ifdef UWSGI_THREADING - if (uwsgi.has_threads && uwsgi.shared->options[UWSGI_OPTION_THREADS] == 1) { - uwsgi._save = PyEval_SaveThread(); - uwsgi.workers[uwsgi.mywid].i_have_gil = 0; - } -#endif + /* all processes ok, doing status scan after N seconds */ check_interval.tv_sec = uwsgi.shared->options[UWSGI_OPTION_MASTER_INTERVAL]; if (!check_interval.tv_sec) @@ -234,6 +230,7 @@ void master_loop(char **argv, char **environ) { #endif else { if (udp_callable && udp_callable_args) { + uwsgi_get_gil(); PyTuple_SetItem(udp_callable_args, 0, PyString_FromString(udp_client_addr)); PyTuple_SetItem(udp_callable_args, 1, PyInt_FromLong(ntohs(udp_client.sin_port))); PyTuple_SetItem(udp_callable_args, 2, PyString_FromStringAndSize(uwsgi.wsgi_req->buffer, rlen)); @@ -243,6 +240,8 @@ void master_loop(char **argv, char **environ) { } if (PyErr_Occurred()) PyErr_Print(); + + uwsgi_release_gil(); } else { // a simple udp logger @@ -271,12 +270,7 @@ void master_loop(char **argv, char **environ) { master_cycles++; working_workers = 0; blocking_workers = 0; -#ifdef UWSGI_THREADING - if (uwsgi.has_threads && !uwsgi.workers[uwsgi.mywid].i_have_gil) { - PyEval_RestoreThread(uwsgi._save); - uwsgi.workers[uwsgi.mywid].i_have_gil = 1; - } -#endif + check_interval.tv_sec = uwsgi.shared->options[UWSGI_OPTION_MASTER_INTERVAL]; if (!check_interval.tv_sec) check_interval.tv_sec = 1; @@ -422,7 +416,6 @@ void master_loop(char **argv, char **environ) { uwsgi.workers[uwsgi.mywid].respawn_count++; uwsgi.workers[uwsgi.mywid].last_spawn = time(NULL); uwsgi.workers[uwsgi.mywid].manage_next_request = 1; - uwsgi.workers[uwsgi.mywid].i_have_gil = 1; break; } else if (pid < 1) { diff --git a/utils.c b/utils.c index fd0bb1aa..f0ee7eb7 100644 --- a/utils.c +++ b/utils.c @@ -460,6 +460,14 @@ inline struct wsgi_request *current_wsgi_req() { void sanitize_args() { + if (uwsgi.async > 0) { + uwsgi.cores = uwsgi.async; + } + + if (uwsgi.threads > 0) { + uwsgi.cores = uwsgi.threads; + } + #ifdef UWSGI_UGREEN #ifdef UWSGI_THREADING if (uwsgi.ugreen) { diff --git a/uwsgi.c b/uwsgi.c index 1188a202..4a778914 100644 --- a/uwsgi.c +++ b/uwsgi.c @@ -153,6 +153,7 @@ static struct option long_options[] = { {"chdir2", required_argument, 0, LONG_ARGS_CHDIR2}, {"mount", required_argument, 0, LONG_ARGS_MOUNT}, {"grunt", no_argument, &uwsgi.grunt, 1}, + {"threads", required_argument, 0, LONG_ARGS_THREADS}, {"no-site", no_argument, &Py_NoSiteFlag, 1}, {"vhost", no_argument, &uwsgi.vhost, 1}, #ifdef UWSGI_ROUTING @@ -484,6 +485,8 @@ int main(int argc, char *argv[], char *envp[]) { uwsgi.shared->after_hooks[i] = unconfigured_after_hook; } + uwsgi.cores = 1; + uwsgi.apps_cnt = 1; uwsgi.default_app = -1; @@ -792,22 +795,30 @@ int main(int argc, char *argv[], char *envp[]) { } } - // allocate more wsgi_req for async mode - uwsgi.wsgi_requests = malloc(sizeof(struct wsgi_request) * uwsgi.async); + // allocate more wsgi_req for async/thread modes + uwsgi.wsgi_requests = malloc(sizeof(struct wsgi_request *) * uwsgi.cores); if (uwsgi.wsgi_requests == NULL) { uwsgi_log("unable to allocate memory for requests.\n"); exit(1); } - memset(uwsgi.wsgi_requests, 0, sizeof(struct wsgi_request) * uwsgi.async); - uwsgi.async_buf = malloc( sizeof(char *) * uwsgi.async); + for(i=0;i 0) { - uwsgi.async_post_buf = malloc( sizeof(char *) * uwsgi.async); + uwsgi.async_post_buf = malloc( sizeof(char *) * uwsgi.cores); if (!uwsgi.async_post_buf) { uwsgi_error("malloc()"); exit(1); @@ -818,7 +829,7 @@ int main(int argc, char *argv[], char *envp[]) { } } - for(i=0;i 1) { - uwsgi_log("allocated %llu bytes (%llu KB) for %d request's buffer.\n", (uint64_t) (sizeof(struct wsgi_request) * uwsgi.async), - (uint64_t)( (sizeof(struct wsgi_request) * uwsgi.async ) / 1024), - uwsgi.async); + if (uwsgi.cores > 1) { + uwsgi_log("allocated %llu bytes (%llu KB) for %d request's buffer.\n", (uint64_t) (sizeof(struct wsgi_request) * uwsgi.cores), + (uint64_t)( (sizeof(struct wsgi_request) * uwsgi.cores ) / 1024), + uwsgi.cores); } if (uwsgi.pyhome != NULL) { @@ -1354,9 +1365,6 @@ int main(int argc, char *argv[], char *envp[]) { uwsgi.workers[1].id = 1; uwsgi.workers[1].last_spawn = time(NULL); uwsgi.workers[1].manage_next_request = 1; -#ifdef UWSGI_THREADING - uwsgi.workers[1].i_have_gil = 1; -#endif uwsgi.mywid = 1; gettimeofday(&last_respawn, NULL); uwsgi.respawn_delta = last_respawn.tv_sec; @@ -1372,9 +1380,6 @@ int main(int argc, char *argv[], char *envp[]) { uwsgi.workers[i].id = i; uwsgi.workers[i].last_spawn = time(NULL); uwsgi.workers[i].manage_next_request = 1; -#ifdef UWSGI_THREADING - uwsgi.workers[i].i_have_gil = 1; -#endif uwsgi.mywid = i; /* check this part if (uwsgi.serverfd != 0 && uwsgi.master_process == 1) { @@ -1491,13 +1496,8 @@ int main(int argc, char *argv[], char *envp[]) { close(uwsgi.erlangfd); #endif -#ifdef UWSGI_THREADING // release the GIL - if (uwsgi.has_threads) { - uwsgi._save = PyEval_SaveThread(); - uwsgi.workers[uwsgi.mywid].i_have_gil = 0; - } -#endif + uwsgi_release_gil(); #ifdef UWSGI_ASYNC @@ -1519,7 +1519,37 @@ int main(int argc, char *argv[], char *envp[]) { #endif // re-initialize wsgi_req (can be full of init_uwsgi_app data) - memset(uwsgi.wsgi_requests, 0, sizeof(struct wsgi_request) * uwsgi.async); + for(i=0;i 1) { + pthread_attr_t pa; + pthread_t *a_thread; + int ret; + + ret = pthread_attr_init(&pa); + if (ret) { + uwsgi_log("pthread_attr_init() = %d\n", ret); + exit(1); + } + + ret = pthread_attr_setdetachstate(&pa, PTHREAD_CREATE_DETACHED); + if (ret) { + uwsgi_log("pthread_attr_setdetachstate() = %d\n", ret); + exit(1); + } + + if (pthread_key_create(&uwsgi.ut_key, NULL)) { + uwsgi_error("pthread_key_create()"); + exit(1); + } + for(i=0;i #include #include @@ -202,6 +210,7 @@ PyAPI_FUNC(PyObject *) PyMarshal_ReadObjectFromString(char *, Py_ssize_t); #define LONG_ARGS_LOG_5xx 17051 #define LONG_ARGS_LOG_BIG 17052 #define LONG_ARGS_MOUNT 17053 +#define LONG_ARGS_THREADS 17054 @@ -512,7 +521,7 @@ struct uwsgi_server { int enable_profiler; // base for all the requests (even on async mode) - struct wsgi_request *wsgi_requests ; + struct wsgi_request **wsgi_requests ; struct wsgi_request *wsgi_req ; PyThreadState *_save; @@ -782,6 +791,10 @@ struct uwsgi_server { char *mounts[MAX_MOUNTPOINTS]; int mounts_cnt; + + int cores; + int threads; + pthread_key_t ut_key; }; struct uwsgi_cluster_node { @@ -849,8 +862,6 @@ struct uwsgi_worker { pid_t pid; uint64_t status; - int i_have_gil; - time_t last_spawn; uint64_t respawn_count; @@ -1247,3 +1258,5 @@ PyObject *uwsgi_mount_loader(void *); char *get_uwsgi_pymodule(char *); PyObject *get_uwsgi_pydict(char *); + +void *simple_loop(void *); diff --git a/uwsgi_pymodule.c b/uwsgi_pymodule.c index 37193257..f3327c62 100644 --- a/uwsgi_pymodule.c +++ b/uwsgi_pymodule.c @@ -1121,8 +1121,8 @@ PyObject *py_uwsgi_grunt(PyObject * self, PyObject * args) { uwsgi.workers[uwsgi.mywid].id = uwsgi.mywid; // this field will be overwrite after each call uwsgi.workers[uwsgi.mywid].pid = uwsgi.mypid; - // take the gil to support threads in grunt (is this useful ?) - uwsgi.workers[uwsgi.mywid].i_have_gil = 1; + // TODO + // manage thread in grunt processes Py_INCREF(Py_True); return Py_True; } diff --git a/uwsgiconfig.py b/uwsgiconfig.py index c0266c48..2d70dcda 100644 --- a/uwsgiconfig.py +++ b/uwsgiconfig.py @@ -89,7 +89,7 @@ gcc_major = int(gcc_version.split('.')[0]) gcc_minor = int(gcc_version.split('.')[1]) -gcc_list = ['utils', 'pyutils', 'pyloader', 'protocol', 'socket', 'logging', 'master', 'wsgi_handlers', 'wsgi_subhandler', 'wsgi_headers', 'uwsgi_handlers', 'plugins', 'uwsgi'] +gcc_list = ['utils', 'pyutils', 'pyloader', 'protocol', 'socket', 'logging', 'master', 'wsgi_handlers', 'wsgi_subhandler', 'wsgi_headers', 'uwsgi_handlers', 'plugins', 'loop', 'uwsgi'] cflags = ['-O2', '-Wall', '-Werror', '-D_LARGEFILE_SOURCE', '-D_FILE_OFFSET_BITS=64'] + os.environ.get("CFLAGS", "").split() diff --git a/wsgi_handlers.c b/wsgi_handlers.c index d7bb9a91..3e8741b0 100644 --- a/wsgi_handlers.c +++ b/wsgi_handlers.c @@ -13,17 +13,9 @@ PyObject *py_uwsgi_write(PyObject * self, PyObject * args) { if (PyString_Check(data)) { content = PyString_AsString(data); len = PyString_Size(data); - -#ifdef UWSGI_THREADING - if (uwsgi.has_threads && uwsgi.shared->options[UWSGI_OPTION_THREADS] == 1) { - Py_BEGIN_ALLOW_THREADS wsgi_req->response_size = write(wsgi_req->poll.fd, content, len); - Py_END_ALLOW_THREADS} - else { -#endif - wsgi_req->response_size = write(wsgi_req->poll.fd, content, len); -#ifdef UWSGI_THREADING - } -#endif + uwsgi_release_gil(); + wsgi_req->response_size = write(wsgi_req->poll.fd, content, len); + uwsgi_get_gil(); } Py_INCREF(Py_None); @@ -87,7 +79,7 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { int tmp_stderr; if (uwsgi_version == NULL) { - uwsgi_version = PyString_FromString(UWSGI_VERSION); + //uwsgi_version = PyString_FromString(UWSGI_VERSION); } char *what; @@ -125,14 +117,6 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { } } - -#ifdef UWSGI_THREADING - if (uwsgi.has_threads && !uwsgi.workers[uwsgi.mywid].i_have_gil) { - PyEval_RestoreThread(uwsgi._save); - uwsgi.workers[uwsgi.mywid].i_have_gil = 1; - } -#endif - if (!uwsgi.ignore_script_name) { if (!wsgi_req->script_name) @@ -163,7 +147,9 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { #endif ) { // a bit of magic: 1-1 = 0 / 0-1 = -1 + uwsgi_get_gil(); wsgi_req->app_id = init_uwsgi_app(LOADER_DYN, (void *) wsgi_req, wsgi_req, uwsgi.single_interpreter-1); + uwsgi_release_gil(); } } } @@ -195,7 +181,9 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { } // set the interpreter + uwsgi_get_gil(); PyThreadState_Swap(wi->interpreter); + uwsgi_release_gil(); if (wi->chdir) { #ifdef UWSGI_DEBUG uwsgi_debug("chdir to %s\n", wi->chdir); @@ -230,6 +218,8 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { wsgi_req->async_environ = wi->wsgi_environ; wsgi_req->async_args = wi->wsgi_args; #endif + + uwsgi_get_gil(); Py_INCREF((PyObject *)wsgi_req->async_environ); @@ -273,6 +263,7 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { // set wsgi vars + uwsgi_release_gil(); if (uwsgi.post_buffering > 0 && wsgi_req->post_cl > (size_t) uwsgi.post_buffering) { wsgi_req->async_post = tmpfile(); @@ -308,6 +299,7 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { wsgi_req->async_post = fdopen(wsgi_req->poll.fd, "r"); } + uwsgi_get_gil(); wsgi_req->async_result = (*wi->request_subhandler)(wsgi_req, wi); @@ -324,6 +316,7 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { } + else if (uwsgi.catch_exceptions) { wsgi_req->response_size += write(wsgi_req->poll.fd, wsgi_req->protocol, wsgi_req->protocol_len); @@ -355,6 +348,7 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { close(tmp_stderr); } + clear: if (uwsgi.single_interpreter == 0 && wsgi_req->app_id > 0) { @@ -362,6 +356,8 @@ clear: PyThreadState_Swap(uwsgi.main_thread); } + uwsgi_release_gil(); + clear2: