Class: MiGA::Daemon

Inherits:
MiGA
  • Object
show all
Extended by:
Common::WithDaemonClass
Includes:
Common::WithDaemon, Base
Defined in:
lib/miga/daemon.rb,
lib/miga/daemon/base.rb

Overview

MiGA Daemons handling job submissions.

Defined Under Namespace

Modules: Base

Constant Summary

Constants included from MiGA

CITATION, VERSION, VERSION_DATE, VERSION_NAME

Instance Attribute Summary collapse

Attributes included from Common::WithDaemon

#declare_alive_pid, #loop_i

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Common::WithDaemonClass

alive_file, last_alive, terminated_file

Methods included from Common::WithDaemon

#active?, #alive_file, #daemon, #declare_alive, #declare_alive_loop, #default_options, #in_loop, #last_alive, #launch_daemon_proc, #output_file, #pid_file, #process_alive?, #run, #start, #status, #stop, #terminate, #terminate_file, #terminated_file, #termination_file?, #write_alive_file

Methods included from Base

#bypass_maintenance?, #latency, #logfh, #maxjobs, #nodelist, #ppn, #runopts, #runopts_for, #show_log!, #show_log?, #show_summary!, #shutdown_when_done?, #skip_maintenance, #verbosity

Methods inherited from MiGA

CITATION, CITATION_ARRAY, DEBUG, DEBUG_OFF, DEBUG_ON, DEBUG_TRACE_OFF, DEBUG_TRACE_ON, FULL_VERSION, LONG_VERSION, VERSION, VERSION_DATE, #advance, debug?, debug_trace?, initialized?, #like_io?, #num_suffix, rc_path, #result_files_exist?

Methods included from Common::Path

#root_path, #script_path

Methods included from Common::Format

#clean_fasta_file, #seqs_length, #tabulate

Methods included from Common::Net

#download_file_ftp, #known_hosts, #remote_connection

Methods included from Common::SystemCall

#run_cmd, #run_cmd_opts

Constructor Details

#initialize(project, json = nil) ⇒ Daemon

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.



44
45
46
47
48
49
50
51
52
53
54
55
 
# 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

Instance Attribute Details

#jobs_running ⇒ Object (readonly)

Array of jobs currently running



36
37
38
 
# File 'lib/miga/daemon.rb', line 36

def jobs_running
  @jobs_running
end

#jobs_to_run ⇒ Object (readonly)

Array of jobs next to be executed



33
34
35
 
# File 'lib/miga/daemon.rb', line 33

def jobs_to_run
  @jobs_to_run
end

#options ⇒ Object (readonly)

Options used to setup the daemon



30
31
32
 
# File 'lib/miga/daemon.rb', line 30

def options
  @options
end

#project ⇒ Object (readonly)

MiGA::Project in which the daemon is running



27
28
29
 
# File 'lib/miga/daemon.rb', line 27

def project
  @project
end

Class Method Details

.daemon_home(project) ⇒ Object

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



19
20
21
22
23
 
# File 'lib/miga/daemon.rb', line 19

def daemon_home(project)
  return project if project.is_a? String

  File.join(project.path, 'daemon')
end

Instance Method Details

#check_datasets ⇒ Object

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



179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
 
# File 'lib/miga/daemon.rb', line 179

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
    k -= 1 unless get_job(:maintenance).nil?
    advance('Datasets:', n - k, n, false)
    miga_say if k == 0
  end
  o
end

#check_project ⇒ Object

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



202
203
204
205
206
207
208
209
210
211
212
213
214
 
# File 'lib/miga/daemon.rb', line 202

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 ⇒ Object

Run only in the first loop



71
72
73
74
75
76
77
78
79
80
 
# 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?
  say 'Configuration options:'
  say @runopts.to_s
  load_status
  queue_maintenance(true)
end

#daemon_home ⇒ Object

Path to the daemon home



59
60
61
 
# File 'lib/miga/daemon.rb', line 59

def daemon_home
  self.class.daemon_home(project)
end

#daemon_loop ⇒ Object

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



84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
 
# 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'
    exit_cleanup
    return false
  end
  flush!
  if (loop_i % 12).zero?
    purge!
    queue_maintenance if (loop_i % (12 * (skip_maintenance + 1))).zero?
  end
  save_status
  sleep(latency)
  l_say(3, 'Daemon loop end')
  true
end

#daemon_name ⇒ Object

Name of the daemon



65
66
67
 
# File 'lib/miga/daemon.rb', line 65

def daemon_name
  "MiGA:#{project.name}"
end

#exit_cleanup ⇒ Object

Remove temporary files on completion



115
116
117
 
# File 'lib/miga/daemon.rb', line 115

def exit_cleanup
  FileUtils.rm_f(File.join(daemon_home, 'status.json'))
end

#flush! ⇒ Object

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



275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
 
# File 'lib/miga/daemon.rb', line 275

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)

  # Prioritize: Project-wide > MiGA Online queries > Other datasets
  @jobs_to_run.sort_by! do |job|
    job[:ds].nil? ? 1 : job[:ds_name] =~ /^qG_/ ? 2 : 3
  end

  # 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) ⇒ Object

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



