IVar subclasses use ns_initialize.

Author: jdantonio

lib/concurrent/future.rb

@@ -24,11 +24,7 @@ class Future < IVar
     # @raise [ArgumentError] if no block is given
     def initialize(opts = {}, &block)
       raise ArgumentError.new('no block given') unless block_given?
-      super(IVar::NO_VALUE, opts, &nil)
-      @state = :unscheduled
-      @task = block
-      @executor = Executor.executor_from_options(opts) || Concurrent.global_io_executor
-      @args = get_arguments_from(opts)
+      super(IVar::NO_VALUE, opts.merge(__task_from_block__: block), &nil)
     end
 
     # Execute an `:unscheduled` `Future`. Immediately sets the state to `:pending` and
@@ -77,7 +73,7 @@ def self.execute(opts = {}, &block)
     # @!macro ivar_set_method
     def set(value = IVar::NO_VALUE, &block)
       check_for_block_or_value!(block_given?, value)
-      mutex.synchronize do
+      synchronize do
         if @state != :unscheduled
           raise MultipleAssignmentError
         else
@@ -87,6 +83,16 @@ def set(value = IVar::NO_VALUE, &block)
       execute
     end
 
+    protected
+
+    def ns_initialize(value, opts)
+      super
+      @state = :unscheduled
+      @task = opts[:__task_from_block__]
+      @executor = Executor.executor_from_options(opts) || Concurrent.global_io_executor
+      @args = get_arguments_from(opts)
+    end
+
     private
 
     # @!visibility private

lib/concurrent/promise.rb

@@ -203,18 +203,7 @@ class Promise < IVar
     # @see http://promises-aplus.github.io/promises-spec/
     def initialize(opts = {}, &block)
       opts.delete_if { |k, v| v.nil? }
-      super(IVar::NO_VALUE, opts, &nil)
-
-      @executor = Executor.executor_from_options(opts) || Concurrent.global_io_executor
-      @args = get_arguments_from(opts)
-
-      @parent = opts.fetch(:parent) { nil }
-      @on_fulfill = opts.fetch(:on_fulfill) { Proc.new { |result| result } }
-      @on_reject = opts.fetch(:on_reject) { Proc.new { |reason| raise reason } }
-
-      @promise_body = block || Proc.new { |result| result }
-      @state = :unscheduled
-      @children = []
+      super(IVar::NO_VALUE, opts.merge(__promise_body_from_block__: block), &nil)
     end
 
     # Create a new `Promise` and fulfill it immediately.
@@ -266,7 +255,7 @@ def execute
     def set(value = IVar::NO_VALUE, &block)
       raise PromiseExecutionError.new('supported only on root promise') unless root?
       check_for_block_or_value!(block_given?, value)
-      mutex.synchronize do
+      synchronize do
         if @state != :unscheduled
           raise MultipleAssignmentError
         else
@@ -319,7 +308,7 @@ def then(rescuer = nil, &block)
         on_reject: rescuer
       )
 
-      mutex.synchronize do
+      synchronize do
         child.state = :pending if @state == :pending
         child.on_fulfill(apply_deref_options(@value)) if @state == :fulfilled
         child.on_reject(@reason) if @state == :rejected
@@ -447,6 +436,21 @@ def self.any?(*promises)
 
     protected
 
+    def ns_initialize(value, opts)
+      super
+
+      @executor = Executor.executor_from_options(opts) || Concurrent.global_io_executor
+      @args = get_arguments_from(opts)
+
+      @parent = opts.fetch(:parent) { nil }
+      @on_fulfill = opts.fetch(:on_fulfill) { Proc.new { |result| result } }
+      @on_reject = opts.fetch(:on_reject) { Proc.new { |reason| raise reason } }
+
+      @promise_body = opts[:__promise_body_from_block__] || Proc.new { |result| result }
+      @state = :unscheduled
+      @children = []
+    end
+
     # Aggregate a collection of zero or more promises under a composite promise,
     # execute the aggregated promises and collect them into a standard Ruby array,
     # call the given Ruby `Ennnumerable` predicate (such as `any?`, `all?`, `none?`,
@@ -472,7 +476,7 @@ def self.aggregate(method, *promises)
 
     # @!visibility private
     def set_pending
-      mutex.synchronize do
+      synchronize do
         @state = :pending
         @children.each { |c| c.set_pending }
       end
@@ -503,7 +507,7 @@ def notify_child(child)
 
     # @!visibility private
     def complete(success, value, reason)
-      children_to_notify = mutex.synchronize do
+      children_to_notify = synchronize do
         set_state!(success, value, reason)
         @children.dup
       end
@@ -528,7 +532,7 @@ def set_state!(success, value, reason)
 
     # @!visibility private
     def synchronized_set_state!(success, value, reason)
-      mutex.synchronize { set_state!(success, value, reason) }
+      synchronize { set_state!(success, value, reason) }
     end
   end
 end

lib/concurrent/scheduled_task.rb

@@ -158,12 +158,14 @@ class ScheduledTask < IVar
     def initialize(delay, opts = {}, &block)
       raise ArgumentError.new('no block given') unless block_given?
       @delay = TimerSet.calculate_delay!(delay)
+      super(IVar::NO_VALUE, opts.merge(__task_from_block__: block), &nil)
+    end
 
-      super(IVar::NO_VALUE, opts, &nil)
-
+    def ns_initialize(value, opts)
+      super
       self.observers = CopyOnNotifyObserverSet.new
       @state         = :unscheduled
-      @task          = block
+      @task          = opts[:__task_from_block__]
       @executor      = Executor.executor_from_options(opts) || Concurrent.global_io_executor
     end