U Mh7@stUddlZddlZddlZddlZddlZddlZddlZddlZddlZddl 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?ej@ejAd eBeCZDGd ddeZEeEddeEddeEddeEddeEddeEddeEddeEddeEdd eEd!d"eEd#d$eEd%d&eEd'd(eEd)d*eEd+d,eEd-d.eEd/d0d1ZFeGd2d3d3ZGd4d5ZHd6d7ZId8d9ZJd:d;ZKdd?ZMd@dAZNdBdCZOdDdEZPdFdGZQdHdIZRdJdKZSdLdMZTdNdOZUdPdQZVdRdSZWdTdUZXe4dVdWdXedYdZdXd[dXfd\d]ZYe8rd^ZZne[e\d_d`ZZdadbiZ]e7rdce]dd<ddedfZ^e[dgdhdiZ_edjdkZ`de[e[e[dldmdnZae[ebdodpdqZcdade e jeefdr<de ebddsdtduZgddgdvdwZhdxZiGdydzdze9Zjeebe"e#fe$e#d{d|d}Zkdalemdgd~dZnddZodeZeifddZpGddde9ZqGddde/jrZsGddde/jrZtedddZuGddde(jvjwj9ZxGdddejZyGddde9ZzdS)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@seZdZUeed<eed<dS)TestSkip exit_codemessageN)__name__ __module__ __qualname__int__annotations__strr*r*\/var/www/html/venv/lib/python3.8/site-packages/torch/testing/_internal/common_distributed.pyr!7s r!Hz5Skipped because distributed backend is not available.Iz Skipped due to small world size.WzSkipped due to odd world size.JzCUDA is not available.KzNeed at least 1 CUDA deviceMzNeed at least 2 CUDA devicesPzNeed at least 3 CUDA devicesQzNeed at least 4 CUDA devicesRzNeed at least 5 CUDA devicesSzNeed at least 6 CUDA devicesTzNeed at least 7 CUDA devicesUzNeed at least 8 CUDA devicesLz#c10d not compiled with NCCL supportNzTest skipped for ROCmOz'Test skipped because no GPU peer accessVzHTest skipped at subprocess level, look at subprocess log for skip reasonXz"Test skipped due to missing import)Zbackend_unavailablesmall_worldsize odd_worldsizeno_cudaz multi-gpu-1z multi-gpu-2z multi-gpu-3z multi-gpu-4z multi-gpu-5z multi-gpu-6z multi-gpu-7z multi-gpu-8nccl skipIfRocmZno_peer_accessgeneric importerrorc@seZdZiZdddhed<eed<ddhed<ddhed<iZdddhed <dddhed <dddhed <dddhed <eed <dS) DistTestCasesr@mpiZuccZallgather_coalescedrzsendrecv anysourcez cpu barrierZglooZgpucudaZddpZsubgrouppluginN)r$r%r&Zskip_collectivesetZbackend_featurer*r*r*r+rDUs   rDcstfdd}|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)``.csVtjsttdjttj d}tj |krLttd|j||S)Nr? WORLD_SIZE multi-gpu-) torchrF is_availablesysexit TEST_SKIPSr"r'osenviron device_count)argskwargs world_sizefuncr*r+wrapperks  zskip_if_no_gpu..wrapperrrWrXr*rVr+skip_if_no_gpugs r[cstfdd}|S)Ncs:tjddkr0ttjddkr0ttdj||S)NBACKENDrErIr=rPrQr'rMrNrOr"rSrTrVr*r+rXys z(skip_if_small_worldsize..wrapperrYrZr*rVr+skip_if_small_worldsizexsr`cstfdd}|S)Ncs>tjddkr4ttjdddkr4ttdj||S)Nr\rErIr]r>r^r_rVr*r+rXs$z&skip_if_odd_worldsize..wrapperrYrZr*rVr+skip_if_odd_worldsizesrbcsfdd}|S)Ncstfdd}|S)Ncs<dkr.tjkr.ttdjn ||SdSNr@rJ)rKrFrRrMrNrOr"r_)backendrWnr*r+rXszCrequire_n_gpus_for_nccl_backend..decorator..wrapperrYrZrdrerVr+ decoratorsz2require_n_gpus_for_nccl_backend..decoratorr*)rerdrgr*rfr+require_n_gpus_for_nccl_backends rhcCs dd}|S)Ncstfdd}|S)NcsFzddlm}m}||WStk r@ttdjYnXdS)Nr)AutoModelForMaskedLM BertConfigrC)Z transformersrirj ImportErrorrMrNrOr")rSrTrirjrVr*r+rXs  z?import_transformers_or_skip..decorator..wrapperrYrZr*rVr+rgs z.import_transformers_or_skip..decoratorr*)rgr*r*r+import_transformers_or_skipsrlcsfdd}|S)Ncstfdd}|S)Ncs<tjr"tjkr"||SttdjdS)NrJrKrFrLrRrMrNrOr"r_)rWxr*r+rXs z4skip_if_lt_x_gpu..decorator..wrapperrYrZrnrVr+rgsz#skip_if_lt_x_gpu..decoratorr*)rnrgr*ror+skip_if_lt_x_gpus rpcsfdd}|S)Ncstfdd}|S)NcsNdkr||Stjr4tjkr4||SttdjdSrcrmr_)rdrWrnr*r+rXs   z9nccl_skip_if_lt_x_gpu..decorator..wrapperrYrZrdrnrVr+rgsz(nccl_skip_if_lt_x_gpu..decoratorr*)rdrnrgr*rqr+nccl_skip_if_lt_x_gpus rrcCst|}d|kstd|ks td|ks,t|d}|ddkrF|n |dd}||ksptd|d|dS) N iterationZ has_errorerrorz Exception raised from rzDid not find expected z in ddp logging data error: )Z_get_ddp_logging_dataAssertionErrorfindsplit)Z model_DDPZ err_substrZddp_logging_dataZ logging_erractualr*r*r+verify_ddp_error_loggeds     rzcstfdd}|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 sztjd}tjd=Wntk r.d}YnXz,ztjd}Wntk rXd}YnXW5dtjd<Xz||}|WS|dk r|tjd<|dk r|tjd<XdS)NZTORCH_NCCL_ASYNC_ERROR_HANDLING1ZTORCH_NCCL_BLOCKING_WAIT)rPrQKeyError)rSrTZ cached_nccl_async_error_handlingZcached_nccl_blocking_waitretrVr*r+rXs0    z(with_nccl_blocking_wait..wrapperrYrZr*rVr+with_nccl_blocking_waits"r~csfdd}|S)zK Runs a test for each distributed debug level specified in levels. cstfdd}|S)NcsRtjdd}D]:}|tjd<t||}t|dk r|tjd<q|S)NZTORCH_DISTRIBUTED_DEBUG)rPrQgetc10dZset_debug_level_from_envbarrier)rSrTZ old_levelr r})rWlevelsr*r+rXs   z:with_dist_debug_levels..decorator..wrapperrYrZrrVr+rgsz)with_dist_debug_levels..decoratorr*)rrgr*rr+with_dist_debug_levels s rcCstt dS)Nz+c10d was not compiled with the Gloo backend)rrZis_gloo_availabler*r*r*r+ requires_gloo&srcCsDtstdSttjj|kd|dtjjd|SdS)N+c10d was not compiled with the NCCL backendz0Requires NCCL version greater than or equal to: z , found: z , reason: )ris_nccl_availablerrrKrFr@version)rmsgr*r*r+requires_nccl_version-srcCstt dS)Nr)rrrr*r*r*r+ requires_nccl9srcCstt dS)Nz*c10d was not compiled with the UCC backend)rrZis_ucc_availabler*r*r*r+ requires_ucc?srcCstt dS)Nz*c10d was not compiled with the MPI backend)rrZis_mpi_availabler*r*r*r+ requires_mpiEsrcsd_tfdd}|S)zSkips a test for ROCmTcs"ts||SttdjdS)NrA)rrMrNrOr"r_rVr*r+rXPs zskip_if_rocm..wrapper) skip_if_rocmrrZr*rVr+rLsrcCsttjdkdS)Nwin32z8This unit test case is not supported on Windows platform)rrMplatformr*r*r*r+ skip_if_win32Ysr localhostraT)minutesFc CsLt}|r2t|tdd}tjj|||||Stj||||||dSdS)zL Creates a TCP store. Retries if the chosen port is already in use. ra)Z milliseconds)wait_for_workers use_libuvN)rr'rrKclassesZ dist_c10dZTCPStorer) addrrU is_mastertimeoutrZ jit_classrportZtimeout_millisecondr*r*r+create_tcp_store`s$ riZ!DISTRIBUTED_TESTS_DEFAULT_TIMEOUTZ300Ztest_ddp_uneven_inputsiZtest_join_kwargscCs2tjdks|dkr tjjddStjj|dSdS)Nrz 127.0.0.1)hostnameZ interface)rMrrZProcessGroupGloo create_devicerr*r*r+rsrreturncCst|ddtSN.ru)TIMEOUT_OVERRIDErrxTIMEOUT_DEFAULT)Ztest_idr*r*r+ get_timeoutsrc csRtt}}tjtj}}z ||t_t_tjtjfVW5||t_t_XdSN)r rMstdoutstderr)Znew_outZnew_errZold_outZold_errr*r*r+captured_outputs r)rankrU num_inputsc sxdttttddd}tdddfd d t|dd t|d d t|d d t|ddt|d dt|d dfDS)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. rar)rrU sparse_dims dense_dimsc Sstt|dd|df}|gddt|D}t|dD](}t|td|df}||qBt|dgddt|D}t|||S)NracSsg|]}dqSr]r*.0_r*r*r+ sz@simple_sparse_reduce_tests..generate..cSsg|]}dqSrr*rr*r*r+rs) rKZreshapeZarangerangecatZzerosappendZonesZsparse_coo_tensor)rrUrrindicesshapervaluesr*r*r+generates "z,simple_sparse_reduce_tests..generate)rUcs ttjfddtDS)Ncsg|]}|qSr*r*)rrfnrUr*r+rszCsimple_sparse_reduce_tests..compute_sum..)roperatoraddrrr*rr+ compute_sumsz/simple_sparse_reduce_tests..compute_sumcsDg|]<fddtDfddtDfqS)cs"g|]}|qSr*r*ri)rrrrUr*r+rsz9simple_sparse_reduce_tests...csg|]}qSr*r*r)rrrrUr*r+rs)r)rrrrrUrr+rs z.simple_sparse_reduce_tests..)rr])r)rar)r'r)rrUrrr*rr+simple_sparse_reduce_testss       r)rUrdcsBtj}t|d||kr&||fddt|D}|S)zMultigpu tests are designed to simulate the multi nodes with multi GPUs on each node. Nccl backend requires equal #GPUs in each process. On a single node, all visible GPUs are evenly divided to subsets, each process only uses a subset. racs*i|]"}|t||dqSra)listrZnGPUs_per_processZvisible_devicesr*r+ sz(init_multigpu_helper..)rKrFrRr)rUrdZnGPUsZ rank_to_GPUr*rr+init_multigpu_helpers  rtmp_dir) init_methodrcCstatjtjd<ttjtjdttjtjdtjtjd}t||dk rn|tjd<nt tj|dtjd<dS)NZTEMP_DIRrZtest_dirZinit_dirZ INIT_METHODZshared_init_file) tempfileTemporaryDirectoryrnamerPrQmkdirpathjoinr)rZ init_dir_pathr*r*r+initialize_temp_directoriess   rcCstdk rtdSr)rcleanupr*r*r*r+cleanup_temp_dirsrcs:eZdZdZdZedddZeedddZ dd Z d.e e d d fd d Z d dfdd Z d dfdd Ze dddZd dddZd dddZGdddeZeedddZeee e d ddd Ze d d!d"d#Zd dd$d%Zd dd&d'Zd dd(d)Zd dd*d+Zeedd,d-ZZS)/MultiProcessTestCaseru rcCsdS)NFr*selfr*r*r+_should_stop_test_suitesz,MultiProcessTestCase._should_stop_test_suitecCstSrDEFAULT_WORLD_SIZErr*r*r+rUszMultiProcessTestCase.world_sizecs tfdd}t||S)Ncs"|j|jkr|ndSr)rMAIN_PROCESS_RANK_join_processesrrr*r+rXs  z1MultiProcessTestCase.join_or_run..wrapperrtypes MethodTyperrrXr*rr+ join_or_runsz MultiProcessTestCase.join_or_runrunTestN) method_name methodNamercs8|dkr |}t|t||}t||||dS)Nrsuper__init__getattrsetattrr)rrrr __class__r*r+r#s   zMultiProcessTestCase.__init__cs8tg|_g|_|j|_tjddj|_ i|_ dS)NF)delete) rsetUpskip_return_code_checks processesrrrNamedTemporaryFiler file_name pid_to_piperrr*r+r,s  zMultiProcessTestCase.setUpcs(t|jD] }|qg|_dSr)rtearDownr terminate)rprr*r+r5s   zMultiProcessTestCase.tearDowncCs|ddSridrxrr*r*r+_current_test_name?sz'MultiProcessTestCase._current_test_namecCsg|_tt|jD]j}tj\}}||jjdt ||| |j |fd}| t d||j||j|j<|j|qdS)Nzprocess )targetrrSzStarted process %s with pid %s)rrr'rUrKmultiprocessingPiper_runr)rrstartloggerinfopidrr)rprocrZ parent_connZ child_connprocessr*r*r+_start_processesCs  z%MultiProcessTestCase._start_processescCstjdj}||dS)Nspawn)rKrZ get_contextProcessr)rrr*r*r+_spawn_processesQsz%MultiProcessTestCase._spawn_processesc@seZdZdZdS)zMultiProcessTestCase.EventraN)r$r%r& GET_TRACEBACKr*r*r*r+EventUsr rc Cstd|tj||g}||kr|jr:td|dS|}td|||tjj krt j dd<}t |||d||td|W5QRX||kr dSq dS)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)rrr connectionwaitclosedrecvrr r rr faulthandlerZdump_tracebackflushseeksendread) parent_pipeZ signal_piperZ ready_pipeseventZtmp_filer*r*r+_event_listenerXs(    z$MultiProcessTestCase._event_listenerr test_namerrcCs$||}||_||_|||dSrrrrun_testclsrrrrrr*r*r+rwszMultiProcessTestCase._run)rrc CsLtjjdd\}}tjtj|||jfdd}|t j dkrTt j dkrTtj ddt jd<zzt||Wntjk r}z*td |j|t|t td jW5d}~XYnVtk r}z6td t|jtj |tt tj W5d}~XYnXW5|dk r(|d|dk s6t||XdS) NF)ZduplexT)rrSdaemonrdarwinr{TORCH_SHOW_CPP_STACKTRACESz4Process %s skipping test %s for following reason: %srBz;Caught exception: %s exiting process %s with exit code: %s)!rKrr threadingThreadrrrrrMr_CZ'_set_print_stack_traces_on_fatal_signalrPrQrrvrcloserunittestSkipTestrrr)rNrOr" Exceptionrt traceback format_excTEST_ERROR_EXIT_CODE)rrrZsignal_recv_pipeZsignal_send_pipeZevent_listener_threadseer*r*r+r~sD    "  zMultiProcessTestCase.run_testc Csg}t|jD]p\}}|jdkr|j|j}z |tjj| ||fWqt k r|}zt d||W5d}~XYqXq|D]\}}zH| dr|jrt d|Wq|}t d||n t d|Wqt k r}zt d||W5d}~XYqXqdS)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) enumeraterexitcoderrrrr r rConnectionErrorrrtpollrrr)rZpipesrrpiper/rr+r*r*r+_get_timedout_process_tracebacksL    z4MultiProcessTestCase._get_timedout_process_tracebackc Cs2t|}t}d}zt|jD]P\}}|jt j kr$t d|d|jdt j }|D] } | q^d}qvq$|r|qtdd|jDrqt|} | |kr|t d|d |jD] }|qqtd qt|} ||jkr|| n || W5|jD]}|qXdS) NFProcess z terminated with exit code z", terminating remaining processes.Tcss|]}|jdk VqdSr)r1)rrr*r*r+ sz7MultiProcessTestCase._join_processes..zTiming out after z" seconds and killing subprocesses.g?)rrtimerrr'r0rr1rr-printrKractive_childrenrallr5sleepr_check_no_test_errors_check_return_codes) rrr start_timeZsubprocess_errorr4rrr:acelapsed elapsed_timer*r*r+rsD            z$MultiProcessTestCase._join_processescCsHt|jD]8\}}|jdkr2td|d|d||j|jq dS)zV Checks that we didn't have any errors thrown in the child processes. Nr6z timed out after  seconds)r0rr1 RuntimeErrorassertNotEqualr-)rrBrrr*r*r+r=s  z*MultiProcessTestCase._check_no_test_errorsc CsH|jstddS|jd}ddt|jD}|r~d}|D]6\}}|j|j}|d|dtjd |d 7}q>t |t|jD]R\}}|j dkrt d|d |d |j |j |j d |d|j d|j dqt D]<} |j | jkrtrtd|| jdSt| jq|j |j dd|j d|jddS)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.r6z exited with error code z and exception:   terminated or timed out after rCzExpect process z+ exit code to match Process 0 exit code of z , but got )r6Skipping %s on sandcastle for the following reason: %sz Expected zero exit code but got z for pid: )rrwarningr0rrrrr-rDr1 assertEqualrOrr"rrrr#r(r)) rrBZ first_processZerrored_processesrtrr error_messagerskipr*r*r+r> sP      z(MultiProcessTestCase._check_return_codescCs |jdkS)Nrr rr*r*r+rIszMultiProcessTestCase.is_master)rr)r$r%r&rr-boolrpropertyr'rUrr)rrrrrr rr  staticmethodr classmethodrrr5rr=r>r __classcell__r*r*rr+rs0   +%0 @r)subtest_configtest_fn test_kwargsc Ost|}dd|D}dd|D}tj|D]T}tt||} |jf| (tj |||| tj W5QRXt q2dS)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``. cSsg|] }|dqS)rr*ritemr*r*r+rdsz run_subtests..cSsg|] }|dqSrr*rVr*r*r+resN) ritems itertoolsproductdictzipZsubTestrK_dynamoresetrr) Zcls_instrSrTZ test_argsrUZsubtest_config_itemsZsubtest_config_keysZsubtest_config_valuesrZsubtest_kwargsr*r*r+ run_subtestsNs  r_cCsLtdk r tSz"tjdddddgddjd kaWntk rFdaYnXtS) 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. NZfi_infoz-pZefaz-tZ FI_EP_RDMF)checkr)EFA_PROBE_RESULT subprocessrun returncodeFileNotFoundErrorr*r*r*r+has_efats rfcCstrddgSdS)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 ZshmZuvN)rfr*r*r*r+ tp_transportssrgcs:dkrtt|dSddtfdd}|S)z+ Wrapper to use with a test method N)rrUcsftt}fddfdd}g}tD]*}tj|||fd}|||q6|S)Ncs tjjkSrrdistributed_c10dZ_worldr*worldr*r+world_is_validszaspawn_threads_and_init_comms.._run_test_method_with_multi_threads..world_is_validc sxtjd||dzPz Wn@tk r^}z"tj|tft |W5d}~XYnXW5rrtXdS)NthreadedrdrrUstore) rinit_process_groupdestroy_process_group BaseExceptionMultiThreadedTestCaseexception_queueputrMexc_inforexception_handle)rZworld_pgroex)callbackrlrUr*r+workers  zYspawn_threads_and_init_comms.._run_test_method_with_multi_threads..workerrrS)rr HashStorerr$r%rr)rUry global_storerzthreadsrtr*)ryrkrlrUr+#_run_test_method_with_multi_threadss   zIspawn_threads_and_init_comms.._run_test_method_with_multi_threadsc sLtjjdz(fdd}t|W5tjjdXdS)NTFcsfSrr*r*)rSrWrTrr*r+z?spawn_threads_and_init_comms..wrapper..)rKr&_distributed_c10d_set_thread_isolation_moders _join_threads)rrSrTr~rrWrU)rSrTrr+rXs z-spawn_threads_and_init_comms..wrapper)rspawn_threads_and_init_commsr)rWrrUrXr*rr+rs rcseZdZdZeZdZddZd'e ddfdd Z d d Z d d Z ddfdd Z fddZddZeddZddZeddZeddZeedddZee ddd Zd(d!d"d#d$Zd)d!d"d%d&ZZS)*rsa5 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? rucs tfdd}t||S)Ncs&|j|jkr||jndSr)rMAIN_THREAD_RANKrr~rrr*r+rXs z2MultiThreadedTestCase.join_or_run..wrapperrrr*rr+rsz!MultiThreadedTestCase.join_or_runrN)rrcs.t|t||d}t||||dSrr)rrrTrr*r+rs  zMultiThreadedTestCase.__init__cCsdSrr*rr*r*r+perThreadSetUpsz$MultiThreadedTestCase.perThreadSetUpcCsdSrr*rr*r*r+perThreadTearDownsz'MultiThreadedTestCase.perThreadTearDownrcs&t|j|_g|_dtjd<dS)z setUp only set up things in the main thread, if you want to configure things in the spawned threads, use perThreadSetUp r{r#N)rrrrr~rPrQrrr*r+rs zMultiThreadedTestCase.setUpcstg|_dS)z tearDown only set up things in the main thread, if you want to configure things in the spawned threads, use perThreadTearDown N)rrr~rrr*r+rs zMultiThreadedTestCase.tearDowncstjjd|j}tt|j_ fdd}|s@t dt |j D]2}t j|jj|||j fd}||j|qJdS)zk class method to spawn threads and run test, use this method in the SetUp of your TestCase Tcs tjjkSrrhr*rjr*r+rlsz.world_is_validz Invalid worldr{N)rKr&rrrrrr|rr}rDrrUr$r%rrr~r)rrrlrrr*rjr+_spawn_threadss  z$MultiThreadedTestCase._spawn_threadscCsH||}||_t|dr6t|_tj|j_tj|j_ | |||dS)N_tls) rhasattrr$localrrZ _precision precisionZ_rel_tolZrel_tolrun_test_with_threaded_pg)r rrrUrr*r*r+rs    zMultiThreadedTestCase._runc Cstjd|||jjd|zVzt||Wn@tk rp}z"|j |t ft |W5d}~XYnXW5t|XdS)zd Run the current test associated with `test_name` using the threaded process group. rmrnN)rrprr}rrqrrrrrtrurMrvrrw)rrrrUrxr*r*r+r)s z/MultiThreadedTestCase.run_test_with_threaded_pgc Cst}z|t|D]B\}}|td|| rt j |t t d|ddffqtg}|j s~|j }||q^W5ttjjdX||||dS)NFrzRank failed to join in under rC)rrrKr&rrr0rmaxis_aliversrtru TimeoutErrorrr^emptyrrr>)r r~rridxthread failed_ranksZfailurer*r*r+r<s.    z#MultiThreadedTestCase._join_threadsc CsNd}d}|D]\}}|d}t|tjrPtd||t||dkrtdj}q t|trd|d|d } t | t | q t|t rd t j|} t d | ||d|d | d 7}q t|tr t|jtkr |dkr |j}q t|dkrt ||dkrJtD]:} || jkrtr:td || jdSt| jqdS)NrFruraz3Thread %s skipping test %s for following reason: %srrBzThread rHz seconds z'Caught exception: %s exiting thread %sz exited with exception: rGrI) isinstancer(r)rrr)rOr"rrtrDr*rr+format_exception SystemExittypecoder'lenrrr#) r rrr error_msgZ skip_coderrvexcrrMr*r*r+r>ZsV            z)MultiThreadedTestCase._check_return_codescCstSrrrr*r*r+rUsz MultiThreadedTestCase.world_sizecCs|ddSrrrr*r*r+rsz(MultiThreadedTestCase._current_test_namerr cCs|j|kr||||dS)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)rrKrrnyrrr*r*r+assertEqualOnRanks z'MultiThreadedTestCase.assertEqualOnRankcCs|j|kr|||dSr)rrErr*r*r+assertNotEqualOnRanks z*MultiThreadedTestCase.assertNotEqualOnRank)r)N)N)r$r%r&__doc__queueQueuertrrr)rrrrrrrQrrrr>rOr'rUrrrrRr*r*rr+rss.      0 rscsFeZdZeejejfeddfdd Z ejejdddZ Z S)SaveForwardInputsModuleNforward_inputscast_forward_inputsrcs(ttdd|_||_||_dS)Nd)rrnnZLinearlrrrrrrr*r+rs z SaveForwardInputsModule.__init__rnrcCs*||j|<||jr$||jjjn|Sr)rrrtoweightZdtyperrnr*r*r+forwards zSaveForwardInputsModule.forward r$r%r&r rModulerKZTensorrNrrrRr*r*rr+rs  rcsFeZdZeejejfeddfdd Z ejejdddZ Z S)SaveForwardInputsModelNrcs,tt|||_t|||_||_dSr)rrrc1c2rrrr*r+rs   zSaveForwardInputsModel.__init__rcCs||j|<|||Sr)rrrrr*r*r+rs zSaveForwardInputsModel.forwardrr*r*rr+rs  rccstj|dtjd<dtjd<|r4tjd||dtjtjj j z dVW5tjtjj j |r|t XdS)Nr MASTER_ADDRZ6789 MASTER_PORTr@rrU) rKrFZ set_devicerPrQrrpr]r^utilsZcountersclearrq)rrUZinit_pgr*r*r+_dynamo_dist_per_rank_inits      rcs4eZdZdZefddZefddZZS)#DynamoDistributedSingleProcTestCasez Test harness for single-process dynamo distributed tests, initializes dist process group. Prefer this for simple tests, as it's easier to debug. csht|jttjdddd|_d|j|_ d|j krHdn|jg|_ t j d|jdd dS) NrZ12355)rrrzcuda:rFr@rar) r setUpClassZ _exit_stack enter_contextrr[rPrQrZdeviceZ device_idsrrpr rr*r+rs  z.DynamoDistributedSingleProcTestCase.setUpClasscsttdSr)rrqr tearDownClassrrr*r+rsz1DynamoDistributedSingleProcTestCase.tearDownClass)r$r%r&rrQrrrRr*r*rr+rs rcsVeZdZdZfddZfddZeedddZe ee e d d d d Z Z 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. cst|dSr)rrr rrr*r+rs z(DynamoDistributedMultiProcTestCase.setUpcs4tzt|jWntk r.YnXdSr)rrrPremoverOSErrorrrr*r+rs  z+DynamoDistributedMultiProcTestCase.tearDownrcCs tjSr)rKrFrRrr*r*r+rU sz-DynamoDistributedMultiProcTestCase.world_sizeNrcCs$||}||_||_|||dSrrrr*r*r+rsz'DynamoDistributedMultiProcTestCase._run) r$r%r&rrrrOr'rUrQr)rrRr*r*rr+rs  rcseZdZUdZeed<dZeed<dZee ed<e e j e ddd Z e dd d Ze fd dZe fddZe deeee dddZZS)MultiProcContinousTestr]rUrurN rdvz_filercCs tddS)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 classN)NotImplementedErrorrr*r*r+ backend_str sz"MultiProcContinousTest.backend_strFcCsdS)z ProcessGroup init options. To be customized by sub test classes, e.g. ProcessGroupNCCLOpTest Here we return None. Nr*)r Zhigh_priority_streamr*r*r+opts*szMultiProcContinousTest.optscstd|jkr"|jkszt|jWntk r<YnXtd|j ddS)z Class-scope test fixture. Run once for entire test class, after all tests finish. Tear down the process group. rz teardown completeN) rrqrrrrPrrr9rrrr*r+rRs z$MultiProcContinousTest.tearDownClass)rrUrcCs||_||_||_tdS)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)rrUrr)r rrUrr*r*r+run_rankbszMultiProcContinousTest.run_rank)F)N)r$r%r&rUr'r(rrr r)rQabcabstractmethodrrrrrrRr*r*rr+rs&    r)N)ra)N)T){rrrYloggingrrPrrbrMrr$r8r+rr( contextlibr dataclassesrdatetimerenumr functoolsrrrior typingr r r r rrrrZ unittest.mockrrKZtorch._dynamo.test_caseZtorch.cuda.ncclZtorch.distributed distributedrZtorch.nnrZ$torch.testing._internal.common_utilsrrrrrrrrrrZ5torch.testing._internal.distributed.multi_threaded_pgrrrr basicConfigINFO getLoggerr$rr!rOrDr[r`rbrhrlrprrrzr~rrrrrrrrrrr'getenvrrrrrr)rrrr(rrrrr_rarNrfrgrrsrrrrr]Z test_caserrrr*r*r*r+s     (   0    .     ,M #  8Z  "