Implement RFC 8305 for TcpClient

This changes TcpClient.new and TcpClient.with_timeout to support
connecting to multiple IP addresses based on RFC 8305, also known as
"Happy Eyeballs version 2".

In addition, TcpClient.new now uses a default timeout of 60 seconds
instead of waiting forever, and std.time.Duration is extended with a few
extra methods such as Duration.positive? and Duration./.

This fixes #795.

Changelog: added

std/src/std/net/socket.inko

@@ -59,17 +59,19 @@
 #
 # For more information about timeouts versus deadlines, consider reading [this
 # article](https://vorpus.org/blog/timeouts-and-cancellation-for-humans/).
-import std.cmp (Equal)
+import std.cmp (Equal, min)
 import std.drop (Drop)
 import std.fmt (Format, Formatter)
 import std.fs.path (Path)
 import std.io (Error, Read, Write, WriteInternal)
+import std.iter (Stream)
 import std.libc
 import std.net.ip (IpAddress)
 import std.string (ToString)
+import std.sync (Channel)
 import std.sys.net
 import std.sys.unix.net (self as sys) if unix
-import std.time (Duration, ToInstant)
+import std.time (Duration, Instant, ToInstant)
 
 # The maximum value valid for a listen() call.
 #
@@ -79,6 +81,10 @@ import std.time (Duration, ToInstant)
 # Linux (for example) would have no effect.
 let MAXIMUM_LISTEN_BACKLOG = 65_535
 
