Merge pull request #828 from AlexanderZagaynov/threads_naming

Threads naming

Author: pitr-ch

lib/concurrent/executor/abstract_executor_service.rb

@@ -16,10 +16,19 @@ class AbstractExecutorService < Synchronization::LockableObject
     # @!macro executor_service_attr_reader_fallback_policy
     attr_reader :fallback_policy
 
+    attr_reader :name
+
     # Create a new thread pool.
-    def initialize(*args, &block)
+    def initialize(opts = {}, &block)
       super(&nil)
-      synchronize { ns_initialize(*args, &block) }
+      synchronize do
+        ns_initialize(opts, &block)
+        @name = opts.fetch(:name) if opts.key?(:name)
+      end
+    end
+
+    def to_s
+      name ? "#{super[0..-2]} name: #{name}>" : super
     end
 
     # @!macro executor_service_method_shutdown

lib/concurrent/executor/fixed_thread_pool.rb

@@ -115,6 +115,9 @@ module Concurrent
   #   Thread pools support several configuration options:
   #
   #   * `idletime`: The number of seconds that a thread may be idle before being reclaimed.
+  #   * `name`: The name of the executor (optional). Printed in the executor's `#to_s` output and
+  #     a `<name>-worker-<id>` name is given to its threads if supported by used Ruby
+  #     implementation. `<id>` is uniq for each thread.
   #   * `max_queue`: The maximum number of tasks that may be waiting in the work queue at
   #     any one time. When the queue size reaches `max_queue` and no new threads can be created,
   #     subsequent tasks will be rejected in accordance with the configured `fallback_policy`.

lib/concurrent/executor/java_executor_service.rb

@@ -18,10 +18,6 @@ class JavaExecutorService < AbstractExecutorService
       }.freeze
       private_constant :FALLBACK_POLICY_CLASSES
 
-      def initialize(*args, &block)
-        super
-      end
-
       def post(*args, &task)
         raise ArgumentError.new('no block given') unless block_given?
         return handle_fallback(*args, &task) unless running?

lib/concurrent/executor/ruby_thread_pool_executor.rb

@@ -131,6 +131,7 @@ def ns_initialize(opts)
       @scheduled_task_count = 0
       @completed_task_count = 0
       @largest_length       = 0
+      @workers_counter      = 0
       @ruby_pid             = $$ # detects if Ruby has forked
 
       @gc_interval  = opts.fetch(:gc_interval, @idletime / 2.0).to_i # undocumented
@@ -224,7 +225,8 @@ def ns_worker_died(worker)
     def ns_add_busy_worker
       return if @pool.size >= @max_length
 
-      @pool << (worker = Worker.new(self))
+      @workers_counter += 1
+      @pool << (worker = Worker.new(self, @workers_counter))
       @largest_length = @pool.length if @pool.length > @largest_length
       worker
     end
@@ -284,6 +286,7 @@ def ns_reset_if_forked
         @scheduled_task_count = 0
         @completed_task_count = 0
         @largest_length       = 0
+        @workers_counter      = 0
         @ruby_pid             = $$
       end
     end
@@ -292,11 +295,15 @@ def ns_reset_if_forked
     class Worker
       include Concern::Logging
 
-      def initialize(pool)
+      def initialize(pool, id)
         # instance variables accessed only under pool's lock so no need to sync here again
         @queue  = Queue.new
         @pool   = pool
         @thread = create_worker @queue, pool, pool.idletime
+
+        if @thread.respond_to?(:name=)
+          @thread.name = [pool.name, 'worker', id].compact.join('-')
+        end
       end
 
       def <<(message)

lib/concurrent/executor/simple_executor_service.rb

@@ -91,7 +91,7 @@ def wait_for_termination(timeout = nil)
 
     private
 
-    def ns_initialize
+    def ns_initialize(*args)
       @running = Concurrent::AtomicBoolean.new(true)
       @stopped = Concurrent::Event.new
       @count = Concurrent::AtomicFixnum.new(0)

spec/concurrent/executor/ruby_thread_pool_executor_spec.rb

@@ -54,5 +54,39 @@ module Concurrent
         block.count_down
       end
     end
+
+    if Concurrent.on_cruby? && Concurrent.ruby_version(:>=, 2, 3, 0)
+      context 'threads naming' do
+        subject do
+          opts = { min_threads: 2 }
+          opts[:name] = pool_name if pool_name
+          described_class.new(opts)
+        end
+
+        let(:names) { Concurrent::Set.new }
+
+        before do
+          subject.post(names) { |names| names << Thread.current.name }
+          subject.post(names) { |names| names << Thread.current.name }
+          subject.shutdown
+          subject.wait_for_termination(pool_termination_timeout)
+          expect(names.size).to eq 2
+        end
+
+        context 'without pool name' do
+          let(:pool_name) { }
+          it 'sets counted name' do
+            expect(names.all? { |name| name =~ /^worker-\d+$/ }).to be true
+          end
+        end
+
+        context 'with pool name' do
+          let(:pool_name) { 'MyExecutor' }
+          it 'sets counted name' do
+            expect(names.all? { |name| name =~ /^MyExecutor-worker-\d+$/ }).to be true
+          end
+        end
+      end
+    end
   end
 end

spec/concurrent/executor/thread_pool_executor_shared.rb

@@ -31,6 +31,10 @@
     it 'defaults :fallback_policy to :abort' do
       expect(subject.fallback_policy).to eq :abort
     end
+
+    it 'defaults :name to nil' do
+      expect(subject.name).to be_nil
+    end
   end
 
   context "#initialize explicit values" do
@@ -100,6 +104,13 @@
         described_class.new(fallback_policy: :bogus)
       }.to raise_error(ArgumentError)
     end
+
+    it 'sets :name' do
+      pool = described_class.new(name: 'MyPool')
+      expect(pool.name).to eq 'MyPool'
+      pool.shutdown
+      pool.wait_for_termination(pool_termination_timeout)
+    end
   end
 
   context '#max_queue' do