class MiGA::Daemon

MiGA Daemons handling job submissions.

Attributes

jobs_running[R]

Array of jobs currently running

jobs_to_run[R]

Array of jobs next to be executed

options[R]

Options used to setup the daemon

project[R]

MiGA::Project in which the daemon is running

Public Class Methods

daemon_home(project) click to toggle source

Daemon's home inside the MiGA::Project project or a String with the full path to the project's 'daemon' folder

# File lib/miga/daemon.rb, line 19
def daemon_home(project)
  return project if project.is_a? String

  File.join(project.path, 'daemon')
end
new(project, json = nil) click to toggle source

Initialize an unactive daemon for the MiGA::Project project. See daemon to wake the daemon. If passed, json must be the path to a daemon definition in json format. Otherwise, the project-stored daemon definition is used. In either case, missing variables are used as defined in ~/.miga_daemon.json.

# File lib/miga/daemon.rb, line 44
def initialize(project, json = nil)
  @project = project
  @runopts = {}
  json ||= File.join(project.path, 'daemon/daemon.json')
  default_json = File.expand_path('.miga_daemon.json', ENV['MIGA_HOME'])
  MiGA::Json.parse(
    json, default: File.exist?(default_json) ? default_json : nil
  ).each { |k, v| runopts(k, v) }
  update_format_0
  @jobs_to_run = []
  @jobs_running = []
end

Public Instance Methods

check_datasets() click to toggle source

Traverse datasets, and returns boolean indicating if at any reference datasets are incomplete

# File lib/miga/daemon.rb, line 169
def check_datasets
  l_say(2, 'Checking datasets')
  o = false
  project.each_dataset do |ds|
    next unless ds.status == :incomplete
    next if ds.next_preprocessing(false).nil?

    o = true if ds.ref?
    queue_job(:d, ds)
  end
  unless show_log?
    n = project.dataset_names.count
    k = jobs_to_run.size + jobs_running.size
    advance('Datasets:', n - k, n, false)
    miga_say if k == 0
  end
  o
end
check_project() click to toggle source

Check if all reference datasets are pre-processed. If yes, check the project-level tasks

# File lib/miga/daemon.rb, line 191
def check_project
  l_say(2, 'Checking project')

  # Ignore task if the project has no datasets
  return if project.dataset_names.empty?

  # Double-check if all datasets are ready
  return unless project.done_preprocessing?

  # Queue project-level job
  to_run = project.next_task(nil, false)
  queue_job(:p) unless to_run.nil?
end
daemon_first_loop() click to toggle source

Run only in the first loop

# File lib/miga/daemon.rb, line 71
def daemon_first_loop
  say '-----------------------------------'
  say 'MiGA:%s launched' % project.name
  say '-----------------------------------'
  miga_say "Saving log to: #{output_file}" unless show_log?
  recalculate_status!
  load_status
  say 'Configuration options:'
  say @runopts.to_s
end
daemon_home() click to toggle source

Path to the daemon home

# File lib/miga/daemon.rb, line 59
def daemon_home
  self.class.daemon_home(project)
end
daemon_loop() click to toggle source

Run one loop step. Returns a Boolean indicating if the loop should continue

# File lib/miga/daemon.rb, line 84
def daemon_loop
  l_say(3, 'Daemon loop start')
  reload_project
  check_datasets or check_project
  if shutdown_when_done? && (jobs_running.size + jobs_to_run.size).zero?
    say 'Nothing else to do, shutting down'
    return false
  end
  flush!
  if (loop_i % 12).zero?
    purge!
    # TEMPORARILY DISABLED:
    # recalculate_status!
  end
  save_status
  sleep(latency)
  l_say(3, 'Daemon loop end')
  true
end
daemon_name() click to toggle source

Name of the daemon

# File lib/miga/daemon.rb, line 65
def daemon_name
  "MiGA:#{project.name}"
end
flush!() click to toggle source

