Class: Fairy::CDirectProduct::CPreFilter

Inherits:
Fairy::CIOFilter show all
Defined in:
lib/fairy/master/c-direct-product.rb

Overview

呼ばれない

def start_create_nodes
  @main_prefilter.start_create_nodes
end

Instance Attribute Summary

Attributes inherited from Fairy::CIOFilter

#input

Attributes included from Fairy::CInputtable

#input

Instance Method Summary collapse

Methods inherited from Fairy::CIOFilter

#node_class, #output=

Methods included from Fairy::CInputtable

#break_running, #inputtable?

Methods inherited from Fairy::CFilter

#abort_create_node, #add_node, #assgin_number_of_nodes?, #break_create_node, #break_running, #create_and_add_node, #create_import, #create_node, #create_nodes, #def_job_pool_variable, #each_node, #each_node_exist_only, #handle_exception, #input, #job_pool_dict, #job_pool_variable, #nodes, #number_of_nodes, #pool_dict, #postmapping_policy, #start_export, #start_watch_node_status, #update_status, watch_status, watch_status=, #watch_status?

Constructor Details

#initialize(controller, opts, block_source) ⇒ CPreFilter

Returns a new instance of CPreFilter.



92
93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/fairy/master/c-direct-product.rb', line 92

def initialize(controller, opts, block_source)
	super
	@block_source = block_source

	@no = 0
	@exports = {}
	@exports_mutex = Mutex.new
#	@exports_cv = XThread::ConditionVariable.new

	@products = nil
	@products_mutex = Mutex.new
	@products_cv = XThread::ConditionVariable.new
end

Instance Method Details

#bind_export(exp, imp) ⇒ Object 



183
184
185
 
# File 'lib/fairy/master/c-direct-product.rb', line 183

def bind_export(exp, imp)
	# do nothing
end

#each_assigned_filter(&block) ⇒ Object 



139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
 
# File 'lib/fairy/master/c-direct-product.rb', line 139

def each_assigned_filter(&block)
	Thread.start do
	  @main.other_prefilters.each do |p| 
	    p.each_node do |n| 
 @exports_mutex.synchronize do 
		@exports[n] = n.exports.dc_dup
#		@exports_cv.broadcast
 end
	    end
	  end
	  @products_mutex.synchronize do 
	    @products = nodes.product(*@main.other_prefilters.collect{|p| p.nodes})
	    @products_cv.broadcast
	  end
	end

	super
end

#each_export_by(njob, mapper, &block) ⇒ Object

main prefilter 用



159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
 
# File 'lib/fairy/master/c-direct-product.rb', line 159

def each_export_by(njob, mapper, &block)
	@exports_mutex.synchronize do
	  @exports[njob] = njob.exports.dc_dup
#	  @exports_cv.broadcast
	end
	@products_mutex.synchronize do
	  while !@products
	    @products_cv.wait(@products_mutex)
	  end
	  
	  post_njob_no = -1
	  @products.each do |main_njob, *others_njobs|
	    post_njob_no += 1
	    next if main_njob != njob
	    @others_njobs = others_njobs
	    
	    block.call(@exports[main_njob].shift,
   :init_njob => proc{|njob| 
			 njob.no = post_njob_no
			 njob.other_inputs = others_njobs.collect{|n| @exports[n].shift}})
	  end
	end
end

#main=(main) ⇒ Object 



106
107
108
 
# File 'lib/fairy/master/c-direct-product.rb', line 106

def main=(main)
	@main = main
end

#njob_creation_params ⇒ Object 



114
115
116
 
# File 'lib/fairy/master/c-direct-product.rb', line 114

def njob_creation_params
	[@block_source]
end

#node_class_name ⇒ Object 



110
111
112
 
# File 'lib/fairy/master/c-direct-product.rb', line 110

def node_class_name
	"PDirectProduct::PPreFilter"
end

#number_of_exports ⇒ Object 



123
124
125
 
# File 'lib/fairy/master/c-direct-product.rb', line 123

def number_of_exports
	@main.no_of_exports_for_prefilter(self)
end

#number_of_nodes=(no) ⇒ Object 



118
119
120
121
 
# File 'lib/fairy/master/c-direct-product.rb', line 118

def number_of_nodes=(no)
	super
	@main.update_prefilter_no_nodes(self)
end

#start_create_nodes ⇒ Object 



127
128
129
130
131
132
133
134
135
136
137
 
# File 'lib/fairy/master/c-direct-product.rb', line 127

def start_create_nodes
 	Log::debug self, "START_CREATE_NODES: #{self}"
 	@main.other_prefilters.each do |other|
 	  Thread.start do
other.each_assigned_filter do |input_filter|
  exp = input_filter.start_export
end
 	  end
 	end
	super
end