Class: Mongo::Socket Private

# File 'lib/mongo/socket.rb', line 270

def connectable?
  true
end

# File 'lib/mongo/socket.rb', line 259

def eof?
  @socket.eof?
rescue IOError, SystemCallError
  true
end

# File 'lib/mongo/socket.rb', line 78

attr_reader :family

# File 'lib/mongo/socket.rb', line 108

def monitor?
  !!options[:monitor]
end

# File 'lib/mongo/socket.rb', line 84

attr_reader :options

# File 'lib/mongo/socket.rb', line 81

attr_reader :socket

# File 'lib/mongo/socket.rb', line 87

attr_reader :timeout

# File 'lib/mongo/socket.rb', line 170

def close
  begin
    # Sometimes it seems the close call can hang for a long time
    ::Timeout.timeout(5) do
      @socket&.close
    end
  rescue
    # Silence all errors
  end
  true
end

# File 'lib/mongo/socket.rb', line 92

def connection_address
  options[:connection_address]
end

# File 'lib/mongo/socket.rb', line 100

def connection_generation
  options[:connection_generation]
end

# File 'lib/mongo/socket.rb', line 473

def do_write(*args, timeout: nil)
  if timeout.nil?
    write_without_timeout(*args)
  else
    write_with_timeout(*args, timeout: timeout)
  end
end

# File 'lib/mongo/socket.rb', line 192

def gets(*args)
  map_exceptions do
    @socket.gets(*args)
  end
end

# File 'lib/mongo/socket.rb', line 619

def human_address
  raise NotImplementedError
end

# File 'lib/mongo/socket.rb', line 607

def map_exceptions
  begin
    yield
  rescue Errno::ETIMEDOUT => e
    raise Error::SocketTimeoutError, "#{e.class}: #{e} (for #{human_address})"
  rescue IOError, SystemCallError, ::SocketError => e
    raise Error::SocketError, "#{e.class}: #{e} (for #{human_address})"
  rescue OpenSSL::SSL::SSLError => e
    raise Error::SocketError, "#{e.class}: #{e} (for #{human_address})"
  end
end

# File 'lib/mongo/socket.rb', line 115

def pipe
  options[:pipe]
end

# File 'lib/mongo/socket.rb', line 623

def raise_timeout_error!(message = nil, csot = false)
  if csot
    raise Mongo::Error::TimeoutError
  else
    raise Errno::ETIMEDOUT, message
  end
end

# File 'lib/mongo/socket.rb', line 215

def read(length, socket_timeout: nil, timeout: nil)
  if !socket_timeout.nil? && !timeout.nil?
    raise ArgumentError, 'Both timeout and socket_timeout cannot be set'
  end
  if !socket_timeout.nil? || timeout.nil?
    read_without_timeout(length, socket_timeout)
  else
    read_with_timeout(length, timeout)
  end
end

# File 'lib/mongo/socket.rb', line 457

def read_buffer_size
  # Buffer size for non-TLS reads
  # 64kb
  65536
end

# File 'lib/mongo/socket.rb', line 334

