#812

Committed 23 May 2014 08:08PM UTC coverage: 39.344% (-57.9%) from 97.237%

Build # #812

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

Coverage Stats

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

1435 existing lines in 57 files now uncovered.

1187 of 3017 relevant lines covered (39.34%)

0.5 hits per line

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

29.68

/lib/concurrent/supervisor.rb

require 'thread'

require 'concurrent/errors'
require 'concurrent/runnable'

module Concurrent

  class Supervisor

    DEFAULT_MONITOR_INTERVAL = 1
    RESTART_STRATEGIES = [:one_for_one, :one_for_all, :rest_for_one]
    DEFAULT_MAX_RESTART = 5
    DEFAULT_MAX_TIME = 60
    WORKER_API = {run: 0, stop: 0, running?: 0}

    CHILD_TYPES = [:worker, :supervisor]
    CHILD_RESTART_OPTIONS = [:permanent, :transient, :temporary]

    WorkerContext = Struct.new(:worker, :type, :restart) do
      attr_accessor :thread
      attr_accessor :terminated

      def alive?() return thread && thread.alive?; end

      def needs_restart?
        return false if thread && thread.alive?
        return false if terminated
        case self.restart
        when :permanent
          return true
        when :transient
          return thread.nil? || thread.status.nil?
        else #when :temporary
          return false
        end
      end
    end

    WorkerCounts = Struct.new(:specs, :supervisors, :workers) do
      attr_accessor :status
      def add(context)
        self.specs += 1
        self.supervisors += 1 if context.type == :supervisor
        self.workers += 1 if context.type == :worker
      end
      def active() sleeping + running + aborting end
      def sleeping() @status.reduce(0){|x, s| x += (s == 'sleep' ? 1 : 0) } end
      def running() @status.reduce(0){|x, s| x += (s == 'run' ? 1 : 0) } end
      def aborting() @status.reduce(0){|x, s| x += (s == 'aborting' ? 1 : 0) } end
      def stopped() @status.reduce(0){|x, s| x += (s == false ? 1 : 0) } end
      def abend() @status.reduce(0){|x, s| x += (s.nil? ? 1 : 0) } end
    end

    attr_reader :monitor_interval
    attr_reader :restart_strategy
    attr_reader :max_restart
    attr_reader :max_time

    alias_method :strategy, :restart_strategy
    alias_method :max_r, :max_restart
    alias_method :max_t, :max_time

    def initialize(opts = {})
      @restart_strategy = opts[:restart_strategy] || opts[:strategy] || :one_for_one
      @monitor_interval = (opts[:monitor_interval] || DEFAULT_MONITOR_INTERVAL).to_f
      @max_restart = (opts[:max_restart] || opts[:max_r] || DEFAULT_MAX_RESTART).to_i
      @max_time = (opts[:max_time] || opts[:max_t] || DEFAULT_MAX_TIME).to_i

      raise ArgumentError.new(":#{@restart_strategy} is not a valid restart strategy") unless RESTART_STRATEGIES.include?(@restart_strategy)
      raise ArgumentError.new(':monitor_interval must be greater than zero') unless @monitor_interval > 0.0
      raise ArgumentError.new(':max_restart must be greater than zero') unless @max_restart > 0
      raise ArgumentError.new(':max_time must be greater than zero') unless @max_time > 0

      @running = false
      @mutex = Mutex.new
      @workers = []
      @monitor = nil

      @count = WorkerCounts.new(0, 0, 0)
      @restart_times = []

      add_worker(opts[:worker]) unless opts[:worker].nil?
      add_workers(opts[:workers]) unless opts[:workers].nil?
    end

    def run!
      @mutex.synchronize do
        raise StandardError.new('already running') if @running
        @running = true
        @monitor = Thread.new do
          Thread.current.abort_on_exception = false
          monitor
        end
      end
      Thread.pass
    end

    def run
      @mutex.synchronize do
        raise StandardError.new('already running') if @running
        @running = true
      end
      monitor
      true
    end

    def stop
      @mutex.synchronize do
        return true unless @running

        @running = false
        unless @monitor.nil?
          @monitor.run if @monitor.status == 'sleep'
          if @monitor.join(0.1).nil?
            @monitor.kill
          end
          @monitor = nil
        end
        @restart_times.clear

        @workers.length.times do |i|
          context = @workers[-1-i]
          terminate_worker(context)
        end
        prune_workers
      end

      true
    end

    def running?
      @mutex.synchronize { @running }
    end

    def length
      @mutex.synchronize { @workers.length }
    end
    alias_method :size, :length

    def current_restart_count
      @restart_times.length
    end

    def count
      @mutex.synchronize do
        @count.status = @workers.collect{|w| w.thread ? w.thread.status : false }
        @count.dup.freeze
      end
    end

    def add_worker(worker, opts = {})
      return nil if worker.nil? || ! behaves_as_worker?(worker)
      @mutex.synchronize {
        restart = opts[:restart] || :permanent
        type = opts[:type] || (worker.is_a?(Supervisor) ? :supervisor : nil) || :worker
        raise ArgumentError.new(":#{restart} is not a valid restart option") unless CHILD_RESTART_OPTIONS.include?(restart)
        raise ArgumentError.new(":#{type} is not a valid child type") unless CHILD_TYPES.include?(type)
        context = WorkerContext.new(worker, type, restart)
        @workers << context
        @count.add(context)
        worker.run if @running
        context.object_id
      }
    end
    alias_method :add_child, :add_worker

    def add_workers(workers, opts = {})
      workers.collect do |worker|
        add_worker(worker, opts)
      end
    end
    alias_method :add_children, :add_workers

    def remove_worker(worker_id)
      @mutex.synchronize do
        index, context = find_worker(worker_id)
        break(nil) if context.nil?
        break(false) if context.alive?
        @workers.delete_at(index)
        context.worker
      end
    end
    alias_method :remove_child, :remove_worker

    def stop_worker(worker_id)
      @mutex.synchronize do
        return true unless @running

        index, context = find_worker(worker_id)
        break(nil) if index.nil?
        context.terminated = true
        terminate_worker(context)
        @workers.delete_at(index) if @workers[index].restart == :temporary
        true
      end
    end
    alias_method :stop_child, :stop_worker

    def start_worker(worker_id)
      @mutex.synchronize do
        return false unless @running

        index, context = find_worker(worker_id)
        break(nil) if context.nil?
        context.terminated = false
        run_worker(context) unless context.alive?
        true
      end
    end
    alias_method :start_child, :start_worker

    def restart_worker(worker_id)
      @mutex.synchronize do
        return false unless @running

        index, context = find_worker(worker_id)
        break(nil) if context.nil?
        break(false) if context.restart == :temporary
        context.terminated = false
        terminate_worker(context)
        run_worker(context)
        true
      end
    end
    alias_method :restart_child, :restart_worker

    private

    def behaves_as_worker?(obj)
      WORKER_API.each do |method, arity|
        break(false) unless obj.respond_to?(method) && obj.method(method).arity == arity
        true
      end
    end

    def monitor
      @workers.each{|context| run_worker(context)}
      loop do
        sleep(@monitor_interval)
        break unless running?
        @mutex.synchronize do
          prune_workers
          self.send(@restart_strategy)
        end
        break unless running?
      end
    rescue MaxRestartFrequencyError => ex
      stop
    end

    def run_worker(context)
      context.thread = Thread.new do
        Thread.current.abort_on_exception = false
        context.worker.run
      end
      context
    end

    def terminate_worker(context)
      if context.alive?
        context.worker.stop
        Thread.pass
      end
    rescue Exception => ex
      begin
        Thread.kill(context.thread)
      rescue
        # suppress
      end
    ensure
      context.thread = nil
    end

    def prune_workers
      @workers.delete_if{|w| w.restart == :temporary && ! w.alive? }
    end

    def find_worker(worker_id)
      index = @workers.find_index{|worker| worker.object_id == worker_id}
      if index.nil?
        [nil, nil]
      else
        [index, @workers[index]]
      end
    end

    def exceeded_max_restart_frequency?
      @restart_times.unshift(Time.now.to_i)
      diff = (@restart_times.first - @restart_times.last).abs
      if @restart_times.length >= @max_restart && diff <= @max_time
        return true
      elsif diff >= @max_time
        @restart_times.pop
      end
      false
    end

    #----------------------------------------------------------------
    # restart strategies

    def one_for_one
      @workers.each do |context|
        if context.needs_restart?
          raise MaxRestartFrequencyError if exceeded_max_restart_frequency?
          run_worker(context)
        end
      end
    end

    def one_for_all
      restart = false

      restart = @workers.each do |context|
        if context.needs_restart?
          raise MaxRestartFrequencyError if exceeded_max_restart_frequency?
          break(true)
        end
      end

      if restart
        @workers.each do |context|
          terminate_worker(context)
        end
        @workers.each{|context| run_worker(context)}
      end
    end

    def rest_for_one
      restart = false

      @workers.each do |context|
        if restart
          terminate_worker(context)
        elsif context.needs_restart?
          raise MaxRestartFrequencyError if exceeded_max_restart_frequency?
          restart = true
        end
      end

      one_for_one if restart
    end
  end
