elixir: refactor to avoid use of dirtyio

Potentially-blocking Rust-side calls now use message passing to respond
to the caller. Elixir now has `Tailscale.Util.await`, which spawns a
task that listens for the relevant response.

Signed-off-by: Nathan Perry <nathan@tailscale.com>
Change-Id: Iafe01153ca52b9f618c80666de5e6f186a6a6964

Author: npry

Cargo.lock

@@ -1,4 +1,6 @@
 defmodule Tailscale do
+  require Tailscale.Util
+
   @moduledoc """
   Elixir bindings for the Tailscale Rust client.
 
@@ -66,8 +68,8 @@ defmodule Tailscale do
 
   See `t:options/0` for details on available options.
   """
-  def connect(key_file_path, options) when is_binary(key_file_path) do
-    case Tailscale.Native.load_key_file(key_file_path) do
+  def connect(key_file_path, options) when is_binary(key_file_path) and is_list(options) do
+    case Tailscale.Util.await(Tailscale.Native.load_key_file(key_file_path)) do
       {:ok, keys} ->
         Keyword.put(options, :keys, keys) |> connect()
 
@@ -86,8 +88,10 @@ defmodule Tailscale do
   """
   def connect(options \\ [])
 
-  def connect(options) when is_list(options),
-    do: :proplists.to_map(options) |> Tailscale.Native.connect()
+  def connect(options) when is_list(options) do
+    options = :proplists.to_map(options)
+    Tailscale.Util.await(Tailscale.Native.connect(options))
+  end
 
   def connect(key_file_path) when is_binary(key_file_path), do: connect(key_file_path, [])
 
@@ -97,7 +101,7 @@ defmodule Tailscale do
     
   Blocks until the address is available.
   """
-  def ipv4_addr(dev), do: Tailscale.Native.ipv4_addr(dev)
+  def ipv4_addr(dev), do: Tailscale.Util.await(Tailscale.Native.ipv4_addr(dev))
 
   @spec ipv6_addr(t()) :: {:ok, :inet.ip6_address()} | {:error, any()}
   @doc """
@@ -108,13 +112,13 @@ defmodule Tailscale do
   Note that this address is in `t::inet.ip6_address/0` format (16-bit segments), which may be
   difficult to read. See `:inet.ntoa/1` to format to a string.
   """
-  def ipv6_addr(dev), do: Tailscale.Native.ipv6_addr(dev)
+  def ipv6_addr(dev), do: Tailscale.Util.await(Tailscale.Native.ipv6_addr(dev))
 
   @spec self_node(t()) :: {:ok, Tailscale.NodeInfo.t()} | {:error, any()}
   @doc """
   Get this node's `m:Tailscale.NodeInfo`.
   """
-  defdelegate self_node(dev), to: Tailscale.Native
+  def self_node(dev), do: Tailscale.Util.await(Tailscale.Native.self_node(dev))
 
   @spec peer_by_name(t(), String.t()) :: {:ok, Tailscale.NodeInfo.t() | nil} | {:error, any()}
   @doc """
@@ -123,7 +127,8 @@ defmodule Tailscale do
   Returns `{:ok, nil}` if there was no such peer, and `{:error, reason}` if the lookup encountered
   an error.
   """
-  def peer_by_name(dev, name), do: Tailscale.Native.peer_by_name(dev, name)
+  def peer_by_name(dev, name),
+    do: Tailscale.Util.await(Tailscale.Native.peer_by_name(dev, name))
 
   @spec peer_by_tailnet_ip(t(), Tailscale.ip_addr()) ::
           {:ok, Tailscale.NodeInfo.t() | nil} | {:error, any()}
