From d4f75ab53c459ba55b66874c91abba5561f03fb8 Mon Sep 17 00:00:00 2001 From: "roberto@natty32" Date: Wed, 16 Mar 2011 07:28:23 +0100 Subject: [PATCH] re-added ugreen support (better and faster) --- async.c | 42 ++--- buildconf/default.ini | 2 +- plugins/python/python_plugin.c | 27 ++- plugins/python/uwsgi_pymodule.c | 2 +- plugins/python/uwsgi_python.h | 8 +- plugins/ugreen/ugreen.c | 315 +++++--------------------------- tests/iobound_async.py | 2 + utils.c | 19 ++ uwsgi.h | 10 + 9 files changed, 113 insertions(+), 314 deletions(-) diff --git a/async.c b/async.c index d8423be6..3803adce 100644 --- a/async.c +++ b/async.c @@ -185,6 +185,9 @@ void async_add_fd_write(struct wsgi_request *wsgi_req, int fd, int timeout) { } +void async_schedule_to_req(void) { + uwsgi.wsgi_req->async_status = uwsgi.p[uwsgi.wsgi_req->uh.modifier1]->request(uwsgi.wsgi_req); +} void *async_loop(void *arg1) { @@ -193,13 +196,16 @@ void *async_loop(void *arg1) { struct uwsgi_rb_timer *min_timeout; int timeout; - struct uwsgi_async_request *current_request = NULL, *next_async_request = NULL; + static struct uwsgi_async_request *current_request = NULL, *next_async_request = NULL; void *events = event_queue_alloc(64); uwsgi.async_runqueue = NULL; uwsgi.async_runqueue_cnt = 0; + // set a default request manager + if (!uwsgi.schedule_to_req) uwsgi.schedule_to_req = async_schedule_to_req; + while (uwsgi.workers[uwsgi.mywid].manage_next_request) { if (uwsgi.async_runqueue_cnt) { @@ -259,35 +265,12 @@ void *async_loop(void *arg1) { #endif - if (wsgi_req_recv(uwsgi.wsgi_req)) { + if (wsgi_req_simple_recv(uwsgi.wsgi_req)) { continue; } - if (uwsgi.wsgi_req->async_status == UWSGI_OK) { - // fast request, close it - // remove all the monitored fds and timeout - while(uwsgi.wsgi_req->waiting_fds) { - event_queue_del_fd(uwsgi.async_queue, uwsgi.wsgi_req->waiting_fds->fd, uwsgi.wsgi_req->waiting_fds->event); - tmp_uaf = uwsgi.wsgi_req->waiting_fds; - uwsgi.async_waiting_fd_table[tmp_uaf->fd] = NULL; - uwsgi.wsgi_req->waiting_fds = tmp_uaf->next; - free(tmp_uaf); - } - uwsgi.wsgi_req->waiting_fds = NULL; - if (uwsgi.wsgi_req->async_timeout) { - rb_erase(&uwsgi.wsgi_req->async_timeout->rbt, uwsgi.rb_async_timeouts); - free(uwsgi.wsgi_req->async_timeout); - uwsgi.wsgi_req->async_timeout = NULL; - } - uwsgi_close_request(uwsgi.wsgi_req); - // push the wsgi_request in the unused stack - uwsgi.async_queue_unused_ptr++; - uwsgi.async_queue_unused[uwsgi.async_queue_unused_ptr] = uwsgi.wsgi_req; - } - else if (!uwsgi.wsgi_req->waiting_fds && !uwsgi.wsgi_req->async_timeout) { - // suspended request put it in the runqueue - runqueue_push(uwsgi.wsgi_req); - } + // put request in the runqueue + runqueue_push(uwsgi.wsgi_req); } else { // app event @@ -322,15 +305,14 @@ void *async_loop(void *arg1) { } // event queue managed, give cpu to runqueue - if (!current_request) current_request = uwsgi.async_runqueue; if (uwsgi.async_runqueue_cnt) { - uwsgi.wsgi_req = current_request->wsgi_req; - uwsgi.wsgi_req->async_status = uwsgi.p[uwsgi.wsgi_req->uh.modifier1]->request(uwsgi.wsgi_req); + + uwsgi.schedule_to_req(); next_async_request = current_request->next; // request ended ? diff --git a/buildconf/default.ini b/buildconf/default.ini index 11b16cad..56616284 100644 --- a/buildconf/default.ini +++ b/buildconf/default.ini @@ -21,7 +21,7 @@ xml_implementation = libxml2 plugins = bin_name = uwsgi plugin_dir = . -embedded_plugins = python, ping, nagios, rpc, fastrouter, http +embedded_plugins = python, ping, nagios, rpc, fastrouter, http, ugreen locking = auto event = auto diff --git a/plugins/python/python_plugin.c b/plugins/python/python_plugin.c index 201c9d81..5ef1a070 100644 --- a/plugins/python/python_plugin.c +++ b/plugins/python/python_plugin.c @@ -710,6 +710,11 @@ int uwsgi_python_mount_app(char *mountpoint, char *app) { void uwsgi_python_init_apps() { + if (uwsgi.async > 1) { + up.current_recursion_depth = uwsgi_malloc(sizeof(int)*uwsgi.async); + up.current_frame = uwsgi_malloc(sizeof(struct _frame)*uwsgi.async); + } + init_pyargv(); #ifdef UWSGI_MINTERPRETERS init_uwsgi_embedded_module(); @@ -841,9 +846,14 @@ void uwsgi_python_suspend(struct wsgi_request *wsgi_req) { PyThreadState *tstate = PyThreadState_GET(); - uwsgi_log("suspending python\n"); - up.current_recursion_depth = tstate->recursion_depth; - up.current_frame = tstate->frame; + if (wsgi_req) { + up.current_recursion_depth[wsgi_req->async_id] = tstate->recursion_depth; + up.current_frame[wsgi_req->async_id] = tstate->frame; + } + else { + up.current_main_recursion_depth = tstate->recursion_depth; + up.current_main_frame = tstate->frame; + } } @@ -958,9 +968,14 @@ void uwsgi_python_resume(struct wsgi_request *wsgi_req) { PyThreadState *tstate = PyThreadState_GET(); - uwsgi_log("resuming python\n"); - tstate->recursion_depth = up.current_recursion_depth; - tstate->frame = up.current_frame; + if (wsgi_req) { + tstate->recursion_depth = up.current_recursion_depth[wsgi_req->async_id]; + tstate->frame = up.current_frame[wsgi_req->async_id]; + } + else { + tstate->recursion_depth = up.current_main_recursion_depth; + tstate->frame = up.current_main_frame; + } } diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index 3e4ef13f..8c9823fa 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/plugins/python/uwsgi_pymodule.c @@ -2242,7 +2242,7 @@ PyObject *py_uwsgi_suspend(PyObject * self, PyObject * args) { struct wsgi_request *wsgi_req = current_wsgi_req(); - uwsgi.schedule_to_main(wsgi_req); + if (uwsgi.schedule_to_main) uwsgi.schedule_to_main(wsgi_req); Py_INCREF(Py_True); return Py_True; diff --git a/plugins/python/uwsgi_python.h b/plugins/python/uwsgi_python.h index ac5d37ad..1c775316 100644 --- a/plugins/python/uwsgi_python.h +++ b/plugins/python/uwsgi_python.h @@ -1,6 +1,7 @@ #include "../../uwsgi.h" #include +#include #define MAX_PYTHONPATH 64 #define MAX_PYMODULE_ALIAS 64 @@ -110,8 +111,11 @@ struct uwsgi_python { int ignore_script_name; int catch_exceptions; - int current_recursion_depth; - struct _frame* current_frame; + int *current_recursion_depth; + struct _frame **current_frame; + + int current_main_recursion_depth; + struct _frame *current_main_frame; void (*swap_ts)(struct wsgi_request *, struct uwsgi_app *); void (*reset_ts)(struct wsgi_request *, struct uwsgi_app *); diff --git a/plugins/ugreen/ugreen.c b/plugins/ugreen/ugreen.c index 5a964fbe..2c4cd770 100644 --- a/plugins/ugreen/ugreen.c +++ b/plugins/ugreen/ugreen.c @@ -12,7 +12,7 @@ struct uwsgi_ugreen { int ugreen; int stackpages; ucontext_t main; - ucontext_t **contexts; + ucontext_t *contexts; size_t u_stack_size; } ug; @@ -27,165 +27,60 @@ struct option ugreen_options[] = { { 0, 0, 0, 0 } }; -void u_green_loop(void); - -void u_green_write_all(char *data, size_t len) { - - struct wsgi_request *wsgi_req; - int i; - ssize_t rlen; - - for(i=0;iasync_status == UWSGI_PAUSED) { - rlen = write(wsgi_req->poll.fd, data, len); - if (rlen < 0) { - uwsgi_error("write()"); - // mark core as plagued - wsgi_req->async_plagued = 1; - } - else { - wsgi_req->response_size += rlen; - } - } - } +void u_green_request() { + uwsgi.wsgi_req->async_status = uwsgi.p[uwsgi.wsgi_req->uh.modifier1]->request(uwsgi.wsgi_req); + uwsgi.wsgi_req->suspended = 0; } -void u_green_unpause_all() { +inline static void u_green_schedule_to_req() { - struct wsgi_request *wsgi_req; - int i; + int id = uwsgi.wsgi_req->async_id; - for(i=0;iasync_status == UWSGI_PAUSED) { - wsgi_req->async_status = UWSGI_AGAIN; - wsgi_req->async_timeout = 0; - } + if (!uwsgi.wsgi_req->suspended) { + ug.contexts[id].uc_link = &ug.main; + makecontext(&ug.contexts[id], u_green_request, 0); + uwsgi.wsgi_req->suspended = 1; } -} + if (uwsgi.p[uwsgi.wsgi_req->uh.modifier1]->suspend) { + uwsgi.p[uwsgi.wsgi_req->uh.modifier1]->suspend(NULL); + } + + swapcontext(&ug.main, &ug.contexts[id] ); -static int u_green_blocking() { - struct wsgi_request* wsgi_req; - int i; + if (uwsgi.p[uwsgi.wsgi_req->uh.modifier1]->resume) { + uwsgi.p[uwsgi.wsgi_req->uh.modifier1]->resume(NULL); + } - for(i=0;iasync_status != UWSGI_ACCEPTING && wsgi_req->async_status != UWSGI_PAUSED && wsgi_req->async_waiting_fd == -1 && !wsgi_req->async_timeout) { - return 0; - } + if (uwsgi.wsgi_req->suspended) { + uwsgi.wsgi_req->async_status = UWSGI_AGAIN; } - return -1; } inline static void u_green_schedule_to_main(struct wsgi_request *wsgi_req) { - if (wsgi_req->async_status != UWSGI_ACCEPTING) { - if (uwsgi.p[wsgi_req->uh.modifier1]->suspend) { - uwsgi.p[wsgi_req->uh.modifier1]->suspend(wsgi_req); - } + if (uwsgi.p[wsgi_req->uh.modifier1]->suspend) { + uwsgi.p[wsgi_req->uh.modifier1]->suspend(wsgi_req); } - uwsgi_log("uGreen EXITING TO MAIN\n"); - swapcontext(ug.contexts[wsgi_req->async_id], &ug.main); + swapcontext(&ug.contexts[wsgi_req->async_id], &ug.main); - uwsgi_log("uGreen RETURNED FROM MAIN\n"); - - if (wsgi_req->async_status != UWSGI_ACCEPTING) { - if (uwsgi.p[wsgi_req->uh.modifier1]->resume) { - uwsgi.p[wsgi_req->uh.modifier1]->resume(wsgi_req); - } - } -} - -inline static void u_green_schedule_to_req(struct wsgi_request *wsgi_req) { - - - if (wsgi_req->async_status != UWSGI_ACCEPTING) { - if (uwsgi.p[wsgi_req->uh.modifier1]->suspend) { - uwsgi.p[wsgi_req->uh.modifier1]->suspend(wsgi_req); - } + if (uwsgi.p[wsgi_req->uh.modifier1]->resume) { + uwsgi.p[wsgi_req->uh.modifier1]->resume(wsgi_req); } uwsgi.wsgi_req = wsgi_req; - uwsgi_log("SWAPCONTEXT to %p\n", ug.contexts[wsgi_req->async_id]); - //wsgi_req->async_switches++; - swapcontext(&ug.main, ug.contexts[wsgi_req->async_id] ); - - uwsgi_log("RESUMED\n"); - - if (wsgi_req->async_status != UWSGI_ACCEPTING) { - if (uwsgi.p[wsgi_req->uh.modifier1]->resume) { - uwsgi.p[wsgi_req->uh.modifier1]->resume(wsgi_req); - } - } - } -void u_green_wait_for_fd(struct wsgi_request *wsgi_req, int fd, int etype, int timeout) { - - if (fd < 0) return; - - if (async_add(uwsgi.async_queue, fd, etype)) return; - - wsgi_req->async_waiting_fd = fd; - wsgi_req->async_waiting_fd_type = etype; - wsgi_req->async_timeout = time(NULL) + timeout; - - u_green_schedule_to_main(wsgi_req); - - async_del(uwsgi.async_queue, wsgi_req->async_waiting_fd, wsgi_req->async_waiting_fd_type); - - wsgi_req->async_waiting_fd = -1; - wsgi_req->async_timeout = 0; -} - -static void u_green_request(struct wsgi_request *wsgi_req, int async_id) { - - uwsgi_log("request handler args %p %d\n", wsgi_req, async_id); - - for(;;) { - uwsgi_log("accept()\n"); - wsgi_req_setup(wsgi_req, async_id); - - wsgi_req->async_status = UWSGI_ACCEPTING; - - u_green_schedule_to_main(wsgi_req); - - if (wsgi_req_accept(wsgi_req)) { - continue; - } - - uwsgi_log("REQUEST ACCEPTED\n"); - wsgi_req->async_status = UWSGI_OK; - - // check here - //u_green_schedule_to_main(wsgi_req); - - if (wsgi_req_recv(wsgi_req)) { - continue; - } - - while(wsgi_req->async_status == UWSGI_AGAIN) { - u_green_schedule_to_main(wsgi_req); - wsgi_req->async_status = uwsgi.p[wsgi_req->uh.modifier1]->request(wsgi_req); - } - - u_green_schedule_to_main(wsgi_req); - - uwsgi_close_request(wsgi_req); - - } - -} int u_green_init() { - struct wsgi_request *wsgi_req; + static int i; - volatile int i; + if (!ug.ugreen) { + return 0; + } ug.u_stack_size = UGREEN_DEFAULT_STACKSIZE; @@ -196,168 +91,38 @@ int u_green_init() { uwsgi_log("initializing %d uGreen threads with stack size of %lu (%lu KB)\n", uwsgi.async, (unsigned long) ug.u_stack_size, (unsigned long) ug.u_stack_size/1024); - ug.contexts = malloc( sizeof(ucontext_t*) * uwsgi.async); - if (!ug.contexts) { - uwsgi_error("malloc()\n"); - exit(1); - } + ug.contexts = uwsgi_malloc( sizeof(ucontext_t) * uwsgi.async); for(i=0;iuc_stack.ss_sp = mmap(NULL, ug.u_stack_size + (uwsgi.page_size*2) , PROT_READ | PROT_WRITE | PROT_EXEC, MAP_ANON | MAP_PRIVATE, -1, 0) + uwsgi.page_size; + getcontext(&ug.contexts[i]); - if (!ug.contexts[i]->uc_stack.ss_sp) { + ug.contexts[i].uc_stack.ss_sp = mmap(NULL, ug.u_stack_size + (uwsgi.page_size*2) , PROT_READ | PROT_WRITE | PROT_EXEC, MAP_ANON | MAP_PRIVATE, -1, 0) + uwsgi.page_size; + + if (!ug.contexts[i].uc_stack.ss_sp) { uwsgi_error("mmap()"); exit(1); } // set guard pages for stack - if (mprotect(ug.contexts[i]->uc_stack.ss_sp - uwsgi.page_size, uwsgi.page_size, PROT_NONE)) { + if (mprotect(ug.contexts[i].uc_stack.ss_sp - uwsgi.page_size, uwsgi.page_size, PROT_NONE)) { uwsgi_error("mprotect()"); exit(1); } - if (mprotect(ug.contexts[i]->uc_stack.ss_sp + ug.u_stack_size, uwsgi.page_size, PROT_NONE)) { + if (mprotect(ug.contexts[i].uc_stack.ss_sp + ug.u_stack_size, uwsgi.page_size, PROT_NONE)) { uwsgi_error("mprotect()"); exit(1); } - ug.contexts[i]->uc_stack.ss_size = ug.u_stack_size; + ug.contexts[i].uc_stack.ss_size = ug.u_stack_size; - ug.contexts[i]->uc_link = NULL; - makecontext(ug.contexts[i], (void(*)(void)) u_green_request, 2, wsgi_req, i); - wsgi_req->async_status = UWSGI_ACCEPTING; - wsgi_req->async_id = i; - uwsgi_log("wsgi_req %d %d %p\n", wsgi_req->async_id, wsgi_req->async_status, ug.contexts[i]); } - uwsgi_register_loop("ugreen", u_green_loop); - - return 0; - -} - -void u_green_expire_timeouts() { - - struct wsgi_request* wsgi_req; - int i; - time_t deadline = time(NULL); - - - for(i=0;iasync_timeout > 0) { - if (wsgi_req->async_timeout <= deadline) { - wsgi_req->async_status = UWSGI_AGAIN; - wsgi_req->async_timeout = 0; - } - } - } -} - -static int u_green_get_timeout() { - - - struct wsgi_request* wsgi_req; - int i; - time_t curtime, tdelta = 0; - int ret = 0; - - if (!uwsgi.async_running) return 0; - - for(i=0;iasync_timeout > 0) { - if (tdelta <= 0 || tdelta > wsgi_req->async_timeout) { - tdelta = wsgi_req->async_timeout; - } - } - } - - curtime = time(NULL); - - ret = tdelta - curtime; - if (ret > 0) { - return ret; - } - - return 0; -} - - -void u_green_loop() { - - struct wsgi_request *wsgi_req; - - int i, current = 0, timeout; - - for(i=0;iasync_status = UWSGI_ACCEPTING; - wsgi_req->async_id = i; - } uwsgi.schedule_to_main = u_green_schedule_to_main; + uwsgi.schedule_to_req = u_green_schedule_to_req; - uwsgi_log("FFAR: %p\n", find_first_accepting_wsgi_req()); - - while(uwsgi.workers[uwsgi.mywid].manage_next_request) { - - //uwsgi_log("i am uGreen...\n"); - - uwsgi.async_running = u_green_blocking(); - timeout = u_green_get_timeout(); - uwsgi.async_nevents = async_wait(uwsgi.async_queue, uwsgi.async_events, uwsgi.async, uwsgi.async_running, timeout); - u_green_expire_timeouts(); - - if (uwsgi.async_nevents < 0) { - continue; - } - - - for(i=0; iasync_id); - u_green_schedule_to_req(wsgi_req); - uwsgi_log("ooops\n"); - } - else { - uwsgi_log("fd ready\n"); - wsgi_req = find_wsgi_req_by_fd(uwsgi.async_events[i].ASYNC_FD, -1); - if (wsgi_req) { - u_green_schedule_to_req(wsgi_req); - } - else { - async_del(uwsgi.async_queue, uwsgi.async_events[i].ASYNC_FD, uwsgi.async_events[i].ASYNC_EV); - } - } - - } - -cycle: - - wsgi_req = uwsgi.wsgi_requests[current]; - uwsgi_log("schedule %d %d ?\n", current, wsgi_req->async_status); - if (wsgi_req->async_status != UWSGI_ACCEPTING && wsgi_req->async_status != UWSGI_PAUSED && wsgi_req->async_waiting_fd == -1 && !wsgi_req->async_timeout) { - uwsgi_log("schedule !\n"); - u_green_schedule_to_req(wsgi_req); - } - current++; - if (current >= uwsgi.async) current = 0; - - } - + return 0; } @@ -376,4 +141,6 @@ struct uwsgi_plugin ugreen_plugin = { .name = "ugreen", .init = u_green_init, + .options = ugreen_options, + .manage_opt = uwsgi_ugreen_manage_opt, }; diff --git a/tests/iobound_async.py b/tests/iobound_async.py index 4cdfff23..595c7881 100644 --- a/tests/iobound_async.py +++ b/tests/iobound_async.py @@ -7,6 +7,8 @@ def send_request(env, client): client.setblocking(1) + # test for suspend/resume + uwsgi.suspend() yield env['x-wsgiorg.fdevent.writable'](client.fileno(), 2) if env['x-wsgiorg.fdevent.timeout']: diff --git a/utils.c b/utils.c index 4c71d5f4..cbeaf325 100644 --- a/utils.c +++ b/utils.c @@ -443,6 +443,25 @@ void wsgi_req_setup(struct wsgi_request *wsgi_req, int async_id) { } +int wsgi_req_simple_recv(struct wsgi_request *wsgi_req) { + + UWSGI_SET_IN_REQUEST; + + gettimeofday(&wsgi_req->start_of_request, NULL); + + + if (!uwsgi_parse_response(&wsgi_req->poll, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT], (struct uwsgi_header *) wsgi_req, wsgi_req->buffer)) { + return -1; + } + + // enter harakiri mode + if (uwsgi.shared->options[UWSGI_OPTION_HARAKIRI] > 0) { + set_harakiri(uwsgi.shared->options[UWSGI_OPTION_HARAKIRI]); + } + + return 0; +} + int wsgi_req_recv(struct wsgi_request *wsgi_req) { UWSGI_SET_IN_REQUEST; diff --git a/uwsgi.h b/uwsgi.h index ca73ea29..b0163f75 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -66,6 +66,9 @@ #include #include #include +#ifdef UWSGI_ASYNC +#include +#endif #ifdef __sun__ #define _XPG4_2 @@ -674,6 +677,9 @@ struct wsgi_request { int async_plagued; + int suspended; + jmp_buf async_jmp_buf; + int *ovector; size_t post_cl; char *post_buffering_buf; @@ -962,8 +968,11 @@ struct uwsgi_server { struct wsgi_request *(*current_wsgi_req) (void); + jmp_buf async_jmp_buf; + // usedby suspend/resume loops void (*schedule_to_main) (struct wsgi_request *); + void (*schedule_to_req) (void); int close_on_exec; @@ -1316,6 +1325,7 @@ void uwsgi_close_request(struct wsgi_request *); void wsgi_req_setup(struct wsgi_request *, int); int wsgi_req_recv(struct wsgi_request *); +int wsgi_req_simple_recv(struct wsgi_request *); int wsgi_req_accept(struct wsgi_request *); int wsgi_req_simple_accept(struct wsgi_request *, int);