Class: Array

Inherits:
Object
  • Object
show all
Defined in:
lib/nysol/mutils.rb,
lib/nysol/mparallel.rb

Overview

/* ////////// LICENSE INFO ////////////////////

  • Copyright © 2013 by NYSOL CORPORATION

*

  • Unless you have received this program directly from NYSOL pursuant

  • to the terms of a commercial license agreement with NYSOL, then

  • this program is licensed to you under the terms of the GNU Affero General

  • Public License (AGPL) as published by the Free Software Foundation,

  • either version 3 of the License, or (at your option) any later version.

*

  • This program is distributed in the hope that it will be useful, but

  • WITHOUT ANY EXPRESS OR IMPLIED WARRANTY, INCLUDING THOSE OF

  • NON-INFRINGEMENT, MERCHANTABILITY OR FITNESS FOR A PARTICULAR PURPOSE.

*

////////// LICENSE INFO ////////////////////*/

MCMD ruby拡張ライブラリ

ユーティリティツール Array.

maddPrefix
maddSuffix
mslidepair
MCMD

errorLog(msg=“”) warningLog(msg=“”) endLog(msg=“”) msgLog(msg=“”,time=true,header=true) mkDir(path,rm=false) chkCmdExe(cmd,type,arg1=nil) chkRexe(libs) mcat(fName,oFile)

Instance Method Summary collapse

Instance Method Details

#maddPrefix(strs) ⇒ Object 



38
39
40
41
42
43
44
45
46
 
# File 'lib/nysol/mutils.rb', line 38

def maddPrefix(strs)
	strs = [strs] if strs.class != Array
	rtn =[]
	params=self.dup
	params.each{|v|
		strs.each{|str| rtn << "#{str}#{v}" }
	}
	return rtn	
end

#maddSuffix(strs) ⇒ Object 



48
49
50
51
52
53
54
55
56
 
# File 'lib/nysol/mutils.rb', line 48

def maddSuffix(strs)
	strs = [strs] if strs.class != Array
	rtn =[]
	params=self.dup
	params.each{|v|
		strs.each{|str| rtn << "#{v}#{str}" }
	}
	return rtn	
end

#mcollect(fn, out, nFlg = false, pclist = "/etc/pclist") ⇒ Object

outの最後が“/”でない場合は１ファイルと見なす 配列にはデータがあるip,pathとが含まれている

[ip1,path1],[ip2,path2

…,[ipn,pathn]]

データ量によっては端末ごとにcat＆圧縮した方が高速化できるかも



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
 
# File 'lib/nysol/mparallel.rb', line 34

def mcollect(fn,out,nFlg=false,pclist="/etc/pclist")
	fn = [fn] if fn.class != Array 
	out = [out] if out.class != Array 
	raise "no match size in-out " if fn.size != out.size 

	temp=MCMD::Mtemp.new
	params=self.dup

	# PC_LIST読み込み
	pcM = MCMD::MpcManager.new(pclist)

	# ip毎に配列集約 
	dlist ={}
	params.each{|ip,path|
		raise "unknown ip " unless pcM.has_IP?(ip) 
		dlist[ip] = [] unless dlist.has_key?(ip)
		dlist[ip] << path
	}
	
	(0...fn.size).each{|i|
		type = out[i] =~ /\/\s*$/ ?  0 : 2
		type += 1 if dlist.size != 0

		#パラメータセット
		nfr  = "/#{fn[i]}"
		nrev = out[i]
		lcin = fn[i]
		to   = out[i] 
		if type == 3 then
			nrev = "#{temp.file}/" 
			lcin = "#{nrev}*"
		end

		# collet 
		dlist.each{|ip,path|
				MCMD::NetTools.recv(ip,pcM.getUIDbyIP(ip),path.maddSuffix(nfr),nrev,pcM.getPWDbyIP(ip))
		}
		case type
		when 0 #出力ディレクトリ(local)
			MCMD::mkDir(to,true)
			system "cp #{lcin} #{to}"
		when 2,3 #出力ファイル	(local)
			if nFlg then
				system "mcat i=#{lcin} o=#{to} -nfn"
			else
				system "mcat i=#{lcin} o=#{to}"
			end			
		end
	}
