himprove port handling

This commit tries to improve two issues

1) A free port is obtained by setting the port value as zero and the
OS will bind to a free port. We immediately close the port and then
later create another socket on the same port. The issue with the
approach is, OS could allocate the same port to another because we
have closed the port. This leads to a situation where more than one
bypass server could listen on the same port (this is possible because
of SO_REUSEPORT flag). The issue is fixed by not closing the socket.

2) Bypass exposes a down api, which closes the socket. The issue here
is the same as above, the OS is free to allocate the port to
others. The current solution tries to fix the issue by keeping track
of which test owns which ports and try not to reuse the same
ports. This is still not foolproof, there is a small interval during
which the socket is active, but better than the old logic.

Author: ananthakumaran

lib/bypass/application.ex

@@ -4,7 +4,11 @@ defmodule Bypass.Application do
   use Application
 
   def start(_type, _args) do
-    opts = [strategy: :one_for_one, name: Bypass.Supervisor]
-    DynamicSupervisor.start_link(opts)
+    children = [
+      Bypass.FreePort,
+      {DynamicSupervisor, strategy: :one_for_one, name: Bypass.Supervisor}
+    ]
+
+    Supervisor.start_link(children, strategy: :one_for_one)
   end
 end

lib/bypass/free_port.ex

@@ -0,0 +1,60 @@
+defmodule Bypass.FreePort do
+  alias Bypass.Utils
+  use GenServer
+
+  defstruct [:ports, :owners]
+
+  def start_link([]) do
+    GenServer.start_link(__MODULE__, [], name: __MODULE__)
+  end
+
+  def reserve(owner) do
+    GenServer.call(__MODULE__, {:reserve, owner})
+  end
+
+  def init([]) do
+    {:ok, %__MODULE__{ports: MapSet.new(), owners: %{}}}
+  end
+
+  def handle_call({:reserve, owner}, _from, state) do
+    ref = Process.monitor(owner)
+    {state, reply} = find_free_port(state, owner, ref, 0)
+    {:reply, reply, state}
+  end
+
+  def handle_info({:DOWN, ref, _type, pid, _reason}, state) do
+    {port, owners} = Map.pop(state.owners, {pid, ref})
+    {:noreply, %{state | ports: MapSet.delete(state.ports, port), owners: owners}}
+  end
+
+  def handle_info(_msg, state) do
+    {:noreply, state}
+  end
+
+  defp find_free_port(state, _owner, _ref, 10 = _attempt),
+    do: {state, {:error, :too_many_attempts}}
+
+  defp find_free_port(state, owner, ref, attempt) do
+    case :ranch_tcp.listen(Utils.so_reuseport() ++ [ip: Utils.listen_ip(), port: 0]) do
+      {:ok, socket} ->
+        {:ok, port} = :inet.port(socket)
+
+        if MapSet.member?(state.ports, port) do
+          true = :erlang.port_close(socket)
+
+          find_free_port(state, owner, ref, attempt + 1)
+        else
+          state = %{
+            state
+            | ports: MapSet.put(state.ports, port),
+              owners: Map.put_new(state.owners, {owner, ref}, port)
+          }
+
+          {state, {:ok, socket}}
+        end
+
+      {:error, reason} ->
+        {state, {:error, reason}}
+    end
+  end
+end

lib/bypass/instance.ex

@@ -24,14 +24,19 @@ defmodule Bypass.Instance do
   # GenServer callbacks
 
   def init([opts]) do
-    # Get a free port from the OS
-    case :ranch_tcp.listen(so_reuseport() ++ [ip: listen_ip(), port: Keyword.get(opts, :port, 0)]) do
-      {:ok, socket} ->
-        {:ok, port} = :inet.port(socket)
-        :erlang.port_close(socket)
+    result =
+      case Keyword.get(opts, :port) do
+        nil ->
+          Bypass.FreePort.reserve(self())
+
+        port ->
+          {:ok, port}
+      end
 
+    case result do
+      {:ok, port_or_socket} ->
         ref = make_ref()
