feat: early deploy at network generation

Rakefile

@@ -28,6 +28,28 @@ def minitest_set_options(test_task, name)
     test_task.options = "#{TESTOPTS} #{minitest_args} -- --simplecov-name=#{name}"
 end
 
+def core(early_deploy: false)
+    s = ":no-early-deploy"
+    if early_deploy
+        s = ":early-deploy"
+        early_deploy_setup = ["test/features/early_deploy.rb"]
+    end
+
+    Rake::TestTask.new("test:core#{s}") do |t|
+        t.libs << "."
+        t.libs << "lib"
+        minitest_set_options(t, "core")
+        test_files = FileList["test/**/test_*.rb", *early_deploy_setup]
+        test_files = test_files
+                     .exclude("test/ros/**/*.rb")
+                     .exclude("test/gui/**/*.rb")
+                     .exclude("test/live/**/*.rb")
+                     .exclude("test/telemetry/**/*.rb")
+        t.test_files = test_files
+        t.warning = false
+    end
+end
+
 Rake::TestTask.new("test:telemetry") do |t|
     t.libs << "."
     t.libs << "lib"
@@ -36,27 +58,16 @@ Rake::TestTask.new("test:telemetry") do |t|
     t.warning = false
 end
 
-Rake::TestTask.new("test:core") do |t|
-    t.libs << "."
-    t.libs << "lib"
-    minitest_set_options(t, "core")
-    test_files = FileList["test/**/test_*.rb"]
-    test_files = test_files
-                 .exclude("test/ros/**/*.rb")
-                 .exclude("test/gui/**/*.rb")
-                 .exclude("test/live/**/*.rb")
-                 .exclude("test/telemetry/**/*.rb")
-    t.test_files = test_files
-    t.warning = false
-end
-
+desc "Run separate tests that require a live syskit instance"
 task "test:live" do
     tests = Dir.enum_for(:glob, "test/live/test_*.rb").to_a
     unless system(File.join("test", "live", "run"), *tests)
         $stderr.puts "live tests failed"
         exit 1
     end
 end
+
+desc "run gui-only tests"
 Rake::TestTask.new("test:gui") do |t|
     t.libs << "."
     t.libs << "lib"
@@ -66,6 +77,12 @@ Rake::TestTask.new("test:gui") do |t|
     t.warning = false
 end
 
+core early_deploy: true
+core
+desc "Run core library tests, excluding GUI and live tests"
+task "test:core" => ["test:core:no-early-deploy", "test:core:early-deploy"]
+
+desc "Run all tests"
 task "test" => ["test:gui", "test:core", "test:live", "test:telemetry"]
 
 task "rubocop" do

lib/syskit.rb

@@ -127,6 +127,7 @@ module ProcessManagers
 require "syskit/actual_data_flow_graph"
 require "syskit/data_flow"
 require "syskit/connection_graphs"
+require "syskit/network_generation_exception_helpers"
 require "syskit/exceptions"
 require "syskit/network_generation"
 require "syskit/runtime"

lib/syskit/base.rb

@@ -2,6 +2,7 @@
 
 require "logger"
 require "utilrb/logger"
+require "syskit/network_generation_exception_helpers"
 require "syskit/exceptions"
 require "facets/string/snakecase"
 

lib/syskit/exceptions.rb

@@ -437,6 +437,8 @@ def pretty_print(pp)
     end
 
     class ConflictingDeviceAllocation < SpecError
+        include Syskit::NetworkGenerationsExceptionHelpers
+
         attr_reader :device, :tasks, :inputs
 
         def can_merge?
@@ -447,38 +449,54 @@ def initialize(device, task0, task1, toplevel_tasks_to_requirements = {})
             @device = device
             @tasks = [task0, task1]
 
-            solver = NetworkGeneration::MergeSolver.new(task0.plan)
-            @merge_result = solver.resolve_merge(task0, task1, {})
             @involved_definitions = @tasks.map do |t|
                 find_all_related_syskit_actions(t, toplevel_tasks_to_requirements)
             end
         end
 
-        def find_all_related_syskit_actions(task, toplevel_tasks_to_requirements)
-            result = []
-            while task
-                result.concat(toplevel_tasks_to_requirements[task] || [])
-                task = task.each_parent_task.first
-            end
-            result
-        end
-
         def pretty_print(pp)
             pp.text "device '#{device.name}' of type #{device.model} is assigned "
             pp.text "to two tasks that cannot be merged"
-            pp.breakable
-            @merge_result.pretty_print_failure(pp)
-            @involved_definitions.each_with_index do |defs, i|
-                next if defs.empty?
+            print_failed_merge_chain(pp, *@tasks)
+            @tasks.zip(@involved_definitions).each do |t, defs|
+                print_dependent_definitions(pp, t, defs)
+            end
+        end
+    end
 