def read_from_socket(length, socket_timeout: nil, csot: false)
  # Just in case
  if length == 0
    return ''.force_encoding('BINARY')
  end

  _timeout = socket_timeout || self.timeout
  if _timeout
    if _timeout > 0
      deadline = Utils.monotonic_time + _timeout
    elsif _timeout < 0
      raise_timeout_error!("Negative timeout #{_timeout} given to socket", csot)
    end
  end

  # We want to have a fixed and reasonably small size buffer for reads
  # because, for example, OpenSSL reads in 16 kb chunks max.
  # Having a 16 mb buffer means there will be 1000 reads each allocating
  # 16 mb of memory and using 16 kb of it.
  buf_size = read_buffer_size
  data = nil

  # If we want to read less than the buffer size, just allocate the
  # memory that is necessary
  if length < buf_size
    buf_size = length
  end

  # The binary encoding is important, otherwise Ruby performs encoding
  # conversions of some sort during the write into the buffer which
  # kills performance
  buf = allocate_string(buf_size)
  retrieved = 0
  begin
    while retrieved < length
      retrieve = length - retrieved
      if retrieve > buf_size
        retrieve = buf_size
      end
      chunk = @socket.read_nonblock(retrieve, buf)

      # If we read the entire wanted length in one operation,
      # return the data as is which saves one memory allocation and
      # one copy per read
      if retrieved == 0 && chunk.length == length
        return chunk
      end

      # If we are here, we are reading the wanted length in
      # multiple operations. Allocate the total buffer here rather
      # than up front so that the special case above won't be
      # allocating twice
      if data.nil?
        data = allocate_string(length)
      end

      # ... and we need to copy the chunks at this point
      data[retrieved, chunk.length] = chunk
      retrieved += chunk.length
    end
  # As explained in https://ruby-doc.com/core-trunk/IO.html#method-c-select,
  # reading from a TLS socket may require writing which may raise WaitWritable
  rescue IO::WaitReadable, IO::WaitWritable => exc
    if deadline
      select_timeout = deadline - Utils.monotonic_time
      if select_timeout <= 0
        raise_timeout_error!("Took more than #{_timeout} seconds to receive data", csot)
      end
    end
    if exc.is_a?(IO::WaitReadable)
      if pipe
        select_args = [[@socket, pipe], nil, [@socket, pipe], select_timeout]
      else
        select_args = [[@socket], nil, [@socket], select_timeout]
      end
    else
      select_args = [nil, [@socket], [@socket], select_timeout]
    end

    rv = Kernel.select(*select_args)
    if Lint.enabled?
      if pipe && rv&.include?(pipe)
        # If the return value of select is the read end of the pipe, and
        # an IOError is not raised, then that means the socket is still
        # open. Select is interrupted be closing the write end of the
        # pipe, which either returns the pipe if the socket is open, or
        # raises an IOError if it isn't. Select is interrupted after all
        # of the pending and checked out connections have been interrupted
        # and closed, and this only happens once the pool is cleared with
        # interrupt_in_use connections flag. This means that in order for
        # the socket to still be open when the select is interrupted, and
        # that socket is being read from, that means after clear was
        # called, a connection from the previous generation was checked
        # out of the pool, for reading on its socket. This should be impossible.
        raise Mongo::LintError, "Select interrupted for live socket. This should be impossible."
      end
    end

    if BSON::Environment.jruby?
      # Ignore the return value of Kernel.select.
      # On JRuby, select appears to return nil prior to timeout expiration
      # (apparently due to a EAGAIN) which then causes us to fail the read
      # even though we could have retried it.
      # Check the deadline ourselves.
      if deadline
        select_timeout = deadline - Utils.monotonic_time
        if select_timeout <= 0
          raise_timeout_error!("Took more than #{_timeout} seconds to receive data", csot)
        end
      end
    elsif rv.nil?
      raise_timeout_error!("Took more than #{_timeout} seconds to receive data (select call timed out)", csot)
    end
    retry
  end

  data
end

# File 'lib/mongo/socket.rb', line 283

def read_with_timeout(length, timeout)
  deadline = Utils.monotonic_time + timeout
  map_exceptions do
    String.new.tap do |data|
      while data.length < length
        socket_timeout = deadline - Utils.monotonic_time
        if socket_timeout <= 0
          raise Mongo::Error::TimeoutError
        end
        chunk = read_from_socket(length - data.length, socket_timeout: socket_timeout, csot: true)
        unless chunk.length > 0
          raise IOError, "Expected to read > 0 bytes but read 0 bytes"
        end
        data << chunk
      end
    end
  end
end

# File 'lib/mongo/socket.rb', line 310

def read_without_timeout(length, socket_timeout = nil)
  map_exceptions do
    String.new.tap do |data|
      while data.length < length
        chunk = read_from_socket(length - data.length, socket_timeout: socket_timeout)
        unless chunk.length > 0
          raise IOError, "Expected to read > 0 bytes but read 0 bytes"
        end
        data << chunk
      end
    end
  end
end

# File 'lib/mongo/socket.rb', line 234

def readbyte
  map_exceptions do
    @socket.readbyte
  end
end

# File 'lib/mongo/socket.rb', line 580

def set_keepalive_opts(sock)
  sock.setsockopt(SOL_SOCKET, SO_KEEPALIVE, true)
  set_option(sock, :TCP_KEEPINTVL, DEFAULT_TCP_KEEPINTVL)
  set_option(sock, :TCP_KEEPCNT, DEFAULT_TCP_KEEPCNT)
  set_option(sock, :TCP_KEEPIDLE, DEFAULT_TCP_KEEPIDLE)
  set_option(sock, :TCP_USER_TIMEOUT, DEFAULT_TCP_USER_TIMEOUT)
