Class: Blender::Scheduler

Inherits:
Object
  • Object
show all
Includes:
Lock, SchedulerDSL
Defined in:
lib/blender/scheduler.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from SchedulerDSL

#add_handler, #append_task, #ask, #build_task, #concurrency, #config, #driver, #ignore_failure, #lock, #lock_options, #log_level, #members, #on, #ruby_task, #scp_download, #scp_upload, #shell_task, #ssh_task, #strategy

Methods included from Discovery

#build_discovery, #old_search, #search, #search_with_config

Methods included from Utils::Refinements

#camelcase, #symbolize

Constructor Details

#initialize(name, tasks = ThreadSafe::Array.new, options = {}) ⇒ Scheduler

options

noop: true/false - No-Op mode, dont invoke job.run
arguments: array or any stock data
rest everything gets merged as metadata



45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/blender/scheduler.rb', line 45

def initialize(name, tasks = ThreadSafe::Array.new, options = {})
  @name = name
  @tasks = tasks
  @events = Blender::EventDispatcher.new
  @config = ThreadSafe::Hash.new{ ThreadSafe::Hash.new }
  @config[:noop] = options[:noop] || false
  @config[:arguments] = options[:arguments] || []
  unless options.delete(:no_doc)
    events.register(Blender::Handlers::Doc.new)
  end
  @metadata = default_metadata.merge(options)
  @scheduling_strategy = nil
  @lock_properties = {driver: nil, driver_options: {}}
end

Instance Attribute Details

#events ⇒ Object (readonly)

Returns the value of attribute events.



38
39
40
 
# File 'lib/blender/scheduler.rb', line 38

def events
  @events
end

#lock_properties ⇒ Object (readonly)

Returns the value of attribute lock_properties.



39
40
41
 
# File 'lib/blender/scheduler.rb', line 39

def lock_properties
  @lock_properties
end

#metadata ⇒ Object (readonly)

Returns the value of attribute metadata.



36
37
38
 
# File 'lib/blender/scheduler.rb', line 36

def metadata
  @metadata
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



36
37
38
 
# File 'lib/blender/scheduler.rb', line 36

def name
  @name
end

#scheduling_strategy ⇒ Object (readonly)

Returns the value of attribute scheduling_strategy.



37
38
39
 
# File 'lib/blender/scheduler.rb', line 37

def scheduling_strategy
  @scheduling_strategy
end

#tasks ⇒ Object (readonly)

Returns the value of attribute tasks.



38
39
40
 
# File 'lib/blender/scheduler.rb', line 38

def tasks
  @tasks
end

Instance Method Details

#blender_config(key) ⇒ Object 



115
116
117
 
# File 'lib/blender/scheduler.rb', line 115

def blender_config(key)
  @config[key]
end

#concurrent_run(jobs) ⇒ Object 



87
88
89
90
91
92
93
94
95
96
97
 
# File 'lib/blender/scheduler.rb', line 87

def concurrent_run(jobs)
  c = metadata[:concurrency]
  Log.debug("Invoking concurrent run with concurrency:#{c}")
  pool = Utils::ThreadPool.new(c)
  jobs.each do |job|
    pool.add_job do
      run_job(job)
    end
  end
  pool.run_till_done
end

#default_metadata ⇒ Object 



124
125
126
127
128
129
130
131
 
# File 'lib/blender/scheduler.rb', line 124

def default_metadata
  temp = ThreadSafe::Hash.new
  temp[:ignore_failure] = false
  temp[:concurrency] = 0
  temp[:handlers] = ThreadSafe::Array.new
  temp[:members] = ThreadSafe::Array.new
  temp
end

#run ⇒ Object 



60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/blender/scheduler.rb', line 60

def run
  @scheduling_strategy ||= SchedulingStrategy::Default.new
  events.run_started(self)
  events.job_computation_started(scheduling_strategy)
  jobs = scheduling_strategy.compute_jobs(@tasks)
  events.job_computation_finished(self, jobs)
  lock do
    if metadata[:concurrency] > 1
      concurrent_run(jobs)
    else
      serial_run(jobs)
    end
    events.run_finished(self)
    jobs
  end
rescue StandardError => e
  events.run_failed(self, e)
  raise e
end

#run_job(job) ⇒ Object 



99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/blender/scheduler.rb', line 99

def run_job(job)
  events.job_started(job)
  Log.debug("Running job #{job.name}")
  unless blender_config(:noop)
    job.run
  end
  events.job_finished(job)
rescue StandardError => e
  events.job_failed(job, e)
  if metadata[:ignore_failure]
    Log.warn("Exception: #{e.inspect} was suppressed, ignoring failure")
  else
    raise e
  end
end

#serial_run(jobs) ⇒ Object 



80
81
82
83
84
85
 
# File 'lib/blender/scheduler.rb', line 80

def serial_run(jobs)
  Log.debug('Invoking serial run')
  jobs.each do |job|
    run_job(job)
  end
end

#update_config(key, value) ⇒ Object 



119
120
121
122
 
# File 'lib/blender/scheduler.rb', line 119

def update_config(key, value)
  @config[key] = value
  @config
end