Class: Fairy::PGroupBy::SimpleCommandSortBuffer

Inherits:

Object

• Object
• Fairy::PGroupBy::SimpleCommandSortBuffer

Defined in:

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

Instance Method Summary

• #each(&block) ⇒ Object
• #initialize(njob, policy) ⇒ SimpleCommandSortBuffer constructor

  A new instance of SimpleCommandSortBuffer.

• #push(value) ⇒ Object

Constructor Details

#initialize(njob, policy) ⇒ SimpleCommandSortBuffer

Returns a new instance of SimpleCommandSortBuffer.

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

def initialize(njob, policy)
	require "fairy/share/fast-tempfile"

	@njob = njob
	@policy = policy

	@buffer_dir = policy[:buffer_dir]
	@buffer_dir ||= CONF.TMP_DIR
	@buffer = FastTempfile.open("mod-group-by-buffer--#{@njob.no}", @buffer_dir)
	@buffer_mutex = Mutex.new
end

Instance Method Details

#each(&block) ⇒ Object

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

def each(&block)
	buffile = @buffer.path
	@buffer.close
	IO::popen("sort #{buffile}") do |io|
	  key = nil
	  values = nil
	  io.each do |line|
	    
#Log::debug(self, line)

	    mk, mv = line.split(" ")
	    k = Marshal.load(mk.tr(":", "\n").unpack("m").first)
	    v = Marshal.load(mv.tr(":", "\n").unpack("m").first)
	    if key == k
 values.push v
	    else
 if values
		values.push_eos
		yield values
 end
 values = KeyValueStream.new(k, self)
 key = k
 values.push v
	    end
	  end
	  if values
	    values.push_eos
	    yield values
	  end
	end
end

#push(value) ⇒ Object

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

def push(value)
	key = @njob.hash_key(value)

	@buffer_mutex.synchronize do
	  @buffer.io << [Marshal.dump(key)].pack("m").tr("\n", ":")
	  @buffer.io << " "
	  @buffer.io << [Marshal.dump(value)].pack("m").tr("\n", ":")
	  @buffer.io << "\n"
	end
end