From a4120efabe881641a2dc409b32558967fdaa95e5 Mon Sep 17 00:00:00 2001 From: "roberto@sirius" Date: Wed, 17 Nov 2010 07:42:51 +0100 Subject: [PATCH] added experimental fiber support for ruby1.9 --- config.ru | 5 +- logging.c | 4 +- plugins/rack/fiber.c | 83 ++++++++++++++-------- plugins/rack/rack_plugin.c | 138 +++++++++++++----------------------- plugins/rack/uwsgi_rack.h | 62 ++++++++++++++++ plugins/rack/uwsgiplugin.py | 4 +- ugreen.ru | 4 +- uwsgi.h | 2 +- uwsgiconfig.py | 10 --- 9 files changed, 175 insertions(+), 137 deletions(-) create mode 100644 plugins/rack/uwsgi_rack.h diff --git a/config.ru b/config.ru index 88cd1ed8..cc873d5d 100644 --- a/config.ru +++ b/config.ru @@ -1,13 +1,14 @@ -#require 'fiber' +require 'fiber' require 'sinatra' get '/hi' do for i in 1..10 puts "ruby" - #UWSGI.suspend() + #Fiber.yield end + "Hello World!" end diff --git a/logging.c b/logging.c index 0de0c27c..374041cf 100644 --- a/logging.c +++ b/logging.c @@ -83,7 +83,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 %d 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 msecs%s(%.*s %d) %d headers in %llu bytes (%d switches on core %d)\n", (int) uwsgi.mypid, wsgi_req->app_id, app_req, @@ -101,7 +101,7 @@ void log_request(struct wsgi_request *wsgi_req) { wsgi_req->protocol_len, wsgi_req->protocol, wsgi_req->status, wsgi_req->header_cnt, - wsgi_req->headers_size, + (unsigned long long) wsgi_req->headers_size, wsgi_req->switches, wsgi_req->async_id); logvec[logvecpos].iov_base = logpkt; diff --git a/plugins/rack/fiber.c b/plugins/rack/fiber.c index f3708d20..784a3e6f 100644 --- a/plugins/rack/fiber.c +++ b/plugins/rack/fiber.c @@ -1,15 +1,20 @@ -#include "../../uwsgi.h" +#include "uwsgi_rack.h" extern struct uwsgi_server uwsgi; - -#include +extern struct uwsgi_rack ur; VALUE fiber_list[200]; +void uwsgi_ruby_exception(void); VALUE fiber_request(VALUE core_id) { + uwsgi_log("i am the fiber\n"); + int async_id = NUM2INT(core_id); + + uwsgi_log("i am the fiber %d\n", async_id); + struct wsgi_request *wsgi_req = uwsgi.wsgi_requests[async_id]; @@ -26,20 +31,30 @@ VALUE fiber_request(VALUE core_id) { if (wsgi_req_accept(wsgi_req)) { continue; } + + uwsgi_log("request on fiber %d accepted\n", async_id); wsgi_req->async_status = UWSGI_OK; + // reinitialize switches counter + + wsgi_req->switches = 0; if (wsgi_req_recv(wsgi_req)) { continue; } + uwsgi_log("FIBER %d HAS DONE\n", async_id); while(wsgi_req->async_status == UWSGI_AGAIN) { + uwsgi_log("ASYNC APP DETECTED\n"); rb_fiber_yield(0, NULL); - wsgi_req->async_status = uwsgi.shared->hook_request[wsgi_req->uh.modifier1](wsgi_req); + wsgi_req->async_status = uwsgi.p[wsgi_req->uh.modifier1]->request(wsgi_req); } + + uwsgi_log("A LAST YIELD FOR %d\n", async_id); rb_fiber_yield(0, NULL); + uwsgi_log("CLOSING REQUEST\n"); uwsgi_close_request(wsgi_req); } @@ -48,42 +63,50 @@ VALUE fiber_request(VALUE core_id) { } -VALUE fiber_create(VALUE core_id) { +VALUE protected_fiber_loop() { - return rb_fiber_new( fiber_request, core_id ); -} + int i, current = 0; + VALUE core_id; -VALUE fiber_resume(VALUE core_id) { + // create a ruby fiber for each async core + + uwsgi_log("create a fiber for each async core...\n"); + for(i=0;iswitches++; + rb_fiber_resume(fiber_list[current], 1, &core_id); + current++; + if (current >= uwsgi.async) current = 0; + } + - rb_fiber_resume( fiber_list[NUM2INT(core_id)], 0, NULL ); - uwsgi_log("fiber yielded\n"); return Qnil; } void fiber_loop() { + int error; - int i; - int current = 0; - // create a ruby fiber for each async core - - uwsgi_log("create a fiber for each async core...\n"); - for(i=0;i= uwsgi.async) current = 0; - } - - + // never here } diff --git a/plugins/rack/rack_plugin.c b/plugins/rack/rack_plugin.c index 2340c3c6..f50d413a 100644 --- a/plugins/rack/rack_plugin.c +++ b/plugins/rack/rack_plugin.c @@ -1,66 +1,8 @@ -#include "../../uwsgi.h" - -#include +#include "uwsgi_rack.h" extern struct uwsgi_server uwsgi; -#define LONG_ARGS_RACK_BASE 17000 + (7 * 100) -#define LONG_ARGS_RAILS LONG_ARGS_RACK_BASE + 1 -#define LONG_ARGS_RUBY_GC_FREQ LONG_ARGS_RACK_BASE + 2 -#define LONG_ARGS_RACK LONG_ARGS_RACK_BASE + 3 - -#ifndef RUBY19 - #define rb_errinfo() ruby_errinfo - #define RUBY_GVL_LOCK - #define RUBY_GVL_UNLOCK -#else - //void fiber_loop(void); - #ifdef UWSGI_THREADING - #define RUBY_GVL_LOCK if (uwsgi.threads > 1) {\ - pthread_mutex_lock(&ur.gvl);\ - } - - #define RUBY_GVL_UNLOCK if (uwsgi.threads > 1) {\ - pthread_mutex_unlock(&ur.gvl);\ - } - #else - #define RUBY_GVL_LOCK - #define RUBY_GVL_UNLOCK - #endif -#endif - -#ifndef RARRAY_LEN -#define RARRAY_LEN(x) RARRAY(x)->len -#endif - -#ifndef RARRAY_PTR -#define RARRAY_PTR(x) RARRAY(x)->ptr -#endif - -#ifndef RSTRING_PTR -#define RSTRING_PTR(x) RSTRING(x)->ptr -#endif - -#ifndef RSTRING_LEN -#define RSTRING_LEN(x) RSTRING(x)->len -#endif - -struct uwsgi_rack { - - char *rails; - char *rack; - int gc_freq; - uint64_t cycles; - - int call_gc; - VALUE dispatcher; - VALUE rb_uwsgi_io_class; - ID call; - VALUE fibers[200]; - - pthread_mutex_t gvl; - -} ur; +struct uwsgi_rack ur; struct option uwsgi_rack_options[] = { @@ -72,7 +14,7 @@ struct option uwsgi_rack_options[] = { }; -static void uwsgi_ruby_exception(void) { +void uwsgi_ruby_exception(void) { VALUE lasterr = rb_gv_get("$!"); VALUE message = rb_obj_as_string(lasterr); @@ -313,11 +255,7 @@ RUBY_GLOBAL_SETUP #endif VALUE require_rack(VALUE arg) { -#ifdef RUBY19 - return rb_require("rack"); -#else return rb_funcall(rb_cObject, rb_intern("require"), 1, rb_str_new2("rack")); -#endif } VALUE require_rails(VALUE arg) { @@ -330,6 +268,12 @@ VALUE require_rails(VALUE arg) { VALUE init_rack_app(VALUE); +#ifdef RUBY19 +VALUE uwsgi_ruby_fiber_yield() { + return rb_fiber_yield(0, NULL); +} +#endif + VALUE uwsgi_ruby_suspend(VALUE *arg) { struct wsgi_request *wsgi_req = current_wsgi_req(); @@ -365,16 +309,20 @@ int uwsgi_rack_init(){ #else ruby_init(); - ruby_script("uwsgi"); ruby_init_loadpath(); #endif + ruby_script("uwsgi"); + #ifdef RUBY19 - //uwsgi_register_loop( (char *) "fiber", fiber_loop); + uwsgi_register_loop( (char *) "fiber", fiber_loop); #endif VALUE rb_uwsgi_embedded = rb_define_module("UWSGI"); rb_define_module_function(rb_uwsgi_embedded, "suspend", uwsgi_ruby_suspend, 0); +#ifdef RUBY19 + rb_define_module_function(rb_uwsgi_embedded, "fiber_yield", uwsgi_ruby_fiber_yield, 0); +#endif if (ur.rack) { @@ -448,7 +396,7 @@ VALUE call_dispatch(VALUE env) { VALUE send_body(VALUE obj) { struct wsgi_request *wsgi_req = current_wsgi_req(); - size_t len; + ssize_t len = 0; int fd = wsgi_req->poll.fd; //uwsgi_log("sending body\n"); @@ -459,22 +407,13 @@ VALUE send_body(VALUE obj) { uwsgi_log("UNMANAGED BODY TYPE %d\n", TYPE(obj)); } + wsgi_req->response_size += len; + + uwsgi_log("body sent forcore %d\n", wsgi_req->async_id); + return Qnil; } -VALUE safe_each(VALUE arg) { - - int error = 0; - VALUE result; - - result = rb_protect(rb_each, arg, &error); - if (error) { - uwsgi_ruby_exception(); - return Qnil; - } - return result; -} - VALUE iterate_body(VALUE body) { #ifdef RUBY19 @@ -562,7 +501,7 @@ int uwsgi_rack_request(struct wsgi_request *wsgi_req) { } - RUBY_GVL_LOCK + //RUBY_GVL_LOCK env = rb_hash_new(); @@ -602,13 +541,19 @@ int uwsgi_rack_request(struct wsgi_request *wsgi_req) { rb_hash_aset(env, rb_str_new2("rack.errors"), rb_funcall( rb_const_get(rb_cObject, rb_intern("IO")), rb_intern("new"), 2, INT2NUM(2), rb_str_new("w",1) )); - ret = rb_protect( call_dispatch, env, &error); + if (ur.unprotected) { + ret = rb_funcall(ur.dispatcher, ur.call, 1, env); + } + else { + ret = rb_protect( call_dispatch, env, &error); + } if (error) { uwsgi_ruby_exception(); //return -1; } + uwsgi_log("ready to manage response\n"); if (TYPE(ret) == T_ARRAY) { if (RARRAY_LEN(ret) != 3) { uwsgi_log("Invalid RACK response size: %d\n", RARRAY_LEN(ret)); @@ -648,7 +593,7 @@ int uwsgi_rack_request(struct wsgi_request *wsgi_req) { wsgi_req->hvec[5].iov_len = 2 ; //RUBY_GVL_UNLOCK - if ( !(wsgi_req->response_size = writev(wsgi_req->poll.fd, wsgi_req->hvec, 6)) ) { + if ( !(wsgi_req->headers_size = writev(wsgi_req->poll.fd, wsgi_req->hvec, 6)) ) { uwsgi_error("writev()"); } //RUBY_GVL_LOCK @@ -681,18 +626,31 @@ int uwsgi_rack_request(struct wsgi_request *wsgi_req) { } else if (rb_respond_to( body, rb_intern("each") )) { - rb_protect( iterate_body, body, &error); - if (error) { - uwsgi_ruby_exception(); + if (ur.unprotected) { +#ifdef RUBY19 + rb_block_call(body, rb_intern("each"), 0, 0, send_body, 0); + uwsgi_log("CORE %d HAS FINISHED LOOPING BODY\n", wsgi_req->async_id); +#else + rb_iterate(rb_each, body, send_body, 0); +#endif + } + else { + rb_protect( iterate_body, body, &error); + if (error) { + uwsgi_ruby_exception(); + } } } if (rb_respond_to( body, rb_intern("close") )) { //uwsgi_log("calling close\n"); + uwsgi_log("CALLING CLOSE ON CORE %d\n", wsgi_req->async_id); rb_funcall( body, rb_intern("close"), 0); } //fine: + + uwsgi_log("UNREGISTERING OBJECTS ON CORE %d\n", wsgi_req->async_id); /* unregister all the objects created */ rb_gc_unregister_address(&status); rb_gc_unregister_address(&headers); @@ -727,7 +685,7 @@ int uwsgi_rack_request(struct wsgi_request *wsgi_req) { #endif } - RUBY_GVL_UNLOCK + //RUBY_GVL_UNLOCK ur.cycles++; @@ -794,12 +752,12 @@ VALUE init_rack_app( VALUE script ) { VALUE rack = rb_const_get(rb_cObject, rb_intern("Rack")); VALUE rackup = rb_funcall( rb_const_get(rack, rb_intern("Builder")), rb_intern("parse_file"), 1, script); if (TYPE(rackup) != T_ARRAY) { - uwsgi_log("unable to parse %s file\n", RSTRING(script)->ptr); + uwsgi_log("unable to parse %s file\n", RSTRING_PTR(script)); return Qnil; } if (RARRAY_LEN(rackup) < 1) { - uwsgi_log("invalid rack config file: %s\n", RSTRING(script)->ptr); + uwsgi_log("invalid rack config file: %s\n", RSTRING_PTR(script)); return Qnil; } diff --git a/plugins/rack/uwsgi_rack.h b/plugins/rack/uwsgi_rack.h new file mode 100644 index 00000000..1dc70b56 --- /dev/null +++ b/plugins/rack/uwsgi_rack.h @@ -0,0 +1,62 @@ +#include "../../uwsgi.h" + +#include + +#define LONG_ARGS_RACK_BASE 17000 + (7 * 100) +#define LONG_ARGS_RAILS LONG_ARGS_RACK_BASE + 1 +#define LONG_ARGS_RUBY_GC_FREQ LONG_ARGS_RACK_BASE + 2 +#define LONG_ARGS_RACK LONG_ARGS_RACK_BASE + 3 + +#ifndef RUBY19 + #define rb_errinfo() ruby_errinfo + #define RUBY_GVL_LOCK + #define RUBY_GVL_UNLOCK +#else + void fiber_loop(void); + #ifdef UWSGI_THREADING + #define RUBY_GVL_LOCK if (uwsgi.threads > 1) {\ + pthread_mutex_lock(&ur.gvl);\ + } + + #define RUBY_GVL_UNLOCK if (uwsgi.threads > 1) {\ + pthread_mutex_unlock(&ur.gvl);\ + } + #else + #define RUBY_GVL_LOCK + #define RUBY_GVL_UNLOCK + #endif +#endif + +#ifndef RARRAY_LEN +#define RARRAY_LEN(x) RARRAY(x)->len +#endif + +#ifndef RARRAY_PTR +#define RARRAY_PTR(x) RARRAY(x)->ptr +#endif + +#ifndef RSTRING_PTR +#define RSTRING_PTR(x) RSTRING(x)->ptr +#endif + +#ifndef RSTRING_LEN +#define RSTRING_LEN(x) RSTRING(x)->len +#endif + +struct uwsgi_rack { + + char *rails; + char *rack; + int gc_freq; + uint64_t cycles; + + int call_gc; + VALUE dispatcher; + VALUE rb_uwsgi_io_class; + ID call; + VALUE fibers[200]; + int unprotected; + + pthread_mutex_t gvl; + +}; diff --git a/plugins/rack/uwsgiplugin.py b/plugins/rack/uwsgiplugin.py index 6d8f8dec..b978435a 100644 --- a/plugins/rack/uwsgiplugin.py +++ b/plugins/rack/uwsgiplugin.py @@ -12,9 +12,12 @@ CFLAGS = os.popen(RUBYPATH + " -e \"require 'rbconfig';print Config::CONFIG['CFL version = os.popen(RUBYPATH + " -e \"print RUBY_VERSION\"").read().rstrip() v = version.split('.') +GCC_LIST = ['rack_plugin'] + if v[0] == '1' and v[1] == '9': CFLAGS.append('-DRUBY19') CFLAGS.append('-Wno-unused-parameter') + GCC_LIST.append('fiber') includedir = os.popen(RUBYPATH + " -e \"require 'rbconfig';print Config::CONFIG['rubyhdrdir']\"").read().rstrip() if includedir == 'nil': @@ -34,5 +37,4 @@ libpath = os.popen(RUBYPATH + " -e \"require 'rbconfig';print Config::CONFIG['li LDFLAGS.append('-L' + libpath ) os.environ['LD_RUN_PATH'] = libpath LIBS = os.popen(RUBYPATH + " -e \"require 'rbconfig';print '-l' + Config::CONFIG['RUBY_SO_NAME']\"").read().rstrip().split() -GCC_LIST = ['rack_plugin'] diff --git a/ugreen.ru b/ugreen.ru index 7b6d3e5b..25593603 100644 --- a/ugreen.ru +++ b/ugreen.ru @@ -1,8 +1,10 @@ +require 'fiber' class SuspendingBody def each - for i in 1..100 + for i in 1..10 yield "numero: #{i}\n" + Fiber.yield end end diff --git a/uwsgi.h b/uwsgi.h index 0991e59b..20ef4b55 100644 --- a/uwsgi.h +++ b/uwsgi.h @@ -513,7 +513,7 @@ struct wsgi_request { int status; size_t response_size; - int headers_size; + ssize_t headers_size; int async_id; int async_status; diff --git a/uwsgiconfig.py b/uwsgiconfig.py index 6c7caa47..52b2caef 100644 --- a/uwsgiconfig.py +++ b/uwsgiconfig.py @@ -171,16 +171,6 @@ class uConf(): def get_gcll(self): kvm_list = ['FreeBSD', 'OpenBSD', 'NetBSD', 'DragonFly'] - if uwsgi_os == 'Darwin': - # build a universal binary on osx - self.cflags.append('-arch i386') - self.cflags.append('-arch x86_64') - self.cflags.append('-arch ppc') - - self.ldflags.append('-arch i386') - self.ldflags.append('-arch x86_64') - self.ldflags.append('-arch ppc') - if uwsgi_os == 'SunOS': self.libs.append('-lsendfile') if not uwsgi_os_v.startswith('Nexenta'):