-        socket = do_up(port, ref)
+        {:ok, port, socket} = do_up(port_or_socket, ref)
 
         state = %{
           expectations: %{},
@@ -77,7 +82,7 @@ defmodule Bypass.Instance do
   end
 
   defp do_handle_call(:up, _from, %{port: port, ref: ref, socket: nil} = state) do
-    socket = do_up(port, ref)
+    {:ok, _port, socket} = do_up(port, ref)
     {:reply, :ok, %{state | socket: socket}}
   end
 
@@ -317,12 +322,17 @@ defmodule Bypass.Instance do
 
   defp match_route(_, _), do: {false, nil}
 
-  defp do_up(port, ref) do
-    plug_opts = [self()]
+  defp do_up(port, ref) when is_integer(port) do
     {:ok, socket} = :ranch_tcp.listen(so_reuseport() ++ [ip: listen_ip(), port: port])
+    do_up(socket, ref)
+  end
+
+  defp do_up(socket, ref) do
+    plug_opts = [self()]
+    {:ok, port} = :inet.port(socket)
     cowboy_opts = cowboy_opts(port, ref, socket)
     {:ok, _pid} = Plug.Cowboy.http(Bypass.Plug, plug_opts, cowboy_opts)
-    socket
+    {:ok, port, socket}
   end
 
   defp do_down(ref, socket) do
@@ -420,46 +430,4 @@ defmodule Bypass.Instance do
   defp cowboy_opts(port, ref, socket) do
     [ref: ref, port: port, transport_options: [num_acceptors: 5, socket: socket]]
   end
-
-  # Use raw socket options to set SO_REUSEPORT so we fix {:error, :eaddrinuse} - where the OS errors
-  # when we attempt to listen on the same port as before, since it's still considered in use.
-  #
-  # See https://lwn.net/Articles/542629/ for details on SO_REUSEPORT.
-  #
-  # See https://github.com/aetrion/erl-dns/blob/0c8d768/src/erldns_server_sup.erl#L81 for an
-  # Erlang library using this approach.
-  #
-  # We want to do this:
-  #
-  #     int optval = 1;
-  #     setsockopt(sfd, SOL_SOCKET, SO_REUSEPORT, &optval, sizeof(optval));
-  #
-  # Use the following C program to find the values on each OS:
-  #
-  #     #include <stdio.h>
-  #     #include <sys/socket.h>
-  #
-  #     int main() {
-  #         printf("SOL_SOCKET: %d\n", SOL_SOCKET);
-  #         printf("SO_REUSEPORT: %d\n", SO_REUSEPORT);
-  #         return 0;
-  #     }
-  defp so_reuseport() do
-    case :os.type() do
-      {:unix, :linux} -> [{:raw, 1, 15, <<1::32-native>>}]
-      {:unix, :darwin} -> [{:raw, 65_535, 512, <<1::32-native>>}]
-      _ -> []
-    end
-  end
-
-  # This is used to override the default behaviour of ranch_tcp
-  # and limit the range of interfaces it will listen on to just
-  # the configured interface. Loopback is a default interface.
-  defp listen_ip do
-    Application.get_env(:bypass, :listen_ip, "127.0.0.1")
-    |> String.split(".")
-    |> Enum.map(&Integer.parse/1)
-    |> Enum.map(&elem(&1, 0))
-    |> List.to_tuple()
-  end
 end

lib/bypass/utils.ex

@@ -15,4 +15,46 @@ defmodule Bypass.Utils do
       :ok
     end
   end
+
+  # This is used to override the default behaviour of ranch_tcp
+  # and limit the range of interfaces it will listen on to just
+  # the configured interface. Loopback is a default interface.
+  def listen_ip do
+    Application.get_env(:bypass, :listen_ip, "127.0.0.1")
+    |> String.split(".")
+    |> Enum.map(&Integer.parse/1)
+    |> Enum.map(&elem(&1, 0))
+    |> List.to_tuple()
+  end
+
+  # Use raw socket options to set SO_REUSEPORT so we fix {:error, :eaddrinuse} - where the OS errors
+  # when we attempt to listen on the same port as before, since it's still considered in use.
+  #
+  # See https://lwn.net/Articles/542629/ for details on SO_REUSEPORT.
+  #
+  # See https://github.com/aetrion/erl-dns/blob/0c8d768/src/erldns_server_sup.erl#L81 for an
+  # Erlang library using this approach.
+  #
+  # We want to do this:
+  #
+  #     int optval = 1;
+  #     setsockopt(sfd, SOL_SOCKET, SO_REUSEPORT, &optval, sizeof(optval));
+  #
+  # Use the following C program to find the values on each OS:
+  #
+  #     #include <stdio.h>
+  #     #include <sys/socket.h>
+  #
+  #     int main() {
+  #         printf("SOL_SOCKET: %d\n", SOL_SOCKET);
+  #         printf("SO_REUSEPORT: %d\n", SO_REUSEPORT);
+  #         return 0;
+  #     }
+  def so_reuseport() do
+    case :os.type() do
+      {:unix, :linux} -> [{:raw, 1, 15, <<1::32-native>>}]
+      {:unix, :darwin} -> [{:raw, 65_535, 512, <<1::32-native>>}]
+      _ -> []
+    end
+  end
 end