262
263
264
265
266
267
268
269
270
 
# File 'lib/miga/daemon.rb', line 262

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

#job_cmd(to_run) ⇒ Object

Construct the command for the given job definition with current daemon settings



233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
 
# File 'lib/miga/daemon.rb', line 233

def job_cmd(to_run)
  what = to_run[:ds].nil? ? :project : :dataset
  vars = {
    'PROJECT' => project.path,
    'RUNTYPE' => runopts_for(:type, what),
    'CORES' => ppn(what),
    'MIGA' => MiGA::MiGA.root_path
  }
  vars['DATASET'] = to_run[:ds].name unless to_run[:ds].nil?
  log_dir = File.expand_path("daemon/#{to_run[:job]}", project.path)
  FileUtils.mkdir_p(log_dir)
  var_hsh = {
    script: MiGA::MiGA.script_path(
              to_run[:job], miga: vars['MIGA'], project: project
            ),
    vars: vars.map do |k, v|
            runopts(:var).miga_variables(key: k, value: v)
          end.join(runopts_for(:varsep, what)),
    cpus: ppn(what),
    log: File.join(log_dir, "#{to_run[:ds_name]}.log"),
    task_name: to_run[:task_name],
    miga: File.join(MiGA::MiGA.root_path, 'bin/miga').shellescape
  }
  runopts_for(:cmd, what).miga_variables(var_hsh)
end

#l_say(level, *msg) ⇒ Object

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



121
122
123
 
# File 'lib/miga/daemon.rb', line 121

def l_say(level, *msg)
  say(*msg) if verbosity >= level
end

#launch_job(job, hostk = nil) ⇒ Object

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



332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
 
# File 'lib/miga/daemon.rb', line 332

def launch_job(job, hostk = nil)
  # Execute job
  job[:cmd] = job_cmd(job)
  MiGA::MiGA.DEBUG "CMD: #{job[:cmd]}"
  case runopts(:type)
  when 'ssh'
    # Remote job
    job[:hostk] = hostk
    job[:cmd] = job[:cmd].miga_variables(host: nodelist[hostk])
    job[:pid] = spawn job[:cmd]
    MiGA::MiGA.DEBUG "Detaching PID: #{job[:pid]}"
    Process.detach(job[:pid]) unless [nil, '', 0].include?(job[:pid])
  when 'bash'
    # Local job
    job[:pid] = spawn job[:cmd]
    MiGA::MiGA.DEBUG "Detaching PID: #{job[:pid]}"
    Process.detach(job[:pid]) unless [nil, '', 0].include?(job[:pid])
  else
    # Schedule cluster job (qsub, msub, slurm)
    job[:pid] = MiGA::MiGA.run_cmd(job[:cmd], return: :output).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 ⇒ Object

Load the status of a previous instance.



152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
 
# File 'lib/miga/daemon.rb', line 152

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 ⇒ Object 



125
 
# File 'lib/miga/daemon.rb', line 125

alias miga_say say

#next_host ⇒ Object

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



311
312
313
314
315
316
317
 
# File 'lib/miga/daemon.rb', line 311

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! ⇒ Object

Remove dead jobs.



321
322
323
324
325
326
327
328
 
# File 'lib/miga/daemon.rb', line 321

def purge!
  say 'Probing running jobs'
  @jobs_running.select! do |job|
    MiGA::MiGA.run_cmd(
      runopts(:alive).miga_variables(pid: job[:pid]), return: :output
    ).chomp.to_i == 1
  end
end

#queue_job(job, ds = nil) ⇒ Object

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!



220
221
222
223
224
225
226
227
228
 
# File 'lib/miga/daemon.rb', line 220

def queue_job(job, ds = nil)
  return nil unless get_job(job, ds).nil?

  ds_name = (ds.nil? ? 'miga-project' : ds.name)
  task_name = "#{project.metadata[:name][0..9]}:#{job}:#{ds_name}"
  to_run = { ds: ds, ds_name: ds_name, job: job, task_name: task_name }
  say 'Queueing %s:%s' % [to_run[:ds_name], to_run[:job]]
  @jobs_to_run << to_run
end

#queue_maintenance(force = false) ⇒ Object

Queue maintenance tasks as an analysis job



106
107
108
109
110
111
 
# File 'lib/miga/daemon.rb', line 106

def queue_maintenance(force = false)
  return if bypass_maintenance? || (!force && shutdown_when_done?)

  say 'Queueing maintenance tasks'
  queue_job(:maintenance)
end

#reload_project ⇒ Object

Reload the project’s metadata



135
136
137
138
 
# File 'lib/miga/daemon.rb', line 135

def reload_project
  l_say(2, 'Reloading project')
  project.load
end

#save_status ⇒ Object

Report status in a JSON file.



142
143
144
145
146
147
148
 
# File 'lib/miga/daemon.rb', line 142

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) ⇒ Object

Same as l_say with level = 1



129
130
131
 
# File 'lib/miga/daemon.rb', line 129

def say(*msg)
  super(logfh, *msg) if verbosity >= 1
end

#update_format_0 ⇒ Object

Update from daemon JSON format 0 to the latest version



368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
 
# File 'lib/miga/daemon.rb', line 368

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