end

#meach(mpCount = 4, msgcnt = 100, tF = false, &block) ⇒ Object

並列処理each



86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
 
# File 'lib/nysol/mparallel.rb', line 86

def meach(mpCount=4,msgcnt=100,tF=false,&block)
	tim = tF ? 5 : -1
	params=self.dup
	ttl    = params.size
	nowcnt = 0 
	mpm = MCMD::MparallelManager.new(mpCount,tim)
	mpm.runStateCheker

	while params.size>0
		param=params.delete_at(0) 
		nowlane = mpm.getLane
		# blockの実行
		pid=fork {
			case block.arity
			when 1
				yield(param)
			when 2
				yield(param,nowcnt)				
			when 3
				yield(param,nowcnt,nowlane)				
			else
				raise "unmatch args size."
			end
		}
		nowcnt+=1
		MCMD::msgLog("meach start #{param} (#{nowcnt}/#{ttl})") if msgcnt!=0 and nowcnt%msgcnt == 0
		mpm.addPid(pid,nowlane) 
	end
	mpm.waitall
	return []
end

#meachc(mpInfo = [4,4], msgcnt = 100, pclist = "/etc/pclist", bg = true, &block) ⇒ Object 



118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
 
# File 'lib/nysol/mparallel.rb', line 118

def meachc(mpInfo=[4,4],msgcnt=100,pclist="/etc/pclist",bg=true,&block)
	# -----------------------------------------------------
	# pc情報設定 & 並列数決定
	# -----------------------------------------------------
	pcM = MCMD::MpcManager.new(pclist)
	mpPC = mpLim = 4
	if mpInfo.class == Array then
		mpPC  = mpInfo[0] if mpInfo[0] 
		mpLim = mpInfo[1] if mpInfo[1] 
	else
		mpPC  = mpInfo
	end
	mpPC  = pcM.pcCount  if mpPC > pcM.pcCount
	mpMax = mpPC * mpLim

	# -----------------------------------------------------
	# 呼び出し元ファイル ＆ ローカル変数取得
	# -----------------------------------------------------
	fn ,lin = block.source_location 
	value = block.binding.eval("local_variables") 
	valueV=[]
	value.each{|v|
		clsinfo = block.binding.eval("#{v}.class")
		case clsinfo.to_s
		when "String"
			valinfo = block.binding.eval("#{v}")
			valueV << "#{v} = \"#{valinfo}\""
		when "Fixnum","Bignum","Float"
			valinfo = block.binding.eval("#{v}")
			valueV << "#{v} = #{valinfo}"
		when "Array"
			valinfo = block.binding.eval("#{v}")
			valueV << "#{v} = #{valinfo}"
		end 
	}

	# -----------------------------------------------------
	# 並列用プログラム生成
	# -----------------------------------------------------
	tempC=MCMD::Mtemp.new
	runspf=tempC.file
	MCMD::MrubyParse.new(fn,lin,"meachc",valueV).output(runspf)		

	# -----------------------------------------------------
	# 作業ディレクトリ決定
	# -----------------------------------------------------
	local_WkD = tempC.file
	net_WkD   = tempC.file
	MCMD::mkDir(local_WkD)

	# -----------------------------------------------------
	# 並列処理
	# -----------------------------------------------------
	params=self.dup
	mpm = MCMD::MparallelManagerByFile.new(mpMax,local_WkD)
	ttl    = params.size
	nowcnt = 0 
	rtnlist =[]

	while params.size>0

		param=params.delete_at(0) 
		nowlane = mpm.getLane

		# blockの実行
		pid=fork {

			stNo = nowlane%mpPC
			param = [param] if param.class != Array
			# -----------------------------------------------------
			# 起動スクリプト生成
			# -----------------------------------------------------
			runinit_Scp = "#{local_WkD}/#{File.basename(runspf)}_#{nowlane}.sh"
			net_nowwkD  = "#{net_WkD}/#{nowcnt}" 
			File.open(runinit_Scp,"w"){|fpw|
				fpw.puts "cd #{net_nowwkD}"
				case block.arity
				when 1
					fpw.puts "ruby #{File.basename(runspf)} #{param.join(",")}" 
				when 2
					fpw.puts "ruby #{File.basename(runspf)} #{param.join(",")} #{nowcnt}" 
				when 3
					fpw.puts "ruby #{File.basename(runspf)} #{param.join(",")} #{nowcnt} #{nowlane}"
				else
					raise "unmatch args size."
				end
				fpw.puts "echo pgmend #{nowcnt} >> #{net_nowwkD}/endlog"
				fpw.puts "ls -lR >> #{net_nowwkD}/endlog"
				fpw.puts "msend.rb i=#{net_nowwkD}/endlog uid=#{pcM.getUID(stNo)} o=#{pcM.localIP}:#{local_WkD}/#{nowlane}.log"
			}			
		
			# -----------------------------------------------------
			# スクリプトファイル&データ転送
			# -----------------------------------------------------
			MCMD::msgLog("meachc snddata #{fn}:#{lin} (#{nowcnt}/#{ttl})") if msgcnt!=0 and nowcnt%msgcnt == 0
			distSrc  = "#{net_nowwkD}/#{File.basename(runspf)}"
			distSrcI = "#{net_nowwkD}/#{File.basename(runspf)}.sh"
			MCMD::NetTools.send(pcM.getIP(stNo),pcM.getUID(stNo),runinit_Scp,distSrcI,pcM.getPWD(stNo))
			MCMD::NetTools.send(pcM.getIP(stNo),pcM.getUID(stNo),runspf     ,distSrc ,pcM.getPWD(stNo))
			param.each{|dt|
				MCMD::NetTools.send(pcM.getIP(stNo),pcM.getUID(stNo),dt,"#{net_nowwkD}/#{dt}",pcM.getPWD(stNo))
			}

			# -----------------------------------------------------
			# プログラム実行
			# -----------------------------------------------------
			MCMD::msgLog("meachc pgmstart #{fn}:#{lin} (#{nowcnt}/#{ttl})") if msgcnt!=0 and nowcnt%msgcnt == 0
			MCMD::NetTools.cmdRun(pcM.getIP(stNo),pcM.getUID(stNo),"nohup bash #{distSrcI} 1> #{net_nowwkD}/RunLog 2>&1 &")
		}
		rtnlist << [ pcM.getIP(nowlane%mpPC),"#{net_WkD}/#{nowcnt}" ]
		mpm.addNo(nowcnt,nowlane) 
		nowcnt+=1
		Process.detach(pid)

	end
	mpm.waitall		
	return rtnlist
