#2744

Committed 05 Nov 2014 06:06PM UTC coverage: 90.672% (-4.1%) from 94.79%

Build # #2744

Build Type

push

Committed by

Commit Message

Update README.md

Run Details

2780 of 3066 relevant lines covered (90.67%)

364.4 hits per line

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

56.52

/lib/concurrent/executor/executor.rb

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

module Concurrent

  module Executor
    
    # @!macro [attach] executor_module_method_can_overflow_question
    #
    #   Does the task queue have a maximum size?
    #
    #   @return [Boolean] True if the task queue has a maximum size else false.
    #
    # @note Always returns `false`
    def can_overflow?
      false
    end

    # @!macro [attach] executor_module_method_serialized_question
    #
    #   Does this executor guarantee serialization of its operations?
    #
    #   @return [Boolean] True if the executor guarantees that all operations
    #     will be post in the order they are received and no two operations may
    #     occur simultaneously. Else false.
    #
    # @note Always returns `false`
    def serialized?
      false
    end
  end

  # Indicates that the including `Executor` or `ExecutorService` guarantees
  # that all operations will occur in the order they are post and that no
  # two operations may occur simultaneously. This module provides no
  # functionality and provides no guarantees. That is the responsibility
  # of the including class. This module exists solely to allow the including
  # object to be interrogated for its serialization status.
  #
  # @example
  #   class Foo
  #     include Concurrent::SerialExecutor
  #   end
  #
  #   foo = Foo.new
  #
  #   foo.is_a? Concurrent::Executor       #=> true
  #   foo.is_a? Concurrent::SerialExecutor #=> true
  #   foo.serialized?                      #=> true
  module SerialExecutor
    include Executor

    # @!macro executor_module_method_serialized_question
    #
    # @note Always returns `true`
    def serialized?
      true
    end
  end

  module RubyExecutor
    include Executor
    include Logging

    # @!macro [attach] executor_method_post
    #
    #   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

    # @!macro [attach] executor_method_left_shift
    #
    #   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

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

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

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

    # @!macro [attach] executor_method_shutdown
    #
    #   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

    # @!macro [attach] executor_method_kill
    #
    #   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

    # @!macro [attach] executor_method_wait_for_termination
    #
    #   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

    # @!macro [attach] executor_method_init_executor
    #
    #   Initialize the executor by creating and initializing all the
    #   internal synchronization objects.
    def init_executor
      @mutex = Mutex.new
      @stop_event = Event.new
      @stopped_event = Event.new
    end

    # @!macro [attach] executor_method_execute
    def execute(*args, &task)
      raise NotImplementedError
    end

    # @!macro [attach] executor_method_shutdown_execution
    # 
    #   Callback method called when an orderly shutdown has completed.
    #   The default behavior is to signal all waiting threads.
    def shutdown_execution
      stopped_event.set
    end

    # @!macro [attach] executor_method_kill_execution
    #
    #   Callback method called when the executor has been killed.
    #   The default behavior is to do nothing.
    def kill_execution
      # do nothing
    end
  end

  if RUBY_PLATFORM == 'java'

    module JavaExecutor
      include Executor
      java_import 'java.lang.Runnable'

      # @!macro executor_method_post
      def post(*args)
        raise ArgumentError.new('no block given') unless block_given?
        if running?
          executor_submit = @executor.java_method(:submit, [Runnable.java_class])
          executor_submit.call { yield(*args) }
          true
        else
          false
        end
      rescue Java::JavaUtilConcurrent::RejectedExecutionException
        raise RejectedExecutionError
      end

      # @!macro executor_method_left_shift
      def <<(task)
        post(&task)
        self
      end

      # @!macro executor_method_running_question
      def running?
        ! (shuttingdown? || shutdown?)
      end

      # @!macro executor_method_shuttingdown_question
      def shuttingdown?
        if @executor.respond_to? :isTerminating
          @executor.isTerminating
        else
          false
        end
      end

      # @!macro executor_method_shutdown_question
      def shutdown?
        @executor.isShutdown || @executor.isTerminated
      end

      # @!macro executor_method_wait_for_termination
      def wait_for_termination(timeout = nil)
        if timeout.nil?
          ok = @executor.awaitTermination(60, java.util.concurrent.TimeUnit::SECONDS) until ok
          true
        else
          @executor.awaitTermination(1000 * timeout, java.util.concurrent.TimeUnit::MILLISECONDS)
        end
      end

      # @!macro executor_method_shutdown
      def shutdown
        @executor.shutdown
        nil
      end

      # @!macro executor_method_kill
      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
