Class: Ripe::WorkerController

Inherits:
Object
  • Object
show all
Defined in:
lib/ripe/worker_controller.rb,
lib/ripe/worker_controller/syncer.rb,
lib/ripe/worker_controller/preparer.rb

Overview

This class controls workers as well as their relationship with regards to the compute cluster: worker preparation, submission, cancellation as well as sync.

Defined Under Namespace

Classes: Preparer, Syncer

Instance Method Summary collapse

Instance Method Details

#cancel(workers) ⇒ Array<DB::Worker>

Cancel worker jobs in the compute cluster system.

Parameters:

Returns:

  • (Array<DB::Worker>) —

    the list of workers given in arguments, with modified states



75
76
77
78
79
80
81
82
83
84
 
# File 'lib/ripe/worker_controller.rb', line 75

def cancel(workers)
  distribute workers do |worker|
    if ['queueing', 'idle', 'blocked', 'active'].include? worker.status
      `canceljob #{worker.moab_id}`
      worker.update(status: :cancelled)
    else
      puts "Worker #{worker.id} could not be cancelled: not started"
    end
  end
end

#distribute(workers, &block) ⇒ Array<DB::Worker>

Apply a block to a list of workers.

Parameters:

Returns:

  • (Array<DB::Worker>) —

    the list of workers given in arguments, with modified states



35
36
37
38
 
# File 'lib/ripe/worker_controller.rb', line 35

def distribute(workers, &block)
  workers = [workers] if workers.is_a? DB::Worker
  workers.map { |w| block.call(w) }
end

#edit(*args) ⇒ void

This method returns an undefined value.

Launch the an interactive text editor from the console.



112
113
114
 
# File 'lib/ripe/worker_controller.rb', line 112

def edit(*args)
  system("$EDITOR #{args.join(' ')}")
end

#list(n = 20) ⇒ Array<DB::Worker>

List the n most recent workers.

Parameters:

  • n (Integer) (defaults to: 20) —

    the number of most recent workers

Returns:

  • (Array<DB::Worker>) —

    the list of n most recent workers



103
104
105
 
# File 'lib/ripe/worker_controller.rb', line 103

def list(n = 20)
  DB::Worker.last(n)
end

#local(workers) ⇒ Array<DB::Worker>

Run worker job code into bash locally.

Parameters:

Returns:

  • (Array<DB::Worker>) —

    the list of workers given in arguments, with modified states



46
47
48
49
50
 
# File 'lib/ripe/worker_controller.rb', line 46

def local(workers)
  distribute workers do |worker|
    `bash #{worker.sh}`
  end
end

#prepare(workflow, samples, params = {}) ⇒ Array<Worker>

Prepare workers by applying the workflow callback and its parameters to each sample.

Parameters:

  • workflow (String) —

    the name of a workflow to apply on the sample list

  • samples (Array<String>) —

    list of samples to apply the callback to

  • params (Hash<Symbol, String>) (defaults to: {}) —

    a list of worker-wide parameters

Returns:

  • (Array<Worker>) —

    workers prepared in current batch

See Also:



23
24
25
 
# File 'lib/ripe/worker_controller.rb', line 23

def prepare(workflow, samples, params = {})
  Preparer.new(workflow, samples, params).workers
end

#start(workers) ⇒ Array<DB::Worker>

Submit worker jobs to the compute cluster system.

Parameters:

Returns:

  • (Array<DB::Worker>) —

    the list of workers given in arguments, with modified states



58
59
60
61
62
63
64
65
66
67
 
# File 'lib/ripe/worker_controller.rb', line 58

def start(workers)
  distribute workers do |worker|
    if worker.status == 'prepared'
      worker.update(status: :queueing,
                    moab_id: `qsub '#{worker.sh}'`.strip.split(/\./).first)
    else
      puts "Worker #{worker.id} could not be started: not prepared"
    end
  end
end

#sync ⇒ Array<DB::Worker>

Synchronize the status of jobs with the internal list of workers.

Returns:

  • (Array<DB::Worker>) —

    the list of updated workers

See Also:



93
94
95
 
# File 'lib/ripe/worker_controller.rb', line 93

def sync
  Syncer.new.workers
end