rescue
  # JRuby 9.2.13.0 and lower do not define TCP_KEEPINTVL etc. constants.
  # JRuby 9.2.14.0 defines the constants but does not allow to get or
  # set them with this error:
  # Errno::ENOPROTOOPT: Protocol not available - Protocol not available
end

# File 'lib/mongo/socket.rb', line 593

def set_option(sock, option, default)
  if Socket.const_defined?(option)
    system_default = sock.getsockopt(IPPROTO_TCP, option).int
    if system_default > default
      sock.setsockopt(IPPROTO_TCP, option, default)
    end
  end
end

# File 'lib/mongo/socket.rb', line 602

def set_socket_options(sock)
  sock.set_encoding(BSON::BINARY)
  set_keepalive_opts(sock)
end

# File 'lib/mongo/socket.rb', line 122

def summary
  fileno = @socket&.fileno rescue '<no socket>' || '<no socket>'
  if monitor?
    indicator = if options[:push]
      'pm'
    else
      'm'
    end
    "#{connection_address};#{indicator};fd=#{fileno}"
  else
    "#{connection_address};c:#{connection_generation};fd=#{fileno}"
  end
end

# File 'lib/mongo/socket.rb', line 568

def unix_socket?(sock)
  defined?(UNIXSocket) && sock.is_a?(UNIXSocket)
end

# File 'lib/mongo/socket.rb', line 551

def wait_for_socket_to_be_writable(deadline)
  select_timeout = deadline - Utils.monotonic_time
  rv = Kernel.select(nil, [@socket], nil, select_timeout)

  if BSON::Environment.jruby?
    # Ignore the return value of Kernel.select.
    # On JRuby, select appears to return nil prior to timeout expiration
    # (apparently due to a EAGAIN) which then causes us to fail the read
    # even though we could have retried it.
    # Check the deadline ourselves.
    select_timeout = deadline - Utils.monotonic_time
    return select_timeout > 0
  end

  !rv.nil?
end

# File 'lib/mongo/socket.rb', line 250

def write(*args, timeout: nil)
  map_exceptions do
    do_write(*args, timeout: timeout)
  end
end

# File 'lib/mongo/socket.rb', line 532

def write_chunk(chunk, timeout)
  deadline = Utils.monotonic_time + timeout

  written = 0
  while written < chunk.length
    begin
      written += @socket.write_nonblock(chunk[written..-1])
    rescue IO::WaitWritable, Errno::EINTR
      if !wait_for_socket_to_be_writable(deadline)
        raise_timeout_error!("Took more than #{timeout} seconds to receive data", true)
      end

      retry
    end
  end

  written
end

# File 'lib/mongo/socket.rb', line 515

def write_with_timeout(*args, timeout:)
  raise ArgumentError, 'timeout cannot be nil' if timeout.nil?
  raise_timeout_error!("Negative timeout #{timeout} given to socket", true) if timeout < 0

  written = 0
  args.each do |buf|
    buf = buf.to_s
    i = 0
    while i < buf.length
      chunk = buf[i...(i + WRITE_CHUNK_SIZE)]
      written += write_chunk(chunk, timeout)
      i += WRITE_CHUNK_SIZE
    end
  end
  written
end

# File 'lib/mongo/socket.rb', line 486

def write_without_timeout(*args)
  # This method used to forward arguments to @socket.write in a
  # single call like so:
  #
  # @socket.write(*args)
  #
  # Turns out, when each buffer to be written is large (e.g. 32 MiB),
  # this write call would take an extremely long time (20+ seconds)
  # while using 100% CPU. Splitting the writes into chunks produced
  # massively better performance (0.05 seconds to write the 32 MiB of
  # data on the same hardware). Unfortunately splitting the data,
  # one would assume, results in it being copied, but this seems to be
  # a much more minor issue compared to CPU cost of writing large buffers.
  args.each do |buf|
    buf = buf.to_s
    i = 0
    while i < buf.length
      chunk = buf[i, WRITE_CHUNK_SIZE]
      i += @socket.write(chunk)
    end
  end
end