chore: refactor handling of the actual dataflow

lib/syskit/data_flow.rb

@@ -469,11 +469,9 @@ def all_inputs_connected?(only_static: false)
                     end
 
                     is_connected =
-                        Runtime::ActualDataFlow
-                            .has_edge?(source_task.orocos_task, orocos_task) &&
-                        Runtime::ActualDataFlow
-                            .edge_info(source_task.orocos_task, orocos_task)
-                            .key?([source_port, sink_port])
+                        Runtime::ActualDataFlow.ports_connected?(
+                            source_task.orocos_task, source_port, orocos_task, sink_port
+                        )
 
                     unless is_connected
                         logger.debug do

lib/syskit/runtime/actual_data_flow_graph.rb

@@ -6,17 +6,21 @@ module Runtime
         #
         # I.e. this is the set of connections that really exist between our
         # components
-        class ActualDataFlowGraph < ConnectionGraph
+        class ActualDataFlowGraph
             # Information about which ports are static and which are not. This
             # information is critical during disconnection to force
             # reconfiguration of the associated tasks
             #
             # @return [Hash<(Orocos::TaskContext,String),Boolean>]
             attr_reader :static_info
 
-            def initialize(*)
-                super
+            def initialize
                 @static_info = {}
+                @graph = ConnectionGraph.new
+            end
+
+            def name=(name)
+                @graph.name = name
             end
 
             # Registers a connection between two tasks
@@ -76,13 +80,19 @@ def add_connections_process_mappings(source_task, sink_task, mappings)
                     @static_info[[sink_task, sink_port]] = sink_static
                     connections[[source_port, sink_port]] = policy
                 end
+                connections
+            end
 
-                if !force_update || !has_edge?(source_task, sink_task)
-                    super(source_task, sink_task, connections)
-                else
-                    set_edge_info(source_task, sink_task,
-                                  edge_info(source_task, sink_task).merge(connections))
-                end
+            def remove_connections(source_task, sink_task, connections)
+                @graph.remove_connections(source_task, sink_task, connections)
+            end
+
+            def clear
+                @graph.clear
+            end
+
+            def empty?
+                @graph.empty?
             end
 
             # Whether the given port is static (per {Port#static?}
@@ -96,6 +106,127 @@ def static?(task, port)
                 raise ArgumentError,
                       "no port #{port} on a task called #{task} is registered on #{self}"
             end
+
+            # Returns whether two ports are connected
+            def tasks_connected?(source_task, sink_task)
+                @graph.has_edge?(source_task, sink_task)
+            end
+
+            # Returns information about the connections existing between two tasks
+            #
+            # @param [Orocos::TaskContext] source_task
+            # @param [Orocos::TaskContext] sink_task
+            # @return [{[String,String] => Hash}] mapping of (source_port,sink_port) pairs
+            #    to the policy of the established connection
+            def connections_of_tasks(source_task, sink_task)
+                return {} unless @graph.has_edge?(source_task, sink_task)
+
+                @graph.edge_info(source_task, sink_task)
+            end
+
+            # Returns whether two ports are connected
+            def ports_connected?(source_task, source_port, sink_task, sink_port)
+                @graph.has_edge?(source_task, sink_task) &&
+                    @graph.edge_info(source_task, sink_task)
+                          .key?([source_port, sink_port])
+            end
+
+            # List connections to the given ports
+            #
+            # @param [Orocos::Task] orocos_task the task whose input port we're inspecting
+            # @param [Array<String>] port_name the name of the input port
+            # @return [{[Orocos::Task,Orocos::Task] => Array<[String,String]>}] matching
+            #   connections, as a mapping of (source_task, sink_task) to the port
+            #   pair (as names)
+            def input_connections_of_ports(orocos_task, port_names)
+                @graph.each_in_neighbour(orocos_task)
+                      .with_object({}) do |source_t, result|
+                    mappings = @graph.edge_info(source_t, orocos_task)
+                    result[[source_t, orocos_task]] =
+                        mappings.each_key.find_all do |_, sink_p|
+                            port_names.include?(sink_p)
+                        end
+                end
+            end
+
+            # Lists the tasks that are present in the graph
+            def each_task
+                @graph.each_vertex
+            end
+
+            # List connections from the given task
+            #
+            # @param [Orocos::Task] orocos_task the task whose output connections are
+            #   expected
+            # @return [{[Orocos::Task,Orocos::Task] => Array<[String,String]>}] matching
+            #   connections, as a mapping of (source_task, sink_task) to the port
+            #   pair (as names)
+            def output_connections_of_task(orocos_task)
+                @graph.each_out_neighbour(orocos_task).with_object({}) do |sink_t, result|
+                    mappings = @graph.edge_info(orocos_task, sink_t)
+                    result[[orocos_task, sink_t]] = mappings.keys
+                end
+            end
+
+            # List connections from the given ports
+            #
+            # @param [Orocos::Task] orocos_task the task whose output ports
+            #   we are inspecting
+            # @param [Array<String>] port_names the names of the output ports
+            # @return [{[Orocos::Task,Orocos::Task] => Array<[String,String]>}] matching
+            #   connections, as a mapping of (source_task, sink_task) to the port
+            #   pair (as names)
+            def output_connections_of_ports(orocos_task, port_names)
+                @graph.each_out_neighbour(orocos_task).with_object({}) do |sink_t, result|
+                    mappings = @graph.edge_info(orocos_task, sink_t)
+                    result[[orocos_task, sink_t]] =
+                        mappings.each_key.find_all do |source_p, _|
+                            port_names.include?(source_p)
+                        end
+                end
+            end
+
+            # List the connections to static input ports
+            #
+            # @param [Orocos::TaskContex] orocos_task
+            # @return [{String=>Set<(Orocos::TaskContext,String)>}] mapping from a
+            #   static input port of orocos_task to its sources, as pairs of tasks and
+            #   port name
+            def static_input_port_connections(orocos_task)
+                @graph.each_in_neighbour(orocos_task)
+                      .with_object({}) do |source_t, result|
+                    connections = @graph.edge_info(source_t, orocos_task)
+                    connections.each_key do |source_p, sink_p|
+                        if static?(orocos_task, sink_p)
+                            sources = (result[sink_p] ||= Set.new)
+                            sources << [source_t, source_p]
+                        end
+                    end
+                end
+            end
+
+            # List the connections to static output ports
+            #
+            # @param [Orocos::TaskContex] orocos_task
+            # @return [{String=>Set<(Orocos::TaskContext,String)>}] mapping from a
+            #   static output port of orocos_task to its sinks, as pairs of tasks and
+            #   port name
+            def static_output_port_connections(orocos_task)
+                @graph.each_out_neighbour(orocos_task).with_object({}) do |sink_t, result|
+                    connections = @graph.edge_info(orocos_task, sink_t)
+                    connections.each_key do |source_p, sink_p|
+                        if static?(orocos_task, source_p)
+                            sinks = (result[source_p] ||= Set.new)
+                            sinks << [sink_t, sink_p]
+                        end
+                    end
+                end
+            end
+
+            # Return the internal graph structure
+            def to_graph
+                @graph
+            end
         end
     end
 end

