Class: Fairy::PSort::PPreSort

Inherits:
Fairy::PBasicGroupBy show all
Defined in:
lib/fairy/node/p-sort.rb

Constant Summary collapse

ST_ALL_IMPORTED = 
:ST_ALL_IMPORTED
ST_WAIT_EXPORT_FINISH = 
:ST_WAIT_EXPORT_FINISH
ST_EXPORT_FINISH = 
:ST_EXPORT_FINISH

Constants inherited from Fairy::PIOFilter

Fairy::PIOFilter::ST_WAIT_IMPORT

Constants inherited from Fairy::PFilter

Fairy::PFilter::END_OF_STREAM, Fairy::PFilter::ST_ACTIVATE, Fairy::PFilter::ST_FINISH, Fairy::PFilter::ST_INIT

Instance Attribute Summary

Attributes inherited from Fairy::PFilter

#IGNORE_EXCEPTION, #id, #log_id, #ntask

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Fairy::PBasicGroupBy

#terminate, #wait_export_finish

Methods inherited from Fairy::PIOFilter

#input=

Methods inherited from Fairy::PFilter

#abort_running, #basic_start, #break_running, #each, #global_break, #global_break_from_other, #handle_exception, #key, #key=, #next, #no, #no=, #notice_status, #processor, #start, #start_watch_status, #status=, #terminate, #terminate_proc

Constructor Details

#initialize(id, ntask, bjob, opts, block_source) ⇒ PPreSort

Returns a new instance of PPreSort.



18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/fairy/node/p-sort.rb', line 18

def initialize(id, ntask, bjob, opts, block_source)
	super
	@block_source = block_source

	@exports = []
	def @exports.each_pair(&block)
	  each_with_index do |item, idx| 
	    block.call(idx, item) if item
	  end
	end
	@exports_queue = Queue.new
	
	@counter = []

	@pvs = nil
	if @opts[:pvs]
	  @pvs = @opts[:pvs].dc_deep_copy
	end

	#start_watch_exports
end

Class Method Details

.each_pair(&block) ⇒ Object 



23
24
25
26
27
 
# File 'lib/fairy/node/p-sort.rb', line 23

def @exports.each_pair(&block)
  each_with_index do |item, idx| 
    block.call(idx, item) if item
  end
end

Instance Method Details

#add_export(key, export) ⇒ Object 



40
41
42
43
44
45
 
# File 'lib/fairy/node/p-sort.rb', line 40

def add_export(key, export)
	@exports[key] = export
	#      @exports_queue.push [key, export]
	# [BUG#171]同期処理でないとまずい.
	@bjob.add_exports(key, export, self)
end

#hash_key(e) ⇒ Object 



146
147
148
 
# File 'lib/fairy/node/p-sort.rb', line 146

def hash_key(e)
	@key_proc.yield(e)
end

#hashing(e) ⇒ Object 



128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
 
# File 'lib/fairy/node/p-sort.rb', line 128

def hashing(e)
	if Import::CTLTOKEN_NULLVALUE === (key = @key_proc.call(e))
	  return
	end
	
	unless idx = @pvs.find_index{|pv| key < @key_proc.call(pv)}
	  idx = @pvs.size
	end

#Log::debug(self, "#{@pvs.inspect}")
#Log::debug(self, "#{idx}")
	

	export = @exports[idx]
	export.push e
	@counter[idx] += 1
end

#init_exports ⇒ Object 



114
115
116
117
118
119
120
121
122
123
124
125
126
 
# File 'lib/fairy/node/p-sort.rb', line 114

def init_exports
	policy = @opts[:postqueuing_policy]
	(@pvs.size+1).times do |idx|
	  export = Export.new(policy)
	  @exports[idx] = export
	  
	  export.njob_id = @id
	  export.no = idx
	  export.add_key(idx)
	  add_export(idx, export)
	  @counter[idx] = 0
	end
end

#start_export ⇒ Object 



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
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
 
# File 'lib/fairy/node/p-sort.rb', line 47

def start_export
	Log::debug(self, "START_EXPORT")

	start do
	  sample_line_no = @opts[:sampling_max]
	  sample_line_no ||= CONF.SORT_SAMPLING_MAX

	  hash_opt = @opts[:cmp_optimize]
	  hash_opt = CONF.SORT_CMP_OPTIMIZE if hash_opt.nil?
	  
	  if hash_opt
	    @key_proc = eval("proc{#{@block_source.source}}", @context.binding)
	  else
	    @key_proc = BBlock.new(@block_source, @context, self)
	  end
	  
	  buf = []
	  no = 0
	  begin
	    if @pvs
 sampling = false
Log::debugf(self, "%s", @pvs.inspect)
 init_exports
	    elsif self.no == 0
 sampling = true
	    else
 sampling = false
 @pvs = @bjob.get_pvs
Log::debugf(self, "%s", @pvs.inspect)
 init_exports
	    end
 
	    @input.each do |e|
 if sampling
		no += 1
		buf.push e
		if no >= sample_line_no
		  sampling = false
		  @pvs = @bjob.get_pvs(buf)
Log::debugf(self, "%s", @pvs.inspect)
		  init_exports
		  buf.each{|e| hashing(e)}
		end
 else
		hashing(e)
 end
	    end
	    if sampling
 @pvs = @bjob.get_pvs(buf)
Log::debugf(self, "%s", @pvs.inspect)
 init_exports
 buf.each{|e| hashing(e)}
	    end
	  rescue
	    Log::debug_exception(self)
	    raise
	  ensure
	    @exports_queue.push nil
	    @exports.each_pair do |key, export| 
 next unless export
 Log::debug(self, "G0 #{key} => #{@counter[key]}")	    
 export.push END_OF_STREAM
	    end
	  end
	end
end