o
    zi'                     @   s<  d dl Z d dlZd dlmZ d dlmZ d dlmZ d dlm	Z	m
Z
mZmZmZmZ d dlZd dlZd dlmZ d dlmZ d dlmZ d dlmZ d d	lmZ d d
lmZ d dlmZ d dl m!Z! d dl"m#Z# d dl$m%Z%m&Z& e	rxd dl'm(Z( G dd deZ)eG dd dZ*dddZ+de
de
fddZ,dddZ-dS )    N)	dataclass)SimpleQueue)dedent)TYPE_CHECKINGAnyCallableDictLiteralOptional)apply_to_collection)Module)override)CPUAccelerator)	_Launcher)move_data_to_device)_set_num_threads_if_needed)_IS_INTERACTIVE)_collect_rng_states_set_rng_states)ParallelStrategyc                   @   s   e Zd ZdZ	dddded ddfd	d
ZeedefddZ	ede
dededefddZ	ddede
dedededed ddfddZdS )_MultiProcessingLaunchera  Launches processes that run a given function in parallel, and joins them all at the end.

    The main process in which this launcher is invoked creates N so-called worker processes (using
    :func:`torch.multiprocessing.start_processes`) that run the given function.
    Worker processes have a rank that ranges from 0 to N - 1.

    Note:
        - This launcher requires all objects to be pickleable.
        - It is important that the entry point to the program/script is guarded by ``if __name__ == "__main__"``.
        - With start method 'fork' the user must ensure that no CUDA context gets created in the main process before
          the launcher is invoked. E.g., one should avoid creating cuda tensors or calling ``torch.cuda.*`` functions
          before calling ``Trainer.fit``.

    Args:
        strategy: A reference to the strategy that is used together with this launcher.
        start_method: The method how to start the processes.
            - 'spawn': The default start method. Requires all objects to be pickleable.
            - 'fork': Preferable for IPython/Jupyter environments where 'spawn' is not available. Not available on
              the Windows platform for example.
            - 'forkserver': Alternative implementation to 'fork'.

    spawnstrategyr   start_method)r   fork
forkserverreturnNc                 C   s<   || _ || _|t vrtd| j ddt  d S )NzThe start method 'z<' is not available on this platform. Available methods are: z, )	_strategy_start_methodmpget_all_start_methods
ValueErrorjoin)selfr   r    r$   i/home/ubuntu/.local/lib/python3.10/site-packages/lightning_fabric/strategies/launchers/multiprocessing.py__init__?   s   
z!_MultiProcessingLauncher.__init__c                 C   s
   | j dkS )Nr   )r   r#   r$   r$   r%   is_interactive_compatibleL   s   
z2_MultiProcessingLauncher.is_interactive_compatiblefunctionargskwargsc                 O   s   | j dv rt  | j dkrt  | jjdusJ t| jjjtjd< t	
| j }| }| j dkr=t }|||||g}n||||g}t	j| j|| jj| j d | S )a  Launches processes that run the given function in parallel.

        The function is allowed to have a return value. However, when all processes join, only the return value
        of worker process 0 gets returned from this `launch` method in the main process.

        Arguments:
            function: The entry point for all launched processes.
            *args: Optional positional arguments to be passed to the given function.
            **kwargs: Optional keyword arguments to be passed to the given function.

        )r   r   r   NMASTER_PORT)r*   nprocsr   )r   _check_bad_cuda_fork_check_missing_main_guardr   cluster_environmentstr	main_portosenvironr   get_contextr   _GlobalStateSnapshotcapturestart_processes_wrapping_functionnum_processesget)r#   r)   r*   r+   contextreturn_queueglobal_statesprocess_argsr$   r$   r%   launchT   s&   


