feat: finalize integration of manager's available? in update_deployment_state

Author: doudou

lib/syskit/runtime/update_deployment_states.rb

@@ -10,14 +10,24 @@ def self.update_deployment_states(plan)
             # #cleanup_dead_connections, thus avoiding to disconnect connections
             # between already-dead processes
 
+            poll_managers
             handle_dead_deployments(plan)
             trigger_ready_deployments(plan)
         end
 
+        def self.poll_managers
+            server_config = Syskit.conf.each_process_server_config.to_a
+            server_config.each do |config|
+                config.client.poll if config.client.respond_to?(:poll)
+            end
+        end
+
         def self.handle_dead_deployments(plan)
             all_dead_deployments = Set.new
             server_config = Syskit.conf.each_process_server_config.to_a
             server_config.each do |config|
+                next unless config.available?
+
                 begin
                     dead_deployments = config.client.wait_termination
                 rescue ::Exception => e
@@ -42,6 +52,8 @@ def self.trigger_ready_deployments(plan)
             not_ready_deployments = find_all_not_ready_deployments(plan)
             not_ready_deployments.each do |process_server_name, deployments|
                 server_config = Syskit.conf.process_server_config_for(process_server_name)
+                next unless server_config.available?
+
                 wait_result = server_config.client.wait_running(
                     *deployments.map { |d| d.arguments[:process_name] }
                 )

test/runtime/fixtures/missing_process_manager_on_start.rb

@@ -0,0 +1,47 @@
+# frozen_string_literal: true
+
+$stdout.sync = true
+require "roby/schedulers/temporal"
+Roby.scheduler = Roby::Schedulers::Temporal.new
+
+TOKEN = ENV.fetch("SYSKIT_TEST_MISSING_PROCESS_SERVER_TOKEN", nil)
+PROCESS_SERVER_PORT = Integer(ENV.fetch("SYSKIT_TEST_MISSING_PROCESS_SERVER_PORT", nil))
+
+Syskit.conf.remote_process_managers_initial_connection_timeout = 1
+Syskit.conf.remote_process_managers_connection_timeout = 1
+Syskit.conf.remote_process_managers_response_timeout = 1
+Syskit.conf.remote_process_managers_accept_failed_connections = true
+Syskit.conf.register_remote_manager(
+    "test", "localhost", port: PROCESS_SERVER_PORT
+)
+test_remote_manager = Syskit.conf.process_server_config_for("test")
+if test_remote_manager.available?
+    puts "#{TOKEN} - failed - process manager available on start"
+    Roby.app.quit
+end
+
+using_task_library "orogen_syskit_tests"
+
+deadline = nil
+spawned = false
+Robot.controller do
+    Roby.execution_engine.each_cycle do
+        if test_remote_manager.available? && !spawned
+            spawned = true
+            task_m =
+                OroGen.orogen_syskit_tests.Empty
+                      .deployed_as("#{Process.pid}_missing_empty", on: "test")
+            task = Roby.plan.add_mission_task(task_m)
+            task.start_event.on do |_event|
+                puts "#{TOKEN} - success"
+                Roby.app.quit
+            end
+        end
+    end
+
+    deadline ||= Time.now + 10
+    if Time.now > deadline
+        puts "#{TOKEN} - failed"
+        Roby.app.quit
+    end
+end

test/runtime/test_update_deployment_state.rb

@@ -1,6 +1,8 @@
 # frozen_string_literal: true
 
 require "syskit/test/self"
+require "syskit/test/roby_app_helpers"
+require "syskit/process_managers/remote/server"
 
 module Syskit
     module Runtime
@@ -38,11 +40,50 @@ def disconnect; end
         end
 
         describe ".update_deployment_states" do
+            describe "system behaviour with remote manager availability" do
+                include Syskit::Test::RobyAppHelpers
+
+                after do
+                    if @remote_process_server_pid
+                        remote_process_server_quit(@remote_process_server_pid)
+                    end
+                end
+
+                def roby_app_fixture_path
+                    File.expand_path("fixtures", __dir__)
+                end
+
+                it "gracefully handles a process manager that appears after start" do
+                    dir = roby_app_setup_single_script(
+                        "missing_process_manager_on_start.rb"
+                    )
+                    port = roby_app_allocate_port
+                    token = SecureRandom.base64(10)
+                    env = {
+                        "SYSKIT_TEST_MISSING_PROCESS_SERVER_PORT" => port.to_s,
+                        "SYSKIT_TEST_MISSING_PROCESS_SERVER_TOKEN" => token
+                    }
+
+                    pid, = roby_app_start(
+                        "run", "-c", "scripts/missing_process_manager_on_start.rb",
+                        capture_output: true, chdir: dir, env: env
+                    )
+                    sleep 5
+
+                    @remote_process_server_pid = remote_process_server_spawn(port)
+                    assert_roby_app_exits(pid)
+                    output = roby_app_captured_output(pid)
+                    refute_match(/#{token} - failed/, output[:out])
+                    assert_match(/#{token} - success/, output[:out])
+                end
+            end
+
             describe "#handle_dead_deployments" do
                 it "calls #dead! on the dead deployments" do
                     client = flexmock
-                    flexmock(Syskit.conf).should_receive(:each_process_server_config)
-                                         .and_return([flexmock(client: client)])
+                    flexmock(Syskit.conf)
+                        .should_receive(:each_process_server_config)
+                        .and_return([flexmock(client: client, available?: true)])
                     client.should_receive(:wait_termination)
                           .and_return([[p = flexmock, s = flexmock]])
                     flexmock(Deployment).should_receive(:deployment_by_process).with(p)
@@ -185,6 +226,22 @@ def mocked_deployment(
                     mock
                 end
             end
+
+            def remote_process_server_spawn(port)
+                log_dir = make_tmpdir
+                pid = spawn(
+                    "syskit", "process_server",
+                    "--log-dir=#{log_dir}", "--port=#{port}",
+                    chdir: log_dir, out: "/dev/null", err: "/dev/null"
+                )
+                register_pid(pid)
+                pid
+            end
+
+            def remote_process_server_quit(pid)
+                Process.kill "INT", pid
+                assert_process_exits(pid)
+            end
         end
     end
 end