primitive fiber aware connection pool

.gitignore

@@ -1,19 +1,9 @@
-ext/*
-!ext/*.cc
-!ext/*.h
-!ext/*.rb
-!ext/swift
-ext/swift/*
-!ext/swift/*.cc
-!ext/swift/*.rb
 benchmarks/gems/ruby/1.9.1/*
 !benchmarks/gems/ruby/1.9.1/cache
 benchmarks/vendor/*
 benchmarks/gems/environment.rb
 benchmarks/bin
 pkg/*
-memory/driver
-memory/driver.o
 dbtest.rb
 benchmarks/adapters/pg_ext/*.*o
 benchmarks/adapters/pg_ext/Makefile
@@ -25,3 +15,4 @@ bin/
 gems/
 .bundle/
 tmp/
+*.gem

Gemfile

@@ -33,4 +33,8 @@ group :test do
   gem 'swift-db-sqlite3'
   gem 'swift-db-postgres'
   gem 'swift-db-mysql'
+
+  # async
+  gem 'eventmachine'
+  gem 'em-synchrony'
 end

Gemfile.lock

@@ -39,6 +39,9 @@ GEM
       data_objects (= 0.10.8)
     do_sqlite3 (0.10.8)
       data_objects (= 0.10.8)
+    em-synchrony (1.0.2)
+      eventmachine (>= 1.0.0.beta.1)
+    eventmachine (1.0.0)
     home_run (1.0.6)
     i18n (0.6.0)
     multi_json (1.3.6)
@@ -74,6 +77,8 @@ DEPENDENCIES
   do_mysql
   do_postgres
   do_sqlite3
+  em-synchrony
+  eventmachine
   home_run
   i18n
   mysql2

README.md

@@ -303,6 +303,24 @@ or use the `em-synchrony` api for `swift`
   end
 ```
 
+### Fibers and Connection Pools
+
+If you intend to use `Swift::Record` with `em-synchrony` you will need to use a fiber aware connection pool.
+
+```ruby
+require 'swift/fiber_connection_pool'
+
+EM.run do
+  Swift.setup_connection_pool 5, :default, Swift::Adapter::Postgres, db: 'swift'
+
+  5.times do
+    EM.synchrony do
+      p User.execute("select * from users").entries
+    end
+  end
+end
+```
+
 ## Performance
 
 Swift prefers performance when it doesn't compromise the Ruby-ish interface. It's unfair to compare Swift to DataMapper

lib/swift/fiber_connection_pool.rb

@@ -0,0 +1,80 @@
+# Based on EM::Synchrony::ConnectionPool
+
+require 'swift/synchrony'
+
+module Swift
+  class FiberConnectionPool
+
+    def initialize opts, &block
+      @reserved  = {}   # map of in-progress connections
+      @available = []   # pool of free connections
+      @pending   = []   # pending reservations (FIFO)
+
+      opts[:size].times do
+        @available.push(block.call)
+      end
+    end
+
+    private
+      # Choose first available connection and pass it to the supplied
+      # block. This will block indefinitely until there is an available
+      # connection to service the request.
+      def __reserve__
+        id    = "#{Fiber.current.object_id}:#{rand}"
+        fiber = Fiber.current
+        begin
+          yield acquire(id, fiber)
+        ensure
+          release(id)
+        end
+      end
+
+      # Acquire a lock on a connection and assign it to executing fiber
+      # - if connection is available, pass it back to the calling block
+      # - if pool is full, yield the current fiber until connection is available
+      def acquire id, fiber
+        if conn = @available.pop
+          @reserved[id] = conn
+        else
+          Fiber.yield @pending.push(fiber)
+          acquire(id, fiber)
+        end
+      end
+
+      # Release connection assigned to the supplied fiber and
+      # resume any other pending connections (which will
+      # immediately try to run acquire on the pool)
+      def release(id)
+        @available.push(@reserved.delete(id))
+        if pending = @pending.shift
+          pending.resume
+        end
+      end
+
+      # Allow the pool to behave as the underlying connection
+      def method_missing method, *args, &blk
+        __reserve__ do |conn|
+          conn.__send__(method, *args, &blk)
+        end
+      end
+  end # FiberConnectionPool
+
+  def self.setup_connection_pool size, name, klass, *args
+    (@repositories ||= {})[name] = FiberConnectionPool.new(size: size) {klass.new(*args)}
+  end
+
+  class Adapter::Sql
+    def serialized_transaction &block
+      Swift.scopes.push(self)
+      execute('begin')
+      res = yield(self)
+      execute('commit')
+      res
+    rescue => e
+      execute('rollback')
+      raise e
+    ensure
+      Swift.scopes.pop
+    end
+  end # Adapter::Sql
+end # Swift

test/test_synchrony.rb

@@ -0,0 +1,41 @@
+require 'helper'
+
+describe 'fiber connection pool' do
+  before do
+    skip 'swift/synchrony re-defines Adapter#execute'
+
+    require 'swift/fiber_connection_pool'
+    EM.synchrony do
+      Swift.setup(:default, Swift::Adapter::Postgres, db: 'swift_test')
+      Swift.db.execute('drop table if exists users')
+      Swift.db.execute('create table users(id serial primary key, name text)')
+
+      @user = Class.new(Swift::Record) do
+        store :users
+        attribute :id,   Swift::Type::Integer, key: true, serial: true
+        attribute :name, Swift::Type::String
+      end
+
+      10.times { @user.create(name: 'test') }
+      EM.stop
+    end
+  end
+
+  it 'can synchronize queries across fibers' do
+    EM.run do
+      Swift.setup_connection_pool 2, :default, Swift::Adapter::Postgres, db: 'swift_test'
+
+      @counts = []
+      5.times do
+        EM.synchrony do
+          @counts << @user.execute('select * from users').selected_rows
+        end
+      end
+
+      EM.add_timer(0.2) { EM.stop }
+    end
+
+    assert_equal 5,    @counts.size
+    assert_equal [10], @counts.uniq
+  end
+end