Initial revision

LICENSE

@@ -0,0 +1,21 @@
+The MIT License
+
+Copyright (c) 2011 Ruben Nine
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

README.md

@@ -0,0 +1,117 @@
+# em-postgresql-adapter
+
+PostgreSQL fiber-based ActiveRecord connection adapter for Ruby 1.9 (compatible with Rails 3).
+
+NOTE: This project is unrelated to mperham's em_postgresql (https://github.com/mperham/em_postgresql).
+
+## Installation
+
+1. Edit your Gemfile:
+
+	gem 'pg'
+	gem 'em-postgresql-adapter', :git => 'git://github.com/leftbee/em-postgresql-adapter.git'
+	gem 'rack-fiber_pool',  :require => 'rack/fiber_pool'
+
+2. Edit your environment (i.e., production.rb, staging.rb, etc.) and make sure threadsafe! is enabled:
+
+{{
+	MyRails3App::Application.configure do
+	  ...
+	  config.threadsafe!
+	  ...
+	end
+}}
+
+3. Finally, edit your config/database.yml:
+
+{{
+  production:
+    adapter: em_postgresql
+    database: myapp_production
+    username: root
+    password:
+    host: localhost
+    pool: 6
+    connections: 6
+}}
+
+## Benchmark
+
+Using a modified aync_rails app (https://github.com/igrigorik/async-rails):
+
+{{
+class WidgetsController < ApplicationController
+	def index
+  	Widget.find_by_sql("select pg_sleep(1)")
+  	render :text => "Oh hai"
+	end
+end
+}}
+
+Results:
+
+{{
+p:~ ruben$ ab -c 220 -n 2000 http://127.0.0.1:3000/widgets
+This is ApacheBench, Version 2.3 <$Revision: 655654 $>
+Copyright 1996 Adam Twiss, Zeus Technology Ltd, http://www.zeustech.net/
+Licensed to The Apache Software Foundation, http://www.apache.org/
+
+Benchmarking 127.0.0.1 (be patient)
+Completed 200 requests
+Completed 400 requests
+Completed 600 requests
+Completed 800 requests
+Completed 1000 requests
+Completed 1200 requests
+Completed 1400 requests
+Completed 1600 requests
+Completed 1800 requests
+Completed 2000 requests
+Finished 2000 requests
+
+
+Server Software:        thin
+Server Hostname:        127.0.0.1
+Server Port:            3000
+
+Document Path:          /widgets
+Document Length:        6 bytes
+
+Concurrency Level:      220
+Time taken for tests:   21.075 seconds
+Complete requests:      2000
+Failed requests:        0
+Write errors:           0
+Total transferred:      558000 bytes
+HTML transferred:       12000 bytes
+Requests per second:    94.90 [#/sec] (mean)
+Time per request:       2318.292 [ms] (mean)
+Time per request:       10.538 [ms] (mean, across all concurrent requests)
+Transfer rate:          25.86 [Kbytes/sec] received
+
+Connection Times (ms)
+              min  mean[+/-sd] median   max
+Connect:        0    1   2.3      0      10
+Processing:  1007 2219 297.2   2123    3571
+Waiting:     1007 2215 298.4   2122    3569
+Total:       1017 2220 297.0   2123    3574
+
+Percentage of the requests served within a certain time (ms)
+  50%   2123
+  66%   2128
+  75%   2134
+  80%   2580
+  90%   2645
+  95%   2702
+  98%   2865
+  99%   3013
+ 100%   3574 (longest request)
+}}
+
+## Credits
+
+Based on brianmario's mysql2 em adapter (https://github.com/brianmario/mysql2) and oldmoe's neverblock-postgresql-adapter (https://github.com/oldmoe/neverblock-postgresql-adapter).
+
+## License
+
+This software is released under the MIT license.

em-postgresql-adapter.gemspec

@@ -0,0 +1,17 @@
+Gem::Specification.new do |s|
+  s.name     = "em-postgresql-adapter"
+  s.version  = "0.1"
+  s.date     = "2011-04-23"
+  s.summary  = "PostgreSQL fiber-based ActiveRecord connection adapter for Ruby 1.9"
+  s.email    = "ruben@leftbee.net"
+  s.homepage = "http://github.com/leftbee/em-postgresql-adapter"
+  s.has_rdoc = false
+  s.authors  = ["Ruben Nine"]
+  s.files    = [
+    "em-postgresql-adapter.gemspec",
+    "README",
+    "lib/active_record/connection_adapters/em_postgresql_adapter.rb",
+    "lib/fibered_postgresql_connection.rb"
+  ]
+  s.add_dependency('pg', '>= 0.8.0')
+end

lib/active_record/connection_adapters/em_postgresql_adapter.rb

@@ -0,0 +1,170 @@
+# Require original mysql adapter as we'll just extend it
+require 'active_record/connection_adapters/postgresql_adapter'
+require 'fibered_postgresql_connection'
+
+module ActiveRecord
+  module ConnectionAdapters
+
+    def self.fiber_pools
+      @fiber_pools ||= []
+    end
+    def self.register_fiber_pool(fp)
+      fiber_pools << fp
+    end
+
+    class FiberedMonitor
+      class Queue
+        def initialize
+          @queue = []
+        end
+
+        def wait(timeout)
+          t = timeout || 5
+          fiber = Fiber.current
+          x = EM::Timer.new(t) do
+            @queue.delete(fiber)
+            fiber.resume(false)
+          end
+          @queue << fiber
+          Fiber.yield.tap do
+            x.cancel
+          end
+        end
+
+        def signal
+          fiber = @queue.pop
+          fiber.resume(true) if fiber
+        end
+      end
+
+      def synchronize
+        yield
+      end
+
+      def new_cond
+        Queue.new
+      end
+    end
+
+    # ActiveRecord's connection pool is based on threads.  Since we are working
+    # with EM and a single thread, multiple fiber design, we need to provide
+    # our own connection pool that keys off of Fiber.current so that different
+    # fibers running in the same thread don't try to use the same connection.
+    class ConnectionPool
+      def initialize(spec)
+        @spec = spec
+
+        # The cache of reserved connections mapped to threads
+        @reserved_connections = {}
+
+        # The mutex used to synchronize pool access
+        @connection_mutex = FiberedMonitor.new
+        @queue = @connection_mutex.new_cond
+
+        # default 5 second timeout unless on ruby 1.9
+        @timeout = spec.config[:wait_timeout] || 5
+
+        # default max pool size to 5
+        @size = (spec.config[:pool] && spec.config[:pool].to_i) || 5
+
+        @connections = []
+        @checked_out = []
+      end
+
+      def clear_stale_cached_connections!
+        cache = @reserved_connections
+        keys = Set.new(cache.keys)
+
+        ActiveRecord::ConnectionAdapters.fiber_pools.each do |pool|
+          pool.busy_fibers.each_pair do |object_id, fiber|
+            keys.delete(object_id)
+          end
+        end
+
+        keys.each do |key|
+          next unless cache.has_key?(key)
+          checkin cache[key]
+          cache.delete(key)
+        end
+      end
+
+      private
+
+      def current_connection_id #:nodoc:
+        Fiber.current.object_id
+      end
+
+      def checkout_and_verify(c)
+        @checked_out << c
+        c.run_callbacks :checkout
+        c.verify!
+        c
+      end
+
+      #   def checkout
+      #     # Checkout an available connection
+      #     loop do
+      #       conn = if @checked_out.size < @connections.size
+      #                checkout_existing_connection
+      #              elsif @connections.size < @size
+      #                checkout_new_connection
+      #              end
+      #       return conn if conn
+      #       # No connections available; wait for one
+      #       @waiting ||= []
+      #       Fiber.yield @waiting << Fiber.current
+      #     end
+      #   end
+      #
+      #   def checkin(conn)
+      #     conn.run_callbacks :checkin
+      #     @checked_out.delete conn
+      #     if @waiting && @waiting.size > 0
+      #       @waiting.shift.resume
+      #     end
+      #   end
+      # end
+    end # ConnectionPool
+
+    class EMPostgreSQLAdapter < ActiveRecord::ConnectionAdapters::PostgreSQLAdapter
+      # Returns 'FiberedPostgreSQL' as adapter name for identification purposes.
+      def adapter_name
+        'EMPostgreSQL'
+      end
+
+      def connect
+        @connection = ::EM::DB::FiberedPostgresConnection.connect(*@connection_parameters[1..(@connection_parameters.length-1)])
+      end
+
+      # Close then reopen the connection.
+      def reconnect!
+        disconnect!
+        connect
+      end
+    end
+
+  end # ConnectionAdapters
+
+  class Base
+    # Establishes a connection to the database that's used by all Active Record objects
+    def self.em_postgresql_connection(config) # :nodoc:
+      config = config.symbolize_keys
+      host     = config[:host]
+      port     = config[:port] || 5432
+      username = config[:username].to_s
+      password = config[:password].to_s
+      size     = config[:connections] || 4
+
+      if config.has_key?(:database)
+        database = config[:database]
+      else
+        raise ArgumentError, "No database specified. Missing argument: database."
+      end
+
+      # The postgres drivers don't allow the creation of an unconnected PGconn object,
+      # so just pass a nil connection object for the time being.
+      ::ActiveRecord::ConnectionAdapters::EMPostgreSQLAdapter.new(nil, logger, [size, host, port, nil, nil, database, username, password], config)
+    end
+  end
+
+end # ActiveRecord

lib/fibered_postgresql_connection.rb

@@ -0,0 +1,64 @@
+require 'fiber'
+require 'eventmachine'
+require 'pg'
+
+module EM
+  module DB
+    class FiberedPostgresConnection < PGconn
+
+      module Watcher
+        def initialize(client, deferable)
+          @client = client
+          @deferable = deferable
+        end
+
+        def notify_readable
+          begin
+            detach
+
+            @client.consume_input while @client.is_busy
+
+            res, data = 0, []
+            while res != nil
+              res = @client.get_result
+              data << res unless res.nil?
+            end
+
+            @deferable.succeed(data.last)
+          rescue Exception => e
+            @deferable.fail(e)
+          end
+        end
+      end
+
+      # Assuming the use of EM fiber extensions and that the exec is run in
+      # the context of a fiber. One that have the value :neverblock set to true.
+      # All neverblock IO classes check this value, setting it to false will force
+      # the execution in a blocking way.
+      def exec(sql)
+        # TODO Still not "killing the query process"-proof
+        # In some cases, the query is simply sent but the fiber never yields
+        if ::EM.reactor_running?
+          send_query sql
+          deferrable = ::EM::DefaultDeferrable.new
+          ::EM.watch(self.socket, Watcher, self, deferrable).notify_readable = true
+          fiber = Fiber.current
+          deferrable.callback do |result|
+            fiber.resume(result)
+          end
+          deferrable.errback do |err|
+            fiber.resume(err)
+          end
+          Fiber.yield.tap do |result|
+            raise result if result.is_a?(Exception)
+          end
+        else
+          super(sql)
+        end
+      end
+
+      alias_method :query, :exec
+
+    end #FiberedPostgresConnection
+  end #DB
+end #EM