From 0497b7934f608f4c28bc6d2c4fa5b140e19cbd76 Mon Sep 17 00:00:00 2001 From: Pranjal Swarup Date: Fri, 27 Jun 2025 11:44:13 -0400 Subject: [PATCH] Update RCCL-tests in `examples` folder (#261) - Create a local copy for ROCm/rccl-tests for our examples. - Update argument parsing to no longer use getopt_long. - Workaround for Dyninst instrumentation. --------- Co-authored-by: David Galiffi [ROCm/rocprofiler-systems commit: 4e5029221b1fc79fab7c05196fb0b72780bbc6d4] --- .../examples/rccl/CMakeLists.txt | 49 +- .../examples/rccl/rccl-tests/Makefile | 23 + .../examples/rccl/rccl-tests/src/Makefile | 174 ++ .../rccl/rccl-tests/src/all_gather.cpp | 110 + .../rccl/rccl-tests/src/all_reduce.cpp | 126 ++ .../examples/rccl/rccl-tests/src/alltoall.cpp | 112 + .../rccl/rccl-tests/src/alltoallv.cpp | 218 ++ .../rccl/rccl-tests/src/broadcast.cpp | 131 ++ .../examples/rccl/rccl-tests/src/common.cpp | 1987 +++++++++++++++++ .../examples/rccl/rccl-tests/src/common.h | 431 ++++ .../examples/rccl/rccl-tests/src/gather.cpp | 143 ++ .../rccl/rccl-tests/src/git_version.h | 6 + .../rccl/rccl-tests/src/hypercube.cpp | 144 ++ .../rccl/rccl-tests/src/nccl1_compat.h | 70 + .../rccl/rccl-tests/src/rccl_float8.h | 1049 +++++++++ .../examples/rccl/rccl-tests/src/reduce.cpp | 142 ++ .../rccl/rccl-tests/src/reduce_scatter.cpp | 131 ++ .../examples/rccl/rccl-tests/src/scatter.cpp | 137 ++ .../examples/rccl/rccl-tests/src/sendrecv.cpp | 135 ++ .../examples/rccl/rccl-tests/src/timer.cc | 35 + .../examples/rccl/rccl-tests/src/timer.h | 17 + .../rccl/rccl-tests/verifiable/Makefile | 74 + .../rccl-tests/verifiable/inexact_regress.cpp | 236 ++ .../rccl/rccl-tests/verifiable/verifiable.cpp | 1563 +++++++++++++ .../rccl/rccl-tests/verifiable/verifiable.h | 67 + .../rccl/rccl-tests/verifiable/verifiable.mk | 33 + 26 files changed, 7330 insertions(+), 13 deletions(-) create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/Makefile create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/src/Makefile create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/src/all_gather.cpp create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/src/all_reduce.cpp create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/src/alltoall.cpp create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/src/alltoallv.cpp create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/src/broadcast.cpp create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/src/common.cpp create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/src/common.h create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/src/gather.cpp create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/src/git_version.h create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/src/hypercube.cpp create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/src/nccl1_compat.h create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/src/rccl_float8.h create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/src/reduce.cpp create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/src/reduce_scatter.cpp create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/src/scatter.cpp create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/src/sendrecv.cpp create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/src/timer.cc create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/src/timer.h create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/verifiable/Makefile create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/verifiable/inexact_regress.cpp create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/verifiable/verifiable.cpp create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/verifiable/verifiable.h create mode 100644 projects/rocprofiler-systems/examples/rccl/rccl-tests/verifiable/verifiable.mk diff --git a/projects/rocprofiler-systems/examples/rccl/CMakeLists.txt b/projects/rocprofiler-systems/examples/rccl/CMakeLists.txt index bdfc878bda..6546183efb 100644 --- a/projects/rocprofiler-systems/examples/rccl/CMakeLists.txt +++ b/projects/rocprofiler-systems/examples/rccl/CMakeLists.txt @@ -1,3 +1,25 @@ +# MIT License +# +# Copyright (c) 2025 Advanced Micro Devices, Inc. All Rights Reserved. +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all +# copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. + cmake_minimum_required(VERSION 3.18.4 FATAL_ERROR) project(rocprofiler-systems-rccl-example LANGUAGES CXX) @@ -28,7 +50,7 @@ function(rccl_message _MSG_TYPE) endif() endfunction() -find_package(hip HINTS ${ROCmVersion_DIR} PATHS ${ROCmVersion_DIR}) +find_package(hip HINTS ${ROCmVersion_DIR} ${ROCM_PATH} /opt/rocm) if(NOT hip_FOUND) rccl_message(AUTHOR_WARNING "${PROJECT_NAME} skipped. Missing HIP...") @@ -39,9 +61,10 @@ if( "${CMAKE_PROJECT_NAME}" STREQUAL "rocprofiler-systems" AND ("$ENV{ROCPROFSYS_CI}" OR ROCPROFSYS_CI OR ROCPROFSYS_BUILD_CI) ) - find_package(rccl QUIET) # avoid generating warning in CI + # avoid generating warning in CI + find_package(rccl HINTS ${ROCmVersion_DIR} ${ROCM_PATH} /opt/rocm QUIET) else() - find_package(rccl) + find_package(rccl HINTS ${ROCmVersion_DIR} ${ROCM_PATH} /opt/rocm) endif() if(NOT rccl_FOUND) @@ -50,19 +73,22 @@ if(NOT rccl_FOUND) endif() if(hip_FOUND AND rccl_FOUND) - include(FetchContent) - FetchContent_Declare(rccl-tests GIT_REPOSITORY https://github.com/ROCm/rccl-tests.git) + # Use existing rccl-tests source in the project binary directory + set(rccl-tests_SOURCE_DIR "${CMAKE_CURRENT_SOURCE_DIR}/rccl-tests") + set(rccl-tests_BUILD_DIR "${CMAKE_CURRENT_BINARY_DIR}/rccl-tests") + file(COPY ${rccl-tests_SOURCE_DIR}/ DESTINATION ${rccl-tests_BUILD_DIR}) - # After the following call, the CMake targets defined by googletest and Catch2 will be - # available to the rest of the build - FetchContent_Populate(rccl-tests) + if(NOT EXISTS "${rccl-tests_SOURCE_DIR}") + rccl_message(AUTHOR_WARNING "${PROJECT_NAME} skipped. Missing rccl-tests source at ${rccl-tests_SOURCE_DIR}") + return() + endif() get_filename_component(rccl_ROOT_DIR "${rccl_INCLUDE_DIR}" DIRECTORY) rccl_message(STATUS "Building rccl-tests...") execute_process( COMMAND make HIP_HOME=${ROCM_PATH} RCCL_HOME=${rccl_ROOT_DIR} - WORKING_DIRECTORY ${CMAKE_BINARY_DIR}/_deps/rccl-tests-src + WORKING_DIRECTORY ${rccl-tests_BUILD_DIR} RESULT_VARIABLE _RCCL_BUILD_RET ERROR_VARIABLE _RCCL_BUILD_ERR OUTPUT_VARIABLE _RCCL_BUILD_OUT @@ -74,10 +100,7 @@ if(hip_FOUND AND rccl_FOUND) rccl_message(${_RCCL_BUILD_OUT}) rccl_message(AUTHOR_WARNING "Failed to build rccl-tests: ${_RCCL_BUILD_ERR}") else() - file( - GLOB RCCL_TEST_EXECUTABLES - ${CMAKE_BINARY_DIR}/_deps/rccl-tests-src/build/*_perf - ) + file(GLOB RCCL_TEST_EXECUTABLES ${rccl-tests_BUILD_DIR}/build/*_perf) set(_RCCL_TEST_TARGETS) foreach(_EXE ${RCCL_TEST_EXECUTABLES}) diff --git a/projects/rocprofiler-systems/examples/rccl/rccl-tests/Makefile b/projects/rocprofiler-systems/examples/rccl/rccl-tests/Makefile new file mode 100644 index 0000000000..f652b78a99 --- /dev/null +++ b/projects/rocprofiler-systems/examples/rccl/rccl-tests/Makefile @@ -0,0 +1,23 @@ +# +# Copyright (c) 2017, NVIDIA CORPORATION. All rights reserved. +# +# See LICENCE.txt for license information +# + +BUILDDIR ?= build +override BUILDDIR := $(abspath $(BUILDDIR)) + +.PHONY: all clean + +default: src.build + +TARGETS=src + +all: ${TARGETS:%=%.build} +clean: ${TARGETS:%=%.clean} + +%.build: + ${MAKE} -C $* build BUILDDIR=${BUILDDIR} + +%.clean: + ${MAKE} -C $* clean BUILDDIR=${BUILDDIR} diff --git a/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/Makefile b/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/Makefile new file mode 100644 index 0000000000..ad6bf4ab7a --- /dev/null +++ b/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/Makefile @@ -0,0 +1,174 @@ +# +# Copyright (c) 2015-2022, NVIDIA CORPORATION. All rights reserved. +# Modifications are Copyright (c) 2019-2025 Advanced Micro Devices, Inc. All rights reserved. +# +# See LICENSE.txt for license information +# + +ROCM_PATH ?= /opt/rocm +MPI_HOME ?= /usr/lib/x86_64-linux-gnu +PREFIX ?= /usr/local +VERBOSE ?= 0 +DEBUG ?= 0 +NCCL_HOME ?= "" +CUSTOM_RCCL_LIB ?= "" + +HIPCC ?= $(ROCM_PATH)/bin/amdclang++ +HIPCONFIG = $(ROCM_PATH)/bin/hipconfig +CXX = $(HIPCC) + +HIPCUFLAGS := -std=c++14 +LDFLAGS := +HIPLDFLAGS := + +HIP_VERSION = $(strip $(shell which $(HIPCONFIG) >/dev/null && $(HIPCONFIG) --version)) +HIP_MAJOR = $(shell echo $(HIP_VERSION) | cut -d "." -f 1) +HIP_MINOR = $(shell echo $(HIP_VERSION) | cut -d "." -f 2) + +# Better define GPU_TARGETS in your environment to the minimal set +# of archs to reduce compile time. +# Currently, supports gfx906,gfx908,gfx90a,gfx942,gfx950,gfx1030,gfx1100,gfx1101,gfx1102,gfx1200,gfx1201 +ifndef GPU_TARGETS +GPU_TARGETS = gfx906 gfx908 gfx90a + ifeq ($(shell test "0$(HIP_MAJOR)" -eq 6; echo $$?),0) + # Include gfx942 support if we're using ROCm 6.0 or above + GPU_TARGETS += gfx942 + ifeq ($(shell test "0$(HIP_MINOR)" -ge 5; echo $$?),0) + # Include gfx950 support if we're using ROCm 6.5 or above + GPU_TARGETS += gfx950 + endif + endif +GPU_TARGETS += gfx1030 gfx1100 gfx1101 gfx1102 gfx1200 gfx1201 +endif + +GPU_TARGETS_FLAGS = $(foreach target,$(GPU_TARGETS),"--offload-arch=$(target)") + +#CUDA_VERSION = $(strip $(shell which $(NVCC) >/dev/null && $(NVCC) --version | grep release | sed 's/.*release //' | sed 's/\,.*//')) +#CUDA_MAJOR = $(shell echo $(CUDA_VERSION) | cut -d "." -f 1) +#CUDA_MINOR = $(shell echo $(CUDA_VERSION) | cut -d "." -f 2) +# +## Better define NVCC_GENCODE in your environment to the minimal set +## of archs to reduce compile time. +#ifeq ($(shell test "0$(CUDA_MAJOR)" -eq 12 -a "0$(CUDA_MINOR)" -ge 8 -o "0$(CUDA_MAJOR)" -ge 13; echo $$?),0) +## Include Blackwell support if we're using CUDA12.8 or above +#NVCC_GENCODE ?= -gencode=arch=compute_80,code=sm_80 \ +# -gencode=arch=compute_90,code=sm_90 \ +# -gencode=arch=compute_100,code=sm_100 \ +# -gencode=arch=compute_120,code=sm_120 \ +# -gencode=arch=compute_120,code=compute_120 +#else ifeq ($(shell test "0$(CUDA_MAJOR)" -ge 12; echo $$?),0) +#NVCC_GENCODE ?= -gencode=arch=compute_60,code=sm_60 \ +# -gencode=arch=compute_61,code=sm_61 \ +# -gencode=arch=compute_70,code=sm_70 \ +# -gencode=arch=compute_80,code=sm_80 \ +# -gencode=arch=compute_90,code=sm_90 \ +# -gencode=arch=compute_90,code=compute_90 +#else ifeq ($(shell test "0$(CUDA_MAJOR)" -ge 11; echo $$?),0) +#NVCC_GENCODE ?= -gencode=arch=compute_60,code=sm_60 \ +# -gencode=arch=compute_61,code=sm_61 \ +# -gencode=arch=compute_70,code=sm_70 \ +# -gencode=arch=compute_80,code=sm_80 \ +# -gencode=arch=compute_80,code=compute_80 +#else +#NVCC_GENCODE ?= -gencode=arch=compute_35,code=sm_35 \ +# -gencode=arch=compute_50,code=sm_50 \ +# -gencode=arch=compute_60,code=sm_60 \ +# -gencode=arch=compute_61,code=sm_61 \ +# -gencode=arch=compute_70,code=sm_70 \ +# -gencode=arch=compute_70,code=compute_70 +#endif + +ifneq ($(NCCL_HOME), "") +HIPCUFLAGS += -I$(NCCL_HOME)/ -I$(NCCL_HOME)/include +HIPLDFLAGS += -Wl,-rpath,$(NCCL_HOME) -L$(NCCL_HOME) -L$(NCCL_HOME)/lib +endif + +HIPCUFLAGS += -I$(ROCM_PATH)/include +HIPCUFLAGS += -I$(ROCM_PATH)/include/hip +HIPCUFLAGS += -x hip -D__HIP_PLATFORM_AMD__ -D__HIPCC__ $(GPU_TARGETS_FLAGS) +LDFLAGS += -L$(ROCM_PATH)/lib -lhsa-runtime64 -lrt +ifneq ($(CUSTOM_RCCL_LIB), "") +HIPLDFLAGS += -L$(CUSTOM_RCCL_LIB) +endif +HIPLDFLAGS += -L$(ROCM_PATH)/lib -lhsa-runtime64 -lamdhip64 -lstdc++ -lrt -pthread + +ifeq ($(DEBUG), 0) +HIPCUFLAGS += -O3 +else +HIPCUFLAGS += -O0 -g -ggdb3 +endif + +ifeq ($(VERBOSE), 0) +.SILENT: +endif + +.PHONY: build clean + +BUILDDIR ?= ../build +HIPIFY_DIR ?= $(BUILDDIR)/hipify + +.PRECIOUS: $(HIPIFY_DIR)/%.cpp $(HIPIFY_DIR)/%.h + +ifeq ($(MPI), 1) +HIPCUFLAGS += -DMPI_SUPPORT -I${MPI_HOME}/include -I${MPI_HOME}/include/openmpi -I${MPI_HOME}/openmpi/include -I${MPI_HOME}/openmpi/include/openmpi +HIPLDFLAGS += -L${MPI_HOME}/lib -L${MPI_HOME}/openmpi/lib -lmpi +else ifeq ($(MPICH), 1) +HIPCUFLAGS += -DMPI_SUPPORT -I${MPI_HOME}/include -I${MPI_HOME}/mpich/include -I/usr/include/x86_64-linux-gnu/mpich +HIPLDFLAGS += -L${MPI_HOME}/lib -L${MPI_HOME}/mpich/lib -lmpich +endif + +LIBRARIES += rccl +HIPLDFLAGS += $(LIBRARIES:%=-l%) + +DST_DIR := $(BUILDDIR) +SRC_FILES := $(wildcard *.cpp) +OBJ_FILES := $(SRC_FILES:%.cpp=${DST_DIR}/%.o) +BIN_FILES_LIST := all_reduce all_gather broadcast reduce_scatter reduce alltoall scatter gather sendrecv alltoallv +BIN_FILES := $(BIN_FILES_LIST:%=${DST_DIR}/%_perf) + +GIT_VERSION_FILE := ${DST_DIR}/src/git_version.cpp +GIT_REV := $(shell git log --pretty=format:'%h' -n 1) +GIT_DIFF := $(shell git diff --quiet --exit-code || echo +) +GIT_BRANCH := $(shell git rev-parse --abbrev-ref HEAD) + +build: ${BIN_FILES} + +clean: + rm -rf ${DST_DIR} + +TEST_VERIFIABLE_SRCDIR := ../verifiable +TEST_VERIFIABLE_BUILDDIR := $(BUILDDIR)/verifiable +include ../verifiable/verifiable.mk + +# Rule to create git_version.cpp +$(GIT_VERSION_FILE): + @mkdir -p ${DST_DIR}/src + @echo 'const char* rcclTestsGitHash = "$(GIT_BRANCH):$(GIT_REV)$(GIT_DIFF)";' > $@ + +${HIPIFY_DIR}/%.cpp: %.cpp + @printf "Hipifying %-35s > %s\n" $< $@ + @mkdir -p ${HIPIFY_DIR} + hipify-perl -quiet-warnings $< > $@ + +${HIPIFY_DIR}/%.h: %.h + @printf "Hipifying %-35s > %s\n" $< $@ + @mkdir -p ${HIPIFY_DIR} + hipify-perl -quiet-warnings $< > $@ + +${DST_DIR}/%.o: ${HIPIFY_DIR}/%.cpp ${HIPIFY_DIR}/common.h $(TEST_VERIFIABLE_HDRS) $(GIT_VERSION_FILE) + @printf "Compiling %-35s > %s\n" $< $@ + @mkdir -p ${DST_DIR} + echo "$(HIPCC) $(HIPCUFLAGS) -I. -c -o $@ $<" + $(HIPCC) $(HIPCUFLAGS) -I. -c -o $@ $< + +${DST_DIR}/timer.o: timer.cc timer.h + @printf "Compiling %-35s > %s\n" $< $@ + @mkdir -p ${DST_DIR} + $(CXX) $(CXXFLAGS) -o $@ -c timer.cc + +${DST_DIR}/%_perf:${DST_DIR}/%.o ${DST_DIR}/common.o ${DST_DIR}/timer.o $(TEST_VERIFIABLE_OBJS) $(DST_DIR)/src/git_version.cpp + @printf "Linking %-35s > %s\n" $< $@ + @mkdir -p ${DST_DIR} + echo "$(HIPCC) -o $@ $^ $(HIPLDFLAGS)" + $(HIPCC) -o $@ $^ $(HIPLDFLAGS) + diff --git a/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/all_gather.cpp b/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/all_gather.cpp new file mode 100644 index 0000000000..a8cc9a1614 --- /dev/null +++ b/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/all_gather.cpp @@ -0,0 +1,110 @@ +/************************************************************************* + * Copyright (c) 2016-2022, NVIDIA CORPORATION. All rights reserved. + * Modifications Copyright (c) 2019-2022 Advanced Micro Devices, Inc. All rights reserved. + * + * See LICENSE.txt for license information + ************************************************************************/ + +#include "common.h" +#include "cuda_runtime.h" + +void +AllGatherGetCollByteCount(size_t* sendcount, size_t* recvcount, size_t* paramcount, + size_t* sendInplaceOffset, size_t* recvInplaceOffset, + size_t count, size_t eltSize, int nranks) +{ + size_t base = (count / nranks) & -(16 / eltSize); + *sendcount = base; + *recvcount = base * nranks; + *sendInplaceOffset = base; + *recvInplaceOffset = 0; + *paramcount = base; +} + +testResult_t +AllGatherInitData(struct threadArgs* args, ncclDataType_t type, ncclRedOp_t op, int root, + int rep, int in_place) +{ + size_t sendcount = args->sendBytes / wordSize(type); + size_t recvcount = args->expectedBytes / wordSize(type); + int nranks = args->nProcs * args->nThreads * args->nGpus; + + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK(cudaSetDevice(args->gpus[i])); + int rank = ((args->proc * args->nThreads + args->thread) * args->nGpus + i); + CUDACHECK(cudaMemset(args->recvbuffs[i], 0, args->expectedBytes)); + void* data = in_place ? ((char*) args->recvbuffs[i]) + rank * args->sendBytes + : args->sendbuffs[i]; + TESTCHECK(InitData(data, sendcount, 0, type, ncclSum, 33 * rep + rank, 1, 0)); + for(int j = 0; j < nranks; j++) + { + TESTCHECK(InitData((char*) args->expected[i] + args->sendBytes * j, sendcount, + 0, type, ncclSum, 33 * rep + j, 1, 0)); + } + CUDACHECK(cudaDeviceSynchronize()); + } + return testSuccess; +} + +void +AllGatherGetBw(size_t count, int typesize, double sec, double* algBw, double* busBw, + int nranks) +{ + double baseBw = (double) (count * typesize * nranks) / 1.0E9 / sec; + + *algBw = baseBw; + double factor = ((double) (nranks - 1)) / ((double) nranks); + *busBw = baseBw * factor; +} + +testResult_t +AllGatherRunColl(void* sendbuff, void* recvbuff, size_t count, ncclDataType_t type, + ncclRedOp_t op, int root, ncclComm_t comm, cudaStream_t stream) +{ + NCCLCHECK(ncclAllGather(sendbuff, recvbuff, count, type, comm, stream)); + return testSuccess; +} + +struct testColl allGatherTest = { "AllGather", AllGatherGetCollByteCount, + AllGatherInitData, AllGatherGetBw, AllGatherRunColl }; + +void +AllGatherGetBuffSize(size_t* sendcount, size_t* recvcount, size_t count, int nranks) +{ + size_t paramcount, sendInplaceOffset, recvInplaceOffset; + AllGatherGetCollByteCount(sendcount, recvcount, ¶mcount, &sendInplaceOffset, + &recvInplaceOffset, count, /*eltSize=*/1, nranks); +} + +testResult_t +AllGatherRunTest(struct threadArgs* args, int root, ncclDataType_t type, + const char* typeName, ncclRedOp_t op, const char* opName) +{ + args->collTest = &allGatherTest; + ncclDataType_t* run_types; + const char** run_typenames; + int type_count; + + if((int) type != -1) + { + type_count = 1; + run_types = &type; + run_typenames = &typeName; + } + else + { + type_count = test_typenum; + run_types = test_types; + run_typenames = test_typenames; + } + + for(int i = 0; i < type_count; i++) + { + TESTCHECK( + TimeTest(args, run_types[i], run_typenames[i], (ncclRedOp_t) 0, "none", -1)); + } + return testSuccess; +} + +struct testEngine ncclTestEngine = { AllGatherGetBuffSize, AllGatherRunTest }; diff --git a/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/all_reduce.cpp b/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/all_reduce.cpp new file mode 100644 index 0000000000..551d2eb8b6 --- /dev/null +++ b/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/all_reduce.cpp @@ -0,0 +1,126 @@ +/************************************************************************* + * Copyright (c) 2016-2022, NVIDIA CORPORATION. All rights reserved. + * Modifications Copyright (c) 2019-2022 Advanced Micro Devices, Inc. All rights reserved. + * + * See LICENSE.txt for license information + ************************************************************************/ + +#include "common.h" +#include "cuda_runtime.h" + +void +AllReduceGetCollByteCount(size_t* sendcount, size_t* recvcount, size_t* paramcount, + size_t* sendInplaceOffset, size_t* recvInplaceOffset, + size_t count, size_t eltSize, int nranks) +{ + *sendcount = count; + *recvcount = count; + *sendInplaceOffset = 0; + *recvInplaceOffset = 0; + *paramcount = *sendcount; +} + +testResult_t +AllReduceInitData(struct threadArgs* args, ncclDataType_t type, ncclRedOp_t op, int root, + int rep, int in_place) +{ + size_t sendcount = args->sendBytes / wordSize(type); + size_t recvcount = args->expectedBytes / wordSize(type); + int nranks = args->nProcs * args->nThreads * args->nGpus; + + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK(cudaSetDevice(args->gpus[i])); + int rank = ((args->proc * args->nThreads + args->thread) * args->nGpus + i); + CUDACHECK(cudaMemset(args->recvbuffs[i], 0, args->expectedBytes)); + void* data = in_place ? args->recvbuffs[i] : args->sendbuffs[i]; + TESTCHECK(InitData(data, sendcount, 0, type, op, rep, nranks, rank)); + TESTCHECK(InitDataReduce(args->expected[i], recvcount, 0, type, op, rep, nranks)); + CUDACHECK(cudaDeviceSynchronize()); + } + return testSuccess; +} + +void +AllReduceGetBw(size_t count, int typesize, double sec, double* algBw, double* busBw, + int nranks) +{ + double baseBw = (double) (count * typesize) / 1.0E9 / sec; + + *algBw = baseBw; + double factor = ((double) (2 * (nranks - 1))) / ((double) nranks); + *busBw = baseBw * factor; +} + +testResult_t +AllReduceRunColl(void* sendbuff, void* recvbuff, size_t count, ncclDataType_t type, + ncclRedOp_t op, int root, ncclComm_t comm, cudaStream_t stream) +{ + NCCLCHECK(ncclAllReduce(sendbuff, recvbuff, count, type, op, comm, stream)); + return testSuccess; +} + +struct testColl allReduceTest = { "AllReduce", AllReduceGetCollByteCount, + AllReduceInitData, AllReduceGetBw, AllReduceRunColl }; + +void +AllReduceGetBuffSize(size_t* sendcount, size_t* recvcount, size_t count, int nranks) +{ + size_t paramcount, sendInplaceOffset, recvInplaceOffset; + AllReduceGetCollByteCount(sendcount, recvcount, ¶mcount, &sendInplaceOffset, + &recvInplaceOffset, count, /*eltSize=*/1, nranks); +} + +testResult_t +AllReduceRunTest(struct threadArgs* args, int root, ncclDataType_t type, + const char* typeName, ncclRedOp_t op, const char* opName) +{ + args->collTest = &allReduceTest; + ncclDataType_t* run_types; + ncclRedOp_t* run_ops; + const char ** run_typenames, **run_opnames; + int type_count, op_count; + + if((int) type != -1) + { + type_count = 1; + run_types = &type; + run_typenames = &typeName; + } + else + { + type_count = test_typenum; + run_types = test_types; + run_typenames = test_typenames; + } + + if((int) op != -1) + { + op_count = 1; + run_ops = &op; + run_opnames = &opName; + } + else + { + op_count = test_opnum; + run_ops = test_ops; + run_opnames = test_opnames; + } + + for(int i = 0; i < type_count; i++) + { + for(int j = 0; j < op_count; j++) + { +#if defined(RCCL_FLOAT8) + if((run_types[i] == ncclFp8E4M3 || run_types[i] == ncclFp8E5M2) && + run_ops[j] == ncclProd) + continue; +#endif + TESTCHECK(TimeTest(args, run_types[i], run_typenames[i], run_ops[j], + run_opnames[j], -1)); + } + } + return testSuccess; +} + +struct testEngine ncclTestEngine = { AllReduceGetBuffSize, AllReduceRunTest }; diff --git a/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/alltoall.cpp b/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/alltoall.cpp new file mode 100644 index 0000000000..37c9d19052 --- /dev/null +++ b/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/alltoall.cpp @@ -0,0 +1,112 @@ +/************************************************************************* + * Copyright (c) 2016-2022, NVIDIA CORPORATION. All rights reserved. + * Modifications Copyright (c) 2019-2022 Advanced Micro Devices, Inc. All rights reserved. + * + * See LICENSE.txt for license information + ************************************************************************/ + +#include "common.h" +#include "cuda_runtime.h" + +void +AlltoAllGetCollByteCount(size_t* sendcount, size_t* recvcount, size_t* paramcount, + size_t* sendInplaceOffset, size_t* recvInplaceOffset, + size_t count, size_t eltSize, int nranks) +{ + *paramcount = (count / nranks) & -(16 / eltSize); + *sendcount = nranks * (*paramcount); + *recvcount = *sendcount; + *sendInplaceOffset = 0; + *recvInplaceOffset = 0; +} + +testResult_t +AlltoAllInitData(struct threadArgs* args, ncclDataType_t type, ncclRedOp_t op, int root, + int rep, int in_place) +{ + size_t sendcount = args->sendBytes / wordSize(type); + size_t recvcount = args->expectedBytes / wordSize(type); + int nranks = args->nProcs * args->nThreads * args->nGpus; + + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK(cudaSetDevice(args->gpus[i])); + int rank = ((args->proc * args->nThreads + args->thread) * args->nGpus + i); + CUDACHECK(cudaMemset(args->recvbuffs[i], 0, args->expectedBytes)); + void* data = in_place ? args->recvbuffs[i] : args->sendbuffs[i]; + TESTCHECK(InitData(data, sendcount, 0, type, ncclSum, 33 * rep + rank, 1, 0)); + for(int j = 0; j < nranks; j++) + { + size_t partcount = sendcount / nranks; + TESTCHECK(InitData((char*) args->expected[i] + j * partcount * wordSize(type), + partcount, rank * partcount, type, ncclSum, 33 * rep + j, + 1, 0)); + } + CUDACHECK(cudaDeviceSynchronize()); + } + // We don't support in-place alltoall + args->reportErrors = in_place ? 0 : 1; + return testSuccess; +} + +void +AlltoAllGetBw(size_t count, int typesize, double sec, double* algBw, double* busBw, + int nranks) +{ + double baseBw = (double) (count * nranks * typesize) / 1.0E9 / sec; + + *algBw = baseBw; + double factor = ((double) (nranks - 1)) / ((double) (nranks)); + *busBw = baseBw * factor; +} + +testResult_t +AlltoAllRunColl(void* sendbuff, void* recvbuff, size_t count, ncclDataType_t type, + ncclRedOp_t op, int root, ncclComm_t comm, cudaStream_t stream) +{ + NCCLCHECK(ncclAllToAll(sendbuff, recvbuff, count, type, comm, stream)); + return testSuccess; +} + +struct testColl alltoAllTest = { "AlltoAll", AlltoAllGetCollByteCount, AlltoAllInitData, + AlltoAllGetBw, AlltoAllRunColl }; + +void +AlltoAllGetBuffSize(size_t* sendcount, size_t* recvcount, size_t count, int nranks) +{ + size_t paramcount, sendInplaceOffset, recvInplaceOffset; + AlltoAllGetCollByteCount(sendcount, recvcount, ¶mcount, &sendInplaceOffset, + &recvInplaceOffset, count, /*eltSize=*/1, nranks); +} + +testResult_t +AlltoAllRunTest(struct threadArgs* args, int root, ncclDataType_t type, + const char* typeName, ncclRedOp_t op, const char* opName) +{ + args->collTest = &alltoAllTest; + ncclDataType_t* run_types; + const char** run_typenames; + int type_count; + + if((int) type != -1) + { + type_count = 1; + run_types = &type; + run_typenames = &typeName; + } + else + { + type_count = test_typenum; + run_types = test_types; + run_typenames = test_typenames; + } + + for(int i = 0; i < type_count; i++) + { + TESTCHECK( + TimeTest(args, run_types[i], run_typenames[i], (ncclRedOp_t) 0, "none", -1)); + } + return testSuccess; +} + +struct testEngine ncclTestEngine = { AlltoAllGetBuffSize, AlltoAllRunTest }; diff --git a/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/alltoallv.cpp b/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/alltoallv.cpp new file mode 100644 index 0000000000..78c90775b9 --- /dev/null +++ b/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/alltoallv.cpp @@ -0,0 +1,218 @@ +/************************************************************************* + * Copyright (c) 2016-2020, NVIDIA CORPORATION. All rights reserved. + * Modifications Copyright (c) 2019 Advanced Micro Devices, Inc. All rights reserved. + * + * See LICENSE.txt for license information + ************************************************************************/ + +#include "common.h" +#include "cuda_runtime.h" + +#define USE_RCCL_GATHER_SCATTER + +void +AlltoAllvGetCollByteCount(size_t* sendcount, size_t* recvcount, size_t* paramcount, + size_t* sendInplaceOffset, size_t* recvInplaceOffset, + size_t count, size_t eltSize, int nranks) +{ + if(count < nranks * nranks / 2) + { + *sendcount = 0; + *recvcount = 0; + *sendInplaceOffset = 0; + *recvInplaceOffset = 0; + *paramcount = 0; + } + else + { + *paramcount = (count / nranks) & -(16 / eltSize); + *sendcount = nranks * (*paramcount); + *recvcount = *sendcount; + *sendInplaceOffset = 0; + *recvInplaceOffset = 0; + } +} + +testResult_t +AlltoAllvInitData(struct threadArgs* args, ncclDataType_t type, ncclRedOp_t op, int root, + int rep, int in_place) +{ + size_t sendcount = args->sendBytes / wordSize(type); + size_t recvcount = args->expectedBytes / wordSize(type); + int nranks = args->nProcs * args->nThreads * args->nGpus; + + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK(cudaSetDevice(args->gpus[i])); + int rank = ((args->proc * args->nThreads + args->thread) * args->nGpus + i); + CUDACHECK(cudaMemset(args->recvbuffs[i], 0, args->expectedBytes)); + void* data = in_place ? args->recvbuffs[i] : args->sendbuffs[i]; + TESTCHECK(InitData(data, sendcount, 0, type, ncclSum, 33 * rep + rank, 1, 0)); + +#if 0 + int *dataHost = (int *)malloc(args->sendBytes); + cudaMemcpy(dataHost, data, args->sendBytes, cudaMemcpyDeviceToHost); + printf(" Rank [%d] Original: ", rank); + for(int j=0; jexpected[i]) + rdisp * wordSize(type), + rcount, sdisp, type, ncclSum, 33 * rep + j, 1, 0)); + rdisp += rcount; + } + CUDACHECK(cudaDeviceSynchronize()); + } + // We don't support in-place alltoall + args->reportErrors = in_place ? 0 : 1; + return testSuccess; +} + +void +AlltoAllvGetBw(size_t count, int typesize, double sec, double* algBw, double* busBw, + int nranks) +{ + double baseBw = (double) (count * nranks * typesize) / 1.0E9 / sec; + + *algBw = baseBw; + double factor = ((double) (nranks - 1)) / ((double) (nranks)); + *busBw = baseBw * factor; +} + +testResult_t +AlltoAllvRunColl(void* sendbuff, void* recvbuff, size_t count, ncclDataType_t type, + ncclRedOp_t op, int root, ncclComm_t comm, cudaStream_t stream) +{ + int nranks; + NCCLCHECK(ncclCommCount(comm, &nranks)); + int rank; + NCCLCHECK(ncclCommUserRank(comm, &rank)); + + if(count == 0) return testSuccess; + + size_t *sendcounts, *recvcounts, *sdispls, *rdispls; + sendcounts = (size_t*) malloc(nranks * nranks * sizeof(size_t)); + recvcounts = (size_t*) malloc(nranks * nranks * sizeof(size_t)); + sdispls = (size_t*) malloc(nranks * nranks * sizeof(size_t)); + rdispls = (size_t*) malloc(nranks * nranks * sizeof(size_t)); + if(sendcounts == nullptr || recvcounts == nullptr || sdispls == nullptr || + rdispls == nullptr) + { + printf("failed to allocate buffers for alltoallv\n"); + return testNcclError; + } + + size_t disp = 0; + size_t chunksize = count * 2 / nranks; + for(int i = 0; i < nranks; i++) + { + size_t scount = ((i + rank) % nranks) * chunksize; + if((i + rank) % nranks == 0) + scount += (count * nranks - chunksize * (nranks - 1) * nranks / 2); + sendcounts[i + rank * nranks] = recvcounts[i + rank * nranks] = scount; + sdispls[i + rank * nranks] = rdispls[i + rank * nranks] = disp; + disp += scount; + // printf("%d->%d: sendcounts/recvcounts %lx sdispls/rdispls %lx\n", rank, i, + // sendcounts[i+rank*nranks]*wordSize(type), + // sdispls[i+rank*nranks]*wordSize(type)); + } + +#if NCCL_MAJOR < 2 || NCCL_MINOR < 7 + printf( + "NCCL 2.7 or later is needed for alltoallv. This test was compiled with %d.%d.\n", + NCCL_MAJOR, NCCL_MINOR); + return testNcclError; +#else +# if defined(RCCL_ALLTOALLV) && defined(USE_RCCL_GATHER_SCATTER) + NCCLCHECK(ncclAllToAllv(sendbuff, sendcounts + rank * nranks, sdispls + rank * nranks, + recvbuff, recvcounts + rank * nranks, rdispls + rank * nranks, + type, comm, stream)); +# else + NCCLCHECK(ncclGroupStart()); + for(int r = 0; r < nranks; r++) + { + if(sendcounts[r + rank * nranks] != 0) + { + NCCLCHECK( + ncclSend(((char*) sendbuff) + sdispls[r + rank * nranks] * wordSize(type), + sendcounts[r + rank * nranks], type, r, comm, stream)); + } + if(recvcounts[r + rank * nranks] != 0) + { + NCCLCHECK( + ncclRecv(((char*) recvbuff) + rdispls[r + rank * nranks] * wordSize(type), + recvcounts[r + rank * nranks], type, r, comm, stream)); + } + } + NCCLCHECK(ncclGroupEnd()); +# endif +#endif + free(sendcounts); + free(recvcounts); + free(sdispls); + free(rdispls); + return testSuccess; +} + +struct testColl alltoAllTest = { "AlltoAllv", AlltoAllvGetCollByteCount, + AlltoAllvInitData, AlltoAllvGetBw, AlltoAllvRunColl }; + +void +AlltoAllvGetBuffSize(size_t* sendcount, size_t* recvcount, size_t count, int nranks) +{ + size_t paramcount, sendInplaceOffset, recvInplaceOffset; + AlltoAllvGetCollByteCount(sendcount, recvcount, ¶mcount, &sendInplaceOffset, + &recvInplaceOffset, count, /*eltSize=*/1, nranks); +} + +testResult_t +AlltoAllvRunTest(struct threadArgs* args, int root, ncclDataType_t type, + const char* typeName, ncclRedOp_t op, const char* opName) +{ + args->collTest = &alltoAllTest; + ncclDataType_t* run_types; + const char** run_typenames; + int type_count; + + if((int) type != -1) + { + type_count = 1; + run_types = &type; + run_typenames = &typeName; + } + else + { + type_count = ncclNumTypes; + run_types = test_types; + run_typenames = test_typenames; + } + + for(int i = 0; i < type_count; i++) + { + TESTCHECK( + TimeTest(args, run_types[i], run_typenames[i], (ncclRedOp_t) 0, "", -1)); + } + return testSuccess; +} + +struct testEngine ncclTestEngine = { AlltoAllvGetBuffSize, AlltoAllvRunTest }; diff --git a/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/broadcast.cpp b/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/broadcast.cpp new file mode 100644 index 0000000000..b98c3a8fd7 --- /dev/null +++ b/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/broadcast.cpp @@ -0,0 +1,131 @@ +/************************************************************************* + * Copyright (c) 2015-2022, NVIDIA CORPORATION. All rights reserved. + * Modifications Copyright (c) 2019-2022 Advanced Micro Devices, Inc. All rights reserved. + * + * See LICENSE.txt for license information + ************************************************************************/ + +#include "common.h" +#include "cuda_runtime.h" + +void +BroadcastGetCollByteCount(size_t* sendcount, size_t* recvcount, size_t* paramcount, + size_t* sendInplaceOffset, size_t* recvInplaceOffset, + size_t count, size_t eltSize, int nranks) +{ + *sendcount = count; + *recvcount = count; + *sendInplaceOffset = 0; + *recvInplaceOffset = 0; + *paramcount = *sendcount; +} + +testResult_t +BroadcastInitData(struct threadArgs* args, ncclDataType_t type, ncclRedOp_t op, int root, + int rep, int in_place) +{ + size_t sendcount = args->sendBytes / wordSize(type); + size_t recvcount = args->expectedBytes / wordSize(type); + + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK(cudaSetDevice(args->gpus[i])); + int rank = ((args->proc * args->nThreads + args->thread) * args->nGpus + i); + CUDACHECK(cudaMemset(args->recvbuffs[i], 0, args->expectedBytes)); + void* data = in_place ? args->recvbuffs[i] : args->sendbuffs[i]; + if(rank == root) + TESTCHECK(InitData(data, sendcount, 0, type, ncclSum, rep, 1, 0)); + TESTCHECK(InitData(args->expected[i], recvcount, 0, type, ncclSum, rep, 1, 0)); + CUDACHECK(cudaDeviceSynchronize()); + } + return testSuccess; +} + +void +BroadcastGetBw(size_t count, int typesize, double sec, double* algBw, double* busBw, + int nranks) +{ + double baseBw = (double) (count * typesize) / 1.0E9 / sec; + + *algBw = baseBw; + double factor = 1; + *busBw = baseBw * factor; +} + +testResult_t +BroadcastRunColl(void* sendbuff, void* recvbuff, size_t count, ncclDataType_t type, + ncclRedOp_t op, int root, ncclComm_t comm, cudaStream_t stream) +{ + int rank; + NCCLCHECK(ncclCommUserRank(comm, &rank)); +#if NCCL_MAJOR >= 2 && NCCL_MINOR >= 2 + NCCLCHECK(ncclBroadcast(sendbuff, recvbuff, count, type, root, comm, stream)); +#else + if(rank == root) + { + NCCLCHECK(ncclBcast(sendbuff, count, type, root, comm, stream)); + } + else + { + NCCLCHECK(ncclBcast(recvbuff, count, type, root, comm, stream)); + } +#endif + return testSuccess; +} + +struct testColl broadcastTest = { "Broadcast", BroadcastGetCollByteCount, + BroadcastInitData, BroadcastGetBw, BroadcastRunColl }; + +void +BroadcastGetBuffSize(size_t* sendcount, size_t* recvcount, size_t count, int nranks) +{ + size_t paramcount, sendInplaceOffset, recvInplaceOffset; + BroadcastGetCollByteCount(sendcount, recvcount, ¶mcount, &sendInplaceOffset, + &recvInplaceOffset, count, /*eltSize=*/1, nranks); +} + +testResult_t +BroadcastRunTest(struct threadArgs* args, int root, ncclDataType_t type, + const char* typeName, ncclRedOp_t op, const char* opName) +{ + args->collTest = &broadcastTest; + ncclDataType_t* run_types; + const char** run_typenames; + int type_count; + int begin_root, end_root; + + if((int) type != -1) + { + type_count = 1; + run_types = &type; + run_typenames = &typeName; + } + else + { + type_count = test_typenum; + run_types = test_types; + run_typenames = test_typenames; + } + + if(root != -1) + { + begin_root = end_root = root; + } + else + { + begin_root = 0; + end_root = args->nProcs * args->nThreads * args->nGpus - 1; + } + + for(int i = 0; i < type_count; i++) + { + for(int j = begin_root; j <= end_root; j++) + { + TESTCHECK(TimeTest(args, run_types[i], run_typenames[i], (ncclRedOp_t) 0, + "none", j)); + } + } + return testSuccess; +} + +struct testEngine ncclTestEngine = { BroadcastGetBuffSize, BroadcastRunTest }; diff --git a/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/common.cpp b/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/common.cpp new file mode 100644 index 0000000000..1e7658b975 --- /dev/null +++ b/projects/rocprofiler-systems/examples/rccl/rccl-tests/src/common.cpp @@ -0,0 +1,1987 @@ + +/************************************************************************* + * Copyright (c) 2016-2022, NVIDIA CORPORATION. All rights reserved. + * Modifications Copyright (c) 2019-2022 Advanced Micro Devices, Inc. All rights reserved. + * Modifications Copyright (c) Microsoft Corporation. Licensed under the MIT License. + * + * See LICENSE.txt for license information + ************************************************************************/ + +#include "common.h" +#include "cuda.h" +#include "cuda_runtime.h" +#include "rccl_float8.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +// #define DEBUG_PRINT + +#include "git_version.h" +#include "verifiable.h" + +int test_ncclVersion = 0; // init'd with ncclGetVersion() +int32_t gpu_block3; +size_t cache_bytes = 192 * 1024 * 1024; // Use 192MB + +#if NCCL_MAJOR >= 2 +ncclDataType_t test_types[ncclNumTypes] = { ncclInt8, + ncclUint8, + ncclInt32, + ncclUint32, + ncclInt64, + ncclUint64, + ncclHalf, + ncclFloat, + ncclDouble +# if RCCL_BFLOAT16 == 1 + , + ncclBfloat16 +# endif +# if RCCL_FLOAT8 == 1 + , + ncclFp8E4M3, + ncclFp8E5M2 +# endif +}; +const char* test_typenames[ncclNumTypes] = { "int8", + "uint8", + "int32", + "uint32", + "int64", + "uint64", + "half", + "float", + "double" +# if RCCL_BFLOAT16 == 1 + , + "bfloat16" +# endif +# if RCCL_FLOAT8 == 1 + , + "fp8_e4m3", + "fp8_e5m2" +# endif +}; +int test_typenum = -1; + +const char* test_opnames[] = { "sum", "prod", "max", "min", "avg", "mulsum" }; +ncclRedOp_t test_ops[] = { + ncclSum, + ncclProd, + ncclMax, + ncclMin +# if NCCL_VERSION_CODE >= NCCL_VERSION(2, 10, 0) + , + ncclAvg +# endif +# if NCCL_VERSION_CODE >= NCCL_VERSION(2, 11, 0) + , + ncclNumOps // stand in for ncclRedOpCreatePreMulSum() created on-demand +# endif +}; +int test_opnum = -1; +#else +ncclDataType_t test_types[ncclNumTypes] = { ncclChar, ncclInt, ncclHalf, ncclFloat, + ncclDouble, ncclInt64, ncclUint64 }; +const char* test_typenames[ncclNumTypes] = { "char", "int", "half", "float", + "double", "int64", "uint64" }; +int test_typenum = 7; +const char* test_opnames[] = { "sum", "prod", "max", "min" }; +ncclRedOp_t test_ops[] = { ncclSum, ncclProd, ncclMax, ncclMin }; +int test_opnum = 4; +#endif + +const char* test_memorytypes[nccl_NUM_MTYPES] = { "coarse", "fine", "host", "managed" }; + +// For libnccl's < 2.13 +extern "C" __attribute__((weak)) char const* +ncclGetLastError(ncclComm_t comm) +{ + return ""; +} + +int is_main_proc = 0; +thread_local int is_main_thread = 0; + +// Command line parameter defaults +static int nThreads = 1; +static int nGpus = 1; +static size_t minBytes = 32 * 1024 * 1024; +static size_t maxBytes = 32 * 1024 * 1024; +static size_t stepBytes = 1 * 1024 * 1024; +static size_t stepFactor = 1; +static int datacheck = 1; +static int warmup_iters = 5; +static int iters = 20; +static int agg_iters = 1; +static int run_cycles = 1; +static int ncclop = ncclSum; +static int nccltype = ncclFloat; +static int ncclroot = 0; +static int parallel_init = 0; +static int blocking_coll = 0; +static int memorytype = 0; +static uint32_t cumask[4]; +static int streamnull = 0; +static int timeout = 0; +static int cudaGraphLaunches = 0; +std::string output_file; +std::string output_format; +static int report_cputime = 0; +// Report average iteration time: (0=RANK0,1=AVG,2=MIN,3=MAX) +static int average = 1; +static int numDevices = 1; +static int delay_inout_place = 0; +static int enable_out_of_place = 1; +static int enable_in_place = 1; +static int enable_cache_flush = 0; +static int enable_rotating_tensor = 0; +#if NCCL_VERSION_CODE >= NCCL_VERSION(2, 19, 0) +static int local_register = 0; +#endif + +Reporter::Reporter(std::string fileName, std::string outputFormat) +: _outputFormat(outputFormat) +{ + if(!fileName.empty()) + { + if(isMainThread()) + { + _out = std::ofstream(fileName, std::ios_base::out); + _outputValid = true; + if(_outputFormat == "csv") + { + _out << "numCycle, "; + _out << "collective, "; +#ifdef MPI_SUPPORT + _out << "ranks, rankspernode, gpusperrank, "; +#else + _out << "gpus, "; +#endif + _out << "size, type, redop, inplace, time, algbw, busbw, #wrong\n"; + } + } + } +} + +void +Reporter::setParameters(const size_t numCycle, const char* name, const char* typeName, + const char* opName) +{ + if(!isMainThread() || !_outputValid) return; + + _numCycle = numCycle; + _collectiveName = name; + _typeName = typeName; + _opName = opName; +} + +void +Reporter::addResult(int gpusPerRank, int ranksPerNode, int totalRanks, size_t numBytes, + int inPlace, double timeUsec, double algBw, double busBw, + int64_t wrongElts) +{ + if(!isMainThread() || !_outputValid) return; + + std::vector> outputValuesKeys; + std::string wrongEltsStr = (wrongElts == -1) ? "N/A" : std::to_string(wrongElts); + int nodes = totalRanks / ranksPerNode; + + outputValuesKeys.push_back(makeValueKeyPair(_numCycle, "numCycle")); + outputValuesKeys.push_back(makeValueKeyPair(_collectiveName, "name")); +#ifdef MPI_SUPPORT + outputValuesKeys.push_back(makeValueKeyPair(nodes, "nodes")); + outputValuesKeys.push_back(makeValueKeyPair(totalRanks, "ranks")); + outputValuesKeys.push_back(makeValueKeyPair(ranksPerNode, "ranksPerNode")); + outputValuesKeys.push_back(makeValueKeyPair(gpusPerRank, "gpusPerRank")); +#else + outputValuesKeys.push_back(makeValueKeyPair(gpusPerRank, "gpus")); +#endif + outputValuesKeys.push_back(makeValueKeyPair(numBytes, "size")); + outputValuesKeys.push_back(makeValueKeyPair(_typeName, "type")); + outputValuesKeys.push_back(makeValueKeyPair(_opName, "redop")); + outputValuesKeys.push_back(makeValueKeyPair(inPlace, "inPlace")); + outputValuesKeys.push_back(makeValueKeyPair(timeUsec, "time")); + outputValuesKeys.push_back(makeValueKeyPair(algBw, "algBw")); + outputValuesKeys.push_back(makeValueKeyPair(busBw, "busBw")); + outputValuesKeys.push_back(makeValueKeyPair(wrongEltsStr, "wrong")); + + for(auto iter = outputValuesKeys.begin(); iter != outputValuesKeys.end(); ++iter) + { + if(_outputFormat == "csv") + { + _out << iter->first; + if(std::next(iter) != outputValuesKeys.end()) + { + _out << ", "; + } + } + else + { // json + if(iter == outputValuesKeys.begin()) + { + _out << "{"; + } + _out << "\"" << iter->second << "\":" << iter->first; + if(std::next(iter) != outputValuesKeys.end()) + { + _out << ", "; + } + else + { + _out << "}"; + } + } + } + _out << std::endl; +} + +bool +Reporter::isMainThread() +{ + return is_main_thread == 1; +} + +#define NUM_BLOCKS 32 + +#ifndef CHECK_HIP_ERROR +# define CHECK_HIP_ERROR(error) \ + if(error != hipSuccess) \ + { \ + fprintf(stderr, "Hip error: '%s'(%d) at %s:%d\n", hipGetErrorString(error), \ + error, __FILE__, __LINE__); \ + exit(EXIT_FAILURE); \ + } +#endif + +extern "C" __global__ void +flush_icache() +{ + asm __volatile__("s_icache_inv \n\t" + "s_nop 0 \n\t" + "s_nop 0 \n\t" + "s_nop 0 \n\t" + "s_nop 0 \n\t" + "s_nop 0 \n\t" + "s_nop 0 \n\t" + "s_nop 0 \n\t" + "s_nop 0 \n\t" + "s_nop 0 \n\t" + "s_nop 0 \n\t" + "s_nop 0 \n\t" + "s_nop 0 \n\t" + "s_nop 0 \n\t" + "s_nop 0 \n\t" + "s_nop 0 \n\t" + "s_nop 0 \n\t" :: + :); +} + +static double +parsesize(const char* value) +{ + long long int units; + double size; + char size_lit[2]; + + int count = sscanf(value, "%lf %1s", &size, size_lit); + + switch(count) + { + case 2: + switch(size_lit[0]) + { + case 'G': + case 'g': units = 1024 * 1024 * 1024; break; + case 'M': + case 'm': units = 1024 * 1024; break; + case 'K': + case 'k': units = 1024; break; + default: return -1.0; + }; + break; + case 1: units = 1; break; + default: return -1.0; + } + + return size * units; +} + +static bool +minReqVersion(int rmajor, int rminor, int rpatch) +{ + int version; + int major, minor, patch, rem; + ncclGetVersion(&version); + + if(version < 10000) + { + major = version / 1000; + rem = version % 1000; + minor = rem / 100; + patch = rem % 100; + } + else + { + major = version / 10000; + rem = version % 10000; + minor = rem / 100; + patch = rem % 100; + } + + if(major < rmajor) + return false; + else if(major > rmajor) + return true; + + // major == rmajor + if(minor < rminor) + return false; + else if(minor > rminor) + return true; + + // major == rmajor && minor == rminor + if(patch < rpatch) return false; + + return true; +} + +testResult_t +CheckDelta(void* results, void* expected, size_t count, size_t offset, + ncclDataType_t type, ncclRedOp_t op, uint64_t seed, int nranks, + int64_t* wrongEltN) +{ + ncclVerifiableVerify(results, expected, count, (int) type, (int) op, nranks, seed, + offset, wrongEltN, cudaStreamDefault); + CUDACHECK(cudaDeviceSynchronize()); + return testSuccess; +} + +testResult_t +InitDataReduce(void* data, const size_t count, const size_t offset, ncclDataType_t type, + ncclRedOp_t op, uint64_t seed, int nranks) +{ + ncclVerifiablePrepareExpected(data, count, (int) type, (int) op, nranks, seed, offset, + cudaStreamDefault); + return testSuccess; +} + +testResult_t +InitData(void* data, const size_t count, size_t offset, ncclDataType_t type, + ncclRedOp_t op, uint64_t seed, int nranks, int rank) +{ + ncclVerifiablePrepareInput(data, count, (int) type, (int) op, nranks, rank, seed, + offset, cudaStreamDefault); + return testSuccess; +} + +void +Barrier(struct threadArgs* args) +{ + thread_local int epoch = 0; + static pthread_mutex_t lock[2] = { PTHREAD_MUTEX_INITIALIZER, + PTHREAD_MUTEX_INITIALIZER }; + static pthread_cond_t cond[2] = { PTHREAD_COND_INITIALIZER, + PTHREAD_COND_INITIALIZER }; + static int counter[2] = { 0, 0 }; + + pthread_mutex_lock(&lock[epoch]); + if(++counter[epoch] == args->nThreads) pthread_cond_broadcast(&cond[epoch]); + + if(args->thread + 1 == args->nThreads) + { + while(counter[epoch] != args->nThreads) + pthread_cond_wait(&cond[epoch], &lock[epoch]); +#ifdef MPI_SUPPORT + MPI_Barrier(MPI_COMM_WORLD); +#endif + counter[epoch] = 0; + pthread_cond_broadcast(&cond[epoch]); + } + else + { + while(counter[epoch] != 0) + pthread_cond_wait(&cond[epoch], &lock[epoch]); + } + pthread_mutex_unlock(&lock[epoch]); + epoch ^= 1; +} + +// Inter-thread/process barrier+allreduce. The quality of the return value +// for average=0 (which means broadcast from rank=0) is dubious. The returned +// value will actually be the result of process-local broadcast from the local thread=0. +template +void +Allreduce(struct threadArgs* args, T* value, int average) +{ + thread_local int epoch = 0; + static pthread_mutex_t lock[2] = { PTHREAD_MUTEX_INITIALIZER, + PTHREAD_MUTEX_INITIALIZER }; + static pthread_cond_t cond[2] = { PTHREAD_COND_INITIALIZER, + PTHREAD_COND_INITIALIZER }; + static T accumulator[2]; + static int counter[2] = { 0, 0 }; + + pthread_mutex_lock(&lock[epoch]); + if(counter[epoch] == 0) + { + if(average != 0 || args->thread == 0) accumulator[epoch] = *value; + } + else + { + switch(average) + { + case /*r0*/ 0: + if(args->thread == 0) accumulator[epoch] = *value; + break; + case /*avg*/ 1: accumulator[epoch] += *value; break; + case /*min*/ 2: + accumulator[epoch] = std::min(accumulator[epoch], *value); + break; + case /*max*/ 3: + accumulator[epoch] = std::max(accumulator[epoch], *value); + break; + case /*sum*/ 4: accumulator[epoch] += *value; break; + } + } + + if(++counter[epoch] == args->nThreads) pthread_cond_broadcast(&cond[epoch]); + + if(args->thread + 1 == args->nThreads) + { + while(counter[epoch] != args->nThreads) + pthread_cond_wait(&cond[epoch], &lock[epoch]); + +#ifdef MPI_SUPPORT + if(average != 0) + { + static_assert(std::is_same::value || + std::is_same::value, + "Allreduce only for T in {long long, double}"); + MPI_Datatype ty = std::is_same::value ? MPI_LONG_LONG + : std::is_same::value ? MPI_DOUBLE + : MPI_Datatype(); + MPI_Op op = average == 1 ? MPI_SUM + : average == 2 ? MPI_MIN + : average == 3 ? MPI_MAX + : average == 4 ? MPI_SUM + : MPI_Op(); + MPI_Allreduce(MPI_IN_PLACE, (void*) &accumulator[epoch], 1, ty, op, + MPI_COMM_WORLD); + } +#endif + + if(average == 1) accumulator[epoch] /= args->totalProcs * args->nThreads; + counter[epoch] = 0; + pthread_cond_broadcast(&cond[epoch]); + } + else + { + while(counter[epoch] != 0) + pthread_cond_wait(&cond[epoch], &lock[epoch]); + } + pthread_mutex_unlock(&lock[epoch]); + + *value = accumulator[epoch]; + epoch ^= 1; +} + +testResult_t +CheckData(struct threadArgs* args, ncclDataType_t type, ncclRedOp_t op, int root, + int in_place, int64_t* wrongElts) +{ + int nranks = args->nProcs * args->nGpus * args->nThreads; + size_t count = args->expectedBytes / wordSize(type); + + int64_t* wrongPerGpu = nullptr; + CUDACHECK(hipHostMalloc((void**) &wrongPerGpu, args->nGpus * sizeof(int64_t), + cudaHostAllocMapped)); + + for(int i = 0; i < args->nGpus; i++) + { + int rank = ((args->proc * args->nThreads + args->thread) * args->nGpus + i); + CUDACHECK(cudaSetDevice(args->gpus[i])); + void* data = in_place ? ((void*) ((uintptr_t) args->recvbuffs[i] + + args->recvInplaceOffset * rank)) + : args->recvbuffs[i]; + + TESTCHECK(CheckDelta(data, args->expected[i], count, 0, type, op, 0, nranks, + wrongPerGpu + i)); + +#if 1 && DEBUG_PRINT + if(args->reportErrors && wrongPerGpu[i] != 0) + { + printf("rank=%d #wrong=%d\n", rank, (int) wrongPerGpu[i]); + char* expectedHost = (char*) malloc(args->expectedBytes); + char* dataHost = (char*) malloc(args->expectedBytes); + int eltsz = wordSize(type); + cudaMemcpy(expectedHost, args->expected[i], args->expectedBytes, + cudaMemcpyDeviceToHost); + cudaMemcpy(dataHost, data, args->expectedBytes, cudaMemcpyDeviceToHost); + + for(int j = 0; j < args->expectedBytes / eltsz; j++) + { + unsigned long long want, got; + want = 0; + memcpy(&want, expectedHost + j * eltsz, eltsz); + got = 0; + memcpy(&got, dataHost + j * eltsz, eltsz); + if(want != got) + { + printf(" rank=%d elt[%d]: want=0x%llx got=0x%llx\n", rank, j, want, + got); + } + } + free(expectedHost); + free(dataHost); + } +#endif + } + + *wrongElts = 0; + for(int i = 0; i < args->nGpus; i++) + *wrongElts += wrongPerGpu[i]; + cudaFreeHost(wrongPerGpu); + + if(args->reportErrors && *wrongElts) args->errors[0]++; + return testSuccess; +} + +testResult_t +testStreamSynchronize(int ngpus, cudaStream_t* streams, ncclComm_t* comms) +{ + cudaError_t cudaErr; + int remaining = ngpus; + int* done = (int*) malloc(sizeof(int) * ngpus); + memset(done, 0, sizeof(int) * ngpus); + timer tim; + + while(remaining) + { + int idle = 1; + for(int i = 0; i < ngpus; i++) + { + if(done[i]) continue; + + cudaErr = cudaStreamQuery(streams[i]); + if(cudaErr == cudaSuccess) + { + done[i] = 1; + remaining--; + idle = 0; + continue; + } + + if(cudaErr != cudaErrorNotReady) CUDACHECK(cudaErr); + +#if NCCL_VERSION_CODE >= NCCL_VERSION(2, 4, 0) + if(test_ncclVersion >= NCCL_VERSION(2, 4, 0) && comms) + { + ncclResult_t ncclAsyncErr; + NCCLCHECK(ncclCommGetAsyncError(comms[i], &ncclAsyncErr)); + if(ncclAsyncErr != ncclSuccess) + { + // An asynchronous error happened. Stop the operation and destroy + // the communicator + for(int i = 0; i < ngpus; i++) + NCCLCHECK(ncclCommAbort(comms[i])); + // Abort the perf test + NCCLCHECK(ncclAsyncErr); + } + } + double delta = tim.elapsed(); + if(delta > timeout && timeout > 0) + { + for(int i = 0; i < ngpus; i++) + NCCLCHECK(ncclCommAbort(comms[i])); + char hostname[1024]; + getHostName(hostname, 1024); + printf("%s: Test timeout (%ds) %s:%d\n", hostname, timeout, __FILE__, + __LINE__); + free(done); + return testTimeout; + } +#endif + } + + // We might want to let other threads (including NCCL threads) use the CPU. + if(idle) sched_yield(); + } + free(done); + return testSuccess; +} + +testResult_t +startColl(struct threadArgs* args, ncclDataType_t type, ncclRedOp_t opIndex, int root, + int in_place, int iter) +{ + size_t count = args->nbytes / wordSize(type); + + // Try to change offset for each iteration so that we avoid cache effects and catch + // race conditions in ptrExchange + size_t shift = 0; + if(enable_rotating_tensor) + { + shift = cache_bytes * (iter % 2); + } + else + { + size_t totalnbytes = std::max(args->sendBytes, args->expectedBytes); + size_t steps = totalnbytes ? args->maxbytes / totalnbytes : 1; + shift = totalnbytes * (iter % steps); + } + + if(args->nGpus > 1) NCCLCHECK(ncclGroupStart()); + for(int i = 0; i < args->nGpus; i++) + { +#ifndef NCCL_MAJOR + CUDACHECK(cudaSetDevice(args->gpus[i])); +#endif + int rank = ((args->proc * args->nThreads + args->thread) * args->nGpus + i); + char* recvBuff = ((char*) args->recvbuffs[i]) + shift; + char* sendBuff = ((char*) args->sendbuffs[i]) + shift; + ncclRedOp_t op; + + if(opIndex < ncclNumOps) + { + op = opIndex; + } +#if NCCL_VERSION_CODE >= NCCL_VERSION(2, 11, 0) + else + { + union + { + int8_t i8; + uint8_t u8; + int32_t i32; + uint32_t u32; + int64_t i64; + uint64_t u64; + half f16; + float f32; + double f64; +# if defined(RCCL_BFLOAT16) + hip_bfloat16 bf16; +# endif +# if defined(RCCL_FLOAT8) + rccl_float8 fp8_e4m3; + rccl_bfloat8 fp8_e5m2; +# endif + }; + switch(type) + { + case ncclInt8: i8 = ncclVerifiablePremulScalar(rank); break; + case ncclUint8: u8 = ncclVerifiablePremulScalar(rank); break; + case ncclInt32: i32 = ncclVerifiablePremulScalar(rank); break; + case ncclUint32: u32 = ncclVerifiablePremulScalar(rank); break; + case ncclInt64: i64 = ncclVerifiablePremulScalar(rank); break; + case ncclUint64: u64 = ncclVerifiablePremulScalar(rank); break; + case ncclFloat16: f16 = ncclVerifiablePremulScalar(rank); break; + case ncclFloat32: f32 = ncclVerifiablePremulScalar(rank); break; + case ncclFloat64: f64 = ncclVerifiablePremulScalar(rank); break; +# if defined(RCCL_BFLOAT16) + case ncclBfloat16: + bf16 = ncclVerifiablePremulScalar(rank); + break; +# endif +# if defined(RCCL_FLOAT8) + case ncclFp8E4M3: + fp8_e4m3 = ncclVerifiablePremulScalar(rank); + break; + case ncclFp8E5M2: + fp8_e5m2 = ncclVerifiablePremulScalar(rank); + break; +# endif + case ncclNumTypes: break; + } + NCCLCHECK(ncclRedOpCreatePreMulSum(&op, &u64, type, ncclScalarHostImmediate, + args->comms[i])); + } +#endif + + if(enable_cache_flush > 0 && ((iter % enable_cache_flush) == 0)) + { + hipLaunchKernelGGL(flush_icache, dim3(gpu_block3), dim3(64), 0, + args->streams[i]); + } + + TESTCHECK(args->collTest->runColl( + (void*) (in_place ? recvBuff + args->sendInplaceOffset * rank : sendBuff), + (void*) (in_place ? recvBuff + args->recvInplaceOffset * rank : recvBuff), + count, type, op, root, args->comms[i], args->streams[i])); + +#if NCCL_VERSION_CODE >= NCCL_VERSION(2, 11, 0) + if(opIndex >= ncclNumOps) + { + NCCLCHECK(ncclRedOpDestroy(op, args->comms[i])); + } +#endif + } + if(args->nGpus > 1) NCCLCHECK(ncclGroupEnd()); + + if(blocking_coll) + { + // Complete op before returning + TESTCHECK(testStreamSynchronize(args->nGpus, args->streams, args->comms)); + } + if(blocking_coll) Barrier(args); + return testSuccess; +} + +testResult_t +completeColl(struct threadArgs* args) +{ + if(blocking_coll) return testSuccess; + + TESTCHECK(testStreamSynchronize(args->nGpus, args->streams, args->comms)); + return testSuccess; +} + +testResult_t +BenchTime(struct threadArgs* args, ncclDataType_t type, ncclRedOp_t op, int root, + int in_place) +{ + size_t count = args->nbytes / wordSize(type); + if(datacheck) + { + // Initialize sendbuffs, recvbuffs and expected + TESTCHECK(args->collTest->initData(args, type, op, root, 99, in_place)); + } + + if(warmup_iters) + { + // Sync + TESTCHECK(startColl(args, type, op, root, in_place, 0)); + TESTCHECK(completeColl(args)); + } + + Barrier(args); + +#if HIP_VERSION >= 50221310 + std::vector graphs(args->nGpus); + std::vector graphExec(args->nGpus); + if(cudaGraphLaunches >= 1) + { + // Begin cuda graph capture + for(int i = 0; i < args->nGpus; i++) + { + // Thread local mdoe is needed for: + // - Multi-thread mode: where graph capture and instantiation can happen + // concurrently across threads + // - P2P pre-connect: when there is no warm-up, P2P pre-connect is done during + // graph capture. + // Since pre-connect calls cudaMalloc, we cannot use global capture mode + CUDACHECK(cudaStreamBeginCapture(args->streams[i], + cudaStreamCaptureModeThreadLocal)); + } + } +#endif + + // Performance Benchmark + timer tim; + for(int iter = 0; iter < iters; iter++) + { + if(agg_iters > 1) NCCLCHECK(ncclGroupStart()); + for(int aiter = 0; aiter < agg_iters; aiter++) + { + TESTCHECK( + startColl(args, type, op, root, in_place, iter * agg_iters + aiter)); + } + if(agg_iters > 1) NCCLCHECK(ncclGroupEnd()); + } + +#if HIP_VERSION >= 50221310 + if(cudaGraphLaunches >= 1) + { + // End cuda graph capture + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK(cudaStreamEndCapture(args->streams[i], graphs.data() + i)); + } + // Instantiate cuda graph + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK( + cudaGraphInstantiate(graphExec.data() + i, graphs[i], NULL, NULL, 0)); + } + // Resync CPU, restart timing, launch cuda graph + Barrier(args); + tim.reset(); + for(int l = 0; l < cudaGraphLaunches; l++) + { + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK(cudaGraphLaunch(graphExec[i], args->streams[i])); + } + } + } +#endif + + double cputimeSec = tim.elapsed() / (iters * agg_iters); + TESTCHECK(completeColl(args)); + + double deltaSec = tim.elapsed(); + deltaSec = deltaSec / (iters * agg_iters); + if(cudaGraphLaunches >= 1) deltaSec = deltaSec / cudaGraphLaunches; + Allreduce(args, &deltaSec, average); + +#if HIP_VERSION >= 50221310 + if(cudaGraphLaunches >= 1) + { + // destroy cuda graph + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK(cudaGraphExecDestroy(graphExec[i])); + CUDACHECK(cudaGraphDestroy(graphs[i])); + } + } +#endif + + double algBw, busBw; + args->collTest->getBw(count, wordSize(type), deltaSec, &algBw, &busBw, + args->nProcs * args->nThreads * args->nGpus); + + Barrier(args); + + int64_t wrongElts = 0; + static __thread int rep = 0; + rep++; + for(int c = 0; c < datacheck; c++) + { + // Initialize sendbuffs, recvbuffs and expected + TESTCHECK(args->collTest->initData(args, type, op, root, rep, in_place)); + +#if HIP_VERSION >= 50221310 + if(cudaGraphLaunches >= 1) + { + // Begin cuda graph capture for data check + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK(cudaStreamBeginCapture(args->streams[i], + args->nThreads > 1 + ? cudaStreamCaptureModeThreadLocal + : cudaStreamCaptureModeGlobal)); + } + } +#endif + + // test validation in single itertion, should ideally be included into the + // multi-iteration run + TESTCHECK(startColl(args, type, op, root, in_place, 0)); + +#if HIP_VERSION >= 50221310 + if(cudaGraphLaunches >= 1) + { + // End cuda graph capture + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK(cudaStreamEndCapture(args->streams[i], graphs.data() + i)); + } + // Instantiate cuda graph + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK( + cudaGraphInstantiate(graphExec.data() + i, graphs[i], NULL, NULL, 0)); + } + // Launch cuda graph + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK(cudaGraphLaunch(graphExec[i], args->streams[i])); + } + } +#endif + + TESTCHECK(completeColl(args)); + +#if HIP_VERSION >= 50221310 + if(cudaGraphLaunches >= 1) + { + // destroy cuda graph + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK(cudaGraphExecDestroy(graphExec[i])); + CUDACHECK(cudaGraphDestroy(graphs[i])); + } + } +#endif + + TESTCHECK(CheckData(args, type, op, root, in_place, &wrongElts)); + + // aggregate delta from all threads and procs + long long wrongElts1 = wrongElts; + // if (wrongElts) fprintf(stderr, "\nERROR: Data corruption : rank %d size %ld + // wrongElts %ld\n", args->proc, args->expectedBytes, wrongElts); + Allreduce(args, &wrongElts1, /*sum*/ 4); + wrongElts = wrongElts1; + if(wrongElts) break; + } + + double timeUsec = (report_cputime ? cputimeSec : deltaSec) * 1.0E6; + char timeStr[100]; + if(timeUsec >= 10000.0) + { + sprintf(timeStr, "%7.0f", timeUsec); + } + else if(timeUsec >= 100.0) + { + sprintf(timeStr, "%7.1f", timeUsec); + } + else + { + sprintf(timeStr, "%7.2f", timeUsec); + } + if(args->reportErrors) + { + PRINT(" %7s %6.2f %6.2f %5g", timeStr, algBw, busBw, (double) wrongElts); + } + else + { + PRINT(" %7s %6.2f %6.2f %5s", timeStr, algBw, busBw, "N/A"); + } + + auto largestMessageSize = std::max(args->sendBytes, args->expectedBytes); + if(args->reporter) + { + if(args->reportErrors) + { + args->reporter->addResult((args->nThreads * args->nGpus), args->nProcs, + args->totalProcs, largestMessageSize, in_place, + timeUsec, algBw, busBw, wrongElts); + } + else + { + args->reporter->addResult((args->nThreads * args->nGpus), args->nProcs, + args->totalProcs, largestMessageSize, in_place, + timeUsec, algBw, busBw); + } + } + + args->bw[0] += busBw; + args->bw_count[0]++; + return testSuccess; +} + +void +setupArgs(size_t size, ncclDataType_t type, struct threadArgs* args) +{ + int nranks = args->nProcs * args->nGpus * args->nThreads; + size_t count, sendCount, recvCount, paramCount, sendInplaceOffset, recvInplaceOffset; + + count = size / wordSize(type); + args->collTest->getCollByteCount(&sendCount, &recvCount, ¶mCount, + &sendInplaceOffset, &recvInplaceOffset, + (size_t) count, wordSize(type), (size_t) nranks); + + args->nbytes = paramCount * wordSize(type); + args->sendBytes = sendCount * wordSize(type); + args->expectedBytes = recvCount * wordSize(type); + args->sendInplaceOffset = sendInplaceOffset * wordSize(type); + args->recvInplaceOffset = recvInplaceOffset * wordSize(type); +} + +testResult_t +TimeTest(struct threadArgs* args, ncclDataType_t type, const char* typeName, + ncclRedOp_t op, const char* opName, int root) +{ + // Sync to avoid first-call timeout + Barrier(args); + + // Warm-up for large size + setupArgs(args->maxbytes, type, args); +#if HIP_VERSION >= 50221310 + std::vector graphs(args->nGpus); + std::vector graphExec(args->nGpus); + if(cudaGraphLaunches >= 1) + { + // Begin cuda graph capture + for(int i = 0; i < args->nGpus; i++) + { + // Thread local mode is needed for: + // - Multi-thread mode: where graph capture and instantiation can happen + // concurrently across threads + // - P2P pre-connect: when there is no warm-up, P2P pre-connect is done during + // graph capture. + // Since pre-connect calls cudaMalloc, we cannot use global capture mode + CUDACHECK(cudaStreamBeginCapture(args->streams[i], + cudaStreamCaptureModeThreadLocal)); + } + } +#endif + for(int iter = 0; iter < warmup_iters; iter++) + { + TESTCHECK(startColl(args, type, op, root, 0, iter)); + } + +#if HIP_VERSION >= 50221310 + if(cudaGraphLaunches >= 1) + { + // End cuda graph capture + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK(cudaStreamEndCapture(args->streams[i], graphs.data() + i)); + } + // Instantiate cuda graph + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK( + cudaGraphInstantiate(graphExec.data() + i, graphs[i], NULL, NULL, 0)); + } + // Resync CPU, restart timing, launch cuda graph + Barrier(args); + for(int l = 0; l < cudaGraphLaunches; l++) + { + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK(cudaGraphLaunch(graphExec[i], args->streams[i])); + } + } + } +#endif + + TESTCHECK(completeColl(args)); + +#if HIP_VERSION >= 50221310 + if(cudaGraphLaunches >= 1) + { + // destroy cuda graph + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK(cudaGraphExecDestroy(graphExec[i])); + CUDACHECK(cudaGraphDestroy(graphs[i])); + } + } +#endif + + // Warm-up for small size + setupArgs(args->minbytes, type, args); +#if HIP_VERSION >= 50221310 + if(cudaGraphLaunches >= 1) + { + // Begin cuda graph capture + for(int i = 0; i < args->nGpus; i++) + { + // Thread local mode is needed for: + // - Multi-thread mode: where graph capture and instantiation can happen + // concurrently across threads + // - P2P pre-connect: when there is no warm-up, P2P pre-connect is done during + // graph capture. + // Since pre-connect calls cudaMalloc, we cannot use global capture mode + CUDACHECK(cudaStreamBeginCapture(args->streams[i], + cudaStreamCaptureModeThreadLocal)); + } + } +#endif + for(int iter = 0; iter < warmup_iters; iter++) + { + TESTCHECK(startColl(args, type, op, root, iter < warmup_iters / 2 ? 0 : 1, iter)); + } + +#if HIP_VERSION >= 50221310 + if(cudaGraphLaunches >= 1) + { + // End cuda graph capture + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK(cudaStreamEndCapture(args->streams[i], graphs.data() + i)); + } + // Instantiate cuda graph + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK( + cudaGraphInstantiate(graphExec.data() + i, graphs[i], NULL, NULL, 0)); + } + // Resync CPU, restart timing, launch cuda graph + Barrier(args); + for(int l = 0; l < cudaGraphLaunches; l++) + { + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK(cudaGraphLaunch(graphExec[i], args->streams[i])); + } + } + } +#endif + + TESTCHECK(completeColl(args)); + +#if HIP_VERSION >= 50221310 + if(cudaGraphLaunches >= 1) + { + // destroy cuda graph + for(int i = 0; i < args->nGpus; i++) + { + CUDACHECK(cudaGraphExecDestroy(graphExec[i])); + CUDACHECK(cudaGraphDestroy(graphs[i])); + } + } +#endif + + // Benchmark + long repeat = run_cycles; + size_t iter = 0; + + do + { + if(run_cycles > 1) PRINT("# Testing %lu cycle.\n", iter + 1); + if(args->reporter) + { + args->reporter->setParameters(iter, args->collTest->name, typeName, opName); + } + for(size_t size = args->minbytes; size <= args->maxbytes; + size = ((args->stepfactor > 1) ? size * args->stepfactor + : size + args->stepbytes)) + { + setupArgs(size, type, args); + char rootName[100]; + sprintf(rootName, "%6i", root); + PRINT("%12li %12li %8s %6s %6s", + std::max(args->sendBytes, args->expectedBytes), + args->nbytes / wordSize(type), typeName, opName, rootName); + if(enable_out_of_place) + { + TESTCHECK(BenchTime(args, type, op, root, 0)); + usleep(delay_inout_place); + } + if(enable_in_place) TESTCHECK(BenchTime(args, type, op, root, 1)); + PRINT("\n"); + } + --repeat; + ++iter; + } while(repeat != 0); + + return testSuccess; +} + +testResult_t +threadRunTests(struct threadArgs* args) +{ + // Set device to the first of our GPUs. If we don't do that, some operations + // will be done on the current GPU (by default : 0) and if the GPUs are in + // exclusive mode those operations will fail. + CUDACHECK(cudaSetDevice(args->gpus[0])); + TESTCHECK(ncclTestEngine.runTest(args, ncclroot, (ncclDataType_t) nccltype, + test_typenames[nccltype], (ncclRedOp_t) ncclop, + test_opnames[ncclop])); + return testSuccess; +} + +testResult_t +threadInit(struct threadArgs* args) +{ + char hostname[1024]; + getHostName(hostname, 1024); + int nranks = args->nProcs * args->nThreads * args->nGpus; + + // set main thread again + is_main_thread = (is_main_proc && args->thread == 0) ? 1 : 0; + + NCCLCHECK(ncclGroupStart()); + for(int i = 0; i < args->nGpus; i++) + { + int rank = + args->proc * args->nThreads * args->nGpus + args->thread * args->nGpus + i; + CUDACHECK(cudaSetDevice(args->gpus[i])); + NCCLCHECK(ncclCommInitRank(args->comms + i, nranks, args->ncclId, rank)); + } + NCCLCHECK(ncclGroupEnd()); +#if NCCL_VERSION_CODE >= NCCL_VERSION(2, 19, 0) + void** sendRegHandles = + (local_register) ? (void**) malloc(sizeof(*sendRegHandles) * args->nGpus) : NULL; + void** recvRegHandles = + (local_register) ? (void**) malloc(sizeof(*recvRegHandles) * args->nGpus) : NULL; + for(int i = 0; i < args->nGpus; i++) + { + if(local_register) + NCCLCHECK(ncclCommRegister(args->comms[i], args->sendbuffs[i], args->maxbytes, + &sendRegHandles[i])); + if(local_register) + NCCLCHECK(ncclCommRegister(args->comms[i], args->recvbuffs[i], args->maxbytes, + &recvRegHandles[i])); + } +#endif + + TESTCHECK(threadRunTests(args)); + + for(int i = 0; i < args->nGpus; i++) + { +#if NCCL_VERSION_CODE >= NCCL_VERSION(2, 19, 0) + if(local_register) + NCCLCHECK(ncclCommDeregister(args->comms[i], sendRegHandles[i])); + if(local_register) + NCCLCHECK(ncclCommDeregister(args->comms[i], recvRegHandles[i])); +#endif + NCCLCHECK(ncclCommDestroy(args->comms[i])); + } + return testSuccess; +} + +void* +threadLauncher(void* thread_) +{ + struct testThread* thread = (struct testThread*) thread_; + thread->ret = thread->func(&thread->args); + return NULL; +} +testResult_t +threadLaunch(struct testThread* thread) +{ + pthread_create(&thread->thread, NULL, threadLauncher, thread); + return testSuccess; +} + +testResult_t +AllocateBuffs(void** sendbuff, size_t sendBytes, void** recvbuff, size_t recvBytes, + void** expected, size_t nbytes) +{ + if(enable_rotating_tensor) + { + recvBytes = recvBytes + cache_bytes; + nbytes = nbytes + cache_bytes; + } + if(memorytype == ncclFine) + { + if(HIP_VERSION >= 50700000) + { + CUDACHECK(hipExtMallocWithFlags(sendbuff, nbytes, hipDeviceMallocUncached)); + CUDACHECK(hipExtMallocWithFlags(recvbuff, nbytes, hipDeviceMallocUncached)); + if(datacheck) + CUDACHECK( + hipExtMallocWithFlags(expected, recvBytes, hipDeviceMallocUncached)); + } + else + { + CUDACHECK( + hipExtMallocWithFlags(sendbuff, nbytes, hipDeviceMallocFinegrained)); + CUDACHECK( + hipExtMallocWithFlags(recvbuff, nbytes, hipDeviceMallocFinegrained)); + if(datacheck) + CUDACHECK(hipExtMallocWithFlags(expected, recvBytes, + hipDeviceMallocFinegrained)); + } + } + else if(memorytype == ncclHost) + { + CUDACHECK(hipHostMalloc(sendbuff, nbytes)); + CUDACHECK(hipHostMalloc(recvbuff, nbytes)); + if(datacheck) CUDACHECK(hipHostMalloc(expected, recvBytes)); + } + else if(memorytype == ncclManaged) + { + CUDACHECK(cudaMallocManaged(sendbuff, nbytes)); + CUDACHECK(cudaMallocManaged(recvbuff, nbytes)); + if(datacheck) CUDACHECK(cudaMallocManaged(expected, recvBytes)); +#if 0 + CUDACHECK(cudaMemset(*sendbuff, 0, nbytes)); + CUDACHECK(cudaMemset(*recvbuff, 0, nbytes)); + if (datacheck) CUDACHECK(cudaMemset(*expected, 0, recvBytes)); +#endif + } + else + { + CUDACHECK(cudaMalloc(sendbuff, nbytes)); + CUDACHECK(cudaMalloc(recvbuff, nbytes)); + if(datacheck) CUDACHECK(cudaMalloc(expected, recvBytes)); + } + CUDACHECK(hipMemset(*sendbuff, 1, nbytes)); + if(datacheck) CUDACHECK(hipMemset(*expected, 1, recvBytes)); + return testSuccess; +} + +testResult_t +run(); // Main function + +int +main(int argc, char* argv[]) +{ + // Make sure everyline is flushed so that we see the progress of the test + setlinebuf(stdout); + +#if NCCL_VERSION_CODE >= NCCL_VERSION(2, 4, 0) + ncclGetVersion(&test_ncclVersion); +#else + test_ncclVersion = NCCL_VERSION_CODE; +#endif + +// printf("# NCCL_VERSION_CODE=%d ncclGetVersion=%d\n", NCCL_VERSION_CODE, +// test_ncclVersion); +#if NCCL_VERSION_CODE >= NCCL_VERSION(2, 0, 0) + test_opnum = 4; + test_typenum = 9; + if(NCCL_VERSION_CODE >= NCCL_VERSION(2, 10, 0) && + test_ncclVersion >= NCCL_VERSION(2, 10, 0)) + { + test_opnum++; // ncclAvg +# if defined(RCCL_BFLOAT16) + test_typenum++; // bfloat16 +# endif +# if defined(RCCL_FLOAT8) + test_typenum++; // fp8_e4m3 + test_typenum++; // fp8_e5m2 +# endif + } + if(NCCL_VERSION_CODE >= NCCL_VERSION(2, 11, 0) && + test_ncclVersion >= NCCL_VERSION(2, 11, 0)) + { + test_opnum++; // PreMulSum + } +#endif + + // Parse args + // Replace getopt_long with manual argument parsing + double parsed; + for(int argi = 1; argi < argc; ++argi) + { + const char* arg = argv[argi]; + if(strcmp(arg, "-t") == 0 || strcmp(arg, "--nthreads") == 0) + { + nThreads = strtol(argv[++argi], NULL, 0); + } + else if(strcmp(arg, "-g") == 0 || strcmp(arg, "--ngpus") == 0) + { + nGpus = strtol(argv[++argi], NULL, 0); + } + else if(strcmp(arg, "-b") == 0 || strcmp(arg, "--minbytes") == 0) + { + parsed = parsesize(argv[++argi]); + if(parsed < 0) + { + fprintf(stderr, "invalid size specified for 'minbytes'\n"); + return -1; + } + minBytes = (size_t) parsed; + } + else if(strcmp(arg, "-e") == 0 || strcmp(arg, "--maxbytes") == 0) + { + parsed = parsesize(argv[++argi]); + if(parsed < 0) + { + fprintf(stderr, "invalid size specified for 'maxbytes'\n"); + return -1; + } + maxBytes = (size_t) parsed; + } + else if(strcmp(arg, "-i") == 0 || strcmp(arg, "--stepbytes") == 0) + { + parsed = parsesize(argv[++argi]); + if(parsed < 0) + { + fprintf(stderr, "invalid size specified for 'stepBytes'\n"); + return -1; + } + stepBytes = (size_t) parsed; + } + else if(strcmp(arg, "-f") == 0 || strcmp(arg, "--stepfactor") == 0) + { + stepFactor = strtol(argv[++argi], NULL, 0); + } + else if(strcmp(arg, "-n") == 0 || strcmp(arg, "--iters") == 0) + { + iters = (int) strtol(argv[++argi], NULL, 0); + } + else if(strcmp(arg, "-m") == 0 || strcmp(arg, "--agg_iters") == 0) + { +#if NCCL_MAJOR > 2 || (NCCL_MAJOR >= 2 && NCCL_MINOR >= 2) + agg_iters = (int) strtol(argv[++argi], NULL, 0); +#else + fprintf(stderr, "Option -m not supported before NCCL 2.2. Ignoring\n"); + ++argi; +#endif + } + else if(strcmp(arg, "-w") == 0 || strcmp(arg, "--warmup_iters") == 0) + { + warmup_iters = (int) strtol(argv[++argi], NULL, 0); + } + else if(strcmp(arg, "-N") == 0 || strcmp(arg, "--run_cycles") == 0) + { + run_cycles = (int) strtol(argv[++argi], NULL, 0); + } + else if(strcmp(arg, "-p") == 0 || strcmp(arg, "--parallel_init") == 0) + { + parallel_init = (int) strtol(argv[++argi], NULL, 0); + } + else if(strcmp(arg, "-c") == 0 || strcmp(arg, "--check") == 0) + { + datacheck = (int) strtol(argv[++argi], NULL, 0); + } + else if(strcmp(arg, "-o") == 0 || strcmp(arg, "--op") == 0) + { + ncclop = ncclstringtoop(argv[++argi]); + } + else if(strcmp(arg, "-d") == 0 || strcmp(arg, "--datatype") == 0) + { + nccltype = ncclstringtotype(argv[++argi]); + } + else if(strcmp(arg, "-r") == 0 || strcmp(arg, "--root") == 0) + { + ncclroot = ncclstringtoroot(argv[++argi]); + } + else if(strcmp(arg, "-z") == 0 || strcmp(arg, "--blocking") == 0) + { + blocking_coll = strtol(argv[++argi], NULL, 0); + } + else if(strcmp(arg, "-y") == 0 || strcmp(arg, "--stream_null") == 0) + { + streamnull = strtol(argv[++argi], NULL, 0); + } + else if(strcmp(arg, "-T") == 0 || strcmp(arg, "--timeout") == 0) + { + timeout = strtol(argv[++argi], NULL, 0); + } + else if(strcmp(arg, "-G") == 0 || strcmp(arg, "--cudagraph") == 0) + { +#if(NCCL_MAJOR > 2 || (NCCL_MAJOR >= 2 && NCCL_MINOR >= 9)) && HIP_VERSION >= 50221310 + cudaGraphLaunches = strtol(argv[++argi], NULL, 0); +#else + printf("Option -G (HIP graph) not supported before NCCL 2.9 + ROCm 5.2 " + "Ignoring\n"); + ++argi; +#endif + } + else if(strcmp(arg, "-C") == 0 || strcmp(arg, "--report_cputime") == 0) + { + report_cputime = strtol(argv[++argi], NULL, 0); + } + else if(strcmp(arg, "-a") == 0 || strcmp(arg, "--average") == 0) + { + average = (int) strtol(argv[++argi], NULL, 0); + } + else if(strcmp(arg, "-R") == 0 || strcmp(arg, "--local_register") == 0) + { +#if NCCL_VERSION_CODE >= NCCL_VERSION(2, 19, 0) + if((int) strtol(argv[++argi], NULL, 0)) + { + local_register = 1; + } +#else + printf("Option -R (register) is not supported before NCCL 2.19. Ignoring\n"); + ++argi; +#endif + } + else if(strcmp(arg, "-Y") == 0 || strcmp(arg, "--memory_type") == 0) + { + memorytype = ncclstringtomtype(argv[++argi]); + } + else if(strcmp(arg, "-u") == 0 || strcmp(arg, "--cumask") == 0) + { + int nmasks = 0; + char* maskstr = argv[++argi]; + char* mask = strtok(maskstr, ","); + while(mask != NULL && nmasks < 4) + { + cumask[nmasks++] = strtol(mask, NULL, 16); + mask = strtok(NULL, ","); + } + } + else if(strcmp(arg, "-O") == 0 || strcmp(arg, "--out_of_place") == 0) + { + enable_out_of_place = strtol(argv[++argi], NULL, 0); + enable_in_place = enable_out_of_place ? 0 : 1; + } + else if(strcmp(arg, "-q") == 0 || strcmp(arg, "--delay_inout_place") == 0) + { + delay_inout_place = (int) strtol(argv[++argi], NULL, 10); + } + else if(strcmp(arg, "-F") == 0 || strcmp(arg, "--cache_flush") == 0) + { + enable_cache_flush = strtol(argv[++argi], NULL, 0); + if(enable_cache_flush > 0) + { + hipDeviceProp_t deviceProps; + CHECK_HIP_ERROR(hipGetDeviceProperties(&deviceProps, 0)); + gpu_block3 = deviceProps.multiProcessorCount * 60; + } + } + else if(strcmp(arg, "-E") == 0 || strcmp(arg, "--rotating_tensor") == 0) + { + enable_rotating_tensor = strtol(argv[++argi], NULL, 0); + } + else if(strcmp(arg, "-x") == 0 || strcmp(arg, "--output_file") == 0) + { + output_file = argv[++argi]; + } + else if(strcmp(arg, "-Z") == 0 || strcmp(arg, "--output_format") == 0) + { + output_format = argv[++argi]; + } + else if(strcmp(arg, "-h") == 0 || strcmp(arg, "--help") == 0) + { + printf("USAGE: %s \n\t" + "[-t,--nthreads ] \n\t" + "[-g,--ngpus ] \n\t" + "[-b,--minbytes ] \n\t" + "[-e,--maxbytes ] \n\t" + "[-i,--stepbytes ] \n\t" + "[-f,--stepfactor ] \n\t" + "[-n,--iters ] \n\t" + "[-m,--agg_iters ] \n\t" + "[-w,--warmup_iters ] \n\t" + "[-N,--run_cycles run & print each cycle (default: 1; " + "0=infinite)] \n\t" + "[-p,--parallel_init <0/1>] \n\t" + "[-c,--check ] \n\t" +#if NCCL_VERSION_CODE >= NCCL_VERSION(2, 11, 0) + "[-o,--op ] \n\t" +#elif NCCL_VERSION_CODE >= NCCL_VERSION(2, 10, 0) + "[-o,--op ] \n\t" +#else + "[-o,--op ] \n\t" +#endif + "[-d,--datatype ] \n\t" + "[-r,--root ] \n\t" + "[-z,--blocking <0/1>] \n\t" + "[-y,--stream_null <0/1>] \n\t" + "[-T,--timeout