o ?߱ie@sddlmZddlZddlZddlZddlZddlZddlZddlm Z m Z ddl m Z ddl mZmZmZmZmZmZddlZddlmZzddlZWn eyWdZYnwddlZddlZddlmZddlm Z dd l!m"Z"m#Z#dd l$m%Z%dd l&m'Z'dd l(m)Z)d_d`ddZ*ddej+fdaddZ,dbd d!Z-dcd$d%Z.ddded*d+Z/ dfdgd1d2Z0dhd4d5Z1d_did8d9Z2ed:ed;efd<Z3Gd=d>d>e)Z4Gd?d@d@e Z5GdAdBdBZ6djdGdHZ7GdIdJdJZ8GdKdLdLZ9dkdOdPZ:dldTdUZ;dmdndYdZZdS)o) annotationsN)ContextDecorator nullcontextfields)AnyCallableListTupleTypeVarUnion)logger)AsyncCollectiveTensor)DTensor) distributedlog)all_gather_tensor)easy_io)TimerTmodeltorch.nn.ModulevalueboolreturnNonecCs|D]}||_qdS)zSet a model to require gradients or not. Args: model (torch.nn.Module): Neural network model. value (bool): Whether the network requires gradients or not. N) parameters requires_grad)rrprT/data/cameron/vidgen/cosmos-predict2.5/cosmos_predict2/_src/imaginaire/utils/misc.pyr/s rdatardevicestr | torch.device | Nonedtypetorch.dtype | None memory_formattorch.memory_formatcs dusdusdusJdttjrOtjkr!dks,tjkr/dkr/tjttr8dkpBttjoBj dk}j | dStt j j rft fddDStt j jrtttfst fd d DSS) aJRecursively cast data into the specified device, dtype, and/or memory_format. The input data can be a tensor, a list of tensors, a dict of tensors. See the documentation for torch.Tensor.to() for details. Args: data (Any): Input data. device (str | torch.device): GPU device (default: None). dtype (torch.dtype): data type (default: None). memory_format (torch.memory_format): memory organization format (default: torch.preserve_format). Returns: data (Any): Data cast to the specified device, dtype, and/or memory_format. Nz@at least one of device, dtype, memory_format should be specifiedcpu)r!r#r% non_blockingc s"i|] }|t|dqSr!r#r%to.0keyr r!r#r%rr es"zto..csg|] }t|dqSr+r-r0elemr,rr gzto..) isinstancetorchTensor channels_lastdimchannels_last_3dpreserve_formatstrr!typer. collectionsabcMappingSequencebytes)r r!r#r%is_cpurr2rr.:s0      r.csttjjrtfddDSttjjr-tttfs-tddDSzt WSt yCtYSw)zSerialize data by hierarchically traversing through iterables. Args: data (Any): Input data. Returns: data (Any): Serialized data. csi|] }|t|qSr serializer/r rrr3vszserialize..cSsg|]}t|qSrrGr4rrrr6xzserialize..) r8rArBrCr@rDr?rEjsondumps TypeErrorrIrrIrrHls   rHenv_vars list[str]c CsZ|D](}|tjvrtdt|dttj|qtdt|dqdS)zPrint a specific list of environment variables. Args: env_vars (list[str]): List of specified environment variables. zEnvironment variable : z not set!N)osenvironrinfoColorgreenyellowwarning)rNenv_varrrrprint_environ_variabless  *rYFseedintby_rankcCsF|r|t7}td|dt|tj|t|dS)zSet random seed. This includes random, numpy, Pytorch. Args: seed (int): Random seed. by_rank (bool): if true, each GPU will use a different random seed. zUsing random seed .N) rget_rankrrSrandomrZnpr9 manual_seed)rZr\rrrset_random_seeds    rbshapeList[int] | Tuple[int] torch.dtypestr | torch.device int | NonecCs2tj|}||tj}t|j||dS)amProduce a GPU-architecture-invariant randomized Torch tensor. Args: shape (list or tuple of ints): Output tensor shape. dtype (torch.dtype): Output tensor type. device (torch.device): Device holding the output. seed (int): Optional randomization seed. Returns: tensor (torch.tensor): Randomly-generated tensor. )r#r!) r`r_ RandomStatestandard_normalastypefloat32r9 from_numpyr.)rcr#r!rZrng random_arrayrrrarch_invariant_rands ro&dict[str, torch.Tensor] | torch.Tensorcs8d fdd |}t|tstd|d||S) zGet the batch size from a data batch, a (possibly hierarchical) dictionary of tensors. Args: data (dict[str, torch.Tensor]): Data batch (dictionary of tensors). Returns: batch_size (int): Data batch size. input_datarrUnion[int, None]cst|tjr t|St|tjjr(|D]\}}|}|dur%|SqdSt|tt frGt|dkrGt|dtjrAt|S|dSdS)z Helper function that recursively finds a tensor in the input data (could be a nested dictionary or list of tensors) and returns its batch size. Nr) r8r9r:lenrArBrCitemslisttuple)rqr1r batch_size_get_batch_sizerrrys   z,get_data_batch_size.._get_batch_sizez Batch size (z) obtained from invalid data: N)rqrrrr)r8r[ ValueError)r rwrrxrget_data_batch_sizes   r{module persistentc Cs~t}|D]\}}|||<q|D](\}}|d}d|dd}||}|d}t|||j|||dqdS)aConvert parameters in a module to buffers. Buffers do not have its own gradients and thus not updated by backpropagation. Args: module (torch.nn.Module): a module to convert parameters persistent (bool): If True, buffers are included in state_dict. r]N)r})dictnamed_parametersrtsplitjoin get_submoduledelattrregister_buffer) r|r} named_paramsnameparamZmodule_hierarchysubmodule_name submodulesubnamerrrparameters_to_buffers    rT.)boundcs$eZdZdZd d fdd ZZS) timerawSimple CPU timer for timing the execution of code. It can be used as either a context manager or a function decorator. The timing result will be logged upon exit. Example: def func_a(): time.sleep(1) with timer("func_a"): func_a() @timer("func_b) def func_b(): time.sleep(1) func_b() Fcontextr?debugrcstj|ddd|ddS)NTFs)tag measure_cpu measure_cudaunitr)super__init__selfrr __class__rrrs ztimer.__init__Frr?rr)__name__ __module__ __qualname____doc__r __classcell__rrrrrsrc@s:eZdZdZddddZdd d Zdd dZdddZdS)memory_checkeraSimple memory checker for a given block of code. It can be used as either a context manager or a function decorator. The memory usage will be logged upon exit. Example: def func_a(): torch.rand([int(1024**2)]).float().cuda() with memory_checker("func_a"): func_a() >>> 0.004GB memory used @memory_checker("func_b") def func_b(): random_var = torch.rand([int(1024**2)]).cuda() func_b() Frr?rrcCs||_||_dSN)rrrrrrr"s zmemory_checker.__init__rrcCs$tjtjtj|_dSr)r9cuda synchronizereset_peak_memory_statsmax_memory_allocatedinitial_memoryrrrr __enter__&s  zmemory_checker.__enter__cCsVtjtj}d|jd||jddd}|jr$t|dSt|dSNzMemory used within rPi@z.4fz GB) r9rrrrrrrrS)rexc_type exc_value traceback final_memorymessagerrr__exit__+s   zmemory_checker.__exit__funcrcstfdd}|S)Ncstjtjtj}|i|}tjtj}djd||ddd}jr9t||St||Sr) r9rrrrrrrrS)argskwargsrresultrrrrrrwrapper5s       z(memory_checker.__call__..wrapper) functoolswraps)rrrrrr__call__4szmemory_checker.__call__Nrrrr)rrrr)rrrrrrrrrrrrrs     rc@s`eZdZdZdddZdddZddd Zdd d ZdddZdddZ dddZ dddZ dS) TrainingTimera_Timer for timing the execution of code, aggregating over multiple training iterations. It is used as a context manager to measure the execution time of code and store the timing results for each function. The context managers can be nested. Attributes: results (dict): A dictionary to store timing results for various code. Example: timer = Timer() for i in range(100): with timer("func_a"): func_a() avg_time = sum(timer.results["func_a"]) / len(timer.results["func_a"]) print(f"func_a() took {avg_time} seconds.") rrcCs(t|_t|_g|_g|_|dSr)rresultsZaverage_resultstimers func_stackresetrrrrrYs  zTrainingTimer.__init__cCsdd|jD|_dS)NcSsi|]}|gqSrrr/rrrr3asz'TrainingTimer.reset..)rrrrrr`szTrainingTimer.resetcCs(tddddd}|j|||S)NTFr)rrrr)rrappendstart)rrrrrrcs zTrainingTimer.__enter__cCsF|j}||}|j}|j|g|j||dSr)rpopendZ get_cpu_timerr setdefaultr)rrrrrrr1rrrris  zTrainingTimer.__exit__ func_namer?cCs|j||Sr)rrrrrrrrqs zTrainingTimer.__call__cC ||Srrrrrr __getattr__u zTrainingTimer.__getattr__cCrrrrrrrnestedxrzTrainingTimer.nesteddict[str, float]cCs2t}|jD]\}}t|t|||<q|Sr)rrrtsumrs)rrr1Z value_listrrrcompute_average_results{sz%TrainingTimer.compute_average_resultsNr)rr)rr?rr)rr) rrrrrrrrrrrrrrrrrGs       rtimeout_periodfloatsignumframecCs8d|d}trddl}|jd||jjdt|)NzTimeout error: more than z) seconds passed since the last iteration.rzTimeout error!)titletextlevel)ris_rank0wandbZalertZ AlertLevelERROR TimeoutError)rrr error_messagerrrrtimeout_handlers rc@sreZdZdZedddZedddZedd d Zedd d Zedd dZ edddZ edddZ dS)rTzA convenience class to colorize strings in the console. Example: import print("This is {Color.red('important')}.") xr?rcCtjt|ddS)Nredcolor termcolorcoloredr?rrrrrz Color.redcCr)NrUrrrrrrrUrz Color.greencCr)Nbluerrrrrrrrz Color.bluecCr)Ncyanrrrrrrrrz Color.cyancCr)NrVrrrrrrrVrz Color.yellowcCr)Nmagentarrrrrrrrz Color.magentacCr)Ngreyrrrrrrrrz Color.greyN)rr?rr?) rrrr staticmethodrrUrrrVrrrrrrrTs       rTc@s8eZdZdZd ddZdddZed d Zd d ZdS) BufferCnta Buffer counter which keeps track of the condition when called and returns True when the condition in met "thres" amount of times, otherwise returns False. Example usage: buf = BufferCnt(thres=3) for _ in range(5): if buf(random.random() > 0.5): print("We got lucky 3 times out of 5.") Args: thres (int): The amount of times the expression needs to be True before returning True. reset_over_thres (bool): Whether to reset the buffer after returning True. FcCsd|_||_||_dSNr)_cntthresreset_over_thres)rrrrrrr zBufferCnt.__init__NcCsL|dur |jd7_nd|_|dur|j}|j|kr$|jr"|dSdS)NTrF)rrrr)rZexprerrrrrs zBufferCnt.__call__cCs|jSrrrrrrcntsz BufferCnt.cntcCs d|_dSrrrrrrrrzBufferCnt.reset)rFr) rrrrrrpropertyrrrrrrrs    r dataclassrcsfddtDS)zConvert a dataclass to a dictionary. Args: dataclass (Any): Dataclass object. Returns: dict: Dictionary representation of the dataclass. csi|] }|jt|jqSr)rgetattr)r0frrrr3r7z.dataclass_instance_to_dict..rrrrrdataclass_instance_to_dicts rtensortorch.Tensor | DTensor torch.tensorcCs,t|tr|}t|tr|S|S|Sr)r8rZto_localrwait)rlocalrrrget_local_tensor_if_DTensors  rrecompile_limituse_duck_shapec Csrz|tjj_|tjjj_WdSty8z|tjj_ |tjjj_WYdSty7}zt d|d}~www)a Set some of the torch compile config options. The default values of arguments are default config values in PyTorch as of 2.10 version. The value recompile_limit=32 is useful for Wan Tokenizer encoding compilation, as the standard value of 8 can easily overflow. The value of use_duck_shape=False is useful for Cosmos3 MoT training to reduce recompilations. Args: recompile_limit (int): Controls the maximum number of cache entries with a guard on same ID_MATCH'd object. use_duck_shape (bool): This flag changes whether we should use the same symbolic variable to represent input sizes that are the same z=torch.compile is not available due to missing config options.N) r9_dynamoconfigrfx experimental_configrAttributeErrorZcache_size_limitrrW)rrerrrset_torch_compile_optionss    rc@s,eZdZdZdddd Zd d Zd d ZdS)NVTXRangeContexta Context manager which inserts NVTX range around the current context and optionally calls torch.cuda.synchronize at the start and the end of the context. Args: name (str): Name of the NVTX range. enabled (bool): Whether the context manager is enabled. When disabled, it does nothing. Default: True. synchronize (bool): Whether to call torch.cuda.synchronize() at the start and the end of the context. Default: True. Trr?enabledrrcCs||_||_||_dSr)rrr)rrrrrrrrrzNVTXRangeContext.__init__cCs.|jsdS|jr tjtjj|jdSr)rrr9rnvtx range_pushrrrrrrs  zNVTXRangeContext.__enter__cCs*|jsdS|jr tjtjjdSr)rrr9rr range_pop)rrexc_valexc_tbrrrr$s  zNVTXRangeContext.__exit__N)TT)rr?rrrr)rrrrrrrrrrrr s   rc@sJeZdZdZ     ddddZddZdd ddZddZddZdS)!StragglerDetectorV2aStragglerDetectorV2 is a class that allows you to easily integrate "straggler" tool: https://gitlab-master.nvidia.com/dl/gwe/fault_tolerance_related/straggler/-/tree/cupti?ref_type=heads. This tool detects stragglers using low-level CUPTI tool, which can gather kernel execution time with very low overhead. The execution times are compared across different ranks, as well as to the execution time of the exact same kernels in the past. This tool can be easily integrated, as it's resilient to any synchronizations, since it captures kernels execution time. It means that we can wrap the entire forward or backward passes and the stragglers will be identified regardless of synchronizations happening during the iteration. Args: enabled (bool): Whether the straggler detection is enabled. When disabled, it does nothing. Default: True. report_freq (int): Generate a report each report_freq iterations that analyzes the GPUs performance. Defaults to 100. profile_freq (int): Enable the CUPTI profiling each profile_freq iterations. Since the overhead is very low, the default value is 1. max_diff (float): Defines the maximum relative difference between the fastest and the slowest rank to determine the slowdown. Defaults to 2.0 raise_error (bool): Whether to raise error when stragglers are detected enough times. Defaults to True.Tdr@rr report_freqr[ profile_freqmax_diffr raise_errorcCs:||_||_||_|jj|_tddd|_||_||_ dS)NrT)rr) rrrrrrrslowdown_countrr)rrrrrrrrrr>s  zStragglerDetectorV2.__init__cCs2|jrts tdtjjddgd|jddSdS)NzPlease install straggler package before using StragglerDetectionV2.Package can be installed from here: https://gitlab-master.nvidia.com/dl/osiris/stragglerZrelative_perf_scoresZindividual_perf_scoresF)Zscores_to_computeZgather_on_rank0Zprofiling_interval)r straggler RuntimeErrorDetector initializerrrrrr!Ns zStragglerDetectorV2.initializerr?section_enabled profile_cudacCs |r |jr tjj||dStS)N)r#)rrr Zdetection_sectionr)rrr"r#rrrprofile_section\s z#StragglerDetectorV2.profile_sectioncCs<g}|D]}|||tjjdqtt|S)Ni) rrZ StatisticMAXrrr9rr)rlocal_section_summariesr r1rrr_aggregate_section_resultsbsz.StragglerDetectorV2._aggregate_section_resultsc sjrF|jdkrHtj}|jt}tt |g } |j }tdkrJ|jdjd}fddt|D}t|j D]\|fddt|DqIt |}t |} |djd| djd t |it|j D]+\t fd d |D}|d dt |d d t |iqddl} | jr| j||dddl} | jjjjj r|djdkrt!"|dj#j$d|ddt!"|dj#j$d|dd|d} |d} | p| }|r+t %t&t'(j)d }t|}|| *+,-d}t./d|0|rLj1rNt2d| d| d|j| ddSdSdSdSdS)Nrr)Zgpu_rel_thresholdcs*i|]\}}jd||dqS)z/relative_gpu_perf_rritem)r0rankZperfrrrr3ssz7StragglerDetectorV2.generate_report..cs2i|]\}}jdd|d|qS)/_03dr()r0r*vr1key_idrrrr3ys2z slowest_rank/_rankZ_relative_perfcsg|]}|qSrr)r0r.)r0rrr6rJz7StragglerDetectorV2.generate_report..zslowest_rank/slowest__time)stepr(z s3://rundir/z/iter_Z09dz.yamlz /report_iter_z.pklZstraggler_gpus_relativeZstraggler_gpus_individualzutf-8zSlowest rank hostname: z Detected GPU zD to be too slow compared to other GPUs. The relative performance of z rank was z. Terminating the training.)3rrrr generate_reportZgpu_relative_perf_scoresrr^rr9rrr'r&Zidentify_stragglersr enumerateupdateargminrr)minargmaxmaxrrunrZ,cosmos_predict2._src.imaginaire.utils.launch_src imaginaireutilsZlaunchZS3_READYrdumprr ByteTensor bytearrayrQunamenodenamer)numpytobytesdecodeloggingcriticalrrr)r iterationreportZgpu_relative_perf_scoreZ#gpu_relative_perf_score_gather_listZlocal_section_dataZ stragglersZ wandb_infoZ data_tensorZslowest_rank_idrcosmos_predict2Zrelative_stragglersZindividual_stragglersZ is_slowdownhostnameZwhole_hostnameZslowest_hostnamerr/rr4is|       9z#StragglerDetectorV2.generate_reportN)TrrrT) rrrr[rr[rrrrT)rr?r"rr#r) rrrrrr!r$r'r4rrrrr,s   rrM)rrrrrr) r rr!r"r#r$r%r&rr)r rrr)rNrOrrr)rZr[r\rrrr)rcrdr#rer!rfrZrg)r rprr[)r|rr}r)rrrr[rr[rr)rrrr)rrrr)rT)rr[rr)? __future__rrAcollections.abcrrKrQr_ contextlibrr dataclassesrtypingrrr r r r rDr`logurur rGr ImportErrorrr9)torch.distributed._functional_collectivesrZtorch.distributed._tensor.apir%cosmos_predict2._src.imaginaire.utilsrrZ1cosmos_predict2._src.imaginaire.utils.distributedrZ-cosmos_predict2._src.imaginaire.utils.easy_iorZ+cosmos_predict2._src.imaginaire.utils.timerrrr>r.rHrYrbror{rrrrrrrTrrrrrrrrrrsb              2    &6 ; % -