From c94206f7baf63b1f9eabe48b69d7b9dc513de9dd Mon Sep 17 00:00:00 2001 From: "roberto@sirius" Date: Thu, 13 Jan 2011 11:42:59 +0100 Subject: [PATCH] a huge list of threading fixes, we are near to ROCK SOLID threading mode --- logging.c | 19 ++++++- master.c | 2 +- plugins/lua/lua_plugin.c | 9 +--- plugins/python/gil.c | 4 +- plugins/python/pyloader.c | 25 +++++----- plugins/python/python_plugin.c | 11 +++++ plugins/python/uwsgi_pymodule.c | 22 ++++++--- plugins/python/uwsgi_python.h | 9 ++++ plugins/python/wsgi_handlers.c | 88 +++++++++++++++++++-------------- protocol.c | 4 +- tests/logger.py | 11 +++++ utils.c | 4 +- uwsgi.c | 15 +++--- uwsgi.h | 9 ++-- 14 files changed, 147 insertions(+), 85 deletions(-) create mode 100644 tests/logger.py diff --git a/logging.c b/logging.c index 374041cf..ffed60d1 100644 --- a/logging.c +++ b/logging.c @@ -34,6 +34,12 @@ void log_request(struct wsgi_request *wsgi_req) { struct iovec logvec[4]; int logvecpos = 0; + const char *msecs = "msecs"; + const char *micros = "micros"; + + long int rt; + char *tsize = (char *) msecs; + #ifdef UWSGI_SENDFILE char *msg1 = " via sendfile() "; #endif @@ -57,6 +63,15 @@ void log_request(struct wsgi_request *wsgi_req) { microseconds = wsgi_req->end_of_request.tv_sec * 1000000 + wsgi_req->end_of_request.tv_usec; microseconds2 = wsgi_req->start_of_request.tv_sec * 1000000 + wsgi_req->start_of_request.tv_usec; + rt = (long int) (microseconds - microseconds2); + + if (uwsgi.log_micros) { + tsize = (char *) micros; + } + else { + rt /= 1000; + } + if (uwsgi.vhost) { logvec[logvecpos].iov_base = wsgi_req->host; logvec[logvecpos].iov_len = wsgi_req->host_len; @@ -83,7 +98,7 @@ void log_request(struct wsgi_request *wsgi_req) { } - rlen = snprintf(logpkt, 4096, "[pid: %d|app: %d|req: %d/%llu] %.*s (%.*s) {%d vars in %d bytes} [%.*s] %.*s %.*s => generated %llu bytes in %ld msecs%s(%.*s %d) %d headers in %llu bytes (%d switches on core %d)\n", + rlen = snprintf(logpkt, 4096, "[pid: %d|app: %d|req: %d/%llu] %.*s (%.*s) {%d vars in %d bytes} [%.*s] %.*s %.*s => generated %llu bytes in %ld %s%s(%.*s %d) %d headers in %llu bytes (%d switches on core %d)\n", (int) uwsgi.mypid, wsgi_req->app_id, app_req, @@ -96,7 +111,7 @@ void log_request(struct wsgi_request *wsgi_req) { wsgi_req->method_len, wsgi_req->method, wsgi_req->uri_len, wsgi_req->uri, (unsigned long long) wsgi_req->response_size, - (long int) (microseconds - microseconds2) / 1000, + rt, tsize, via, wsgi_req->protocol_len, wsgi_req->protocol, wsgi_req->status, diff --git a/master.c b/master.c index 28dd9295..a9eeebd1 100644 --- a/master.c +++ b/master.c @@ -520,7 +520,7 @@ void master_loop(char **argv, char **environ) { for(i=1;irequests; + tmp_counter += uwsgi.core[j]->requests; } uwsgi.workers[i].requests = tmp_counter; } diff --git a/plugins/lua/lua_plugin.c b/plugins/lua/lua_plugin.c index 460adc74..20da1485 100644 --- a/plugins/lua/lua_plugin.c +++ b/plugins/lua/lua_plugin.c @@ -42,20 +42,13 @@ static void ulua_check_args(lua_State *L, const char *func, int n) { static int uwsgi_api_log(lua_State *L) { - time_t tt; const char *logline ; lca(L, 1); if (lua_isstring(L, 1)) { logline = lua_tolstring(L, 1, NULL); - tt = time(NULL); - if (logline[strlen(logline)] != '\n') { - uwsgi_log( UWSGI_LOGBASE " %.*s] %s\n", 24, ctime(&tt), logline); - } - else { - uwsgi_log( UWSGI_LOGBASE " %.*s] %s", 24, ctime(&tt), logline); - } + uwsgi_log( "%s\n", logline); } return 0; diff --git a/plugins/python/gil.c b/plugins/python/gil.c index 8b6a8d97..dfcd0d7f 100644 --- a/plugins/python/gil.c +++ b/plugins/python/gil.c @@ -5,11 +5,11 @@ extern struct uwsgi_python up; void gil_real_get() { PyEval_AcquireLock(); - PyThreadState_Swap((PyThreadState *) pthread_getspecific(up.upt_save_key)); + PyThreadState_Swap((PyThreadState *) pthread_getspecific(up.upt_gil_key)); } void gil_real_release() { - pthread_setspecific(up.upt_save_key, (void *) PyThreadState_Swap(NULL)); + pthread_setspecific(up.upt_gil_key, (void *) PyThreadState_Swap(NULL)); PyEval_ReleaseLock(); } diff --git a/plugins/python/pyloader.c b/plugins/python/pyloader.c index 9b1374c0..09b69841 100644 --- a/plugins/python/pyloader.c +++ b/plugins/python/pyloader.c @@ -158,7 +158,6 @@ int init_uwsgi_app(int loader, void *arg1, struct wsgi_request *wsgi_req, PyThre } #endif - // check function args // by defaut it is a WSGI app wi->argc = 2; @@ -247,22 +246,22 @@ int init_uwsgi_app(int loader, void *arg1, struct wsgi_request *wsgi_req, PyThre #endif } - if (interpreter == NULL && id) { + if (uwsgi.threads > 1 && id) { // if we have multiple threads we need to initialize a PyThreadState for each one - if (uwsgi.threads > 1) { - for(i=0;its[id] = PyThreadState_New( ((PyThreadState *)wi->interpreter)->interp); - if (!uwsgi.workers[uwsgi.mywid].cores[i]->ts[id]) { - uwsgi_log("unable to allocate new PyThreadState structure for app %s", mountpoint); - goto doh; - } + for(i=0;its[id] = PyThreadState_New( ((PyThreadState *)wi->interpreter)->interp); + if (!uwsgi.core[i]->ts[id]) { + uwsgi_log("unable to allocate new PyThreadState structure for app %s", mountpoint); + goto doh; } - PyThreadState_Swap((PyThreadState *) pthread_getspecific(up.upt_save_key)); - } - else { - PyThreadState_Swap(up.main_thread); } + PyThreadState_Swap((PyThreadState *) pthread_getspecific(up.upt_save_key) ); } + else if (interpreter == NULL && id) { + PyThreadState_Swap(up.main_thread); + } + + uwsgi_log("done\n"); if (wi->argc == 1) { uwsgi_log( "Web3 application %d (SCRIPT_NAME=%.*s) ready on interpreter %p", id, wi->mountpoint_len, wi->mountpoint, wi->interpreter); diff --git a/plugins/python/python_plugin.c b/plugins/python/python_plugin.c index 9492ca13..686b666a 100644 --- a/plugins/python/python_plugin.c +++ b/plugins/python/python_plugin.c @@ -124,6 +124,9 @@ int uwsgi_python_init() { up.gil_get = gil_fake_get; up.gil_release = gil_fake_release; + up.swap_ts = simple_swap_ts; + up.reset_ts = simple_reset_ts; + return 1; } @@ -671,12 +674,19 @@ int uwsgi_python_mount_app(char *mountpoint, char *app) { uwsgi_error("pthread_key_create()"); exit(1); } + if (pthread_key_create(&up.upt_gil_key, NULL)) { + uwsgi_error("pthread_key_create()"); + exit(1); + } pthread_setspecific(up.upt_save_key, (void *) PyThreadState_Get()); + pthread_setspecific(up.upt_gil_key, (void *) PyThreadState_Get()); pthread_mutex_init(&up.lock_pyloaders, NULL); pthread_atfork(uwsgi_python_pthread_prepare, uwsgi_python_pthread_parent, uwsgi_python_pthread_child); up.gil_get = gil_real_get; up.gil_release = gil_real_release; + up.swap_ts = threaded_swap_ts; + up.reset_ts = threaded_reset_ts; uwsgi_log("threads support enabled\n"); } @@ -686,6 +696,7 @@ int uwsgi_python_mount_app(char *mountpoint, char *app) { PyThreadState *pts; pts = PyThreadState_New(up.main_thread->interp); pthread_setspecific(up.upt_save_key, (void *) pts); + pthread_setspecific(up.upt_gil_key, (void *) pts); } diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index a551a5c5..83c8460d 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/plugins/python/uwsgi_pymodule.c @@ -410,6 +410,16 @@ PyObject *py_uwsgi_signal(PyObject * self, PyObject * args) { } +PyObject *py_uwsgi_log_this(PyObject * self, PyObject * args) { + + struct wsgi_request *wsgi_req = current_wsgi_req(); + + wsgi_req->log_this = 1; + + Py_INCREF(Py_None); + return Py_None; +} + PyObject *py_uwsgi_recv_frame(PyObject * self, PyObject * args) { struct wsgi_request *wsgi_req = current_wsgi_req(); @@ -763,19 +773,12 @@ PyObject *py_uwsgi_warning(PyObject * self, PyObject * args) { PyObject *py_uwsgi_log(PyObject * self, PyObject * args) { char *logline; - time_t tt; if (!PyArg_ParseTuple(args, "s:log", &logline)) { return NULL; } - tt = time(NULL); - if (logline[strlen(logline)] != '\n') { - uwsgi_log( UWSGI_LOGBASE " %.*s] %s\n", 24, ctime(&tt), logline); - } - else { - uwsgi_log( UWSGI_LOGBASE " %.*s] %s", 24, ctime(&tt), logline); - } + uwsgi_log( "%s\n", logline); Py_INCREF(Py_True); return Py_True; @@ -1837,7 +1840,9 @@ PyObject *py_uwsgi_cl(PyObject * self, PyObject * args) { struct wsgi_request *wsgi_req = current_wsgi_req(); +#ifdef UWSGI_DEBUG uwsgi_log( "disconnecting worker %d (pid :%d) from session...\n", uwsgi.mywid, uwsgi.mypid); +#endif fclose(wsgi_req->async_post); wsgi_req->fd_closed = 1; @@ -2111,6 +2116,7 @@ static PyMethodDef uwsgi_advanced_methods[] = { {"request_id", py_uwsgi_request_id, METH_VARARGS, ""}, {"worker_id", py_uwsgi_worker_id, METH_VARARGS, ""}, {"log", py_uwsgi_log, METH_VARARGS, ""}, + {"log_this_request", py_uwsgi_log_this, METH_VARARGS, ""}, {"disconnect", py_uwsgi_disconnect, METH_VARARGS, ""}, {"grunt", py_uwsgi_grunt, METH_VARARGS, ""}, {"load_plugin", py_uwsgi_load_plugin, METH_VARARGS, ""}, diff --git a/plugins/python/uwsgi_python.h b/plugins/python/uwsgi_python.h index 3d03799e..791fe1e1 100644 --- a/plugins/python/uwsgi_python.h +++ b/plugins/python/uwsgi_python.h @@ -108,8 +108,12 @@ struct uwsgi_python { int current_recursion_depth; struct _frame* current_frame; + void (*swap_ts)(struct wsgi_request *, struct uwsgi_app *); + void (*reset_ts)(struct wsgi_request *, struct uwsgi_app *); + #ifdef UWSGI_THREADING pthread_key_t upt_save_key; + pthread_key_t upt_gil_key; pthread_mutex_t lock_pyloaders; void (*gil_get) (void); void (*gil_release) (void); @@ -209,3 +213,8 @@ void init_uwsgi_module_sharedarea(PyObject *); void init_uwsgi_module_cache(PyObject *); PyObject *uwsgi_pyimport_by_filename(char *, char *); + +void threaded_swap_ts(struct wsgi_request *, struct uwsgi_app *); +void simple_swap_ts(struct wsgi_request *, struct uwsgi_app *); +void threaded_reset_ts(struct wsgi_request *, struct uwsgi_app *); +void simple_reset_ts(struct wsgi_request *, struct uwsgi_app *); diff --git a/plugins/python/wsgi_handlers.c b/plugins/python/wsgi_handlers.c index a20570ba..ccb9055a 100644 --- a/plugins/python/wsgi_handlers.c +++ b/plugins/python/wsgi_handlers.c @@ -191,29 +191,15 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { wi = &uwsgi.apps[wsgi_req->app_id]; - if (uwsgi.single_interpreter == 0 && wi->interpreter != up.main_thread) { - if (!wi->interpreter) { - internal_server_error(wsgi_req->poll.fd, "wsgi application's %d interpreter not found"); - goto clear2; - } - - // set the interpreter - UWSGI_GET_GIL - if (uwsgi.threads > 1) { - PyThreadState_Swap(uwsgi.workers[uwsgi.mywid].cores[wsgi_req->async_id]->ts[wsgi_req->app_id]); - } - else { - PyThreadState_Swap(wi->interpreter); - } - UWSGI_RELEASE_GIL - if (wi->chdir) { + up.swap_ts(wsgi_req, wi); + + if (wi->chdir) { #ifdef UWSGI_DEBUG - uwsgi_debug("chdir to %s\n", wi->chdir); + uwsgi_debug("chdir to %s\n", wi->chdir); #endif - if (chdir(wi->chdir)) { - uwsgi_error("chdir()"); - } - } + if (chdir(wi->chdir)) { + uwsgi_error("chdir()"); + } } @@ -241,8 +227,8 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { UWSGI_GET_GIL - // no fear of race conditions for this counter as it is already protected by the GIL - wi->requests++; + // no fear of race conditions for this counter as it is already protected by the GIL + wi->requests++; Py_INCREF((PyObject *)wsgi_req->async_environ); @@ -291,10 +277,11 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { wsgi_req->async_result = wi->request_subhandler(wsgi_req, wi); + UWSGI_RELEASE_GIL + if (wsgi_req->async_result) { - UWSGI_RELEASE_GIL while (wi->response_subhandler(wsgi_req) != UWSGI_OK) { wsgi_req->switches++; #ifdef UWSGI_ASYNC @@ -332,7 +319,9 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { goto clear; } // print the error + UWSGI_GET_GIL PyErr_Print(); + UWSGI_RELEASE_GIL // ...resume the original stderr, in case of error we are damaged forever !!! if (dup2(tmp_stderr, 2) < 0) { uwsgi_error("dup2()"); @@ -342,19 +331,7 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) { clear: - UWSGI_GET_GIL - - if (uwsgi.single_interpreter == 0 && wi->interpreter != up.main_thread) { - // restoring main interpreter - if (uwsgi.threads > 1) { - PyThreadState_Swap((PyThreadState *) pthread_getspecific(up.upt_save_key)); - } - else { - PyThreadState_Swap(up.main_thread); - } - } - - UWSGI_RELEASE_GIL + up.reset_ts(wsgi_req, wi); clear2: @@ -365,7 +342,7 @@ clear2: void uwsgi_after_request_wsgi(struct wsgi_request *wsgi_req) { - if (uwsgi.shared->options[UWSGI_OPTION_LOGGING]) { + if (uwsgi.shared->options[UWSGI_OPTION_LOGGING] || wsgi_req->log_this) { log_request(wsgi_req); } else { @@ -416,3 +393,38 @@ PyObject *py_uwsgi_sendfile(PyObject * self, PyObject * args) { return (PyObject *) wsgi_req->sendfile_obj; } #endif + +void threaded_swap_ts(struct wsgi_request *wsgi_req, struct uwsgi_app *wi) { + + if (uwsgi.single_interpreter == 0 && wi->interpreter != up.main_thread) { + UWSGI_GET_GIL + PyThreadState_Swap(uwsgi.core[wsgi_req->async_id]->ts[wsgi_req->app_id]); + UWSGI_RELEASE_GIL + } + +} + +void threaded_reset_ts(struct wsgi_request *wsgi_req, struct uwsgi_app *wi) { + if (uwsgi.single_interpreter == 0 && wi->interpreter != up.main_thread) { + UWSGI_GET_GIL + PyThreadState_Swap((PyThreadState *) pthread_getspecific(up.upt_save_key)); + UWSGI_RELEASE_GIL + } +} + + +void simple_reset_ts(struct wsgi_request *wsgi_req, struct uwsgi_app *wi) { + if (uwsgi.single_interpreter == 0 && wi->interpreter != up.main_thread) { + // restoring main interpreter + PyThreadState_Swap(up.main_thread); + } +} + + +void simple_swap_ts(struct wsgi_request *wsgi_req, struct uwsgi_app *wi) { + + if (uwsgi.single_interpreter == 0 && wi->interpreter != up.main_thread) { + // set the interpreter + PyThreadState_Swap(wi->interpreter); + } +} diff --git a/protocol.c b/protocol.c index 8f9e54ff..ee8e79bf 100644 --- a/protocol.c +++ b/protocol.c @@ -523,7 +523,7 @@ int uwsgi_parse_vars(struct wsgi_request *wsgi_req) { script_name = wsgi_req->var_cnt; } for(i=0;i= uwsgi.apps[i].mountpoint_len) { if (!uwsgi_startswith(orig_path_info, uwsgi.apps[i].mountpoint, uwsgi.apps[i].mountpoint_len) && uwsgi.apps[i].mountpoint_len > best_found) { best_found = uwsgi.apps[i].mountpoint_len; @@ -537,7 +537,9 @@ int uwsgi_parse_vars(struct wsgi_request *wsgi_req) { wsgi_req->hvec[path_info].iov_base = wsgi_req->path_info; wsgi_req->hvec[path_info].iov_len = wsgi_req->path_info_len; +#ifdef UWSGI_DEBUG uwsgi_log("managed SCRIPT_NAME = %.*s PATH_INFO = %.*s\n", wsgi_req->script_name_len, wsgi_req->script_name, wsgi_req->path_info_len, wsgi_req->path_info); +#endif } } } diff --git a/tests/logger.py b/tests/logger.py new file mode 100644 index 00000000..530022dd --- /dev/null +++ b/tests/logger.py @@ -0,0 +1,11 @@ +import uwsgi + +uwsgi.log("I am uWSGI %s" % uwsgi.version) + +def application(env, start_response): + start_response('200 OK', [('Content-Type','text/html')]) + uwsgi.log(str(env)) + + if env['PATH_INFO'] == '/logme': + uwsgi.log_this_request() + return "log written" diff --git a/utils.c b/utils.c index 0af08308..1a28df2d 100644 --- a/utils.c +++ b/utils.c @@ -393,7 +393,7 @@ void uwsgi_close_request(struct wsgi_request *wsgi_req) { uwsgi.workers[uwsgi.mywid].requests++; if (uwsgi.cores > 1) { - uwsgi.workers[uwsgi.mywid].cores[wsgi_req->async_id]->requests++; + uwsgi.core[wsgi_req->async_id]->requests++; } // after_request hook @@ -946,7 +946,7 @@ int uwsgi_get_app_id(char *script_name, int script_name_len, int modifier1) { int i; for(i=0;i 1) { - for (i = 1; i < uwsgi.numproc + 1; i++) { - uwsgi.workers[i].cores = (struct uwsgi_core **) mmap(NULL, sizeof(struct uwsgi_core *) * uwsgi.cores, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0); - if (!uwsgi.workers[i].cores) { + for (i = 0; i < uwsgi.numproc+1; i++) { + uwsgi.core = (struct uwsgi_core **) mmap(NULL, sizeof(struct uwsgi_core *) * uwsgi.cores, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0); + if (!uwsgi.core) { uwsgi_error("mmap()"); exit(1); } - memset(uwsgi.workers[i].cores, 0, sizeof(struct uwsgi_core *) * uwsgi.cores); + memset(uwsgi.core, 0, sizeof(struct uwsgi_core *) * uwsgi.cores); for (j = 0; j < uwsgi.cores; j++) { - uwsgi.workers[i].cores[j] = (struct uwsgi_core *) mmap(NULL, sizeof(struct uwsgi_core), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0); - if (!uwsgi.workers[i].cores[j]) { + uwsgi.core[j] = (struct uwsgi_core *) mmap(NULL, sizeof(struct uwsgi_core), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANON, -1, 0); + if (!uwsgi.core[j]) { uwsgi_error("mmap()"); exit(1); } - memset(uwsgi.workers[i].cores[j], 0, sizeof(struct uwsgi_core)); + memset(uwsgi.core[j], 0, sizeof(struct uwsgi_core)); } } } diff --git a/uwsgi.h b/uwsgi.h index 65b365e8..63551bd4 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -4,8 +4,6 @@ #define UWSGI_VERSION "0.9.7-dev" -#define UWSGI_LOGBASE "[- uWSGI -" - #define uwsgi_error(x) uwsgi_log("%s: %s [%s line %d]\n", x, strerror(errno), __FILE__, __LINE__); #define uwsgi_debug(x, ...) uwsgi_log("[uWSGI DEBUG] " x, __VA_ARGS__); #define uwsgi_rawlog(x) if (write(2, x, strlen(x)) != strlen(x)) uwsgi_error("write()") @@ -589,6 +587,8 @@ struct wsgi_request { off_t frame_pos; int frame_len; + int log_this; + }; struct uwsgi_fmon { @@ -650,6 +650,7 @@ struct uwsgi_server { int manage_script_name; int no_default_app; int logdate; + int log_micros; char *log_strftime; #ifdef UWSGI_PROXY @@ -841,6 +842,9 @@ struct uwsgi_server { int mounts_cnt; int cores; + + struct uwsgi_core **core; + int threads; //this key old the u_request structure per core / thread @@ -1041,7 +1045,6 @@ struct uwsgi_worker { int manage_next_request; - struct uwsgi_core **cores; };