Class: Fairy::PGroupBy::DirectMergeSortBuffer

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

Defined Under Namespace

Classes: CachedBuffer, KeyValueStream, Merger

Instance Attribute Summary

Attributes inherited from DirectOnMemoryBuffer

#log_id

Instance Method Summary collapse

Constructor Details

#initialize(njob, policy) ⇒ DirectMergeSortBuffer

Returns a new instance of DirectMergeSortBuffer.



895
896
897
898
899
900
901
902
 
# File 'lib/fairy/node/p-group-by.rb', line 895

def initialize(njob, policy)
  super

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

  @buffers = nil
end

Instance Method Details

#each(&block) ⇒ Object 



960
961
962
963
964
965
966
 
# File 'lib/fairy/node/p-group-by.rb', line 960

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

#each_2ndmemory(&block) ⇒ Object 



968
969
970
971
972
973
974
975
976
977
978
 
# File 'lib/fairy/node/p-group-by.rb', line 968

def each_2ndmemory(&block)
  unless @key_values.empty?
    store_2ndmemory(@key_values)
    @key_values = nil
  end
  Log::info(self, "Merge Start: #{@buffers.size} files")
  Log::debug(self, @buffers.collect{|b| b.path}.join(" "))
  
  m = Merger.new(@njob, @buffers)
  m.each(&block)
end

#init_2ndmemory ⇒ Object 



904
905
906
907
908
909
910
911
 
# File 'lib/fairy/node/p-group-by.rb', line 904

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

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

  @buffers = []
end

#open_buffer(&block) ⇒ Object 



913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
 
# File 'lib/fairy/node/p-group-by.rb', line 913

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 



932
933
934
935
936
937
938
939
940
941
942
943
944
945
 
# File 'lib/fairy/node/p-group-by.rb', line 932

def push(value)
  super

  key_values = nil
  @key_values_mutex.synchronize do
    if @key_values.size > @threshold
      key_values = @key_values
      @key_values = []
    end
    if key_values
      store_2ndmemory(key_values)
    end
  end
end

#store_2ndmemory(key_values) ⇒ Object 



947
948
949
950
951
952
953
954
955
956
957
958
 
# File 'lib/fairy/node/p-group-by.rb', line 947

def store_2ndmemory(key_values)
  Log::debug(self, "START STORE")
  key_values = key_values.sort_by{|e| @njob.hash_key(e)}
  
  open_buffer do |io|
    key_values.each_slice(@CHUNK_SIZE) do |ary|
      Marshal.dump(ary, io)
    end
  end
  sorted = nil
  Log::debug(self, "FINISH STORE")
end