lib/syskit/runtime/connection_management.rb

@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 
 module Syskit
-    module Runtime
+    module Runtime # :nodoc:
         # (see ConnectionGraph)
         ActualDataFlow = ActualDataFlowGraph.new
         ActualDataFlow.name = "Syskit::ActualDataFlow"
@@ -126,16 +126,19 @@ def compute_connection_changes(tasks)
 
                 update_required_dataflow_graph(tasks)
                 new_edges, removed_edges, updated_edges =
-                    RequiredDataFlow.difference(ActualDataFlow, tasks, &:orocos_task)
+                    RequiredDataFlow.difference(ActualDataFlow.to_graph, tasks, &:orocos_task)
 
                 new = {}
                 new_edges.each do |source_task, sink_task|
-                    new[[source_task, sink_task]] = RequiredDataFlow.edge_info(source_task, sink_task)
+                    new[[source_task, sink_task]] =
+                        RequiredDataFlow.edge_info(source_task, sink_task)
                 end
 
                 removed = {}
                 removed_edges.each do |source_task, sink_task|
-                    removed[[source_task, sink_task]] = ActualDataFlow.edge_info(source_task, sink_task).keys.to_set
+                    removed[[source_task, sink_task]] =
+                        ActualDataFlow
+                        .connections_of_tasks(source_task, sink_task).keys.to_set
                 end
 
                 # We have to work on +updated+. The graphs are between tasks,
@@ -149,7 +152,9 @@ def compute_connection_changes(tasks)
                 # recreate other ones between other components
                 updated_edges.each do |source_task, sink_task|
                     new_mapping = RequiredDataFlow.edge_info(source_task, sink_task)
-                    old_mapping = ActualDataFlow.edge_info(source_task.orocos_task, sink_task.orocos_task)
+                    old_mapping = ActualDataFlow.connections_of_tasks(
+                        source_task.orocos_task, sink_task.orocos_task
+                    )
 
                     new_connections     = {}
                     removed_connections = Set.new
@@ -173,7 +178,8 @@ def compute_connection_changes(tasks)
                         new[[source_task, sink_task]] = new_connections
                     end
                     unless removed_connections.empty?
-                        removed[[source_task.orocos_task, sink_task.orocos_task]] = removed_connections
+                        removed[[source_task.orocos_task, sink_task.orocos_task]] =
+                            removed_connections
                     end
                 end
 
@@ -640,16 +646,12 @@ def apply_connection_changes(new, removed)
             #
             # @return [Hash]
             def dangling_task_cleanup
