o 0 h@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,^/var/www/vscode/kcb/lib/python3.10/site-packages/torch/testing/_internal/common_distributed.pyr#<s  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 importno_acceleratorYzaccelerator is not available.c@seZdZiZhded<eed<ddhed<ddhed<iZhded <hded <hded <hded <eed <erCdhed<erLdhed<dSdS) DistTestCases>mpiuccr>allgather_coalescedrr>rLzsendrecv anysourcez cpu barrier>rLgloor>gpucudaddpsubgrouppluginhcclhpuxcclxpuN)r&r'r(skip_collectivesetbackend_featurerrr,r,r,r-rJ[s"          rJctfdd}|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)Nr4 WORLD_SIZE multi-gpu-z multi-xpu-)torchrP is_availablesysexit TEST_SKIPSr$r)osenviron device_countrrUrrW)argskwargs world_sizefuncr,r-wrapperus zskip_if_no_gpu..wrapperrrjrkr,rir-skip_if_no_gpuqs rncr[)Ncs>tjddkrttjddkrttdj|i|S)NBACKENDrKr\r0rcrdr)r`rarbr$rfrgrir,r-rks z(skip_if_small_worldsize..wrapperrlrmr,rir-skip_if_small_worldsizerscr[)NcsBtjddkrttjdddkrttdj|i|S)NrorKr\r2rqrrrir,r-rks$z&skip_if_odd_worldsize..wrapperrlrmr,rir-skip_if_odd_worldsizertrwcfdd}|S)Nctfdd}|S)Ncs>dkrtjkrttdjdS|i|SNr>r])r^rPrer`rarbr$rr)backendrjnr,r-rkszCrequire_n_gpus_for_nccl_backend..decorator..wrapperrlrmr{r|rir- decoratorsz2require_n_gpus_for_nccl_backend..decoratorr,)r|r{r~r,r}r-require_n_gpus_for_nccl_backends rcCs dd}|S)Ncr[)NcsFzddlm}m}|i|WSty"ttdjYdSw)Nr)AutoModelForMaskedLM BertConfigrF) transformersrr ImportErrorr`rarbr$)rfrgrrrir,r-rks  z?import_transformers_or_skip..decorator..wrapperrlrmr,rir-r~s z.import_transformers_or_skip..decoratorr,)r~r,r,r-import_transformers_or_skipsrcCstjo tj|kSN)r^rPr_rexr,r,r-at_least_x_gpusrcfdd}|S)Nctfdd}|S)Ncstjrtjkr|i|Str#tjkr#|i|Str3tjkr3|i|St t dj dS)Nr]) r^rPr_rerrUrrWr`rarbr$rr)rjrr,r-rksz4skip_if_lt_x_gpu..decorator..wrapperrlrmrrir-r~s z#skip_if_lt_x_gpu..decoratorr,)rr~r,rr-skip_if_lt_x_gpus  rcrx)Ncry)NcsVdkr |i|Stjrtjkr|i|SttdjdSrz)r^rPr_rer`rarbr$rr)r{rjrr,r-rks z9nccl_skip_if_lt_x_gpu..decorator..wrapperrlrmr{rrir-r~sz(nccl_skip_if_lt_x_gpu..decoratorr,)r{rr~r,rr-nccl_skip_if_lt_x_gpus rcCst|}d|vs Jd|vsJd|vsJ|d}|ddkr#|n|dd}||vs8Jd|d|dS) 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_erractualr,r,r-verify_ddp_error_loggeds     rcr[)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)NTORCH_NCCL_ASYNC_ERROR_HANDLINGTORCH_NCCL_BLOCKING_WAIT1)rcrdKeyError)rfrg cached_nccl_async_error_handlingcached_nccl_blocking_waitretrir,r-rksF     z(with_nccl_blocking_wait..wrapperrlrmr,rir-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)NTORCH_DISTRIBUTED_DEBUG)rcrdgetc10dset_debug_level_from_envbarrier)rfrg old_levelr"r)rjlevelsr,r-rk.s  z:with_dist_debug_levels..decorator..wrapperrlrmrrir-r~-sz)with_dist_debug_levels..decoratorr,)rr~r,rr-with_dist_debug_levels(s rcCtt dS)Nz+c10d was not compiled with the Gloo backend)rris_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_availablerrr^rPr>version)rmsgr,r,r-requires_nccl_versionIsrcCr)Nr)rrrr,r,r,r- requires_ncclUrrcCr)Nz*c10d was not compiled with the UCC backend)rris_ucc_availabler,r,r,r- requires_ucc[rrcCr)Nz*c10d was not compiled with the MPI backend)rris_mpi_availabler,r,r,r- requires_mpiarrcCs$tjo ttjd}t| dS)Nrz"multicast support is not available)r^rPr_rhas_multicast_supportrCUDAr)rr,r,r-requires_multicast_supporths  rcsd_tfdd}|S)zSkips a test for ROCmTcs&ts |i|SttdjdS)Nr@)rr`rarbr$rrrir,r-rkwsz*skip_if_rocm_multiprocess..wrapper)skip_if_rocm_multiprocessrrmr,rir-rssrcCsttjdkdS)Nwin32z8This unit test case is not supported on Windows platform)rr`platformr,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. rPz3sm_is_or_later() is only supported for CUDA devicesNF)type ValueErrorr^rhiprPget_device_capability)rrrr,r,r-sm_is_or_higher_thans  r localhostrvT)minutesFc CsHt}|rt|tdd}tjj|||||Stj||||||dS)zL Creates a TCP store. Retries if the chosen port is already in use. rv) milliseconds)wait_for_workers use_libuv)rr)rr^classes dist_c10dTCPStorer) addrrh is_mastertimeoutr jit_classrporttimeout_millisecondr,r,r-create_tcp_stores   ri!DISTRIBUTED_TESTS_DEFAULT_TIMEOUT300test_ddp_uneven_inputsitest_join_kwargscCs.tjdks |durtjjddStjj|dS)Nrz 127.0.0.1)hostname interface)r`rrProcessGroupGloo create_devicerr,r,r-rsrcCst|ddtSN.r)TIMEOUT_OVERRIDErrTIMEOUT_DEFAULT)test_idr,r,r- get_timeoutsrc cs`tt}}tjtj}}z||t_t_tjtjfVW||t_t_dS||t_t_wr)r r`stdoutstderr)new_outnew_errold_outold_errr,r,r-captured_outputs"rrankrh 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. rvrrrh sparse_dims dense_dimsc Sstt|dd|df}|gddt|D}t|dD]}t|td|df}||q!t|dgddt|D}t|||S)NrvcSg|]}dqSrur,.0_r,r,r- z@simple_sparse_reduce_tests..generate..cSrrr,rr,r,r-rr) r^reshapearangerangecatzerosappendonessparse_coo_tensor)rrhrrindicesshapervaluesr,r,r-generates "z,simple_sparse_reduce_tests..generatecs ttjfddtDS)Ncsg|]}|qSr,r,)rrfnrhr,r-rzCsimple_sparse_reduce_tests..compute_sum..)roperatoraddrrr,rr- compute_sumsz/simple_sparse_reduce_tests..compute_sumcsDg|]fddtDfddtDfqS)cs"g|] }|qSr,r,ri)r rrrhr,r-rsz9simple_sparse_reduce_tests...csg|] }qSr,r,r)r r rrhr,r-rs)r)rr rrrhr r-rsz.simple_sparse_reduce_tests..)rru)rN)rvr)r)r)rrhrrr,rr-simple_sparse_reduce_testss      rr{cs^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. rvcs*i|]}|t||dqSrv)listrnGPUs_per_processvisible_devicesr,r- sz(init_multigpu_helper..)r^rPrerrUrrWr)rhr{nGPUs rank_to_GPUr,rr-init_multigpu_helpers    rtmp_dir init_methodcCstatjtjd<ttjtjdttjtjdtjtjd}t||dur8|tjd<dSt tj|dtjd<dS)NTEMP_DIRrtest_dirinit_dir INIT_METHODshared_init_file) tempfileTemporaryDirectoryrnamercrdmkdirpathjoinr)r init_dir_pathr,r,r-initialize_temp_directoriess  r+cCstdur tdSdSr)rcleanupr,r,r,r-cleanup_temp_dir,s r-c 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_suitecCr1)NTr,r2r,r,r-destroy_pg_upon_exitQz)MultiProcessTestCase.destroy_pg_upon_exitcCtSrDEFAULT_WORLD_SIZEr2r,r,r-rhUr7zMultiProcessTestCase.world_sizec tfdd}t||S)Ncs$|j|jkr |dSdSr)rMAIN_PROCESS_RANK_join_processesr2rr,r-rkZs  z1MultiProcessTestCase.join_or_run..wrapperrtypes MethodTyper3r rkr,rr- join_or_runY z MultiProcessTestCase.join_or_runrunTest method_name methodNameNc |dkr|}t|zt||}t||||WdSty@}z|dkr5td|jd||WYd}~dSd}~wwNrDzno such test method in z: super__init__getattrsetattrrBAttributeErrorr __class__r3rErFr erOr,r-rKg  zMultiProcessTestCase.__init__cs8tg|_g|_|j|_tjddj|_ i|_ dS)NF)delete) rJsetUpskip_return_code_checks processesr<rr$NamedTemporaryFiler& file_name pid_to_piper2rRr,r-rUvs  zMultiProcessTestCase.setUpcs(t|jD]}|qg|_dSr)rJtearDownrW terminate)r3prRr,r-r[s    zMultiProcessTestCase.tearDowncC|ddSridrr2r,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)targetr&rfrgzStarted process %s with pid %s)rWrr)rhr^multiprocessingPiperO_runr+rarYrLstartloggerinfopidrZr)r3procr parent_conn child_connprocessr,r,r-_start_processess   z%MultiProcessTestCase._start_processescCstjdj}||dS)Nspawn)r^rd get_contextProcessro)r3rkr,r,r-_spawn_processessz%MultiProcessTestCase._spawn_processesc@seZdZdZdS)zMultiProcessTestCase.EventrvN)r&r'r( GET_TRACEBACKr,r,r,r-EventsrurcCstd| 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)rhrird connectionwaitclosedrecvr/rurtr$rX faulthandlerdump_tracebackflushseeksendread) parent_pipe signal_piper ready_pipeseventtmp_filer,r,r-_event_listeners,     z$MultiProcessTestCase._event_listener test_namerYcKs$||}||_||_|||dSr)rrYrun_testclsrrrYrrgr3r,r,r-rfszMultiProcessTestCase._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)duplexT)rcrfdaemonrdarwinrTORCH_SHOW_CPP_STACKTRACESz4Process %s skipping test %s for following reason: %srDz;Caught exception: %s exiting process %s with exit code: %s)%r^rdre threadingThreadr/rrrgr`r_C'_set_print_stack_traces_on_fatal_signalrcrdrLunittestSkipTestrhrir+rarbr$ Exceptionr traceback format_excTEST_ERROR_EXIT_CODErr)closer6rdestroy_process_groupAssertionErrorr)r3rrsignal_recv_pipesignal_send_pipeevent_listener_threadser,r,r-rsX          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) enumeraterWexitcoderZrjrr/rurtrConnectionErrorrhrpollryrirz)r3pipesrrnpiperQrrr,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)r)rr]r,r,r- 9sz7MultiProcessTestCase._join_processes..zTiming out after z" seconds and killing subprocesses.g?)rr`timerrWrr/rprintr^rdactive_childrenr\allrsleeprV_check_no_test_errors_check_return_codesrZrr) r3r r start_timesubprocess_errorrr]racelapsed elapsed_timerr,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. Nrz timed out after  seconds)rrWr RuntimeErrorassertNotEqualr)r3rrr]r,r,r-rSs 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.rz exited with error code z and exception:   terminated or timed out after rzExpect 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: )rWrhwarningrrZrjrzr/rrr assertEqualrbrr$rrir`r%rr) r3r first_processerrored_processesrrrn error_messager]skipr,r,r-r^sR          z(MultiProcessTestCase._check_return_codescCs |jdkS)Nrrr2r,r,r-r zMultiProcessTestCase.is_masterrDrDrN) r&r'r(r<rboolr4propertyr6r)rhrBr+rKrUr[rarorsrru staticmethodr classmethodrfrrr=rrr __classcell__r,r,rRr-r/@s6    3 % 0 @r/csBeZdZfddZddZdefddZdd Zd d ZZ S) DistributedTestBasect|dSrrJrUrsr2rRr,r-rU  zDistributedTestBase.setUpcCs(z t|jWdStyYdSwr)rcremoverYOSErrorr2r,r,r-r[s  zDistributedTestBase.tearDownrcCs(d|vrdSd|vr dSd|vrdSdS)NrPr>rUrTrWrVrNr,r3rr,r,r-r{szDistributedTestBase.backendcCsrt|}tj|j|}tjj|||j|j |dd||vs,d||vr3tj |j tjj S)Nr{rhrstorer>rV)r^get_device_modulere distributed FileStorerYinit_process_groupr{rhr acceleratorset_device_indexdistributed_c10d_get_default_group)r3rnum_visible_devicesrr,r,r- create_pgs zDistributedTestBase.create_pgcs&t|fddt|jDS)Ncsi|]}||gqSr,r,rrr,r-rsz6DistributedTestBase.rank_to_device..)r^rrerrhrr,rr-rank_to_devicesz"DistributedTestBase.rank_to_device) r&r'r(rUr[r+r{rrrr,r,rRr-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 itertoolsproductdictzipsubTestr^_dynamoresetrr) cls_instrr test_argsrsubtest_config_itemssubtest_config_keyssubtest_config_valuesrsubtest_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)fi_infoz-pefaz-t 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 shmuvN)rr,r,r,r- tp_transportssrcs:dur tt|dSddtfdd}|S)z+ Wrapper to use with a test method N)rrhcsftt}fddfdd}g}tD]}tj|||fd}|||q|S)Nc tjjkSrrr_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)Nthreadedr{rrhr) rr BaseExceptionMultiThreadedTestCaseexception_queueputr`exc_infor!exception_handler)rworld_pgrex)callbackrrhr,r-worker$s    zYspawn_threads_and_init_comms.._run_test_method_with_multi_threads..workerrcrf)rr HashStorerrrrgr)rhr global_storerthreadsrtr,)rrrrhr-#_run_test_method_with_multi_threadss   zIspawn_threads_and_init_comms.._run_test_method_with_multi_threadsc sXtjjdzfdd}t|WtjjddStjjdw)NTcsgRiSrr,r,)rfrjrgr3r,r-@r z?spawn_threads_and_init_comms..wrapper..F)r^r_distributed_c10d_set_thread_isolation_moder  _join_threads)r3rfrgrrrjrh)rfrgr3r-rk;s "z-spawn_threads_and_init_comms..wrapper)rspawn_threads_and_init_commsr)rjrrhrkr,r r-r!s r!cseZdZdZeZdZddZd(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)+r a5 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_RANKrrr2rr,r-rk\s  z2MultiThreadedTestCase.join_or_run..wrapperr>rAr,rr-rB[rCz!MultiThreadedTestCase.join_or_runrDrErFrNc rGrHrIrPrRr,r-rKerSzMultiThreadedTestCase.__init__cCdSrr,r2r,r,r-perThreadSetUptr7z$MultiThreadedTestCase.perThreadSetUpcCr#rr,r2r,r,r-perThreadTearDownxr5z'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 rrN)rJrUr"rrrcrdr2rRr,r-rU{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)rJr[rr2rRr,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 Tcrrrr,rr,r-rr z.world_is_validz Invalid worldrN)r^rrrrarrrrOrrrrhrrrfrgrr)r3rrrrr,rr-_spawn_threadss  z$MultiThreadedTestCase._spawn_threadscKsH||}||_t|drt|_tj|j_tj|j_ | |||dS)N_tls) rhasattrrlocalr'r _precision precision_rel_tolrel_tolrun_test_with_threaded_pg)rrrrhrgr3r,r,r-rfs    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. r r N)rrrOrr$rLr rrr`rr!rrr%)r3rrrhrr,r,r-r.s&    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 rF)rrr)maxis_aliver rr TimeoutErrorr!remptyrrr r^rrrr)rrr ridxthread failed_ranksfailurer,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)Nrrrvz3Thread %s skipping test %s for following reason: %srrDzThread rz seconds z'Caught exception: %s exiting thread %sz exited with exception: rr) isinstancerrrhrir+rbr$r1rrrr)rformat_exception SystemExitrcoder)lenrrr%) rr5rr  error_msg skip_coderrexcrrr,r,r-rsR            z)MultiThreadedTestCase._check_return_codescCr8rr9r2r,r,r-rhr7z MultiThreadedTestCase.world_sizecCr^rr_r2r,r,r-rasz(MultiThreadedTestCase._current_test_namerrcCs |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)rrr3ryrrr,r,r-assertEqualOnRanks z'MultiThreadedTestCase.assertEqualOnRankcCs|j|kr |||dSdSr)rrr?r,r,r-assertNotEqualOnRank's z*MultiThreadedTestCase.assertNotEqualOnRankrrr)r&r'r(__doc__queueQueuerr"rBr+rKr$r%rUr[r&rrfr.rrrr)rhrarArBrr,r,rRr-r Is0      0 r cLeZdZdeejejfdeddffdd Z dejdejfdd Z Z S) SaveForwardInputsModuleforward_inputscast_forward_inputsrNcs(ttdd|_||_||_dS)Nd)rJrKnnLinearlrHrIr3rHrIrRr,r-rK-s  z SaveForwardInputsModule.__init__rcCs,||j|<||jr||jjjS|Sr)rHrMrItoweightdtyper3rr,r,r-forward7s "zSaveForwardInputsModule.forward r&r'r(rrKModuler^TensorrrKrSrr,r,rRr-rG, rGcrF) SaveForwardInputsModelrHrIrNcs,tt|||_t|||_||_dSr)rJrKrGc1c2rHrNrRr,r-rK=s    zSaveForwardInputsModel.__init__rcCs||j|<|||Sr)rHrZrYrRr,r,r-rSGs zSaveForwardInputsModel.forwardrTr,r,rRr-rX<rWrXccs|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_ADDR6789 MASTER_PORTfakerr>rrh)r^rrrcrdtesting _internalrrb FakeStorerrrrutilscountersclearr)rrhinit_pgrbrr,r,r-_dynamo_dist_per_rank_initKs:        rgcs4eZdZdZefddZefddZZS)#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) Nr12355)r[r]rzcuda:rPr>rvr_) rJ setUpClass _exit_stack enter_contextrrrcrdrr device_idsrrrrRr,r-rjqs  z.DynamoDistributedSingleProcTestCase.setUpClasscsttdSr)rrrJ tearDownClassrnrRr,r-rosz1DynamoDistributedSingleProcTestCase.tearDownClass)r&r'r(rCrrjrorr,r,rRr-rhis rhc 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. crrrr2rRr,r-rUrz(DynamoDistributedMultiProcTestCase.setUpcs2tz t|jWdStyYdSwr)rJr[rcrrYrr2rRr,r-r[s  z+DynamoDistributedMultiProcTestCase.tearDownrcCs tjSr)r^rPrer2r,r,r-rhrz-DynamoDistributedMultiProcTestCase.world_sizerrrYNcKs2tt||}||_||_|||dSr)r addHandlerlogging NullHandlerrrYrrr,r,r-rfs z'DynamoDistributedMultiProcTestCase._run) r&r'r(rCrUr[rr)rhrr+rfrr,r,rRr-rps $rpc seZdZUdZeed<dZeed<dZee ed<e ddZ e ed <e e jd e fd d Ze dddZe fddZe fddZe ddededee fddZZS)MultiProcContinousTestrurhrrN 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)NotImplementedErrorrnr,r,r- backend_strsz"MultiProcContinousTest.backend_strFcCr1)z ProcessGroup init options. To be customized by sub test classes, e.g. ProcessGroupNCCLOpTest Here we return None. Nr,)rhigh_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=)r{rhrr pg_optionsrRank z setup complete)rJrjrrhrrurrr{ryrrrrrpg)rrr{r{rRr,r-rjs2  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. r}z teardown completeN) rrrJrorurcrrrrrnrRr,r-ros  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)rrhrur)rrrhrur,r,r-run_ranks zMultiProcContinousTest.run_rank)Fr)r&r'r(rhr)r*rrur r+rrrabcabstractmethodryr{rjrorrr,r,rRr-rts.   rtrrr)TF)rr{rrrrdrcrDrr`r$rrrr?r contextlibr dataclassesrdatetimerenumr functoolsrrrior typingr r r r r unittest.mockrtorch._logging._internalrr^torch._dynamo.test_casetorch.cuda.nccltorch.distributedrrtorch._C._autogradrtorch._C._distributed_c10drtorch.nnrK$torch.testing._internal.common_utilsrrrrrrrrrrrr5torch.testing._internal.distributed.multi_threaded_pgrr r!r  basicConfigINFO getLoggerr&rhr#rbrJrnrsrwrrrrrrrrrrrrrrrrrr)rrrrgetenvrrrrrr+rrr%r*r+r-r:r/rrrrrrrr!r rUrGrXrgr test_caserhrprtr,r,r,r-s,            8                   .       , i' # 8d  $