code: def run(
      local_rank: int,
      work_run: Callable,
      main_address: str,
      main_port: int,
      num_nodes: int,
      node_rank: int,
      nprocs: int,
  ):
      import torch
  
      # 1. Setting distributed environment
      global_rank = local_rank + node_rank * nprocs
      world_size = num_nodes * nprocs
  
      if torch.distributed.is_available():
          if not torch.distributed.is_initialized():
              torch.distributed.init_process_group(
                  "nccl" if torch.cuda.is_available() else "gloo",
                  rank=global_rank,
                  world_size=world_size,
                  init_method=f"tcp://{main_address}:{main_port}",
              )
      elif world_size > 1:
          raise Exception("Torch distributed should be available.")
  
      return work_run(world_size, node_rank, global_rank, local_rank)
  
file_path: pytorch-lightning/src/lightning/app/components/multi_node/pytorch_spawn.py
imports: from typing import Any, Callable, Protocol, Type, runtime_checkable
  from lightning.app.components.multi_node.base import MultiNode
  from lightning.app.core.queues import MultiProcessQueue
  from lightning.app.core.work import LightningWork
  from lightning.app.utilities.packaging.cloud_compute import CloudCompute
  from lightning.app.utilities.proxies import WorkRunExecutor, WorkStateObserver, _proxy_setattr, unwrap
label_desc: Use image node_id 3 for calling the _PyTorchSpawnRunExecutor obj's underlying member method code with example usage: obj.run(local_rank, work_run, main_address, main_port, num_nodes, node_rank, nprocs) and returns: work_run