-                pp.breakable
-                pp.text "Chain #{i + 1} is needed by the following definitions:"
-                pp.nest(2) do
-                    defs.each do |d|
-                        pp.breakable
-                        pp.text d.to_s
-                    end
+    class ConflictingDeploymentAllocation < SpecError
+        include Syskit::NetworkGenerationsExceptionHelpers
+
+        attr_reader :deployment_to_tasks
+
+        def initialize(deployment_to_tasks, toplevel_tasks_to_requirements = {})
+            @deployment_to_tasks = deployment_to_tasks
+            @toplevel_tasks_to_requirements = toplevel_tasks_to_requirements
+            @deployment_to_execution_agent = \
+                deployment_to_tasks.transform_values do |tasks|
+                    tasks.first.execution_agent
+                end
+        end
+
+        def pretty_print(pp)
+            deployment_to_tasks.each do |orocos_name, tasks|
+                agent = @deployment_to_execution_agent[orocos_name]
+                deployment_m = agent.deployed_orogen_model_by_name(orocos_name)
+                pp.text(
+                    "deployed task '#{orocos_name}' from deployment " \
+                    "'#{deployment_m.name}' defined in " \
+                    "'#{deployment_m.project.name}' on '#{agent.process_server_name}' " \
+                    "is assigned to #{tasks.size} tasks. Below is the list of " \
+                    "the dependent non-deployed actions. Right after the list " \
+                    "is a detailed explanation of why the first two tasks are not merged:"
+                )
+                tasks.each do |t|
+                    defs = find_all_related_syskit_actions(
+                        t, @toplevel_tasks_to_requirements
+                    )
+                    print_dependent_definitions(pp, t, defs)
                 end
+                print_failed_merge_chain(pp, tasks[0], tasks[1])
             end
         end
     end

lib/syskit/network_generation/engine.rb

@@ -401,7 +401,7 @@ def finalize_deployed_tasks
 
                 # This is required to merge the already existing compositions
                 # with the ones in the plan
-                merge_solver.merge_identical_tasks
+                merge_solver.merge_compositions
                 log_timepoint "merge"
 
                 [selected_deployment_tasks, reused_deployed_tasks | newly_deployed_tasks]
@@ -703,18 +703,27 @@ def compute_system_network(
                     Engine.discover_requirement_tasks_from_plan(real_plan),
                 garbage_collect: true,
                 validate_abstract_network: true,
-                validate_generated_network: true
+                validate_generated_network: true,
+                default_deployment_group: Syskit.conf.deployment_group,
+                validate_deployed_network: (true if Syskit.conf.early_deploy?),
+                early_deploy: Syskit.conf.early_deploy?
             )
                 requirement_tasks = requirement_tasks.to_a
                 instance_requirements = requirement_tasks.map(&:requirements)
+                merge_solver.merge_task_contexts_with_same_agent = early_deploy
                 system_network_generator = SystemNetworkGenerator.new(
-                    work_plan, event_logger: event_logger, merge_solver: merge_solver
+                    work_plan,
+                    event_logger: event_logger,
+                    merge_solver: merge_solver,
+                    default_deployment_group: default_deployment_group,
+                    early_deploy: early_deploy
                 )
                 toplevel_tasks = system_network_generator.generate(
                     instance_requirements,
                     garbage_collect: garbage_collect,
                     validate_abstract_network: validate_abstract_network,
-                    validate_generated_network: validate_generated_network
+                    validate_generated_network: validate_generated_network,
+                    validate_deployed_network: validate_deployed_network
                 )
 
                 Hash[requirement_tasks.zip(toplevel_tasks)]
@@ -749,14 +758,19 @@ def resolve_system_network(
                 validate_deployed_network: true,
                 compute_deployments: true,
                 default_deployment_group: Syskit.conf.deployment_group,
-                compute_policies: true
+                compute_policies: true,
+                early_deploy: Syskit.conf.early_deploy?
             )
 
+                merge_solver.merge_task_contexts_with_same_agent = early_deploy
                 required_instances = compute_system_network(
                     requirement_tasks,
                     garbage_collect: garbage_collect,
                     validate_abstract_network: validate_abstract_network,
-                    validate_generated_network: validate_generated_network
+                    validate_generated_network: validate_generated_network,
+                    default_deployment_group: (default_deployment_group if early_deploy),
+                    validate_deployed_network: validate_deployed_network,
+                    early_deploy: early_deploy && compute_deployments
                 )
 
                 if compute_deployments
@@ -804,8 +818,10 @@ def resolve(
                 validate_abstract_network: true,
                 validate_generated_network: true,
                 validate_deployed_network: true,
-                validate_final_network: true
+                validate_final_network: true,
+                early_deploy: Syskit.conf.early_deploy?
             )
