Method: Socket.tcp_with_fast_fallback

Defined in:

lib/socket.rb

.tcp_with_fast_fallback(host, port, local_host = nil, local_port = nil, connect_timeout: nil, resolv_timeout: nil) ⇒ Object

# File 'lib/socket.rb', line 677

def self.tcp_with_fast_fallback(host, port, local_host = nil, local_port = nil, connect_timeout: nil, resolv_timeout: nil)
  if local_host || local_port
    local_addrinfos = Addrinfo.getaddrinfo(local_host, local_port, nil, :STREAM, timeout: resolv_timeout)
    resolving_family_names = local_addrinfos.map { |lai| ADDRESS_FAMILIES.key(lai.afamily) }.uniq
  else
    local_addrinfos = []
    resolving_family_names = ADDRESS_FAMILIES.keys
  end

  hostname_resolution_threads = []
  resolution_store = HostnameResolutionStore.new(resolving_family_names)
  connecting_sockets = {}
  is_windows_environment ||= (RUBY_PLATFORM =~ /mswin|mingw|cygwin/)

  now = current_clock_time
  resolution_delay_expires_at = nil
  connection_attempt_delay_expires_at = nil
  user_specified_connect_timeout_at = nil
  last_error = nil

  if resolving_family_names.size == 1
    family_name = resolving_family_names.first
    addrinfos = Addrinfo.getaddrinfo(host, port, family_name, :STREAM, timeout: resolv_timeout)
    resolution_store.add_resolved(family_name, addrinfos)
    hostname_resolution_result = nil
    hostname_resolution_notifier = nil
    user_specified_resolv_timeout_at = nil
  else
    hostname_resolution_result = HostnameResolutionResult.new(resolving_family_names.size)
    hostname_resolution_notifier = hostname_resolution_result.notifier

    hostname_resolution_threads.concat(
      resolving_family_names.map { |family|
        thread_args = [family, host, port, hostname_resolution_result]
        thread = Thread.new(*thread_args) { |*thread_args| resolve_hostname(*thread_args) }
        Thread.pass
        thread
      }
    )

    user_specified_resolv_timeout_at = resolv_timeout ? now + resolv_timeout : Float::INFINITY
  end

  loop do
    if resolution_store.any_addrinfos? &&
        !resolution_delay_expires_at &&
        !connection_attempt_delay_expires_at
      while (addrinfo = resolution_store.get_addrinfo)
        if local_addrinfos.any?
          local_addrinfo = local_addrinfos.find { |lai| lai.afamily == addrinfo.afamily }

          if local_addrinfo.nil? # Connecting addrinfoと同じアドレスファミリのLocal addrinfoがない
            if resolution_store.any_addrinfos?
              # Try other Addrinfo in next "while"
              next
            elsif connecting_sockets.any? || resolution_store.any_unresolved_family?
              # Exit this "while" and wait for connections to be established or hostname resolution in next loop
              # Or exit this "while" and wait for hostname resolution in next loop
              break
            else
              raise SocketError.new 'no appropriate local address'
            end
          end
        end

        begin
          if resolution_store.any_addrinfos? ||
             connecting_sockets.any? ||
             resolution_store.any_unresolved_family?
            socket = Socket.new(addrinfo.pfamily, addrinfo.socktype, addrinfo.protocol)
            socket.bind(local_addrinfo) if local_addrinfo
            result = socket.connect_nonblock(addrinfo, exception: false)
          else
            result = socket = local_addrinfo ?
              addrinfo.connect_from(local_addrinfo, timeout: connect_timeout) :
              addrinfo.connect(timeout: connect_timeout)
          end

          if result == :wait_writable
            connection_attempt_delay_expires_at = now + CONNECTION_ATTEMPT_DELAY
            if resolution_store.empty_addrinfos?
              user_specified_connect_timeout_at = connect_timeout ? now + connect_timeout : Float::INFINITY
            end

            connecting_sockets[socket] = addrinfo
            break
          else
            return socket # connection established
          end
        rescue SystemCallError => e
          socket&.close
          last_error = e

          if resolution_store.any_addrinfos?
            # Try other Addrinfo in next "while"
            next
          elsif connecting_sockets.any? || resolution_store.any_unresolved_family?
            # Exit this "while" and wait for connections to be established or hostname resolution in next loop
            # Or exit this "while" and wait for hostname resolution in next loop
            break
          else
            raise last_error
          end
        end
      end
    end

    ends_at =
      if resolution_store.any_addrinfos?
        resolution_delay_expires_at || connection_attempt_delay_expires_at
      else
        [user_specified_resolv_timeout_at, user_specified_connect_timeout_at].compact.max
      end

    hostname_resolved, writable_sockets, except_sockets = IO.select(
      hostname_resolution_notifier,
      connecting_sockets.keys,
      # Use errorfds to wait for non-blocking connect failures on Windows
      is_windows_environment ? connecting_sockets.keys : nil,
      second_to_timeout(current_clock_time, ends_at),
    )
    now = current_clock_time
    resolution_delay_expires_at = nil if expired?(now, resolution_delay_expires_at)
    connection_attempt_delay_expires_at = nil if expired?(now, connection_attempt_delay_expires_at)

    if writable_sockets&.any?
      while (writable_socket = writable_sockets.pop)
        is_connected = is_windows_environment || (
          sockopt = writable_socket.getsockopt(Socket::SOL_SOCKET, Socket::SO_ERROR)
          sockopt.int.zero?
        )

        if is_connected
          connecting_sockets.delete writable_socket
          return writable_socket
        else
          failed_ai = connecting_sockets.delete writable_socket
          writable_socket.close
          ip_address = failed_ai.ipv6? ? "[#{failed_ai.ip_address}]" : failed_ai.ip_address
          last_error = SystemCallError.new("connect(2) for #{ip_address}:#{failed_ai.ip_port}", sockopt.int)

          if writable_sockets.any? || connecting_sockets.any?
            # Try other writable socket in next "while"
            # Or exit this "while" and wait for connections to be established or hostname resolution in next loop
          elsif resolution_store.any_addrinfos? || resolution_store.any_unresolved_family?
            # Exit this "while" and try other connection attempt
            # Or exit this "while" and wait for hostname resolution in next loop
            connection_attempt_delay_expires_at = nil
            user_specified_connect_timeout_at = nil
          else
            raise last_error
          end
        end
      end
    end

    if except_sockets&.any?
      except_sockets.each do |except_socket|
        failed_ai = connecting_sockets.delete except_socket
        sockopt = except_socket.getsockopt(Socket::SOL_SOCKET, Socket::SO_ERROR)
        except_socket.close
        ip_address = failed_ai.ipv6? ? "[#{failed_ai.ip_address}]" : failed_ai.ip_address
        last_error = SystemCallError.new("connect(2) for #{ip_address}:#{failed_ai.ip_port}", sockopt.int)

        if except_sockets.any? || connecting_sockets.any?
          # Cleanup other except socket in next "each"
          # Or exit this "while" and wait for connections to be established or hostname resolution in next loop
        elsif resolution_store.any_addrinfos? || resolution_store.any_unresolved_family?
          # Exit this "while" and try other connection attempt
          # Or exit this "while" and wait for hostname resolution in next loop
          connection_attempt_delay_expires_at = nil
          user_specified_connect_timeout_at = nil
        else
          raise last_error
        end
      end
    end

    if hostname_resolved&.any?
      while (family_and_result = hostname_resolution_result.get)
        family_name, result = family_and_result

        if result.is_a? Exception
          resolution_store.add_error(family_name, result)

          unless (Socket.const_defined?(:EAI_ADDRFAMILY)) &&
            (result.is_a?(Socket::ResolutionError)) &&
            (result.error_code == Socket::EAI_ADDRFAMILY)
            other = family_name == :ipv6 ? :ipv4 : :ipv6
            if !resolution_store.resolved?(other) || !resolution_store.resolved_successfully?(other)
              last_error = result
            end
          end
        else
          resolution_store.add_resolved(family_name, result)
        end
      end

      if resolution_store.resolved?(:ipv4)
        if resolution_store.resolved?(:ipv6)
          hostname_resolution_notifier = nil
          resolution_delay_expires_at = nil
          user_specified_resolv_timeout_at = nil
        elsif resolution_store.resolved_successfully?(:ipv4)
          resolution_delay_expires_at = now + RESOLUTION_DELAY
        end
      end
    end

    if resolution_store.empty_addrinfos?
      if connecting_sockets.empty? && resolution_store.resolved_all_families?
        raise last_error
      end

      if (expired?(now, user_specified_resolv_timeout_at) || resolution_store.resolved_all_families?) &&
         (expired?(now, user_specified_connect_timeout_at) || connecting_sockets.empty?)
        raise Errno::ETIMEDOUT, 'user specified timeout'
      end
    end
  end
ensure
  hostname_resolution_threads.each do |thread|
    thread.exit
  end

  hostname_resolution_result&.close

  connecting_sockets.each_key do |connecting_socket|
    connecting_socket.close
  end
end