Class: Fairy::PGroupBy::SimpleCommandSortBuffer

Inherits:
Object
  • Object
show all
Defined in:
lib/fairy/node/p-group-by.rb

Instance Method Summary collapse

Constructor Details

#initialize(njob, policy) ⇒ SimpleCommandSortBuffer

Returns a new instance of SimpleCommandSortBuffer.



258
259
260
261
262
263
264
265
266
267
268
 
# 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 



281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
 
# 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 



270
271
272
273
274
275
276
277
278
279
 
# 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