U <cw@sDUddlZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl m Z ddlmZddlmZddlmZddlmZmZmZddlmZddlmZmZmZddlZddlZddl m!Z"dd l#m$Z$m%Z%m&Z&m'Z'm(Z(m)Z)m*Z*m+Z+m,Z,ej-ej.d e/e0Z1Gd d d eZ2e2d de2dde2dde2dde2dde2dde2dde2dde2dde2dd e2d!d"e2d#d$e2d%d&e2d'd(e2d)d*e2d+d,d-Z3eGd.d/d/Z4d0d1Z5d2d3Z6d4d5Z7d6d7Z8d8d9Z9d:d;Z:dd?ZdDdEZ?dFdGZ@dHdIZAdJdKZBdLdMZCe)dNdOdPedQdRdPdSfdTdUZDe&rrdVZEneFeGdWdXZEdYdZiZHe%rd[eHd\<dvd]d^ZIeFd_d`daZJe dbdcZKdweFeFeFdddedfZLeFeMdgdhdiZNdaOeejPeQdj<dxeeMddkdldmZRdd_dndoZSGdpdqdqe$ZTdaUeVd_drdsZWdtduZXdS)yN)contextmanager) dataclass) timedelta)Enum)partialreducewraps)StringIO) NamedTupleOptionalUnion) TestCaseTEST_WITH_ROCMTEST_WITH_TSAN FILE_SCHEMAfind_free_portretry_on_connect_failures IS_SANDCASTLEsandcastle_skip_ifsandcastle_skip)levelc@seZdZUeed<eed<dS)TestSkip exit_codemessageN)__name__ __module__ __qualname__int__annotations__strr r N/tmp/pip-unpacked-wheel-gikjz4vx/torch/testing/_internal/common_distributed.pyr,s rHz5Skipped 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 reason)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_accessgenericc@seZdZiZdddhed<eed<ddhed<ddhed<iZddhed <dddhed <dddhed <dddhed <eed <dS) DistTestCasesr5mpiZuccZallgather_coalescedrzsendrecv anysourcez cpu barrierZglooZgpucudaZddpZsubgroupZpluginN)rrrZskip_collectivesetZbackend_featurer r r r!r8Hs    r8cstfdd}|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)Nr4 WORLD_SIZE multi-gpu-) torchr: is_availablesysexit TEST_SKIPSrrosenviron device_count)argskwargs world_sizefuncr r!wrapper]s  zskip_if_no_gpu..wrapperrrJrKr rIr!skip_if_no_gpuZs rNcstfdd}|S)Ncs:tjddkr0ttjddkr0ttdj||S)NBACKENDr9r<r2rCrDrr@rArBrrFrGrIr r!rKks z(skip_if_small_worldsize..wrapperrLrMr rIr!skip_if_small_worldsizejsrScstfdd}|S)Ncs>tjddkr4ttjdddkr4ttdj||S)NrOr9r<rPr3rQrRrIr r!rKus$z&skip_if_odd_worldsize..wrapperrLrMr rIr!skip_if_odd_worldsizetsrUcsfdd}|S)Ncstfdd}|S)Ncs<dkr.tjkr.ttdjn ||SdSNr5r=)r>r:rEr@rArBrrR)backendrJnr r!rKszCrequire_n_gpus_for_nccl_backend..decorator..wrapperrLrMrWrXrIr! decoratorsz2require_n_gpus_for_nccl_backend..decoratorr )rXrWrZr rYr!require_n_gpus_for_nccl_backend~s r[csfdd}|S)Ncstfdd}|S)Ncs<tjr"tjkr"||SttdjdS)Nr=r>r:r?rEr@rArBrrR)rJxr r!rKs z4skip_if_lt_x_gpu..decorator..wrapperrLrMr]rIr!rZsz#skip_if_lt_x_gpu..decoratorr )r]rZr r^r!skip_if_lt_x_gpus r_csfdd}|S)Ncstfdd}|S)NcsNdkr||Stjr4tjkr4||SttdjdSrVr\rR)rWrJr]r r!rKs   z9nccl_skip_if_lt_x_gpu..decorator..wrapperrLrMrWr]rIr!rZsz(nccl_skip_if_lt_x_gpu..decoratorr )rWr]rZr r`r!nccl_skip_if_lt_x_gpus racCst|}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    ricstfdd}|S)a2 Convenience decorator to set/unset NCCL_BLOCKING_WAIT flag. Note that use of this decorator will override the setting of NCCL_ASYNC_ERROR_HANDLING for the particular test. After the test, both NCCL_BLOCKING_WAIT and 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)NZNCCL_ASYNC_ERROR_HANDLING1ZNCCL_BLOCKING_WAIT)rCrDKeyError)rFrGZ cached_nccl_async_error_handlingZcached_nccl_blocking_waitretrIr r!rKs0    z(with_nccl_blocking_wait..wrapperrLrMr rIr!with_nccl_blocking_waits"rmcsfdd}|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)rCrDgetc10dZset_debug_level_from_envbarrier)rFrGZ old_levelrrl)rJlevelsr r!rKs   z:with_dist_debug_levels..decorator..wrapperrLrMrqrIr!rZsz)with_dist_debug_levels..decoratorr )rqrZr rrr!with_dist_debug_levelss rscCstt dS)Nz+c10d was not compiled with the Gloo backend)rroZis_gloo_availabler r r r! requires_gloosrtcCs<tstdSttjj|kd|tjj|SdS)N+c10d was not compiled with the NCCL backendzIRequires NCCL version greater than or equal to: {}, found: {}, reason: {}) rois_nccl_availablerrr>r:r5versionformat)rwmsgr r r!requires_nccl_versions rzcCstt dS)Nru)rrorvr r r r! requires_ncclsr{cCstt dS)Nz*c10d was not compiled with the MPI backend)rroZis_mpi_availabler r r r! requires_mpisr|csd_tfdd}|S)zSkips a test for ROCmTcs"ts||SttdjdS)Nr6)rr@rArBrrRrIr r!rK's zskip_if_rocm..wrapper) skip_if_rocmrrMr rIr!r}#sr}cCsttjdkdS)Nwin32z9This unit test case is not supportted on Windows platform)rr@platformr r r r! skip_if_win320sr localhostrTT)minutesFcCsJt}|r2t|tdd}tjj|||||Stj|||||dSdS)zL Creates a TCP store. Retries if the chosen port is already in use. rT)Z milliseconds)wait_for_workersN)rrrr>classesZ dist_c10dZTCPStorero)addrrH is_mastertimeoutrZ jit_classportZtimeout_millisecondr r r!create_tcp_store7s" riZ!DISTRIBUTED_TESTS_DEFAULT_TIMEOUTZ300Ztest_ddp_uneven_inputsiZtest_join_kwargscCs2tjdks|dkr tjjddStjj|dSdS)Nr~z 127.0.0.1)hostnameZ interface)r@rroZProcessGroupGloo create_devicerr r r!rZsrreturncCst|ddtSN.rd)TIMEOUT_OVERRIDErnrgTIMEOUT_DEFAULT)Ztest_idr r r! get_timeoutasrc csRtt}}tjtj}}z ||t_t_tjtjfVW5||t_t_XdSN)r r@stdoutstderr)Znew_outZnew_errZold_outZold_errr r r!captured_outputes r)rankrH 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. rTr)rrH sparse_dims dense_dimsc Sstt|dd|df}|gddt|D}t|dD](}t|td|df}||qBt|dgddt|D}t|||S)NrTcSsg|]}dqSrPr .0_r r r! |sz@simple_sparse_reduce_tests..generate..cSsg|]}dqSrr rr r r!rs) r>ZreshapeZarangerangecatzerosappendZonesZsparse_coo_tensor)rrHrrindicesshapervaluesr r r!generatews "z,simple_sparse_reduce_tests..generate)rHcs"tddfddtDS)NcSs||Srr )abr r r!zAsimple_sparse_reduce_tests..compute_sum..csg|]}|qSr r )rrfnrHr r!rszCsimple_sparse_reduce_tests..compute_sum..)rrrr rr! compute_sumsz/simple_sparse_reduce_tests..compute_sumcsDg|]<fddtDfddtDfqS)cs"g|]}|qSr r ri)rrrrHr r!rsz9simple_sparse_reduce_tests...csg|]}qSr r r)rrrrHr r!rs)r)rrrrrHrr!rs z.simple_sparse_reduce_tests..)rrP)r)rTr)rr)rrHrrr rr!simple_sparse_reduce_testsps       r)rHrWcsTtj}t||dkr$dtjd<d||kr8||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. r5rjZNCCL_MAX_NRINGSrTcs*i|]"}|t||dqS)rT)listrZnGPUs_per_processZvisible_devicesr r! s z(init_multigpu_helper..)r>r:rErrCrD)rHrWZnGPUsZ rank_to_GPUr rr!init_multigpu_helpers   rtmp_dir) init_methodrcCstatjtjd<ttjtjdttjtjdtjtjd}t||dk rn|tjd<nt tj|dtjd<dS)NZTEMP_DIRrpZtest_dirZinit_dirZ INIT_METHODZshared_init_file) tempfileTemporaryDirectoryrnamerCrDmkdirpathjoinr)rZ init_dir_pathr r r!initialize_temp_directoriess   rcCstdk rtdSr)rcleanupr r r r!cleanup_temp_dirsrcs8eZdZdZdZedddZeedddZ dd Z d.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)/MultiProcessTestCaserd rcCsdS)NFr selfr r r!_should_stop_test_suitesz,MultiProcessTestCase._should_stop_test_suitecCsdS)Nr rr r r!rHszMultiProcessTestCase.world_sizecs tfdd}t||S)Ncs"|j|jkr|ndSr)rMAIN_PROCESS_RANK_join_processesrrr r!rKs  z1MultiProcessTestCase.join_or_run..wrapper)rtypes MethodType)rrrKr rr! join_or_runsz MultiProcessTestCase.join_or_runrunTestN) method_namercs,t|t||}t||||dSr)super__init__getattrsetattrr)rrr __class__r r!rs  zMultiProcessTestCase.__init__cs8tg|_g|_|j|_tjddj|_ i|_ dS)NF)delete) rsetUpskip_return_code_checks processesrrrNamedTemporaryFiler file_name pid_to_piperrr r!rs  zMultiProcessTestCase.setUpcs(t|jD] }|qg|_dSr)rtearDownr terminate)rprr r!r s   zMultiProcessTestCase.tearDowncCs|ddSr)idrgrr r r!_current_test_namesz'MultiProcessTestCase._current_test_namecCsg|_tt|jD]r}tj\}}||jjdt ||| |j |fd}| t d|d|j||j|j<|j|qdS)Nzprocess )targetrrFzStarted process z with pid )rrrrHr>multiprocessingPiper_runrrrstartloggerinfopidrr)rprocrZ parent_connZ child_connprocessr r r!_start_processess  z%MultiProcessTestCase._start_processescCstjdj}||dS)Nspawn)r>rZ get_contextProcessr)rrr r r!_spawn_processes)sz%MultiProcessTestCase._spawn_processesc@seZdZdZdS)zMultiProcessTestCase.EventrTN)rrr GET_TRACEBACKr r r r!Event-srrc Cstd|tj||g}||kr|jrDtd|ddS|}td|d||tjj krt j ddB}t |||d||td |d W5QRX||krdSqdS) Nz(Starting event listener thread for rank Pipe closed for process z , stopping event listener threadzReceived event z on process zr+)moderProcess z sent traceback)rrr connectionwaitclosedrecvrrrrr faulthandlerZdump_tracebackflushseeksendread) parent_pipeZ signal_piperZ ready_pipeseventZtmp_filer r r!_event_listener0s&    z$MultiProcessTestCase._event_listener)r test_namerrcCs8ddlm}|d||}||_||_|||dS)Nr)_set_ddp_with_replicated_tensorT)Z.torch.nn.parallel._replicated_tensor_ddp_utilsrrrrun_test)clsrrrrrrr r r!rOs  zMultiProcessTestCase._run)rrc Csdtjjdd\}}tjtj|||jfdd}|t j dkrTt j dkrTtj ddt jd<zzt||Wntjk r}z6td |jd |d t|t td jW5d}~XYnbtk r&}zBtd td|jdtj |tt tj W5d}~XYnXW5|dk r@|d|dk sNt||XdS)NF)ZduplexT)rrFdaemonr~darwinrjZTORCH_SHOW_CPP_STACKTRACESrz skipping test z for following reason: r7zCaught exception: z exiting process z with exit code: )!r>rr threadingThreadrrrrr@rZ_CZ'_set_print_stack_traces_on_fatal_signalrCrDrrercloserunittestSkipTestrrrrArBr Exceptionrc traceback format_excTEST_ERROR_EXIT_CODE)rrrZsignal_recv_pipeZsignal_send_pipeZevent_listener_threadseer r r!r[s8    "  zMultiProcessTestCase.run_testc Cs0g}t|jD]x\}}|jdkr|j|j}z |tjj| ||fWqt k r}zt d|d|W5d}~XYqXq|D]\}}zZ| dr|jrt d|dWq|}t d|d|nt d|Wqt k r(}zt d|d|W5d}~XYqXqdS) Nz.zTiming out after z" seconds and killing subprocesses.g?)rrtimeritemsrrrrrrprintr>ractive_childrenrallrsleepr_check_no_test_errors_check_return_codes) rrr start_timeZsubprocess_errorrrrrr acelapsed elapsed_timer r r!rsD            z$MultiProcessTestCase._join_processescCsBt|jD]2\}}|jdkr,td||||j|jq dS)zV Checks that we didn't have any errors thrown in the child processes. Nz%Process {} timed out after {} seconds)rrr RuntimeErrorrxZassertNotEqualr)rr(rrr r r!r#s   z*MultiProcessTestCase._check_no_test_errorsc Cs|jd}ddt|jD}|r`d}|D],\}}|j|j}|d|tj|7}q*t|t|jD]D\}}|j dkrtd|||j |j |j d||j |j d qjt D]B} |j | j krtrtd |d | jdSt| jq|j |j dd |j |jd dS) z Checks that the return codes of all spawned processes match, and skips tests if they returned a return code indicating a skipping condition. rcSs$g|]\}}|jtjkr||fqSr )rrr)rrrr r r!rs z.z7Process {} exited with error code {} and exception: {} Nz3Process {} terminated or timed out after {} secondszJExpect process {} exit code to match Process 0 exit code of {}, but got {})ryz Skipping z) on sandcastle for the following reason: z.Expected zero exit code but got {} for pid: {})rrrrrrxrrr)r assertEqualrBrrrrrrrrr) rr(Z first_processZerrored_processesrcrr error_messagerskipr r r!r$s\     z(MultiProcessTestCase._check_return_codescCs |jdkS)Nrrrr r r!r%szMultiProcessTestCase.is_master)r)rrrrrboolrpropertyrrHrrrrrrrrrr staticmethodr classmethodrrrrr#r$r __classcell__r r rr!rs0   *%0 @rcCsHtdk r tSztdddddgjdkaWntk rBdaYnXtS) 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_RDMrF)EFA_PROBE_RESULT subprocessrun returncodeFileNotFoundErrorr r r r!has_efa,s r8cCstrddgSdS)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)r8r r r r! tp_transports=sr9)N)rT)N)YrloggingrrCr@rr r4rrrr contextlibrZ dataclassesrdatetimerenumr functoolsrrrior typingr r r r>Ztorch.cuda.ncclZtorch.distributedZ distributedroZ$torch.testing._internal.common_utilsr rrrrrrrr basicConfigINFO getLoggerrrrrBr8rNrSrUr[r_rarirmrsrtrzr{r|r}rrrrgetenvrrrrrrrrrrrrrr3r.r8r9r r r r!s      ,    .    , K