@@ -132,12 +137,14 @@ defmodule Tailscale do
 
   Returns `{:ok, nil}` if there was no such peer. `:error` if the lookup encountered an error.
   """
-  defdelegate peer_by_tailnet_ip(dev, ip), to: Tailscale.Native
+  def peer_by_tailnet_ip(dev, ip),
+    do: Tailscale.Util.await(Tailscale.Native.peer_by_tailnet_ip(dev, ip))
 
   @spec peers_with_route(t(), Tailscale.ip_addr()) ::
           {:ok, [Tailscale.NodeInfo.t()]} | {:error, any()}
   @doc """
   Retrieve the most narrow set of peers that accept packets for the specified IP.
   """
-  defdelegate peers_with_route(dev, ip), to: Tailscale.Native
+  def peers_with_route(dev, ip),
+    do: Tailscale.Util.await(Tailscale.Native.peers_with_route(dev, ip))
 end

ts_elixir/lib/tailscale.ex

@@ -32,14 +32,39 @@ defmodule Tailscale.Native do
   """
   @opaque tcp_stream :: reference()
 
+  @typedoc """
+  NIFs provided here may have asynchronous effects that would typically block and require the use of
+  the DirtyIO scheduler. This is undesirable as we may have a large number of concurrent calls into
+  the NIFs, which could exhaust the DirtyIO thread pool. Instead, we use message passing on the Rust
+  side to send replies back into the BEAM. Functions that use this model return `async_reply`
+  without blocking. The `:async` case means the reply will be sent asynchronously using a message of
+  the format `{:tailscale, REF, PAYLOAD}`, where `REF` is the reference associated with the `:async`
+  response, guaranteed unique per call.
+
+  The `:error` response means that an error was encountered before dispatching the asynchronous
+  call.
+
+  The `:nif_panic` response means that the NIF panicked during execution; the second parameter is
+  the reason for the panic (if given).
+
+  `{:raise, TERM}` means `TERM` should be raised as an exception.
+
+  `m:Tailscale.Util` has helpers for decoding messages of this form.
+  """
+  @type async_reply() ::
+          {:async, reference()}
+          | {:error, any()}
+          | {:nif_panic, String.t() | {}}
+          | {:raise, any()}
+
   defp err, do: :erlang.nif_error(:nif_not_loaded)
 
   @doc """
   Open a new tailnet connection.
 
   See `t:Tailscale.options/0` for details on what options are supported.
   """
-  @spec connect(%{}) :: {:ok, device()} | {:error, any()}
+  @spec connect(%{}) :: async_reply()
   def connect(_opts), do: err()
 
   @doc """
@@ -51,7 +76,7 @@ defmodule Tailscale.Native do
   - `port`: the port to which the socket should bind.
   """
   @spec udp_bind(device(), Tailscale.ip_addr() | :ip4 | :ip6, :inet.port_number()) ::
-          {:ok, udp_socket()} | {:error, any()}
+          async_reply()
   def udp_bind(_dev, _addr, _port), do: err()
 
   @doc """
@@ -65,14 +90,14 @@ defmodule Tailscale.Native do
   - `msg`: the packet to send.
   """
   @spec udp_send(udp_socket(), Tailscale.ip_addr(), :inet.port_number(), binary()) ::
-          :ok | {:error, any()}
+          async_reply()
   def udp_send(_sock, _ip, _port, _msg), do: err()
 
   @doc """
   Receive an incoming UDP packet on the given socket.
   """
   @spec udp_recv(udp_socket()) ::
-          {:ok, :inet.ip_address(), :inet.port_number(), binary()} | {:error, any()}
+          async_reply()
   def udp_recv(_sock), do: err()
 
   @doc """
@@ -92,7 +117,7 @@ defmodule Tailscale.Native do
   Start a TCP listener on the given device, address, and port.
   """
   @spec tcp_listen(device(), Tailscale.ip_addr() | :ip4 | :ip6, :inet.port_number()) ::
-          {:ok, tcp_listener()} | {:error, any()}
+          async_reply()
   def tcp_listen(_dev, _addr, _port), do: err()
 
   @doc """
@@ -105,13 +130,13 @@ defmodule Tailscale.Native do
   Connect to the given TCP endpoint using the given device.
   """
   @spec tcp_connect(device(), Tailscale.ip_addr(), :inet.port_number()) ::