+# The amount of milliseconds to wait for a socket when performing connections in
+# parallel.
+let PARALLEL_CONNECT_TIMEOUT = 250
+
 fn pub try_ips[R, E](
   ips: ref Array[IpAddress],
   fun: fn (IpAddress) -> Result[R, E],
@@ -101,12 +107,132 @@ fn pub try_ips[R, E](
   Result.Error(last.or_panic('at least one IP address must be specified'))
 }
 
+# Returns an iterator that yields IP addresses in alternating order, starting
+# with an IPv6 address.
+fn interleave_ips(ips: ref Array[IpAddress]) -> Stream[IpAddress] {
+  let mut v6_idx = 0
+  let mut v4_idx = 0
+  let mut v6 = true
+
+  Stream.new(fn move {
+    if v6 := v6.false? {
+      loop {
+        match ips.opt(v6_idx := v6_idx + 1) {
+          case Some(V6(ip)) -> return Option.Some(IpAddress.V6(ip))
+          case Some(_) -> {}
+          case _ -> break
+        }
+      }
+    }
+
+    loop {
+      match ips.opt(v4_idx := v4_idx + 1) {
+        case Some(V4(ip)) -> return Option.Some(IpAddress.V4(ip))
+        case Some(_) -> {}
+        case _ -> return Option.None
+      }
+    }
+  })
+}
+
 trait RawSocketOperations {
   fn mut raw_socket -> Pointer[net.RawSocket]
 
   fn raw_deadline -> Int
 }
 
+# A type for connecting a `TcpClient` asynchronously.
+type async ParallelConnection {
+  let @id: Int
+  let @ip: IpAddress
+  let @port: Int
+  let @deadline: Instant
+  let @output: Channel[Result[TcpClient, Int]]
+  let @run: Bool
+
+  fn static new(
+    id: Int,
+    ip: IpAddress,
+    port: Int,
+    deadline: Instant,
+    output: uni Channel[Result[TcpClient, Int]],
+  ) -> ParallelConnection {
+    ParallelConnection(
+      id: id,
+      ip: ip,
+      port: port,
+      deadline: deadline,
+      output: output,
+      run: true,
+    )
+  }
+
+  fn async mut cancel {
+    @run = false
+  }
+
+  fn async connect {
+    if @run.false? { return }
+
+    # To support cancellation we use an internal timeout. This way we don't just
+    # sit around for e.g. 60 seconds even though another socket connected
+    # successfully.
+    let interval = Duration.from_millis(PARALLEL_CONNECT_TIMEOUT)
+    let deadline = min(interval.to_instant, @deadline)
+    let res = recover {
+      match TcpClient.connect(@ip, @port, deadline) {
+        case Ok(v) -> Result.Ok(v)
+        case Error(TimedOut) if @deadline.remaining.to_nanos > 0 -> {
+          # We finished one cycle but there's still time left, so let's try
+          # again until the user-provided deadline is also exceeded.
+          return connect
+        }
+        case Error(_) -> {
+          # We wan out of time or encountered a non-timeout error (e.g. the
+          # connection is refused). In this case we need to report back to the
+          # parent process such that it doesn't hang waiting for a result
+          # forever.
+          Result.Error(@id)
+        }
+      }
+    }
+
+    @output.send(res)
+  }
+}
+
+type inline ParallelConnections {
+  let @port: Int
+  let @timeout_after: Instant
+  let @channel: Channel[Result[TcpClient, Int]]
+  let @connections: Array[ParallelConnection]
+
+  fn static new(port: Int, timeout_after: Instant) -> ParallelConnections {
+    ParallelConnections(
+      port: port,
+      timeout_after: timeout_after,
+      channel: Channel.new,
+      connections: [],
+    )
+  }
+
+  fn mut connect(ip: IpAddress) -> Int {
+    let id = @connections.size
+    let chan = recover @channel.clone
+    let proc = ParallelConnection.new(id, ip, @port, @timeout_after, chan)
+
+    proc.connect
+    @connections.push(proc)
+    id
+  }
+}
+
+impl Drop for ParallelConnections {
+  fn mut drop {
+    @connections.iter_mut.each(fn (c) { c.cancel })
+  }
+}
+
 # An IPv4 or IPv6 socket address.
 type pub copy SocketAddress {
   # The IPv4/IPv6 address of this socket address.
@@ -811,24 +937,102 @@ type pub TcpClient {
     Result.Ok(TcpClient(socket))
   }
 
-  # Creates a new `TcpClient` that's connected to an IP address and port number.
-  #
-  # If multiple IP addresses are given, this method attempts to connect to them
-  # in order, returning upon the first successful connection. If no connection
-  # can be established, the error of the last attempt is returned.
-  #
-  # This method doesn't enforce a deadline on establishing the connection. If
-  # you need to limit the amount of time spent waiting to establish the
-  # connection, use `TcpClient.with_timeout` instead.
+  fn static connect(
+    ip: IpAddress,
+    port: Int,
+    timeout_after: Instant,
+  ) -> Result[TcpClient, Error] {
+    let socket = try Socket.stream(ip.v6?)
+
+    socket.timeout_after = timeout_after
+    try socket.connect(ip, port)
+    socket.reset_deadline
+    from(socket)
+  }
+
+  fn static connect_parallel(
+    ips: ref Array[IpAddress],
+    port: Int,
+    timeout_after: Instant,
+  ) -> Result[TcpClient, Error] {
+    let size = ips.size
+
+    # It's possible the list of IPs is passed directly from "user input" such as
+    # a DNS record. If this list is empty we don't want to panic and abort, but
+    # instead give callers a chance to handle the error. As such, we return
+    # `InvalidArgument` instead.
+    #
+    # When there's only one IP address we can skip the Happy Eyeballs algorithm
+    # and just connect to it directly.
+    match size {
+      case 0 -> throw Error.InvalidArgument
+      case 1 -> return connect(ips.get(0), port, timeout_after)
+      case _ -> {}
+    }
+
+    let ips = interleave_ips(ips)
+    let cons = ParallelConnections.new(port, timeout_after)
+    let mut pending = 0
+
+    while timeout_after.remaining.positive? {
+      let id = match ips.next {
+        case Some(ip) -> {
+          pending += 1
+          cons.connect(ip)
+        }
+        case _ -> break
+      }
+
+      let wait = Duration.from_millis(PARALLEL_CONNECT_TIMEOUT)
+      let deadline = min(timeout_after, wait.to_instant)
+
+      while deadline.remaining.positive? {
+        match cons.channel.receive_until(deadline) {
+          case Some(Ok(v)) -> return Result.Ok(v)
+          # If the socket we're waiting for produces an error then there's no
+          # point in waiting any longer, so we just move on.
+          case Some(Error(v)) if v == id -> {
+            pending -= 1
+            break
+          }
+          # If a socket we tried to use previously produces an error we just
+          # ignore it and continue waiting for the current socket.
+          case Some(_) -> pending -= 1
+          case _ -> break
+        }
+      }
+    }
+
+    # None of the sockets could connect within the initial timeout, but they
+    # might connect before our supplied deadline (if this hasn't already expired
+    # at this point).
+    while pending > 0 and timeout_after.remaining.positive? {
+      match cons.channel.receive_until(timeout_after) {
+        case Some(Ok(v)) -> return Result.Ok(v)
+        case Some(_) -> pending -= 1
+        case _ -> break
+      }
+    }
+
+    Result.Error(
+      if timeout_after.remaining.positive? {
+        Error.ConnectionRefused
+      } else {
+        Error.TimedOut
+      },
+    )
+  }
+
+  # Creates a new `TcpClient` that's connected to an IP address and port number,
+  # using a default timeout.
   #
-  # # Panics
+  # This method uses a default timeout of 60 seconds. If you wish to use a
+  # custom timeout/deadline, use `TcpClient.with_timeout` instead.
   #
-  # This method panics if `ips` is empty.
+  # For more details, refer to the documentation of `TcpClient.with_timeout`.
   #
   # # Examples
   #
-  # Connecting a `TcpClient`:
-  #
   # ```inko
   # import std.net.socket (TcpClient)
   # import std.net.ip (IpAddress)
@@ -839,29 +1043,45 @@ type pub TcpClient {
     ips: ref Array[IpAddress],
     port: Int,
   ) -> Result[TcpClient, Error] {
-    try_ips(ips, fn (ip) {
-      let socket = try Socket.stream(ip.v6?)
-
-      try socket.connect(ip, port)
-      from(socket)
-    })
+    with_timeout(ips, port, Duration.from_secs(60))
   }
 
   # Creates a new `TcpClient` but limits the amount of time spent waiting for
   # the connection to be established.
   #
+  # The `timeout_after` argument specifies the deadline after which the
+  # `connect()` system call times out. This deadline is _not_ inherited by the
+  # returned `TcpClient`.
+  #
+  # # Connecting to multiple IP addresses
+  #
   # If multiple IP addresses are given, this method attempts to connect to them
-  # in order, returning upon the first successful connection. If no connection
-  # can be established, the error of the last attempt is returned.
+  # in accordance with [RFC 8305](https://datatracker.ietf.org/doc/html/rfc8305)
+  # (also known as "Happy Eyeballs version 2"), with the following differences:
   #
-  # The `timeout_after` argument specifies the deadline after which the
-  # `connect()` times out. The deadline is cleared once connected.
+  # - DNS requests are performed separately and thus not subject to the Happy
+  #   Eyeballs algorithm.
+  # - We always interleave IPv6 and IPv4 addresses, starting with an IPv6
+  #   address (so `IPv6, IPv4, IPv6, IPv4, ...`).
+  # - There's no way to configure this behavior, nor is it planned to add the
+  #   ability to do so.
+  #
+  # # Errors
+  #
+  # If the connection can't be established, a `std.io.Error` error is returned.
   #
-  # See `TcpClient.new` for more information.
+  # If `ips` contains multiple IP addresses and a connection can't be
+  # established to any of the addresses, one of the following errors is
+  # returned:
   #
-  # # Panics
+  # - `Error.ConnectionRefused` if no connection could be established before the
+  #   deadline expired
+  # - `Error.TimedOut` if the deadline expired
   #
-  # This method panics if `ips` is empty.
+  # If `ips` is empty, an `Error.InvalidArgument` error is returned instead of
+  # producing a panic. This is to allow handling of cases where one passes the
+  # output of e.g. `std.net.dns.Resolver.resolve` directly to this method
+  # without checking if the DNS record actually contains any IP addresses.
   #
   # # Examples
   #
@@ -883,14 +1103,7 @@ type pub TcpClient {
     port: Int,
     timeout_after: ref T,
   ) -> Result[TcpClient, Error] {
-    try_ips(ips, fn (ip) {
-      let socket = try Socket.stream(ip.v6?)
-
-      socket.timeout_after = timeout_after
-      try socket.connect(ip, port)
-      socket.reset_deadline
-      from(socket)
-    })
+    connect_parallel(ips, port, timeout_after.to_instant)
   }
 
   # Returns the local address of this socket.