#2731

Committed 10 Jun 2014 02:39PM UTC coverage: 89.322% (-4.3%) from 93.581%

Build # #2731

Build Type

push

Committed by

Commit Message

Merge branch 'billdueber-isolated_threadpool_fix'

Run Details

3 of 3 new or added lines in 1 file covered. (100.0%)

130 existing lines in 12 files now uncovered.

2518 of 2819 relevant lines covered (89.32%)

594.75 hits per line

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

59.26

/lib/concurrent/executor/executor.rb

require 'concurrent/errors'
require 'concurrent/logging'
require 'concurrent/atomic/event'

module Concurrent

  module Executor
    def can_overflow?
      false
    end
  end

  module RubyExecutor
    include Executor
    include Logging

    # Submit a task to the executor for asynchronous processing.
    #
    # @param [Array] args zero or more arguments to be passed to the task
    #
    # @yield the asynchronous task to perform
    #
    # @return [Boolean] `true` if the task is queued, `false` if the executor
    #   is not running
    #
    # @raise [ArgumentError] if no task is given
    def post(*args, &task)
      raise ArgumentError.new('no block given') unless block_given?
      mutex.synchronize do
        return false unless running?
        execute(*args, &task)
        true
      end
    end

    # Submit a task to the executor for asynchronous processing.
    #
    # @param [Proc] task the asynchronous task to perform
    #
    # @return [self] returns itself
    def <<(task)
      post(&task)
      self
    end

    # Is the executor running?
    #
    # @return [Boolean] `true` when running, `false` when shutting down or shutdown
    def running?
      ! stop_event.set?
    end

    # Is the executor shuttingdown?
    #
    # @return [Boolean] `true` when not running and not shutdown, else `false`
    def shuttingdown?
      ! (running? || shutdown?)
    end

    # Is the executor shutdown?
    #
    # @return [Boolean] `true` when shutdown, `false` when shutting down or running
    def shutdown?
      stopped_event.set?
    end

    # Begin an orderly shutdown. Tasks already in the queue will be executed,
    # but no new tasks will be accepted. Has no additional effect if the
    # thread pool is not running.
    def shutdown
      mutex.synchronize do
        break unless running?
        stop_event.set
        shutdown_execution
      end
      true
    end

    # Begin an immediate shutdown. In-progress tasks will be allowed to
    # complete but enqueued tasks will be dismissed and no new tasks
    # will be accepted. Has no additional effect if the thread pool is
    # not running.
    def kill
      mutex.synchronize do
        break if shutdown?
        stop_event.set
        kill_execution
        stopped_event.set
      end
      true
    end

    # Block until executor shutdown is complete or until `timeout` seconds have
    # passed.
    #
    # @note Does not initiate shutdown or termination. Either `shutdown` or `kill`
    #   must be called before this method (or on another thread).
    #
    # @param [Integer] timeout the maximum number of seconds to wait for shutdown to complete
    #
    # @return [Boolean] `true` if shutdown complete or false on `timeout`
    def wait_for_termination(timeout = nil)
      stopped_event.wait(timeout)
    end

    protected

    attr_reader :mutex, :stop_event, :stopped_event

    def init_executor
      @mutex = Mutex.new
      @stop_event = Event.new
      @stopped_event = Event.new
    end

    def execute(*args, &task)
      raise NotImplementedError
    end

    def shutdown_execution
      stopped_event.set
    end

    def kill_execution
      # do nothing
    end
  end

  if RUBY_PLATFORM == 'java'

    module JavaExecutor
      include Executor

      # Submit a task to the executor for asynchronous processing.
      #
      # @param [Array] args zero or more arguments to be passed to the task
      #
      # @yield the asynchronous task to perform
      #
      # @return [Boolean] `true` if the task is queued, `false` if the executor
      #   is not running
      #
      # @raise [ArgumentError] if no task is given
      def post(*args)
        raise ArgumentError.new('no block given') unless block_given?
        if running?
          @executor.submit{ yield(*args) }
          true
        else
          false
        end
      rescue Java::JavaUtilConcurrent::RejectedExecutionException => ex
        raise RejectedExecutionError
      end

      # Submit a task to the executor for asynchronous processing.
      #
      # @param [Proc] task the asynchronous task to perform
      #
      # @return [self] returns itself
      def <<(task)
        post(&task)
        self
      end

      # Is the executor running?
      #
      # @return [Boolean] `true` when running, `false` when shutting down or shutdown
      def running?
        ! (shuttingdown? || shutdown?)
      end

      # Is the executor shuttingdown?
      #
      # @return [Boolean] `true` when not running and not shutdown, else `false`
      def shuttingdown?
        if @executor.respond_to? :isTerminating
          @executor.isTerminating
        else
          false
        end
      end

      # Is the executor shutdown?
      #
      # @return [Boolean] `true` when shutdown, `false` when shutting down or running
      def shutdown?
        @executor.isShutdown || @executor.isTerminated
      end

      # Block until executor shutdown is complete or until `timeout` seconds have
      # passed.
      #
      # @note Does not initiate shutdown or termination. Either `shutdown` or `kill`
      #   must be called before this method (or on another thread).
      #
      # @param [Integer] timeout the maximum number of seconds to wait for shutdown to complete
      #
      # @return [Boolean] `true` if shutdown complete or false on `timeout`
      def wait_for_termination(timeout)
        @executor.awaitTermination(1000 * timeout, java.util.concurrent.TimeUnit::MILLISECONDS)
      end

      # Begin an orderly shutdown. Tasks already in the queue will be executed,
      # but no new tasks will be accepted. Has no additional effect if the
      # executor is not running.
      def shutdown
        @executor.shutdown
        nil
      end

      # Begin an immediate shutdown. In-progress tasks will be allowed to
      # complete but enqueued tasks will be dismissed and no new tasks
      # will be accepted. Has no additional effect if the executor is
      # not running.
      def kill
        @executor.shutdownNow
        nil
      end

      protected

      def set_shutdown_hook
        # without this the process may fail to exit
        at_exit { self.kill }
      end
    end
  end
