class Aerospike::Recordset

Recordset implements a queue for a producer-consumer pattern a producer is a thread that fetches records from one node and puts them on this queue a consumer fetches records from this queue so the production and the consumptoin are decoupled there can be an unlimited count of producer threads and consumer threads

Attributes

records[R]

Public Class Methods

new(queue_size = 5000, thread_count = 1, type) click to toggle source
# File lib/aerospike/query/recordset.rb, line 27
def initialize(queue_size = 5000, thread_count = 1, type)
  queue_size = thread_count if queue_size < thread_count
  @records = SizedQueue.new(queue_size)

  # holds the count of active threads.
  # when it reaches zero it means the whole operations of fetching records from server nodes is finished
  @active_threads = Atomic.new(thread_count)

  # operation cancelled by user or an exception occured in one of the threads
  @cancelled = Atomic.new(false)

  # saves the exception that occurred inside one of the threads to reraise it in the main thread
  # and also is a signal to terminate other threads as the whole operation is assumed as failed
  @thread_exception = Atomic.new(nil)

  # type of the operation. it is either :scan or :query
  @type = type
end

Public Instance Methods

active?() click to toggle source

recordset is active unless it is cancelled by the user or an exception has occurred in of threads

# File lib/aerospike/query/recordset.rb, line 62
def active?
  !@cancelled.get
end
cancel(expn = nil) click to toggle source

this is called by a thread who faced an exception to singnal to terminate the whole operation it also may be called by the user to terminate the command in the middle of fetching records from server nodes it clears the queue so that if any threads are waiting for the queue get unblocked and find out about the cancellation

# File lib/aerospike/query/recordset.rb, line 82
def cancel(expn = nil)
  set_exception(expn)
  @cancelled.set(true)
  @records.clear
end
each(&block) click to toggle source

fetches and returns all the records from the queue until the whole operation is finished and it reaches an EOF mark calling cancel inside the each block raises an exception to signal other consumer threads

# File lib/aerospike/query/recordset.rb, line 90
def each(&block)
  r = true
  while r
    r = next_record
    # nil means EOF
    unless r.nil?
      block.call(r)
    else
      # reached the EOF
      break
    end
  end
end
is_scan?() click to toggle source

the command is a scan if there are no filters applied otherwise it is a query

# File lib/aerospike/query/recordset.rb, line 105
def is_scan?
  @filters.nil? || @filters.empty?
end
next_record() click to toggle source

fetches and return the first record from the queue if the operation is not finished and the queue is empty it blocks and waits for new records it sets the exception if it reaches the EOF mark, and returns nil EOF means the operation has finished and no more records are comming from server nodes it re-raises the exception occurred in threads, or which was set after reaching the EOF in the previous call

# File lib/aerospike/query/recordset.rb, line 51
def next_record
  raise @thread_exception.get unless @thread_exception.get.nil?

  r = @records.deq

  set_exception if r.nil?

  r
end
thread_finished(expn = nil) click to toggle source

this is called by working threads to signal their job is finished it decreases the count of active threads and puts an EOF on queue when all threads are finished e is an exception that has happened in the exceutor, and outside of the threads themselves

# File lib/aerospike/query/recordset.rb, line 69
def thread_finished(expn = nil)
  @active_threads.update do |v|
    v -= 1
    @records.enq(nil) if v == 0
    v
  end

  raise expn unless expn.nil?
end

Private Instance Methods

set_exception(expn = nil) click to toggle source
# File lib/aerospike/query/recordset.rb, line 111
def set_exception(expn = nil)
  expn ||= (@type == :scan ? SCAN_TERMINATED_EXCEPTION : QUERY_TERMINATED_EXCEPTION)
  @thread_exception.set(expn)
end