code: def run_fault_tolerant_loop():
      """Entrance function to the fairseq library, providing fault-tolerance."""
  
      # Parse the command line arguments.
      parser = options.get_training_parser()
      add_ray_args(parser)
      args = options.parse_args_and_arch(parser)
      original_args = copy.deepcopy(args)
  
      # Main loop for fault-tolerant training.
      retry = True
      while retry:
          args = copy.deepcopy(original_args)
  
          # Initialize Ray.
          ray.init(address=args.ray_address)
  
          set_num_resources(args)
          set_batch_size(args)
  
          # Set up Ray distributed actors.
          Actor = ray.remote(num_cpus=1, num_gpus=int(not args.cpu))(
              RayDistributedActor
          )
          workers = [
              Actor.remote() for i in range(args.distributed_world_size)
          ]
  
          # Get the IP address and a free port of actor 0, which is used for
          # fairseq distributed training.
          ip = ray.get(workers[0].get_node_ip.remote())
          port = ray.get(workers[0].find_free_port.remote())
          address = "tcp://{ip}:{port}".format(ip=ip, port=port)
  
          # Start the remote processes, and check whether their are any process
          # fails. If so, restart all the processes.
          unfinished = [
              worker.run.remote(address, i, args)
              for i, worker in enumerate(workers)
          ]
          try:
              while len(unfinished) > 0:
                  finished, unfinished = ray.wait(unfinished)
                  finished = ray.get(finished)
              retry = False
          except Exception as inst:
              print("Ray restart because following error occurs:")
              print(inst)
              retry = True
          ray.shutdown()
  
file_path: ray/doc/source/ray-core/examples/lm/ray_train.py
imports: import copy
  import math
  import socket
  import time
  from contextlib import closing
  import fairseq
  from fairseq import options
  from fairseq_cli.train import main
  import ray
label_desc: Use image node_id 1 for calling a global function with example usage: run_fault_tolerant_loop() without return types