defmodule Rx.Runtime do @moduledoc false use GenServer def start_link(opts \\ []) do GenServer.start_link(__MODULE__, opts, name: __MODULE__) end def system_init(config) do GenServer.call(__MODULE__, {:system_init, config}, :infinity) |> handle_init_reply() end def use_backend(config) do GenServer.call(__MODULE__, {:use_backend, config}, :infinity) |> handle_init_reply() end def ensure_system_init(config) do GenServer.call(__MODULE__, {:ensure_system_init, config}, :infinity) |> handle_init_reply() end defp handle_init_reply(reply) do case reply do :ok -> :ok {:error, %_{} = exception} -> raise exception {:error, {:backend_not_switchable, message}} when is_binary(message) -> raise RuntimeError, message {:error, reason} -> raise RuntimeError, "R backend initialization failed: #{inspect(reason)}" end end def eval(source, globals, opts) do case GenServer.call(__MODULE__, {:eval, source, globals, opts}, :infinity) do {:ok, eval_payload} -> wrap_eval_result(eval_payload, opts) {:error, :backend_crashed} -> raise RuntimeError, "R backend crashed" {:error, :backend_not_running} -> raise RuntimeError, "R backend not running" {:error, {:backend_exit, _}} -> raise RuntimeError, "R backend crashed" {:error, %ArgumentError{} = error} -> raise error {:error, %Rx.Error{} = error} -> raise error {:error, error_map} when is_map(error_map) -> raise_r_error(error_map) {:error, reason} -> raise RuntimeError, "R eval failed: #{inspect(reason)}" end end def plot(source, globals, opts) do case GenServer.call(__MODULE__, {:plot, source, globals, opts}, :infinity) do {:ok, plot_payload} -> wrap_plot_result(plot_payload, opts) {:error, :backend_crashed} -> raise RuntimeError, "R backend crashed" {:error, :backend_not_running} -> raise RuntimeError, "R backend not running" {:error, {:backend_exit, _}} -> raise RuntimeError, "R backend crashed" {:error, %ArgumentError{} = error} -> raise error {:error, %Rx.Error{} = error} -> raise error {:error, error_map} when is_map(error_map) -> raise_r_error(error_map) {:error, reason} -> raise RuntimeError, "R plot failed: #{inspect(reason)}" end end def encode(type, value) do case GenServer.call(__MODULE__, {:encode, type, value}, :infinity) do {:ok, %Rx.Object{} = object} -> object %Rx.Object{} = object -> object {:error, reason} -> raise ArgumentError, "failed to encode R value: #{inspect(reason)}" end end def print(%Rx.Object{} = object, opts) do case GenServer.call(__MODULE__, {:print, object, opts}, :infinity) do {:ok, output} -> wrap_print_result(output, opts) {:error, :backend_crashed} -> raise RuntimeError, "R backend crashed" {:error, :backend_not_running} -> raise RuntimeError, "R backend not running" {:error, {:backend_exit, _}} -> raise RuntimeError, "R backend crashed" {:error, %ArgumentError{} = error} -> raise error {:error, %Rx.Error{} = error} -> raise error {:error, error_map} when is_map(error_map) -> raise_r_error(error_map) {:error, reason} -> raise RuntimeError, "R print failed: #{inspect(reason)}" end end def decode(%Rx.Object{} = object) do if inline_object?(object) do decode_inline_object(object) else decode_backend_object(object) end end defp decode_backend_object(%Rx.Object{} = object) do case GenServer.call(__MODULE__, {:decode, object}, :infinity) do {:ok, value} -> normalize_decoded(value) {:opaque, object} -> object {:error, reason} -> raise ArgumentError, "failed to decode R value: #{inspect(reason)}" end end defp decode_inline_object(%Rx.Object{} = object) do case Rx.Backends.PortArrow.decode(object) do {:ok, value} -> value |> normalize_decoded() |> restore_inline_source_values(object) {:opaque, object} -> object {:error, reason} -> raise ArgumentError, "failed to decode R value: #{inspect(reason)}" end end defp restore_inline_source_values(decoded, %Rx.Object{ remote_info: {:inline_named_list, source_map} }) when is_map(source_map) and not is_struct(source_map) do restore_inline_source_value(decoded, source_map) end defp restore_inline_source_values(decoded, %Rx.Object{}), do: decoded defp restore_inline_source_value(decoded, source_map) when is_map(source_map) and not is_struct(source_map) do Map.new(decoded, fn {key, value} -> {key, restore_inline_source_value(value, Map.get(source_map, key))} end) end defp restore_inline_source_value( _decoded, %Rx.Object{id: <<"__inline__:", _json::binary>>} = object ), do: decode_inline_object(object) defp restore_inline_source_value(_decoded, %Rx.Object{} = object), do: object defp restore_inline_source_value(decoded, _source), do: decoded defp inline_object?(%Rx.Object{id: "__null__"}), do: true defp inline_object?(%Rx.Object{id: <<"__inline__:", _json::binary>>}), do: true defp inline_object?(%Rx.Object{}), do: false def decode_arrow(%Rx.Object{} = object) do case GenServer.call(__MODULE__, {:decode_arrow, object}, :infinity) do {:ok, binary} -> {:ok, binary} {:error, reason} -> {:error, reason} end end def decode_data_frame(%Rx.Object{} = object, opts) when is_list(opts) do case GenServer.call(__MODULE__, {:decode_data_frame, object, opts}, :infinity) do {:ok, wire} -> {:ok, wire} {:error, reason} -> {:error, reason} end end def encode_dataframe(ipc_bytes) when is_binary(ipc_bytes) do case GenServer.call(__MODULE__, {:encode_dataframe, ipc_bytes}, :infinity) do {:ok, %Rx.Object{} = obj} -> {:ok, obj} {:error, reason} -> {:error, reason} end end def encode_data_frame(wire, opts) when is_map(wire) and is_list(opts) do case GenServer.call(__MODULE__, {:encode_data_frame, wire, opts}, :infinity) do {:ok, %Rx.Object{} = obj} -> {:ok, obj} {:error, reason} -> {:error, reason} end end def backend do GenServer.call(__MODULE__, :backend) end @impl true def init(_opts) do {:ok, %{init: nil, backend: nil}} end @impl true def handle_call({:system_init, config}, _from, %{init: nil} = state) do {reply, state} = init_backend(config, state) {:reply, reply, state} end def handle_call({:use_backend, config}, _from, %{init: nil} = state) do {reply, state} = init_backend(config, state) {:reply, reply, state} end def handle_call({:use_backend, config}, _from, %{init: _existing} = state) do if state.init == init_identity(config) do {reply, state} = init_backend(config, state) {:reply, reply, state} else case shutdown_backend(state.backend) do :ok -> {reply, state} = init_backend(config, %{state | init: nil, backend: nil}) {:reply, reply, state} {:error, reason} -> {:reply, {:error, reason}, state} end end end def handle_call({:ensure_system_init, config}, _from, %{init: nil} = state) do configs = List.wrap(config) {reply, state} = ensure_first_available(configs, state) {:reply, reply, state} end def handle_call({:ensure_system_init, _config}, _from, %{init: existing} = state) do # Re-run system_init if the backend process died since last init. case safe_backend_call(fn -> state.backend.system_init(existing) end) do :ok -> {:reply, :ok, state} {:error, reason} -> {:reply, {:error, reason}, state} end end def handle_call({:system_init, config}, _from, %{init: existing} = state) do if existing == init_identity(config) do # Re-run system_init if the backend process died since last init. {reply, state} = init_backend(config, state) {:reply, reply, state} else {:reply, {:error, RuntimeError.exception("Rx is already initialized with different options")}, state} end end def handle_call(:backend, _from, state) do {:reply, backend_name(state.backend), state} end def handle_call({:eval, _source, _globals, _opts}, _from, %{backend: nil} = state) do {:reply, {:error, :backend_not_running}, state} end def handle_call({:eval, source, globals, opts}, _from, state) do case safe_backend_call(fn -> state.backend.eval(source, globals, opts) end) do {:ok, payload} -> {:reply, {:ok, payload}, state} {:error, :backend_crashed} -> {:reply, {:error, :backend_crashed}, state} {:error, :backend_not_running} -> {:reply, {:error, :backend_not_running}, state} {:error, {:backend_exit, _} = reason} -> {:reply, {:error, reason}, state} {:error, reason} -> {:reply, {:error, reason}, state} end end def handle_call({:plot, _source, _globals, _opts}, _from, %{backend: nil} = state) do {:reply, {:error, :backend_not_running}, state} end def handle_call({:plot, source, globals, opts}, _from, state) do case safe_backend_call(fn -> state.backend.plot(source, globals, opts) end) do {:ok, payload} -> {:reply, {:ok, payload}, state} {:error, :backend_crashed} -> {:reply, {:error, :backend_crashed}, state} {:error, :backend_not_running} -> {:reply, {:error, :backend_not_running}, state} {:error, {:backend_exit, _} = reason} -> {:reply, {:error, reason}, state} {:error, reason} -> {:reply, {:error, reason}, state} end end def handle_call({:encode, _type, _value}, _from, %{backend: nil} = state) do {:reply, {:error, :backend_not_running}, state} end def handle_call({:encode, type, value}, _from, state) do {:reply, safe_backend_call(fn -> state.backend.encode(type, value) end), state} end def handle_call({:print, %Rx.Object{}, _opts}, _from, %{backend: nil} = state) do {:reply, {:error, :backend_not_running}, state} end def handle_call({:print, %Rx.Object{} = object, opts}, _from, state) do {:reply, safe_backend_call(fn -> state.backend.print(object, opts) end), state} end def handle_call({:decode, %Rx.Object{}}, _from, %{backend: nil} = state) do {:reply, {:error, :backend_not_running}, state} end def handle_call({:decode, %Rx.Object{} = object}, _from, state) do case safe_backend_call(fn -> state.backend.decode(object) end) do {:ok, value} -> {:reply, {:ok, value}, state} {:opaque, obj} -> {:reply, {:opaque, obj}, state} {:error, reason} -> {:reply, {:error, reason}, state} end end def handle_call({:decode_arrow, %Rx.Object{}}, _from, %{backend: nil} = state) do {:reply, {:error, :backend_not_running}, state} end def handle_call({:decode_arrow, %Rx.Object{} = object}, _from, state) do result = route_decode_arrow(object, state) {:reply, result, state} end def handle_call({:decode_data_frame, %Rx.Object{}, _opts}, _from, %{backend: nil} = state) do {:reply, {:error, :backend_not_running}, state} end def handle_call({:decode_data_frame, %Rx.Object{} = object, opts}, _from, state) do result = route_decode_data_frame(object, opts, state) {:reply, result, state} end def handle_call({:encode_dataframe, _ipc_bytes}, _from, %{backend: nil} = state) do {:reply, {:error, :backend_not_running}, state} end def handle_call({:encode_dataframe, ipc_bytes}, _from, state) when is_binary(ipc_bytes) do result = route_encode_dataframe(ipc_bytes, state) {:reply, result, state} end def handle_call({:encode_data_frame, _wire, _opts}, _from, %{backend: nil} = state) do {:reply, {:error, :backend_not_running}, state} end def handle_call({:encode_data_frame, wire, opts}, _from, state) when is_map(wire) and is_list(opts) do result = route_encode_data_frame(wire, opts, state) {:reply, result, state} end defp route_decode_arrow(%Rx.Object{} = object, state) do if native_backend?(state.backend) do route_native_decode_arrow(object, state) else safe_backend_call(fn -> state.backend.decode_arrow(object) end) end end defp route_native_decode_arrow(%Rx.Object{} = object, state) do case object.backend do :port_arrow -> with :ok <- ensure_dataframe_process_backend(state) do safe_backend_call(fn -> dataframe_process_backend().decode_arrow(object) end) end _other -> safe_backend_call(fn -> state.backend.decode_arrow(object) end) end end defp route_decode_data_frame(%Rx.Object{} = object, opts, state) do if native_backend?(state.backend) do route_native_decode_data_frame(object, opts, state) else safe_backend_call(fn -> state.backend.decode_data_frame(object, opts) end) end end defp route_native_decode_data_frame(%Rx.Object{backend: :port_arrow} = object, opts, state) do with :ok <- ensure_dataframe_process_backend(state) do safe_backend_call(fn -> dataframe_process_backend().decode_data_frame(object, opts) end) end end defp route_native_decode_data_frame(%Rx.Object{} = object, opts, state) do safe_backend_call(fn -> state.backend.decode_data_frame(object, opts) end) end defp route_encode_dataframe(ipc_bytes, state) do if native_backend?(state.backend) do route_native_encode_dataframe(ipc_bytes, state) else safe_backend_call(fn -> state.backend.encode_dataframe(ipc_bytes) end) end end defp route_native_encode_dataframe(ipc_bytes, state) do safe_backend_call(fn -> state.backend.encode_dataframe(ipc_bytes) end) end defp route_encode_data_frame(wire, opts, state) do safe_backend_call(fn -> state.backend.encode_data_frame(wire, opts) end) end defp ensure_dataframe_process_backend(state) do config = dataframe_process_config(state) safe_backend_call(fn -> dataframe_process_backend().system_init(config) end) end defp dataframe_process_config(%{init: init}) when is_list(init) do config = [backend: dataframe_process_backend()] case Keyword.fetch(init, :lib_paths) do {:ok, lib_paths} -> Keyword.put(config, :lib_paths, lib_paths) :error -> config end end defp dataframe_process_config(_state), do: [backend: dataframe_process_backend()] defp dataframe_process_backend do Application.get_env(:rx, :dataframe_process_backend, Rx.Backends.PortArrow) end defp native_backend?(backend) do backend == Application.get_env(:rx, :runtime_native_backend, Rx.Backends.Native) end defp ensure_first_available([], state), do: {{:error, :backend_not_running}, state} defp ensure_first_available([config | rest], state) do case init_backend(config, state) do {:ok, state} -> {:ok, state} {{:error, reason}, _state} when rest != [] -> if continue_init_fallback?(config, reason) do ensure_first_available(rest, state) else {{:error, reason}, state} end {{:error, reason}, _state} -> {{:error, reason}, state} end end defp continue_init_fallback?(config, reason) do backend = Keyword.fetch!(config, :backend) if native_backend?(backend) do safe_native_init_fallback_reason?(reason) else true end end defp safe_native_init_fallback_reason?({:embedded_nif_unavailable, _message}), do: true defp safe_native_init_fallback_reason?({:not_loaded, _message}), do: true defp safe_native_init_fallback_reason?(:missing_r_home), do: true defp safe_native_init_fallback_reason?(:missing_lib_r), do: true defp safe_native_init_fallback_reason?({:native_init_failed, %{retryable: true}}), do: true defp safe_native_init_fallback_reason?({:native_init_failed, diagnostics}) when is_map(diagnostics), do: false defp safe_native_init_fallback_reason?({:native_init_mismatch, _diagnostics}), do: false defp safe_native_init_fallback_reason?(_reason), do: false defp init_backend(config, state) do backend = Keyword.fetch!(config, :backend) case safe_backend_call(fn -> backend.system_init(config) end) do :ok -> {:ok, %{state | init: init_identity(config), backend: backend}} {:error, reason} -> {{:error, reason}, state} end end defp shutdown_backend(nil), do: :ok defp shutdown_backend(backend) do if function_exported?(backend, :shutdown, 0) do safe_backend_call(fn -> backend.shutdown() end) else {:error, {:backend_not_switchable, "#{inspect(backend)} cannot be switched because it does not expose shutdown/0"}} end end defp safe_backend_call(fun) do fun.() rescue error in [ArgumentError, RuntimeError, Rx.Error] -> {:error, error} catch :exit, {:noproc, _} -> {:error, :backend_not_running} :exit, {:normal, _} -> {:error, :backend_not_running} :exit, reason -> {:error, {:backend_exit, reason}} end defp backend_name(nil), do: nil defp backend_name(Rx.Backends.PortArrow), do: :port_arrow defp backend_name(Rx.Backends.Native), do: :native defp backend_name(backend), do: backend defp init_identity(config) when is_list(config) do Keyword.update(config, :renv, nil, &renv_identity/1) end defp init_identity(config), do: config defp renv_identity(nil), do: nil defp renv_identity(%{identity_env: identity_env} = renv) do %{renv | env: identity_env} end defp renv_identity(%{env: env} = renv) when is_list(env) do %{renv | env: Enum.sort_by(env, fn {name, value} -> {name, value} end)} end defp renv_identity(renv), do: renv defp wrap_eval_result({result, globals, output}, opts) do if opts[:capture] do %Rx.EvalResult{ result: result, globals: globals, stdout: output.stdout, messages: output.messages, warnings: output.warnings } else write_output!(output, opts) {result, globals} end end defp wrap_print_result(output, opts) do if opts[:capture] do %Rx.PrintResult{ stdout: output.stdout, messages: output.messages, warnings: output.warnings } else output.stdout end end defp wrap_plot_result(%{plots: plots, output: output}, opts) do if opts[:capture] do %Rx.PlotResult{ plots: plots, stdout: output.stdout, messages: output.messages, warnings: output.warnings } else write_output!(output, opts) plots end end defp write_output!(output, opts) do unless output.stdout == "", do: IO.write(opts[:stdout_device], output.stdout) stderr = opts[:stderr_device] unless output.messages == "", do: IO.write(stderr, output.messages) unless output.warnings == "", do: IO.write(stderr, output.warnings) end defp raise_r_error(%{message: message} = error_map) do raise Rx.Error, message: message, r_class: normalize_r_character_vector(Map.get(error_map, :r_class)), call: normalize_r_character_scalar(Map.get(error_map, :call)), traceback: normalize_r_character_vector(Map.get(error_map, :traceback)), output: Map.get(error_map, :output) end defp normalize_r_character_scalar(nil), do: nil defp normalize_r_character_scalar([value | _rest]) when is_binary(value), do: value defp normalize_r_character_scalar(value) when is_binary(value), do: value defp normalize_r_character_scalar(value), do: inspect(value) defp normalize_r_character_vector(nil), do: nil defp normalize_r_character_vector(values) when is_list(values), do: values defp normalize_r_character_vector(value) when is_binary(value), do: [value] defp normalize_r_character_vector(value), do: [inspect(value)] defp normalize_decoded({:na, type}), do: %Rx.NA{type: type} defp normalize_decoded(values) when is_list(values), do: Enum.map(values, &normalize_decoded/1) defp normalize_decoded(%Rx.RList{items: items} = rlist) do %{rlist | items: Enum.map(items, fn {k, v} -> {k, normalize_decoded(v)} end)} end defp normalize_decoded(value) when is_map(value) and not is_struct(value) do Map.new(value, fn {k, v} -> {k, normalize_decoded(v)} end) end defp normalize_decoded(value), do: value end