#2764

Committed 08 Dec 2014 03:40PM UTC coverage: 91.388% (-0.4%) from 91.753%

Build # #2764

Build Type

push

Committed by

Commit Message

Merge pull request #201 from rkday/fallback_handling

Posting to a shutdown thread pool - JRuby consistency and better naming

Run Details

18 of 26 new or added lines in 5 files covered. (69.23%)

212 existing lines in 36 files now uncovered.

2812 of 3077 relevant lines covered (91.39%)

369.8 hits per line

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

95.65

/lib/concurrent/executor/per_thread_executor.rb

require 'concurrent/atomics'
require 'concurrent/executor/executor'

module Concurrent

  # An executor service in which every operation spawns a new,
  # independently operating thread.
  #
  # This is perhaps the most inefficient executor service in this
  # library. It exists mainly for testing an debugging. Thread creation
  # and management is expensive in Ruby and this executor performs no
  # resource pooling. This can be very beneficial during testing and
  # debugging because it decouples the using code from the underlying
  # executor implementation. In production this executor will likely
  # lead to suboptimal performance.
  #
  # @note Intended for use primarily in testing and debugging.
  class PerThreadExecutor
    include Executor

    # Creates a new executor
    def initialize
      @running = Concurrent::AtomicBoolean.new(true)
      @stopped = Concurrent::Event.new
      @count = Concurrent::AtomicFixnum.new(0)
    end

    # @!macro executor_method_post
    def self.post(*args)
      raise ArgumentError.new('no block given') unless block_given?
      Thread.new(*args) do
        Thread.current.abort_on_exception = false
        yield(*args)
      end
      true
    end

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

    # @!macro executor_method_post
    def post(*args, &task)
      raise ArgumentError.new('no block given') unless block_given?
      return false unless running?
      @count.increment
      Thread.new(*args) do
        Thread.current.abort_on_exception = false
        begin
          yield(*args)
        ensure
          @count.decrement
          @stopped.set if @running.false? && @count.value == 0
        end
      end
    end

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

    # @!macro executor_method_running_question
    def running?
      @running.true?
    end

    # @!macro executor_method_shuttingdown_question
    def shuttingdown?
      @running.false? && ! @stopped.set?
    end

    # @!macro executor_method_shutdown_question
    def shutdown?
      @stopped.set?
    end

    # @!macro executor_method_shutdown
    def shutdown
      @running.make_false
      @stopped.set if @count.value == 0
      true
    end

    # @!macro executor_method_kill
    def kill
      @running.make_false
      @stopped.set
      true
    end

    # @!macro executor_method_wait_for_termination
    def wait_for_termination(timeout = nil)
      @stopped.wait(timeout)
    end
  end
end

1	require 'concurrent/atomics'	1✔
2	require 'concurrent/executor/executor'	1✔
3
4	module Concurrent	1✔
5
6	# An executor service in which every operation spawns a new,
7	# independently operating thread.
8	#
9	# This is perhaps the most inefficient executor service in this
10	# library. It exists mainly for testing an debugging. Thread creation
11	# and management is expensive in Ruby and this executor performs no
12	# resource pooling. This can be very beneficial during testing and
13	# debugging because it decouples the using code from the underlying
14	# executor implementation. In production this executor will likely
15	# lead to suboptimal performance.
16	#
17	# @note Intended for use primarily in testing and debugging.
18	class PerThreadExecutor	1✔
19	include Executor	1✔
20
21	# Creates a new executor
22	def initialize	1✔
23	@running = Concurrent::AtomicBoolean.new(true)	211✔
24	@stopped = Concurrent::Event.new	211✔
25	@count = Concurrent::AtomicFixnum.new(0)	211✔
26	end
27
28	# @!macro executor_method_post
29	def self.post(*args)	1✔
30	raise ArgumentError.new('no block given') unless block_given?	6✔
31	Thread.new(*args) do	6✔
32	Thread.current.abort_on_exception = false	3✔
33	yield(*args)	3✔
34	end
35	true	6✔
36	end
37
38	# @!macro executor_method_left_shift
39	def self.<<(task)	1✔
40	post(&task)	1✔
41	self	1✔
42	end
43
44	# @!macro executor_method_post
45	def post(*args, &task)	1✔
46	raise ArgumentError.new('no block given') unless block_given?	392✔
47	return false unless running?	391✔
48	@count.increment	385✔
49	Thread.new(*args) do	385✔
50	Thread.current.abort_on_exception = false	347✔
51	begin
52	yield(*args)	347✔
53	ensure
54	@count.decrement	347✔
55	@stopped.set if @running.false? && @count.value == 0	347✔
56	end
57	end
58	end
59
60	# @!macro executor_method_left_shift
61	def <<(task)	1✔
62	post(&task)	12✔
63	self	12✔
64	end
65
66	# @!macro executor_method_running_question
67	def running?	1✔
68	@running.true?	395✔
69	end
70
71	# @!macro executor_method_shuttingdown_question
72	def shuttingdown?	1✔
UNCOV 73	@running.false? && ! @stopped.set?	×
74	end
75
76	# @!macro executor_method_shutdown_question
77	def shutdown?	1✔
UNCOV 78	@stopped.set?	×
79	end
80
81	# @!macro executor_method_shutdown
82	def shutdown	1✔
83	@running.make_false	15✔
84	@stopped.set if @count.value == 0	15✔
85	true	15✔
86	end
87
88	# @!macro executor_method_kill
89	def kill	1✔
90	@running.make_false	26✔
91	@stopped.set	26✔
92	true	26✔
93	end
94
95	# @!macro executor_method_wait_for_termination
96	def wait_for_termination(timeout = nil)	1✔
97	@stopped.wait(timeout)	11✔
98	end
99	end
100	end

ruby-concurrency / concurrent-ruby / #2764

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