Class: Fairy::CDirectProduct

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

Defined Under Namespace

Classes: CPostFilter, CPreFilter

Instance Attribute Summary collapse

Attributes inherited from CIOFilter

#input

Attributes included from CInputtable

#input

Instance Method Summary collapse

Methods inherited from CIOFilter

#node_class, #output=

Methods included from CInputtable

#break_running, #inputtable?

Methods inherited from CFilter

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

Constructor Details

#initialize(controller, opts, others, block_source) ⇒ CDirectProduct

Returns a new instance of CDirectProduct.



22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/fairy/master/c-direct-product.rb', line 22

def initialize(controller, opts, others, block_source)
  super

  @others = others
  @block_source = block_source

  @main_prefilter = CPreFilter.new(@controller, @opts, block_source)
  @main_prefilter.main = self
  @other_prefilters = []
  @others.each do |other|
	prefilter = CPreFilter.new(@controller, @opts, block_source)
	prefilter.main = self
	@other_prefilters.push prefilter
  end
  @postfilter = CPostFilter.new(@controller, @opts, block_source)

  @prefilter_no_nodes = {}
  @prefilter_no_nodes_mutex = Mutex.new
  @prefilter_no_nodes_cv = XThread::ConditionVariable.new
end

Instance Attribute Details

#other_prefilters ⇒ Object (readonly)

Returns the value of attribute other_prefilters.



43
44
45
 
# File 'lib/fairy/master/c-direct-product.rb', line 43

def other_prefilters
  @other_prefilters
end

Instance Method Details

#all_prefilters ⇒ Object 



45
46
47
 
# File 'lib/fairy/master/c-direct-product.rb', line 45

def all_prefilters
  [@main_prefilter, *@other_prefilters]
end

#each_assigned_filter(&block) ⇒ Object 



53
54
55
 
# File 'lib/fairy/master/c-direct-product.rb', line 53

def each_assigned_filter(&block)
  @postfilter.each_assigned_filter &block
end

#input=(other) ⇒ Object 



57
58
59
60
61
62
63
64
 
# File 'lib/fairy/master/c-direct-product.rb', line 57

def input=(other)
  @main_prefilter.input = other
  @others.zip(@other_prefilters) do |o, prefilter|
	prefilter.input = o
  end

  @postfilter.input = @main_prefilter
end

#njob_creation_params ⇒ Object 



49
50
51
 
# File 'lib/fairy/master/c-direct-product.rb', line 49

def njob_creation_params
  [@block_source]
end

#no_of_exports_for_prefilter(prefilter) ⇒ Object 



73
74
75
76
77
78
79
80
81
82
 
# File 'lib/fairy/master/c-direct-product.rb', line 73

def no_of_exports_for_prefilter(prefilter)
  all_prefilters.reject{|f| prefilter==f}.inject(1){|dp, f|
	@prefilter_no_nodes_mutex.synchronize do
	  while (v = @prefilter_no_nodes[f]).nil?
 @prefilter_no_nodes_cv.wait(@prefilter_no_nodes_mutex)
	  end
	  dp *= v
	end
  }
end

#update_prefilter_no_nodes(prefilter) ⇒ Object 



66
67
68
69
70
71
 
# File 'lib/fairy/master/c-direct-product.rb', line 66

def update_prefilter_no_nodes(prefilter)
  @prefilter_no_nodes_mutex.synchronize do
	@prefilter_no_nodes[prefilter] = prefilter.number_of_nodes
	@prefilter_no_nodes_cv.broadcast
  end
end