#2783

Committed 25 Jan 2015 12:39AM UTC coverage: 90.646% (-1.2%) from 91.857%

Build # #2783

Build Type

push

Committed by

Commit Message

Released 0.7.2

Run Details

2878 of 3175 relevant lines covered (90.65%)

356.65 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.08

/lib/concurrent/executor/ruby_thread_pool_executor.rb

require 'thread'

require_relative 'executor'
require 'concurrent/atomic/event'
require 'concurrent/executor/ruby_thread_pool_worker'

module Concurrent

  # @!macro thread_pool_executor
  class RubyThreadPoolExecutor
    include RubyExecutor

    # Default maximum number of threads that will be created in the pool.
    DEFAULT_MAX_POOL_SIZE      = 2**15 # 32768

    # Default minimum number of threads that will be retained in the pool.
    DEFAULT_MIN_POOL_SIZE      = 0

    # Default maximum number of tasks that may be added to the task queue.
    DEFAULT_MAX_QUEUE_SIZE     = 0

    # Default maximum number of seconds a thread in the pool may remain idle
    # before being reclaimed.
    DEFAULT_THREAD_IDLETIMEOUT = 60

    # The maximum number of threads that may be created in the pool.
    attr_reader :max_length

    # The minimum number of threads that may be retained in the pool.
    attr_reader :min_length

    # The largest number of threads that have been created in the pool since construction.
    attr_reader :largest_length

    # The number of tasks that have been scheduled for execution on the pool since construction.
    attr_reader :scheduled_task_count

    # The number of tasks that have been completed by the pool since construction.
    attr_reader :completed_task_count

    # The number of seconds that a thread may be idle before being reclaimed.
    attr_reader :idletime

    # The maximum number of tasks that may be waiting in the work queue at any one time.
    # When the queue size reaches `max_queue` subsequent tasks will be rejected in
    # accordance with the configured `fallback_policy`.
    attr_reader :max_queue

    # Create a new thread pool.
    #
    # @param [Hash] opts the options which configure the thread pool
    #
    # @option opts [Integer] :max_threads (DEFAULT_MAX_POOL_SIZE) the maximum
    #   number of threads to be created
    # @option opts [Integer] :min_threads (DEFAULT_MIN_POOL_SIZE) the minimum
    #   number of threads to be retained
    # @option opts [Integer] :idletime (DEFAULT_THREAD_IDLETIMEOUT) the maximum
    #   number of seconds a thread may be idle before being reclaimed
    # @option opts [Integer] :max_queue (DEFAULT_MAX_QUEUE_SIZE) the maximum
    #   number of tasks allowed in the work queue at any one time; a value of
    #   zero means the queue may grow without bound
    # @option opts [Symbol] :fallback_policy (:abort) the policy for handling new
    #   tasks that are received when the queue size has reached
    #   `max_queue` or the executor has shut down
    #
    # @raise [ArgumentError] if `:max_threads` is less than one
    # @raise [ArgumentError] if `:min_threads` is less than zero
    # @raise [ArgumentError] if `:fallback_policy` is not one of the values specified
    #   in `FALLBACK_POLICIES`
    #
    # @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ThreadPoolExecutor.html
    def initialize(opts = {})
      @min_length      = opts.fetch(:min_threads, DEFAULT_MIN_POOL_SIZE).to_i
      @max_length      = opts.fetch(:max_threads, DEFAULT_MAX_POOL_SIZE).to_i
      @idletime        = opts.fetch(:idletime, DEFAULT_THREAD_IDLETIMEOUT).to_i
      @max_queue       = opts.fetch(:max_queue, DEFAULT_MAX_QUEUE_SIZE).to_i
      @fallback_policy = opts.fetch(:fallback_policy, opts.fetch(:overflow_policy, :abort))
      warn '[DEPRECATED] :overflow_policy is deprecated terminology, please use :fallback_policy instead' if opts.has_key?(:overflow_policy)

      raise ArgumentError.new('max_threads must be greater than zero') if @max_length <= 0
      raise ArgumentError.new('min_threads cannot be less than zero') if @min_length < 0
      raise ArgumentError.new("#{fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICIES.include?(@fallback_policy)
      raise ArgumentError.new('min_threads cannot be more than max_threads') if min_length > max_length

      init_executor

      @pool                 = []
      @queue                = Queue.new
      @scheduled_task_count = 0
      @completed_task_count = 0
      @largest_length       = 0

      @gc_interval  = opts.fetch(:gc_interval, 1).to_i # undocumented
      @last_gc_time = Time.now.to_f - [1.0, (@gc_interval * 2.0)].max
    end

    # @!macro executor_module_method_can_overflow_question
    def can_overflow?
      @max_queue != 0
    end

    # The number of threads currently in the pool.
    #
    # @return [Integer] the length
    def length
      mutex.synchronize { running? ? @pool.length : 0 }
    end

    alias_method :current_length, :length

    # The number of tasks in the queue awaiting execution.
    #
    # @return [Integer] the queue_length
    def queue_length
      mutex.synchronize { running? ? @queue.length : 0 }
    end

    # Number of tasks that may be enqueued before reaching `max_queue` and rejecting
    # new tasks. A value of -1 indicates that the queue may grow without bound.
    #
    # @return [Integer] the remaining_capacity
    def remaining_capacity
      mutex.synchronize { @max_queue == 0 ? -1 : @max_queue - @queue.length }
    end

    # Returns an array with the status of each thread in the pool
    #
    # This method is deprecated and will be removed soon.
    def status
      warn '[DEPRECATED] `status` is deprecated and will be removed soon.'
      mutex.synchronize { @pool.collect { |worker| worker.status } }
    end

    # Run on task completion.
    #
    # @!visibility private
    def on_end_task
      mutex.synchronize do
        @completed_task_count += 1 #if success
        break unless running?
      end
    end

    # Run when a thread worker exits.
    #
    # @!visibility private
    def on_worker_exit(worker)
      mutex.synchronize do
        @pool.delete(worker)
        if @pool.empty? && !running?
          stop_event.set
          stopped_event.set
        end
      end
    end

    protected

    # @!visibility private
    def execute(*args, &task)
      prune_pool
      if ensure_capacity?
        @scheduled_task_count += 1
        @queue << [args, task]
      else
        handle_fallback(*args, &task) if @max_queue != 0 && @queue.length >= @max_queue
      end
    end

    # @!visibility private
    def shutdown_execution
      if @pool.empty?
        stopped_event.set
      else
        @pool.length.times { @queue << :stop }
      end
    end

    # @!visibility private
    def kill_execution
      @queue.clear
      drain_pool
    end

    # Check the thread pool configuration and determine if the pool
    # has enought capacity to handle the request. Will grow the size
    # of the pool if necessary.
    #
    # @return [Boolean] true if the pool has enough capacity else false
    #
    # @!visibility private
    def ensure_capacity?
      additional = 0
      capacity   = true

      if @pool.size < @min_length
        additional = @min_length - @pool.size
      elsif @queue.empty? && @queue.num_waiting >= 1
        additional = 0
      elsif @pool.size == 0 && @min_length == 0
        additional = 1
      elsif @pool.size < @max_length || @max_length == 0
        additional = 1
      elsif @max_queue == 0 || @queue.size < @max_queue
        additional = 0
      else
        capacity = false
      end

      additional.times do
        @pool << create_worker_thread
      end

      if additional > 0
        @largest_length = [@largest_length, @pool.length].max
      end

      capacity
    end

    # Scan all threads in the pool and reclaim any that are dead or
    # have been idle too long. Will check the last time the pool was
    # pruned and only run if the configured garbage collection
    # interval has passed.
    #
    # @!visibility private
    def prune_pool
      if Time.now.to_f - @gc_interval >= @last_gc_time
        @pool.delete_if { |worker| worker.dead? }
        # send :stop for each thread over idletime
        @pool.
            select { |worker| @idletime != 0 && Time.now.to_f - @idletime > worker.last_activity }.
            each { @queue << :stop }
        @last_gc_time = Time.now.to_f
      end
    end

    # Reclaim all threads in the pool.
    #
    # @!visibility private
    def drain_pool
      @pool.each { |worker| worker.kill }
      @pool.clear
    end

    # Create a single worker thread to be added to the pool.
    #
    # @return [Thread] the new thread.
    #
    # @!visibility private
    def create_worker_thread
      wrkr = RubyThreadPoolWorker.new(@queue, self)
      Thread.new(wrkr, self) do |worker, parent|
        Thread.current.abort_on_exception = false
        worker.run
        parent.on_worker_exit(worker)
      end
      return wrkr
    end
  end
end

1	require 'thread'	1✔
2
3	require_relative 'executor'	1✔
4	require 'concurrent/atomic/event'	1✔
5	require 'concurrent/executor/ruby_thread_pool_worker'	1✔
6
7	module Concurrent	1✔
8
9	# @!macro thread_pool_executor
10	class RubyThreadPoolExecutor	1✔
11	include RubyExecutor	1✔
12
13	# Default maximum number of threads that will be created in the pool.
14	DEFAULT_MAX_POOL_SIZE = 2**15 # 32768	1✔
15
16	# Default minimum number of threads that will be retained in the pool.
17	DEFAULT_MIN_POOL_SIZE = 0	1✔
18
19	# Default maximum number of tasks that may be added to the task queue.
20	DEFAULT_MAX_QUEUE_SIZE = 0	1✔
21
22	# Default maximum number of seconds a thread in the pool may remain idle
23	# before being reclaimed.
24	DEFAULT_THREAD_IDLETIMEOUT = 60	1✔
25
26	# The maximum number of threads that may be created in the pool.
27	attr_reader :max_length	1✔
28
29	# The minimum number of threads that may be retained in the pool.
30	attr_reader :min_length	1✔
31
32	# The largest number of threads that have been created in the pool since construction.
33	attr_reader :largest_length	1✔
34
35	# The number of tasks that have been scheduled for execution on the pool since construction.
36	attr_reader :scheduled_task_count	1✔
37
38	# The number of tasks that have been completed by the pool since construction.
39	attr_reader :completed_task_count	1✔
40
41	# The number of seconds that a thread may be idle before being reclaimed.
42	attr_reader :idletime	1✔
43
44	# The maximum number of tasks that may be waiting in the work queue at any one time.
45	# When the queue size reaches `max_queue` subsequent tasks will be rejected in
46	# accordance with the configured `fallback_policy`.
47	attr_reader :max_queue	1✔
48
49	# Create a new thread pool.
50	#
51	# @param [Hash] opts the options which configure the thread pool
52	#
53	# @option opts [Integer] :max_threads (DEFAULT_MAX_POOL_SIZE) the maximum
54	# number of threads to be created
55	# @option opts [Integer] :min_threads (DEFAULT_MIN_POOL_SIZE) the minimum
56	# number of threads to be retained
57	# @option opts [Integer] :idletime (DEFAULT_THREAD_IDLETIMEOUT) the maximum
58	# number of seconds a thread may be idle before being reclaimed
59	# @option opts [Integer] :max_queue (DEFAULT_MAX_QUEUE_SIZE) the maximum
60	# number of tasks allowed in the work queue at any one time; a value of
61	# zero means the queue may grow without bound
62	# @option opts [Symbol] :fallback_policy (:abort) the policy for handling new
63	# tasks that are received when the queue size has reached
64	# `max_queue` or the executor has shut down
65	#
66	# @raise [ArgumentError] if `:max_threads` is less than one
67	# @raise [ArgumentError] if `:min_threads` is less than zero
68	# @raise [ArgumentError] if `:fallback_policy` is not one of the values specified
69	# in `FALLBACK_POLICIES`
70	#
71	# @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ThreadPoolExecutor.html
72	def initialize(opts = {})	1✔
73	@min_length = opts.fetch(:min_threads, DEFAULT_MIN_POOL_SIZE).to_i	484✔
74	@max_length = opts.fetch(:max_threads, DEFAULT_MAX_POOL_SIZE).to_i	484✔
75	@idletime = opts.fetch(:idletime, DEFAULT_THREAD_IDLETIMEOUT).to_i	484✔
76	@max_queue = opts.fetch(:max_queue, DEFAULT_MAX_QUEUE_SIZE).to_i	484✔
77	@fallback_policy = opts.fetch(:fallback_policy, opts.fetch(:overflow_policy, :abort))	484✔
78	warn '[DEPRECATED] :overflow_policy is deprecated terminology, please use :fallback_policy instead' if opts.has_key?(:overflow_policy)	484✔
79
80	raise ArgumentError.new('max_threads must be greater than zero') if @max_length <= 0	484✔
81	raise ArgumentError.new('min_threads cannot be less than zero') if @min_length < 0	483✔
82	raise ArgumentError.new("#{fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICIES.include?(@fallback_policy)	482✔
83	raise ArgumentError.new('min_threads cannot be more than max_threads') if min_length > max_length	481✔
84
85	init_executor	480✔
86
87	@pool = []	480✔
88	@queue = Queue.new	480✔
89	@scheduled_task_count = 0	480✔
90	@completed_task_count = 0	480✔
91	@largest_length = 0	480✔
92
93	@gc_interval = opts.fetch(:gc_interval, 1).to_i # undocumented	480✔
94	@last_gc_time = Time.now.to_f - [1.0, (@gc_interval * 2.0)].max	480✔
95	end
96
97	# @!macro executor_module_method_can_overflow_question
98	def can_overflow?	1✔
99	@max_queue != 0	×
100	end
101
102	# The number of threads currently in the pool.
103	#
104	# @return [Integer] the length
105	def length	1✔
106	mutex.synchronize { running? ? @pool.length : 0 }	54✔
107	end
108
109	alias_method :current_length, :length	1✔
110
111	# The number of tasks in the queue awaiting execution.
112	#
113	# @return [Integer] the queue_length
114	def queue_length	1✔
115	mutex.synchronize { running? ? @queue.length : 0 }	10✔
116	end
117
118	# Number of tasks that may be enqueued before reaching `max_queue` and rejecting
119	# new tasks. A value of -1 indicates that the queue may grow without bound.
120	#
121	# @return [Integer] the remaining_capacity
122	def remaining_capacity	1✔
123	mutex.synchronize { @max_queue == 0 ? -1 : @max_queue - @queue.length }	10✔
124	end
125
126	# Returns an array with the status of each thread in the pool
127	#
128	# This method is deprecated and will be removed soon.
129	def status	1✔
130	warn '[DEPRECATED] `status` is deprecated and will be removed soon.'	2✔
131	mutex.synchronize { @pool.collect { \|worker\| worker.status } }	4✔
132	end
133
134	# Run on task completion.
135	#
136	# @!visibility private
137	def on_end_task	1✔
138	mutex.synchronize do	3,414✔
139	@completed_task_count += 1 #if success	3,414✔
140	break unless running?	3,414✔
141	end
142	end
143
144	# Run when a thread worker exits.
145	#
146	# @!visibility private
147	def on_worker_exit(worker)	1✔
148	mutex.synchronize do	594✔
149	@pool.delete(worker)	594✔
150	if @pool.empty? && !running?	594✔
151	stop_event.set	82✔
152	stopped_event.set	82✔
153	end
154	end
155	end
156
157	protected	1✔
158
159	# @!visibility private
160	def execute(*args, &task)	1✔
161	prune_pool	3,873✔
162	if ensure_capacity?	3,873✔
163	@scheduled_task_count += 1	3,746✔
164	@queue << [args, task]	3,746✔
165	else
166	handle_fallback(*args, &task) if @max_queue != 0 && @queue.length >= @max_queue	127✔
167	end
168	end
169
170	# @!visibility private
171	def shutdown_execution	1✔
172	if @pool.empty?	61✔
173	stopped_event.set	16✔
174	else
175	@pool.length.times { @queue << :stop }	445✔
176	end
177	end
178
179	# @!visibility private
180	def kill_execution	1✔
181	@queue.clear	103✔
182	drain_pool	103✔
183	end
184
185	# Check the thread pool configuration and determine if the pool
186	# has enought capacity to handle the request. Will grow the size
187	# of the pool if necessary.
188	#
189	# @return [Boolean] true if the pool has enough capacity else false
190	#
191	# @!visibility private
192	def ensure_capacity?	1✔
193	additional = 0	3,873✔
194	capacity = true	3,873✔
195
196	if @pool.size < @min_length	3,873✔
197	additional = @min_length - @pool.size	225✔
198	elsif @queue.empty? && @queue.num_waiting >= 1	3,648✔
199	additional = 0	2,095✔
200	elsif @pool.size == 0 && @min_length == 0	1,553✔
201	additional = 1	38✔
202	elsif @pool.size < @max_length \|\| @max_length == 0	1,515✔
203	additional = 1	895✔
204	elsif @max_queue == 0 \|\| @queue.size < @max_queue	620✔
205	additional = 0	493✔
206	else
207	capacity = false	127✔
208	end
209
210	additional.times do	3,873✔
211	@pool << create_worker_thread	12,281✔
212	end
213
214	if additional > 0	3,873✔
215	@largest_length = [@largest_length, @pool.length].max	1,158✔
216	end
217
218	capacity	3,873✔
219	end
220
221	# Scan all threads in the pool and reclaim any that are dead or
222	# have been idle too long. Will check the last time the pool was
223	# pruned and only run if the configured garbage collection
224	# interval has passed.
225	#
226	# @!visibility private
227	def prune_pool	1✔
228	if Time.now.to_f - @gc_interval >= @last_gc_time	3,875✔
229	@pool.delete_if { \|worker\| worker.dead? }	505✔
230	# send :stop for each thread over idletime
231	@pool.	274✔
232	select { \|worker\| @idletime != 0 && Time.now.to_f - @idletime > worker.last_activity }.	228✔
233	each { @queue << :stop }	4✔
234	@last_gc_time = Time.now.to_f	274✔
235	end
236	end
237
238	# Reclaim all threads in the pool.
239	#
240	# @!visibility private
241	def drain_pool	1✔
242	@pool.each { \|worker\| worker.kill }	416✔
243	@pool.clear	103✔
244	end
245
246	# Create a single worker thread to be added to the pool.
247	#
248	# @return [Thread] the new thread.
249	#
250	# @!visibility private
251	def create_worker_thread	1✔
252	wrkr = RubyThreadPoolWorker.new(@queue, self)	12,281✔
253	Thread.new(wrkr, self) do \|worker, parent\|	12,281✔
254	Thread.current.abort_on_exception = false	10,329✔
255	worker.run	10,329✔
256	parent.on_worker_exit(worker)	297✔
257	end
258	return wrkr	12,281✔
259	end
260	end
261	end

ruby-concurrency / concurrent-ruby / #2783

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous