def _train_worker(
    process_rank: int,
    params: Params,
    serialization_dir: Union[str, PathLike],
    include_package: List[str] = None,
    dry_run: bool = False,
    node_rank: int = 0,
    primary_addr: str = "127.0.0.1",
    primary_port: int = 29500,
    world_size: int = 1,
    distributed_device_ids: List[int] = None,
    file_friendly_logging: bool = False,
    include_in_archive: List[str] = None,
    distributed_params: Optional[Params] = None,
) -> Optional[Model]:
    """
    Helper to train the configured model/experiment. In distributed mode, this is spawned as a
    worker process. In a single GPU experiment, this returns the `Model` object and in distributed
    training, nothing is returned.

    # Parameters

    process_rank : `int`
        The process index that is initialized using the GPU device id.
    params : `Params`
        A parameter object specifying an AllenNLP Experiment.
    serialization_dir : `str`
        The directory in which to save results and logs.
    include_package : `List[str]`, optional
        In distributed mode, since this function would have been spawned as a separate process,
        the extra imports need to be done again. NOTE: This does not have any effect in single
        GPU training.
    dry_run : `bool`, optional (default=`False`)
        Do not train a model, but create a vocabulary, show dataset statistics and other training
        information.
    node_rank : `int`, optional
        Rank of the node.
    primary_addr : `str`, optional (default=`"127.0.0.1"`)
        Address of the primary node for distributed training.
    primary_port : `str`, optional (default=`"29500"`)
        Port of the primary node for distributed training.
    world_size : `int`, optional
        The number of processes involved in distributed training.
    distributed_device_ids: `List[str]`, optional
        IDs of the devices used involved in distributed training.
    file_friendly_logging : `bool`, optional (default=`False`)
        If `True`, we add newlines to tqdm output, even on an interactive terminal, and we slow
        down tqdm's output to only once every 10 seconds.
    include_in_archive : `List[str]`, optional
        Paths relative to `serialization_dir` that should be archived in addition to the default ones.
    distributed_params : `Optional[Params]`, optional
        Additional distributed params.

    # Returns

    best_model : `Optional[Model]`
        The model with the best epoch weights or `None` if in distributed training or in dry run.
    """
    common_logging.FILE_FRIENDLY_LOGGING = file_friendly_logging

    common_logging.prepare_global_logging(
        serialization_dir,
        rank=process_rank,
        world_size=world_size,
    )
    common_util.prepare_environment(params)

    distributed = world_size > 1

    primary = process_rank == 0

    include_package = include_package or []

    ddp_accelerator: Optional[DdpAccelerator] = None

    if distributed:
        assert distributed_device_ids is not None
        assert distributed_params is not None

        # Since the worker is spawned and not forked, the extra imports need to be done again.
        # Both the ones from the plugins and the ones from `include_package`.
        import_plugins()
        for package_name in include_package:
            common_util.import_module_and_submodules(package_name)

        num_procs_per_node = len(distributed_device_ids)
        # The Unique identifier of the worker process among all the processes in the
        # distributed training group is computed here. This is used while initializing
        # the process group using `init_process_group`
        global_rank = node_rank * num_procs_per_node + process_rank

        # Number of processes per node is useful to know if a process
        # is a primary in the local node(node in which it is running)
        os.environ["ALLENNLP_PROCS_PER_NODE"] = str(num_procs_per_node)

        # In distributed training, the configured device is always going to be a list.
        # The corresponding gpu id for the particular worker is obtained by picking the id
        # from the device list with the rank as index
        gpu_id = int(distributed_device_ids[process_rank])  # type: ignore

        # Till now, "cuda_device" might not be set in the trainer params.
        # But a worker trainer needs to only know about its specific GPU id.
        params["trainer"]["local_rank"] = process_rank
        params["trainer"]["cuda_device"] = gpu_id
        params["trainer"]["world_size"] = world_size
        params["trainer"]["distributed"] = True

        if gpu_id >= 0:
            torch.cuda.set_device(gpu_id)
            dist.init_process_group(
                backend="nccl",
                init_method=f"tcp://{primary_addr}:{primary_port}",
                world_size=world_size,
                rank=global_rank,
            )
        else:
            dist.init_process_group(
                backend="gloo",
                init_method=f"tcp://{primary_addr}:{primary_port}",
                world_size=world_size,
                rank=global_rank,
            )

        if "ddp_accelerator" in distributed_params:
            ddp_accelerator_params = distributed_params.pop("ddp_accelerator")
            ddp_accelerator = DdpAccelerator.from_params(
                ddp_accelerator_params,
                local_rank=process_rank,
                world_size=world_size,
                cuda_device=gpu_id,
            )

        logging.info(
            f"Process group of world size {world_size} initialized "
            f"for distributed training in worker {global_rank}"
        )

    train_loop = TrainModel.from_params(
        params=params,
        serialization_dir=serialization_dir,
        local_rank=process_rank,
        ddp_accelerator=ddp_accelerator,
    )

    if dry_run:
        return None

    try:
        if distributed:  # let the setup get ready for all the workers
            dist.barrier()

        metrics = train_loop.run()
    except (KeyboardInterrupt, common_util.SigTermReceived):
        # if we have completed an epoch, try to create a model archive.
        if primary:
            best_weights_path = train_loop.trainer.get_best_weights_path()
            if best_weights_path is None:
                logging.info(
                    "Training interrupted by the user, and no best model has been saved. "
                    "No model archive created."
                )
            else:
                logging.info(
                    "Training interrupted by the user. Attempting to create "
                    "a model archive using the current best epoch weights."
                )
                archive_model(
                    serialization_dir,
                    weights=best_weights_path,
                    include_in_archive=include_in_archive,
                )
        raise

    if primary:
        train_loop.finish(metrics)

    if not distributed:
        return train_loop.model

    return None
Helper to train the configured model/experiment. In distributed mode, this is spawned as a
worker process. In a single GPU experiment, this returns the `Model` object and in distributed
training, nothing is returned.

# Parameters

process_rank : `int`
The process index that is initialized using the GPU device id.
params : `Params`
A parameter object specifying an AllenNLP Experiment.
serialization_dir : `str`
The directory in which to save results and logs.
include_package : `List[str]`, optional
In distributed mode, since this function would have been spawned as a separate process,
the extra imports need to be done again. NOTE: This does not have any effect in single
GPU training.
dry_run : `bool`, optional (default=`False`)
Do not train a model, but create a vocabulary, show dataset statistics and other training
information.
node_rank : `int`, optional
Rank of the node.
primary_addr : `str`, optional (default=`"127.0.0.1"`)
Address of the primary node for distributed training.
primary_port : `str`, optional (default=`"29500"`)
Port of the primary node for distributed training.
world_size : `int`, optional
The number of processes involved in distributed training.
distributed_device_ids: `List[str]`, optional
IDs of the devices used involved in distributed training.
file_friendly_logging : `bool`, optional (default=`False`)
If `True`, we add newlines to tqdm output, even on an interactive terminal, and we slow
down tqdm's output to only once every 10 seconds.
include_in_archive : `List[str]`, optional
Paths relative to `serialization_dir` that should be archived in addition to the default ones.
distributed_params : `Optional[Params]`, optional
Additional distributed params.

# Returns

best_model : `Optional[Model]`
The model with the best epoch weights or `None` if in distributed training or in dry run.
allennlp/commands/train.py