#2700

Committed 12 Feb 2015 06:34PM UTC coverage: 92.31% (+0.6%) from 91.69%

Build # #2700

Build Type

push

Committed by

Commit Message

Merge pull request #238 from ruby-concurrency/semaphore-pure-java

Implementation of Concurrent::JavaSemaphore in pure Java.

Run Details

2881 of 3121 relevant lines covered (92.31%)

393.08 hits per line

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

97.22

/lib/concurrent/executor/ruby_single_thread_executor.rb

require_relative 'executor'

module Concurrent

  # @!macro single_thread_executor
  class RubySingleThreadExecutor
    include RubyExecutor
    include SerialExecutor

    # Create a new thread pool.
    #
    # @option opts [Symbol] :fallback_policy (:discard) the policy for
    #   handling new tasks that are received when the queue size has
    #   reached `max_queue` or after the executor has shut down
    #
    # @see http://docs.oracle.com/javase/tutorial/essential/concurrency/pools.html
    # @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Executors.html
    # @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html
    def initialize(opts = {})
      @queue = Queue.new
      @thread = nil
      @fallback_policy = opts.fetch(:fallback_policy, :discard)
      raise ArgumentError.new("#{@fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICIES.include?(@fallback_policy)
      init_executor
    end

    protected

    # @!visibility private
    def execute(*args, &task)
      supervise
      @queue << [args, task]
    end

    # @!visibility private
    def shutdown_execution
      @queue << :stop
      stopped_event.set unless alive?
    end

    # @!visibility private
    def kill_execution
      @queue.clear
      @thread.kill if alive?
    end

    # @!visibility private
    def alive?
      @thread && @thread.alive?
    end

    # @!visibility private
    def supervise
      @thread = new_worker_thread unless alive?
    end

    # @!visibility private
    def new_worker_thread
      Thread.new do
        Thread.current.abort_on_exception = false
        work
      end
    end

    # @!visibility private
    def work
      loop do
        task = @queue.pop
        break if task == :stop
        begin
          task.last.call(*task.first)
        rescue => ex
          # let it fail
          log DEBUG, ex
        end
      end
      stopped_event.set
    end
  end
end

1	require_relative 'executor'	1✔
2
3	module Concurrent	1✔
4
5	# @!macro single_thread_executor
6	class RubySingleThreadExecutor	1✔
7	include RubyExecutor	1✔
8	include SerialExecutor	1✔
9
10	# Create a new thread pool.
11	#
12	# @option opts [Symbol] :fallback_policy (:discard) the policy for
13	# handling new tasks that are received when the queue size has
14	# reached `max_queue` or after the executor has shut down
15	#
16	# @see http://docs.oracle.com/javase/tutorial/essential/concurrency/pools.html
17	# @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Executors.html
18	# @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html
19	def initialize(opts = {})	1✔
20	@queue = Queue.new	126✔
21	@thread = nil	126✔
22	@fallback_policy = opts.fetch(:fallback_policy, :discard)	126✔
23	raise ArgumentError.new("#{@fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICIES.include?(@fallback_policy)	126✔
24	init_executor	126✔
25	end
26
27	protected	1✔
28
29	# @!visibility private
30	def execute(*args, &task)	1✔
31	supervise	240✔
32	@queue << [args, task]	240✔
33	end
34
35	# @!visibility private
36	def shutdown_execution	1✔
37	@queue << :stop	14✔
38	stopped_event.set unless alive?	14✔
39	end
40
41	# @!visibility private
42	def kill_execution	1✔
43	@queue.clear	38✔
44	@thread.kill if alive?	38✔
45	end
46
47	# @!visibility private
48	def alive?	1✔
49	@thread && @thread.alive?	292✔
50	end
51
52	# @!visibility private
53	def supervise	1✔
54	@thread = new_worker_thread unless alive?	240✔
55	end
56
57	# @!visibility private
58	def new_worker_thread	1✔
59	Thread.new do	99✔
60	Thread.current.abort_on_exception = false	82✔
61	work	82✔
62	end
63	end
64
65	# @!visibility private
66	def work	1✔
67	loop do	82✔
68	task = @queue.pop	229✔
69	break if task == :stop	189✔
70	begin
71	task.last.call(*task.first)	179✔
72	rescue => ex
73	# let it fail
74	log DEBUG, ex	×
75	end
76	end
77	stopped_event.set	10✔
78	end
79	end
80	end

ruby-concurrency / concurrent-ruby / #2700

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