Class: Yoda::Server::Scheduler

Inherits:
Object
  • Object
show all
Defined in:
lib/yoda/server/scheduler.rb

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(thread_pool: nil) ⇒ Scheduler

Returns a new instance of Scheduler.

Parameters:

  • thread_pool (Concurrent::ThreadPoolExecutor) (defaults to: nil) 


18
19
20
21
 
# File 'lib/yoda/server/scheduler.rb', line 18

def initialize(thread_pool: nil)
  @thread_pool = thread_pool || self.class.default_thread_pool
  @future_map = Concurrent::Map.new
end

Instance Attribute Details

#future_map ⇒ Concurrent::Map{String => Concurrent::Future} (readonly)

Returns:

  • (Concurrent::Map{String => Concurrent::Future}) 


10
11
12
 
# File 'lib/yoda/server/scheduler.rb', line 10

def future_map
  @future_map
end

#thread_pool ⇒ Concurrent::ThreadPoolExecutor (readonly)

Returns:

  • (Concurrent::ThreadPoolExecutor) 


7
8
9
 
# File 'lib/yoda/server/scheduler.rb', line 7

def thread_pool
  @thread_pool
end

Class Method Details

.default_thread_pool ⇒ Concurrent::ThreadPoolExecutor

Returns:

  • (Concurrent::ThreadPoolExecutor) 


13
14
15
 
# File 'lib/yoda/server/scheduler.rb', line 13

def self.default_thread_pool
  Concurrent.global_fast_executor
end

Instance Method Details

#async(id:, &block) ⇒ Concurrent::Future

Parameters:

  • id (String)

Returns:

  • (Concurrent::Future) 


25
26
27
28
29
30
31
 
# File 'lib/yoda/server/scheduler.rb', line 25

def async(id:, &block)
  future = Concurrent::Future.new(executor: thread_pool) { block.call }
  future.add_observer { |_time, value, reason| future_map.delete(id) }
  future_map.put_if_absent(id, future)
  future.execute
  future
end

#cancel(id) ⇒ Object

Parameters:

  • id (String) 


34
35
36
 
# File 'lib/yoda/server/scheduler.rb', line 34

def cancel(id)
  future_map[id]&.cancel
end

#cancel_all ⇒ Object 



44
45
46
 
# File 'lib/yoda/server/scheduler.rb', line 44

def cancel_all
  future_map.each_value { |future| future&.cancel }
end

#wait_for_termination(timeout:) ⇒ Object

Parameters:

  • timeout (Integer) —

    the maximum number of seconds to wait for shutdown to complete.



39
40
41
42
 
# File 'lib/yoda/server/scheduler.rb', line 39

def wait_for_termination(timeout:)
  thread_pool.shutdown
  thread_pool.wait_for_termination(timeout)
end