end

1	require 'concurrent/errors'	1✔
2	require 'concurrent/logging'	1✔
3	require 'concurrent/atomic/event'	1✔
4
5	module Concurrent	1✔
6
7	module Executor	1✔
8	def can_overflow?	1✔
9	false	×
10	end
11	end
12
13	module RubyExecutor	1✔
14	include Executor	1✔
15	include Logging	1✔
16
17	# Submit a task to the executor for asynchronous processing.
18	#
19	# @param [Array] args zero or more arguments to be passed to the task
20	#
21	# @yield the asynchronous task to perform
22	#
23	# @return [Boolean] `true` if the task is queued, `false` if the executor
24	# is not running
25	#
26	# @raise [ArgumentError] if no task is given
27	def post(*args, &task)	1✔
28	raise ArgumentError.new('no block given') unless block_given?	5,121✔
29	mutex.synchronize do	5,117✔
30	return false unless running?	5,117✔
31	execute(*args, &task)	5,081✔
32	true	5,061✔
33	end
34	end
35
36	# Submit a task to the executor for asynchronous processing.
37	#
38	# @param [Proc] task the asynchronous task to perform
39	#
40	# @return [self] returns itself
41	def <<(task)	1✔
42	post(&task)	1,217✔
43	self	1,207✔
44	end
45
46	# Is the executor running?
47	#
48	# @return [Boolean] `true` when running, `false` when shutting down or shutdown
49	def running?	1✔
50	! stop_event.set?	7,676✔
51	end
52
53	# Is the executor shuttingdown?
54	#
55	# @return [Boolean] `true` when not running and not shutdown, else `false`
56	def shuttingdown?	1✔
57	! (running? \|\| shutdown?)	×
58	end
59
60	# Is the executor shutdown?
61	#
62	# @return [Boolean] `true` when shutdown, `false` when shutting down or running
63	def shutdown?	1✔
64	stopped_event.set?	626✔
65	end
66
67	# Begin an orderly shutdown. Tasks already in the queue will be executed,
68	# but no new tasks will be accepted. Has no additional effect if the
69	# thread pool is not running.
70	def shutdown	1✔
71	mutex.synchronize do	121✔
72	break unless running?	121✔
73	stop_event.set	112✔
74	shutdown_execution	112✔
75	end
76	true	121✔
77	end
78
79	# Begin an immediate shutdown. In-progress tasks will be allowed to
80	# complete but enqueued tasks will be dismissed and no new tasks
81	# will be accepted. Has no additional effect if the thread pool is
82	# not running.
83	def kill	1✔
84	mutex.synchronize do	623✔
85	break if shutdown?	622✔
86	stop_event.set	186✔
87	kill_execution	186✔
88	stopped_event.set	186✔
89	end
90	true	622✔
91	end
92
93	# Block until executor shutdown is complete or until `timeout` seconds have
94	# passed.
95	#
96	# @note Does not initiate shutdown or termination. Either `shutdown` or `kill`
97	# must be called before this method (or on another thread).
98	#
99	# @param [Integer] timeout the maximum number of seconds to wait for shutdown to complete
100	#
101	# @return [Boolean] `true` if shutdown complete or false on `timeout`
102	def wait_for_termination(timeout = nil)	1✔
103	stopped_event.wait(timeout)	60✔
104	end
105
106	protected	1✔
107
108	attr_reader :mutex, :stop_event, :stopped_event	1✔
109
110	def init_executor	1✔
111	@mutex = Mutex.new	678✔
112	@stop_event = Event.new	678✔
113	@stopped_event = Event.new	678✔
114	end
115
116	def execute(*args, &task)	1✔
117	raise NotImplementedError	×
118	end
119
120	def shutdown_execution	1✔
121	stopped_event.set	×
122	end
123
124	def kill_execution	1✔
125	# do nothing
126	end
127	end
128
129	if RUBY_PLATFORM == 'java'	1✔
130
UNCOV 131	module JavaExecutor	×
UNCOV 132	include Executor	×
133
134	# Submit a task to the executor for asynchronous processing.
135	#
136	# @param [Array] args zero or more arguments to be passed to the task
137	#
138	# @yield the asynchronous task to perform
139	#
140	# @return [Boolean] `true` if the task is queued, `false` if the executor
141	# is not running
142	#
143	# @raise [ArgumentError] if no task is given
UNCOV 144	def post(*args)	×
UNCOV 145	raise ArgumentError.new('no block given') unless block_given?	×
UNCOV 146	if running?	×
UNCOV 147	@executor.submit{ yield(*args) }	×
UNCOV 148	true	×
149	else
UNCOV 150	false	×
151	end
152	rescue Java::JavaUtilConcurrent::RejectedExecutionException => ex
153	raise RejectedExecutionError	×
154	end
155
156	# Submit a task to the executor for asynchronous processing.
157	#
158	# @param [Proc] task the asynchronous task to perform
159	#
160	# @return [self] returns itself
UNCOV 161	def <<(task)	×
UNCOV 162	post(&task)	×
UNCOV 163	self	×
164	end
165
166	# Is the executor running?
167	#
168	# @return [Boolean] `true` when running, `false` when shutting down or shutdown
UNCOV 169	def running?	×
UNCOV 170	! (shuttingdown? \|\| shutdown?)	×
171	end
172
173	# Is the executor shuttingdown?
174	#
175	# @return [Boolean] `true` when not running and not shutdown, else `false`
UNCOV 176	def shuttingdown?	×
UNCOV 177	if @executor.respond_to? :isTerminating	×
UNCOV 178	@executor.isTerminating	×
179	else
UNCOV 180	false	×
181	end
182	end
183
184	# Is the executor shutdown?
185	#
186	# @return [Boolean] `true` when shutdown, `false` when shutting down or running
UNCOV 187	def shutdown?	×
UNCOV 188	@executor.isShutdown \|\| @executor.isTerminated	×
189	end
190
191	# Block until executor shutdown is complete or until `timeout` seconds have
192	# passed.
193	#
194	# @note Does not initiate shutdown or termination. Either `shutdown` or `kill`
195	# must be called before this method (or on another thread).
196	#
197	# @param [Integer] timeout the maximum number of seconds to wait for shutdown to complete
198	#
199	# @return [Boolean] `true` if shutdown complete or false on `timeout`
UNCOV 200	def wait_for_termination(timeout)	×
UNCOV 201	@executor.awaitTermination(1000 * timeout, java.util.concurrent.TimeUnit::MILLISECONDS)	×
202	end
203
204	# Begin an orderly shutdown. Tasks already in the queue will be executed,
205	# but no new tasks will be accepted. Has no additional effect if the
206	# executor is not running.
UNCOV 207	def shutdown	×
UNCOV 208	@executor.shutdown	×
209	nil
210	end
211
212	# Begin an immediate shutdown. In-progress tasks will be allowed to
213	# complete but enqueued tasks will be dismissed and no new tasks
214	# will be accepted. Has no additional effect if the executor is
215	# not running.
UNCOV 216	def kill	×
UNCOV 217	@executor.shutdownNow	×
218	nil
219	end
220
UNCOV 221	protected	×
222
UNCOV 223	def set_shutdown_hook	×
224	# without this the process may fail to exit
UNCOV 225	at_exit { self.kill }	×
226	end
227	end
228	end
229	end

ruby-concurrency / concurrent-ruby / #2731

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