better error handling for connections

lib/never_block/db/fibered_mysql_connection.rb

@@ -1,4 +1,4 @@
-require 'mysql'
+require 'mysqlplus'
 
 module NeverBlock
 
@@ -18,27 +18,48 @@ class FiberedMysqlConnection < Mysql
       # Creates a new mysql connection, sets it
       # to nonblocking and wraps the descriptor in an IO
       # object.
-	    def real_connect(*args)
-        super(*args)
-        @fd = socket
-        @io = IO.new(socket)
+	    def self.real_connect(*args)
+        me = super(*args)
+        me.init_descriptor
+        me
       end
       #alias :real_connect :initialize
       #alias :connect :initialize
+
+      def init_descriptor
+        @fd = socket
+        @io = IO.new(socket)
+      end
 
       # Assuming the use of NeverBlock 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 query(sql)
-        if Fiber.respond_to? :current and Fiber.current[:neverblock]		      
-          send_query sql
-          @fiber = Fiber.current		      
-          Fiber.yield 
-        else		      
-          super(sql)
+        begin
+          if Fiber.respond_to? :current and Fiber.current[:neverblock]		      
+            send_query sql
+            @fiber = Fiber.current		      
+            Fiber.yield
+            get_result 
+          else		      
+            super(sql)
+          end
+        rescue Exception => e
+          reconnect if e.msg.include? "not connected"
+          raise e
         end		
       end
+
+      # reset the connection
+      # and reattach to the
+      # event loop
+      def reconnect
+        unregister_from_event_loop
+        super
+        init_descriptor
+        register_with_event_loop(@loop)    
+      end
 
       # Attaches the connection socket to an event loop.
       # Currently only supports EM, but Rev support will be
@@ -73,8 +94,8 @@ def unregister_from_event_loop
 
       # The callback, this is called whenever
       # there is data available at the socket
-      def process_command
-        @fiber.resume get_result
+      def resume_command
+        @fiber.resume
       end
 
     end #FiberedPostgresConnection 
@@ -86,7 +107,7 @@ def initialize connection
         @connection = connection
       end
       def notify_readable
-        @connection.process_command
+        @connection.resume_command
       end
     end
 

lib/never_block/db/fibered_postgres_connection.rb

@@ -20,24 +20,52 @@ class FiberedPostgresConnection < PGconn
       # object.
 	    def initialize(*args)
         super(*args)
-        @fd = socket
-        @io = IO.new(socket)
+        init_descriptor
         #setnonblocking(true)
       end
-
+
+      def init_descriptor
+        @fd = socket
+        @io = IO.new(socket)
+      end  
       # Assuming the use of NeverBlock 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)
-        if Fiber.respond_to? :current and Fiber.current[:neverblock]		      
-          self.send_query sql
-          @fiber = Fiber.current		      
-          Fiber.yield 
-        else		      
-          super(sql)
+        begin
+          if Fiber.respond_to? :current and Fiber.current[:neverblock]		      
+            send_query sql
+            @fiber = Fiber.current		      
+            Fiber.yield 
+            while is_busy
+              consume_input
+              Fiber.yield if is_busy
+            end
+            res, data = 0, []
+            while res != nil
+              res = self.get_result
+              data << res unless res.nil?
+            end
+            data.last          
+          else		      
+            super(sql)
+          end
+        rescue Exception => e
+          reset if e.msg.include? "not connected"
+          raise e
         end		
       end
+
+      # reset the connection
+      # and reattach to the
+      # event loop
+      def reset
+        unregister_from_event_loop
+        super
+        init_descriptor
+        register_with_event_loop(@loop)    
+      end
 
       # Attaches the connection socket to an event loop.
       # Currently only supports EM, but Rev support will be
@@ -73,20 +101,9 @@ def unregister_from_event_loop
 
       # The callback, this is called whenever
       # there is data available at the socket
-      def process_command
-        # make sure all commands are sent
-        # before attempting to read
-        #return unless self.flush
-        self.consume_input
-        unless is_busy		
-          res, data = 0, []
-          while res != nil
-            res = self.get_result
-            data << res unless res.nil?
-          end
-          #let the fiber continue its work		      
-          @fiber.resume(data.last)
-        end
+      def resume_command
+        #let the fiber continue its work		      
+        @fiber.resume
       end
 
     end #FiberedPostgresConnection 
@@ -98,7 +115,7 @@ def initialize connection
         @connection = connection
       end
       def notify_readable
-        @connection.process_command
+        @connection.resume_command
       end
     end
 

test/test_mysql.rb

@@ -2,22 +2,26 @@
 require 'neverblock'
 require 'neverblock-mysql'
 
-@count = 10
+class Mysql
+  attr_accessor :fiber
+end
+
+@count = 100
 @connections = {}
 @fpool = NB::Pool::FiberPool.new(@count)
 @cpool = NB::Pool::FiberedConnectionPool.new(size:@count, eager:true) do
-   c = Mysql.real_connect('localhost','root',nil)
-   @connections[IO.new(c.socket)] = c
+   c = NB::DB::FiberedMysqlConnection.real_connect('localhost','root',nil)
+   @connections[c.io] = c
    c
 end
 
 @break = false
 @done = 0
 @t = Time.now
 @count.times do
-  @fpool.spawn(false) do
+  @fpool.spawn do
     @cpool.hold do |conn|
-      conn.query('select sleep(1)').each{|r| r}
+      conn.query('select sleep(1) as sleep').each{|r|p r}
       @done = @done + 1
       puts "done in #{Time.now - @t}" if @done == @count
     end
@@ -27,6 +31,6 @@
 loop do
   res = select(@sockets,nil,nil,nil)
   if res
-    res.first.each{|c|@connections[c].process_command}
+    res.first.each{|s|@connections[s].resume_command}
   end
 end

test/test_pg.rb

@@ -97,6 +97,6 @@ def stop_loop
 run_evented
 loop do
 	res = select($sockets,nil,nil,nil)
-	res.first.each{ |s|$connections[s].process_command } if res
+	res.first.each{ |s|$connections[s].resume_command } if res
 	break if $done
 end