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 315

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 386

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 394

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 326

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 335

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 354

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 371

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