Remove finished jobs from the internal queue and launch as many as possible respecting maxjobs or nodelist (if set).

# File lib/miga/daemon.rb, line 254
def flush!
  # Check for finished jobs
  l_say(2, 'Checking for finished jobs')
  @jobs_running.select! do |job|
    ongoing =
      case job[:job].to_s
      when 'd'
        !job[:ds].nil? && !job[:ds].next_preprocessing(false).nil?
      when 'p'
        !project.next_task(nil, false).nil?
      else
        (job[:ds].nil? ? project : job[:ds]).add_result(job[:job], false).nil?
      end
    say "Completed pid:#{job[:pid]} for #{job[:task_name]}" unless ongoing
    ongoing
  end

  # Avoid single datasets hogging resources
  @jobs_to_run.rotate! rand(jobs_to_run.size)

  # Launch as many +jobs_to_run+ as possible
  while (hostk = next_host)
    break if jobs_to_run.empty?

    launch_job(@jobs_to_run.shift, hostk)
  end
end
get_job(job, ds = nil) click to toggle source

Get the taks with key symbol job in dataset ds. For project-wide tasks let ds be nil.

# File lib/miga/daemon.rb, line 241
def get_job(job, ds = nil)
  (jobs_to_run + jobs_running).find do |j|
    if ds.nil?
      j[:ds].nil? && j[:job] == job
    else
      !j[:ds].nil? && j[:ds].name == ds.name && j[:job] == job
    end
  end
end
l_say(level, *msg) click to toggle source

Send msg to say as long as level is at most verbosity

# File lib/miga/daemon.rb, line 111
def l_say(level, *msg)
  say(*msg) if verbosity >= level
end
launch_job(job, hostk = nil) click to toggle source

Launch the job described by Hash job to hostk-th host

# File lib/miga/daemon.rb, line 304
def launch_job(job, hostk = nil)
  # Execute job
  case runopts(:type)
  when 'ssh'
    # Remote job
    job[:hostk] = hostk
    job[:cmd] = job[:cmd].miga_variables(host: nodelist[hostk])
    job[:pid] = spawn job[:cmd]
    Process.detach job[:pid] unless [nil, '', 0].include?(job[:pid])
  when 'bash'
    # Local job
    job[:pid] = spawn job[:cmd]
    Process.detach job[:pid] unless [nil, '', 0].include?(job[:pid])
  else
    # Schedule cluster job (qsub, msub, slurm)
    job[:pid] = `#{job[:cmd]}`.chomp
  end

  # Check if registered
  if [nil, '', 0].include? job[:pid]
    job[:pid] = nil
    @jobs_to_run << job
    say "Unsuccessful #{job[:task_name]}, rescheduling"
  else
    @jobs_running << job
    job_host = " to #{job[:hostk]}:#{nodelist[job[:hostk]]}" if job[:hostk]
    say "Spawned pid:#{job[:pid]}#{job_host} for #{job[:task_name]}"
  end
end
load_status() click to toggle source

Load the status of a previous instance.

# File lib/miga/daemon.rb, line 142
def load_status
  f_path = File.join(daemon_home, 'status.json')
  return unless File.size? f_path

  say 'Loading previous status in daemon/status.json:'
  status = MiGA::Json.parse(f_path)
  status.each_key do |i|
    status[i].map! do |j|
      j.tap do |k|
        unless k[:ds].nil? || k[:ds_name] == 'miga-project'
          k[:ds] = project.dataset(k[:ds_name])
        end
        k[:job] = k[:job].to_sym unless k[:job].nil?
      end
    end
  end
  @jobs_running = status[:jobs_running]
  @jobs_to_run  = status[:jobs_to_run]
  say "- jobs left running: #{@jobs_running.size}"
  purge!
  say "- jobs running: #{@jobs_running.size}"
  say "- jobs to run: #{@jobs_to_run.size}"
end
miga_say(*msg)
Alias for: say
next_host() click to toggle source

