Class: Aerospike::Command

Inherits:

Object

• Object
• Aerospike::Command

Defined in:

lib/aerospike/command/command.rb

Overview

:nodoc:

Direct Known Subclasses

MultiCommand, SingleCommand

Instance Method Summary

• #execute ⇒ Object
• #initialize(node = nil) ⇒ Command constructor

  A new instance of Command.

• #set_delete(policy, key) ⇒ Object

  Writes the command for delete operations.

• #set_exists(policy, key) ⇒ Object

  Writes the command for exist operations.

• #set_operate(policy, key, operations) ⇒ Object

  Implements different command operations.

• #set_read(policy, key, bin_names) ⇒ Object

  Writes the command for get operations (specified bins).

• #set_read_for_key_only(policy, key) ⇒ Object

  Writes the command for get operations (all bins).

• #set_read_header(policy, key) ⇒ Object

  Writes the command for getting metadata operations.

• #set_scan(policy, namespace, set_name, bin_names) ⇒ Object
• #set_touch(policy, key) ⇒ Object

  Writes the command for touch operations.

• #set_udf(policy, key, package_name, function_name, args) ⇒ Object
• #set_write(policy, operation, key, bins) ⇒ Object

  Writes the command for write operations.

• #write_bins ⇒ Object

  List of all bins that this command will write to - sub-classes should override this as appropriate.

Constructor Details

#initialize(node = nil) ⇒ Command

Returns a new instance of Command.

# File 'lib/aerospike/command/command.rb', line 89

def initialize(node=nil)
  @data_offset = 0
  @data_buffer = nil

  @node = node

  @compress = false

  # will add before use
  @sequence = Atomic.new(-1)

  self
end

Instance Method Details

#execute ⇒ Object

Raises:

• (Aerospike::Exceptions::Timeout)

# File 'lib/aerospike/command/command.rb', line 427

def execute
  iterations = 0

  # set timeout outside the loop
  limit = Time.now + @policy.timeout

  # Execute command until successful, timed out or maximum iterations have been reached.
  while true
    # too many retries
    iterations += 1
    break if (@policy.max_retries > 0) && (iterations > @policy.max_retries+1)

    # Sleep before trying again, after the first iteration
    sleep(@policy.sleep_between_retries) if iterations > 1 && @policy.sleep_between_retries > 0

    # check for command timeout
    break if @policy.timeout > 0 && Time.now > limit

    begin
      @node = get_node
      @conn = @node.get_connection(@policy.timeout)
    rescue => e
      # Socket connection error has occurred. Decrease health and retry.
      @node.decrease_health

      Aerospike.logger.error("Node #{@node.to_s}: #{e}")
      next
    end

    # Draw a buffer from buffer pool, and make sure it will be put back
    begin
      @data_buffer = Buffer.get

      # Set command buffer.
      begin
        write_buffer
      rescue => e
        Aerospike.logger.error(e)

        # All runtime exceptions are considered fatal. Do not retry.
        # Close socket to flush out possible garbage. Do not put back in pool.
        @conn.close if @conn
        raise e
      end

      # Reset timeout in send buffer (destined for server) and socket.
      @data_buffer.write_int32((@policy.timeout * 1000).to_i, 22)

      # Send command.
      begin
        @conn.write(@data_buffer, @data_offset)
      rescue => e
        # IO errors are considered temporary anomalies. Retry.
        # Close socket to flush out possible garbage. Do not put back in pool.
        @conn.close if @conn

        Aerospike.logger.error("Node #{@node.to_s}: #{e}")
        # IO error means connection to server @node is unhealthy.
        # Reflect cmd status.
        @node.decrease_health
        next
      end

      # Parse results.
      begin
        parse_result
      rescue => e
        case e
        # do not log the following exceptions
        when Aerospike::Exceptions::ScanTerminated
        when Aerospike::Exceptions::QueryTerminated
        else
          Aerospike.logger.error(e)
        end

        # close the connection
        # cancelling/closing the batch/multi commands will return an error, which will
        # close the connection to throw away its data and signal the server about the
        # situation. We will not put back the connection in the buffer.
        @conn.close if @conn
        raise e
      end

      # Reflect healthy status.
      @node.restore_health

      # Put connection back in pool.
      @node.put_connection(@conn)

      # command has completed successfully.  Exit method.
      return
    ensure
      Buffer.put(@data_buffer)
    end

  end # while

  # execution timeout
  raise Aerospike::Exceptions::Timeout.new(limit, iterations)
end

#set_delete(policy, key) ⇒ Object

Writes the command for delete operations

# File 'lib/aerospike/command/command.rb', line 136

def set_delete(policy, key)
  begin_cmd
  field_count = estimate_key_size(key)
  
  predexp_size = estimate_predexp(policy.predexp)
  field_count += 1 if predexp_size > 0

  size_buffer
  write_header_with_policy(policy, 0, INFO2_WRITE|INFO2_DELETE, field_count, 0)
  write_key(key)
  write_predexp(policy.predexp, predexp_size)
  end_cmd
end

#set_exists(policy, key) ⇒ Object

