Class: Fairy::PGroupBy::ExtMergeSortBuffer

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

Instance Attribute Summary

Attributes inherited from OnMemoryBuffer

#log_id

Instance Method Summary collapse

Methods inherited from MergeSortBuffer

#store_2ndmemory

Methods inherited from CommandMergeSortBuffer

#each, #init_2ndmemory, #initialize, #open_buffer, #push, #store_2ndmemory

Methods inherited from OnMemoryBuffer

#each, #initialize, #push

Constructor Details

This class inherits a constructor from Fairy::PGroupBy::CommandMergeSortBuffer

Instance Method Details

#each_2ndmemory(&block) ⇒ Object 



525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
 
# File 'lib/fairy/node/p-group-by.rb', line 525

def each_2ndmemory(&block)
  require "deep-connect/deep-fork"
  
  unless @key_values.empty?
    store_2ndmemory(@key_values)
  end

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

  df = DeepConnect::DeepFork.fork(@njob.processor.deepconnect){|dc, ds|
    $0 = "fairy processor sorter"

    dc.export("Sorter", self)

    finish_wait
#   ds.close
#   dc.stop
    sleep 1
  }
  sorter = df.peer_deep_space.import("Sorter", true)
  sorter.sub_each {|key, values|
# sorter.sub_each {|bigstr|
#     values = bigstr.split("\t").collect{|e| 
#       e.gsub(/(\\t|\\\\)/){|v| v == "\\t" ? "\t" : "\\"}
#     }
#     key = values.shift
    block.call values
    nil  # referenceが戻らないようにしている
  }
  sorter.finish
# df.peer_deep_space.close
  @buffers.each{|buf| buf.close!}
  Process.waitpid(df.peer_pid)
end

#finish ⇒ Object 



604
605
606
 
# File 'lib/fairy/node/p-group-by.rb', line 604

def finish
  @cv.signal
end

#finish_wait ⇒ Object

DeepConnect.def_method_spec(self, “REF sub_each()DVAL”)



596
597
598
599
600
601
602
 
# File 'lib/fairy/node/p-group-by.rb', line 596

def finish_wait
  @mx = Mutex.new
  @cv = XThread::ConditionVariable.new
  @mx.synchronize do
    @cv.wait(@mx)
  end
end

#sub_each(&block) ⇒ Object 



560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
 
# File 'lib/fairy/node/p-group-by.rb', line 560

def sub_each(&block)
  bufs = @buffers.collect{|buf|
    buf.open
    kv = read_line(buf.io)
    [kv, buf]
  }.select{|kv, buf| !kv.nil?}.sort_by{|kv, buf| kv[0]}
  
  key = nil
  values = []
  while buf_min = bufs.shift
    kv, buf = buf_min

    if key == kv[0]
      values.concat kv[1]
    else
      yield key, values unless values.empty?
      key = kv[0]
      values = kv[1]
    end

    next unless line = read_line(buf.io)
    idx = bufs.rindex{|kv, b| kv[0] <= line[0]}
    idx ? bufs.insert(idx+1, [line, buf]) : bufs.unshift([line, buf])
  end
  unless values.empty?
    yield values
#     values.unshift key
#     bigstr = values.collect{|e| 
#       e.gsub(/[\\\t]/){|v| v == "\t" ? "\\t" : '\\\\'}
#     }.join("\t")
#     yield bigstr
  end
  nil  # referenceが戻らないようにしている
end