#716

Committed 23 May 2014 08:08PM UTC coverage: 80.966% (-16.3%) from 97.237%

Build # #716

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%)

469 existing lines in 32 files now uncovered.

2280 of 2816 relevant lines covered (80.97%)

592.71 hits per line

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

55.56

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

jdantonio / concurrent-ruby / #716

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