+                merge_solver.merge_task_contexts_with_same_agent = early_deploy
                 required_instances = resolve_system_network(
                     requirement_tasks,
                     garbage_collect: garbage_collect,
@@ -814,7 +830,8 @@ def resolve(
                     compute_deployments: compute_deployments,
                     default_deployment_group: default_deployment_group,
                     compute_policies: compute_policies,
-                    validate_deployed_network: validate_deployed_network
+                    validate_deployed_network: validate_deployed_network,
+                    early_deploy: early_deploy
                 )
 
                 apply_system_network_to_plan(

lib/syskit/network_generation/merge_solver.rb

@@ -7,7 +7,7 @@ module NetworkGeneration
         #
         # This is the core of the system deployment algorithm implemented in
         # Engine
-        class MergeSolver
+        class MergeSolver # rubocop:disable Metrics/ClassLength
             extend Logger::Hierarchy
             include Logger::Hierarchy
             include Roby::DRoby::EventLogging
@@ -34,6 +34,8 @@ class MergeSolver
             # information
             attr_reader :event_logger
 
+            attr_writer :merge_task_contexts_with_same_agent
+
             def initialize(plan, event_logger: plan.event_logger)
                 @plan = plan
                 @event_logger = event_logger
@@ -43,6 +45,7 @@ def initialize(plan, event_logger: plan.event_logger)
                 @task_replacement_graph = Roby::Relations::BidirectionalDirectedAdjacencyGraph.new
                 @resolved_replacements = {}
                 @invalid_merges = Set.new
+                @merge_task_contexts_with_same_agent = false
             end
 
             def clear
@@ -51,6 +54,10 @@ def clear
                 @invalid_merges.clear
             end
 
+            def merge_task_contexts_with_same_agent?
+                @merge_task_contexts_with_same_agent
+            end
+
             # Returns the task that is used in place of the given task
             #
             # @param [Roby::Task] the task for which we want to know the
@@ -198,16 +205,7 @@ def self.merge_identical_tasks(plan)
                 solver.merge_identical_tasks
             end
 
-            # Tests whether task.merge(target_task) is a valid operation
-            #
-            # @param [Syskit::TaskContext] task
-            # @param [Syskit::TaskContext] target_task
-            #
-            # @return [false,true] if false, the merge is not possible. If
-            #   true, it is possible. If nil, the only thing that makes the
-            #   merge impossible are missing inputs, and these tasks might
-            #   therefore be merged if there was a dataflow cycle
-            def may_merge_task_contexts?(merged_task, task)
+            def may_merge_components?(merged_task, task)
                 can_merge = log_nest(2) do
                     task.can_merge?(merged_task)
                 end
@@ -220,16 +218,39 @@ def may_merge_task_contexts?(merged_task, task)
                     return false
                 end
 
+                true
+            end
+
+            # Tests whether task.merge(target_task) is a valid operation
+            #
+            # @param [Syskit::TaskContext] task
+            # @param [Syskit::TaskContext] target_task
+            #
+            # @return [false,true] if false, the merge is not possible. If
+            #   true, it is possible.
+            def may_merge_task_contexts?(merged_task, task)
+                return false unless may_merge_components?(merged_task, task)
+
                 # Merges involving a deployed task can only involve a
                 # non-deployed task as well
-                if task.execution_agent && merged_task.execution_agent
+                unless mergeable_agents?(merged_task, task)
                     info "rejected: deployment attribute mismatches"
                     return false
                 end
 
                 true
             end
 
+            def mergeable_agents?(merged_task, task)
+                unless merge_task_contexts_with_same_agent?
+                    return !(task.execution_agent && merged_task.execution_agent)
+                end
+
+                return false unless task.execution_agent && merged_task.execution_agent
+
+                task.orocos_name == merged_task.orocos_name
+            end
+
             def each_component_merge_candidate(task)
                 # Get the set of candidates. We are checking if the tasks in
                 # this set can be replaced by +task+
@@ -329,9 +350,7 @@ def composition_children_by_role(task)
             end
 
             def may_merge_compositions?(merged_task, task)
-                unless may_merge_task_contexts?(merged_task, task)
-                    return false
-                end
+                return false unless may_merge_components?(merged_task, task)
 
                 merged_task_children = composition_children_by_role(merged_task)
                 task_children        = composition_children_by_role(task)
@@ -446,6 +465,22 @@ def pretty_print_failure(pp)
                 :source_task, :source_port, :policy, :sink_port, :sink_task
             )
 
+            def may_merge?(merged_task, task)
+                case merged_task
+                when TaskContext
+                    may_merge_task_contexts?(merged_task, task)
+                when Composition
+                    may_merge_compositions?(merged_task, task)
+                when Placeholder
+                    may_merge_components?(merged_task, task)
+                else
+                    raise ArgumentError,
+                          "may_merge? called with #{merged_task} of type " \
+                          "#{merged_task.class}, expected either TaskContext, " \
+                          "Composition or Placeholder"
+                end
+            end
+
             # Resolve merge between N tasks with the given tasks as seeds
             #
             # The method will cycle through the task's mismatching inputs (if
@@ -462,7 +497,7 @@ def pretty_print_failure(pp)
             #
             # @return [MergeResolution]
             def resolve_merge(merged_task, task, mappings)
-                unless may_merge_task_contexts?(merged_task, task)
+                unless may_merge?(merged_task, task)
                     return MergeResolution.new(mappings, merged_task, task, [], [])
                 end