ȅi4SSKrSSKJrJr SSKJr SSKJrJr SSK r SSK J r /SQr "SS5r"SS \5r"S S \5r"S S 5rg)N)ABCabstractmethod) TracebackType)Any NamedTuple)JoinHookJoinableJoinc4\rSrSrSrS SjrS\SS4SjrSrg) ra This defines a join hook, which provides two entry points in the join context manager. Entry points : 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. returnNcg)zCall this hook while there exists a non-joined process to shadow collective communications in a training iteration. Training iteration i.e., in one forward pass, backward pass, and optimizer step. Nselfs [/home/james-whalen/.local/lib/python3.13/site-packages/torch/distributed/algorithms/join.py main_hookJoinHook.main_hookis_last_joinercg)a Call hook 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)rrs r post_hookJoinHook.post_hook rrrr N) __name__ __module__ __qualname____firstlineno____doc__rboolr__static_attributes__rrrrrs       rrc^\rSrSrSr\S U4Sjj5r\S\4Sj5r\ \S\ R4Sj55r \ \S\ 4Sj55rSrU=r$) r ,aC 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. r cT>[TU]5 [R5UlgN)super__init__ _JoinConfigconstruct_disabled_join_config _join_config)r __class__s rr(Joinable.__init__7s 'FFHrc g)aV Return 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)rkwargss r join_hookJoinable.join_hook<s rcg)zeReturn the device from which to perform collective communications needed by the join context manager.Nrrs r join_deviceJoinable.join_deviceI rcg)zfReturns the process group for the collective communications needed by the join context manager itself.Nrrs rjoin_process_groupJoinable.join_process_groupOr5r)r+r)rrrrr rr(rr0propertytorchdevicer3rr7r" __classcell__)r,s@rr r ,sII  X     U\\   C  rr cH\rSrSr%Sr\\S'\\S'\\S'\S5rSr g) r)VzdThis includes all fields needed from a :class:`Joinable` instance for the join context manager side.enablethrow_on_early_terminationis_first_joinablec[SSSS9$)zReturn 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@rA)r)rrrr**_JoinConfig.construct_disabled_join_config]s Ue  rrN) rrrrr r!__annotations__ staticmethodr*r"rrrr)r)Vs(o L $$  rr)c\rSrSrSrSS\\S\S\4SjjrSSjr SS jr S r S \ \ S-S \ S-S \S-4SjrSrSr\S\4Sj5rSrg)r ha 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 joinablesr?r@c [U5S:Xa [S5eXlURVs/sHoUR"S0UD6PM snUlX lX0lUR5 UR5 gs snf)Nrz7The join context manager requires at least one joinabler) len ValueError _joinablesr0 _join_hooks_enable_throw_on_early_termination_set_joinable_configs_extract_dist_info)rrIr?r@r/joinables rr( Join.__init__sw y>Q VW W#9= 9HX   ( (  +E( ""$ !  sA?Nc[UR5S:deSnURH)n[URURUS9UlSnM+ g)zESet the :class:`_JoinConfig` of each participating :class:`Joinable`.rTrCFN)rKrMr)rOrPr+)rrArSs rrQJoin._set_joinable_configssU4??#a''' H$/||+/+K+K"3%H ! !&  (rc SnSnURH>nUc URnOXR:wa [S5eUbM2URnM@ Xl[ R "UR5UlX lg)as Extract 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) rMr7rLr3_process_groupdistget_rank_rank_device)r process_groupr;rSs rrRJoin._extract_dist_infos~ H$ ( ; ; "="== M~!--(,]]4#6#67  rcgr&rrs r __enter__Join.__enter__srtypevalue tracebackc UR(aU(agSnSnSnSn[R"S5 U(dXg:a)[R"SUSURS US 3S S 9 UR 5nUS:XaSnOKUR (aUR5 URHn U R5 M SnUS - nU(dMURHn U RU5 M g)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.) stacklevel) rOwarnings simplefilterwarnr[_get_num_nonjoined_procsrP_notify_procs_to_terminaterNrr) rrbrcrdall_procs_joinedriWARN_THRESHOLDnum_nonjoined_procsr0s r__exit__ Join.__exit__s||t   f%"! A%&&=zzl.0@A33 !#'"?"?"A "a'#' 33335"&!1!1I'')"2"'Q1#"6))I    /*rc[R"SURS9n[R"XR S9 UR 5$)zaReturn the number of non-joined processes by shadowing an all-reduce in the non-joined processes.rir;group)r:zerosr\rY all_reducerXitem)rrrs rrmJoin._get_num_nonjoined_procss9#kk!DLLA +3F3FG"''))rc[R"SURS9n[R"XR S9 [ SURS35e)zSchedule an all-reduce to notify non-joined processes to terminate. Also raise a ``RuntimeError`` indicating that the current process has exhausted its inputs. rirvrwzRank z exhausted all inputs.)r:onesr\rYrzrX RuntimeErrorr[)rr~s rrnJoin._notify_procs_to_terminate sC zz!DLL1 $7$78U4::,.DEFFrrSc[US5(dS[U5S35eURnUR(aUR(dgUR nUR n[R"SUS9n[R"XCSS9nUR(aK[R"SUS9n[R"XcS 9 UR5nU(a [S 5eU$) a 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`` constructorNrirvT)rxasync_oprwzLDetected at least one rank that exhausted inputs. Throwing across all ranks.)hasattrrbr+rAr?r3r7r:r~rYrzr@ryr{r)rS join_configr;r]r~workry should_throws rnotify_join_contextJoin.notify_join_context)s4x00 d8n-.' ' 0 ++ ,,K4F4F%% 33 zz!F+t4H  1 1KK&1E OOE 7 ::""%) "$ &>30=!D(30t#30!4' 30j* G4h44rr )rjabcrrtypesrtypingrrr:torch.distributed distributedrY__all__rr r)r rrrrsM#"  +  <' s' T * $vvr