-          {:ok, tcp_stream()} | {:error, any()}
+          async_reply()
   def tcp_connect(_dev, _addr, _port), do: err()
 
   @doc """
   Accept an incoming TCP connection. Blocks until one is available.
   """
-  @spec tcp_accept(tcp_listener()) :: {:ok, tcp_stream()} | {:error, any()}
+  @spec tcp_accept(tcp_listener()) :: async_reply()
   def tcp_accept(_listener), do: err()
 
   @doc """
@@ -120,13 +145,13 @@ defmodule Tailscale.Native do
 
   Returns the number of bytes actually written to the remote.
   """
-  @spec tcp_send(tcp_stream(), binary()) :: {:ok, integer()} | {:error, any()}
+  @spec tcp_send(tcp_stream(), binary()) :: async_reply()
   def tcp_send(_stream, _msg), do: err()
 
   @doc """
   Receive incoming data from the tcp socket, blocking until at least one byte can be received.
   """
-  @spec tcp_recv(tcp_stream()) :: {:ok, binary()} | {:error, any()}
+  @spec tcp_recv(tcp_stream()) :: async_reply()
   def tcp_recv(_stream), do: err()
 
   @doc """
@@ -146,44 +171,76 @@ defmodule Tailscale.Native do
 
   Blocks until the device is connected and gets its address from control.
   """
-  @spec ipv4_addr(device()) :: {:ok, :inet.ip4_address()} | {:error, any()}
+  @spec ipv4_addr(device()) :: async_reply()
   def ipv4_addr(_dev), do: err()
 
   @doc """
   Retrieve the IPv6 address for the given tailscale device.
 
   Blocks until the device is connected and gets its address from control.
   """
-  @spec ipv6_addr(device()) :: {:ok, :inet.ip6_address()} | {:error, any()}
+  @spec ipv6_addr(device()) :: async_reply()
   def ipv6_addr(_dev), do: err()
 
   @doc """
   Retrieve a peer by name.
   """
-  @spec peer_by_name(device(), String.t()) :: {:ok, %{} | nil} | {:error, any()}
+  @spec peer_by_name(device(), String.t()) :: async_reply()
   def peer_by_name(_dev, _name), do: err()
 
   @doc """
   Retrieve this node's info
   """
-  @spec self_node(device()) :: {:ok, %{}} | {:error, any()}
+  @spec self_node(device()) :: async_reply()
   def self_node(_dev), do: err()
 
   @doc """
   Retrieve a peer by its tailnet IP.
   """
-  @spec peer_by_tailnet_ip(device(), Tailscale.ip_addr()) :: {:ok, %{} | nil} | {:error, any()}
+  @spec peer_by_tailnet_ip(device(), Tailscale.ip_addr()) :: async_reply()
   def peer_by_tailnet_ip(_dev, _ip), do: err()
 
   @doc """
   Retrieve the most narrow set of peers that accept packets for the specified IP.
   """
-  @spec peers_with_route(device(), Tailscale.ip_addr()) :: {:ok, [%{}]} | {:error, any()}
+  @spec peers_with_route(device(), Tailscale.ip_addr()) :: async_reply()
   def peers_with_route(_dev, _ip), do: err()
 
   @doc """
   Load key state from the specified path, generating a new state if the file doesn't exist.
   """
-  @spec load_key_file(String.t()) :: {:ok, Tailscale.Keystate.t()} | {:error, any()}
+  @spec load_key_file(String.t()) :: async_reply()
   def load_key_file(_path), do: err()
