Merge pull request #333 from pitr-ch/futures

Performance improvements

Author: Petr Chalupa

examples/benchmark_new_futures.rb

@@ -1,7 +1,5 @@
 #!/usr/bin/env ruby
 
-$: << File.expand_path('../../lib', __FILE__)
-
 require 'benchmark/ips'
 require 'concurrent'
 require 'concurrent-edge'
@@ -33,7 +31,7 @@
 
 Benchmark.ips(time, warmup) do |x|
   x.report('flat-old') { Concurrent::Promise.execute { 1 }.flat_map { |v| Concurrent::Promise.execute { v + 2 } }.value! }
-  x.report('flat-new') { Concurrent.future(:fast) { 1 }.then { |v| Concurrent.future(:fast) { v+ 1 } }.flat.value! }
+  x.report('flat-new') { Concurrent.future(:fast) { 1 }.then { |v| Concurrent.future(:fast) { v + 2 } }.flat.value! }
   x.compare!
 end
 

lib/concurrent/edge/future.rb

@@ -120,11 +120,7 @@ def post_on(executor, *args, &job)
     class Event < Synchronization::Object
       include Concern::Deprecation
 
-      class State < Synchronization::Object
-        def initialize
-          ensure_ivar_visibility!
-        end
-
+      class State
         def completed?
           raise NotImplementedError
         end
@@ -331,19 +327,22 @@ def touched
         @Touched.value
       end
 
+      # only for debugging inspection
+      def waiting_threads
+        @Waiters.each.to_a
+      end
+
       private
 
       def wait_until_complete(timeout)
-        lock = Synchronization::Lock.new
-
         while true
           last_waiter = @Waiters.peek # waiters' state before completion
           break if completed?
 
           # synchronize so it cannot be signaled before it waits
           synchronize do
             # ok only if completing thread did not start signaling
-            next unless @Waiters.compare_and_push last_waiter, lock
+            next unless @Waiters.compare_and_push last_waiter, Thread.current
             ns_wait_until(timeout) { completed? }
             break
           end
@@ -407,7 +406,6 @@ def reason
       class Success < CompletedWithResult
         def initialize(value)
           @Value = value
-          super()
         end
 
         def success?
@@ -440,7 +438,6 @@ def apply(block)
       class Failed < CompletedWithResult
         def initialize(reason)
           @Reason = reason
-          super()
         end
 
         def success?
@@ -764,7 +761,6 @@ def hide_completable
     # @!visibility private
     class AbstractPromise < Synchronization::Object
       def initialize(future)
-        super(&nil)
         @Future = future
         ensure_ivar_visibility!
       end
@@ -873,7 +869,7 @@ def initialize(future, blocked_by_futures, countdown, &block)
         @Countdown = AtomicFixnum.new countdown
 
         super(future)
-        blocked_by.each { |future| future.add_callback :pr_callback_notify_blocked, self }
+        @BlockedBy.each { |future| future.add_callback :pr_callback_notify_blocked, self }
       end
 
       # @api private
@@ -893,7 +889,7 @@ def touch
         blocked_by.each(&:touch)
       end
 
-      # @api private
+      # !visibility private
       # for inspection only
       def blocked_by
         @BlockedBy
@@ -906,7 +902,7 @@ def inspect
       private
 
       def initialize_blocked_by(blocked_by_futures)
-        (@BlockedBy = Array(blocked_by_futures).freeze).size
+        @BlockedBy = Array(blocked_by_futures)
       end
 
       def clear_blocked_by!
@@ -999,8 +995,6 @@ def on_completable(done_future)
     # @!visibility private
     class ImmediatePromise < InnerPromise
       def initialize(default_executor, *args)
-        # FIXME optimize, create completed futures directly, with/without args
-
         super(if args.empty?
                 Event.new(self, default_executor).complete
               else
@@ -1011,6 +1005,8 @@ def initialize(default_executor, *args)
 
     # @!visibility private
     class FlattingPromise < BlockedPromise
+
+      # !visibility private
       def blocked_by
         @BlockedBy.each.to_a
       end
@@ -1044,7 +1040,6 @@ def initialize(blocked_by_future, levels, default_executor)
 
       def initialize_blocked_by(blocked_by_future)
         @BlockedBy = LockFreeStack.new.push(blocked_by_future)
-        1
       end
 
       def on_completable(done_future)

lib/concurrent/edge/lock_free_stack.rb

@@ -2,20 +2,26 @@ module Concurrent
   module Edge
     class LockFreeStack < Synchronization::Object
 
-      Node = ImmutableStruct.new(:value, :next) do
+      class Node
+        attr_reader :value, :next_node
+
+        def initialize(value, next_node)
+          @value     = value
+          @next_node = next_node
+        end
+
         singleton_class.send :alias_method, :[], :new
       end
 
       class Empty < Node
-        def next
+        def next_node
           self
         end
       end
 
       EMPTY = Empty[nil, nil]
 
       def initialize
-        super()
         @Head = AtomicReference.new EMPTY
         ensure_ivar_visibility!
       end
@@ -29,22 +35,25 @@ def compare_and_push(head, value)
       end
 
       def push(value)
-        @Head.update { |head| Node[value, head] }
-        self
+        while true
+          head = @Head.get
+          return self if @Head.compare_and_set head, Node[value, head]
+        end
       end
 
       def peek
         @Head.get
       end
 
       def compare_and_pop(head)
-        @Head.compare_and_set head, head.next
+        @Head.compare_and_set head, head.next_node
       end
 
       def pop
-        popped = nil
-        @Head.update { |head| (popped = head).next }
-        popped.value
+        while true
+          head = @Head.get
+          return head.value if @Head.compare_and_set head, head.next_node
+        end
       end
 
       def compare_and_clear(head)
@@ -66,7 +75,7 @@ def each
         it = peek
         until it.equal?(EMPTY)
           yield it.value
-          it = it.next
+          it = it.next_node
         end
         self
       end