def refresh_todo
@todo = Job.queue
+ @todo_pipelines = PipelineInstance.queue
end
def sinfo
job_done = j_done[:job]
$stderr.puts "dispatch: child #{pid_done} exit"
$stderr.puts "dispatch: job #{job_done.uuid} end"
- $redis.publish job_done.uuid, "end"
# Ensure every last drop of stdout and stderr is consumed
read_pipes
# Wait the thread
j_done[:wait_thr].value
+ jobrecord = Job.find_by_uuid(job_done.uuid)
+ jobrecord.running = false
+ jobrecord.finished_at ||= Time.now
+ # Don't set 'jobrecord.success = false' because if the job failed to run due to an
+ # issue with crunch-job or slurm, we want the job to stay in the queue.
+ jobrecord.save!
+
# Invalidate the per-job auth token
j_done[:job_auth].update_attributes expires_at: Time.now
+ $redis.publish job_done.uuid, "end"
+
@running.delete job_done.uuid
end
+ def update_pipelines
+ @todo_pipelines.each do |p|
+ pipe_auth = ApiClientAuthorization.
+ new(user: User.where('uuid=?', p.modified_by_user_uuid).first,
+ api_client_id: 0)
+ pipe_auth.save
+
+ puts `export ARVADOS_API_TOKEN=#{pipe_auth.api_token} && arv-run-pipeline-instance --run-here --no-wait --instance #{p.uuid}`
+ end
+ end
+
def run
act_as_system_user
@running ||= {}
unless @todo.empty? or did_recently(:start_jobs, 1.0) or $signal[:term]
start_jobs
end
+ unless @todo_pipelines.empty? or did_recently(:update_pipelines, 5.0)
+ update_pipelines
+ end
end
reap_children
select(@running.values.collect { |j| [j[:stdout], j[:stderr]] }.flatten,