In SSH daemons, retrieve the host index of an available node, nil if none. In any other daemons, returns true as long as maxjobs is not reached

# File lib/miga/daemon.rb, line 285
def next_host
  return jobs_running.size < maxjobs if runopts(:type) != 'ssh'

  allk = (0..nodelist.size - 1).to_a
  busyk = jobs_running.map { |k| k[:hostk] }
  (allk - busyk).first
end
purge!() click to toggle source

Remove dead jobs.

# File lib/miga/daemon.rb, line 295
def purge!
  say 'Probing running jobs'
  @jobs_running.select! do |job|
    `#{runopts(:alive).miga_variables(pid: job[:pid])}`.chomp.to_i == 1
  end
end
queue_job(job, ds = nil) click to toggle source

Add the task to the internal queue with symbol key job. If the task is dataset-specific, ds specifies the dataset. To submit jobs to the scheduler (or to bash or ssh) see flush!

# File lib/miga/daemon.rb, line 209
def queue_job(job, ds = nil)
  return nil unless get_job(job, ds).nil?

  ds_name = (ds.nil? ? 'miga-project' : ds.name)
  say 'Queueing %s:%s' % [ds_name, job]
  vars = {
    'PROJECT' => project.path,
    'RUNTYPE' => runopts(:type),
    'CORES' => ppn,
    'MIGA' => MiGA::MiGA.root_path
  }
  vars['DATASET'] = ds.name unless ds.nil?
  log_dir = File.expand_path("daemon/#{job}", project.path)
  Dir.mkdir(log_dir) unless Dir.exist? log_dir
  task_name = "#{project.metadata[:name][0..9]}:#{job}:#{ds_name}"
  to_run = { ds: ds, ds_name: ds_name, job: job, task_name: task_name }
  to_run[:cmd] = runopts(:cmd).miga_variables(
    script: MiGA::MiGA.script_path(job, miga: vars['MIGA'], project: project),
    vars: vars.map do |k, v|
            runopts(:var).miga_variables(key: k, value: v)
          end.join(runopts(:varsep)),
    cpus: ppn,
    log: File.expand_path("#{ds_name}.log", log_dir),
    task_name: task_name,
    miga: File.expand_path('bin/miga', MiGA::MiGA.root_path).shellescape
  )
  @jobs_to_run << to_run
end
recalculate_status!() click to toggle source
# File lib/miga/daemon.rb, line 104
def recalculate_status!
  say 'Recalculating status for all datasets'
  project.each_dataset(&:recalculate_status)
end
reload_project() click to toggle source

Reload the project's metadata

# File lib/miga/daemon.rb, line 125
def reload_project
  l_say(2, 'Reloading project')
  project.load
end
save_status() click to toggle source

Report status in a JSON file.

# File lib/miga/daemon.rb, line 132
def save_status
  l_say(2, 'Saving current status')
  MiGA::Json.generate(
    { jobs_running: @jobs_running, jobs_to_run: @jobs_to_run },
    File.join(daemon_home, 'status.json')
  )
end
say(*msg) click to toggle source

Same as l_say with +level = 1+

Calls superclass method MiGA::MiGA#say
# File lib/miga/daemon.rb, line 119
def say(*msg)
  super(logfh, *msg) if verbosity >= 1
end
Also aliased as: miga_say
update_format_0() click to toggle source

Update from daemon JSON format 0 to the latest version

# File lib/miga/daemon.rb, line 336
def update_format_0
  {
    cmd: %w[script vars cpus log task_name],
    var: %w[key value],
    alive: %w[pid],
    kill: %w[pid]
  }.each do |k, v|
    if !runopts(k).nil? && runopts(k) =~ /%(\d+\$)?[ds]/
      runopts(
        k, runopts(k).gsub(/%(\d+\$)?d/, '%\\1s') % v.map { |i| "{{#{i}}}" }
      )
    end
  end
  runopts(:format_version, 1)
end