Class: Fairy::ChunkedSizedPoolQueue

Inherits:

ChunkedPoolQueue

• Object
• ChunkedPoolQueue
• Fairy::ChunkedSizedPoolQueue

Defined in:

lib/fairy/share/port.rb

Instance Attribute Summary

Attributes inherited from ChunkedPoolQueue

#fib_cv

Instance Method Summary

• #initialize(policy, queues_mon = XThread::Monitor.new, queues_cv = queues_mon.new_cond) ⇒ ChunkedSizedPoolQueue constructor

  A new instance of ChunkedSizedPoolQueue.

• #pop ⇒ Object
• #pop_all ⇒ Object
• #push(e) ⇒ Object
• #push_all(buf) ⇒ Object

Constructor Details

#initialize(policy, queues_mon = XThread::Monitor.new, queues_cv = queues_mon.new_cond) ⇒ ChunkedSizedPoolQueue

Returns a new instance of ChunkedSizedPoolQueue.

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

def initialize(policy, queues_mon = XThread::Monitor.new, queues_cv = queues_mon.new_cond)
  super
  @max_size = policy[:size]
  @max_size ||= CONF.ONMEMORY_SIZEDQUEUE_SIZE

  @queue_size = 0

  @pop_cv = @queues_cv
  @push_cv = @queues_mon.new_cond
end

Instance Method Details

#pop ⇒ Object

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

def pop
  e = super
  @queues_mon.synchronize do
  @queue_size -= 1
  @push_cv.broadcast if @queue_size <= @max_size
  end
  e
end

#pop_all ⇒ Object

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

def pop_all
  buf = super
  @queues_mon.synchronize do
  @queue_size -= buf.size
  @push_cv.broadcast if @queue_size <= @max_size
  end
  buf
end

#push(e) ⇒ Object

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

def push(e)
  @queues_mon.synchronize do
  @push_cv.wait_while{@queue_size > @max_size}
  @queue_size += 1
  end
  super
end

#push_all(buf) ⇒ Object

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

def push_all(buf)
  @queues_mon.synchronize do
  @push_cv.wait_while{@queue_size > @max_size}
  @queue_size += buf.size
  end
  super
end