code: def test_continue_training_on_failure(self):
      # We tolerate failing workers and pause training
      config = (
          PPOConfig()
          .environment("CartPole-v1")
          .rollouts(
              num_rollout_workers=6,
              recreate_failed_workers=True,
              validate_workers_after_construction=True,
          )
          .training(
              train_batch_size=300,
          )
      )
      ppo = PPO(config=config)
  
      # One step with all nodes up, enough to satisfy resource requirements
      ppo.train()
  
      self.assertEqual(ppo.workers.num_healthy_remote_workers(), 6)
      self.assertEqual(ppo.workers.num_remote_workers(), 6)
  
      # Remove the first non-head node.
      node_to_kill = get_other_nodes(self.cluster, exclude_head=True)[0]
      self.cluster.remove_node(node_to_kill)
  
      # step() should continue with 4 rollout workers.
      ppo.train()
  
      self.assertEqual(ppo.workers.num_healthy_remote_workers(), 4)
      self.assertEqual(ppo.workers.num_remote_workers(), 6)
  
      # node comes back immediately.
      self.cluster.add_node(
          redis_port=None,
          num_redis_shards=None,
          num_cpus=2,
          num_gpus=0,
          object_store_memory=object_store_memory,
          redis_max_memory=redis_max_memory,
          dashboard_host="0.0.0.0",
      )
  
      # Now, let's wait for Ray to restart all the RolloutWorker actors.
      while True:
          states = [
              a["state"] == "ALIVE"
              for a in list_actors()
              if a["class_name"] == "RolloutWorker"
          ]
          if all(states):
              break
          # Otherwise, wait a bit.
          time.sleep(1)
  
      # This step should continue with 4 workers, but by the end
      # of weight syncing, the 2 recovered rollout workers should
      # be back.
      ppo.train()
  
      # Workers should be back up, everything back to normal.
      self.assertEqual(ppo.workers.num_healthy_remote_workers(), 6)
      self.assertEqual(ppo.workers.num_remote_workers(), 6)
  
file_path: ray/python/ray/rllib/tests/test_node_failure.py
imports: import time
  import unittest
  import ray
  from ray._private.test_utils import get_other_nodes
  from ray.cluster_utils import Cluster
  from ray.util.state import list_actors
  from ray.rllib.algorithms.ppo import PPO, PPOConfig
label_desc: Use image node_id 3 for calling the NodeFailureTests obj's underlying member method code with example usage: obj.test_continue_training_on_failure() without return types