alias Maru.Entity.Struct.Batch alias Maru.Entity.Struct.Instance alias Maru.Entity.Struct.Serializer alias Maru.Entity.Struct.Exposure defmodule Maru.Entity.Runtime do @moduledoc """ Serialize an instance concurrently. """ @type id :: reference @type state :: %{ data: :ets.tab, old_link: :ets.tab, new_link: :ets.tab, old_batch: :ets.tab, new_batch: :ets.tab, max_concurrency: pos_integer, } @doc false @spec serialize(Serializer.t, Maru.Entity.instance, Keyword.t) :: Maru.Entity.object def serialize(serializer, instance, options) do state = init(options) id = save_link(serializer, instance, state.old_link) state = do_loop(state) terminate(id, serializer.type, state) end @spec init(Keyword.t) :: state defp init(options) do max_concurrency = Keyword.get(options, :max_concurrency) || Application.get_env(:maru_entity, :default_max_concurrency) || 4 %{ data: create_ets(:duplicate_bag), old_link: create_ets(:duplicate_bag), new_link: create_ets(:duplicate_bag), old_batch: create_ets(:set), new_batch: create_ets(:set), max_concurrency: max_concurrency, } end @spec terminate(id, Maru.Entity.one_or_many, state) :: Maru.Entity.object defp terminate(id, type, state) do :ets.delete(state.old_link) :ets.delete(state.new_link) :ets.delete(state.old_batch) :ets.delete(state.new_batch) data = do_build(id, type, state.data) :ets.delete(state.data) data end @spec create_ets(:set | :duplicate_bag) :: :ets.tab defp create_ets(type) when type in [:set, :duplicate_bag] do :ets.new(:maru_entity_serializer, [ type, :public, heir: :none, write_concurrency: true, read_concurrency: true ]) end @spec save_link(Serializer.t, Maru.Entity.instance, :ets.tab) :: id defp save_link(%Serializer{type: :one}=s, instance, ets) do id = make_ref() :ets.insert(ets, {id, s, instance, nil}) id end defp save_link(%Serializer{type: :many}=s, instances, ets) do id = make_ref() instances |> Stream.with_index |> Enum.each(fn {i, idx} -> :ets.insert(ets, {id, s, i, idx}) end) id end @spec do_build(id, Maru.Entity.one_or_many, :ets.tab) :: Maru.Entity.object | [Maru.Entity.object] defp do_build(id, :one, ets) do [{_id, i, nil}] = :ets.lookup(ets, id) do_build_one(i, ets) end defp do_build(id, :many, ets) do :ets.lookup(ets, id) |> Enum.sort(fn {_, _, idx1}, {_, _, idx2} -> idx1 < idx2 end) |> Enum.map(fn {_id, i, _idx} -> do_build_one(i, ets) end) end @spec do_build_one(Instance.t, :ets.tab) :: Maru.Entity.object defp do_build_one(%Instance{data: data, links: []}, _ets), do: data defp do_build_one(%Instance{data: data, links: links}, ets) do Enum.reduce(links, data, fn {attr_group, type, id}, acc -> put_in(acc, attr_group, do_build(id, type, ets)) end) end @spec do_loop(state) :: state defp do_loop(state) do unless :ets.info(state.old_link)[:size] == 0 do do_loop_link_monitor(state) end unless :ets.info(state.old_batch)[:size] == 0 do :ets.delete_all_objects(state.old_link) do_loop_batch(state) do_loop_link_monitor(state) end case :ets.info(state.new_link)[:size] + :ets.info(state.new_batch)[:size] do 0 -> state _ -> :ets.delete_all_objects(state.old_link) :ets.delete_all_objects(state.old_batch) %{ state | old_link: state.new_link, new_link: state.old_link, old_batch: state.new_batch, new_batch: state.old_batch, } |> do_loop end end @spec do_loop_link_monitor(state) :: :ok defp do_loop_link_monitor(state) do parent = self() {pid, ref} = Process.spawn(fn -> do_loop_link(parent, state) end, [:monitor]) receive do {:DOWN, ^ref, :process, ^pid, :normal} -> :ok {:DOWN, ^ref, :process, ^pid, reason} -> {:error, exception, stack} = reason :erlang.raise(:error, exception, stack) end end @spec do_loop_link(pid, state) :: :ok defp do_loop_link(parent, state) do Process.flag(:trap_exit, true) parent_ref = Process.monitor(parent) sync = fn(continue) -> receive do {:EXIT, _pid, :normal} -> continue.() {:EXIT, _pid, reason} -> exit(reason) {:DOWN, ^parent_ref, :process, ^parent, _} -> exit(:kill) end end {_, rest} = :ets.foldl(fn term, {0, rest} -> sync.(fn -> Process.spawn(__MODULE__, :do_serialize, [term, state], [:link]) end) {0, rest} term, {num, rest} -> Process.spawn(__MODULE__, :do_serialize, [term, state], [:link]) {num - 1, rest + 1} end, {state.max_concurrency, 0}, state.old_link) for _ <- 1..rest do sync.(fn -> :ok end) end :ok end @spec do_loop_batch(state) :: :ok defp do_loop_batch(state) do data = :ets.foldl(fn {id, serializer, %Batch{module: module, key: key}}, acc -> case acc do %{^module => value} -> value = Map.put(value, key, {id, serializer}) Map.put(acc, module, value) _ -> Map.put(acc, module, %{ key => {id, serializer} }) end end, %{}, state.old_batch) |> Enum.map(fn {module, value} -> {module, Task.async(module, :resolve, [Map.keys(value)])} end) |> Enum.map(fn {module, task} -> {module, Task.await(task)} end) |> Enum.into(%{}) :ets.foldl(fn {id, %Serializer{type: :many}=s, %Batch{module: module, key: key}}, _ -> data |> Map.get(module, %{}) |> Map.get(key, []) |> Stream.with_index |> Enum.each(fn {instance, idx} -> :ets.insert(state.old_link, {id, %{s | type: :one}, instance, idx}) end) :ok {id, %Serializer{type: :one}=s, %Batch{module: module, key: key}}, _ -> instance = get_in(data, [module, key]) :ets.insert(state.old_link, {id, s, instance, nil}) :ok {id, nil, %Batch{module: module, key: key}}, _ -> instance = get_in(data, [module, key]) :ets.insert(state.old_link, {id, nil, instance, nil}) :ok end, :ok, state.old_batch) end @doc false @spec do_serialize({id, Serializer.t | nil, Maru.Entity.instance, integer}, state) :: :ok def do_serialize({id, nil, instance, idx}, state) do result = %Instance{data: instance, links: []} :ets.insert(state.data, {id, result, idx}) :ok end def do_serialize({id, serializer, instance, idx}, state) do exposures = serializer.module.__exposures__ result = do_serialize(exposures, %Instance{}, instance, serializer.options, state) :ets.insert(state.data, {id, result, idx}) :ok rescue e -> exit({:error, e, System.stacktrace()}) end @spec do_serialize(list(Exposure.Runtime.t), Maru.Entity.object, Maru.Entity.instance, Maru.Entity.options, state) :: Maru.Entity.object defp do_serialize([], result, _instance, _options, _state), do: result defp do_serialize([h | t], result, instance, options, state) do result = if h.if_func.(instance, options) do do_update(h.serializer, h.do_func.(instance, options), h, result, options, state) else result end do_serialize(t, result, instance, options, state) end @spec do_update(Serializer.t | nil, Batch.t | Maru.Entity.instance, Exposure.Runtime.t, Maru.Entity.instance, Maru.Entity.options, state) :: Maru.Entity.instance defp do_update(nil, %Batch{}=batch, field, result, _options, state) do id = make_ref() :ets.insert(state.new_batch, {id, nil, batch}) %Instance{ data: put_in(result.data, field.attr_group, nil), links: [{field.attr_group, :one, id} | result.links], } end defp do_update(serializer, %Batch{}=batch, field, result, options, state) do id = make_ref() s = %{serializer | options: options} :ets.insert(state.new_batch, {id, s, batch}) %Instance{ data: put_in(result.data, field.attr_group, nil), links: [{field.attr_group, serializer.type, id} | result.links], } end defp do_update(nil, instance, field, result, _options, _state) do %{ result | data: put_in(result.data, field.attr_group, instance), } end defp do_update(serializer, instance, field, result, options, state) do s = %{serializer | options: options} id = save_link(s, instance, state.new_link) %Instance{ data: put_in(result.data, field.attr_group, nil), links: [{field.attr_group, serializer.type, id} | result.links], } end end