#2730

Committed 24 Jun 2014 03:45PM UTC coverage: 91.456% (-4.9%) from 96.381%

Build # #2730

Build Type

push

Committed by

Commit Message

Merge pull request #118 from ShaneWilton/master

Typo in documenation, brake -> break

Run Details

2280 of 2493 relevant lines covered (91.46%)

473.72 hits per line

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

98.28

/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 set of possible overflow policies that may be set at thread pool creation.
    OVERFLOW_POLICIES = [:abort, :discard, :caller_runs]

    # 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 `overflow_policy`.
    attr_reader :max_queue

    # The policy defining how rejected tasks (tasks received once the queue size reaches
    # the configured `max_queue`) are handled. Must be one of the values specified in
    # `OVERFLOW_POLICIES`.
    attr_reader :overflow_policy

    # 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 bounnd
    # @option opts [Symbol] :overflow_policy (:abort) the policy for handling new
    #   tasks that are received when the queue size has reached `max_queue`
    #
    # @raise [ArgumentError] if `:max_threads` is less than one
    # @raise [ArgumentError] if `:min_threads` is less than zero
    # @raise [ArgumentError] if `:overflow_policy` is not one of the values specified
    #   in `OVERFLOW_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
      @overflow_policy = opts.fetch(:overflow_policy, :abort)

      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("#{overflow_policy} is not a valid overflow policy") unless OVERFLOW_POLICIES.include?(@overflow_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

    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_overflow(*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

    # Handler which executes the `overflow_policy` once the queue size
    # reaches `max_queue`.
    #
    # @param [Array] args the arguments to the task which is being handled.
    #
    # @!visibility private
    def handle_overflow(*args)
      case @overflow_policy
      when :abort
        raise RejectedExecutionError
      when :discard
        false
      when :caller_runs
        begin
          yield(*args)
        rescue => ex
          # let it fail
          log DEBUG, ex
        end
        true
      end
    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 do |worker|
          worker.dead? ||
            (@idletime == 0 ? false : Time.now.to_f - @idletime > worker.last_activity)
        end
        @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 set of possible overflow policies that may be set at thread pool creation.
27	OVERFLOW_POLICIES = [:abort, :discard, :caller_runs]	1✔
28
29	# The maximum number of threads that may be created in the pool.
30	attr_reader :max_length	1✔
31
32	# The minimum number of threads that may be retained in the pool.
33	attr_reader :min_length	1✔
34
35	# The largest number of threads that have been created in the pool since construction.
36	attr_reader :largest_length	1✔
37
38	# The number of tasks that have been scheduled for execution on the pool since construction.
39	attr_reader :scheduled_task_count	1✔
40
41	# The number of tasks that have been completed by the pool since construction.
42	attr_reader :completed_task_count	1✔
43
44	# The number of seconds that a thread may be idle before being reclaimed.
45	attr_reader :idletime	1✔
46
47	# The maximum number of tasks that may be waiting in the work queue at any one time.
48	# When the queue size reaches `max_queue` subsequent tasks will be rejected in
49	# accordance with the configured `overflow_policy`.
50	attr_reader :max_queue	1✔
51
52	# The policy defining how rejected tasks (tasks received once the queue size reaches
53	# the configured `max_queue`) are handled. Must be one of the values specified in
54	# `OVERFLOW_POLICIES`.
55	attr_reader :overflow_policy	1✔
56
57	# Create a new thread pool.
58	#
59	# @param [Hash] opts the options which configure the thread pool
60	#
61	# @option opts [Integer] :max_threads (DEFAULT_MAX_POOL_SIZE) the maximum
62	# number of threads to be created
63	# @option opts [Integer] :min_threads (DEFAULT_MIN_POOL_SIZE) the minimum
64	# number of threads to be retained
65	# @option opts [Integer] :idletime (DEFAULT_THREAD_IDLETIMEOUT) the maximum
66	# number of seconds a thread may be idle before being reclaimed
67	# @option opts [Integer] :max_queue (DEFAULT_MAX_QUEUE_SIZE) the maximum
68	# number of tasks allowed in the work queue at any one time; a value of
69	# zero means the queue may grow without bounnd
70	# @option opts [Symbol] :overflow_policy (:abort) the policy for handling new
71	# tasks that are received when the queue size has reached `max_queue`
72	#
73	# @raise [ArgumentError] if `:max_threads` is less than one
74	# @raise [ArgumentError] if `:min_threads` is less than zero
75	# @raise [ArgumentError] if `:overflow_policy` is not one of the values specified
76	# in `OVERFLOW_POLICIES`
77	#
78	# @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ThreadPoolExecutor.html
79	def initialize(opts = {})	1✔
80	@min_length = opts.fetch(:min_threads, DEFAULT_MIN_POOL_SIZE).to_i	345✔
81	@max_length = opts.fetch(:max_threads, DEFAULT_MAX_POOL_SIZE).to_i	345✔
82	@idletime = opts.fetch(:idletime, DEFAULT_THREAD_IDLETIMEOUT).to_i	345✔
83	@max_queue = opts.fetch(:max_queue, DEFAULT_MAX_QUEUE_SIZE).to_i	345✔
84	@overflow_policy = opts.fetch(:overflow_policy, :abort)	345✔
85
86	raise ArgumentError.new('max_threads must be greater than zero') if @max_length <= 0	345✔
87	raise ArgumentError.new('min_threads cannot be less than zero') if @min_length < 0	344✔
88	raise ArgumentError.new("#{overflow_policy} is not a valid overflow policy") unless OVERFLOW_POLICIES.include?(@overflow_policy)	343✔
89	raise ArgumentError.new('min_threads cannot be more than max_threads') if min_length > max_length	342✔
90
91	init_executor	341✔
92
93	@pool = []	341✔
94	@queue = Queue.new	341✔
95	@scheduled_task_count = 0	341✔
96	@completed_task_count = 0	341✔
97	@largest_length = 0	341✔
98
99	@gc_interval = opts.fetch(:gc_interval, 1).to_i # undocumented	341✔
100	@last_gc_time = Time.now.to_f - [1.0, (@gc_interval * 2.0)].max	341✔
101	end
102
103	def can_overflow?	1✔
104	@max_queue != 0	×
105	end
106
107	# The number of threads currently in the pool.
108	#
109	# @return [Integer] the length
110	def length	1✔
111	mutex.synchronize{ running? ? @pool.length : 0 }	54✔
112	end
113	alias_method :current_length, :length	1✔
114
115	# The number of tasks in the queue awaiting execution.
116	#
117	# @return [Integer] the queue_length
118	def queue_length	1✔
119	mutex.synchronize{ running? ? @queue.length : 0 }	10✔
120	end
121
122	# Number of tasks that may be enqueued before reaching `max_queue` and rejecting
123	# new tasks. A value of -1 indicates that the queue may grow without bound.
124	#
125	# @return [Integer] the remaining_capacity
126	def remaining_capacity	1✔
127	mutex.synchronize { @max_queue == 0 ? -1 : @max_queue - @queue.length }	10✔
128	end
129
130	# Returns an array with the status of each thread in the pool
131	#
132	# This method is deprecated and will be removed soon.
133	def status	1✔
134	warn '[DEPRECATED] `status` is deprecated and will be removed soon.'	2✔
135	mutex.synchronize { @pool.collect { \|worker\| worker.status } }	4✔
136	end
137
138	# Run on task completion.
139	#
140	# @!visibility private
141	def on_end_task	1✔
142	mutex.synchronize do	4,328✔
143	@completed_task_count += 1 #if success	4,327✔
144	break unless running?	4,327✔
145	end
146	end
147
148	# Run when a thread worker exits.
149	#
150	# @!visibility private
151	def on_worker_exit(worker)	1✔
152	mutex.synchronize do	578✔
153	@pool.delete(worker)	578✔
154	if @pool.empty? && ! running?	578✔
155	stop_event.set	74✔
156	stopped_event.set	74✔
157	end
158	end
159	end
160
161	protected	1✔
162
163	# @!visibility private
164	def execute(*args, &task)	1✔
165	prune_pool	6,944✔
166	if ensure_capacity?	6,944✔
167	@scheduled_task_count += 1	4,639✔
168	@queue << [args, task]	4,639✔
169	else
170	handle_overflow(*args, &task) if @max_queue != 0 && @queue.length >= @max_queue	2,305✔
171	end
172	end
173
174	# @!visibility private
175	def shutdown_execution	1✔
176	if @pool.empty?	51✔
177	stopped_event.set	8✔
178	else
179	@pool.length.times{ @queue << :stop }	445✔
180	end
181	end
182
183	# @!visibility private
184	def kill_execution	1✔
185	@queue.clear	109✔
186	drain_pool	109✔
187	end
188
189	# Check the thread pool configuration and determine if the pool
190	# has enought capacity to handle the request. Will grow the size
191	# of the pool if necessary.
192	#
193	# @return [Boolean] true if the pool has enough capacity else false
194	#
195	# @!visibility private
196	def ensure_capacity?	1✔
197	additional = 0	6,944✔
198	capacity = true	6,944✔
199
200	if @pool.size < @min_length	6,944✔
201	additional = @min_length - @pool.size	138✔
202	elsif @queue.empty? && @queue.num_waiting >= 1	6,806✔
203	additional = 0	2,213✔
204	elsif @pool.size == 0 && @min_length == 0	4,593✔
205	additional = 1	39✔
206	elsif @pool.size < @max_length \|\| @max_length == 0	4,554✔
207	additional = 1	1,797✔
208	elsif @max_queue == 0 \|\| @queue.size < @max_queue	2,757✔
209	additional = 0	452✔
210	else
211	capacity = false	2,305✔
212	end
213
214	additional.times do	6,944✔
215	@pool << create_worker_thread	7,494✔
216	end
217
218	if additional > 0	6,944✔
219	@largest_length = [@largest_length, @pool.length].max	1,974✔
220	end
221
222	capacity	6,944✔
223	end
224
225	# Handler which executes the `overflow_policy` once the queue size
226	# reaches `max_queue`.
227	#
228	# @param [Array] args the arguments to the task which is being handled.
229	#
230	# @!visibility private
231	def handle_overflow(*args)	1✔
232	case @overflow_policy	2,301✔
233	when :abort
234	raise RejectedExecutionError	21✔
235	when :discard
236	false	2,271✔
237	when :caller_runs
238	begin
239	yield(*args)	9✔
240	rescue => ex
241	# let it fail
242	log DEBUG, ex	×
243	end
244	true	9✔
245	end
246	end
247
248	# Scan all threads in the pool and reclaim any that are dead or
249	# have been idle too long. Will check the last time the pool was
250	# pruned and only run if the configured garbage collection
251	# interval has passed.
252	#
253	# @!visibility private
254	def prune_pool	1✔
255	if Time.now.to_f - @gc_interval >= @last_gc_time	6,944✔
256	@pool.delete_if do \|worker\|	184✔
257	worker.dead? \|\|	145✔
258	(@idletime == 0 ? false : Time.now.to_f - @idletime > worker.last_activity)	78✔
259	end
260	@last_gc_time = Time.now.to_f	184✔
261	end
262	end
263
264	# Reclaim all threads in the pool.
265	#
266	# @!visibility private
267	def drain_pool	1✔
268	@pool.each {\|worker\| worker.kill }	433✔
269	@pool.clear	109✔
270	end
271
272	# Create a single worker thread to be added to the pool.
273	#
274	# @return [Thread] the new thread.
275	#
276	# @!visibility private
277	def create_worker_thread	1✔
278	wrkr = RubyThreadPoolWorker.new(@queue, self)	7,494✔
279	Thread.new(wrkr, self) do \|worker, parent\|	7,494✔
280	Thread.current.abort_on_exception = false	6,591✔
281	worker.run	6,591✔
282	parent.on_worker_exit(worker)	289✔
283	end
284	return wrkr	7,494✔
285	end
286	end
287	end

ruby-concurrency / concurrent-ruby / #2730

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