z_MultiProcessingLauncher.launchprocess_idxr=   r>   r6   c                 C   s~   |r|   | jdkrt| jjtrt||f\}}t| jjd t	|t
jd< ||i |}|dkr=|t|d d S d S )Nr   )r:   
LOCAL_RANKr   cpu)restorer   
isinstancer   acceleratorr   _disable_module_memory_sharingr   r:   r1   r3   r4   putr   )r#   rA   r)   r*   r+   r=   r>   resultsr$   r$   r%   r9   }   s   	z+_MultiProcessingLauncher._wrapping_function)r   N)__name__
__module____qualname____doc__r	   r&   propertyr   boolr(   r   r   r@   intr   r
   r9   r$   r$   r$   r%   r   '   s>    
/r   c                   @   sR   e Zd ZU dZeed< eed< eed< eeef ed< e	dddZ
dd
dZd	S )r6   a  Captures a hand-selected set of (global) variables in modules and provides a way to restore them.

    It facilitates and encapsulates the transfer of globals like PyTorch's deterministic flags or random generator state
    across process boundaries when launching processes with :func:`torch.multiprocessing.spawn`.

    Example:

        .. code-block:: python

            # in main process
            snapshot = _GlobalStateSnapshot.capture()

            # in worker process
            snapshot.restore()

    use_deterministic_algorithms&use_deterministic_algorithms_warn_onlycudnn_benchmark
rng_statesr   c                 C   s    | t  t  t jjjt dS )ziCapture a few global states from torch, numpy, etc., that we want to restore in a spawned worker process.)rR   rS   rT   rU   )torch$are_deterministic_algorithms_enabled-is_deterministic_algorithms_warn_only_enabledbackendscudnn	benchmarkr   )clsr$   r$   r%   r7      s   z_GlobalStateSnapshot.captureNc                 C   s,   t j| j| jd | jt jj_t| j dS )zJRestores all globals to the values captured in the :meth:`capture` method.)	warn_onlyN)	rV   rR   rS   rT   rY   rZ   r[   r   rU   r'   r$   r$   r%   rD      s
   z_GlobalStateSnapshot.restore)r   r6   r   N)rK   rL   rM   rN   rP   __annotations__r   r1   r   classmethodr7   rD   r$   r$   r$   r%   r6      s   
 	r6   r   c                  C   s&   t j sdS d} tr| d7 } t| )zChecks whether it is safe to fork and initialize CUDA in the new processes, and raises an exception if not.

    The error message replaces PyTorch's 'Cannot re-initialize CUDA in forked subprocess' with helpful advice for
    Lightning users.

    Na  Lightning can't create new processes if CUDA is already initialized. Did you manually call `torch.cuda.*` functions, have moved the model to the device, or allocated memory on the GPU any other way? Please remove any such calls, or change the selected strategy.z, You will have to restart the Python kernel.)rV   cudais_initializedr   RuntimeErrormessager$   r$   r%   r.      s   
r.   datac                 C   s(   t  dtdtfdd}t| |tdS )zDisables memory sharing on parameters and buffers of `nn.Module`s contained in the given collection.

    Note: This is only required when running on CPU.

    moduler   c                 S   s*   t |  |  D ]}|j |_q
| S rJ   )	itertoolschain
parametersbuffersrf   clone)rg   tensorr$   r$   r%   unshare   s   z/_disable_module_memory_sharing.<locals>.unshare)r)   dtype)rV   no_gradr   r   )rf   rn   r$   r$   r%   rG      s   
rG   c                  C   s$   t t dds
dS td} t| )zGRaises an exception if the ``__name__ == "__main__"`` guard is missing._inheritingFNa  
        Launching multiple processes with the 'spawn' start method requires that your script guards the main
        function with an `if __name__ == "__main__"` clause. For example:

        def main():
            # Put your code here
            ...

        if __name__ == "__main__":
            main()

        Alternatively, you can run with `strategy="ddp"` to avoid this error.
        )getattrr   current_processr   rc   rd   r$   r$   r%   r/      s   r/   r^   ).rh   r3   dataclassesr   multiprocessing.queuesr   textwrapr   typingr   r   r   r   r	   r
   rV   torch.backends.cudnntorch.multiprocessingmultiprocessingr   lightning_utilitiesr   torch.nnr   typing_extensionsr   !lightning_fabric.accelerators.cpur   .lightning_fabric.strategies.launchers.launcherr   %lightning_fabric.utilities.apply_funcr   &lightning_fabric.utilities.distributedr   "lightning_fabric.utilities.importsr   lightning_fabric.utilities.seedr   r   lightning_fabric.strategiesr   r   r6   r.   rG   r/   r$   r$   r$   r%   <module>   s4    m
*