[]
end
end
- def self.create(attributes)
+ def self.create(job, create_params)
@cache ||= {}
result = $client.execute(:api_method => $arvados.jobs.create,
:parameters => {
:api_token => ENV['ARVADOS_API_TOKEN'],
- :job => attributes.to_json
- },
+ :job => job.to_json
+ }.merge(create_params),
:authenticated => false)
j = JSON.parse result.body, :symbolize_names => true
if j.is_a? Hash and j[:uuid]
@cache[j[:uuid]] = j
else
- debuglog "create job: #{j[:errors] rescue nil} with attribute #{attributes}", 0
+ debuglog "create job: #{j[:errors] rescue nil} with attributes #{job}", 0
nil
end
end
@instance ||= PipelineInstance.
create(:components => @components,
:pipeline_template_uuid => @template[:uuid],
- :active => true)
+ :state => 'RunningOnClient')
self
end
def run
moretodo = true
+ interrupted = false
+
while moretodo
moretodo = false
@components.each do |cname, c|
job = nil
+ # Is the job satisfying this component already known to be
+ # finished? (Already meaning "before we query API server about
+ # the job's current state")
c_already_finished = (c[:job] &&
c[:job][:uuid] &&
!c[:job][:success].nil?)
# No job yet associated with this component and is component inputs
# are fully specified (any output_of script_parameters are resolved
# to real value)
- job = JobCache.create({:script => c[:script],
- :script_parameters => c[:script_parameters],
- :script_version => c[:script_version],
- :repository => c[:repository],
- :minimum_script_version => c[:minimum_script_version],
- :exclude_script_versions => c[:exclude_minimum_script_versions],
- :nondeterministic => c[:nondeterministic],
- :no_reuse => @options[:no_reuse],
- :output_is_persistent => c[:output_is_persistent] || false})
+ job = JobCache.create({
+ :script => c[:script],
+ :script_parameters => c[:script_parameters],
+ :script_version => c[:script_version],
+ :repository => c[:repository],
+ :nondeterministic => c[:nondeterministic],
+ :output_is_persistent => c[:output_is_persistent] || false,
+ # TODO: Delete the following three attributes when
+ # supporting pre-20140418 API servers is no longer
+ # important. New API servers take these as flags that
+ # control behavior of create, rather than job attributes.
+ :minimum_script_version => c[:minimum_script_version],
+ :exclude_script_versions => c[:exclude_minimum_script_versions],
+ :no_reuse => @options[:no_reuse] || c[:nondeterministic],
+ }, {
+ # This is the right place to put these attributes when
+ # dealing with new API servers.
+ :minimum_script_version => c[:minimum_script_version],
+ :exclude_script_versions => c[:exclude_minimum_script_versions],
+ :find_or_create => !(@options[:no_reuse] || c[:nondeterministic]),
+ })
if job
debuglog "component #{cname} new job #{job[:uuid]}"
c[:job] = job
if (c[:job][:running] or
not (c[:job][:finished_at] or c[:job][:cancelled_at]))
# Job is running so update copy of job record
- c[:job] = JobCache.get(c[:job][:uuid])
+ c[:job] = JobCache.get(c[:job][:uuid])
end
if c[:job][:success]
end
end
unless c_already_finished
+ # This is my first time discovering that the job
+ # succeeded. (At the top of this loop, I was still
+ # waiting for it to finish.)
if c[:output_is_persistent]
- # This is my first time discovering that the job
- # succeeded. I need to make sure a resources/wants
- # link is in place to protect the output from garbage
+ # I need to make sure a resources/wants link is in
+ # place to protect the output from garbage
# collection. (Normally Crunch does this for me, but
# here I might be reusing the output of someone else's
# job and I need to make sure it's understood that the
sleep 10
rescue Interrupt
debuglog "interrupt", 0
- abort
+ interrupted = true
+ break
+ #abort
end
end
end
end
end
end
-
- if ended == @components.length or failed > 0
- @instance[:active] = false
- @instance[:success] = (succeeded == @components.length)
+
+ success = (succeeded == @components.length)
+
+ if interrupted
+ if success
+ @instance[:active] = false
+ @instance[:success] = success
+ @instance[:state] = "Complete"
+ else
+ @instance[:active] = nil
+ @instance[:success] = nil
+ @instance[:state] = 'Paused'
+ end
+ else
+ if ended == @components.length or failed > 0
+ @instance[:active] = false
+ @instance[:success] = success
+ @instance[:state] = success ? "Complete" : "Failed"
+ end
end
+ # set components_summary
+ components_summary = {"todo" => @components.length - ended, "done" => succeeded, "failed" => failed}
+ @instance[:components_summary] = components_summary
+
@instance.save
end
projects / arvados.git / blobdiff