Class: Fairy::PDirectProduct::PPostFilter

Inherits:
Fairy::PSingleExportFilter show all
Defined in:
lib/fairy/node/p-direct-product.rb

Constant Summary

Constants included from Fairy::PSingleExportable

Fairy::PSingleExportable::END_OF_STREAM, Fairy::PSingleExportable::ST_EXPORT_FINISH, Fairy::PSingleExportable::ST_WAIT_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 included from Fairy::PSingleExportable

#export

Attributes inherited from Fairy::PFilter

#IGNORE_EXCEPTION, #id, #log_id, #ntask

Instance Method Summary collapse

Methods included from Fairy::PSingleExportable

#start, #start_export, #terminate, #wait_export_finish

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_export, #start_watch_status, #status=, #terminate, #terminate_proc

Constructor Details

#initialize(id, processor, bjob, opts, block_source) ⇒ PPostFilter

Returns a new instance of PPostFilter.



85
86
87
88
89
90
91
92
 
# File 'lib/fairy/node/p-direct-product.rb', line 85

def initialize(id, processor, bjob, opts, block_source)
	super
	@block_source = block_source
	
	@other_imports = nil
	@other_imports_mutex = Mutex.new
	@other_imports_cv = XThread::ConditionVariable.new
end

Instance Method Details

#basic_each(&block) ⇒ Object 



130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/fairy/node/p-direct-product.rb', line 130

def basic_each(&block)
	@map_proc = BBlock.new(@block_source, @context, self)

	elements = []
	elements.push @input.to_a
	elements.push *other_imports.collect{|i| i.to_a}

	idxs = elements.collect{|e| 0}
	max_idxs = elements.collect{|e| e.size}

	cont = true
	while cont
	  e = elements.zip(idxs).collect{|ary, idx| ary[idx]}
	  block.call @map_proc.yield *e

	  (idxs.size-1).downto(0) do |idx|
	    idxs[idx] += 1
	    break if idxs[idx] < max_idxs[idx]
	    idxs[idx] = 0
	    cont = false if idx == 0 && idxs[idx] == 0
	  end
	end
end

#input=(input) ⇒ Object 



94
95
96
97
98
 
# File 'lib/fairy/node/p-direct-product.rb', line 94

def input=(input)
	@input = input
#	self.no = input.no
	self.key = input.key
end

#other_imports ⇒ Object 



121
122
123
124
125
126
127
128
 
# File 'lib/fairy/node/p-direct-product.rb', line 121

def other_imports
	@other_imports_mutex.synchronize do
	  while !@other_imports
	    @other_imports_cv.wait(@other_imports_mutex)
	  end
	  @other_imports
	end
end

#other_inputs=(exports) ⇒ Object 



100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
 
# File 'lib/fairy/node/p-direct-product.rb', line 100

def other_inputs=(exports)
	@other_imports_mutex.synchronize do
	  @other_imports = exports.collect{|exp|
# 後で検討する
#           policy = @opts[:prequeuing_policy]
#	    imp = Import.new(policy)
	    imp = Import.new
	    imp.no = exp.no
	    imp.add_key(exp.key)
	    imp.no_import = 1
	    imp.set_log_callback do |n, key| 
 Log::verbose(self, "IMPORT POP key=#{key}: #{n}")
	    end

	    exp.output = imp
	    imp
	  }
	  @other_imports_cv.broadcast
	end
end