-                removed = {}
-                ActualDataFlow.each_vertex do |parent_t|
-                    unless @orocos_task_to_syskit_tasks.key?(parent_t)
-                        ActualDataFlow.each_out_neighbour(parent_t) do |child_t|
-                            mappings = ActualDataFlow.edge_info(parent_t, child_t)
-                            removed[[parent_t, child_t]] = mappings.keys.to_set
-                        end
+                ActualDataFlow
+                    .each_task
+                    .find_all { |t| !@orocos_task_to_syskit_tasks.key?(t) }
+                    .each_with_object({}) do |t, result|
+                        result.merge!(ActualDataFlow.output_connections_of_task(t))
                     end
-                end
-                removed
             end
 
             def active_task?(t)

lib/syskit/task_context.rb

@@ -813,7 +813,6 @@ def clean_dynamic_port_connections(port_names)
         # Helper for {#prepare_for_setup} that enumerates the inbound
         # connections originating from a dynamic output port
         def dynamic_input_port_connections(existing_port_names)
-            to_remove = {}
             real_model = model.concrete_model
             dynamic_ports = model.each_input_port.find_all do |p|
                 !real_model.find_input_port(p.name)
@@ -829,22 +828,15 @@ def dynamic_input_port_connections(existing_port_names)
                 end
             end
 
-            Runtime::ActualDataFlow.each_in_neighbour(orocos_task) do |source_t|
-                mappings = Runtime::ActualDataFlow.edge_info(source_t, orocos_task)
-                to_remove[[source_t, orocos_task]] =
-                    mappings.each_key.find_all do |_, sink_p|
-                        dynamic_ports.include?(sink_p)
-                    end
-            end
-            to_remove
+            Runtime::ActualDataFlow
+                .input_connections_of_ports(orocos_task, dynamic_ports)
         end
 
         # @api private
         #
         # Helper for {#prepare_for_setup} that enumerates the outbound
         # connections originating from a dynamic output port
         def dynamic_output_port_connections(existing_port_names)
-            to_remove = {}
             real_model = model.concrete_model
             dynamic_ports = model.each_output_port.find_all do |p|
                 !real_model.find_output_port(p.name)
@@ -860,14 +852,8 @@ def dynamic_output_port_connections(existing_port_names)
                 end
             end
 
-            Runtime::ActualDataFlow.each_out_neighbour(orocos_task) do |sink_t|
-                mappings = Runtime::ActualDataFlow.edge_info(orocos_task, sink_t)
-                to_remove[[orocos_task, sink_t]] =
-                    mappings.each_key.find_all do |source_p, _|
-                        dynamic_ports.include?(source_p)
-                    end
-            end
-            to_remove
+            Runtime::ActualDataFlow
+                .output_connections_of_ports(orocos_task, dynamic_ports)
         end
 
         # @api private
@@ -1408,31 +1394,19 @@ def transaction_modifies_static_ports?
                     end
                 end
 
-                current_connections_to_static = {}
-                Runtime::ActualDataFlow.each_in_neighbour(orocos_task) do |source_t|
-                    # Transactions neither touch ActualDataFlow nor the
-                    # task-to-orocos_task mapping. It's safe to check it
-                    # straight.
-                    connections = Runtime::ActualDataFlow.edge_info(source_t, orocos_task)
-                    connections.each_key do |source_p, sink_p|
-                        if Runtime::ActualDataFlow.static?(orocos_task, sink_p)
-                            sources = (current_connections_to_static[sink_p] ||= Set.new)
-                            sources << [source_t.name, source_p]
-                        end
+                # Transactions neither touch ActualDataFlow nor the
+                # task-to-orocos_task mapping. It's safe to check it
+                # straight.
+                current_input_connections_to_static =
+                    Runtime::ActualDataFlow.static_input_port_connections(orocos_task)
+                current_output_connections_to_static =
+                    Runtime::ActualDataFlow.static_output_port_connections(orocos_task)
+                current_connections_to_static =
+                    current_input_connections_to_static
+                    .merge(current_output_connections_to_static)
+                    .transform_values do |v|
+                        v.to_set { |task, port_name| [task.name, port_name] }
                     end
-                end
-                Runtime::ActualDataFlow.each_out_neighbour(orocos_task) do |sink_t|
-                    # Transactions neither touch ActualDataFlow nor the
-                    # task-to-orocos_task mapping. It's safe to check it
-                    # straight.
-                    connections = Runtime::ActualDataFlow.edge_info(orocos_task, sink_t)
-                    connections.each_key do |source_p, sink_p|
-                        if Runtime::ActualDataFlow.static?(orocos_task, source_p)
-                            sinks = (current_connections_to_static[source_p] ||= Set.new)
-                            sinks << [sink_t.name, sink_p]
-                        end
-                    end
-                end
 
                 current_connections_to_static != new_connections_to_static
             end