Changeset 611

Timestamp:

12/06/07 07:31:39 (1 year ago)

Author:

blackhedd

Message:

added heartbeat and inactivity timers to pure Ruby

Files:

• version_0/lib/pr_eventmachine.rb (modified) (13 diffs)
• version_0/tests/test_pure.rb (modified) (1 diff)

version_0/lib/pr_eventmachine.rb

@@ -194 +194 @@
-    def set_comm_inactivity_timeout sig, tm
-      r = Reactor.instance.get_selectable( sig ) or raise "unknown set_comm_inactivity_timeout target"
-p "?????????????" # TODO, IMPLEMENT THIS.
+r.set_inactivity_timeout tm
-    end
-  end
@@ -248 +248 @@
-
-  TimerFired = 100
-       
-       
-       
-       
-       
+  
+  
+  
+  
+  
-
-end
@@ -261 +261 @@
-class Reactor
-  include Singleton
+
+  HeartbeatInterval = 2
+
+  attr_reader :current_loop_time
-
-  def initialize
@@ -284 +288 @@
-    @timers = SortedSet.new # []
-    set_timer_quantum(0.1)
+    @current_loop_time = Time.now
+    @next_heartbeat = @current_loop_time + HeartbeatInterval
-  end
-
@@ -302 +308 @@
-
-      loop {
+        @current_loop_time = Time.now
+
-        break if @stop_scheduled
-        run_timers
-        break if @stop_scheduled
-        crank_selectables
+        break if @stop_scheduled
+        run_heartbeats
-      }
-    ensure
@@ -318 +328 @@
-
-  def run_timers
-    now = Time.now
-    @timers.each {|t|
-now
+@current_loop_time
-        @timers.delete t
-        EventMachine::event_callback "", TimerFired, t.last
@@ -331 +340 @@
-    #  EventMachine::event_callback "", TimerFired, t.last
-    #end
+  end
+
+  def run_heartbeats
+    if @next_heartbeat <= @current_loop_time
+      @next_heartbeat = @current_loop_time + HeartbeatInterval
+      @selectables.each {|k,io| io.heartbeat}
+    end
-  end
-
@@ -397 +413 @@
-  def_delegator :@my_selectable, :send_datagram
-  def_delegator :@my_selectable, :get_outbound_data_size
+  def_delegator :@my_selectable, :set_inactivity_timeout
+  def_delegator :@my_selectable, :heartbeat
-end
-
@@ -409 +427 @@
-      @uuid = UuidGenerator.generate
-      @io = io
+      @last_activity = Reactor.instance.current_loop_time
-
-      m = @io.fcntl(Fcntl::F_GETFL, 0)
@@ -437 +456 @@
-    end
-
+    def set_inactivity_timeout tm
+      @inactivity_timeout = tm
+    end
+
+    def heartbeat
+    end
-  end
-
@@ -477 +502 @@
-    # If we have it, then we can read multiple times safely to improve
-    # performance.
+    # The last-activity clock ASSUMES that we only come here when we
+    # have selected readable.
-    # TODO, coalesce multiple reads into a single event.
-    # TODO, do the function check somewhere else and cache it.
-    def eventable_read
+      @last_activity = Reactor.instance.current_loop_time
-      begin
-        if io.respond_to?(:read_nonblock)
@@ -509 +537 @@
-    # connections. Also we should coalesce small writes.
-    # URGENT TODO: Coalesce small writes. They are a performance killer.
+    # The last-activity recorder ASSUMES we'll only come here if we've
+    # selected writable.
-    def eventable_write
-      # coalesce the outbound array here, perhaps
+      @last_activity = Reactor.instance.current_loop_time
-      while data = @outbound_q.shift do
-        begin
@@ -565 +596 @@
-    end
-
+    def heartbeat
+      if @inactivity_timeout and (@last_activity + @inactivity_timeout) < Reactor.instance.current_loop_time
+        schedule_close true
+      end
+    end
-  end
-

version_0/tests/test_pure.rb

@@ -113 +113 @@
-  end
-
+
+
+  def test_reactor_running
+    a = false
+    EM.run {
+      a = EM.reactor_running?
+      EM.next_tick {EM.stop}
+    }
+    assert a
+  end
+
+
-end