@@ -74,10 +74,5 @@ end if $0 == __FILE__ require DIR + '/../ruby_lib/ebb' - require 'rubygems' - require 'ruby-debug' - Debugger.start - - puts "Ebb started on http://0.0.0.0:4001/" server = Ebb::start_server(SimpleApp.new, :port => 4001) end \ No newline at end of file benchmark/application.rb @@ -80,7 +80,7 @@ class ServerTest end def kill - Process.kill('KILL', @pid) + Process.kill('KILL', @pid) if @pid end def running? @@ -149,4 +149,4 @@ class ServerTest :ab_cmd => cmd } end -end \ No newline at end of file +end benchmark/server_test.rb @@ -1,18 +1,13 @@ # A ruby binding to the ebb web server # Copyright (c) 2007 Ry Dahl <ry.d4hl@gmail.com> # This software is released under the "MIT License". See README file for details. + module Ebb LIBDIR = File.dirname(__FILE__) + require LIBDIR + '/../src/ebb_ext' VERSION = File.read(LIBDIR + "/../VERSION").gsub(/\s/,'') autoload :Runner, LIBDIR + '/ebb/runner' -end - -require Ebb::LIBDIR + '/../src/ebb_ext' - -module Ebb - # "Gasp! No Server class! But this is Object Oriented Programming - we - # classes for servers!", you say. Not when there always will be - # exactly one server per virtual machine. + def self.start_server(app, options={}) port = (options[:port] || 4001).to_i #socket = options[:socket] @@ -20,20 +15,17 @@ module Ebb trap('INT') { @running = false } - FFI::server_listen_on_port(port) - + notify_fd = FFI::server_listen_on_port(port) + notifier = IO.new(notify_fd) puts "Ebb listening at http://0.0.0.0:#{port}/" - - @running = true - while FFI::server_process_connections() and @running - unless FFI::waiting_clients.empty? - if $DEBUG and $ebb_waiting_clients.length > 1 - puts "#{FFI::waiting_clients.length} waiting clients" - end - client = FFI::waiting_clients.shift + while notifier.read(1) + if client = FFI::server_next_client() + # puts "#{FFI::waiting_clients.length} waiting clients" + # p client process_client(app, client) end end + puts "Ebb unlistening" FFI::server_unlisten() end @@ -70,10 +62,8 @@ module Ebb client.finished end - module FFI - def self.waiting_clients - @waiting_clients - end + def FFI.waiting_clients + @waiting_clients end class Client ruby_lib/ebb.rb @@ -124,6 +124,47 @@ void content_length_cb(void *data, const char *at, size_t length) } +static void +on_client_writable(struct ev_loop *loop, ev_io *watcher, int revents) +{ + ebb_client *client = (ebb_client*)(watcher->data); + ssize_t sent; + + if(client->headers_sent == FALSE) + return; + + if(EV_ERROR & revents) { + g_message("on_client_writable() got error event, closing peer"); + return; + } + + //if(client->written != 0) + // g_debug("total written: %d", (int)(client->written)); + + sent = send( client->fd + , client->response_buffer->str + sizeof(gchar)*(client->written) + , client->response_buffer->len - client->written + , 0 + ); + if(sent < 0) { +#ifdef DEBUG + g_message("Error writing: %s", strerror(errno)); +#endif + ebb_client_close(client); + return; + } + client->written += sent; + + assert(client->written <= client->response_buffer->len); + //g_message("wrote %d bytes. total: %d", (int)sent, (int)(client->written)); + + ev_timer_again(loop, &(client->timeout_watcher)); + + if(client->written == client->response_buffer->len) + ebb_client_close(client); +} + + const char* localhost_str = "0.0.0.0"; void dispatch(ebb_client *client) { @@ -143,6 +184,12 @@ void dispatch(ebb_client *client) , strlen(server->port) ); } + + client->write_watcher.data = client; + ev_init (&client->write_watcher, on_client_writable); + ev_io_set (&client->write_watcher, client->fd, EV_WRITE | EV_ERROR); + ev_io_start(server->loop, &client->write_watcher); + server->request_cb(client, server->request_cb_data); } @@ -533,42 +580,6 @@ void ebb_client_close(ebb_client *client) } -void on_client_writable(struct ev_loop *loop, ev_io *watcher, int revents) -{ - ebb_client *client = (ebb_client*)(watcher->data); - ssize_t sent; - - if(EV_ERROR & revents) { - g_message("on_client_writable() got error event, closing peer"); - return; - } - - //if(client->written != 0) - // g_debug("total written: %d", (int)(client->written)); - - sent = send( client->fd - , client->response_buffer->str + sizeof(gchar)*(client->written) - , client->response_buffer->len - client->written - , 0 - ); - if(sent < 0) { -#ifdef DEBUG - g_message("Error writing: %s", strerror(errno)); -#endif - ebb_client_close(client); - return; - } - client->written += sent; - - assert(client->written <= client->response_buffer->len); - //g_message("wrote %d bytes. total: %d", (int)sent, (int)(client->written)); - - ev_timer_again(loop, &(client->timeout_watcher)); - - if(client->written == client->response_buffer->len) - ebb_client_close(client); -} - void ebb_client_write_status(ebb_client *client, int status, const char *human_status) { assert(client->status_sent == FALSE); @@ -600,7 +611,7 @@ void ebb_client_write(ebb_client *client, const char *data, int length) void ebb_client_finished(ebb_client *client) { assert(client->open); - assert(FALSE == ev_is_active(&(client->write_watcher))); + assert(ev_is_active(&client->write_watcher)); /* assure the socket is still in non-blocking mode * in the ruby binding, for example, i change this flag @@ -611,12 +622,8 @@ void ebb_client_finished(ebb_client *client) ebb_client_close(client); return; } - client->written = 0; - client->write_watcher.data = client; - ev_init (&(client->write_watcher), on_client_writable); - ev_io_set (&(client->write_watcher), client->fd, EV_WRITE | EV_ERROR); - ev_io_start(client->server->loop, &(client->write_watcher)); + client->headers_sent = TRUE; } src/ebb.c @@ -7,6 +7,8 @@ #include <fcntl.h> #include <ebb.h> #include <ev.h> +#include <pthread.h> +#include <glib.h> static VALUE cClient; static VALUE global_http_prefix; @@ -29,6 +31,10 @@ static VALUE global_http_host; */ static ebb_server *server; struct ev_loop *loop; +static notify_fd; +static pthread_mutex_t waiting_clients_lock = PTHREAD_MUTEX_INITIALIZER; +static GQueue *waiting_clients; + /* Variables with a leading underscore are C-level variables */ @@ -38,39 +44,60 @@ struct ev_loop *loop; # define RSTRING_LEN(s) (RSTRING(s)->len) #endif -void request_cb(ebb_client *client, void *data) +void request_cb(ebb_client *client, void *_) { - VALUE waiting_clients = (VALUE)data; - VALUE rb_client = Data_Wrap_Struct(cClient, 0, 0, client); - rb_ary_push(waiting_clients, rb_client); -} - -VALUE server_listen_on_port(VALUE _, VALUE port) -{ - if(ebb_server_listen_on_port(server, FIX2INT(port)) < 0) - rb_sys_fail("Problem listening on port"); - return Qnil; + pthread_mutex_lock(&waiting_clients_lock); + g_queue_push_tail(waiting_clients, (void*)client); + pthread_mutex_unlock(&waiting_clients_lock); + + assert(notify_fd > 2); + assert(1 == write(notify_fd, "N", 1)); } static void oneshot_timeout (struct ev_loop *loop, struct ev_timer *w, int revents) {;} -VALUE server_process_connections(VALUE _) +static void* process_connections(void *_) +{ + ev_loop(loop, 0); +} + +VALUE server_next_client(VALUE _) { - ev_timer timeout; - ev_timer_init (&timeout, oneshot_timeout, 0.01, 0.); - ev_timer_start (loop, &timeout); - - ev_loop(loop, EVLOOP_ONESHOT); - /* XXX: Need way to know when the loop is finished... - * should return true or false */ - - ev_timer_stop(loop, &timeout); + ebb_client *client; + VALUE rb_client; + + pthread_mutex_lock(&waiting_clients_lock); + client = g_queue_pop_head(waiting_clients); + pthread_mutex_unlock(&waiting_clients_lock); - if(server->open) - return Qtrue; + if(client == NULL) + rb_client = Qnil; else - return Qfalse; + rb_client = Data_Wrap_Struct(cClient, 0, 0, client); + return rb_client; +} + +VALUE server_listen_on_port(VALUE _, VALUE port) +{ + if(ebb_server_listen_on_port(server, FIX2INT(port)) < 0) + rb_sys_fail("Problem listening on port"); + pthread_t thread; + int fildes[2]; + assert(0 <= pipe(fildes)); + + /* the read side is blocking */ + int flags = fcntl(fildes[0], F_GETFL, 0); + assert(0 <= fcntl(fildes[0], F_SETFL, flags & ~O_NONBLOCK)); + /* the write side is nonblocking */ + flags = fcntl(fildes[1], F_GETFL, 0); + assert(0 <= fcntl(fildes[1], F_SETFL, flags | O_NONBLOCK)); + notify_fd = fildes[1]; + + assert(0 <= pthread_create(&thread, NULL, process_connections, NULL)); + pthread_detach(thread); + + return INT2FIX(fildes[0]); } @@ -215,8 +242,9 @@ void Init_ebb_ext() DEF_GLOBAL(path_info, "PATH_INFO"); DEF_GLOBAL(content_length, "CONTENT_LENGTH"); DEF_GLOBAL(http_host, "HTTP_HOST"); + + rb_define_singleton_method(mFFI, "server_next_client", server_next_client, 0); - rb_define_singleton_method(mFFI, "server_process_connections", server_process_connections, 0); rb_define_singleton_method(mFFI, "server_listen_on_port", server_listen_on_port, 1); rb_define_singleton_method(mFFI, "server_unlisten", server_unlisten, 0); @@ -231,7 +259,8 @@ void Init_ebb_ext() /* initialize ebb_server */ loop = ev_default_loop (0); server = ebb_server_alloc(); - VALUE waiting_clients = rb_ary_new(); - rb_iv_set(mFFI, "@waiting_clients", waiting_clients); - ebb_server_init(server, loop, request_cb, (void*)waiting_clients); + + waiting_clients = g_queue_new(); + + ebb_server_init(server, loop, request_cb, NULL); } src/ebb_ruby.c @@ -34,7 +34,7 @@ end dir = File.dirname(__FILE__) libev_dir = File.expand_path(dir + '/../libev') -$LDFLAGS << " -lpthread " +$LDFLAGS << " -lpthread -lprofiler " $CFLAGS << " -I#{libev_dir} " << flags.join(' ') $defs << "-DRUBY_VERSION_CODE=#{RUBY_VERSION.gsub(/\D/, '')}" src/extconf.rb @@ -10,7 +10,7 @@ PORT = 4044 class EbbTest < Test::Unit::TestCase def setup @pid = fork do - STDOUT.reopen "/dev/null", "a" + #STDOUT.reopen "/dev/null", "a" server = Ebb::start_server(self, :port => PORT) end sleep 0.5 test/basic_test.rb c023fa46c39157b09728f57fa5014d71bfcb935f Ryan Dahl ry@lakshmi.local http://github.com/ry/ebb/commit/5068077d187211464654c5b95ff4fa63c91b2c48 5068077d187211464654c5b95ff4fa63c91b2c48 2008-03-09T15:57:21-07:00 2008-03-09T15:57:21-07:00 Another attempt at moving ev_loop to a seperate thread It turns out this is signifigantly slower than the original and I will leave these changes out of the master. See http://four.livejournal.com/847870.html for some details and benchmarks 8810eb01570d41f6292525c887df38fbe2ca7621 Ryan Dahl ry@lakshmi.local