Writes the command for exist operations

# File 'lib/aerospike/command/command.rb', line 168

def set_exists(policy, key)
  begin_cmd
  field_count = estimate_key_size(key)
  
  predexp_size = estimate_predexp(policy.predexp)
  field_count += 1 if predexp_size > 0

  size_buffer
  write_header(policy, INFO1_READ|INFO1_NOBINDATA, 0, field_count, 0)
  write_key(key)
  write_predexp(policy.predexp, predexp_size)
  end_cmd
end

#set_operate(policy, key, operations) ⇒ Object

Implements different command operations

# File 'lib/aerospike/command/command.rb', line 250

def set_operate(policy, key, operations)
  begin_cmd
  field_count = estimate_key_size(key, policy)
  
  predexp_size = estimate_predexp(policy.predexp)
  field_count += 1 if predexp_size > 0

  read_attr = 0
  write_attr = 0
  read_header = false
  record_bin_multiplicity = policy.record_bin_multiplicity == RecordBinMultiplicity::ARRAY

  operations.each do |operation|
    case operation.op_type
    when Aerospike::Operation::READ, Aerospike::Operation::CDT_READ, 
          Aerospike::Operation::HLL_READ, Aerospike::Operation::BIT_READ
      read_attr |= INFO1_READ

      # Read all bins if no bin is specified.
      read_attr |= INFO1_GET_ALL unless operation.bin_name

    when Aerospike::Operation::READ_HEADER
      # The server does not currently return record header data with _INFO1_NOBINDATA attribute set.
      # The workaround is to request a non-existent bin.
      # TODO: Fix this on server.
      # read_attr |= _INFO1_READ | _INFO1_NOBINDATA
      read_attr |= INFO1_READ
      read_header = true

    else
      write_attr = INFO2_WRITE
    end

    if [Aerospike::Operation::HLL_MODIFY, Aerospike::Operation::HLL_READ, 
          Aerospike::Operation::BIT_MODIFY, Aerospike::Operation::BIT_READ].include?(operation.op_type)
      record_bin_multiplicity = true
    end

    estimate_operation_size_for_operation(operation)
  end
  size_buffer


  write_attr |= INFO2_RESPOND_ALL_OPS if write_attr != 0 && record_bin_multiplicity

  if write_attr != 0
    write_header_with_policy(policy, read_attr, write_attr, field_count, operations.length)
  else
    write_header(policy, read_attr, write_attr, field_count, operations.length)
  end
  write_key(key, policy)
  write_predexp(policy.predexp, predexp_size)

  operations.each do |operation|
    write_operation_for_operation(operation)
  end

  write_operation_for_bin(nil, Aerospike::Operation::READ) if read_header

  end_cmd
  mark_compressed(policy)
end

#set_read(policy, key, bin_names) ⇒ Object

Writes the command for get operations (specified bins)

# File 'lib/aerospike/command/command.rb', line 198

def set_read(policy, key, bin_names)
  if bin_names && bin_names.length > 0
    begin_cmd
    field_count = estimate_key_size(key)
    
    predexp_size = estimate_predexp(policy.predexp)
    field_count += 1 if predexp_size > 0


    bin_names.each do |bin_name|
      estimate_operation_size_for_bin_name(bin_name)
    end

    size_buffer
    write_header(policy, INFO1_READ, 0, field_count, bin_names.length)
    write_key(key)
    write_predexp(policy.predexp, predexp_size)

    bin_names.each do |bin_name|
      write_operation_for_bin_name(bin_name, Aerospike::Operation::READ)
    end

    end_cmd
  else
    set_read_for_key_only(policy, key)
  end
end

#set_read_for_key_only(policy, key) ⇒ Object

Writes the command for get operations (all bins)

# File 'lib/aerospike/command/command.rb', line 183

def set_read_for_key_only(policy, key)
  begin_cmd
  field_count = estimate_key_size(key)
  
  predexp_size = estimate_predexp(policy.predexp)
  field_count += 1 if predexp_size > 0

  size_buffer
  write_header(policy, INFO1_READ|INFO1_GET_ALL, 0, field_count, 0)
  write_key(key)
  write_predexp(policy.predexp, predexp_size)
  end_cmd
end

#set_read_header(policy, key) ⇒ Object

Writes the command for getting metadata operations

# File 'lib/aerospike/command/command.rb', line 227

def set_read_header(policy, key)
  begin_cmd
  field_count = estimate_key_size(key)
  
  predexp_size = estimate_predexp(policy.predexp)
  field_count += 1 if predexp_size > 0

  estimate_operation_size_for_bin_name('')
  size_buffer

  # The server does not currently return record header data with _INFO1_NOBINDATA attribute set.
  # The workaround is to request a non-existent bin.
  # TODO: Fix this on server.
  #command.set_read(INFO1_READ | _INFO1_NOBINDATA);
  write_header(policy, INFO1_READ, 0, field_count, 1)

  write_key(key)
  write_predexp(policy.predexp, predexp_size)
  write_operation_for_bin_name('', Aerospike::Operation::READ)
  end_cmd
end

