U 9%e!S@sddlZddlZddlmZmZddlmZmZmZddl Z ddl m Z ddl m Z ddl mZmZddlmZeeZe jedZd d ZeGd d d Zeed ddZeeedddZddZGdddZdS)N) dataclassfield)AnyListOptional)fx)GraphCompileReason)deepcopy_to_fake_tensordetect_fake_mode)NodeZ ddp_graphscCspt|rd|jdSt|tr>dddd|DdSt|trdddd d|DdSt|SdS) NzT[]ztuple(z, cSsg|] }t|qSargs_str.0xr r a/var/www/html/Darija-Ai-API/env/lib/python3.8/site-packages/torch/_dynamo/backends/distributed.py szargs_str..)zlist(cSsg|] }t|qSr rrr r rrs)torchZ is_tensorshape isinstancetuplejoinliststr)argsr r rrs   rc@sreZdZUdZeed<eedZe e ed<eedZ e e j ed<eedZe ed<dZeed<dZeed<d S) Bucketrsize)default_factoryparamsnodes param_ids,opcount_increased_to_capture_external_output!paramsize_before_opcount_increaseN)__name__ __module__ __qualname__rint__annotations__rrr!rrr"rr r#r$r%r r r rrs   r)bucketreturncCs<t}|jD]*}|||jD]}||kr dSq q dS)NTF)setr"addZusers)r+Znodes_in_bucketnodeuserr r rbucket_has_external_output*s    r1)bucketsbucket_bytes_capc Cs6d}g}g}tt|D]v\}}t|jdkrj|||j|jdf|jddD]}|dd|fqT|jdkr|||j|j|jfqt|r(t d|t|t|rt dzFddl m }t d|||dd t|rt d ||d dd Wn t k r$t d YnXn t d dS)N)IndexzSize (b)z Param Namesrzd DDPOptimizer used bucket cap %s and created %d buckets. Enable debug logs for detailed bucket info.aSome buckets were extended beyond their requested parameter capacities in order to ensure each subgraph has an output node, required for fx graph partitioning. This can be the case when a subgraph would have only contained nodes performing inplace mutation, and returning no logical outputs. This should not be a problem, unless it results in too few graph partitions for optimal DDP performance.)tabulatez; DDPOptimizer produced the following bucket assignments: %sZ simple_grid)headersZtablefmtzKDDPOptimizer extended these buckets to ensure per-subgraph output nodes: %s)r4z Extra OpszExtra Param Size (b)z^Please `pip install tabulate` in order to display ddp bucket sizes and diagnostic information.zADDPOptimizer captured no parameters and did not split this graph.) enumeratereversedlenr!appendrr$r%loginfowarningr6debug ImportError) r2r3r7rowsZextended_bucketsidxr+paramr6r r rpretty_print_buckets7sZ       rDcCs<|jjD].}|jdkrt||j}t|tjjrdSqdS)Nget_attrTF) graphr"opgetattrtargetrrr GraphModule)gmr/ maybe_paramr r rhas_higher_order_oprs    rMc@sFeZdZdZd eeedddZddZej e e j dd d Z dS) DDPOptimizeraNote [DDPOptimizer] DDPOptimizer applies when dynamo compiles models wrapped in DistributedDataParallel (DDP), breaking the dynamo graph into chunks to compile separately, with the breaks aligning to the boundaries of gradient-allreduce buckets chosen by DDP. Background/Motivation - DDP uses allreduce collectives to synchronize partial gradients computed on different workers - DDP groups gradient allreduces into 'buckets' to optimize communication efficiency of all-reduce - Parameters grouped into buckets are assumed to be adjacent in time, so they become ready at around the same time during backward and thus can share the same allreduce efficiently - Allreduces must overlap with backward compute for optimal training performance - DDP schedules allreduces using 'hooks' fired from the c++ autograd engine in pytorch, which operates when individual grads become 'ready' - Dynamo+AOTAutograd produces a single fused graph that runs 'atomically' from the perspective of the autograd engine, such that all gradients become 'ready' at the same time. Hooks fire after the whole fused backward function executes, preventing any overlap of compute and communication Algorithm - DDPOptimizer starts off with an FX graph traced by dynamo which represents forward. It can traverse this graph in reverse order to determine the true order that gradients will become ready during backward. - Parameter sizes are counted in reverse order, up to a bucket size limit, at which point a new bucket is started and a graph break introduced - Each of the subgraphs is compiled by the compiler provided to dynamo by the user, and then fused back together into an outer module that is returned to the user Notes - It would be better to enforce (by adding an API to DDP) that the bucket splits chosen here are used by DDP, and that DDP does not need to detect or optimize bucket order by observing execution at runtime, as it does in eager. - If Dynamo can't capture a whole graph for the portion of the model wrapped by DDP, this algorithm will currently produce splits that do not necessarily align with the buckets used by DDP. This should result in performance degradation approaching the baseline case where graph-splits are not used, but not worse. - If the backend compiler fails to compile a single subgraph, it will execute eagerly despite the rest of the subgraphs being compiled - DDP has a 'parameters_and_buffers_to_ignore' field, which DDPOptimizer attempts to honor by reading markers left by DDP on individual parameters. In cases where other transformations, such as reparameterization, are also used, the ignore markers could be lost. If DDPOptimizer fails to ignore a parameter ignored by DDP, it is not catastrophic but could impact performance by choosing sub-optimal bucket splits. - DDPOptimizer always ignores all buffers, regardless of their ignore flag, since buffers do not require gradients, and therefore aren't allreduced by DDP. (They are broadcast during forward, but this is not covered by DDPOptimizer) Debugging - Generally, it is easiest to debug DDPOptimizer in a single process program, using pdb. - In many cases, the log messages are helpful (they show bucket size assignments)- just configure torch._dynamo.config.log_level to info or debug. - See `benchmarks/dynamo/distributed.py` for a simple harness that will run a toy model or a torchbench model in a single process (or with torchrun, in multiple processes) Args: bucket_bytes_cap (int): Controls the size of buckets, in bytes, used to determine graphbreaks. Should be set to match the equivalent parameter on the original DDP module. backend_compile_fn (callable): A dynamo compiler function, to be invoked to compile each subgraph. first_bucket_cap (int): Controls the size of the first bucket. Should match DDP's first bucket cap. DDP special-cases the first bucket size since it is sometimes optimal to start a small allreduce early. N)r3first_bucket_capcCsP|dk r||_ntjr&tjj|_n||_||_|j|jksFtd||_dS)NzQFirst bucket should be smaller/equal to other buckets to get comms warmed up ASAP)rOr distributedZ is_availableZ_DEFAULT_FIRST_BUCKET_BYTESr3AssertionErrorbackend_compile_fn)selfr3rRrOr r r__init__s   zDDPOptimizer.__init__cCst|do|jS)N _ddp_ignored)hasattrrU)rSZ parameterr r r_ignore_parameterszDDPOptimizer._ignore_parameter)rKexample_inputscsPt|dkrtjjt|r,tdtg}t|j j D]}|j dkrRq@|dj |j ks~t|dkr|dj |jkrt|dr|dtn0|djdkr|dj |d_|djd7_|j dkrP||j}|D]b\}}|jr||s|dj |7_ |dj|jd||djt|qnl|j dkrt||j}|jr||s|dj |7_ |dj|j|djt||dj |q@t|dkr$|dj dkr$|dj |dj t|djdkst!d |d=||_"t#||j t|dkrP|$||Sit%|D] \} } | j D]}| |<qjq\t&j'j((|dfd d } d |j d d| j d } | )D]6\}} d|krt|r| d|d| j d 7} q| d7} t*+| Gfdddtj&j,j-}|| |j$}|j.|| /t*+d| j | S)a1 Implements graph splitting, first determining a set of of buckets by counting parameter sizes in reverse graph order, then invoking the user/backend compiler to compile each subgraph. Finally, stiches compiled graphs into one graphmodule and returns its callable. Na]DDPOptimizer backend: Found a higher order op in the graph. This is not supported. Please turn off DDP optimizer using torch._dynamo.config.optimize_ddp=False. Note that this can cause performance degradation because there will be one bucket for the entire Dynamo graph. Please refer to this issue - https://github.com/pytorch/pytorch/issues/104674.)output placeholderrr5 call_module_rEz#Params should be empty if size is 0cs|SNr )r/) partition_mapr r,z)DDPOptimizer.compile_fn..z ---orig graph---  z ---split graph--- .z ---z graph--- z --------------- cs8eZdZfddZddZeedfdd ZZS)z/DDPOptimizer.compile_fn..SubmodCompilercst|||_dSr])superrTcompiler)rSmodulerd __class__r rrT<s z8DDPOptimizer.compile_fn..SubmodCompiler.__init__cSst|dkstdGdddtjj}d}|jjD],}|jdkr4t|j dt s4d}|j f|_ q4| t dt tdtg|_|||||}|S) z Compile the submodule, using a wrapper to make sure its output is always a tuple, which is required by AotAutograd based compilers rz%We assume only args for these modulescs$eZdZfddZddZZS)zUDDPOptimizer.compile_fn..SubmodCompiler.compile_submod..WrapperModulecst||_||_dSr])rcrTsubmodunwrap_singleton_tuple)rSrhrirfr rrTIs z^DDPOptimizer.compile_fn..SubmodCompiler.compile_submod..WrapperModule.__init__cWs*|j|}|jr&t|ttfr&|dS|S)Nr)rhrirrr)rSrrr r rforwardNs z]DDPOptimizer.compile_fn..SubmodCompiler.compile_submod..WrapperModule.forward)r&r'r(rTrj __classcell__r r rfr WrapperModuleHs rlFrYTz{DDPOptimizer intentional graph-break (See Note [DDPOptimizer]). Set `torch._dynamo.config.optimize_ddp = False` to disable.)r:rQrnnModulerFr"rGrrr recompiler traceback FrameSummary__file__rNZcompile_subgraph_reasonrd)rSZ input_modrkwargsrlriZsnwrapperr r rcompile_submod@s&     z>DDPOptimizer.compile_fn..SubmodCompiler.compile_submod)nr,c sB||\}}g}st|D]:}t|tjrNt|tjjsN||q||qt d|j |j t |t|tstt|tst|j dkr(||j }rt|}n|}t d|j |j||||}|j|j d|j |_ |j|j ||||W5QRSQRXnt||j |j ||SdS)Nzrun_node %s, %s got args %sr[z ---%s graph--- %sZ compiled_)Zfetch_args_kwargs_from_envrQrrTensor _subclassesZ FakeTensorr;Z from_tensorr<r?rGrIrrdictZ fetch_attrr ddp_graph_logrFrureZdelete_submoduleZ add_submodulerH) rSrvrrsnew_argsargZreal_modZ curr_submodZcompiled_submod_real fake_moder rrun_nodesD     z8DDPOptimizer.compile_fn..SubmodCompiler.run_node) r&r'r(rTrur rrrkr r}rfrSubmodCompiler;s Brz& ---final graph--- %s --------------- )0r rrxZ fake_tensorZFakeTensorModerMNotImplementedErrorrr9rFr"rGrr3r:rOr1insertr$r%Z get_submodulerIZnamed_parametersZ requires_gradrWZuntyped_storagenbytesr!r;r#idrHextendrQr2rDrRr8rZpassesZ split_moduleZ named_modulesrzr? interpreterZ Interpreterrunro)rSrKrXr2r/rInamerCrLrBbZsplit_gmZ debug_strrerZsubmod_compilerr )r~r^r compile_fns                 x  zDDPOptimizer.compile_fn)N)r&r'r(__doc__r)rrTrWrrJrrrwrr r r rrN|s@ rN) loggingrp dataclassesrrtypingrrrrrZtorch._dynamo.output_graphrZtorch._dynamo.utilsr r Z torch.fx.noder getLoggerr&r<Z_loggingZgetArtifactLoggerrzrrboolr1r)rDrMrNr r r rs"       ;