end

#meachcN(mpInfo = [4,4], msgcnt = 100, pclist = "/etc/pclist", bg = true, &block) ⇒ Object 



238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
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
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
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
 
# File 'lib/nysol/mparallel.rb', line 238

def meachcN(mpInfo=[4,4],msgcnt=100,pclist="/etc/pclist",bg=true,&block)
	# -----------------------------------------------------
	# pc情報設定 & 並列数決定
	# -----------------------------------------------------
	pcM = MCMD::MpcManager.new(pclist)
	mpPC = mpLim = 4
	if mpInfo.class == Array then
		mpPC  = mpInfo[0] if mpInfo[0] 
		mpLim = mpInfo[1] if mpInfo[1] 
	else
		mpPC  = mpInfo
	end
	mpPC  = pcM.pcCount  if mpPC > pcM.pcCount
	mpMax = mpPC * mpLim

	# -----------------------------------------------------
	# 呼び出し元ファイル ＆ ローカル変数取得
	# -----------------------------------------------------
	fn ,lin = block.source_location 
	value = block.binding.eval("local_variables") 
	valueV=[]
	value.each{|v|
		clsinfo = block.binding.eval("#{v}.class")
		case clsinfo.to_s
		when "String"
			valinfo = block.binding.eval("#{v}")
			valueV << "#{v} = \"#{valinfo}\""
		when "Fixnum","Bignum","Float"
			valinfo = block.binding.eval("#{v}")
			valueV << "#{v} = #{valinfo}"
		when "Array"
			valinfo = block.binding.eval("#{v}")
			valueV << "#{v} = #{valinfo}"
		end 
	}

	# -----------------------------------------------------
	# 並列用プログラム生成
	# -----------------------------------------------------
	tempC=MCMD::Mtemp.new
	runspf=tempC.file
	MCMD::MrubyParse.new(fn,lin,"meachcN",valueV).output(runspf)		

	# -----------------------------------------------------
	# 作業ディレクトリ決定
	# -----------------------------------------------------
	local_WkD = tempC.file
	net_WkD   = tempC.file
	net_HMD   = ENV['PWD'].sub(/#{ENV['HOME']}\//,"")
	MCMD::mkDir(local_WkD)

	# -----------------------------------------------------
	# 並列処理
	# -----------------------------------------------------
	params=self.dup
	mpm = MCMD::MparallelManagerByFile.new(mpMax,local_WkD)
	ttl    = params.size
	nowcnt = 0 
	rtnlist =[]

	while params.size>0

		param=params.delete_at(0) 
		nowlane = mpm.getLane

		# blockの実行
		pid=fork {

			stNo = nowlane%mpPC
			param = [param] if param.class != Array
			# -----------------------------------------------------
			# 起動スクリプト生成
			# -----------------------------------------------------
			runinit_Scp = "#{local_WkD}/#{File.basename(runspf)}_#{nowlane}.sh"
			net_nowwkD  = "#{net_WkD}/#{nowcnt}" 
			distSrc  = "#{net_nowwkD}/#{File.basename(runspf)}"
			distSrcI = "#{net_nowwkD}/#{File.basename(runspf)}.sh"


			File.open(runinit_Scp,"w"){|fpw|
				fpw.puts "mkdir -p #{net_HMD}"
				fpw.puts "cd #{net_HMD}"
				case block.arity
				when 1
					fpw.puts "ruby #{distSrc} #{param.join(",")}" 
				when 2
					fpw.puts "ruby #{distSrc} #{param.join(",")} #{nowcnt}" 
				when 3
					fpw.puts "ruby #{distSrc} #{param.join(",")} #{nowcnt} #{nowlane}"
				else
					raise "unmatch args size."
				end
				fpw.puts "echo pgmend #{nowcnt} >> #{net_nowwkD}/endlog"
				fpw.puts "ls -lR >> #{net_nowwkD}/endlog"
				fpw.puts "msend.rb i=#{net_nowwkD}/endlog uid=#{pcM.getUID(stNo)} o=#{pcM.localIP}:#{local_WkD}/#{nowlane}.log"
			}			
		
			# -----------------------------------------------------
			# スクリプトファイル&データ転送
			# -----------------------------------------------------
			MCMD::msgLog("meachcN snddata #{fn}:#{lin} (#{nowcnt}/#{ttl})") if msgcnt!=0 and nowcnt%msgcnt == 0
			MCMD::NetTools.send(pcM.getIP(stNo),pcM.getUID(stNo),runinit_Scp,distSrcI,pcM.getPWD(stNo))
			MCMD::NetTools.send(pcM.getIP(stNo),pcM.getUID(stNo),runspf     ,distSrc ,pcM.getPWD(stNo))

			# -----------------------------------------------------
			# プログラム実行
			# -----------------------------------------------------
			MCMD::msgLog("meachcN pgmstart #{fn}:#{lin} (#{nowcnt}/#{ttl})") if msgcnt!=0 and nowcnt%msgcnt == 0
			MCMD::NetTools.cmdRun(pcM.getIP(stNo),pcM.getUID(stNo),"nohup bash #{distSrcI} 1> #{net_nowwkD}/RunLog 2>&1 &")
		}
		rtnlist << [ pcM.getIP(nowlane%mpPC),"#{net_WkD}/#{nowcnt}" ]
		mpm.addNo(nowcnt,nowlane) 
		nowcnt+=1
		Process.detach(pid)

	end
	mpm.waitall		
	return rtnlist
end

#mslidepair ⇒ Object 



58
59
60
61
62
63
64
65
 
# File 'lib/nysol/mutils.rb', line 58

def mslidepair
	rtn =[]
	params=self.dup
	(1...params.size).each{|i|
		rtn << [params[i-1],params[i]]	
	}
	return rtn	
end