Class: Contender::LinkedQueue

Inherits:
Queue
  • Object
show all
Defined in:
lib/contender/linked_queue.rb

Defined Under Namespace

Classes: Node

Instance Method Summary collapse

Methods inherited from Queue

#empty?

Constructor Details

#initialize(capacity = FIXNUM_MAX) ⇒ LinkedQueue

Returns a new instance of LinkedQueue.

Raises:

  • (ArgumentError) 


3
4
5
6
7
8
9
10
11
12
13
14
15
16
 
# File 'lib/contender/linked_queue.rb', line 3

def initialize(capacity = FIXNUM_MAX)
  raise ArgumentError if capacity <= 0

  @capacity = capacity
  # Invariant: head.item == nil
  # Invariant: tail.next == nil
  @head = @tail = Node.new
  @count = Counter.new

  @take_lock = Monitor.new
  @not_empty = @take_lock.new_cond
  @put_lock = Monitor.new
  @not_full = @put_lock.new_cond
end

Instance Method Details

#capacity_remaining ⇒ Integer

Returns:

  • (Integer) 


221
222
223
 
# File 'lib/contender/linked_queue.rb', line 221

def capacity_remaining
  @capacity - @count.get
end

#clear ⇒ undefined

Returns:

  • (undefined)

Raises:

  • (NotImplementedError) 


139
140
141
 
# File 'lib/contender/linked_queue.rb', line 139

def clear
  raise NotImplementedError
end

#delete(element) ⇒ Boolean

Parameters:

  • element (Object)

Returns:

  • (Boolean) 


151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
 
# File 'lib/contender/linked_queue.rb', line 151

def delete(element)
  return false if element.nil?

  full_synchronize do
    trail = @head
    interior = trail.next

    while interior
      if element == interior.element
        unlink interior, trail
        return true
      end

      trail = interior
      interior = interior.next
    end
  end

  return false
end

#drain_to(target, max_elements = FIXNUM_MAX) ⇒ Integer

Parameters:

  • target (Array)
  • max_elements (Integer) (defaults to: FIXNUM_MAX)

Returns:

  • (Integer) 


175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
 
# File 'lib/contender/linked_queue.rb', line 175

def drain_to(target, max_elements = FIXNUM_MAX)
  should_signal = false

  @take_lock.synchronize do
    n = [max_elements, size].min
    i = 0

    head = @head

    begin
      while i < n
        interior = head.next
        target.add interior.element!
        head.next = head
        head = interior

        i += 1
      end

      return n
    ensure
      if i > 0
        @head = head
        should_signal = @count.get_and_add(-i) == capacity
      end
    end
  end
ensure
  signal_not_full if should_signal
end

#each {|Object| ... } ⇒ undefined

Yields:

  • (Object)

Returns:

  • (undefined) 


145
146
147
 
# File 'lib/contender/linked_queue.rb', line 145

def each(&block)
  to_a.each &block
end

#include?(element) ⇒ Boolean

Returns:

  • (Boolean) 


206
207
208
209
210
211
212
213
214
215
216
217
218
 
# File 'lib/contender/linked_queue.rb', line 206

def include?(element)
  return false if element.nil?

  full_synchronize do
    current = @head.next
    while current
      return true if element == current.element
      current = current.next
    end
  end

  return false
end

#offer(element, timeout = nil) ⇒ Boolean

Parameters:

  • element (Object)
  • timeout (Float) (defaults to: nil)

Returns:

  • (Boolean) 


21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
 
# File 'lib/contender/linked_queue.rb', line 21

def offer(element, timeout = nil)
  validate_element element

  count = 0
  node = Node.new element

  @put_lock.synchronize do
    if size == @capacity
      return false unless timeout

      @not_full.wait timeout
      return false if size == @capacity
    end

    enqueue node
    count = @count.increment
    if count < @capacity
      @not_full.signal
    end
  end

  if count > 0
    signal_not_empty
  end

  return true
end

#peek ⇒ Object

Returns:

  • (Object) 


129
130
131
132
133
134
135
136
 
# File 'lib/contender/linked_queue.rb', line 129

def peek
  return if size == 0

  @take_lock.synchronize do
    first = @head.next
    return first.element if first
  end
end

#poll(timeout = nil) ⇒ Object

Parameters:

  • timeout (Float) (defaults to: nil)

Returns:

  • (Object) 


76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/contender/linked_queue.rb', line 76

def poll(timeout = nil)
  count = 0
  element = nil

  @take_lock.synchronize do
    if size == 0
      return unless timeout

      @not_empty.wait timeout
      return if size == 0
    end

    element = dequeue
    count = @count.decrement

    if count > 0
      @not_empty.signal
    end
  end

  if count < @capacity
    signal_not_full
  end

  return element
end

#put(element) ⇒ undefined

Parameters:

  • element (Object)

Returns:

  • (undefined) 


51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/contender/linked_queue.rb', line 51

def put(element)
  validate_element element

  count = 0
  node = Node.new element

  @put_lock.synchronize do
    while size == @capacity
      @not_full.wait
    end

    enqueue node
    count = @count.increment
    if count < @capacity
      @not_full.signal
    end
  end

  if count > 0
    signal_not_empty
  end
end

#size ⇒ Integer

Returns:

  • (Integer) 


226
227
228
 
# File 'lib/contender/linked_queue.rb', line 226

def size
  @count.get
end

#take ⇒ Object

Returns:

  • (Object) 


104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
 
# File 'lib/contender/linked_queue.rb', line 104

def take
  count = 0
  element = nil

  @take_lock.synchronize do
    while size == 0
      @not_empty.wait
    end

    element = dequeue
    count = @count.decrement

    if count > 0
      @not_empty.signal
    end
  end

  if count < @capacity
    signal_not_full
  end

  return element
end

#to_a ⇒ Array

Returns:

  • (Array) 


231
232
233
234
235
236
237
238
239
240
241
242
243
 
# File 'lib/contender/linked_queue.rb', line 231

def to_a
  full_synchronize do
    result = Array.new size

    current = @head.next
    while current
      result.push current.element
      current = current.next
    end

    result
  end
end