Module: HPC::Orchestration
Instance Method Summary collapse
- #get_chains(job, rules, chains = {}) ⇒ Object
- #get_job_dependencies(job, job_rules = nil) ⇒ Object
- #get_recursive_job_dependencies(job) ⇒ Object
- #job_rules(rules, job) ⇒ Object
- #orchestrate_job(job, options) ⇒ Object
- #piggyback(job, job_rules, job_deps) ⇒ Object
- #workload(job, rules, chains, options, seen = nil) ⇒ Object
Instance Method Details
#get_chains(job, rules, chains = {}) ⇒ Object
75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 |
# File 'lib/rbbt/hpc/orchestrate.rb', line 75 def get_chains(job, rules, chains = {}) job_rules = self.job_rules(rules, job) job_deps = get_job_dependencies(job) input_deps = [] job.rec_dependencies.each do |dep| input_deps.concat dep.input_dependencies end job_deps.each do |dep| input_deps.concat dep.input_dependencies get_chains(dep, rules, chains) end job_deps.select do |dep| chained = job_rules["chain_tasks"] && job_rules["chain_tasks"][job.workflow.to_s] && job_rules["chain_tasks"][job.workflow.to_s].include?(job.task_name.to_s) && job_rules["chain_tasks"][dep.workflow.to_s] && job_rules["chain_tasks"][dep.workflow.to_s].include?(dep.task_name.to_s) dep_skip = dep.done? && ! input_deps.include?(dep) && self.job_rules(rules, dep)["skip"] chained || dep_skip end.each do |dep| chains[job] ||= [] chains[job] << dep chains[job].concat chains[dep] if chains[dep] end chains end |
#get_job_dependencies(job, job_rules = nil) ⇒ Object
56 57 58 59 60 |
# File 'lib/rbbt/hpc/orchestrate.rb', line 56 def get_job_dependencies(job, job_rules = nil) deps = job.dependencies || [] deps += job.input_dependencies || [] deps end |
#get_recursive_job_dependencies(job) ⇒ Object
62 63 64 65 |
# File 'lib/rbbt/hpc/orchestrate.rb', line 62 def get_recursive_job_dependencies(job) deps = get_job_dependencies(job) (deps + deps.collect{|dep| get_recursive_job_dependencies(dep) }).flatten end |
#job_rules(rules, job) ⇒ Object
5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 |
# File 'lib/rbbt/hpc/orchestrate.rb', line 5 def job_rules(rules, job) workflow = job.workflow.to_s task_name = job.task_name.to_s task_name = job.overriden.to_s if Symbol === job.overriden defaults = rules["defaults"] || {} defaults = defaults.merge(rules[workflow]["defaults"] || {}) if rules[workflow] job_rules = IndiferentHash.setup(defaults.dup) rules["chains"].each do |name,info| IndiferentHash.setup(info) chain_tasks = info[:tasks].split(/,\s*/) chain_tasks.each do |task| task_workflow, chain_task = task.split("#") chain_task, task_workflow = task_workflow, info[:workflow] if chain_task.nil? or chain_tasks.empty? job_rules["chain_tasks"] ||= {} job_rules["chain_tasks"][task_workflow] ||= [] job_rules["chain_tasks"][task_workflow] << chain_task next unless task_name == chain_task.to_s && workflow == task_workflow.to_s config_keys = job_rules.delete :config_keys job_rules = IndiferentHash.setup(job_rules.merge(info)) if config_keys config_keys.gsub!(/,\s+/,',') job_rules[:config_keys] = job_rules[:config_keys] ? config_keys + "," + job_rules[:config_keys] : config_keys end end if job_rules["chain_tasks"][workflow] && job_rules["chain_tasks"][workflow].include?(task_name) break else job_rules.delete "chain_tasks" end end if rules["chains"] config_keys = job_rules.delete :config_keys job_rules = IndiferentHash.setup(job_rules.merge(rules[workflow][task_name])) if rules[workflow] && rules[workflow][task_name] if config_keys config_keys.gsub!(/,\s+/,',') job_rules[:config_keys] = job_rules[:config_keys] ? config_keys + "," + job_rules[:config_keys] : config_keys end if rules["skip"] && rules["skip"][workflow] job_rules["skip"] = true if rules["skip"][workflow].split(/,\s*/).include? task_name end job_rules end |
#orchestrate_job(job, options) ⇒ Object
193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 |
# File 'lib/rbbt/hpc/orchestrate.rb', line 193 def orchestrate_job(job, ) .delete "recursive_clean" .delete "clean_task" .delete "clean" .delete "tail" .delete "printpath" .delete "detach" .delete "jobname" rules = YAML.load(Open.read([:orchestration_rules])) if [:orchestration_rules] rules ||= {} IndiferentHash.setup(rules) chains = get_chains(job, rules) workload(job, rules, chains, ) end |
#piggyback(job, job_rules, job_deps) ⇒ Object
67 68 69 70 71 72 73 |
# File 'lib/rbbt/hpc/orchestrate.rb', line 67 def piggyback(job, job_rules, job_deps) return false unless job_rules["skip"] final_deps = job_deps - job_deps.collect{|dep| get_recursive_job_dependencies(dep)}.flatten.uniq final_deps = final_deps.reject{|dep| dep.done? } return final_deps.first if final_deps.length == 1 return false end |
#workload(job, rules, chains, options, seen = nil) ⇒ Object
105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 |
# File 'lib/rbbt/hpc/orchestrate.rb', line 105 def workload(job, rules, chains, , seen = nil) return [] if job.done? if seen.nil? seen = {} target_job = true end job_rules = self.job_rules(rules, job) job_deps = get_job_dependencies(job) chain = chains[job] chain = chain.reject{|j| seen.include? j.path} if chain chain = chain.reject{|dep| dep.done? } if chain piggyback = piggyback(job, job_rules, job_deps) dep_ids = job_deps.collect do |dep| seen[dep.path] ||= nil if chain && chain.include?(dep) #&& ! job.input_dependencies.include?(dep) = IndiferentHash.setup(.dup) if piggyback and piggyback == dep [:piggyback] ||= [] [:piggyback].push job ids = workload(dep, rules, chains, , seen) else .delete :piggyback ids = workload(dep, rules, chains, , seen) end ids = [ids].flatten.compact.collect{|id| ['canfail', id] * ":"} if job.canfail_paths.include? dep.path seen[dep.path] = ids ids end.compact.flatten.uniq return seen[job.path] || dep_ids if seen.include?(job.path) if piggyback and seen[piggyback.path] return seen[job.path] = seen[piggyback.path] end job_rules.delete :chain_tasks job_rules.delete :tasks job_rules.delete :workflow = IndiferentHash.setup(.merge(job_rules).merge(:batch_dependencies => dep_ids)) .delete :orchestration_rules config_keys = job_rules.delete(:config_keys) if config_keys config_keys.gsub!(/,\s+/,',') [:config_keys] = [:config_keys] ? config_keys + "," + [:config_keys] : config_keys end if [:piggyback] manifest = [:piggyback].uniq manifest += [job] manifest.concat chain if chain job = [:piggyback].first job_rules = self.job_rules(rules, job) new_config_keys = self.job_rules(rules, job)[:config_keys] if new_config_keys new_config_keys = new_config_keys.gsub(/,\s+/,',') [:config_keys] = [:config_keys] ? new_config_keys + "," + [:config_keys] : new_config_keys end .delete :piggyback else manifest = [job] manifest.concat chain if chain end manifest.uniq! [:manifest] = manifest.collect{|j| j.task_signature } [:config_keys] = [:config_keys].split(",").uniq * "," if [:config_keys] if [:dry_run] puts Log.color(:magenta, "Manifest: ") + Log.color(:blue, [:manifest] * ", ") + " - tasks: #{[:task_cpus] || 1} - time: #{[:time]} - config: #{[:config_keys]}" puts Log.color(:yellow, "Deps: ") + Log.color(:blue, [:batch_dependencies]*", ") [:manifest].first else run_job(job, ) end end |