Class: Pwrake::Communicator
- Inherits:
-
Object
- Object
- Pwrake::Communicator
- Defined in:
- lib/pwrake/branch/communicator.rb
Defined Under Namespace
Classes: ConnectError
Instance Attribute Summary collapse
-
#channel ⇒ Object
readonly
Returns the value of attribute channel.
-
#handler ⇒ Object
readonly
Returns the value of attribute handler.
-
#host ⇒ Object
readonly
Returns the value of attribute host.
-
#id ⇒ Object
readonly
Returns the value of attribute id.
-
#ncore ⇒ Object
readonly
Returns the value of attribute ncore.
-
#reader ⇒ Object
readonly
Returns the value of attribute reader.
-
#shells ⇒ Object
readonly
Returns the value of attribute shells.
-
#writer ⇒ Object
readonly
Returns the value of attribute writer.
Instance Method Summary collapse
- #common_line(s) ⇒ Object
- #connect(worker_code) ⇒ Object
- #dropout(exc = nil) ⇒ Object
- #finish ⇒ Object
- #finish_shells ⇒ Object
-
#initialize(set, id, host, ncore, selector, option) ⇒ Communicator
constructor
A new instance of Communicator.
- #inspect ⇒ Object
- #new_channel ⇒ Object
Constructor Details
#initialize(set, id, host, ncore, selector, option) ⇒ Communicator
Returns a new instance of Communicator.
42 43 44 45 46 47 48 49 50 |
# File 'lib/pwrake/branch/communicator.rb', line 42 def initialize(set,id,host,ncore,selector,option) @set = set @id = id @host = host @ncore = @ncore_given = ncore @selector = selector @option = option @shells = {} end |
Instance Attribute Details
#channel ⇒ Object (readonly)
Returns the value of attribute channel.
38 39 40 |
# File 'lib/pwrake/branch/communicator.rb', line 38 def channel @channel end |
#handler ⇒ Object (readonly)
Returns the value of attribute handler.
39 40 41 |
# File 'lib/pwrake/branch/communicator.rb', line 39 def handler @handler end |
#host ⇒ Object (readonly)
Returns the value of attribute host.
38 39 40 |
# File 'lib/pwrake/branch/communicator.rb', line 38 def host @host end |
#id ⇒ Object (readonly)
Returns the value of attribute id.
38 39 40 |
# File 'lib/pwrake/branch/communicator.rb', line 38 def id @id end |
#ncore ⇒ Object (readonly)
Returns the value of attribute ncore.
38 39 40 |
# File 'lib/pwrake/branch/communicator.rb', line 38 def ncore @ncore end |
#reader ⇒ Object (readonly)
Returns the value of attribute reader.
39 40 41 |
# File 'lib/pwrake/branch/communicator.rb', line 39 def reader @reader end |
#shells ⇒ Object (readonly)
Returns the value of attribute shells.
40 41 42 |
# File 'lib/pwrake/branch/communicator.rb', line 40 def shells @shells end |
#writer ⇒ Object (readonly)
Returns the value of attribute writer.
39 40 41 |
# File 'lib/pwrake/branch/communicator.rb', line 39 def writer @writer end |
Instance Method Details
#common_line(s) ⇒ Object
106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 |
# File 'lib/pwrake/branch/communicator.rb', line 106 def common_line(s) x = "Communicator#common_line(id=#{@id},host=#{@host})" case s when /^heartbeat$/ Log.debug "#{x}: #{s.inspect}" @selector.heartbeat(@reader.io) when /^exited$/ Log.debug "#{x}: #{s.inspect}" return false when /^log:(.*)$/ Log.info "#{x}: log>#{$1}" when String Log.warn "#{x}: out>#{s.inspect}" when Exception Log.warn "#{x}: err>#{s.class}: #{s.}" dropout(s) return false else raise ConnectError, "#{x}: invalid for read: #{s.inspect}" end true end |
#connect(worker_code) ⇒ Object
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 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 |
# File 'lib/pwrake/branch/communicator.rb', line 61 def connect(worker_code) rb_cmd = "ruby -e 'eval ARGF.read(#{worker_code.size})'" if ['localhost','localhost.localdomain','127.0.0.1'].include? @host #if /^localhost/ =~ @host cmd = rb_cmd else cmd = "ssh -x -T #{@option[:ssh_option]} #{@host} \"#{rb_cmd}\"" end # @ior,w0 = IO.pipe @ioe,w1 = IO.pipe r2,@iow = IO.pipe @pid = Kernel.spawn(cmd,:pgroup=>true,:out=>w0,:err=>w1,:in=>r2) w0.close w1.close r2.close sel = @set.selector @reader = NBIO::MultiReader.new(sel,@ior) @rd_err = NBIO::Reader.new(sel,@ioe) @writer = NBIO::Writer.new(sel,@iow) @handler = NBIO::Handler.new(@reader,@writer,@host) # @writer.write(worker_code) @writer.write(Marshal.dump(@ncore)) @writer.write(Marshal.dump(@option)) # read ncore while s = @reader.get_line if /^ncore:(.*)$/ =~ s a = $1 Log.debug "ncore=#{a} @#{@host}" if /^(\d+)$/ =~ a @ncore = $1.to_i return false else raise ConnectError, "invalid for ncore: #{a.inspect}" end else return false if !common_line(s) end end raise ConnectError, "fail to connect #{cmd.inspect}" rescue => e dropout(e) end |
#dropout(exc = nil) ⇒ Object
133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 |
# File 'lib/pwrake/branch/communicator.rb', line 133 def dropout(exc=nil) # Error output err_out = [] begin finish_shells @handler.exit while s = @rd_err.get_line err_out << s end rescue => e m = Log.bt(e) #$stderr.puts m Log.error(m) end # Error output if !err_out.empty? $stderr.puts err_out.join("\n") Log.error((["process error output:"]+err_out).join("\n ")) end # Exception if exc m = Log.bt(exc) #$stderr.puts m Log.error m end ensure @set.delete(self) end |
#finish ⇒ Object
162 163 164 165 166 167 168 169 170 |
# File 'lib/pwrake/branch/communicator.rb', line 162 def finish @iow.close while s=@ior.gets puts "out=#{s.chomp}" end while s=@ioe.gets puts "err=#{s.chomp}" end end |
#finish_shells ⇒ Object
129 130 131 |
# File 'lib/pwrake/branch/communicator.rb', line 129 def finish_shells @shells.keys.each{|sh| sh.finish_task_q} end |
#inspect ⇒ Object
52 53 54 |
# File 'lib/pwrake/branch/communicator.rb', line 52 def inspect "#<#{self.class} @id=#{@id},@host=#{@host},@ncore=#{@ncore}>" end |
#new_channel ⇒ Object
56 57 58 59 |
# File 'lib/pwrake/branch/communicator.rb', line 56 def new_channel i,q = @reader.new_queue CommChannel.new(@host,i,q,@writer,[@ior,@iow,@ioe]) end |