Class: Fairy::SortedQueue

Inherits:

Object

• Object
• Fairy::SortedQueue

Defined in:

lib/fairy/share/port.rb

Instance Method Summary

• #init_2ndmemory ⇒ Object
• #initialize(policy, queue_mon = XThread::Monitor.new, queue_cv = queue_mon.new_cond) ⇒ SortedQueue constructor

  A new instance of SortedQueue.

• #open_2ndmemory(&block) ⇒ Object
• #pop ⇒ Object
• #pop_2ndmemory ⇒ Object
• #pop_all ⇒ Object
• #push(e) ⇒ Object
• #store_2ndmemory(ary) ⇒ Object

Constructor Details

#initialize(policy, queue_mon = XThread::Monitor.new, queue_cv = queue_mon.new_cond) ⇒ SortedQueue

Returns a new instance of SortedQueue.

# File 'lib/fairy/share/port.rb', line 1392

def initialize(policy, queue_mon = XThread::Monitor.new, queue_cv = queue_mon.new_cond)
  @policy = policy

  @pool_threshold = policy[:pool_threshold]
  @pool_threshold ||= CONF.SORTEDQUEUE_POOL_THRESHOLD
  
  @threshold = policy[:threshold]
  @threshold ||= CONF.SORTEDQUEUE_THRESHOLD

  @push_queue = []
  @pop_queue = nil
  @buffers = nil

  @queue_mon = queue_mon
  @queue_cv = queue_cv

  @sort_by = policy[:sort_by]
  @sort_by ||= CONF.SORTEDQUEUE_SORTBY   

  if @sort_by.kind_of?(String)
	@sort_by = eval("proc{#{@sort_by}}")
  end
end

Instance Method Details

#init_2ndmemory ⇒ Object

# File 'lib/fairy/share/port.rb', line 1486

def init_2ndmemory
  require "tempfile"

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

  @buffers = []
  @merge_io = nil
end

#open_2ndmemory(&block) ⇒ Object

# File 'lib/fairy/share/port.rb', line 1496

def open_2ndmemory(&block)
  unless @buffers
	init_2ndmemory
  end
  buffer = Tempfile.open("port-buffer-", @buffer_dir)
  begin
	# ruby BUG#2390の対応のため.
#	yield buffer
	yield buffer.instance_eval{@tmpfile}
  ensure
	buffer.close
  end
  @buffers.push buffer
  buffer
end

#pop ⇒ Object

# File 'lib/fairy/share/port.rb', line 1442

def pop
  @queue_mon.synchronize do
	@queue_cv.wait_while{@pop_queue.nil?}

	if @buffers.nil?
#Log::debug(self, @pop_queue.inspect)
	  return @pop_queue.shift
	else
	  pop_2ndmemory
	end
  end
end

#pop_2ndmemory ⇒ Object

# File 'lib/fairy/share/port.rb', line 1524

def pop_2ndmemory
  unless @merge_io
	@buffers.each{|tf| tf.open}
	@merge_io = @buffers.map{|io| 
	  e = nil
	  begin
 e = Marshal.load(io)
	  rescue EOFError
 io.close!
	  end
	  [io, e]}.select{|io, v| !v.nil?}.sort_by{|io, v| @sort_by.call(v)}
  end
  unless io_min = @merge_io.shift
	return :END_OF_STREAM
  end
  
  io, min = io_min
  begin
	e = Marshal.load(io)
	@merge_io.push [io, e] 
	@merge_io = @merge_io.sort_by{|io, e| @sort_by.call(e)}
  rescue EOFError
	io.close!
  end
  min
end

#pop_all ⇒ Object

# File 'lib/fairy/share/port.rb', line 1455

def pop_all
  buf = []
  while e = pop
	buf.push e
	return buf if buf.size > @pool_threshold
  end
  buf
end

#push(e) ⇒ Object

# File 'lib/fairy/share/port.rb', line 1416

def push(e)
  @queue_mon.synchronize do
	@push_queue.push e
	if e == :END_OF_STREAM
	  @push_queue.pop
	  if @buffers
 store_2ndmemory(@push_queue)
 @push_queue = []
 @pop_queue = []
	  else
 begin
   @pop_queue = @push_queue.sort_by{|e| @sort_by.call(e)}
   @pop_queue.push :END_OF_STREAM
 rescue
   Log::debug_exception
 end
	  end
	  @queue_cv.broadcast
	end
	if @push_queue.size >= @threshold
	  store_2ndmemory(@push_queue)
	  @push_queue = []
	end
  end
end

#store_2ndmemory(ary) ⇒ Object

# File 'lib/fairy/share/port.rb', line 1512

def store_2ndmemory(ary)
  Log::debug(self, "start store: ")
  open_2ndmemory do |io|
	ary = ary.sort_by{|e| @sort_by.call(e)}
	while !ary.empty?
	  e = ary.shift
	  Marshal.dump(e, io)
	end
  end
  Log::debug(self, "end store")
end