#set_scan(policy, namespace, set_name, bin_names) ⇒ Object

# File 'lib/aerospike/command/command.rb', line 336

def set_scan(policy, namespace, set_name, bin_names)
  # Estimate buffer size
  begin_cmd
  field_count = 0

  if namespace
    @data_offset += namespace.bytesize + FIELD_HEADER_SIZE
    field_count += 1
  end

  if set_name
    @data_offset += set_name.bytesize + FIELD_HEADER_SIZE
    field_count += 1
  end
  
  if policy.records_per_second > 0
    @data_offset += 4 + FIELD_HEADER_SIZE
    field_count += 1
  end

  predexp_size = estimate_predexp(policy.predexp)
  field_count += 1 if predexp_size > 0

  # Estimate scan options size.
  @data_offset += 2 + FIELD_HEADER_SIZE
  field_count += 1

  # Estimate scan timeout size.
  @data_offset += 4 + FIELD_HEADER_SIZE
  field_count += 1

  if bin_names
    bin_names.each do |bin_name|
      estimate_operation_size_for_bin_name(bin_name)
    end
  end

  size_buffer
  read_attr = INFO1_READ

  if !policy.include_bin_data
    read_attr |= INFO1_NOBINDATA
  end

  operation_count = 0
  if bin_names
    operation_count = bin_names.length
  end

  write_header(policy, read_attr, 0, field_count, operation_count)

  if namespace
    write_field_string(namespace, Aerospike::FieldType::NAMESPACE)
  end

  if set_name
    write_field_string(set_name, Aerospike::FieldType::TABLE)
  end

  if policy.records_per_second > 0
    write_field_int(policy.records_per_second, Aerospike::FieldType::RECORDS_PER_SECOND)
  end

  write_predexp(policy.predexp, predexp_size)

  write_field_header(2, Aerospike::FieldType::SCAN_OPTIONS)

  priority = policy.priority & 0xFF
  priority <<= 4
  if policy.fail_on_cluster_change
    priority |= 0x08
  end

  @data_buffer.write_byte(priority, @data_offset)
  @data_offset += 1
  @data_buffer.write_byte(policy.scan_percent.to_i.ord, @data_offset)
  @data_offset += 1

  write_field_header(4, Aerospike::FieldType::SCAN_TIMEOUT)
  @data_buffer.write_uint32(policy.socket_timeout.to_i, @data_offset)
  @data_offset += 4

  if bin_names
    bin_names.each do |bin_name|
      write_operation_for_bin_name(bin_name, Aerospike::Operation::READ)
    end
  end

  end_cmd
end

#set_touch(policy, key) ⇒ Object

Writes the command for touch operations

# File 'lib/aerospike/command/command.rb', line 151

def set_touch(policy, key)
  begin_cmd
  field_count = estimate_key_size(key)
  
  predexp_size = estimate_predexp(policy.predexp)
  field_count += 1 if predexp_size > 0

  estimate_operation_size
  size_buffer
  write_header_with_policy(policy, 0, INFO2_WRITE, field_count, 1)
  write_key(key)
  write_predexp(policy.predexp, predexp_size)
  write_operation_for_operation_type(Aerospike::Operation::TOUCH)
  end_cmd
end

#set_udf(policy, key, package_name, function_name, args) ⇒ Object

# File 'lib/aerospike/command/command.rb', line 313

def set_udf(policy, key, package_name, function_name, args)
  begin_cmd
  field_count = estimate_key_size(key, policy)
  
  predexp_size = estimate_predexp(policy.predexp)
  field_count += 1 if predexp_size > 0

  arg_bytes = args.to_bytes

  field_count += estimate_udf_size(package_name, function_name, arg_bytes)
  size_buffer

  write_header(policy, 0, INFO2_WRITE, field_count, 0)
  write_key(key, policy)
  write_predexp(policy.predexp, predexp_size)
  write_field_string(package_name, Aerospike::FieldType::UDF_PACKAGE_NAME)
  write_field_string(function_name, Aerospike::FieldType::UDF_FUNCTION)
  write_field_bytes(arg_bytes, Aerospike::FieldType::UDF_ARGLIST)

  end_cmd
  mark_compressed(policy)
end

#set_write(policy, operation, key, bins) ⇒ Object

Writes the command for write operations

# File 'lib/aerospike/command/command.rb', line 110

def set_write(policy, operation, key, bins)
  begin_cmd
  field_count = estimate_key_size(key, policy)
  
  predexp_size = estimate_predexp(policy.predexp)
  field_count += 1 if predexp_size > 0

  bins.each do |bin|
    estimate_operation_size_for_bin(bin)
  end

  size_buffer

  write_header_with_policy(policy, 0, INFO2_WRITE, field_count, bins.length)
  write_key(key, policy)
  write_predexp(policy.predexp, predexp_size)

  bins.each do |bin|
    write_operation_for_bin(bin, operation)
  end

  end_cmd
  mark_compressed(policy)
end

#write_bins ⇒ Object

List of all bins that this command will write to - sub-classes should override this as appropriate.

# File 'lib/aerospike/command/command.rb', line 105

def write_bins
  []
end