end

1	require 'thread'	1✔
2
3	require 'concurrent/errors'	1✔
4	require 'concurrent/runnable'	1✔
5
6	module Concurrent	1✔
7
8	class Supervisor	1✔
9
10	DEFAULT_MONITOR_INTERVAL = 1	1✔
11	RESTART_STRATEGIES = [:one_for_one, :one_for_all, :rest_for_one]	1✔
12	DEFAULT_MAX_RESTART = 5	1✔
13	DEFAULT_MAX_TIME = 60	1✔
14	WORKER_API = {run: 0, stop: 0, running?: 0}	1✔
15
16	CHILD_TYPES = [:worker, :supervisor]	1✔
17	CHILD_RESTART_OPTIONS = [:permanent, :transient, :temporary]	1✔
18
19	WorkerContext = Struct.new(:worker, :type, :restart) do	1✔
20	attr_accessor :thread	1✔
21	attr_accessor :terminated	1✔
22
23	def alive?() return thread && thread.alive?; end	1✔
24
25	def needs_restart?	1✔
UNCOV 26	return false if thread && thread.alive?	×
UNCOV 27	return false if terminated	×
UNCOV 28	case self.restart	×
29	when :permanent
UNCOV 30	return true	×
31	when :transient
UNCOV 32	return thread.nil? \|\| thread.status.nil?	×
33	else #when :temporary
34	return false	×
35	end
36	end
37	end
38
39	WorkerCounts = Struct.new(:specs, :supervisors, :workers) do	1✔
40	attr_accessor :status	1✔
41	def add(context)	1✔
UNCOV 42	self.specs += 1	×
UNCOV 43	self.supervisors += 1 if context.type == :supervisor	×
UNCOV 44	self.workers += 1 if context.type == :worker	×
45	end
46	def active() sleeping + running + aborting end	1✔
47	def sleeping() @status.reduce(0){\|x, s\| x += (s == 'sleep' ? 1 : 0) } end	1✔
48	def running() @status.reduce(0){\|x, s\| x += (s == 'run' ? 1 : 0) } end	1✔
49	def aborting() @status.reduce(0){\|x, s\| x += (s == 'aborting' ? 1 : 0) } end	1✔
50	def stopped() @status.reduce(0){\|x, s\| x += (s == false ? 1 : 0) } end	1✔
51	def abend() @status.reduce(0){\|x, s\| x += (s.nil? ? 1 : 0) } end	1✔
52	end
53
54	attr_reader :monitor_interval	1✔
55	attr_reader :restart_strategy	1✔
56	attr_reader :max_restart	1✔
57	attr_reader :max_time	1✔
58
59	alias_method :strategy, :restart_strategy	1✔
60	alias_method :max_r, :max_restart	1✔
61	alias_method :max_t, :max_time	1✔
62
63	def initialize(opts = {})	1✔
UNCOV 64	@restart_strategy = opts[:restart_strategy] \|\| opts[:strategy] \|\| :one_for_one	×
UNCOV 65	@monitor_interval = (opts[:monitor_interval] \|\| DEFAULT_MONITOR_INTERVAL).to_f	×
UNCOV 66	@max_restart = (opts[:max_restart] \|\| opts[:max_r] \|\| DEFAULT_MAX_RESTART).to_i	×
UNCOV 67	@max_time = (opts[:max_time] \|\| opts[:max_t] \|\| DEFAULT_MAX_TIME).to_i	×
68
UNCOV 69	raise ArgumentError.new(":#{@restart_strategy} is not a valid restart strategy") unless RESTART_STRATEGIES.include?(@restart_strategy)	×
UNCOV 70	raise ArgumentError.new(':monitor_interval must be greater than zero') unless @monitor_interval > 0.0	×
UNCOV 71	raise ArgumentError.new(':max_restart must be greater than zero') unless @max_restart > 0	×
UNCOV 72	raise ArgumentError.new(':max_time must be greater than zero') unless @max_time > 0	×
73
UNCOV 74	@running = false	×
UNCOV 75	@mutex = Mutex.new	×
UNCOV 76	@workers = []	×
UNCOV 77	@monitor = nil	×
78
UNCOV 79	@count = WorkerCounts.new(0, 0, 0)	×
UNCOV 80	@restart_times = []	×
81
UNCOV 82	add_worker(opts[:worker]) unless opts[:worker].nil?	×
UNCOV 83	add_workers(opts[:workers]) unless opts[:workers].nil?	×
84	end
85
86	def run!	1✔
UNCOV 87	@mutex.synchronize do	×
UNCOV 88	raise StandardError.new('already running') if @running	×
UNCOV 89	@running = true	×
UNCOV 90	@monitor = Thread.new do	×
UNCOV 91	Thread.current.abort_on_exception = false	×
UNCOV 92	monitor	×
93	end
94	end
UNCOV 95	Thread.pass	×
96	end
97
98	def run	1✔
UNCOV 99	@mutex.synchronize do	×
UNCOV 100	raise StandardError.new('already running') if @running	×
UNCOV 101	@running = true	×
102	end
UNCOV 103	monitor	×
UNCOV 104	true	×
105	end
106
107	def stop	1✔
UNCOV 108	@mutex.synchronize do	×
UNCOV 109	return true unless @running	×
110
UNCOV 111	@running = false	×
UNCOV 112	unless @monitor.nil?	×
UNCOV 113	@monitor.run if @monitor.status == 'sleep'	×
UNCOV 114	if @monitor.join(0.1).nil?	×
UNCOV 115	@monitor.kill	×
116	end
UNCOV 117	@monitor = nil	×
118	end
UNCOV 119	@restart_times.clear	×
120
UNCOV 121	@workers.length.times do \|i\|	×
UNCOV 122	context = @workers[-1-i]	×
UNCOV 123	terminate_worker(context)	×
124	end
UNCOV 125	prune_workers	×
126	end
127
UNCOV 128	true	×
129	end
130
131	def running?	1✔
UNCOV 132	@mutex.synchronize { @running }	×
133	end
134
135	def length	1✔
UNCOV 136	@mutex.synchronize { @workers.length }	×
137	end
138	alias_method :size, :length	1✔
139
140	def current_restart_count	1✔
UNCOV 141	@restart_times.length	×
142	end
143
144	def count	1✔
UNCOV 145	@mutex.synchronize do	×
UNCOV 146	@count.status = @workers.collect{\|w\| w.thread ? w.thread.status : false }	×
UNCOV 147	@count.dup.freeze	×
148	end
149	end
150
151	def add_worker(worker, opts = {})	1✔
UNCOV 152	return nil if worker.nil? \|\| ! behaves_as_worker?(worker)	×
UNCOV 153	@mutex.synchronize {	×
UNCOV 154	restart = opts[:restart] \|\| :permanent	×
UNCOV 155	type = opts[:type] \|\| (worker.is_a?(Supervisor) ? :supervisor : nil) \|\| :worker	×
UNCOV 156	raise ArgumentError.new(":#{restart} is not a valid restart option") unless CHILD_RESTART_OPTIONS.include?(restart)	×
UNCOV 157	raise ArgumentError.new(":#{type} is not a valid child type") unless CHILD_TYPES.include?(type)	×
UNCOV 158	context = WorkerContext.new(worker, type, restart)	×
UNCOV 159	@workers << context	×
UNCOV 160	@count.add(context)	×
UNCOV 161	worker.run if @running	×
UNCOV 162	context.object_id	×
163	}
164	end
165	alias_method :add_child, :add_worker	1✔
166
167	def add_workers(workers, opts = {})	1✔
UNCOV 168	workers.collect do \|worker\|	×
UNCOV 169	add_worker(worker, opts)	×
170	end
171	end
172	alias_method :add_children, :add_workers	1✔
173
174	def remove_worker(worker_id)	1✔
UNCOV 175	@mutex.synchronize do	×
UNCOV 176	index, context = find_worker(worker_id)	×
UNCOV 177	break(nil) if context.nil?	×
UNCOV 178	break(false) if context.alive?	×
UNCOV 179	@workers.delete_at(index)	×
UNCOV 180	context.worker	×
181	end
182	end
183	alias_method :remove_child, :remove_worker	1✔
184
185	def stop_worker(worker_id)	1✔
UNCOV 186	@mutex.synchronize do	×
UNCOV 187	return true unless @running	×
188
UNCOV 189	index, context = find_worker(worker_id)	×
UNCOV 190	break(nil) if index.nil?	×
UNCOV 191	context.terminated = true	×
UNCOV 192	terminate_worker(context)	×
UNCOV 193	@workers.delete_at(index) if @workers[index].restart == :temporary	×
UNCOV 194	true	×
195	end
196	end
197	alias_method :stop_child, :stop_worker	1✔
198
199	def start_worker(worker_id)	1✔
UNCOV 200	@mutex.synchronize do	×
UNCOV 201	return false unless @running	×
202
UNCOV 203	index, context = find_worker(worker_id)	×
UNCOV 204	break(nil) if context.nil?	×
UNCOV 205	context.terminated = false	×
UNCOV 206	run_worker(context) unless context.alive?	×
UNCOV 207	true	×
208	end
209	end
210	alias_method :start_child, :start_worker	1✔
211
212	def restart_worker(worker_id)	1✔
UNCOV 213	@mutex.synchronize do	×
UNCOV 214	return false unless @running	×
215
UNCOV 216	index, context = find_worker(worker_id)	×
UNCOV 217	break(nil) if context.nil?	×
UNCOV 218	break(false) if context.restart == :temporary	×
UNCOV 219	context.terminated = false	×
UNCOV 220	terminate_worker(context)	×
UNCOV 221	run_worker(context)	×
UNCOV 222	true	×
223	end
224	end
225	alias_method :restart_child, :restart_worker	1✔
226
227	private	1✔
228
229	def behaves_as_worker?(obj)	1✔
UNCOV 230	WORKER_API.each do \|method, arity\|	×
UNCOV 231	break(false) unless obj.respond_to?(method) && obj.method(method).arity == arity	×
UNCOV 232	true	×
233	end
234	end
235
236	def monitor	1✔
UNCOV 237	@workers.each{\|context\| run_worker(context)}	×
UNCOV 238	loop do	×
UNCOV 239	sleep(@monitor_interval)	×
UNCOV 240	break unless running?	×
UNCOV 241	@mutex.synchronize do	×
UNCOV 242	prune_workers	×
UNCOV 243	self.send(@restart_strategy)	×
244	end
UNCOV 245	break unless running?	×
246	end
247	rescue MaxRestartFrequencyError => ex
UNCOV 248	stop	×
249	end
250
251	def run_worker(context)	1✔
UNCOV 252	context.thread = Thread.new do	×
UNCOV 253	Thread.current.abort_on_exception = false	×
UNCOV 254	context.worker.run	×
255	end
UNCOV 256	context	×
257	end
258
259	def terminate_worker(context)	1✔
UNCOV 260	if context.alive?	×
UNCOV 261	context.worker.stop	×
UNCOV 262	Thread.pass	×
263	end
264	rescue Exception => ex
265	begin
UNCOV 266	Thread.kill(context.thread)	×
267	rescue
268	# suppress
269	end
270	ensure
UNCOV 271	context.thread = nil	×
272	end
273
274	def prune_workers	1✔
UNCOV 275	@workers.delete_if{\|w\| w.restart == :temporary && ! w.alive? }	×
276	end
277
278	def find_worker(worker_id)	1✔
UNCOV 279	index = @workers.find_index{\|worker\| worker.object_id == worker_id}	×
UNCOV 280	if index.nil?	×
UNCOV 281	[nil, nil]	×
282	else
UNCOV 283	[index, @workers[index]]	×
284	end
285	end
286
287	def exceeded_max_restart_frequency?	1✔
UNCOV 288	@restart_times.unshift(Time.now.to_i)	×
UNCOV 289	diff = (@restart_times.first - @restart_times.last).abs	×
UNCOV 290	if @restart_times.length >= @max_restart && diff <= @max_time	×
UNCOV 291	return true	×
UNCOV 292	elsif diff >= @max_time	×
UNCOV 293	@restart_times.pop	×
294	end
UNCOV 295	false	×
296	end
297
298	#----------------------------------------------------------------
299	# restart strategies
300
301	def one_for_one	1✔
UNCOV 302	@workers.each do \|context\|	×
UNCOV 303	if context.needs_restart?	×
UNCOV 304	raise MaxRestartFrequencyError if exceeded_max_restart_frequency?	×
UNCOV 305	run_worker(context)	×
306	end
307	end
308	end
309
310	def one_for_all	1✔
UNCOV 311	restart = false	×
312
UNCOV 313	restart = @workers.each do \|context\|	×
UNCOV 314	if context.needs_restart?	×
UNCOV 315	raise MaxRestartFrequencyError if exceeded_max_restart_frequency?	×
UNCOV 316	break(true)	×
317	end
318	end
319
UNCOV 320	if restart	×
UNCOV 321	@workers.each do \|context\|	×
UNCOV 322	terminate_worker(context)	×
323	end
UNCOV 324	@workers.each{\|context\| run_worker(context)}	×
325	end
326	end
327
328	def rest_for_one	1✔
UNCOV 329	restart = false	×
330
UNCOV 331	@workers.each do \|context\|	×
UNCOV 332	if restart	×
UNCOV 333	terminate_worker(context)	×
UNCOV 334	elsif context.needs_restart?	×
UNCOV 335	raise MaxRestartFrequencyError if exceeded_max_restart_frequency?	×
UNCOV 336	restart = true	×
337	end
338	end
339
UNCOV 340	one_for_one if restart	×
341	end
342	end
343	end

jdantonio / concurrent-ruby / #812

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