class Arvados::V1::JobsController < ApplicationController
+ accept_attribute_as_json :components, Hash
accept_attribute_as_json :script_parameters, Hash
accept_attribute_as_json :runtime_constraints, Hash
accept_attribute_as_json :tasks_summary, Hash
skip_before_filter :find_object_by_uuid, :only => [:queue, :queue_size]
skip_before_filter :render_404_if_no_object, :only => [:queue, :queue_size]
+ include DbCurrentTime
+
def create
[:repository, :script, :script_version, :script_parameters].each do |r|
if !resource_attrs[r]
params[:find_or_create] = !resource_attrs.delete(:no_reuse)
end
- if params[:find_or_create]
- return if false.equal?(load_filters_param)
- if @filters.empty? # Translate older creation parameters into filters.
- @filters =
- [["repository", "=", resource_attrs[:repository]],
- ["script", "=", resource_attrs[:script]],
- ["script_version", "in git",
- params[:minimum_script_version] || resource_attrs[:script_version]],
- ["script_version", "not in git", params[:exclude_script_versions]],
- ].reject { |filter| filter.last.nil? or filter.last.empty? }
- if image_search = resource_attrs[:runtime_constraints].andand["docker_image"]
- if image_tag = resource_attrs[:runtime_constraints]["docker_image_tag"]
- image_search += ":#{image_tag}"
- end
- @filters.append(["docker_image_locator", "in docker", image_search])
- else
- @filters.append(["docker_image_locator", "=", nil])
- end
- if sdk_version = resource_attrs[:runtime_constraints].andand["arvados_sdk_version"]
- @filters.append(["arvados_sdk_version", "in git", sdk_version])
- end
- begin
- load_job_specific_filters
- rescue ArgumentError => error
- return send_error(error.message)
+ return super if !params[:find_or_create]
+ return if !load_filters_param
+
+ if @filters.empty? # Translate older creation parameters into filters.
+ @filters =
+ [["repository", "=", resource_attrs[:repository]],
+ ["script", "=", resource_attrs[:script]],
+ ["script_version", "not in git", params[:exclude_script_versions]],
+ ].reject { |filter| filter.last.nil? or filter.last.empty? }
+ if !params[:minimum_script_version].blank?
+ @filters << ["script_version", "in git",
+ params[:minimum_script_version]]
+ else
+ add_default_git_filter("script_version", resource_attrs[:repository],
+ resource_attrs[:script_version])
+ end
+ if image_search = resource_attrs[:runtime_constraints].andand["docker_image"]
+ if image_tag = resource_attrs[:runtime_constraints]["docker_image_tag"]
+ image_search += ":#{image_tag}"
end
+ image_locator = Collection.
+ for_latest_docker_image(image_search).andand.portable_data_hash
+ else
+ image_locator = nil
+ end
+ @filters << ["docker_image_locator", "=", image_locator]
+ if sdk_version = resource_attrs[:runtime_constraints].andand["arvados_sdk_version"]
+ add_default_git_filter("arvados_sdk_version", "arvados", sdk_version)
+ end
+ begin
+ load_job_specific_filters
+ rescue ArgumentError => error
+ return send_error(error.message)
end
+ end
- # Check specified filters for some reasonableness.
- filter_names = @filters.map { |f| f.first }.uniq
- ["repository", "script"].each do |req_filter|
- if not filter_names.include?(req_filter)
- return send_error("#{req_filter} filter required")
- end
+ # Check specified filters for some reasonableness.
+ filter_names = @filters.map { |f| f.first }.uniq
+ ["repository", "script"].each do |req_filter|
+ if not filter_names.include?(req_filter)
+ return send_error("#{req_filter} filter required")
end
+ end
- # Search for a reusable Job, and return it if found.
- @objects = Job.readable_by(current_user)
- apply_filters
- @object = nil
- incomplete_job = nil
- @objects.each do |j|
- if j.nondeterministic != true and
- ["Queued", "Running", "Complete"].include?(j.state) and
- j.script_parameters == resource_attrs[:script_parameters]
- if j.state != "Complete" && j.owner_uuid == current_user.uuid
- # We'll use this if we don't find a job that has completed
- incomplete_job ||= j
- else
- if Collection.readable_by(current_user).find_by_portable_data_hash(j.output)
- # Record the first job in the list
- if !@object
- @object = j
- end
- # Ensure that all candidate jobs actually did produce the same output
- if @object.output != j.output
- @object = nil
- break
- end
- end
- end
- end
- @object ||= incomplete_job
- if @object
- return show
+ # Search for a reusable Job, and return it if found.
+ @objects = Job.
+ readable_by(current_user).
+ where('state = ? or (owner_uuid = ? and state in (?))',
+ Job::Complete, current_user.uuid, [Job::Queued, Job::Running]).
+ where('script_parameters = ?', resource_attrs[:script_parameters].to_yaml).
+ where('nondeterministic is distinct from ?', true).
+ order('state desc, created_at') # prefer Running jobs over Queued
+ apply_filters
+ @object = nil
+ incomplete_job = nil
+ @objects.each do |j|
+ if j.state != Job::Complete
+ # We'll use this if we don't find a job that has completed
+ incomplete_job ||= j
+ next
+ end
+
+ if @object
+ if @object.output != j.output
+ # If two matching jobs produced different outputs, just run
+ # a new job instead of choosing one arbitrarily.
+ @object = nil
+ return super
end
+ # ...and that's the only thing we need to do once we've chosen
+ # an @object to reuse.
+ elsif !Collection.readable_by(current_user).find_by_portable_data_hash(j.output)
+ # As soon as the output we will end up returning (if any) is
+ # decided, check whether it will be visible to the user; if
+ # not, any further investigation of reusable jobs is futile.
+ return super
+ else
+ @object = j
end
end
- super
+ @object ||= incomplete_job
+ if @object
+ show
+ else
+ super
+ end
end
def cancel
while not @job.started_at
# send a summary (job queue + available nodes) to the client
# every few seconds while waiting for the job to start
- last_ack_at ||= Time.now - Q_UPDATE_INTERVAL - 1
- if Time.now - last_ack_at >= Q_UPDATE_INTERVAL
+ current_time = db_current_time
+ last_ack_at ||= current_time - Q_UPDATE_INTERVAL - 1
+ if current_time - last_ack_at >= Q_UPDATE_INTERVAL
nodes_in_state = {idle: 0, alloc: 0}
ActiveRecord::Base.uncached do
Node.where('hostname is not ?', nil).collect do |n|
end
end
end
- job_queue = Job.queue
+ job_queue = Job.queue.select(:uuid)
n_queued_before_me = 0
job_queue.each do |j|
break if j.uuid == @job.uuid
n_queued_before_me += 1
end
- yield "#{Time.now}" \
+ yield "#{db_current_time}" \
" job #{@job.uuid}" \
" queue_position #{n_queued_before_me}" \
- " queue_size #{job_queue.size}" \
+ " queue_size #{job_queue.count}" \
" nodes_idle #{nodes_in_state[:idle]}" \
" nodes_alloc #{nodes_in_state[:alloc]}\n"
- last_ack_at = Time.now
+ last_ack_at = db_current_time
end
sleep 3
ActiveRecord::Base.uncached do
load_limit_offset_order_params
load_where_param
@where.merge!({state: Job::Queued})
- return if false.equal?(load_filters_param)
+ return if !load_filters_param
find_objects_for_index
index
end
protected
+ def add_default_git_filter(attr_name, repo_name, refspec)
+ # Add a filter to @filters for `attr_name` = the latest commit available
+ # in `repo_name` at `refspec`. No filter is added if refspec can't be
+ # resolved.
+ commits = Commit.find_commit_range(repo_name, nil, refspec, nil)
+ if commit_hash = commits.first
+ @filters << [attr_name, "=", commit_hash]
+ end
+ end
+
def load_job_specific_filters
# Convert Job-specific @filters entries into general SQL filters.
script_info = {"repository" => nil, "script" => nil}
else
raise ArgumentError.new("unknown attribute for git filter: #{attr}")
end
- version_range = Commit.find_commit_range(current_user,
- filter["repository"],
- filter["min_version"],
- filter["max_version"],
- filter["exclude_versions"])
- if version_range.nil?
+ revisions = Commit.find_commit_range(filter["repository"],
+ filter["min_version"],
+ filter["max_version"],
+ filter["exclude_versions"])
+ if revisions.empty?
raise ArgumentError.
new("error searching #{filter['repository']} from " +
"'#{filter['min_version']}' to '#{filter['max_version']}', " +
"excluding #{filter['exclude_versions']}")
end
- @filters.append([attr, "in", version_range])
+ @filters.append([attr, "in", revisions])
end
end
rescue ArgumentError => error
send_error(error.message)
false
+ else
+ true
end
end
end
projects / arvados.git / blobdiff