U 9%e4@sddlZddlmZmZddlmZddlmZmZm Z m Z m Z ddl Z ddl mZdddgZGdddZGd ddeZGd d d e ZGd ddZdS) N)ABCabstractmethod) TracebackType)AnyList NamedTupleOptionalTypeJoinHookJoinableJoinc@s.eZdZdZddddZeddddZdS) r a This defines a join hook, which provides two entry points in the join context manager: a main hook, which is called repeatedly while there exists a non-joined process, and a post-hook, which is called once all processes have joined. To implement a join hook for the generic join context manager, define a class that inherits from :class:`JoinHook` and override ``main_hook()`` and ``post_hook()`` as appropriate. NreturncCsdS)z This hook is called repeatedly while there exists a non-joined process to shadow collective communications in one training iteration (i.e. in one forward pass, backward pass, and optimizer step). Nselfrr`/var/www/html/Darija-Ai-API/env/lib/python3.8/site-packages/torch/distributed/algorithms/join.py main_hookszJoinHook.main_hook)is_last_joinerrcCsdS)a\ This hook is called after all processes have joined. It is passed an additional ``bool`` argument ``is_last_joiner``, which indicates if the rank is one of the last to join. Arguments: is_last_joiner (bool): ``True`` if the rank is one of the last to join; ``False`` otherwise. Nr)rrrrr post_hooks zJoinHook.post_hook)__name__ __module__ __qualname____doc__rboolrrrrrr s csdeZdZdZefddZeedddZeee j dddZ eee dd d Z ZS) r aZ This defines an abstract base class for joinable classes. A joinable class (inheriting from :class:`Joinable`) should implement :meth:`join_hook`, which returns a :class:`JoinHook` instance, in addition to :meth:`join_device` and :meth:`join_process_group` that return device and process group information, respectively. cstt|_dSN)super__init__ _JoinConfigconstruct_disabled_join_config _join_configr __class__rrr3s zJoinable.__init__r cKsdS)a Returns a :class:`JoinHook` instance for the given :class:`Joinable`. Arguments: kwargs (dict): a :class:`dict` containing any keyword arguments to modify the behavior of the join hook at run time; all :class:`Joinable` instances sharing the same join context manager are forwarded the same value for ``kwargs``. Nr)rkwargsrrr join_hook8s zJoinable.join_hookcCsdS)z Returns the device from which to perform collective communications needed by the join context manager implementation itself. Nrrrrr join_deviceEszJoinable.join_devicecCsdS)z Returns the process group for the collective communications needed by the join context manager itself. Nrrrrrjoin_process_groupNszJoinable.join_process_group)rrrrrrr r$propertytorchdevicer%rr& __classcell__rrr!rr +s c@s6eZdZUdZeed<eed<eed<eddZdS)rzr This includes all fields needed from a :class:`Joinable` instance for the join context manager side. enablethrow_on_early_terminationis_first_joinablecCstddddS)z Returns a :class:`_JoinConfig` instance indicating that join-related logic should be disabled, e.g. if the caller is not in a join context manager. Fr+r,r-)rrrrrras z*_JoinConfig.construct_disabled_join_configN)rrrrr__annotations__ staticmethodrrrrrrXs rc@seZdZdZdeeeedddZddd d Zddd d Z d dZ e e e e e e edddZddZddZeedddZdS)r a This class defines the generic join context manager, which allows custom hooks to be called after a process joins. These hooks should shadow the collective communications of non-joined processes to prevent hanging and erroring and to ensure algorithmic correctness. Refer to :class:`JoinHook` for details about the hook definition. .. warning:: The context manager requires each participating :class:`Joinable` to call the method :meth:`notify_join_context()` before its own per- iteration collective communications to ensure correctness. .. warning:: The context manager requires that all ``process_group`` attributes in the :class:`JoinHook` objects are the same. If there are multiple :class:`JoinHook` objects, then the ``device`` of the first is used. The process group and device information is used for checking for non- joined processes and for notifying processes to throw an exception if ``throw_on_early_termination`` is enabled, both of which using an all- reduce. Arguments: joinables (List[Joinable]): a list of the participating :class:`Joinable` s; their hooks are iterated over in the given order. enable (bool): a flag enabling uneven input detection; setting to ``False`` disables the context manager's functionality and should only be set when the user knows the inputs will not be uneven (default: ``True``). throw_on_early_termination (bool): a flag controlling whether to throw an exception upon detecting uneven inputs (default: ``False``). Example:: >>> import os >>> import torch >>> import torch.distributed as dist >>> import torch.multiprocessing as mp >>> # xdoctest: +SKIP >>> import torch.nn.parallel.DistributedDataParallel as DDP >>> import torch.distributed.optim.ZeroRedundancyOptimizer as ZeRO >>> from torch.distributed.algorithms.join import Join >>> >>> # On each spawned worker >>> def worker(rank): >>> dist.init_process_group("nccl", rank=rank, world_size=2) >>> model = DDP(torch.nn.Linear(1, 1).to(rank), device_ids=[rank]) >>> optim = ZeRO(model.parameters(), torch.optim.Adam, lr=0.01) >>> # Rank 1 gets one more input than rank 0 >>> inputs = [torch.tensor([1.]).to(rank) for _ in range(10 + rank)] >>> with Join([model, optim]): >>> for input in inputs: >>> loss = model(input).sum() >>> loss.backward() >>> optim.step() >>> # All ranks reach here without hanging/erroring TF) joinablesr+r,c sPt|dkrtd||_fdd|jD|_||_||_||dS)Nrz7The join context manager requires at least one joinablecsg|]}|jfqSr)r$).0joinabler#rr sz!Join.__init__..)len ValueError _joinables _join_hooks_enable_throw_on_early_termination_set_joinable_configs_extract_dist_info)rr1r+r,r#rr4rrs z Join.__init__Nr cCs>t|jdkstd}|jD]}t|j|j|d|_d}qdS)zX Sets the :class:`_JoinConfig` of each participating :class:`Joinable`. rTr.FN)r6r8AssertionErrorrr:r;r )rr-r3rrrr<s zJoin._set_joinable_configscCsbd}d}|jD]4}|dkr"|j}n||jkr4td|dkr|j}q||_t|j|_||_dS)a Extracts the process group and device information from the joinables. If there are multiple joinables, then the context manager uses the first specified device. Preconditions: ``self._joinables`` is not ``None`` and is non-empty. Raises: ValueError If there are multiple conflicting ``process_group`` attributes among the ``Joinable`` objects. Nz7Using join context manager with multiple process groups) r8r&r7r%_process_groupdistZget_rank_rank_device)r process_groupr)r3rrrr=s  zJoin._extract_dist_infocCsdSrrrrrr __enter__szJoin.__enter__)typevalue tracebackc Cs|jr |rdSd}d}d}d}td|s||krTtd|d|jd |d |}|dkrjd}q(|jrx||jD] } | q~d}|d 7}q(|jD]} | |qdS) z Repeatedly runs the main hooks until all processes join; then, runs the post-hooks. Raises: RuntimeError If ``throw_on_early_termination=True``. NFTrioncez+Detected uneven input skew of greater than z. This means that rank z has at least zz fewer inputs than other currently-active ranks. This level of skew could lead to performance degradation during training.) r:warnings simplefilterwarnrA_get_num_nonjoined_procsr;_notify_procs_to_terminater9rr) rrErFrGZall_procs_joinedriZWARN_THRESHOLDnum_nonjoined_procsr$rrr__exit__s.       z Join.__exit__cCs(tjd|jd}tj||jd|S)z| Returns the number of non-joined processes by shadowing an all-reduce in the non-joined processes. rIr)group)r(zerosrBr@ all_reducer?item)rrPrrrrMszJoin._get_num_nonjoined_procscCs6tjd|jd}tj||jdtd|jddS)z Schedules an all-reduce to notify non-joined processes to terminate and raises a ``RuntimeError`` indicating that the current process has exhausted its inputs. rIrRrSzRank z exhausted all inputs.N)r(onesrBr@rVr? RuntimeErrorrA)rrXrrrrN&szJoin._notify_procs_to_terminate)r3cCst|dstdt|d|j}|jr0|js4dS|j}|j}tj d|d}t j ||dd}|j rtj d|d}t j ||d |}|rtd |S) aO Notifies the join context manager that the calling process has not yet joined; then, if ``throw_on_early_termination=True``, checks if uneven inputs have been detected (i.e. if one process has already joined) and throws an exception if so. This method should be called from a :class:`Joinable` object before its per-iteration collective communications. For example, this should be called at the beginning of the forward pass in :class:`DistributedDataParallel`. Only the first :class:`Joinable` object passed into the context manager performs the collective communications in this method, and for the others, this method is vacuous. Arguments: joinable (Joinable): the :class:`Joinable` object calling this method. Returns: An async work handle for the all-reduce meant to notify the context manager that the process has not yet joined if ``joinable`` is the first one passed into the context manager; ``None`` otherwise. r zCheck that the z/ constructor calls the ``Joinable`` constructorNrIrRT)rTZasync_oprSzLDetected at least one rank that exhausted inputs. Throwing across all ranks.)hasattrr>rEr r-r+r%r&r(rXr@rVr,rUrWrY)r3Z join_configr)rCrXZworkrUZ should_throwrrrnotify_join_context0s&  zJoin.notify_join_context)TF)rrrrrr rrr<r=rDrr BaseExceptionrrQrMrNr0r[rrrrr ps&>   5  )rJabcrrtypesrtypingrrrrr r(Ztorch.distributed distributedr@__all__r r rr rrrrs    -