sgBUddlZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl Z ddl Z ddlZddlmZddlmZddlmZddlmZddlmZmZmZddlmZddlmZmZm Z m!Z!m"Z"m#Z#m$Z$m%Z%dd l&m'Z'ddl(Z(ddl)Z(ddl*Z(ddl+m,Z-ddl.m/Z/dd l0m1Z1m2Z2m3Z3m4Z4m5Z5m6Z6m7Z7m8Z8m9Z9m:Z:dd l;mZ>ddl?Z?ejej ejeCZDGd deZEideEdddeEdddeEdddeEdddeEdddeEdd d!eEd"d#d$eEd%d&d'eEd(d)d*eEd+d,d-eEd.d/d0eEd1d2d3eEd4d5d6eEd7d8d9eEd:d;d<eEd=d>d?eEd@dAZFeGdBdCZGdDZHdEZIdFZJdGZKdHZLdIZMdJZNdKZOdLZPdMZQdNZRdOZSdPZTdQZUdRZVdSZWdTZXdUZYe4dVdWdXedYZdXd[dXfd\ZZe8rd]Z[ne\ejd^d_Z[d`daiZ^e7rdbe^dc<dddZ_dee\fdfZ`edgZaddhe\die\dje\fdkZbdie\dlecfdmZddaee e jegdn<ddoe ecdedfdpZhddqZidrZjGdsdte9Zkdueece"e#fdve$dwe#fdxZldamdeenfdyZodzZpde[ejfd{ZqGd|d}e9ZrGd~de/jZtGdde/jZueddZvGdde(jjjrZyGddekZzGdde9Z{y)N)contextmanager) dataclass) timedelta)Enum)partialreducewraps)StringIO)Dict NamedTupleOptionalUnionListAnyCallableTuple)patch) FILE_SCHEMAfind_free_port IS_SANDCASTLEretry_on_connect_failuresskip_but_pass_in_sandcastleskip_but_pass_in_sandcastle_ifTEST_WITH_ROCMTEST_WITH_TSANTestCase run_tests)_install_threaded_pg_uninstall_threaded_pgProcessLocalGroup)levelc"eZdZUeed<eed<y)TestSkip exit_codemessageN)__name__ __module__ __qualname__int__annotations__str]/var/www/html/venv/lib/python3.12/site-packages/torch/testing/_internal/common_distributed.pyr#r#7s N Lr-r#backend_unavailableHz5Skipped because distributed backend is not available.small_worldsizeIz Skipped due to small world size. odd_worldsizeWzSkipped due to odd world size.no_cudaJzCUDA is not available.z multi-gpu-1KzNeed at least 1 CUDA devicez multi-gpu-2MzNeed at least 2 CUDA devicesz multi-gpu-3PzNeed at least 3 CUDA devicesz multi-gpu-4QzNeed at least 4 CUDA devicesz multi-gpu-5RzNeed at least 5 CUDA devicesz multi-gpu-6SzNeed at least 6 CUDA devicesz multi-gpu-7TzNeed at least 7 CUDA devicesz multi-gpu-8UzNeed at least 8 CUDA devicesncclLz#c10d not compiled with NCCL support skipIfRocmNzTest skipped for ROCmno_peer_accessOz'Test skipped because no GPU peer accessgenericVzHTest skipped at subprocess level, look at subprocess log for skip reason importerrorXz"Test skipped due to missing importceZdZiZhded<eed<ddhed<ddhed<iZhded <hded <hded <hded <eed <y) DistTestCases>mpiuccr?allgather_coalescedrr?rLzsendrecv anysourcez cpu barrier>rLgloor?gpucudaddpsubgrouppluginN)r&r'r(skip_collectivesetbackend_featurer,r-r.rJrJUsxO-CO)* #OH-3UOO()&,e_OM"O4OE5OF4OE"9OJ #OHr-rJc.tfd}|S)zSkips if the world size exceeds the number of GPUs, ensuring that if the test is run, each rank has its own GPU via ``torch.cuda.device(rank)``.chtjjs&tjt dj ttjd}tjj|kr)tjt d|j |i|S)Nr5 WORLD_SIZE multi-gpu-) torchrP is_availablesysexit TEST_SKIPSr$r)osenviron device_count)argskwargs world_sizefuncs r.wrapperzskip_if_no_gpu..wrapperkszz&&( HHZ *44 5L12 :: " " $z 1 HHZ*ZL 9:DD ET$V$$r-r rfrgs` r.skip_if_no_gpurjgs" 4[%% Nr-c.tfd}|S)Nctjddk7rEttjddkr&tjt dj |i|S)NBACKENDrKrYr1r`rar)r]r^r_r$rcrdrfs r.rgz(skip_if_small_worldsize..wrapperysR JJy !U *BJJ|4L0MQR0R HHZ 12<< =T$V$$r-rhris` r.skip_if_small_worldsizerqx  4[%% Nr-c.tfd}|S)Nctjddk7rHttjddzdk(r&tjt dj |i|S)NrmrKrYrnr3rorps r.rgz&skip_if_odd_worldsize..wrappersW JJy !U *BJJ|4L0MPQ0QUV0V HHZ0:: ;T$V$$r-rhris` r.skip_if_odd_worldsizervrrr-cfd}|S)Nc4tfd}|S)Ncdk(rKtjjkr*tjt dj y|i|SNr?rZ)r[rPrbr]r^r_r$)rcrdbackendrfns r.rgzCrequire_n_gpus_for_nccl_backend..decorator..wrappersM& UZZ%<%<%>%Bj$45??@T,V,,r-rh)rfrgr{r|s` r. decoratorz2require_n_gpus_for_nccl_backend..decorators t -  - r-r,)r|r{r}s`` r.require_n_gpus_for_nccl_backendr~s r-c d}|S)Nc.tfd}|S)Nc ddlm}m}|i|S#t$r)t j t djYywxYw)Nr)AutoModelForMaskedLM BertConfigrG) transformersrr ImportErrorr]r^r_r$)rcrdrrrfs r.rgz?import_transformers_or_skip..decorator..wrappersE > T,V,, >M2<<= >s/AArhris` r.r}z.import_transformers_or_skip..decorators t >  >r-r,)r}s r.import_transformers_or_skiprs  r-ctjjxr!tjj|k\SN)r[rPr\rb)xs r.at_least_x_gpurs, :: " " $ G)@)@)Ba)GGr-cfd}|S)Nc2tfd}|S)Nctjjr)tjjk\r|i|St j t djy)NrZr[rPr\rbr]r^r_r$)rcrdrfrs r.rgz4skip_if_lt_x_gpu..decorator..wrappersUzz&&(UZZ-D-D-F!-KT,V,, HHZ*QC 01;; .decorators t =  = r-r,)rr}s` r.skip_if_lt_x_gpurs r-cfd}|S)Nc4tfd}|S)Ncdk7r|i|Stjjr)tjjk\r|i|St j t djyrzr)rcrdr{rfrs r.rgz9nccl_skip_if_lt_x_gpu..decorator..wrappersm& T,V,,zz&&(UZZ-D-D-F!-KT,V,, HHZ*QC 01;; .decorators t =  =r-r,)r{rr}s`` r.nccl_skip_if_lt_x_gpurs  r-c|j}d|vsJd|vsJd|vsJ|d}|jddk(r|n|jdd}||vs Jd|d|y) N iteration has_errorerrorz Exception raised from rzDid not find expected z in ddp logging data error: )_get_ddp_logging_datafindsplit) model_DDP err_substrddp_logging_data logging_erractuals r.verify_ddp_error_loggedrs 668 * ** * * ** * & && &"7+K ??5 6" <    8 9! < +R x'CK=QRr-c.tfd}|S)aJ Convenience decorator to set/unset TORCH_NCCL_BLOCKING_WAIT flag. Note that use of this decorator will override the setting of TORCH_NCCL_ASYNC_ERROR_HANDLING for the particular test. After the test, both TORCH_NCCL_BLOCKING_WAIT and TORCH_NCCL_ASYNC_ERROR_HANDLING will be restored to their original values. c tjd}tjd= tjd}dtjd< |i|}|||tjd<||tjd<SS#t$rd}YjwxYw#t$rd}YgwxYw#dtjd<wxYw#||tjd<||tjd<wwxYw)NTORCH_NCCL_ASYNC_ERROR_HANDLINGTORCH_NCCL_BLOCKING_WAIT1)r`raKeyError)rcrd cached_nccl_async_error_handlingcached_nccl_blocking_waitretrfs r.rgz(with_nccl_blocking_wait..wrappers 4AC1B , <=  9:<***; % 69BJJ1 2 S''C0;5 5)49R 5651 4/3 , 4 -(, % -69BJJ1 20;5 5)49R 565s@$BB B> BB B# B&"B##B&&B;>-C+rhris` r.with_nccl_blocking_waitrs% 4[ S SD Nr-cfd}|S)zK Runs a test for each distributed debug level specified in levels. c2tfd}|S)Nctjjdd}D][}|tjd<tj|i|}tj |I|tjd<]S)NTORCH_DISTRIBUTED_DEBUG)r`ragetc10dset_debug_level_from_envbarrier)rcrd old_levelr!rrflevelss r.rgz:with_dist_debug_levels..decorator..wrappersx '@$GI F8= 45--/D+F+ (.decorators t   r-r,)rr}s` r.with_dist_debug_levelsrs $ r-c@ttj dS)Nz+c10d was not compiled with the Gloo backend)rris_gloo_availabler,r-r. requires_gloor*! )  " " $$5 r-c tjs tdSttj j j|kd|dtj j jd|S)N+c10d was not compiled with the NCCL backendz0Requires NCCL version greater than or equal to: z , found: z , reason: )ris_nccl_availablerrr[rPr?version)rmsgs r.requires_nccl_versionr1sv  ! ! #* 9  . JJOO # # % />wiyQVQ[Q[Q`Q`QhQhQjPkkuvyuz {  r-c@ttj dS)Nr)rrrr,r-r. requires_ncclr=rr-c@ttj dS)Nz*c10d was not compiled with the UCC backend)rris_ucc_availabler,r-r. requires_uccrC! )  ! ! ##4 r-c@ttj dS)Nz*c10d was not compiled with the MPI backend)rris_mpi_availabler,r-r. requires_mpirIrr-c<d_tfd}|S)zSkips a test for ROCmTcnts|i|Stjtdjy)NrA)rr]r^r_r$rps r.rgz*skip_if_rocm_multiprocess..wrapperTs-(( ( L)334r-)skip_if_rocm_multiprocessr ris` r.rrPs(%)D" 4[55 Nr-c<ttjdk(dS)Nwin32z8This unit test case is not supported on Windows platform)rr]platformr,r-r. skip_if_win32r]s ) B r- localhostruT)minutesFct}|rEt|tdz }tjj j |||||Stj ||||||S)zL Creates a TCP store. Retries if the chosen port is already in use. ru) milliseconds)wait_for_workers use_libuv)rr)rr[classes dist_c10dTCPStorer) addrre is_mastertimeoutr jit_classrporttimeout_milliseconds r.create_tcp_storerdso  D!'I1,E"EF}}&&// $ I/B  }} $ I@P\e  r-i!DISTRIBUTED_TESTS_DEFAULT_TIMEOUT300test_ddp_uneven_inputsitest_join_kwargsctjdk(s| tjj dStjj |S)Nrz 127.0.0.1)hostname interface)r]rrProcessGroupGloo create_devicers r.rrsG ||w)"3$$22K2HH$$22Y2GGr-returncZtj|jddtSN.r)TIMEOUT_OVERRIDErrTIMEOUT_DEFAULT)test_ids r. get_timeoutrs#    c 22 6 HHr-c#NKtt}}tjtj}} ||ct_t_tjtjf||ct_t_y#||ct_t_wxYwwr)r r]stdoutstderr)new_outnew_errold_outold_errs r.captured_outputrslz8:WGzz3::WG2!(' CJjj#**$$!(' CJ' CJs5B%9B 1B% B""B%rankre num_inputsc d dtdtdtdtfd}dtfd}t|dt|d t|d t|d t|d t|d fDcgc]N}t|Dcgc]}|||z|z||zc}t|Dcgc]}||||zc}fPc}}Scc}wcc}wcc}}w) z Generate a number of basic test cases for sparse reduction. These cover tensors with a varying number of sparse dimensions and a varying number of dense dimensions. The only reduction operation we support is sum. rurre sparse_dims dense_dimsc tjtj|dzd|dzf}|gt|Dcgc]}dc}z}t|dz D]A}tj|tj d|dzf}|j |Ctj|dzgt|Dcgc]}dc}z}tj|||Scc}wcc}w)Nrurn) r[reshapearangerangecatzerosappendonessparse_coo_tensor)rrerrindices_shapevaluess r.generatez,simple_sparse_reduce_tests..generates-- TAX 6D1H F 5+<=a=={Q' %Aii%++a*B CDG LL $ %TAXJU:5F)G!)GGH&&w>> >*Hs  C+ C0 c |ttjt|Dcgc] }||| c}Scc}wr)roperatoraddr)fnrers r. compute_sumz/simple_sparse_reduce_tests..compute_sums2 LLE*.wrappers(yyD222$$R(r-r types MethodTyper8rrgs ` r. join_or_runz MultiProcessTestCase.join_or_run. r    ..r- method_name methodNameNc|dk7r|}t||t||}t|||j |yNrunTestsuper__init__getattrsetattrrGr8rIrJr __class__s r.rPzMultiProcessTestCase.__init__'B  "$K % T; 'k4#3#3B#78r-ct|g|_g|_|j|_t jdj|_ i|_ y)NF)delete) rOsetUpskip_return_code_checks processesr@rr'NamedTemporaryFiler) file_name pid_to_piper8rTs r.rXzMultiProcessTestCase.setUp0sH  ')$** !44EBGGr-crt||jD]}|jg|_yr)rOtearDownrZ terminate)r8prTs r.r`zMultiProcessTestCase.tearDown9s3  A KKM  r-cF|jjddSridrr7s r._current_test_namez'MultiProcessTestCase._current_test_nameCswwys#B''r-c g|_tt|jD]}tj j \}}||jjdt|z||j|j|fdt|ddi}|jtjd||j ||j"|j <|jj%|y)Nzprocess fake_pgF)targetr)rcrdzStarted process %s with pid %s)rZrr)rer[multiprocessingPiperT_runr+rfr\rQstartloggerinfopidr]r)r8procr parent_conn child_connprocesss r._start_processesz%MultiProcessTestCase._start_processesGs#doo./ +D&+&;&;&@&@&B #K~~**#d)+D335t~~zRwtY> G MMO KK8$ L,7D  W[[ ) NN ! !' * +r-cxtjjdj}|j |y)Nspawn)r[rj get_contextProcessru)r8rqs r._spawn_processesz%MultiProcessTestCase._spawn_processesXs,$$009AA d#r-ceZdZdZy)MultiProcessTestCase.EventruN)r&r'r( GET_TRACEBACKr,r-r.Eventr|\s r-r~rctjd| tjj ||g}||vr|j rtjd|y|j }tjd|||tjjk(rtjd5}tj||j|jd|j!|j#tjd|ddd||vry#1swYxYw)Nz*Starting event listener thread for rank %sz:Pipe closed for process %s, stopping event listener threadzReceived event %s on process %szr+)moderzProcess %s sent traceback)rnrorj connectionwaitclosedrecvr4r~r}r'r[ faulthandlerdump_tracebackflushseeksendread) parent_pipe signal_piper ready_pipeseventtmp_files r._event_listenerz$MultiProcessTestCase._event_listener_s  @$G)4499; :TUKk)%%KKTVZ#((* =udK066DDD!44$?G8$33H= ( a(#((9 $?FGk)5 GGs :A,D55D> test_namer\c T||}||_||_|j||yrrr\run_testclsrrr\rrdr8s r.rlzMultiProcessTestCase._run~s'9~ " i-r-c tjjd\}}tjt j |||jfd}|jtjdk7r2tjdk7rtjjddtjd< t||||j;d|J|j=|j?y#t j"$r[}t$j'd |j|t)|tj*t,d j.Yd}~d}~wt0$r}t$j3d t5j6|jt j8|j;t5j6tj*t j8Yd}~;d}~wwxYw#||j;d|J|j=|j?wxYw) NF)duplexT)rircdaemonrdarwinrTORCH_SHOW_CPP_STACKTRACESz4Process %s skipping test %s for following reason: %srEz;Caught exception: %s exiting process %s with exit code: %s) r[rjrk threadingThreadr4rrrmr]r_C'_set_print_stack_traces_on_fatal_signalr`rarQunittestSkipTestrnror+r^r_r$ Exceptionr traceback format_excTEST_ERROR_EXIT_CODErr,close)r8rrsignal_recv_pipesignal_send_pipeevent_listener_threadsees r.rzMultiProcessTestCase.run_tests-2-B-B-G-Gu-G-U** ) 0 0'77/;!  ##% <<7 "s||x'? HH < > ? ? @ + %%d+(4 44 ! & & (    s>D HAE2-H2 H>BH H HH9I cg}t|jD]h\}}|j|j|j} |j t jj|j||fj|D]x\}} |jdrK|jrtjd|;|j!}tjd||ntjd|zy#t$r"}tjd||Yd}~d}~wwxYw#t$r!}tjd||Yd}~d}~wwxYw)NzBEncountered error while trying to get traceback for process %s: %srz5Pipe closed for process %s, cannot retrieve tracebackz)Process %s timed out with traceback: %sz6Could not retrieve traceback for timed out process: %s) enumeraterZexitcoder]rprr4r~r}rConnectionErrorrnrpollrror)r8pipesrrtpiperrrs r._get_timedout_process_tracebackz4MultiProcessTestCase._get_timedout_process_tracebacks>#DNN3 JAw''' 4II288FFGLL!T+   JD$ 99Q<{{ SUY! $ ILLEtYLLPRV  'LL\^_ab0#  XZ^`a s6D/ D, D''D,/ E8EEct|j}tj}d} t|jD]w\}}|j t jk(s$td|d|j dtjj}|D]}|jd}n|rntd|jDrnttj|z } | |kDrA|jtd|d|jD]}|jntjd #tj|z } ||j vr|j#| n|j%| |j&j)D]} | j+y#|j&j)D]} | j+wxYw) NFTProcess z terminated with exit code z", terminating remaining processes.c38K|]}|jduywr)r).0rbs r. z7MultiProcessTestCase._join_processes..sF!qzz-FszTiming out after z" seconds and killing subprocesses.g?)rretimerrZrr4rprintr[rjactive_childrenraallrsleeprY_check_no_test_errors_check_return_codesr]rr) r8rr start_timesubprocess_errorrrbracelapsed elapsed_timers r.rAz$MultiProcessTestCase._join_processessdggi(YY[  * '7 FQzz%9%N%NN&qc)DQZZLPrs+0*?*?*O*O*Q"1+BLLN++/( $Ft~~FF))+ 2W$88:+G94VW"^^& & 3=@ 99;3LT111**<8((6((//1   ((//1   s9G +D2G 1G>ct|jD]I\}}|jtd|d|d|j |j |jKy)zV Checks that we didn't have any errors thrown in the child processes. Nrz timed out after  seconds)rrZr RuntimeErrorassertNotEqualr)r8rrrbs r.rz*MultiProcessTestCase._check_no_test_errorsshdnn- GDAqzz!"qc!2<.I    9 91:: F  Gr-c 8|jstjdy|jd}t|jDcgc]&\}}|jt j k(r||f(}}}|r[d}|D]I\}}|j|jj}|d|dt j d|dz }Kt|t|jD]h\}}|jtd|d |d |j|j|jd |d |jd |jjtjD]q} |j| jk(str1tj!d|j#| j$yt'j(| j$|j|jdd|jd|jycc}}w)z Checks that the return codes of all spawned processes match, and skips tests if they returned a return code indicating a skipping condition. z @4r-r4subtest_configtest_fn test_kwargsct|j}|Dcgc]}|d }}|Dcgc]}|d }}tj|D]} t t || } |j di| 5tjj||i|| tjjdddtjycc}wcc}w#1swY*xYw)a\ Runs a test function given by ``test_fn`` as a subtest according to the configurations specified by ``subtest_config``. This amortizes the costly setup overhead (including process spawn and initializing the process group) over the subtests. Args: subtest_config (Dict[str, List[Any]]): A mapping from subtest keyword argument name to a list of its possible values. test_fn (Callable): A callable that runs the actual test. test_args: Positional arguments to pass to ``test_fn``. test_kwargs: Keyword arguments to pass to ``test_fn``. rruNr,) ritems itertoolsproductdictzipsubTestr[_dynamoresetrr) cls_instrr test_argsrsubtest_config_itemsitemsubtest_config_keyssubtest_config_valuesrsubtest_kwargss r. run_subtestsrUs*9=^=Q=Q=S8T:N%O$d1g%O%OBV-W$d1g-W-W##%:;c"5v>? X   / / " MM   ! Y @+ @ @ MM   ! " &P-W " "s C  C%:AC**C3 cttS tjgddjdk(atS#t$r daYtSwxYw)a If shell command `fi_info -p efa -t FI_EP_RDM` returns exit code 0 then we assume that the machine has Libfabric EFA interfaces and EFA software components installed, see https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/efa-start.html. )fi_infoz-pefaz-t FI_EP_RDMF)checkr)EFA_PROBE_RESULT subprocessrun returncodeFileNotFoundErrorr,r-r.has_efar{sZ#! NNFe T _ _cd d   !  !s&: A  A c"trddgSdS)a If the machine has Libfabric EFA interfaces and EFA software components installed it may cause 'RuntimeError: In operator() at tensorpipe/common/ibv.h:172 "": Operation not supported' if tensorpipe uses InfiniBand transport, so we exclude it from tensorpipe transports, see https://github.com/pytorch/pytorch/issues/73885 and https://github.com/pytorch/pytorch/issues/65022 shmuvN)rr,r-r. tp_transportsrs$IE4=/4/r-cdtt|Sdtfd}|S)z+ Wrapper to use with a test method )rrecttj}fdfd}g}tD]=}t j |||f}|j |j|?|S)Nc>tjjk(Srrdistributed_c10d_worldworldsr.world_is_validzaspawn_threads_and_init_comms.._run_test_method_with_multi_threads..world_is_validD11888 8r-c~tjd|| rtjyy#t$rR}tjj |t jftj|Yd}~sd}~wwxYw#rtjwwxYw)Nthreadedr{rrestore) rinit_process_group BaseExceptionMultiThreadedTestCaseexception_queueputr]exc_infor exception_handledestroy_process_group)rworld_pgrexcallbackr res r.workerzYspawn_threads_and_init_comms.._run_test_method_with_multi_threads..workers  # #"*E  1 "#..0$ ! 7%55994:PQ!22266 7 "#..0$s*A B ABBBBB<rirc)rr HashStorerrrrmr) rer global_storerthreadsrtrr s `` @@r.#_run_test_method_with_multi_threadszIspawn_threads_and_init_comms.._run_test_method_with_multi_threadssq$&~~'  9 1*% D  dE<5PQA GGI NN1   r-cXtjjjd fd}tj |tjjjdy#tjjjdwxYw)NTcgiSrr,)rcrfrdr8sr.z?spawn_threads_and_init_comms..wrapper..sdSWNiZ^NibhNir-F)r[r_distributed_c10d_set_thread_isolation_moder _join_threads)r8rcrdrr rfres``` r.rgz-spawn_threads_and_init_comms..wrappersq ""==dC I9*FijG ! / / > HH & & A A% HEHH & & A A% Hs %A>>+B))rspawn_threads_and_init_commsr )rfrrergr s` ` @r.r'r'sD  | ('j  < 4[II Nr-ceZdZdZej ZdZdZdde de ddffd Z d Z d Z dfd Z fd Zd ZedZdZedZedZedefdZede fdZddddZddddZxZS)ra5 Test runner that runs all tests with the in-proc process group using multiple threads with the threaded process group. Each test spawns world_size threads and run the test method in each thread. Difference from regular MultiProcess test runner: Must explicitly defines SetUp and call self._spawn_threads() to run the tests. Cannot use setUp / tearDown (must use perThreadSetup / perThreadShutdown) to set up / tear down each thread when running each test. No global state possible How bad of a limitation is this? rcVtfd}tj||S)Nc|j|jk(r|j|jyyr)rMAIN_THREAD_RANKr&rrBs r.rgz2MultiThreadedTestCase.join_or_run..wrappers.yyD111""4<<4r-rCrFs ` r.rGz!MultiThreadedTestCase.join_or_runrHr-rIrJrNc|dk7r|}t||t||}t|||j |yrLrNrSs r.rPzMultiThreadedTestCase.__init__rUr-cyrr,r7s r.perThreadSetUpz$MultiThreadedTestCase.perThreadSetUps r-cyrr,r7s r.perThreadTearDownz'MultiThreadedTestCase.perThreadTearDowns r-cxt||j|_g|_dt j d<y)z setUp only set up things in the main thread, if you want to configure things in the spawned threads, use perThreadSetUp rrN)rOrXr+rrr`rar^s r.rXzMultiThreadedTestCase.setUps1  ))  36 /0r-c0t|g|_y)z tearDown only set up things in the main thread, if you want to configure things in the spawned threads, use perThreadTearDown N)rOr`rr^s r.r`zMultiThreadedTestCase.tearDowns  r-ctjjjd|j}t t j|j_ fd}|s tdt|jD]e}tj|jj|||jf}|j!|j"j%|gy)zk class method to spawn threads and run test, use this method in the SetUp of your TestCase Tc>tjjk(Srrrsr.r z.world_is_validr r-z Invalid worldrN)r[rr$r%rfrrrrTrrrrerrrlrmrr)r8rr rrrs @r._spawn_threadsz$MultiThreadedTestCase._spawn_threadss ""==dC++ $&&*nn&6# 9/ /$//* #D  (;(;9dTXTcTcBdeA GGI LL   " #r-c ||}||_t|drWtj|_t j |j_t j|j_ |j|||y)N_tls) rhasattrrlocalr7r _precision precision_rel_tolrel_tolrun_test_with_threaded_pg)rrrrerdr8s r.rlzMultiThreadedTestCase._run$sb9~ 4 !)DI"*"5"5DII  ( 1 1DII  &&y$ Cr-ctjd|||jj|j  t ||tj|jy#t $rN}|jj|tjftj|Yd}~wd}~wwxYw#tj|jwxYw)zd Run the current test associated with `test_name` using the threaded process group. r r N)rrrTrr.rQrrrr]rr rrr0)r8rrrers r.r>z/MultiThreadedTestCase.run_test_with_threaded_pg4s TjHcHc   % $GD) $ &  & & (  " " $  3  $ $dCLLN%; <  . .r 2 2 3  & & (  " " $s*A55 C >ACCC  C&C5c t} t|D]f\}}|jtd||j s2t j j|ttd|ddffhtjg}|j jsF|j j}|j||j jsFttj j"j%d|j'|||y#ttj j"j%dwxYw)NrzRank failed to join in under rF)rrr,maxis_aliverrr TimeoutErrorr remptyrrrr[rr$r%r)rrrridxthread failed_ranksfailures r.r&z#MultiThreadedTestCase._join_threadsGs-! I(1  V C7O,??$)99== , ,&CG9H$U!"!%      # # %L))//1--113##G,))//1 # $ HH & & A A% H  gr: # $ HH & & A A% HsTdV#S(Y] #I- )2 y>A y) ) q="))+ >.$ TVXZ^ZfZf&// == > r-ctSrr;r7s r.rez MultiThreadedTestCase.world_sizer=r-cF|jjddSrrdr7s r.rfz(MultiThreadedTestCase._current_test_nameswwys#B''r-rrcJ|j|k(r|j|||yy)z The reason why we have this util function instead of self.assertEqual is all threads are sharing one CPU RNG so the assertion result is only reliable on rank 0 N)rrr8ryrrs r.assertEqualOnRankz'MultiThreadedTestCase.assertEqualOnRanks' 99    Q3 ' r-cH|j|k(r|j||yyr)rrrVs r.assertNotEqualOnRankz*MultiThreadedTestCase.assertNotEqualOnRanks# 99    1 % r-rrr)r&r'r(__doc__queueQueuerr+rGr+rPr.r0rXr`r5rrlr>r&rrr)rerfrXrZrrs@r.rrs "ekkmO/9C99UY9   7#* D D%&;;:.>.>`"C""(C((((&1&&r-rceZdZdeej ejfdeddffd Z dejdejfdZ xZ S)SaveForwardInputsModuleforward_inputscast_forward_inputsrNctt|tjdd|_||_||_y)Nd)rOrPnnLinearlr`rar8r`rarTs r.rPz SaveForwardInputsModule.__init__s2 3$,#6 r-rc||j|<|j|jr3|j|jjj S|Sr)r`rfratoweightdtyper8rs r.forwardzSaveForwardInputsModule.forwardsI$%D!vv43K3Kadd466==../SSQRSSr- r&r'r(r rdModuler[TensorrrPrmrrs@r.r_r_sT7RYY 457"7  7TT%,,Tr-r_ceZdZdeej ejfdeddffd Z dejdejfdZ xZ S)SaveForwardInputsModelr`rarNctt|t|||_t|||_||_yr)rOrPr_c1c2r`rgs r.rPzSaveForwardInputsModel.__init__s6 ).:MN).:MN,r-rc`||j|<|j|j|Sr)r`rurtrls r.rmzSaveForwardInputsModel.forwards)$%D!wwtwwqz""r-rnrs@r.rrrrsQ-RYY 45-"-  -##%,,#r-rrc#K|stjj|dtjd<dtjd<|rp|rVtj j jjj}tjd|||ntjd||tjjtjjjj! dtjjtjjjj!|rtj"yy#tjjtjjjj!|rtj"wwxYww) Nr MASTER_ADDR6789 MASTER_PORTfake)r{rerrr?rre)r[rP set_devicer`ratesting _internal distributedrh FakeStorerrrrutilscountersclearr)rreinit_pgrhrs r._dynamo_dist_per_rank_initrsH  d# +BJJ} &BJJ} MM++77??IIKE  # #%    # #F* M MM MM  &&()   $$**,   & & (    $$**,   & & ( s D G% E9A(G%9A)G""G%c@eZdZdZefdZefdZxZS)#DynamoDistributedSingleProcTestCasez Test harness for single-process dynamo distributed tests, initializes dist process group. Prefer this for simple tests, as it's easier to debug. c`t||jjt j t jdddd|_d|j|_ d|jvrdn |jg|_ tjd|jd y) Nr12355)rxrzrzcuda:rPr?rur|) rO setUpClass _exit_stack enter_contextrrr`rardevice device_idsrrrrTs r.rz.DynamoDistributedSingleProcTestCase.setUpClasss  %% JJ #.#*  SXXJ' !'3::!5CHH: SXX!Dr-cJtjt| yr)rrrO tearDownClassrs r.rz1DynamoDistributedSingleProcTestCase.tearDownClasss ""$ r-)r&r'r(r[rrrrrs@r.rrs2EE"  r-rc deZdZdZfdZfdZedefdZe dede de dd fd Z xZ S) "DynamoDistributedMultiProcTestCasea  Use this for tests that actually run on multiple GPUs. Decorate tests with @skip_if_lt_x_gpu(ngpu) Note: MultiProcTestCase spawns processes per test and is slow. Prefer MultiThreadedTestCase for most tests. Perhaps use this one sparingly for integration tests. cBt||jyr)rOrXrzr^s r.rXz(DynamoDistributedMultiProcTestCase.setUps   r-ct| tj|jy#t $rYywxYwr)rOr`r`remover\OSErrorr^s r.r`z+DynamoDistributedMultiProcTestCase.tearDowns5   IIdnn %   s 1 ==rc>tjjSr)r[rPrbr7s r.rez-DynamoDistributedMultiProcTestCase.world_sizeszz&&((r-rrr\Nc T||}||_||_|j||yrrrs r.rlz'DynamoDistributedMultiProcTestCase._run#s)9~ " i-r-) r&r'r(r[rXr`rr)rerr+rlrrs@r.rr s^ )C))....W[..r-rc eZdZUdZeed<dZeed<dZee ed<e e jde fdZ e d d Ze fd Ze fd Ze ddededee fd ZxZS)MultiProcContinousTestrnrerrN rdvz_filerctd)z ProcessGroup backend str. To be customized by sub test classes, e.g. "nccl". Here we raise error. z/Please implement backend_str in your test class)NotImplementedError)rs r. backend_strz"MultiProcContinousTest.backend_str5s""STTr-cy)z ProcessGroup init options. To be customized by sub test classes, e.g. ProcessGroupNCCLOpTest Here we return None. Nr,)rhigh_priority_streams r.optszMultiProcContinousTest.opts?sr-cPt|d|jcxkr|jks'nt d|jd|j|j r+t j|j |j}nd}|j}|j}td|t j||j|j||t jj|_td|jdy) z Class-scope test fixture. Run once for entire test class, before any test starts. Set up the process group. rzBRank must be set and in the range of 0 to world_size. World size: z Rank: NzTesting backend=)r{rerr pg_optionsRank z setup complete)rOrrrerrr FileStorerrrrr_get_default_grouppg)rrrr{rTs r.rz!MultiProcContinousTest.setUpClassHs CHH-s~~-"~~.gchhZA  ==NN3==#..AEExxz//# ! #$ ~~  &&99; chhZ/0r-ctjt| |jr t j |jtd|jdy#t$rY%wxYw)z Class-scope test fixture. Run once for entire test class, after all tests finish. Tear down the process group. rz teardown completeN) rrrOrrr`rrrrrs r.rz$MultiProcContinousTest.tearDownClassgsd ""$  ==  #--( chhZ123  sA** A65A6cB||_||_||_ty)ad This is an entry point for each rank to run the tests in `MultiProcContinousTest`. In this entry point, we set the class variables for the test class. Then we run all tests. Note: - This helper only works for a subclass of `MultiProcContinousTest`. Example: - See `test_c10d_ops_nccl.py`. N)rrerr)rrrers r.run_rankzMultiProcContinousTest.run_rankws&#!  r-)Fr)r&r'r(rer)r*rrr r+rabcabstractmethodrrrrrrrs@r.rr,sJD#N#Ix}#UCUU11< 4 4 $( C= r-rr)rur)TF)|rrrloggingrjr`r\rr]r'rrrrDr contextlibr dataclassesrdatetimerenumr functoolsrrr ior typingr r r rrrrr unittest.mockrr[torch._dynamo.test_casetorch.cuda.nccltorch.distributedrrtorch.nnrd$torch.testing._internal.common_utilsrrrrrrrrrr5torch.testing._internal.distributed.multi_threaded_pgrrr r basicConfigINFO getLoggerr&rnr#r_rJrjrqrvr~rrrrrrrrrrrrrrrrr)getenvrrrrrr+rrr(r*r.r1r<r4rrrrrr'rror_rrrr test_caserrrr,r-r.rsk     %!,,PPP      ',,'   8 $z 8 Cx$FG  Xb"BC  x45 8B =>8B >?8B >?8B >?8B >?8B >?8B >?8B >? HR> ? (267!"hr#LM#$x V%*8B DE+ 2 && &"" &H  R$+\4       a   0O)"))$GOPO,c2+.'(HICI22(S(c(s(XS3*26(-- .5 Xc] d " I8IX d3i(  F&0 3E5pZ&HZ&z Tbii T #RYY #)): %--*A*A*J*J @.)=.DbXbr-