Merge pull request #283 from ruby-concurrency/synchronization-executor-service

Synchronization ExecutorService

Author: jdantonio

examples/benchmark_thread_pool_implementations.rb

@@ -0,0 +1,28 @@
+#!/usr/bin/env ruby
+
+$: << File.expand_path('../../lib', __FILE__)
+
+require 'benchmark'
+require 'concurrent/executors'
+
+COUNT = 100_000
+
+EXECUTORS = [
+  [Concurrent::JavaCachedThreadPool],
+  [Concurrent::JavaFixedThreadPool, 10],
+  [Concurrent::JavaSingleThreadExecutor],
+  [Concurrent::JavaThreadPoolExecutor]
+]
+
+Benchmark.bmbm do |x|
+  EXECUTORS.each do |executor_class, *args|
+    x.report(executor_class.to_s) do
+      if args.empty?
+        executor = executor_class.new
+      else
+        executor = executor_class.new(*args)
+      end
+      COUNT.times { executor.post{} }
+    end
+  end
+end

ext/com/concurrent_ruby/ext/SynchronizationLibrary.java

@@ -55,14 +55,14 @@ public IRubyObject initialize(ThreadContext context, IRubyObject[] args, Block b
             }
         }
 
-        @JRubyMethod(name = "synchronize", visibility = Visibility.PRIVATE)
+        @JRubyMethod(name = "synchronize", visibility = Visibility.PROTECTED)
         public IRubyObject rubySynchronize(ThreadContext context, Block block) {
             synchronized (this) {
                 return block.yield(context, null);
             }
         }
 
