perf(runner): introduce RuntimeCpnet to replace linear scans (audit #2) (#79)

Author: fahchen

lib/coloured_flow/enabled_binding_elements/computation.ex

@@ -5,14 +5,14 @@ defmodule ColouredFlow.EnabledBindingElements.Computation do
 
   alias ColouredFlow.Definition.Arc
   alias ColouredFlow.Definition.ColourSet
-  alias ColouredFlow.Definition.ColouredPetriNet
   alias ColouredFlow.Definition.Expression
   alias ColouredFlow.Definition.Place
   alias ColouredFlow.Definition.Transition
   alias ColouredFlow.EnabledBindingElements.Binding
   alias ColouredFlow.Enactment.BindingElement
   alias ColouredFlow.Enactment.Marking
   alias ColouredFlow.MultiSet
+  alias ColouredFlow.Runner.RuntimeCpnet
 
   import ColouredFlow.EnabledBindingElements.Utils
 
@@ -22,12 +22,12 @@ defmodule ColouredFlow.EnabledBindingElements.Computation do
   """
   @spec list(
           transition :: Transition.t(),
-          cpnet :: ColouredPetriNet.t(),
+          runtime_cpnet :: RuntimeCpnet.t(),
           markings :: %{Place.name() => Marking.t()}
         ) :: [BindingElement.t()]
-  def list(transition, cpnet, markings) do
-    inputs = get_arcs_with_place(transition, :p_to_t, cpnet)
-    constants = build_constants(cpnet)
+  def list(transition, %RuntimeCpnet{} = runtime_cpnet, markings) do
+    inputs = get_arcs_with_place(transition, :p_to_t, runtime_cpnet)
+    constants = runtime_cpnet.constants
 
     arc_bindings =
       inputs
@@ -46,35 +46,35 @@ defmodule ColouredFlow.EnabledBindingElements.Computation do
           Binding.match_bag(marking.tokens, arc_bind_expr)
         end)
       end)
-      |> reject_invalid_bandings(cpnet)
+      |> reject_invalid_bandings(runtime_cpnet)
 
     binding_combinations = Binding.combine(arc_bindings)
 
     Enum.flat_map(binding_combinations, fn binding ->
-      build_binding_element(inputs, binding, transition, constants, markings, cpnet)
+      build_binding_element(inputs, binding, transition, constants, markings, runtime_cpnet)
     end)
   end
 
-  defp build_binding_element(inputs, binding, transition, constants, markings, cpnet) do
-    case collect_consumption(inputs, binding, transition, constants, markings, cpnet) do
+  defp build_binding_element(inputs, binding, transition, constants, markings, runtime_cpnet) do
+    case collect_consumption(inputs, binding, transition, constants, markings, runtime_cpnet) do
       :error -> []
       to_consume -> [BindingElement.new(transition.name, binding, to_consume)]
     end
   end
 
-  defp collect_consumption(inputs, binding, transition, constants, markings, cpnet) do
+  defp collect_consumption(inputs, binding, transition, constants, markings, runtime_cpnet) do
     Enum.reduce_while(inputs, [], fn {arc, place}, acc ->
-      try_consume(arc, place, binding, transition, constants, markings, cpnet, acc)
+      try_consume(arc, place, binding, transition, constants, markings, runtime_cpnet, acc)
     end)
   end
 
-  defp try_consume(arc, place, binding, transition, constants, markings, cpnet, acc) do
+  defp try_consume(arc, place, binding, transition, constants, markings, runtime_cpnet, acc) do
     arc_binding = merge_constants(binding, arc, constants)
 
     with(
       {:ok, {coefficient, value}} <- eval_arc(arc, arc_binding),
-      colour_set = fetch_colour_set!(place.colour_set, cpnet),
-      of_type_context = build_of_type_context(cpnet),
+      colour_set = fetch_colour_set!(place.colour_set, runtime_cpnet),
+      of_type_context = runtime_cpnet.of_type_context,
       {:ok, ^value} <- ColourSet.Of.of_type(value, colour_set.type, of_type_context),
       guard_binding = merge_constants(binding, transition, constants),
       {:ok, true} <- eval_transition_guard(transition, guard_binding),
@@ -88,14 +88,14 @@ defmodule ColouredFlow.EnabledBindingElements.Computation do
     end
   end
 
-  @spec reject_invalid_bandings([[[BindingElement.binding()]]], ColouredPetriNet.t()) ::
+  @spec reject_invalid_bandings([[[BindingElement.binding()]]], RuntimeCpnet.t()) ::
           [[[BindingElement.binding()]]]
-  defp reject_invalid_bandings(bindings_list, cpnet) do
-    of_type_context = build_of_type_context(cpnet)
+  defp reject_invalid_bandings(bindings_list, %RuntimeCpnet{} = runtime_cpnet) do
+    of_type_context = runtime_cpnet.of_type_context
 
     valid_binding? = fn binding ->
       Enum.all?(binding, fn {name, value} ->
-        variable = fetch_variable!(name, cpnet)
+        variable = fetch_variable!(name, runtime_cpnet)
 
         match?(
           {:ok, _value},

lib/coloured_flow/enabled_binding_elements/occurrence.ex

@@ -6,12 +6,12 @@ defmodule ColouredFlow.EnabledBindingElements.Occurrence do
 
   alias ColouredFlow.Definition.Arc
   alias ColouredFlow.Definition.ColourSet
-  alias ColouredFlow.Definition.ColouredPetriNet
   alias ColouredFlow.Definition.Variable
   alias ColouredFlow.Enactment.BindingElement
   alias ColouredFlow.Enactment.Marking
   alias ColouredFlow.Enactment.Occurrence
   alias ColouredFlow.MultiSet
+  alias ColouredFlow.Runner.RuntimeCpnet
 
   import ColouredFlow.EnabledBindingElements.Utils
 
@@ -22,23 +22,23 @@ defmodule ColouredFlow.EnabledBindingElements.Occurrence do
   @spec occur(
           binding_element :: BindingElement.t(),
           free_binding :: [{Variable.name(), ColourSet.value()}],
-          cpnet :: ColouredPetriNet.t()
+          runtime_cpnet :: RuntimeCpnet.t()
         ) :: {:ok, Occurrence.t()} | {:error, [Exception.t()]}
-  def occur(binding_element, free_binding, cpnet) do
-    transition = fetch_transition!(binding_element.transition, cpnet)
+  def occur(binding_element, free_binding, %RuntimeCpnet{} = runtime_cpnet) do
+    transition = fetch_transition!(binding_element.transition, runtime_cpnet)
 
-    constants = build_constants(cpnet)
+    constants = runtime_cpnet.constants
     binding = Enum.concat(binding_element.binding, free_binding)
 
-    outputs = get_arcs_with_place(transition, :t_to_p, cpnet)
+    outputs = get_arcs_with_place(transition, :t_to_p, runtime_cpnet)
 
     {to_produce, exceptions} =
       Enum.map_reduce(outputs, [], fn {arc, place}, acc ->
         # place constants into binding
         binding = merge_constants(binding, arc, constants)
 
         with {:ok, result} <- ColouredFlow.Expression.eval(arc.expression.expr, binding),
-             {:ok, tokens} <- build_tokens(result, place.colour_set, cpnet) do
+             {:ok, tokens} <- build_tokens(result, place.colour_set, runtime_cpnet) do
           {%Marking{place: place.name, tokens: tokens}, acc}
         else
           {:error, exceptions} -> {:error, acc ++ exceptions}
@@ -70,15 +70,15 @@ defmodule ColouredFlow.EnabledBindingElements.Occurrence do
           # credo:disable-for-next-line JetCredo.Checks.ExplicitAnyType
           binding :: term(),
           colour_set :: ColourSet.name(),
-          cpnet :: ColouredPetriNet.t()
+          runtime_cpnet :: RuntimeCpnet.t()
         ) :: {:ok, MultiSet.t()} | {:error, [Exception.t()]}
-  defp build_tokens({coefficient, value}, colour_set, cpnet)
+  defp build_tokens({coefficient, value}, colour_set, %RuntimeCpnet{} = runtime_cpnet)
        when is_integer(coefficient) and coefficient >= 0 do
     alias ColouredFlow.Definition.ColourSet.ColourSetMismatch
     alias ColouredFlow.Definition.ColourSet.Of
 
-    colour_set = fetch_colour_set!(colour_set, %ColouredPetriNet{} = cpnet)
-    context = build_of_type_context(cpnet)
+    colour_set = fetch_colour_set!(colour_set, runtime_cpnet)
+    context = runtime_cpnet.of_type_context
 
     case Of.of_type(value, colour_set.type, context) do
       {:ok, value} ->

lib/coloured_flow/enabled_binding_elements/utils.ex

@@ -1,71 +1,48 @@
 defmodule ColouredFlow.EnabledBindingElements.Utils do
   @moduledoc """
   The utils for the enabled binding elements.
+
+  These helpers operate on a `ColouredFlow.Runner.RuntimeCpnet` — the indexed view
+  built from a raw `ColouredFlow.Definition.ColouredPetriNet` at the runner-call
+  boundary. Lookups are `O(1)` against the precomputed indexes.
   """
 
   alias ColouredFlow.Definition.Arc
   alias ColouredFlow.Definition.ColourSet
-  alias ColouredFlow.Definition.ColouredPetriNet
   alias ColouredFlow.Definition.Place
   alias ColouredFlow.Definition.Transition
   alias ColouredFlow.Definition.Variable
   alias ColouredFlow.Enactment.Marking
   alias ColouredFlow.MultiSet
+  alias ColouredFlow.Runner.RuntimeCpnet
 
-  @spec fetch_colour_set!(colour_set :: ColourSet.name(), cpnet :: ColouredPetriNet.t()) ::
+  @spec fetch_colour_set!(colour_set :: ColourSet.name(), runtime_cpnet :: RuntimeCpnet.t()) ::
           ColourSet.t()
-  def fetch_colour_set!(colour_set, %ColouredPetriNet{} = cpnet) do
-    Enum.find(
-      cpnet.colour_sets,
-      &match?(%ColourSet{name: ^colour_set}, &1)
-    ) || raise "Colour set with name #{inspect(colour_set)} not found in the petri net."
-  end
-
-  @spec build_of_type_context(cpnet :: ColouredPetriNet.t()) ::
-          ColouredFlow.Definition.ColourSet.Of.context()
-  def build_of_type_context(%ColouredPetriNet{} = cpnet) do
-    types =
-      Map.new(cpnet.colour_sets, fn %ColourSet{name: name, type: type} ->
-        {name, type}
-      end)
-
-    %{
-      fetch_type: fn name ->
-        case Map.fetch(types, name) do
-          :error -> raise "Colour set with name #{inspect(name)} not found in the petri net."
-          {:ok, type} -> {:ok, type}
-        end
-      end
-    }
+  def fetch_colour_set!(colour_set, %RuntimeCpnet{colour_sets: colour_sets}) do
+    case Map.fetch(colour_sets, colour_set) do
+      {:ok, cs} -> cs
+      :error -> raise "Colour set with name #{inspect(colour_set)} not found in the petri net."
+    end
   end
 
-  @spec fetch_variable!(variable :: Variable.name(), cpnet :: ColouredPetriNet.t()) ::
+  @spec fetch_variable!(variable :: Variable.name(), runtime_cpnet :: RuntimeCpnet.t()) ::
           Variable.t()
-  def fetch_variable!(variable, %ColouredPetriNet{} = cpnet) do
-    Enum.find(
-      cpnet.variables,
-      &match?(%Variable{name: ^variable}, &1)
-    ) || raise "Variable with name #{inspect(variable)} not found in the petri net."
+  def fetch_variable!(variable, %RuntimeCpnet{variables: variables}) do
+    case Map.fetch(variables, variable) do
+      {:ok, var} -> var
+      :error -> raise "Variable with name #{inspect(variable)} not found in the petri net."
+    end
   end
 
-  @spec get_arcs_with_place(Transition.t(), Arc.orientation(), ColouredPetriNet.t()) ::
-          Enumerable.t({Arc.t(), Place.t()})
-  def get_arcs_with_place(%Transition{} = transition, orientation, %ColouredPetriNet{} = cpnet)
+  @spec get_arcs_with_place(Transition.t(), Arc.orientation(), RuntimeCpnet.t()) ::
+          [{Arc.t(), Place.t()}]
+  def get_arcs_with_place(
+        %Transition{name: transition_name},
+        orientation,
+        %RuntimeCpnet{arcs_by_transition_orientation: index}
+      )
       when orientation in [:p_to_t, :t_to_p] do
-    %{name: transition_name} = transition
-
-    cpnet.arcs
-    |> Stream.flat_map(fn
-      %Arc{orientation: ^orientation, transition: ^transition_name} = arc -> [arc]
-      %Arc{} -> []
-    end)
-    |> Stream.map(fn %Arc{place: place_name} = arc ->
-      place =
-        Enum.find(cpnet.places, &match?(%Place{name: ^place_name}, &1)) ||
-          raise "Place with name #{inspect(place_name)} not found in the petri net."
-
-      {arc, place}
-    end)
+    Map.get(index, {transition_name, orientation}, [])
   end
 
   @spec get_marking(Place.t(), %{Place.name() => Marking.t()}) :: Marking.t()
@@ -77,40 +54,22 @@ defmodule ColouredFlow.EnabledBindingElements.Utils do
     )
   end
 
-  @spec fetch_transition!(name :: Transition.name(), cpnet :: ColouredPetriNet.t()) ::
+  @spec fetch_transition!(name :: Transition.name(), runtime_cpnet :: RuntimeCpnet.t()) ::
           Transition.t()
-  def fetch_transition!(name, %ColouredPetriNet{} = cpnet) do
-    case Enum.find(cpnet.transitions, &(&1.name == name)) do
-      nil -> raise "Transition not found: #{name}"
-      transition -> transition
+  def fetch_transition!(name, %RuntimeCpnet{transitions: transitions}) do
+    case Map.fetch(transitions, name) do
+      {:ok, transition} -> transition
+      :error -> raise "Transition not found: #{name}"
     end
   end
 
-  @spec list_transitions(in_places :: Enumerable.t(Place.name()), cpnet :: ColouredPetriNet.t()) ::
-          Enumerable.t(Transition.t())
-  def list_transitions(in_places, %ColouredPetriNet{} = cpnet) when is_list(in_places) do
-    in_places = MapSet.new(in_places)
-
-    cpnet.arcs
-    |> Enum.flat_map(fn
-      %Arc{orientation: :p_to_t} = arc ->
-        if arc.place in in_places do
-          [arc.transition]
-        else
-          []
-        end
-
-      %Arc{} ->
-        []
-    end)
-    |> MapSet.new()
-    |> then(fn transition_names ->
-      Enum.filter(cpnet.transitions, &MapSet.member?(transition_names, &1.name))
-    end)
-  end
-
-  @spec build_constants(ColouredPetriNet.t()) :: %{ColourSet.name() => ColourSet.value()}
-  def build_constants(%ColouredPetriNet{} = cpnet) do
-    Map.new(cpnet.constants, &{&1.name, &1.value})
+  @spec list_transitions(
+          in_places :: Enumerable.t(Place.name()),
+          runtime_cpnet :: RuntimeCpnet.t()
+        ) :: [Transition.t()]
+  def list_transitions(in_places, %RuntimeCpnet{transitions_by_input_place: index}) do
+    in_places
+    |> Enum.flat_map(fn place_name -> Map.get(index, place_name, []) end)
+    |> Enum.uniq_by(& &1.name)
   end
 end

lib/coloured_flow/runner/enactment/enactment.ex

@@ -50,6 +50,7 @@ defmodule ColouredFlow.Runner.Enactment do
   alias ColouredFlow.Runner.Enactment.WorkitemCompletion
   alias ColouredFlow.Runner.Enactment.WorkitemConsumption
   alias ColouredFlow.Runner.Exceptions
+  alias ColouredFlow.Runner.RuntimeCpnet
   alias ColouredFlow.Runner.Storage
   alias ColouredFlow.Runner.Telemetry
 
@@ -150,12 +151,12 @@ defmodule ColouredFlow.Runner.Enactment do
   end
 
   def handle_continue(:calibrate_workitems, %__MODULE__{} = state) do
-    cpnet = Storage.get_flow_by_enactment(state.enactment_id)
-    calibration = WorkitemCalibration.initial_calibrate(state, cpnet)
+    runtime_cpnet = build_runtime_cpnet(state.enactment_id)
+    calibration = WorkitemCalibration.initial_calibrate(state, runtime_cpnet)
     state = apply_calibration(calibration)
 
     # try to terminate at the start
-    case check_termination(state, cpnet) do
+    case check_termination(state, runtime_cpnet.definition) do
       {:stop, reason} -> {:stop, {:shutdown, reason}, state}
       :cont -> {:noreply, state, Lifespan.timeout(state)}
     end
@@ -170,8 +171,11 @@ defmodule ColouredFlow.Runner.Enactment do
     state = apply_calibration(calibration)
 
     if transition in [:complete, :complete_e] do
-      cpnet = Storage.get_flow_by_enactment(state.enactment_id)
-      # try to terminate when the transition is `:complete` or `:complete_e`
+      # try to terminate when the transition is `:complete` or `:complete_e`.
+      # `complete_workitems/3` always threads the runtime cpnet through the
+      # calibration options, so we reuse it here instead of re-fetching.
+      %RuntimeCpnet{definition: cpnet} = Keyword.fetch!(options, :runtime_cpnet)
+
       case check_termination(state, cpnet) do
         {:stop, reason} -> {:stop, {:shutdown, reason}, state}
         :cont -> {:noreply, state, Lifespan.timeout(state)}
@@ -497,10 +501,14 @@ defmodule ColouredFlow.Runner.Enactment do
         Enum.map(workitem_id_and_outputs, fn {workitem_id, free_binding} ->
           {Map.fetch!(started_workitems, workitem_id), free_binding}
         end),
-      cpnet = Storage.get_flow_by_enactment(state.enactment_id),
-      {:ok, workitem_occurrences} <- WorkitemCompletion.complete(workitem_and_outputs, cpnet)
+      runtime_cpnet = build_runtime_cpnet(state.enactment_id),
+      {:ok, workitem_occurrences} <-
+        WorkitemCompletion.complete(workitem_and_outputs, runtime_cpnet)
     ) do
-      calibration_options = [cpnet: cpnet, workitem_occurrences: workitem_occurrences]
+      calibration_options = [
+        runtime_cpnet: runtime_cpnet,
+        workitem_occurrences: workitem_occurrences
+      ]
 
       {
         :ok,
@@ -516,6 +524,13 @@ defmodule ColouredFlow.Runner.Enactment do
     end
   end
 
+  @spec build_runtime_cpnet(enactment_id()) :: RuntimeCpnet.t()
+  defp build_runtime_cpnet(enactment_id) do
+    enactment_id
+    |> Storage.get_flow_by_enactment()
+    |> RuntimeCpnet.from_definition()
+  end
+
   @impl GenServer
   def handle_info(:take_snapshot, %__MODULE__{} = state) do
     drain_take_snapshot_messages()