Class: Contender::Pool::TaskQueue

Inherits:
Object
  • Object
show all
Defined in:
lib/contender/pool/task_queue.rb

Overview

Simple thread-safe queue that holds tasks waiting for execution

Instance Method Summary collapse

Constructor Details

#initialize ⇒ undefined 



6
7
8
9
10
 
# File 'lib/contender/pool/task_queue.rb', line 6

def initialize
  @condition = ConditionVariable.new
  @mutex = Mutex.new
  @tasks = Array.new
end

Instance Method Details

#backlog ⇒ Integer

Performs a volatile read of the size of the task queue

Returns:

  • (Integer) 


16
17
18
 
# File 'lib/contender/pool/task_queue.rb', line 16

def backlog
  @tasks.size
end

#dequeue(non_block = false) ⇒ Task

Dequeues the first task in line

If this operation is performed as non-blocking and the queue is empty, then the result of the dequeue will be nil.

If this operation is performed as blocking, then it will wait until a task is added to the queue. In the case that the thread pool is shutdown during while a worker is waiting for the operation to complete, nil will be returned immediately.

Parameters:

  • non_block (Boolean) (defaults to: false)

Returns:



62
63
64
65
66
67
68
69
70
71
 
# File 'lib/contender/pool/task_queue.rb', line 62

def dequeue(non_block = false)
  @mutex.synchronize do
    if @tasks.empty?
      return if non_block
      @condition.wait @mutex
    end

    @tasks.shift
  end
end

#empty? ⇒ Boolean

Returns true if the task queue is empty

Returns:

  • (Boolean) 


24
25
26
 
# File 'lib/contender/pool/task_queue.rb', line 24

def empty?
  @tasks.empty?
end

#enqueue(task) ⇒ undefined

Enqueues the given task for execution

Parameters:

Returns:

  • (undefined) 


43
44
45
46
47
48
 
# File 'lib/contender/pool/task_queue.rb', line 43

def enqueue(task)
  @mutex.synchronize do
    @tasks.push task
    @condition.signal
  end
end

#purge ⇒ undefined

Removes any tasks that are queued for execution

Returns:

  • (undefined) 


32
33
34
35
36
 
# File 'lib/contender/pool/task_queue.rb', line 32

def purge
  @mutex.synchronize do
    @tasks.clear
  end
end

#wakeup ⇒ undefined

Wakes up any threads waiting for tasks to be queued

Returns:

  • (undefined) 


75
76
77
 
# File 'lib/contender/pool/task_queue.rb', line 75

def wakeup
  @condition.broadcast
end