Class: Fairy::SizedMarshaledQueue

Inherits:
MarshaledQueue show all
Defined in:
lib/fairy/share/port-marshaled-queue.rb

Instance Attribute Summary

Attributes inherited from MarshaledQueue

#fib_cv

Instance Method Summary collapse

Constructor Details

#initialize(policy, queues_mon = Monitor.new, queues_cv = queues_mon.new_cond) ⇒ SizedMarshaledQueue

Returns a new instance of SizedMarshaledQueue.



127
128
129
130
131
132
133
134
 
# File 'lib/fairy/share/port-marshaled-queue.rb', line 127

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

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

Instance Method Details

#pop ⇒ Object 



157
158
159
160
161
162
163
 
# File 'lib/fairy/share/port-marshaled-queue.rb', line 157

def pop
  e = super
  @queues_mon.synchronize do
	@push_cv.broadcast if @queues.size <= @max_size
  end
  e
end

#pop_all ⇒ Object 



165
166
167
168
169
170
171
 
# File 'lib/fairy/share/port-marshaled-queue.rb', line 165

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

#pop_raw ⇒ Object 



173
174
175
176
177
178
179
 
# File 'lib/fairy/share/port-marshaled-queue.rb', line 173

def pop_raw
  raw = super
  @queues_mon.synchronize do
	@push_cv.broadcast if @queues.size <= @max_size
  end
  raw
end

#push(e) ⇒ Object 



136
137
138
139
140
141
 
# File 'lib/fairy/share/port-marshaled-queue.rb', line 136

def push(e)
  @queues_mon.synchronize do
	@push_cv.wait_while{@queues.size > @max_size}
  end
  super
end

#push_all(buf) ⇒ Object 



143
144
145
146
147
148
 
# File 'lib/fairy/share/port-marshaled-queue.rb', line 143

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

#push_raw(raw) ⇒ Object 



150
151
152
153
154
155
 
# File 'lib/fairy/share/port-marshaled-queue.rb', line 150

def push_raw(raw)
  @queues_mon.synchronize do
	@push_cv.wait_while{@queues.size > @max_size}
  end
  super
end