Class: Fairy::PGroupBy::CommandMergeSortBuffer

Inherits:

OnMemoryBuffer

• Object
• OnMemoryBuffer
• Fairy::PGroupBy::CommandMergeSortBuffer

Defined in:

lib/fairy/node/p-group-by.rb

Direct Known Subclasses

DirectKBMergeSortBuffer, MergeSortBuffer

Instance Attribute Summary

Attributes inherited from OnMemoryBuffer

#log_id

Instance Method Summary

• #each(&block) ⇒ Object
• #each_2ndmemory(&block) ⇒ Object
• #init_2ndmemory ⇒ Object
• #initialize(njob, policy) ⇒ CommandMergeSortBuffer constructor

  A new instance of CommandMergeSortBuffer.

• #open_buffer(&block) ⇒ Object
• #push(value) ⇒ Object
• #store_2ndmemory(key_values) ⇒ Object

Constructor Details

#initialize(njob, policy) ⇒ CommandMergeSortBuffer

Returns a new instance of CommandMergeSortBuffer.

# File 'lib/fairy/node/p-group-by.rb', line 309

def initialize(njob, policy)
	super

	@key_values_size = 0

	@threshold = policy[:threshold]
	@threshold ||= CONF.GROUP_BY_CMSB_THRESHOLD

	@buffers = nil
end

Instance Method Details

#each(&block) ⇒ Object

# File 'lib/fairy/node/p-group-by.rb', line 380

def each(&block)
	if @buffers
	  each_2ndmemory &block
	else
	  super
	end
end

#each_2ndmemory(&block) ⇒ Object

# File 'lib/fairy/node/p-group-by.rb', line 388

def each_2ndmemory(&block)
	unless @key_values.empty?
	  store_2ndmemory(@key_values)
	end

	Log::debug(self, @buffers.collect{|b| b.path}.join(" "))

	IO::popen("sort -m -k1,1 #{@buffers.collect{|b| b.path}.join(' ')}") do |io|
	  key = nil
	  values = nil
	  io.each do |line|
	    mk, mv = line.split(/\s+/)
	    k = Marshal.load(mk.tr(":", "\n").unpack("m").first)
	    v = Marshal.load(mv.tr(":", "\n").unpack("m").first)
	    if key == k
 values.concat v
	    else
 if values
		values.push_eos
		yield values
 end
 key = k
 values = KeyValueStream.new(key, self)
 values.concat v
	    end
	  end
	  if values
	    values.push_eos
	    yield values
	  end
	end
end

#init_2ndmemory ⇒ Object

# File 'lib/fairy/node/p-group-by.rb', line 320

def init_2ndmemory
	require "fairy/share/fast-tempfile"

	@buffer_dir = @policy[:buffer_dir]
	@buffer_dir ||= CONF.TMP_DIR

	@buffers = []
end

#open_buffer(&block) ⇒ Object

# File 'lib/fairy/node/p-group-by.rb', line 329

def open_buffer(&block)
	unless @buffers
	  init_2ndmemory
	end
	buffer = FastTempfile.open("mod-group-by-buffer-#{@njob.no}-", @buffer_dir)
	@buffers.push buffer
	if block_given?
	  begin
	    # ruby BUG#2390の対応のため.
	    # yield buffer
	    yield buffer.io
	  ensure
	    buffer.close
	  end
	else
	  buffer
	end
end

#push(value) ⇒ Object

# File 'lib/fairy/node/p-group-by.rb', line 348

def push(value)
	super

	@key_values_size += 1
	key_values = nil
	@key_values_mutex.synchronize do
	  if @key_values_size > @threshold
	    key_values = @key_values
	    @key_values_size = 0
	    @key_values = {}
	  end
	  if key_values
	    store_2ndmemory(key_values)
	  end
	end
end

#store_2ndmemory(key_values) ⇒ Object

# File 'lib/fairy/node/p-group-by.rb', line 365

def store_2ndmemory(key_values)
	Log::info(self, "start store")
	sorted = key_values.collect{|key, values| 
	  [[Marshal.dump(key)].pack("m").tr("\n", ":"), 
	    [Marshal.dump(values)].pack("m").tr("\n", ":")]}.sort_by{|e| e.first}

	open_buffer do |io|
	  sorted.each do |k, v|
	    io.puts "#{k}\t#{v}"
	  end
	end
	sorted = nil
	Log::info(self, "end store")
end