Class: RQ::JobRunner

Inherits:
Object
  • Object
show all
Includes:
DRbUndumped
Defined in:
lib/rq/jobrunner.rb

Overview

the JobRunner class is responsible for pre-forking a process/shell in which to run a job. this class is utilized by the JobRunnerDaemon so processes can be forked via a drb proxy to avoid actual forking during an sqlite transaction - which has undefined behaviour

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(q, job) ⇒ JobRunner 



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
 
# File 'lib/rq/jobrunner.rb', line 33

def initialize q, job
#--{{{
  @q = q
  @job = job
  @jid = job['jid']
  @command = job['command']
  @shell = job['shell'] || 'bash'
  @sh_like = File::basename(@shell) == 'bash' || File::basename(@shell) == 'sh' 
  @r,@w = IO::pipe

  @env = {}
  @env["PATH"] = [@q.bin, ENV["PATH"]].join(":")
  @job.fields.each do |field|
    key = "RQ_#{ field }".upcase.gsub(%r/\s+/,'_')
    val = @job[field]
    val = File.expand_path(File.join(@q.path,val)) if %w( stdin stdout stderr data).include?(field.to_s)
    @env[key] = "#{ val }"
  end
  @env['RQ'] = File.expand_path @q.path
  @env['RQ_JOB'] = @job.to_hash.to_yaml 

  @stdin = @job['stdin']
  @stdout = @job['stdout']
  @stderr = @job['stderr']
  @data = @job['data']

  @stdin &&= File::join(@q.path, @stdin) # assume path relative to queue 
  @stdout &&= File::join(@q.path, @stdout) # assume path relative to queue 
  @stderr &&= File::join(@q.path, @stderr) # assume path relative to queue
  @data &&= File::join(@q.path, @data) # assume path relative to queue 

  [@stdin, @stdout, @stderr].each do |path|
    FileUtils::mkdir_p(FileUtils::dirname(path)) rescue nil
    FileUtils::touch(path) unless File.exist?(path)
  end

  @cid = 
    Util::fork do
      @env.each{|k,v| ENV[k] = v}
      ENV['RQ_PID'] = "#{ $$ }"
      @w.close
      STDIN.reopen @r
      argv =
        if @sh_like 
          [ [@shell, "__rq_job__#{ @jid }__#{ File::basename(@shell) }__"], '--login' ]
        else
          [ [@shell, "__rq_job__#{ @jid }__#{ File::basename(@shell) }__"], '-l' ]
        end
      exec *argv
    end
  @r.close
#--}}}
end

Instance Attribute Details

#cid ⇒ Object (readonly) Also known as: pid

Returns the value of attribute cid.



25
26
27
 
# File 'lib/rq/jobrunner.rb', line 25

def cid
  @cid
end

#command ⇒ Object (readonly)

Returns the value of attribute command.



27
28
29
 
# File 'lib/rq/jobrunner.rb', line 27

def command
  @command
end

#data ⇒ Object (readonly)

Returns the value of attribute data.



31
32
33
 
# File 'lib/rq/jobrunner.rb', line 31

def data
  @data
end

#jid ⇒ Object (readonly)

Returns the value of attribute jid.



24
25
26
 
# File 'lib/rq/jobrunner.rb', line 24

def jid
  @jid
end

#job ⇒ Object (readonly)

Returns the value of attribute job.



23
24
25
 
# File 'lib/rq/jobrunner.rb', line 23

def job
  @job
end

#q ⇒ Object (readonly)

Returns the value of attribute q.



22
23
24
 
# File 'lib/rq/jobrunner.rb', line 22

def q
  @q
end

#shell ⇒ Object (readonly)

Returns the value of attribute shell.



26
27
28
 
# File 'lib/rq/jobrunner.rb', line 26

def shell
  @shell
end

#stderr ⇒ Object (readonly)

Returns the value of attribute stderr.



30
31
32
 
# File 'lib/rq/jobrunner.rb', line 30

def stderr
  @stderr
end

#stdin ⇒ Object (readonly)

Returns the value of attribute stdin.



28
29
30
 
# File 'lib/rq/jobrunner.rb', line 28

def stdin
  @stdin
end

#stdout ⇒ Object (readonly)

Returns the value of attribute stdout.



29
30
31
 
# File 'lib/rq/jobrunner.rb', line 29

def stdout
  @stdout
end

Instance Method Details

#run ⇒ Object

–}}}



86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
 
# File 'lib/rq/jobrunner.rb', line 86

def run
#--{{{
  command = @command.gsub %r/#.*/o, '' # kill comments
  path = @q.bin

  command =
    if @sh_like 
      sin = "0<#{ @stdin }" if @stdin
      sout = "1>#{ @stdout }" if @stdout
      serr = "2>#{ @stderr }" if @stderr
      "( PATH=#{ path }:$PATH #{ command } ;) #{ sin } #{ sout } #{ serr }"
    else
      sin = "<#{ @stdin }" if @stdin
      sout = ">#{ @stdout }" if @stdout
      serr = ">&#{ @stderr }" if @stderr
      "( ( #{ command } ;) #{ sin } #{ sout } ) #{ serr }"
    end

  @w.puts command
  @w.close
#--}}}
end