o wZh@s(UddlZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl Z ddl Z ddlZddlmZddlmZddlmZddlmZddlmZmZmZddlmZddlmZmZm Z m!Z!m"Z"dd l#m$Z$dd l%m&Z&ddl'Z'ddl(Z'ddl)Z'ddl*m+Z,dd l-m.Z.dd l/m0Z0ddl1m2Z2dd l3m4Z4m5Z5m6Z6m7Z7m8Z8m9Z9m:Z:m;Z;mZ>m?Z?ddl@mAZAmBZBmCZCddlDZDejEejFdeGeHZIGdddeZJideJdddeJdddeJdddeJdddeJdd d!eJd"d#d$eJd%d&d'eJd(d)d*eJd+d,d-eJd.d/d0eJd1d2d3eJd4d5d6eJd7d8d9eJd:d;dd?eJd@dAdBeJdCdDdEeJdFdGiZKeGdHdIdIZLdJdKZMdLdMZNdNdOZOdPdQZPdRdSZQdTdUZRdVdWZSdXdYZTdZd[ZUd\d]ZVd^d_ZWd`daZXdbdcZYdddeZZdfdgZ[dhdiZ\djdkZ]dldmZ^dndoZ_dpe'j`dqeadreadsebfdtduZce7dvdwdxedydzdxd{dxfd|d}Zde;rd~ZeneaefddZeddiZge:rdegd<dddZhdseafddZieddZjddeadeadeafddZkdeadelfddZmdanee joepd<ddeeldsdfddZqdddZrdZsGddde<ZtGdddetZudevelewe!fde"de!fddZxdaydsebfddZzddZ{deeesfddZ|Gddde<Z}Gddde2j~ZGddde2j~ZedddZGddde'jjj<ZGdddetZGddde<ZdS)N)contextmanager) dataclass) timedelta)Enum)partialreducewraps)StringIO) NamedTupleOptionalUnionAnyCallable)patch) trace_log) DeviceType)_SymmetricMemory) FILE_SCHEMAfind_free_port IS_SANDCASTLEretry_on_connect_failuresskip_but_pass_in_sandcastleskip_but_pass_in_sandcastle_ifTEST_WITH_ROCMTEST_WITH_TSANTestCase run_testsTEST_HPUTEST_XPU)_install_threaded_pg_uninstall_threaded_pgProcessLocalGroup)levelc@seZdZUeed<eed<dS)TestSkip exit_codemessageN)__name__ __module__ __qualname__int__annotations__strr,r,Y/var/www/auris/lib/python3.10/site-packages/torch/testing/_internal/common_distributed.pyr#<s  r#Zbackend_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 ROCmZno_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 importZno_acceleratorYzaccelerator is not available.c@seZdZiZhded<eed<ddhed<ddhed<iZhded <hded <hded <hded <eed <erCdhed<erLdhed<dSdS) DistTestCases>uccr=mpiZallgather_coalescedrr=rHzsendrecv anysourcez cpu barrier>rHr=glooZgpucudaZddpZsubgrouppluginhcclhpuxcclxpuN)r&r'r(Zskip_collectivesetZbackend_featurerrr,r,r,r-rG[s"          rGctfdd}|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)``.cstjs ttdjttj d}tj |kr&ttd|jt r9tj j |kr9ttd|jt rLtjj |krLttd|j|i|S)Nr3 WORLD_SIZE multi-gpu-z multi-xpu-)torchrK is_availablesysexit TEST_SKIPSr$r)osenviron device_countrrNrrP)argskwargs world_sizefuncr,r-wrapperus zskip_if_no_gpu..wrapperrrarbr,r`r-skip_if_no_gpuqs recrR)Ncs>tjddkrttjddkrttdj|i|S)NBACKENDrIrSr/rZr[r)rWrXrYr$r]r^r`r,r-rbs z(skip_if_small_worldsize..wrapperrcrdr,r`r-skip_if_small_worldsizerjcrR)NcsBtjddkrttjdddkrttdj|i|S)NrfrIrSr1rhrir`r,r-rbs$z&skip_if_odd_worldsize..wrapperrcrdr,r`r-skip_if_odd_worldsizerkrncfdd}|S)Nctfdd}|S)Ncs>dkrtjkrttdjdS|i|SNr=rT)rUrKr\rWrXrYr$ri)backendranr,r-rbszCrequire_n_gpus_for_nccl_backend..decorator..wrapperrcrdrrrsr`r- decoratorsz2require_n_gpus_for_nccl_backend..decoratorr,)rsrrrur,rtr-require_n_gpus_for_nccl_backends rvcCs dd}|S)NcrR)NcsFzddlm}m}|i|WSty"ttdjYdSw)Nr)AutoModelForMaskedLM BertConfigrD)Z transformersrwrx ImportErrorrWrXrYr$)r]r^rwrxr`r,r-rbs  z?import_transformers_or_skip..decorator..wrapperrcrdr,r`r-rus z.import_transformers_or_skip..decoratorr,)rur,r,r-import_transformers_or_skipsrzcCstjo tj|kSN)rUrKrVr\xr,r,r-at_least_x_gpusr~cfdd}|S)Nctfdd}|S)Ncstjrtjkr|i|Str#tjkr#|i|Str3tjkr3|i|St t dj dS)NrT) rUrKrVr\rrNrrPrWrXrYr$ri)rar}r,r-rbsz4skip_if_lt_x_gpu..decorator..wrapperrcrdr|r`r-rus z#skip_if_lt_x_gpu..decoratorr,)r}rur,r|r-skip_if_lt_x_gpus  rcro)Ncrp)NcsVdkr |i|Stjrtjkr|i|SttdjdSrq)rUrKrVr\rWrXrYr$ri)rrrar}r,r-rbs z9nccl_skip_if_lt_x_gpu..decorator..wrapperrcrdrrr}r`r-rusz(nccl_skip_if_lt_x_gpu..decoratorr,)rrr}rur,rr-nccl_skip_if_lt_x_gpus rcCst|}d|vs Jd|vsJd|vsJ|d}|ddkr#|n|dd}||vs8Jd|d|dS) N iterationZ has_errorerrorz Exception raised from rzDid not find expected z in ddp logging data error: )Z_get_ddp_logging_datafindsplit)Z model_DDPZ err_substrZddp_logging_dataZ logging_erractualr,r,r-verify_ddp_error_loggeds     rcrR)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 sz tjd}tjd=Wn tyd}Ynwzztjd}Wn ty*d}YnwWdtjd<ndtjd<wz|i|}|W|durK|tjd<|durU|tjd<SS|dur_|tjd<|durh|tjd<w)NZTORCH_NCCL_ASYNC_ERROR_HANDLINGZTORCH_NCCL_BLOCKING_WAIT1)rZr[KeyError)r]r^Z cached_nccl_async_error_handlingZcached_nccl_blocking_waitretr`r,r-rbsF     z(with_nccl_blocking_wait..wrapperrcrdr,r`r-with_nccl_blocking_waits"rcr)zK Runs a test for each distributed debug level specified in levels. cr)NcsVtjdd}D]}|tjd<t|i|}t|dur(|tjd<q |S)NZTORCH_DISTRIBUTED_DEBUG)rZr[getc10dZset_debug_level_from_envbarrier)r]r^Z old_levelr"r)ralevelsr,r-rb.s  z:with_dist_debug_levels..decorator..wrapperrcrdrr`r-ru-sz)with_dist_debug_levels..decoratorr,)rrur,rr-with_dist_debug_levels(s rcCtt dS)Nz+c10d was not compiled with the Gloo backend)rrZis_gloo_availabler,r,r,r- requires_glooBrcCs@tstdSttjj|kd|dtjjd|S)N+c10d was not compiled with the NCCL backendz0Requires NCCL version greater than or equal to: z , found: z , reason: )ris_nccl_availablerrrUrKr=version)rmsgr,r,r-requires_nccl_versionIsrcCr)Nr)rrrr,r,r,r- requires_ncclUrrcCr)Nz*c10d was not compiled with the UCC backend)rrZis_ucc_availabler,r,r,r- requires_ucc[rrcCr)Nz*c10d was not compiled with the MPI backend)rrZis_mpi_availabler,r,r,r- requires_mpiarrcCs$tjo ttjd}t| dS)Nrz"multicast support is not available)rUrKrVrhas_multicast_supportrCUDAr)rr,r,r-requires_multicast_supporths  rcsd_tfdd}|S)zSkips a test for ROCmTcs&ts |i|SttdjdS)Nr?)rrWrXrYr$rir`r,r-rbwsz*skip_if_rocm_multiprocess..wrapper)skip_if_rocm_multiprocessrrdr,r`r-rssrcCsttjdkdS)Nwin32z8This unit test case is not supported on Windows platform)rrWplatformr,r,r,r- skip_if_win32rrdevicemajorminorreturncCs6|jdkr tdtjjdurdStj|||fkS)z Returns True if the device's compute capability is (major, minor) or higher. Error out if the device is not a CUDA device. Returns False if device is a RoCM device. rKz3sm_is_or_later() is only supported for CUDA devicesNF)type ValueErrorrUrZhiprKZget_device_capability)rrrr,r,r-sm_is_or_higher_thans  r localhostrmT)minutesFc CsHt}|rt|tdd}tjj|||||Stj||||||dS)zL Creates a TCP store. Retries if the chosen port is already in use. rm)Z milliseconds)wait_for_workers use_libuv)rr)rrUclassesZ dist_c10dZTCPStorer) addrr_ is_mastertimeoutrZ jit_classrportZtimeout_millisecondr,r,r-create_tcp_stores   riZ!DISTRIBUTED_TESTS_DEFAULT_TIMEOUTZ300Ztest_ddp_uneven_inputsiZtest_join_kwargscCs.tjdks |durtjjddStjj|dS)Nrz 127.0.0.1)hostnameZ interface)rWrrZProcessGroupGloo create_devicerr,r,r-rsrcCst|ddtSN.r)TIMEOUT_OVERRIDErrTIMEOUT_DEFAULT)Ztest_idr,r,r- get_timeoutsrc cs`tt}}tjtj}}z||t_t_tjtjfVW||t_t_dS||t_t_wr{)r rWstdoutstderr)Znew_outZnew_errZold_outZold_errr,r,r-captured_outputs"rrankr_ num_inputsc s~ddtdtdtdtfdd}dtfd d fd d t|dd t|dd t|dd t|ddt|ddt|ddfDS)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. rmrrr_ sparse_dims dense_dimsc Sstt|dd|df}|gddt|D}t|dD]}t|td|df}||q!t|dgddt|D}t|||S)NrmcSg|]}dqSrlr,.0_r,r,r- z@simple_sparse_reduce_tests..generate..cSrrr,rr,r,r-rr) rUZreshapeZarangerangecatZzerosappendZonesZsparse_coo_tensor)rr_rrindicesshapervaluesr,r,r-generates "z,simple_sparse_reduce_tests..generatecs ttjfddtDS)Ncsg|]}|qSr,r,)rrfnr_r,r-rzCsimple_sparse_reduce_tests..compute_sum..)roperatoraddrrr,rr- compute_sumsz/simple_sparse_reduce_tests..compute_sumcsDg|]fddtDfddtDfqS)cs"g|] }|qSr,r,ri)rrrr_r,r-rsz9simple_sparse_reduce_tests...csg|] }qSr,r,r)rrrr_r,r-rs)r)rrrrr_rr-rsz.simple_sparse_reduce_tests..)rrl)rN)rmr)r)r)rr_rrr,rr-simple_sparse_reduce_testss      rrrcs^tj}tr tj}trtj}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. rmcs*i|]}|t||dqSrm)listrZnGPUs_per_processZvisible_devicesr,r- sz(init_multigpu_helper..)rUrKr\rrNrrPr)r_rrZnGPUsZ rank_to_GPUr,rr-init_multigpu_helpers    rtmp_dir init_methodcCstatjtjd<ttjtjdttjtjdtjtjd}t||dur8|tjd<dSt tj|dtjd<dS)NZTEMP_DIRrZtest_dirZinit_dirZ INIT_METHODZshared_init_file) tempfileTemporaryDirectoryrnamerZr[mkdirpathjoinr)rZ init_dir_pathr,r,r-initialize_temp_directoriess  rcCstdur tdSdSr{)rcleanupr,r,r,r-cleanup_temp_dir,s rc s8eZdZdZdZdefddZedefddZede fdd Z d d Z d1d e de ddffdd Z d2fdd Zd2fdd Zde fddZd2ddZd2ddZGdddeZede fdd Zede d!e d"e ddfd#d$Zd!e ddfd%d&Zd2d'd(Zd2d)d*Zd2d+d,Zd2d-d.Zedefd/d0ZZS)3MultiProcessTestCaser rcCdS)NFr,selfr,r,r-_should_stop_test_suiteIz,MultiProcessTestCase._should_stop_test_suitecCr)NTr,rr,r,r-destroy_pg_upon_exitQz)MultiProcessTestCase.destroy_pg_upon_exitcCtSr{DEFAULT_WORLD_SIZErr,r,r-r_UrzMultiProcessTestCase.world_sizec tfdd}t||S)Ncs$|j|jkr |dSdSr{)rMAIN_PROCESS_RANK_join_processesrrr,r-rbZs  z1MultiProcessTestCase.join_or_run..wrapperrtypes MethodTyperrrbr,rr- join_or_runY z MultiProcessTestCase.join_or_runrunTest method_name methodNameNc |dkr|}t|zt||}t||||WdSty@}z|dkr5td|jd||WYd}~dSd}~wwNr zno such test method in z: super__init__getattrsetattrrAttributeErrorr __class__rr r rerr,r-rg  zMultiProcessTestCase.__init__cs8tg|_g|_|j|_tjddj|_ i|_ dS)NF)delete) rsetUpskip_return_code_checks processesrrrNamedTemporaryFiler file_name pid_to_piperrr,r-rvs  zMultiProcessTestCase.setUpcs(t|jD]}|qg|_dSr{)rtearDownr terminate)rprr,r-r s    zMultiProcessTestCase.tearDowncC|ddSridrrr,r,r-_current_test_namesz'MultiProcessTestCase._current_test_namec Csg|_tt|jD]<}tj\}}||jjdt ||| |j |fdt |ddid}| td||j||j|j<|j|q dS)Nzprocess fake_pgF)targetrr]r^zStarted process %s with pid %s)rrr)r_rUmultiprocessingPiper_runr+r&rrstartloggerinfopidrr)rprocrZ parent_connZ child_connprocessr,r,r-_start_processess   z%MultiProcessTestCase._start_processescCstjdj}||dS)NZspawn)rUr)Z get_contextProcessr2)rr0r,r,r-_spawn_processessz%MultiProcessTestCase._spawn_processesc@seZdZdZdS)zMultiProcessTestCase.EventrmN)r&r'r( GET_TRACEBACKr,r,r,r-Eventsr6rcCstd| tj||g}||vr`|jrtd|dS|}td|||tjj kr`t j dd#}t |||d||td|Wdn1s[wY||vrfdSq) Nz*Starting event listener thread for rank %sTz:Pipe closed for process %s, stopping event listener threadzReceived event %s on process %szr+)moderzProcess %s sent traceback)r-r.r) connectionwaitclosedrecvrr6r5rr faulthandlerZdump_tracebackflushseeksendread) parent_pipeZ signal_piperZ ready_pipeseventZtmp_filer,r,r-_event_listeners,     z$MultiProcessTestCase._event_listener test_namercKs$||}||_||_|||dSr{)rrrun_testclsrrDrrAr^rr,r,r-r+szMultiProcessTestCase._runc Cstjjdd\}}tjtj|||jfdd}|t j dkr*t j dkr*tj ddt jd<zizt||WnGtjy^}ztd |j|t|t td jWYd}~n&d}~wtytd t|jtj|tt tjYnwW|dur|d|dusJ||n|dur|d|dusJ||w|j rzt!"WdSt#t$fyYdSwdS) NF)ZduplexT)r(r]daemonrdarwinrTORCH_SHOW_CPP_STACKTRACESz4Process %s skipping test %s for following reason: %srBz;Caught exception: %s exiting process %s with exit code: %s)%rUr)r* threadingThreadrrCrr,rWr_CZ'_set_print_stack_traces_on_fatal_signalrZr[runittestSkipTestr-r.r+rXrYr$ Exceptionr traceback format_excTEST_ERROR_EXIT_CODEr?rcloserrdestroy_process_groupAssertionErrorr)rrDrAZsignal_recv_pipeZsignal_send_pipeZevent_listener_threadser,r,r-rEsX          zMultiProcessTestCase.run_testc Csg}t|jD]9\}}|jdur@|j|j}z|tjj| ||fWqt y?}z t d||WYd}~qd}~wwq|D]A\}}z$| drd|jrXt d|WqC|}t d||nt d|WqCt y}z t d||WYd}~qCd}~wwdS)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) enumeraterexitcoderr/r?rr6r5rConnectionErrorr-rpollr:r.r;)rZpipesrr1piperrrQr,r,r-_get_timedout_process_tracebacksJ    z4MultiProcessTestCase._get_timedout_process_tracebackc CsFt|}t}d}z t|jD](\}}|jtjkr;td|d|jdt j }|D]}| q0d}nq|r?n2t dd|jDrJn't|} | |krk|td|d |jD]}| qcntd qt|} ||jvr|| n|| W|jD]} | qdS|jD]} | qw) NFTProcess z terminated with exit code z", terminating remaining processes.css|]}|jduVqdSr{)rY)rr"r,r,r- 9sz7MultiProcessTestCase._join_processes..zTiming out after z" seconds and killing subprocesses.g?)rr%timerXrrYrrSprintrUr)active_childrenr!allr]sleepr_check_no_test_errors_check_return_codesrrrT) rrr start_timeZsubprocess_errorrr"rbacelapsed elapsed_timer\r,r,r-r#sR               z$MultiProcessTestCase._join_processescCsHt|jD]\}}|jdurtd|d|d||j|jqdS)zV Checks that we didn't have any errors thrown in the child processes. Nr^z timed out after  seconds)rXrrY RuntimeErrorassertNotEqualrS)rrjrr"r,r,r-reSs z*MultiProcessTestCase._check_no_test_errorsc CsF|js tddS|jd}ddt|jD}|r?d}|D]\}}|j|j}|d|dtjd |d 7}qt |t|jD])\}}|j durXt d|d |d |j |j |j d |d|j d|j dqDt D]} |j | jkrtrtd|| jdSt| jqr|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.r^z exited with error code z and exception:   terminated or timed out after rkzExpect 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: )rr-warningrXrr/r;rrSrlrY assertEqualrYrr$rr.r%r%rNrO) rrjZ first_processZerrored_processesrrr1 error_messager"skipr,r,r-rf^sR          z(MultiProcessTestCase._check_return_codescCs |jdkS)Nrrrr,r,r-r zMultiProcessTestCase.is_masterr r rN) r&r'r(rrSboolrpropertyrr)r_rr+rrr r&r2r4rr6 staticmethodrC classmethodr+rEr]rrerfr __classcell__r,r,rr-r@s6    3 % 0 @rcsBeZdZfddZddZdefddZdd Zd d ZZ S) DistributedTestBasect|dSr{rrr4rrr,r-r  zDistributedTestBase.setUpcCs(z t|jWdStyYdSwr{)rZremoverOSErrorrr,r,r-r s  zDistributedTestBase.tearDownrcCs(d|vrdSd|vr dSd|vrdSdS)NrKr=rNrMrPrOrJr,rrr,r,r-rrszDistributedTestBase.backendcCsrt|}tj|j|}tjj|||j|j |dd||vs,d||vr3tj |j tjj S)Nrrr_rstorer=rO)rUget_device_moduler\ distributed FileStorerinit_process_grouprrr_r acceleratorset_device_indexdistributed_c10d_get_default_group)rrnum_visible_devicesrr,r,r- create_pgs zDistributedTestBase.create_pgcs&t|fddt|jDS)Ncsi|]}||gqSr,r,rrr,r-rsz6DistributedTestBase.rank_to_device..)rUrr\rr_rr,rr-rank_to_devicesz"DistributedTestBase.rank_to_device) r&r'r(rr r+rrrrr~r,r,rr-rs    rsubtest_configtest_fn test_kwargsc Ost|}dd|D}dd|D}tj|D]8}tt||} |jdi| tj ||i|| tj Wdn1sHwYt qdS)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``. cSg|]}|dqS)rr,ritemr,r,r-rz run_subtests..cSrrr,rr,r,r-rrNr,) ritems itertoolsproductdictzipZsubTestrU_dynamoresetrr) Zcls_instrrZ test_argsrZsubtest_config_itemsZsubtest_config_keysZsubtest_config_valuesrZsubtest_kwargsr,r,r- run_subtestss    rcCsDtdurtSztjgdddjdkaWtSty!daYtSw)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. N)Zfi_infoz-pZefaz-tZ FI_EP_RDMF)checkr)EFA_PROBE_RESULT subprocessrun returncodeFileNotFoundErrorr,r,r,r-has_efas rcCstrddgSdS)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)rr,r,r,r- tp_transportssrcs:dur tt|dSddtfdd}|S)z+ Wrapper to use with a test method N)rr_csftt}fddfdd}g}tD]}tj|||fd}|||q|S)Nc tjjkSr{rrZ_worldr,worldr,r-world_is_valid! zaspawn_threads_and_init_comms.._run_test_method_with_multi_threads..world_is_validc stjd||dz3zWn!ty0}ztj|tft |WYd}~nd}~wwWr;t dSdSrEt ww)Nthreadedrrrr_r) rr BaseExceptionMultiThreadedTestCaseexception_queueputrWexc_infor!exception_handlerU)rZworld_pgrex)callbackrr_r,r-worker$s    zYspawn_threads_and_init_comms.._run_test_method_with_multi_threads..workerr(r])rr HashStorerrKrLr,r)r_r global_storerthreadsrtr,)rrrr_r-#_run_test_method_with_multi_threadss   zIspawn_threads_and_init_comms.._run_test_method_with_multi_threadsc sXtjjdzfdd}t|WtjjddStjjdw)NTcsgRiSr{r,r,)r]rar^rr,r-@rz?spawn_threads_and_init_comms..wrapper..F)rUrM_distributed_c10d_set_thread_isolation_moder _join_threads)rr]r^rrrar_)r]r^rr-rb;s "z-spawn_threads_and_init_comms..wrapper)rspawn_threads_and_init_commsr)rarr_rbr,rr-rs rcseZdZdZeZdZddZd(de de dd ffd d Z d d Z ddZ d)fdd Z fddZddZeddZddZeddZeddZedefddZede fd d!Zd*d"d#d$d%Zd*d"d#d&d'ZZS)+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? rcr)Ncs(|j|jkr||jdSdSr{)rMAIN_THREAD_RANKrrrrr,r-rb\s  z2MultiThreadedTestCase.join_or_run..wrapperrrr,rr-r[rz!MultiThreadedTestCase.join_or_runr r r rNc r r rrrr,r-rerzMultiThreadedTestCase.__init__cCdSr{r,rr,r,r-perThreadSetUptrz$MultiThreadedTestCase.perThreadSetUpcCrr{r,rr,r,r-perThreadTearDownxrz'MultiThreadedTestCase.perThreadTearDowncs&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 rrJN)rrrrrrZr[rrr,r-r{s 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)rr rrrr,r-r s  zMultiThreadedTestCase.tearDowncstjjd|j}tt|j_ fdd}|s t dt |j D]}t j|jj|||j fd}||j|q%dS)zk class method to spawn threads and run test, use this method in the SetUp of your TestCase Tcrr{rr,rr,r-rrz.world_is_validz Invalid worldrN)rUrMrrr&rrrrrrlrr_rKrLr+r,rr)rrDrrrr,rr-_spawn_threadss  z$MultiThreadedTestCase._spawn_threadscKsH||}||_t|drt|_tj|j_tj|j_ | |||dS)N_tls) rhasattrrKlocalrrZ _precision precisionZ_rel_tolZrel_tolrun_test_with_threaded_pg)rGrDrr_r^rr,r,r-r+s    zMultiThreadedTestCase._runc Cstjd|||jjd|z@zt||Wn!ty9}z|j|t ft |WYd}~nd}~wwWt |dSWt |dSt |w)zd Run the current test associated with `test_name` using the threaded process group. rrN)rrrrrrrrrrWrr!rrUr)rrDrr_rr,r,r-rs&    z/MultiThreadedTestCase.run_test_with_threaded_pgc Cst}zLt|D]!\}}|td||r(tj|ttd|ddffqt g}|j sC|j }| ||j r4Wttjjdn ttjjdw||||dS)NrzRank failed to join in under rkF)rrXrmaxis_aliverrr TimeoutErrorr!remptyrrr rUrMrrrf)rGrrridxthread failed_ranksZfailurer,r,r-rs8      z#MultiThreadedTestCase._join_threadsc CsHd}d}|D]l\}}|d}t|tjr(td||t||dkr'tdj}qt|tr?d|d|d } t | t | t|t r_d t j|} t d | ||d|d | d 7}qt|trrt|jtkrr|dkrr|j}qt|dkr}t ||dkrtD]} || jkrtrtd || jdSt| jqdSdS)Nrnrrmz3Thread %s skipping test %s for following reason: %srrBzThread rpz seconds z'Caught exception: %s exiting thread %sz exited with exception: rorq) isinstancerNrOr-r.r+rYr$rrrlrPrrQformat_exception SystemExitrcoder)lenrrr%) rGrrr error_msgZ skip_coderrexcrrur,r,r-rfsR            z)MultiThreadedTestCase._check_return_codescCrr{rrr,r,r-r_rz MultiThreadedTestCase.world_sizecCr#rr$rr,r,r-r&sz(MultiThreadedTestCase._current_test_namerrvcCs |j|kr||||dSdS)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)rrsrr}yrrr,r,r-assertEqualOnRanks z'MultiThreadedTestCase.assertEqualOnRankcCs|j|kr |||dSdSr{)rrmrr,r,r-assertNotEqualOnRank's z*MultiThreadedTestCase.assertNotEqualOnRankrxryr{)r&r'r(__doc__queueQueuerrrr+rrrrr rr}r+rrrfr{r)r_r&rrr~r,r,rr-rIs0      0 rcLeZdZdeejejfdeddffdd Z dejdejfdd Z Z S) SaveForwardInputsModuleforward_inputscast_forward_inputsrNcs(ttdd|_||_||_dS)Nd)rrnnZLinearlrrrrrrr,r-r-s  z SaveForwardInputsModule.__init__r}cCs,||j|<||jr||jjjS|Sr{)rrrtoweightZdtyperr}r,r,r-forward7s "zSaveForwardInputsModule.forward r&r'r(rrModulerUZTensorrzrrr~r,r,rr-r, rcr) SaveForwardInputsModelrrrNcs,tt|||_t|||_||_dSr{)rrrc1c2rrrr,r-r=s    zSaveForwardInputsModel.__init__r}cCs||j|<|||Sr{)rrrrr,r,r-rGs zSaveForwardInputsModel.forwardrr,r,rr-r<rrccs|s tj|dtjd<dtjd<|r1|r)tjjjj }t j d|||dnt j d||dtj tj jjzdVWtj tj jj|rVt dSdStj tj jj|rkt ww) Nr MASTER_ADDRZ6789 MASTER_PORTZfakerr=rr_)rUrrrZr[testing _internalrr'Z FakeStorerrrrutilsZcountersclearrU)rr_Zinit_pgr'rr,r,r-_dynamo_dist_per_rank_initKs:        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 vr$dn|jg|_ t j d|jdd dS) NrZ12355)rrrzcuda:rKr=rmr) r setUpClassZ _exit_stack enter_contextrrrZr[rrZ device_idsrrrGrr,r-rqs  z.DynamoDistributedSingleProcTestCase.setUpClasscsttdSr{)rrUr tearDownClassrrr,r-rsz1DynamoDistributedSingleProcTestCase.tearDownClass)r&r'r(rr}rrr~r,r,rr-ris rc s\eZdZdZfddZfddZedefddZe d ed e d e dd fd dZ 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. crr{rrrr,r-rrz(DynamoDistributedMultiProcTestCase.setUpcs2tz t|jWdStyYdSwr{)rr rZrrrrrr,r-r s  z+DynamoDistributedMultiProcTestCase.tearDownrcCs tjSr{)rUrKr\rr,r,r-r_rwz-DynamoDistributedMultiProcTestCase.world_sizerrDrNcKs2tt||}||_||_|||dSr{)r addHandlerlogging NullHandlerrrrErFr,r,r-r+s z'DynamoDistributedMultiProcTestCase._run) r&r'r(rrr r{r)r_r}r+r+r~r,r,rr-rs $rc seZdZUdZeed<dZeed<dZee ed<e ddZ e ed <e e jd e fd d Ze dddZe fddZe fddZe ddededee fddZZS)MultiProcContinousTestrlr_rrN rdvz_filex)secondsrrcCstd)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)NotImplementedErrorrr,r,r- backend_strsz"MultiProcContinousTest.backend_strFcCr)z ProcessGroup init options. To be customized by sub test classes, e.g. ProcessGroupNCCLOpTest Here we return None. Nr,)rGZhigh_priority_streamr,r,r-optsszMultiProcContinousTest.optscstd|jkr|jksntd|jd|j|jr*t|j|j}nd}|}| }t d|tj ||j|j|||j dtj |_t d|jddS) 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=)rrr_rrZ pg_optionsrRank z setup complete)rrrr_rlr rrrrrarrrrZpg)rGrrrrrr,r-rs2  z!MultiProcContinousTest.setUpClasscsRtt|jrzt|jWn tyYnwtd|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) rrUrrr rZrrrarrrr,r-rs  z$MultiProcContinousTest.tearDownClasscCs||_||_||_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)rr_r r)rGrr_r r,r,r-run_ranks zMultiProcContinousTest.run_rank)Fr{)r&r'r(r_r)r*rr r r+rrr}abcabstractmethodrrrrrr~r,r,rr-r s.   r r{rry)TF)rr<rrr)rZrrrWrrKr`rQrrN contextlibr dataclassesrdatetimerenumr functoolsrrrior typingr r r r rZ unittest.mockrZtorch._logging._internalrrUZtorch._dynamo.test_caseZtorch.cuda.ncclZtorch.distributedrrZtorch._C._autogradrZtorch._C._distributed_c10drZtorch.nnrZ$torch.testing._internal.common_utilsrrrrrrrrrrrrZ5torch.testing._internal.distributed.multi_threaded_pgrr r!r basicConfigINFO getLoggerr&r-r#rYrGrerjrnrvrzr~rrrrrrrrrrrrrrr)rzrrrgetenvrrrrrr+rrrr*rrrrrrrrrrrrrrrrrrZ test_caserrr r,r,r,r-s,            8                   .       , i' # 8d  $