#713

Committed 23 May 2014 08:08PM UTC coverage: 79.572% (-17.7%) from 97.237%

Build # #713

Build Type

push

Committed by

Commit Message

Merge pull request #96 from jdantonio/refactor/errors

Moved all custom errors into a single file and into the Concurrent module

Run Details

17 of 18 new or added lines in 10 files covered. (94.44%)

450 existing lines in 31 files now uncovered.

2010 of 2526 relevant lines covered (79.57%)

578.32 hits per line

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

54.05

/lib/concurrent/executor/executor.rb

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

module Concurrent

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

jdantonio / concurrent-ruby / #713

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