9	# @!macro [attach] executor_module_method_can_overflow_question
10	#
11	# Does the task queue have a maximum size?
12	#
13	# @return [Boolean] True if the task queue has a maximum size else false.
14	#
15	# @note Always returns `false`
16	def can_overflow?	1✔
17	false	×
18	end
19
20	# @!macro [attach] executor_module_method_serialized_question
21	#
22	# Does this executor guarantee serialization of its operations?
23	#
24	# @return [Boolean] True if the executor guarantees that all operations
25	# will be post in the order they are received and no two operations may
26	# occur simultaneously. Else false.
27	#
28	# @note Always returns `false`
29	def serialized?	1✔
30	false	×
31	end
32	end
33
34	# Indicates that the including `Executor` or `ExecutorService` guarantees
35	# that all operations will occur in the order they are post and that no
36	# two operations may occur simultaneously. This module provides no
37	# functionality and provides no guarantees. That is the responsibility
38	# of the including class. This module exists solely to allow the including
39	# object to be interrogated for its serialization status.
40	#
41	# @example
42	# class Foo
43	# include Concurrent::SerialExecutor
44	# end
45	#
46	# foo = Foo.new
47	#
48	# foo.is_a? Concurrent::Executor #=> true
49	# foo.is_a? Concurrent::SerialExecutor #=> true
50	# foo.serialized? #=> true
51	module SerialExecutor	1✔
52	include Executor	1✔
53
54	# @!macro executor_module_method_serialized_question
55	#
56	# @note Always returns `true`
57	def serialized?	1✔
58	true	×
59	end
60	end
61
62	module RubyExecutor	1✔
63	include Executor	1✔
64	include Logging	1✔
65
66	# @!macro [attach] executor_method_post
67	#
68	# Submit a task to the executor for asynchronous processing.
69	#
70	# @param [Array] args zero or more arguments to be passed to the task
71	#
72	# @yield the asynchronous task to perform
73	#
74	# @return [Boolean] `true` if the task is queued, `false` if the executor
75	# is not running
76	#
77	# @raise [ArgumentError] if no task is given
78	def post(*args, &task)	1✔
79	raise ArgumentError.new('no block given') unless block_given?	6,145✔
80	mutex.synchronize do	6,142✔
81	return false unless running?	6,142✔
82	execute(*args, &task)	6,115✔
83	true	6,094✔
84	end
85	end
86
87	# @!macro [attach] executor_method_left_shift
88	#
89	# Submit a task to the executor for asynchronous processing.
90	#
91	# @param [Proc] task the asynchronous task to perform
92	#
93	# @return [self] returns itself
94	def <<(task)	1✔
95	post(&task)	1,196✔
96	self	1,186✔
97	end
98
99	# @!macro [attach] executor_method_running_question
100	#
101	# Is the executor running?
102	#
103	# @return [Boolean] `true` when running, `false` when shutting down or shutdown
104	def running?	1✔
105	! stop_event.set?	9,871✔
106	end
107
108	# @!macro [attach] executor_method_shuttingdown_question
109	#
110	# Is the executor shuttingdown?
111	#
112	# @return [Boolean] `true` when not running and not shutdown, else `false`
113	def shuttingdown?	1✔
114	! (running? \|\| shutdown?)	×
115	end
116
117	# @!macro [attach] executor_method_shutdown_question
118	#
119	# Is the executor shutdown?
120	#
121	# @return [Boolean] `true` when shutdown, `false` when shutting down or running
122	def shutdown?	1✔
123	stopped_event.set?	550✔
124	end
125
126	# @!macro [attach] executor_method_shutdown
127	#
128	# Begin an orderly shutdown. Tasks already in the queue will be executed,
129	# but no new tasks will be accepted. Has no additional effect if the
130	# thread pool is not running.
131	def shutdown	1✔
132	mutex.synchronize do	104✔
133	break unless running?	104✔
134	stop_event.set	95✔
135	shutdown_execution	95✔
136	end
137	true	104✔
138	end
139
140	# @!macro [attach] executor_method_kill
141	#
142	# Begin an immediate shutdown. In-progress tasks will be allowed to
143	# complete but enqueued tasks will be dismissed and no new tasks
144	# will be accepted. Has no additional effect if the thread pool is
145	# not running.
146	def kill	1✔
147	mutex.synchronize do	546✔
148	break if shutdown?	546✔
149	stop_event.set	194✔
150	kill_execution	194✔
151	stopped_event.set	194✔
152	end
153	true	546✔
154	end
155
156	# @!macro [attach] executor_method_wait_for_termination
157	#
158	# Block until executor shutdown is complete or until `timeout` seconds have
159	# passed.
160	#
161	# @note Does not initiate shutdown or termination. Either `shutdown` or `kill`
162	# must be called before this method (or on another thread).
163	#
164	# @param [Integer] timeout the maximum number of seconds to wait for shutdown to complete
165	#
166	# @return [Boolean] `true` if shutdown complete or false on `timeout`
167	def wait_for_termination(timeout = nil)	1✔
168	stopped_event.wait(timeout)	49✔
169	end
170
171	protected	1✔
172
173	attr_reader :mutex, :stop_event, :stopped_event	1✔
174
175	# @!macro [attach] executor_method_init_executor
176	#
177	# Initialize the executor by creating and initializing all the
178	# internal synchronization objects.
179	def init_executor	1✔
180	@mutex = Mutex.new	747✔
181	@stop_event = Event.new	747✔
182	@stopped_event = Event.new	747✔
183	end
184
185	# @!macro [attach] executor_method_execute
186	def execute(*args, &task)	1✔
187	raise NotImplementedError	×
188	end
189
190	# @!macro [attach] executor_method_shutdown_execution
191	#
192	# Callback method called when an orderly shutdown has completed.
193	# The default behavior is to signal all waiting threads.
194	def shutdown_execution	1✔
195	stopped_event.set	×
196	end
197
198	# @!macro [attach] executor_method_kill_execution
199	#
200	# Callback method called when the executor has been killed.
201	# The default behavior is to do nothing.
202	def kill_execution	1✔
203	# do nothing
204	end
205	end
206
207	if RUBY_PLATFORM == 'java'	1✔
208
209	module JavaExecutor	×
210	include Executor	×
211	java_import 'java.lang.Runnable'	×
212
213	# @!macro executor_method_post
214	def post(*args)	×
215	raise ArgumentError.new('no block given') unless block_given?	×
216	if running?	×
217	executor_submit = @executor.java_method(:submit, [Runnable.java_class])	×
218	executor_submit.call { yield(*args) }	×
219	true	×
220	else
221	false	×
222	end
223	rescue Java::JavaUtilConcurrent::RejectedExecutionException
224	raise RejectedExecutionError	×
225	end
226
227	# @!macro executor_method_left_shift
228	def <<(task)	×
229	post(&task)	×
230	self	×
231	end
232
233	# @!macro executor_method_running_question
234	def running?	×
235	! (shuttingdown? \|\| shutdown?)	×
236	end
237
238	# @!macro executor_method_shuttingdown_question
239	def shuttingdown?	×
240	if @executor.respond_to? :isTerminating	×
241	@executor.isTerminating	×
242	else
243	false	×
244	end
245	end
246
247	# @!macro executor_method_shutdown_question
248	def shutdown?	×
249	@executor.isShutdown \|\| @executor.isTerminated	×
250	end
251
252	# @!macro executor_method_wait_for_termination
253	def wait_for_termination(timeout = nil)	×
254	if timeout.nil?	×
255	ok = @executor.awaitTermination(60, java.util.concurrent.TimeUnit::SECONDS) until ok	×
256	true	×
257	else
258	@executor.awaitTermination(1000 * timeout, java.util.concurrent.TimeUnit::MILLISECONDS)	×
259	end
260	end
261
262	# @!macro executor_method_shutdown
263	def shutdown	×
264	@executor.shutdown	×
265	nil
266	end
267
268	# @!macro executor_method_kill
269	def kill	×
270	@executor.shutdownNow	×
271	nil
272	end
273
274	protected	×
275
276	def set_shutdown_hook	×
277	# without this the process may fail to exit
278	at_exit { self.kill }	×
279	end
280	end
281	end
282	end

ruby-concurrency / concurrent-ruby / #2744

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