+
+  @doc """
+  Raise a `:badarg` exception.
+  """
+  @spec raise_badarg() :: nil
+  def raise_badarg(), do: err()
+
+  if @testing_nifs do
+    @doc """
+    DEV ONLY: trigger an async panic in the Rust code with the given message (if provided).
+    """
+    @spec async_panic(String.t() | nil) :: async_reply()
+    def async_panic(_msg \\ nil), do: err()
+
+    @doc """
+    DEV ONLY: trigger a raised exception in the Rust code with the given message.
+    """
+    @spec async_raise(String.t(), boolean()) :: async_reply()
+    def async_raise(_msg, _atom \\ false), do: err()
+
+    @doc """
+    DEV ONLY: trigger an asynchronous error in the Rust code with the given message.
+    """
+    @spec async_error(String.t(), boolean()) :: async_reply()
+    def async_error(_msg, _atom \\ false), do: err()
+
+    @doc """
+    DEV ONLY: trigger an asynchronous `:badarg` in the Rust code with the given message.
+    """
+    @spec async_badarg() :: async_reply()
+    def async_badarg(), do: err()
+  end
 end

ts_elixir/lib/tailscale/native.ex

@@ -1,4 +1,6 @@
 defmodule Tailscale.Tcp do
+  require Tailscale.Util
+
   @moduledoc """
   Functionality to create tailscale TCP sockets.
 
@@ -19,7 +21,7 @@ defmodule Tailscale.Tcp do
   @spec listen(Tailscale.t(), Tailscale.ip_addr() | :ip4 | :ip6, :inet.port_number()) ::
           {:ok, Tailscale.Tcp.Listener.t()} | {:error, any()}
   def listen(dev, addr, port) do
-    Tailscale.Native.tcp_listen(dev, addr, port)
+    Tailscale.Util.await(Tailscale.Native.tcp_listen(dev, addr, port))
   end
 
   @doc """
@@ -28,6 +30,6 @@ defmodule Tailscale.Tcp do
   @spec connect(Tailscale.t(), Tailscale.ip_addr(), :inet.port_number()) ::
           {:ok, Tailscale.Tcp.Stream.t()} | {:error, any()}
   def connect(dev, addr, port) do
-    Tailscale.Native.tcp_connect(dev, addr, port)
+    Tailscale.Util.await(Tailscale.Native.tcp_connect(dev, addr, port))
   end
 end

ts_elixir/lib/tailscale/tcp.ex

@@ -1,4 +1,6 @@
 defmodule Tailscale.Tcp.Listener do
+  require Tailscale.Util
+
   @moduledoc """
   Tailscale TCP listening socket functionality.
   """
@@ -15,7 +17,7 @@ defmodule Tailscale.Tcp.Listener do
   Blocks until a connection is ready.
   """
   def accept(res) do
-    Tailscale.Native.tcp_accept(res)
+    Tailscale.Util.await(Tailscale.Native.tcp_accept(res))
   end
 
   @doc """

ts_elixir/lib/tailscale/tcp/listener.ex

@@ -3,6 +3,8 @@ defmodule Tailscale.Tcp.Stream do
   Tailscale TCP sockets (connected).
   """
 
+  require Tailscale.Util
+
   @typedoc """
   A handle to a TCP stream (connected socket).
   """
@@ -15,7 +17,7 @@ defmodule Tailscale.Tcp.Stream do
   Returns the number of bytes actually sent.
   """
   def send(res, msg) do
-    Tailscale.Native.tcp_send(res, msg)
+    Tailscale.Util.await(Tailscale.Native.tcp_send(res, msg))
   end
 
   @spec send_all(t(), binary()) :: :ok | {:error, any()}
@@ -27,7 +29,7 @@ defmodule Tailscale.Tcp.Stream do
 
     case Tailscale.Tcp.Stream.send(res, msg) do
       {:ok, ^len} -> :ok
-      {:ok, n} -> Tailscale.Tcp.Stream.send_all(res, binary_slice(msg, n..len))
+      {:ok, n} -> send_all(res, binary_slice(msg, n..len))
       err -> err
     end
   end
@@ -37,7 +39,7 @@ defmodule Tailscale.Tcp.Stream do
   Receive data from the TCP socket, blocking until at least one byte can be received.
   """
   def recv(res) do
-    Tailscale.Native.tcp_recv(res)
+    Tailscale.Util.await(Tailscale.Native.tcp_recv(res))
   end
 
   @spec local_addr(t()) :: {:inet.ip_address(), :inet.port_number()}