-        @JRubyMethod(name = "ns_wait", optional = 1, visibility = Visibility.PRIVATE)
+        @JRubyMethod(name = "ns_wait", optional = 1, visibility = Visibility.PROTECTED)
         public IRubyObject nsWait(ThreadContext context, IRubyObject[] args) {
             Ruby runtime = context.runtime;
             if (args.length > 1) {
@@ -94,19 +94,19 @@ public IRubyObject nsWait(ThreadContext context, IRubyObject[] args) {
             return this;
         }
 
-        @JRubyMethod(name = "ns_signal", visibility = Visibility.PRIVATE)
+        @JRubyMethod(name = "ns_signal", visibility = Visibility.PROTECTED)
         public IRubyObject nsSignal(ThreadContext context) {
             notify();
             return this;
         }
 
-        @JRubyMethod(name = "ns_broadcast", visibility = Visibility.PRIVATE)
+        @JRubyMethod(name = "ns_broadcast", visibility = Visibility.PROTECTED)
         public IRubyObject nsBroadcast(ThreadContext context) {
             notifyAll();
             return this;
         }
 
-        @JRubyMethod(name = "ensure_ivar_visibility!", visibility = Visibility.PRIVATE)
+        @JRubyMethod(name = "ensure_ivar_visibility!", visibility = Visibility.PROTECTED)
         public IRubyObject ensureIvarVisibilityBang(ThreadContext context) {
             return context.nil;
         }

lib/concurrent/actor/core.rb

@@ -1,3 +1,5 @@
+require 'concurrent/executors'
+
 module Concurrent
   module Actor
 
@@ -47,58 +49,7 @@ class Core < Synchronization::Object
       #   any logging system
       # @param [Proc] block for class instantiation
       def initialize(opts = {}, &block)
-        super(&nil) # TODO use ns_initialize
-        synchronize do
-          @mailbox              = Array.new
-          @serialized_execution = SerializedExecution.new
-          @children             = Set.new
-
-          @context_class = Child! opts.fetch(:class), AbstractContext
-          allocate_context
-
-          @executor = Type! opts.fetch(:executor, Concurrent.global_io_executor), Executor
-          raise ArgumentError, 'ImmediateExecutor is not supported' if @executor.is_a? ImmediateExecutor
-
-          @reference = (Child! opts[:reference_class] || @context.default_reference_class, Reference).new self
-          @name      = (Type! opts.fetch(:name), String, Symbol).to_s
-
-          parent       = opts[:parent]
-          @parent_core = (Type! parent, Reference, NilClass) && parent.send(:core)
-          if @parent_core.nil? && @name != '/'
-            raise 'only root has no parent'
-          end
-
-          @path   = @parent_core ? File.join(@parent_core.path, @name) : @name
-          @logger = opts[:logger]
-
-          @parent_core.add_child reference if @parent_core
-
-          initialize_behaviours opts
-
-          @args       = opts.fetch(:args, [])
-          @block      = block
-          initialized = Type! opts[:initialized], Edge::CompletableFuture, NilClass
-
-          messages = []
-          messages << :link if opts[:link]
-          messages << :supervise if opts[:supervise]
-
-          schedule_execution do
-            begin
-              build_context
-
-              messages.each do |message|
-                handle_envelope Envelope.new(message, nil, parent, reference)
-              end
-
-              initialized.success reference if initialized
-            rescue => ex
-              log ERROR, ex
-              @first_behaviour.terminate!
-              initialized.fail ex if initialized
-            end
-          end
-        end
+        super
       end
 
       # @return [Reference, nil] of parent actor
@@ -199,6 +150,60 @@ def build_context
         @context.send :initialize, *@args, &@block
       end
 
+      protected
+
+      def ns_initialize(opts, &block)
+        @mailbox              = Array.new
+        @serialized_execution = SerializedExecution.new
+        @children             = Set.new
+
+        @context_class = Child! opts.fetch(:class), AbstractContext
+        allocate_context
+
+        @executor = Type! opts.fetch(:executor, Concurrent.global_io_executor), Concurrent::AbstractExecutorService
+        raise ArgumentError, 'ImmediateExecutor is not supported' if @executor.is_a? ImmediateExecutor
+
+        @reference = (Child! opts[:reference_class] || @context.default_reference_class, Reference).new self
+        @name      = (Type! opts.fetch(:name), String, Symbol).to_s
+
+        parent       = opts[:parent]
+        @parent_core = (Type! parent, Reference, NilClass) && parent.send(:core)
+        if @parent_core.nil? && @name != '/'
+          raise 'only root has no parent'
+        end
+
+        @path   = @parent_core ? File.join(@parent_core.path, @name) : @name
+        @logger = opts[:logger]
+
+        @parent_core.add_child reference if @parent_core
+
+        initialize_behaviours opts
+
+        @args       = opts.fetch(:args, [])
+        @block      = block
+        initialized = Type! opts[:initialized], Edge::CompletableFuture, NilClass
+
+        messages = []
+        messages << :link if opts[:link]
+        messages << :supervise if opts[:supervise]
+
+        schedule_execution do
+          begin
+            build_context
+
+            messages.each do |message|
+              handle_envelope Envelope.new(message, nil, parent, reference)
+            end
+
+            initialized.success reference if initialized
+          rescue => ex
+            log ERROR, ex
+            @first_behaviour.terminate!
+            initialized.fail ex if initialized
+          end
+        end
+      end
+
       private
 
       def handle_envelope(envelope)
@@ -215,7 +220,7 @@ def initialize_behaviours(opts)
         end
         @behaviours           = {}
         @first_behaviour      = @behaviour_definition.reverse.
-            reduce(nil) { |last, (behaviour, args)| @behaviours[behaviour] = behaviour.new(self, last, *args) }
+          reduce(nil) { |last, (behaviour, args)| @behaviours[behaviour] = behaviour.new(self, last, *args) }
       end
     end
   end

lib/concurrent/agent.rb

@@ -1,8 +1,8 @@
 require 'thread'
-
 require 'concurrent/dereferenceable'
 require 'concurrent/observable'
 require 'concurrent/logging'
+require 'concurrent/executor/executor'
 
 module Concurrent
 

lib/concurrent/atomic/count_down_latch.rb

@@ -58,7 +58,7 @@ def count
       synchronize { @count }
     end
 
-    private
+    protected
 
     def ns_initialize(count)
       @count = count

lib/concurrent/atomic/cyclic_barrier.rb

@@ -62,7 +62,6 @@ def wait(timeout = nil)
       end
     end
 
-
     # resets the barrier to its initial state
     # If there is at least one waiting thread, it will be woken up, the `wait`
     # method will return false and the barrier will be broken
@@ -83,7 +82,7 @@ def broken?
       synchronize { @generation.status != :waiting }
     end
 
-    private
+    protected
 
     def ns_generation_done(generation, status, continue = true)
       generation.status = status
@@ -101,6 +100,5 @@ def ns_initialize(parties, &block)
       @action  = block
       ns_next_generation
     end
-
   end
 end

lib/concurrent/atomic/event.rb

@@ -70,7 +70,7 @@ def wait(timeout = nil)
       end
     end
 
-    private
+    protected
 
     def ns_set
       unless @set

lib/concurrent/dataflow.rb

@@ -1,6 +1,5 @@
 require 'concurrent/future'
 require 'concurrent/atomic/atomic_fixnum'
-require 'concurrent/executor/per_thread_executor'
 
 module Concurrent
 

lib/concurrent/delay.rb

@@ -1,6 +1,7 @@
 require 'thread'
 require 'concurrent/configuration'
 require 'concurrent/obligation'
+require 'concurrent/executor/executor'
 require 'concurrent/executor/immediate_executor'
 require 'concurrent/synchronization'
 
@@ -68,19 +69,9 @@ class Delay < Synchronization::Object
     # @raise [ArgumentError] if no block is given
     def initialize(opts = {}, &block)
       raise ArgumentError.new('no block given') unless block_given?
-
-      super()
-      init_obligation(self)
-      set_deref_options(opts)
-      @task_executor = Executor.executor_from_options(opts)
-
-      @task      = block
-      @state     = :pending
-      @computing = false
+      super(opts, &block)
     end
 
-    protected :synchronize
-
     # Return the value this object represents after applying the options
     # specified by the `#set_deref_options` method. If the delayed operation
     # raised an exception this method will return nil. The execption object
@@ -169,6 +160,18 @@ def reconfigure(&block)
       end
     end
 
+    protected
+
+    def ns_initialize(opts, &block)
+      init_obligation(self)
+      set_deref_options(opts)
+      @task_executor = Executor.executor_from_options(opts)
+
+      @task      = block
+      @state     = :pending
+      @computing = false
+    end
+
     private
 
     # @!visibility private