- candidates = Container.
- where('command = ?', attrs[:command].to_yaml).
- where('cwd = ?', attrs[:cwd]).
- where('environment = ?', self.deep_sort_hash(attrs[:environment]).to_yaml).
- where('output_path = ?', attrs[:output_path]).
- where('container_image = ?', attrs[:container_image]).
- where('mounts = ?', self.deep_sort_hash(attrs[:mounts]).to_yaml).
- where('runtime_constraints = ?', self.deep_sort_hash(attrs[:runtime_constraints]).to_yaml).
- where('state in (?)', [Queued, Locked, Running, Complete]).
- reject {|c| c.state == Complete and
- (c.exit_code != 0 or c.output.nil? or c.log.nil?)}
- if candidates.empty?
- nil
- elsif candidates.count == 1
- candidates.first
+ log_reuse_info { "starting with #{Container.all.count} container records in database" }
+ candidates = Container.where_serialized(:command, attrs[:command])
+ log_reuse_info(candidates) { "after filtering on command #{attrs[:command].inspect}" }
+
+ candidates = candidates.where('cwd = ?', attrs[:cwd])
+ log_reuse_info(candidates) { "after filtering on cwd #{attrs[:cwd].inspect}" }
+
+ candidates = candidates.where_serialized(:environment, attrs[:environment])
+ log_reuse_info(candidates) { "after filtering on environment #{attrs[:environment].inspect}" }
+
+ candidates = candidates.where('output_path = ?', attrs[:output_path])
+ log_reuse_info(candidates) { "after filtering on output_path #{attrs[:output_path].inspect}" }
+
+ image = resolve_container_image(attrs[:container_image])
+ candidates = candidates.where('container_image = ?', image)
+ log_reuse_info(candidates) { "after filtering on container_image #{image.inspect} (resolved from #{attrs[:container_image].inspect})" }
+
+ candidates = candidates.where_serialized(:mounts, resolve_mounts(attrs[:mounts]))
+ log_reuse_info(candidates) { "after filtering on mounts #{attrs[:mounts].inspect}" }
+
+ candidates = candidates.where('secret_mounts_md5 = ?', Digest::MD5.hexdigest(SafeJSON.dump(self.deep_sort_hash(attrs[:secret_mounts]))))
+ log_reuse_info(candidates) { "after filtering on mounts #{attrs[:mounts].inspect}" }
+
+ candidates = candidates.where_serialized(:runtime_constraints, resolve_runtime_constraints(attrs[:runtime_constraints]))
+ log_reuse_info(candidates) { "after filtering on runtime_constraints #{attrs[:runtime_constraints].inspect}" }
+
+ log_reuse_info { "checking for state=Complete with readable output and log..." }
+
+ select_readable_pdh = Collection.
+ readable_by(current_user).
+ select(:portable_data_hash).
+ to_sql
+
+ usable = candidates.where(state: Complete, exit_code: 0)
+ log_reuse_info(usable) { "with state=Complete, exit_code=0" }
+
+ usable = usable.where("log IN (#{select_readable_pdh})")
+ log_reuse_info(usable) { "with readable log" }
+
+ usable = usable.where("output IN (#{select_readable_pdh})")
+ log_reuse_info(usable) { "with readable output" }
+
+ usable = usable.order('finished_at ASC').limit(1).first
+ if usable
+ log_reuse_info { "done, reusing container #{usable.uuid} with state=Complete" }
+ return usable
+ end
+
+ # Check for Running candidates and return the most likely to finish sooner.
+ log_reuse_info { "checking for state=Running..." }
+ running = candidates.where(state: Running).
+ order('progress desc, started_at asc').
+ limit(1).first
+ if running
+ log_reuse_info { "done, reusing container #{running.uuid} with state=Running" }
+ return running
+ else
+ log_reuse_info { "have no containers in Running state" }
+ end
+
+ # Check for Locked or Queued ones and return the most likely to start first.
+ locked_or_queued = candidates.
+ where("state IN (?)", [Locked, Queued]).
+ order('state asc, priority desc, created_at asc').
+ limit(1).first
+ if locked_or_queued
+ log_reuse_info { "done, reusing container #{locked_or_queued.uuid} with state=#{locked_or_queued.state}" }
+ return locked_or_queued