Class: Fairy::PSegJoin

Inherits:
PSingleExportFilter show all
Defined in:
lib/fairy/node/p-seg-join.rb

Constant Summary

Constants included from PSingleExportable

Fairy::PSingleExportable::END_OF_STREAM, Fairy::PSingleExportable::ST_EXPORT_FINISH, Fairy::PSingleExportable::ST_WAIT_EXPORT_FINISH

Constants inherited from PIOFilter

Fairy::PIOFilter::ST_WAIT_IMPORT

Constants inherited from PFilter

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

Instance Attribute Summary

Attributes included from PSingleExportable

#export

Attributes inherited from PFilter

#IGNORE_EXCEPTION, #id, #log_id, #ntask

Instance Method Summary collapse

Methods included from PSingleExportable

#start, #start_export, #terminate, #wait_export_finish

Methods inherited from PIOFilter

#input=

Methods inherited from 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, ntask, bjob, opts, block_source) ⇒ PSegJoin

Returns a new instance of PSegJoin.



14
15
16
17
18
19
20
21
 
# File 'lib/fairy/node/p-seg-join.rb', line 14

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

  @join_imports = nil
  @join_imports_mutex = Mutex.new
  @join_imports_cv = XThread::ConditionVariable.new
end

Instance Method Details

#basic_each(&block) ⇒ Object 



58
59
60
61
62
63
64
65
 
# File 'lib/fairy/node/p-seg-join.rb', line 58

def basic_each(&block)
  @map_proc = BBlock.new(@block_source, @context, self)
  arg = [@input]
  arg.push *join_imports
  arg.push block
	  
  @map_proc.yield(*arg)
end

#join_imports ⇒ Object 



23
24
25
26
27
28
29
30
 
# File 'lib/fairy/node/p-seg-join.rb', line 23

def join_imports
  @join_imports_mutex.synchronize do
	while !@join_imports
	  @join_imports_cv.wait(@join_imports_mutex)
	end
  end
  @join_imports
end

#join_inputs=(jinputs) ⇒ Object 



34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/fairy/node/p-seg-join.rb', line 34

def join_inputs=(jinputs)
  policy = @opts[:prequeuing_policy]
  @join_imports_mutex.synchronize do
	@join_imports = jinputs.collect{|jinput| 
	  if jinput
 import = Import.new(policy)
 import.no = jinput.no
 import.add_key(jinput.key)
 import.set_log_callback do |n, key| 
   Log::verbose(self, "IMPORT POP key=#{key}: #{n}")
 end

 import
	  else
 nil
	  end
	}
	@join_imports_cv.broadcast
  end
  @join_imports
end