#699

Committed 23 May 2014 10:34AM UTC coverage: 92.412% (-5.1%) from 97.494%

Build # #699

Build Type

push

Committed by

Commit Message

Another brittle supervisor test.

Run Details

2326 of 2517 relevant lines covered (92.41%)

664.38 hits per line

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

58.11

/lib/concurrent/executor/executor.rb

require 'concurrent/atomic/event'

module Concurrent

  # An exception class raised when the maximum queue size is reached and the
  # `overflow_policy` is set to `:abort`.
  RejectedExecutionError = Class.new(StandardError)

  module 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, &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

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

jdantonio / concurrent-ruby / #699

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