class Syskit::DataFlow
Class out of which the Flows::DataFlow graph object is made
see {ConnectionGraph} for a description of the roles of each connection graph
Attributes
If non-nil, this holds the set of concrete connections for this data flow graph. It MUST be maintained by some external entity, and as such is set only in contexts where the set of modifications to the graph is known (e.g. {NetworkGeneration::MergeSolver}
@return [ConnectionGraph,nil]
Returns the set of tasks whose data flow has been changed that has not yet been applied.
It is maintained only on executable plans, through the added/removed/updated hooks {TaskContext#added_sink}, {TaskContext#removed_sink}, {TaskContext#updated_sink}, {Composition#added_sink}, {Composition#removing_sink} and {Composition#updated_sink}
The set of connection changes that have been applied to the DataFlow relation graph, but not yet applied on the actual components (i.e. not yet present in the ActualDataFlow graph).
Public Class Methods
# File lib/syskit/data_flow.rb, line 36 def initialize(*args, **options) super @modified_tasks = Set.new @concrete_connection_graph = nil end
Public Instance Methods
@api private
Computes the concrete connection graph from the DataFlow information
# File lib/syskit/data_flow.rb, line 66 def compute_concrete_connection_graph current_graph, @concrete_connection_graph = @concrete_connection_graph, nil graph = ConcreteConnectionGraph.new each_vertex do |task| next if !task.kind_of?(Syskit::TaskContext) task_to_task = Hash.new each_concrete_in_connection(task) do |source_task, source_port, sink_port, policy| port_to_port = (task_to_task[source_task] ||= Hash.new) port_to_port[[source_port, sink_port]] = policy end task_to_task.each do |source_task, mappings| graph.add_edge(source_task, task, mappings) end end graph ensure @concrete_connection_graph = current_graph end
# File lib/syskit/data_flow.rb, line 91 def concrete_connection_graph_enabled? !!@concrete_connection_graph end
# File lib/syskit/data_flow.rb, line 87 def disable_concrete_connection_graph @concrete_connection_graph = nil end
Yield or enumerates the connections that exist towards the input ports of self. It does not include connections to composition ports (i.e. exported ports): these connections are followed until a concrete port (a port on an actual Syskit::TaskContext) is found.
@param [#name,String,nil] port if non-nil, the port for
which we want to enumerate the connections (in which case the sink_port yield parameter is guaranteed to be this name). Otherwise, all ports are enumerated.
@yield each connections @yieldparam [Syskit::TaskContext] source_task the source task in
the connection
@yieldparam [String] source_port the source port name on source_task @yieldparam [String] sink_port the sink port name on self. If
the port argument is non-nil, it is guaranteed to be the same.
@yieldparam [Hash] policy the connection policy
@see each_input_connection each_concrete_output_connection
each_output_connection
# File lib/syskit/data_flow.rb, line 141 def each_concrete_in_connection(task, port = nil) return enum_for(__method__, task, port) if !block_given? if concrete_connection_graph return concrete_connection_graph.each_in_connection(task, port, &proc) end each_in_connection(task, port) do |source_task, source_port, sink_port, policy| # Follow the forwardings while +sink_task+ is a composition if source_task.kind_of?(Composition) each_concrete_in_connection(source_task, source_port) do |source_task, source_port, _, connection_policy| begin this_policy = Syskit.update_connection_policy(policy, connection_policy) rescue ArgumentError => e raise SpecError, "incompatible policies in input chain for #{self}:#{sink_port}: #{e.message}" end policy_copy = this_policy.dup yield(source_task, source_port, sink_port, this_policy) if policy_copy != this_policy connection_policy.clear connection_policy.merge!(this_policy) end end else yield(source_task, source_port, sink_port, policy) end end self end
Yield or enumerates the connections that exist from the output ports of self. It does not include connections to composition ports (i.e. exported ports): these connections are followed until a concrete port (a port on an actual Syskit::TaskContext) is found.
@param [#name,String,nil] port if non-nil, the port for
which we want to enumerate the connections (in which case the source_port yield parameter is guaranteed to be this name). Otherwise, all ports are enumerated.
@yield each connections @yieldparam [String] source_port the source port name on self. If
the port argument is non-nil, it is guaranteed to be the same.
@yieldparam [String] sink_port the sink port name on sink_task. @yieldparam [Syskit::TaskContext] sink_task the sink task in
the connection
@yieldparam [Hash] policy the connection policy
@see each_concrete_input_connection each_input_connection
each_output_connection
# File lib/syskit/data_flow.rb, line 194 def each_concrete_out_connection(task, port = nil) return enum_for(__method__, task, port) if !block_given? if concrete_connection_graph return concrete_connection_graph.each_out_connection(task, port, &proc) end each_out_connection(task, port) do |source_port, sink_port, sink_task, policy| # Follow the forwardings while +sink_task+ is a composition if sink_task.kind_of?(Composition) each_concrete_out_connection(sink_task, sink_port) do |_, sink_port, sink_task, connection_policy| begin this_policy = Syskit.update_connection_policy(policy, connection_policy) rescue ArgumentError => e raise SpecError, "incompatible policies in output chain for #{self}:#{source_port}: #{e.message}" end policy_copy = this_policy.dup yield(source_port, sink_port, sink_task, this_policy) if policy_copy != this_policy connection_policy.clear connection_policy.merge!(this_policy) end end else yield(source_port, sink_port, sink_task, policy) end end self end
# File lib/syskit/data_flow.rb, line 54 def enable_concrete_connection_graph(compute: true) @concrete_connection_graph = if compute compute_concrete_connection_graph else ConcreteConnectionGraph.new end end
Called by the relation graph management to update the DataFlow edge information when connections are added or removed.
# File lib/syskit/data_flow.rb, line 97 def merge_info(source, sink, current_mappings, additional_mappings) current_mappings.merge(additional_mappings) do |_, old_options, new_options| Syskit.update_connection_policy(old_options, new_options) end end