AI/rocm-systems
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/develop' into 2.8.3

Browse Source 
[ROCm/rccl commit: ab1e7a0318]
This commit is contained in:
Wenkai Du
2021-02-04 19:46:48 -05:00
parent 1dfb88f554 16d625ca27
commit ae5779702a
36 changed files with 2820 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
projects/rccl/CMakeLists.txt
+15 -1
View File
@@ -11,7 +11,21 @@ set(CMAKE_INSTALL_PREFIX "/opt/rocm" CACHE PATH "")
project(rccl CXX)
set(AMDGPU_TARGETS gfx803;gfx900;gfx906;gfx908 CACHE STRING "List of specific machine types for library to target")
# Detect compiler support for target ID
if( CMAKE_CXX_COMPILER MATCHES ".*/hipcc$" )
execute_process(COMMAND ${CMAKE_CXX_COMPILER} "--help"
OUTPUT_VARIABLE CXX_OUTPUT
OUTPUT_STRIP_TRAILING_WHITESPACE
ERROR_STRIP_TRAILING_WHITESPACE)
string(REGEX MATCH ".mcode\-object\-version" TARGET_ID_SUPPORT ${CXX_OUTPUT})
endif()
# Use target ID syntax if supported for AMDGPU_TARGETS
if(TARGET_ID_SUPPORT)
set(AMDGPU_TARGETS gfx803;gfx900;gfx906:xnack-;gfx908:xnack- CACHE STRING "List of specific machine types for library to target")
else()
set(AMDGPU_TARGETS gfx803;gfx900;gfx906;gfx908 CACHE STRING "List of specific machine types for library to target")
endif()
option(BUILD_TESTS "Build test programs" OFF)
option(INSTALL_DEPENDENCIES "Force install dependencies" OFF)
projects/rccl/install.sh
+3 -1
View File
@@ -1,5 +1,5 @@
#!/bin/bash
# Copyright (c) 2019-2020 Advanced Micro Devices, Inc. All rights reserved.
# Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
# #################################################
# helper functions
@@ -200,8 +200,10 @@ if ($run_tests); then
if (test -f "./test/UnitTests"); then
if ($run_tests_all); then
./test/UnitTests
NCCL_COMM_ID=$HOSTNAME:55512 ./test/UnitTestsMultiProcess
else
./test/UnitTests --gtest_filter="BroadcastCorrectnessSweep*:*float32*"
NCCL_COMM_ID=$HOSTNAME:55512 ./test/UnitTestsMultiProcess --gtest_filter="BroadcastMultiProcessCorrectnessSweep*:*float32*"
fi
else
echo "Unit tests have not been built yet; please re-run script with -t to build unit tests."
projects/rccl/src/clique/CliqueManager.cc
+18 -6
View File
@@ -46,8 +46,8 @@ int* CliqueManager::m_staticGpuBarrierMem = NULL;
// Define some environment variables that affect clique-based kernels
RCCL_PARAM(EnableClique, "ENABLE_CLIQUE", 0); // Opt-in environment variable for clique-based kernels
RCCL_PARAM(AllReduceCliqueByteLimit, "CLIQUE_ALLREDUCE_BYTE_LIMIT", 2097152); // Max number of bytes to use clique-based kernels for all reduce
RCCL_PARAM(AllReduceNumChannels, "CLIQUE_ALLREDUCE_NCHANNELS", 4); // Number of channels to use for all-reduce
RCCL_PARAM(AllReduceCliqueByteLimit, "CLIQUE_ALLREDUCE_BYTE_LIMIT", 16777216); // Max number of bytes to use clique-based kernels for all reduce
RCCL_PARAM(AllReduceNumChannels, "CLIQUE_ALLREDUCE_NCHANNELS", 0); // Number of channels to use for all-reduce. (0 for auto-select)
CliqueManager::CliqueManager(int const rank,
int const numRanks,
@@ -315,7 +315,22 @@ ncclResult_t CliqueManager::GetNumChannelsToUse(ncclFunc_t const coll,
*numChannelstoUse = 1;
if (coll == ncclFuncAllReduce) {
*numChannelstoUse = std::min((int)rcclParamAllReduceNumChannels(), totalNumChannels);
if (rcclParamAllReduceNumChannels() == 0)
{
// NOTE: These are currently based on collected data and not necessarily ideal for all hardware
int numChannels;
if (totalBytes <= 65536) numChannels = 1;
else if (totalBytes <= 262144) numChannels = 2;
else if (totalBytes <= 524288) numChannels = 4;
else if (totalBytes <= 2097152) numChannels = 8;
else numChannels = 11;
*numChannelstoUse = std::min(numChannels, totalNumChannels);
}
else
{
*numChannelstoUse = std::min((int)rcclParamAllReduceNumChannels(), totalNumChannels);
}
}
return ncclSuccess;
@@ -337,9 +352,6 @@ ncclResult_t CliqueManager::SetCliqueArgs(ncclWorkElem* args)
int opIndex = args->opCount % NCCL_MAX_OPS;
args->clique.ptrs = &m_pinnedCliquePtrs[opIndex];
// Determine number of channels to use for this collective
args->clique.nChannels = rcclParamAllReduceNumChannels();
return ncclSuccess;
}
projects/rccl/src/collectives/device/gather.h
+75
View File
@@ -0,0 +1,75 @@
/*************************************************************************
* Copyright (c) 2015-2019, NVIDIA CORPORATION. All rights reserved.
* Modifications Copyright (c) 2019-2020 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#include "devcomm.h"
#include "primitives.h"
#include "collectives.h"
template<int UNROLL, class FUNC, typename T>
__attribute__((noinline))
__device__ void ncclGatherKernel(struct CollectiveArgs* args) {
const int tid = threadIdx.x;
const int nthreads = args->coll.nThreads;
const int nChannels = args->coll.nChannels;
struct ncclDevComm* comm = args->comm;
struct ncclChannel* channel = comm->channels+blockIdx.x;
struct ncclRing* ring = &channel->ring;
const ssize_t size = args->coll.count;
const int nranks = comm->nRanks;
const int bid = args->coll.bid;
const int rank = ring->devUserRanks[0];
const int stepSize = comm->buffSizes[NCCL_PROTO_SIMPLE] / (sizeof(T)*NCCL_STEPS);
const int chunkSize = stepSize * GATHER_CHUNKSTEPS;
const int peersPerChan = DIVUP(nranks, nChannels);
const ssize_t loopSize = (peersPerChan == 1 ? (nChannels/nranks)*(ssize_t)chunkSize : (ssize_t)chunkSize);
const int root = args->coll.root;
// Compute pointers
const T * __restrict__ thisInput = (const T*)args->sendbuff;
T * __restrict__ thisOutput = (T*)args->recvbuff;
for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += loopSize) {
for (int i = 0; i < peersPerChan; i++) {
if ((peersPerChan == 1 && blockIdx.x >= (nChannels/nranks)*nranks) ||
(peersPerChan > 1 && blockIdx.x*peersPerChan+i >= nranks))
continue;
int realChunkSize = min(chunkSize, DIVUP(size-gridOffset, (peersPerChan == 1 ? (nChannels/nranks) : 1)));
ALIGN_SIZE(realChunkSize, nthreads*sizeof(uint64_t)/sizeof(T));
ssize_t chunkOffset = gridOffset + (peersPerChan == 1 ? (bid/nranks)*realChunkSize : 0);
int nelem = min(realChunkSize, size-chunkOffset);
if ((blockIdx.x*peersPerChan+i)%nranks == 0 && rank == root) {
const T* sendbuff = thisInput+chunkOffset;
T* recvbuff = thisOutput+chunkOffset+rank*size;
if (tid < nthreads && sendbuff != recvbuff) {
// local copy
ReduceOrCopyMulti<UNROLL, FUNC, T, 1, 1, 1, 1>(tid, nthreads, 1, &sendbuff, 1, &recvbuff, nelem);
}
}
else {
int peerSend = (rank+(blockIdx.x*peersPerChan)+i)%nranks;
int peerRecv = (2*nranks+rank-((blockIdx.x*peersPerChan)%nranks)-(i%nranks))%nranks;
int peerNone = -1;
if (rank == root) {
ncclPrimitives<UNROLL, GATHER_CHUNKSTEPS/GATHER_SLICESTEPS, GATHER_SLICESTEPS, T, 1, 1, 0, FUNC>
prims(tid, nthreads, &peerRecv, &peerNone, NULL, stepSize, channel, comm);
ssize_t recv_offset = chunkOffset + peerRecv*size;
prims.recv(thisOutput+recv_offset, nelem);
}
else {
if (peerSend == root) {
ncclPrimitives<UNROLL, GATHER_CHUNKSTEPS/GATHER_SLICESTEPS, GATHER_SLICESTEPS, T, 1, 1, 0, FUNC>
prims(tid, nthreads, &peerNone, &peerSend, NULL, stepSize, channel, comm);
ssize_t send_offset = chunkOffset;
prims.send(thisInput+send_offset, nelem);
}
}
}
}
}
}
projects/rccl/src/collectives/device/scatter.h
+75
View File
@@ -0,0 +1,75 @@
/*************************************************************************
* Copyright (c) 2015-2019, NVIDIA CORPORATION. All rights reserved.
* Modifications Copyright (c) 2019-2020 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#include "devcomm.h"
#include "primitives.h"
#include "collectives.h"
template<int UNROLL, class FUNC, typename T>
__attribute__((noinline))
__device__ void ncclScatterKernel(struct CollectiveArgs* args) {
const int tid = threadIdx.x;
const int nthreads = args->coll.nThreads;
const int nChannels = args->coll.nChannels;
struct ncclDevComm* comm = args->comm;
struct ncclChannel* channel = comm->channels+blockIdx.x;
struct ncclRing* ring = &channel->ring;
const ssize_t size = args->coll.count;
const int nranks = comm->nRanks;
const int bid = args->coll.bid;
const int rank = ring->devUserRanks[0];
const int stepSize = comm->buffSizes[NCCL_PROTO_SIMPLE] / (sizeof(T)*NCCL_STEPS);
const int chunkSize = stepSize * SCATTER_CHUNKSTEPS;
const int peersPerChan = DIVUP(nranks, nChannels);
const ssize_t loopSize = (peersPerChan == 1 ? (nChannels/nranks)*(ssize_t)chunkSize : (ssize_t)chunkSize);
const int root = args->coll.root;
// Compute pointers
const T * __restrict__ thisInput = (const T*)args->sendbuff;
T * __restrict__ thisOutput = (T*)args->recvbuff;
for (ssize_t gridOffset = 0; gridOffset < size; gridOffset += loopSize) {
for (int i = 0; i < peersPerChan; i++) {
if ((peersPerChan == 1 && blockIdx.x >= (nChannels/nranks)*nranks) ||
(peersPerChan > 1 && blockIdx.x*peersPerChan+i >= nranks))
continue;
int realChunkSize = min(chunkSize, DIVUP(size-gridOffset, (peersPerChan == 1 ? (nChannels/nranks) : 1)));
ALIGN_SIZE(realChunkSize, nthreads*sizeof(uint64_t)/sizeof(T));
ssize_t chunkOffset = gridOffset + (peersPerChan == 1 ? (bid/nranks)*realChunkSize : 0);
int nelem = min(realChunkSize, size-chunkOffset);
if ((blockIdx.x*peersPerChan+i)%nranks == 0 && rank == root) {
const T* sendbuff = thisInput+chunkOffset+rank*size;
T* recvbuff = thisOutput+chunkOffset;
if (tid < nthreads && sendbuff != recvbuff) {
// local copy
ReduceOrCopyMulti<UNROLL, FUNC, T, 1, 1, 1, 1>(tid, nthreads, 1, &sendbuff, 1, &recvbuff, nelem);
}
}
else {
int peerSend = (rank+(blockIdx.x*peersPerChan)+i)%nranks;
int peerRecv = (2*nranks+rank-((blockIdx.x*peersPerChan)%nranks)-(i%nranks))%nranks;
int peerNone = -1;
if (rank == root) {
ncclPrimitives<UNROLL, SCATTER_CHUNKSTEPS/SCATTER_SLICESTEPS, SCATTER_SLICESTEPS, T, 1, 1, 0, FUNC>
prims(tid, nthreads, &peerNone, &peerSend, NULL, stepSize, channel, comm);
ssize_t send_offset = chunkOffset + peerSend*size;
prims.send(thisInput+send_offset, nelem);
}
else {
if (peerRecv == root) {
ncclPrimitives<UNROLL, SCATTER_CHUNKSTEPS/SCATTER_SLICESTEPS, SCATTER_SLICESTEPS, T, 1, 1, 0, FUNC>
prims(tid, nthreads, &peerRecv, &peerNone, NULL, stepSize, channel, comm);
ssize_t recv_offset = chunkOffset;
prims.recv(thisOutput+recv_offset, nelem);
}
}
}
}
}
}
projects/rccl/src/graph/rome_models.h
+24
View File
@@ -213,6 +213,28 @@ static struct rcclRomeModel rome_model_44 = {
.ringBase = "5 4 7 6 2 1 3 0|5 6 7 4 1 0 2 3",
};
static struct rcclRomeModel rome_model_45 = {
.nGpus = 8, .nCpus = 7, .nNics = 0, .nLinks = 3,
.gpuIds = { 0x43000, 0x23000, 0x26000, 0x3000, 0xc3000, 0xc6000, 0xa3000, 0x83000, },
.nicIds = { },
.gpuNuma = { 1, 2, 2, 3, 5, 5, 6, 7, },
.nicNuma = { },
.connMatrix = { 0, 1, 1, 1, 0, 0, 0, 0, 1, 0, 1, 1, 0, 0, 0, 0, 1, 1, 0, 1, 0, 0, 0, 0, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 1, 0, 0, 0, 0, 1, 0, 1, 1, 0, 0, 0, 0, 1, 1, 0, 1, 0, 0, 0, 0, 1, 1, 1, 0, },
.pattern = "00102010002010",
.ringBase = "0 1 2 3 4 5 6 7|0 2 5 7 4 6 1 3|0 3 1 6 4 7 5 2|0 7 6 5 4 3 2 1",
};
static struct rcclRomeModel rome_model_46 = {
.nGpus = 8, .nCpus = 7, .nNics = 1, .nLinks = 3,
.gpuIds = { 0x43000, 0x23000, 0x26000, 0x3000, 0xc3000, 0xc6000, 0xa3000, 0x83000, },
.nicIds = { 0xe1000, },
.gpuNuma = { 1, 2, 2, 3, 5, 5, 6, 7, },
.nicNuma = { 4, },
.connMatrix = { 0, 1, 1, 1, 0, 0, 0, 0, 1, 0, 1, 1, 0, 0, 0, 0, 1, 1, 0, 1, 0, 0, 0, 0, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 1, 0, 0, 0, 0, 1, 0, 1, 1, 0, 0, 0, 0, 1, 1, 0, 1, 0, 0, 0, 0, 1, 1, 1, 0, },
.pattern = "00102010012010",
.ringBase = "6 5 7 4 1 2 3 0|7 4 6 5 1 0 3 2",
};
static struct rcclRomeModel romeTopoModels[] = {
rome_model_22,
rome_model_25,
@@ -230,4 +252,6 @@ static struct rcclRomeModel romeTopoModels[] = {
rome_model_40,
rome_model_42,
rome_model_44,
rome_model_45,
rome_model_46,
};
projects/rccl/src/graph/search.cc
+4 -2
View File
@@ -462,7 +462,7 @@ ncclResult_t ncclTopoSearchRecNet(struct ncclTopoSystem* system, struct ncclTopo
for (int i = 0; i<system->nodes[GPU].count; i++)
if (paths[i].count < paths[f].count) f = i;
int t = 1 << 10;
NCCLCHECK(ncclTopoSearchTryGpu(system, graph, saveGraph, 0, backToNet, backToFirstRank, FORCED_ORDER_PCI, &t, NET, n, f));
NCCLCHECK(ncclTopoSearchTryGpu(system, graph, saveGraph, 0, backToNet, backToFirstRank, (f == 0) ? FORCED_ORDER_PCI : 0, &t, NET, n, f));
if (t == -1) *time = -1;
}
@@ -998,8 +998,10 @@ static bool permuteGpuIds(int *g, int n, int last, struct rcclRomeModel* ref, st
if (i < ref->nGpus) return false;
// match XGMI connection
for (i = 0; i < ref->nGpus; i++) {
for (j = 0; j < ref->nGpus; j++)
for (j = 0; j < ref->nGpus; j++) {
if (ref->connMatrix[i*ref->nGpus+j] != topo->connMatrix[g[i]*ref->nGpus+g[j]]) break;
if ((ref->gpuIds[i]-ref->gpuIds[j])*(topo->gpuIds[g[i]]-topo->gpuIds[g[j]]) < 0) break;
}
if (j < ref->nGpus) break;
}
if (i < ref->nGpus) return false;
projects/rccl/src/graph/xml.cc
+9 -2
View File
@@ -492,7 +492,10 @@ ncclResult_t ncclTopoGetXmlFromSys(struct ncclXmlNode* pciNode, struct ncclXml*
char portSpeedStr[MAX_STR_LEN];
float portSpeed;
NCCLCHECK(ncclTopoGetStrFromSys(path, "../max_link_speed", portSpeedStr));
sscanf(portSpeedStr, "%f GT/s", &portSpeed);
if (portSpeedStr[0])
sscanf(portSpeedStr, "%f GT/s", &portSpeed);
else
portSpeed = deviceSpeed;
NCCLCHECK(xmlSetAttr(pciNode, "link_speed", portSpeed < deviceSpeed ? portSpeedStr : deviceSpeedStr));
}
NCCLCHECK(xmlGetAttrIndex(pciNode, "link_width", &index));
@@ -502,7 +505,11 @@ ncclResult_t ncclTopoGetXmlFromSys(struct ncclXmlNode* pciNode, struct ncclXml*
NCCLCHECK(ncclTopoGetStrFromSys(path, "max_link_width", strValue));
int deviceWidth = strtol(strValue, NULL, 0);
NCCLCHECK(ncclTopoGetStrFromSys(path, "../max_link_width", strValue));
int portWidth = strtol(strValue, NULL, 0);
int portWidth;
if (strValue[0])
portWidth = strtol(strValue, NULL, 0);
else
portWidth = deviceWidth;
NCCLCHECK(xmlSetAttrInt(pciNode, "link_width", std::min(deviceWidth,portWidth)));
}
struct ncclXmlNode* parent = pciNode->parent;
projects/rccl/test/CMakeLists.txt
+26 -5
View File
@@ -1,14 +1,14 @@
# Copyright (c) 2019-2020 Advanced Micro Devices, Inc. All rights reserved.
# Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
cmake_minimum_required(VERSION 2.8.12)
if(BUILD_TESTS)
message("Going to build unit tests (Installed in /test/UnitTests)")
include_directories(${GTEST_INCLUDE_DIRS})
include_directories(${GTEST_INCLUDE_DIRS})
# Collect source files for tests
set(TEST_SOURCES
set(TEST_SOURCES_SINGLE_PROCESS
test_AllGather.cpp
test_AllReduce.cpp
test_Broadcast.cpp
@@ -24,18 +24,39 @@ if(BUILD_TESTS)
test_AllToAllv.cpp
)
add_executable(UnitTests ${TEST_SOURCES})
set(TEST_SOURCES_MULTI_PROCESS
test_AllGatherMultiProcess.cpp
test_AllReduceMultiProcess.cpp
test_AllToAllMultiProcess.cpp
test_BroadcastMultiProcess.cpp
test_CombinedCallsMultiProcess.cpp
test_GatherMultiProcess.cpp
test_GroupCallsMultiProcess.cpp
test_ReduceMultiProcess.cpp
test_ReduceScatterMultiProcess.cpp
test_ScatterMultiProcess.cpp
)
add_executable(UnitTests ${TEST_SOURCES_SINGLE_PROCESS})
target_include_directories(UnitTests PRIVATE /opt/rocm ${GTEST_INCLUDE_DIRS})
target_link_libraries(UnitTests PRIVATE ${GTEST_BOTH_LIBRARIES})
add_executable(UnitTestsMultiProcess ${TEST_SOURCES_MULTI_PROCESS})
target_include_directories(UnitTestsMultiProcess PRIVATE /opt/rocm ${GTEST_INCLUDE_DIRS})
target_link_libraries(UnitTestsMultiProcess PRIVATE ${GTEST_BOTH_LIBRARIES})
# UnitTests using static library of rccl requires passing rccl
# through -l and -L instead of command line input.
if(BUILD_STATIC)
add_dependencies(UnitTests rccl)
target_link_libraries(UnitTests PRIVATE dl rt numa -lrccl -L${CMAKE_BINARY_DIR})
target_link_libraries(UnitTests PRIVATE amdhip64 amd_comgr hsa-runtime64::hsa-runtime64)
add_dependencies(UnitTestsMultiProcess rccl)
target_link_libraries(UnitTestsMultiProcess PRIVATE dl rt numa -lrccl -L${CMAKE_BINARY_DIR})
target_link_libraries(UnitTestsMultiProcess PRIVATE amdhip64 amd_comgr hsa-runtime64::hsa-runtime64)
else()
target_link_libraries(UnitTests PRIVATE rccl)
target_link_libraries(UnitTestsMultiProcess PRIVATE rt rccl)
endif()
else()
message("Not building unit tests")
projects/rccl/test/CorrectnessTest.hpp
+515 -23
View File
@@ -1,15 +1,28 @@
/*************************************************************************
* Copyright (c) 2019-2020 Advanced Micro Devices, Inc. All rights reserved.
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#ifndef CORRECTNESSTEST_HPP
#define CORRECTNESSTEST_HPP
#include <algorithm>
#include <cstdio>
#include <cstdlib>
#include <errno.h>
#include <fcntl.h>
#include <semaphore.h>
#include <stdio.h>
#include <string>
#include <sys/mman.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <tuple>
#include <unistd.h>
#include <vector>
#include <gtest/gtest.h>
#include "rccl.h"
#include "../include/rccl_bfloat16.h"
@@ -88,11 +101,12 @@ namespace CorrectnessTests
return numElements * DataTypeToBytes(dataType);
}
void Initialize(int const numDevices_,
size_t const numElements_,
ncclDataType_t const dataType_,
bool const inPlace_,
ncclFunc_t const func_ = ncclCollBroadcast)
// To be used in multi-process tests, in the parent process before forking children.
void InitializeRootProcess(int const numDevices_,
size_t const numElements_,
ncclDataType_t const dataType_,
bool const inPlace_,
ncclFunc_t const func_ = ncclCollBroadcast)
{
numDevices = numDevices_;
numElements = numElements_;
@@ -100,22 +114,68 @@ namespace CorrectnessTests
inPlace = inPlace_;
function = func_;
inputs.resize(numDevices);
outputs.resize(numDevices);
expected.resize(numDevices);
for (int i = 0; i < numDevices_; i++)
{
void* ptr = (void*)mmap(NULL, sizeof(void*), PROT_READ|PROT_WRITE, MAP_SHARED|MAP_ANONYMOUS, -1, 0);
inputs.push_back(ptr);
}
for (int i = 0; i < numDevices_; i++)
{
void* ptr = (void*)mmap(NULL, sizeof(void*), PROT_READ|PROT_WRITE, MAP_SHARED|MAP_ANONYMOUS, -1, 0);
outputs.push_back(ptr);
}
for (int i = 0; i < numDevices_; i++)
{
void* ptr = (void*)mmap(NULL, NumBytes(ncclOutputBuffer), PROT_READ|PROT_WRITE, MAP_SHARED|MAP_ANONYMOUS, -1, 0);
expected.push_back(ptr);
}
}
void Initialize(int const numDevices_,
size_t const numElements_,
ncclDataType_t const dataType_,
bool const inPlace_,
ncclFunc_t const func_ = ncclCollBroadcast,
int const multiProcessRank_ = -1)
{
numDevices = numDevices_;
numElements = numElements_;
dataType = dataType_;
inPlace = inPlace_;
function = func_;
if (multiProcessRank_ == -1)
{
inputs.resize(numDevices);
outputs.resize(numDevices);
expected.resize(numDevices);
}
// Allocate per-device memory
for (int i = 0; i < numDevices; i++)
if (multiProcessRank_ > -1)
{
HIP_CALL(hipSetDevice(i));
HIP_CALL(hipMalloc((void **)&inputs[i], NumBytes(ncclInputBuffer)));
HIP_CALL(hipSetDevice(multiProcessRank_));
HIP_CALL(hipMalloc((void **)&inputs[multiProcessRank_], NumBytes(ncclInputBuffer)));
if (inPlace)
outputs[i] = inputs[i];
outputs[multiProcessRank_] = inputs[multiProcessRank_];
else
HIP_CALL(hipMalloc((void **)&outputs[i], NumBytes(ncclOutputBuffer)));
expected[i] = malloc(NumBytes(ncclOutputBuffer));
HIP_CALL(hipMalloc((void **)&outputs[multiProcessRank_], NumBytes(ncclOutputBuffer)));
}
else
{
for (int i = 0; i < numDevices; i++)
{
HIP_CALL(hipSetDevice(i));
HIP_CALL(hipMalloc((void **)&inputs[i], NumBytes(ncclInputBuffer)));
if (inPlace)
outputs[i] = inputs[i];
else
HIP_CALL(hipMalloc((void **)&outputs[i], NumBytes(ncclOutputBuffer)));
expected[i] = malloc(NumBytes(ncclOutputBuffer));
}
}
}
// Explicit memory release to avoid double-free from subDatasets
@@ -131,11 +191,19 @@ namespace CorrectnessTests
outputs.clear();
}
// Multi-process version of Release() where each process frees its own data
void Release(int rank)
{
if (!inPlace) hipFree(outputs[rank]);
hipFree(inputs[rank]);
}
// Creates a dataset by pointing to an existing dataset
// Primarily to allow for testing with different starting byte-alignments
void ExtractSubDataset(size_t const startElement,
size_t const lastElement,
Dataset& subDataset)
Dataset& subDataset,
int const multiProcessRank = -1)
{
ASSERT_LE(startElement, lastElement);
ASSERT_LT(lastElement, numElements);
@@ -150,15 +218,182 @@ namespace CorrectnessTests
subDataset.expected.resize(numDevices);
size_t const byteOffset = (startElement * DataTypeToBytes(dataType));
for (int i = 0; i < numDevices; i++)
if (multiProcessRank != -1)
{
subDataset.inputs[i] = (int8_t *)inputs[i] + byteOffset;
subDataset.outputs[i] = (int8_t *)outputs[i] + byteOffset;
subDataset.expected[i] = (int8_t *)expected[i] + byteOffset;
subDataset.inputs[multiProcessRank] = (int8_t *)inputs[multiProcessRank] + byteOffset;
subDataset.outputs[multiProcessRank] = (int8_t *)outputs[multiProcessRank] + byteOffset;
subDataset.expected[multiProcessRank] = (int8_t *)expected[multiProcessRank] + byteOffset;
}
else
{
for (int i = 0; i < numDevices; i++)
{
subDataset.inputs[i] = (int8_t *)inputs[i] + byteOffset;
subDataset.outputs[i] = (int8_t *)outputs[i] + byteOffset;
subDataset.expected[i] = (int8_t *)expected[i] + byteOffset;
}
}
}
};
class Barrier
{
public:
Barrier(){};
Barrier(int rank, int numRanks, int uniqueId)
{
this->numRanks = numRanks;
std::string uniqueIdString = std::to_string(uniqueId);
mutexName = std::string("mutex").append(uniqueIdString);
turnstile1Name = std::string("turnstile1").append(uniqueIdString);
turnstile2Name = std::string("turnstile2").append(uniqueIdString);
counterName = std::string("counter").append(uniqueIdString);
tinyBarrierName = std::string("tinyBarrier").append(uniqueIdString);
size_t smSize = sizeof(sem_t);
if (rank == 0)
{
InitSemaphore(smSize, mutexName, 1, mutex);
InitSemaphore(smSize, turnstile1Name, 0, turnstile1);
InitSemaphore(smSize, turnstile2Name, 0, turnstile2);
OpenSharedMemoryVariable(sizeof(int), counterName, true, counter);
OpenSharedMemoryVariable(smSize, tinyBarrierName, true, tinyBarrier);
}
else
{
OpenSharedMemoryVariable(smSize, tinyBarrierName, false, tinyBarrier);
OpenSemaphore(smSize, mutexName, mutex);
OpenSemaphore(smSize, turnstile1Name, turnstile1);
OpenSemaphore(smSize, turnstile2Name, turnstile2);
OpenSharedMemoryVariable(sizeof(int), counterName, false, counter);
}
}
void Wait()
{
Part1();
Part2();
}
~Barrier()
{
shm_unlink(mutexName.c_str());
shm_unlink(turnstile1Name.c_str());
shm_unlink(turnstile2Name.c_str());
shm_unlink(counterName.c_str());
shm_unlink(tinyBarrierName.c_str());
}
static void ClearShmFiles(int uniqueId)
{
std::string uniqueIdString = std::to_string(uniqueId);
std::vector<std::string> names;
names.push_back(std::string("mutex").append(uniqueIdString));
names.push_back(std::string("turnstile1").append(uniqueIdString));
names.push_back(std::string("turnstile2").append(uniqueIdString));
names.push_back(std::string("counter").append(uniqueIdString));
names.push_back(std::string("tinyBarrier").append(uniqueIdString));
std::string shmDir = "/dev/shm/";
for (auto it = names.begin(); it != names.end(); it++)
{
struct stat fileStatus;
std::string shmFullPath = shmDir + *it;
// Check if shm file already exists; if so, unlink it
if (stat(shmFullPath.c_str(), &fileStatus) == 0)
{
shm_unlink(it->c_str());
}
}
}
private:
template <typename T>
void OpenSharedMemoryVariable(size_t size, std::string name, bool create, T& val)
{
int protection = PROT_READ | PROT_WRITE;
int visibility = MAP_SHARED;
int fd;
if (create)
{
fd = shm_open(name.c_str(), O_CREAT | O_RDWR, S_IRUSR | S_IWUSR);
ftruncate(fd, size);
}
else
{
do
{
// TODO: Error checking so we don't just infinite loop
fd = shm_open(name.c_str(), O_RDWR, S_IRUSR | S_IWUSR);
} while (fd == -1 && errno == ENOENT);
}
val = (T)mmap(NULL, size, protection, visibility, fd, 0);
close(fd);
}
void InitSemaphore(size_t size, std::string name, int semValue, sem_t*& semaphore)
{
OpenSharedMemoryVariable<sem_t*>(size, name, true, semaphore);
sem_init(semaphore, 1, semValue);
}
void OpenSemaphore(size_t size, std::string name, sem_t*& semaphore)
{
OpenSharedMemoryVariable<sem_t*>(size, name, false, semaphore);
}
void Part1()
{
sem_wait(mutex);
if (++(*counter) == numRanks)
{
sem_post_batch(turnstile1, numRanks);
}
sem_post(mutex);
sem_wait(turnstile1);
}
void Part2()
{
sem_wait(mutex);
if (--(*counter) == 0)
{
sem_post_batch(turnstile2, numRanks);
}
sem_post(mutex);
sem_wait(turnstile2);
}
int sem_post_batch(sem_t*& sem, int n)
{
int ret = 0;
for (int i = 0; i < n; i++)
{
ret = sem_post(sem);
if (ret != 0) break;
}
return ret;
}
int numRanks;
int* counter;
sem_t* mutex;
sem_t* turnstile1;
sem_t* turnstile2;
sem_t* tinyBarrier;
std::string mutexName;
std::string turnstile1Name;
std::string turnstile2Name;
std::string tinyBarrierName;
std::string counterName;
};
typedef std::tuple<ncclRedOp_t /* op */,
ncclDataType_t /* dataType */,
size_t /* numElements */,
@@ -196,7 +431,6 @@ namespace CorrectnessTests
{ncclMax, "max"},
{ncclMin, "min"}
};
std::map<ncclDataType_t, std::string> dataTypeStrings
{
{ncclInt8, "int8"},
@@ -216,7 +450,6 @@ namespace CorrectnessTests
};
};
protected:
// This code is called per test-tuple
void SetUp() override
{
@@ -466,6 +699,265 @@ namespace CorrectnessTests
char* savedEnv[MAX_ENV_TOKENS/2];
};
class MultiProcessCorrectnessTest : public CorrectnessTest
{
protected:
void SetUp() override
{
// Check for NCCL_COMM_ID env variable (otherwise will not init)
if (!getenv("NCCL_COMM_ID"))
{
printf("Must set NCCL_COMM_ID prior to execution\n");
exit(0);
}
// Make the test tuple parameters accessible
std::tie(op, dataType, numElements, numDevices, inPlace, envVals) = GetParam();
envString = 0;
numTokens = 0;
if (strcmp(envVals, "")) {
// enable RCCL env vars testing
setenv("RCCL_TEST_ENV_VARS", "ENABLE", 1);
envString = strdup(envVals);
tokens[numTokens] = strtok(envString, "=, ");
numTokens++;
while (tokens[numTokens-1] != NULL && numTokens < MAX_ENV_TOKENS)
tokens[numTokens++] = strtok(NULL, "=, ");
for (int i = 0; i < numTokens/2; i++) {
char *val = getenv(tokens[i*2]);
if (val)
savedEnv[i] = strdup(val);
else
savedEnv[i] = 0;
setenv(tokens[i*2], tokens[i*2+1], 1);
fprintf(stdout, "[ ] setting environmental variable %s to %s\n", tokens[i*2], getenv(tokens[i*2]));
}
}
comms.resize(numDevices);
streams.resize(numDevices);
}
void TearDown() override
{
// Restore env vars after tests
for (int i = 0; i < numTokens/2; i++) {
if (savedEnv[i]) {
setenv(tokens[i*2], savedEnv[i], 1);
fprintf(stdout, "[ ] restored environmental variable %s to %s\n", tokens[i*2], getenv(tokens[i*2]));
free(savedEnv[i]);
}
else {
unsetenv(tokens[i*2]);
fprintf(stdout, "[ ] removed environmental variable %s\n", tokens[i*2]);
}
}
// Cleanup
unsetenv("RCCL_TEST_ENV_VARS");
free(envString);
}
void SetUpPerProcessHelper(int rank, ncclComm_t& comm, hipStream_t& stream)
{
// Check for NCCL_COMM_ID env variable (otherwise will not init)
if (!getenv("NCCL_COMM_ID"))
{
printf("Must set NCCL_COMM_ID prior to execution\n");
exit(0);
}
// Collect the number of available GPUs
HIP_CALL(hipGetDeviceCount(&numDevicesAvailable));
// Only proceed with testing if there are enough GPUs
if (numDevices > numDevicesAvailable)
{
fprintf(stdout, "[ SKIPPED ] Test requires %d devices (only %d available)\n",
numDevices, numDevicesAvailable);
// Modify the number of devices so that tear-down doesn't occur
// This is temporary until GTEST_SKIP() becomes available
numDevices = 0;
numDevicesAvailable = -1;
return;
}
HIP_CALL(hipSetDevice(rank));
HIP_CALL(hipStreamCreate(&stream));
ncclUniqueId id;
NCCL_CALL(ncclGetUniqueId(&id));
ncclResult_t res;
res = ncclCommInitRank(&comm, numDevices, id, rank); // change to local comm and stream per process
if (res != ncclSuccess)
{
printf("Test failure:%s %d '%s' numRanks:%d\n", __FILE__,__LINE__,ncclGetErrorString(res), numDevices);
ASSERT_EQ(res, hipSuccess);
}
}
// To be called by each process individually
void SetUpPerProcess(int rank, ncclFunc_t const func, ncclComm_t& comm, hipStream_t& stream, Dataset& dataset)
{
SetUpPerProcessHelper(rank, comm, stream);
dataset.Initialize(numDevices, numElements, dataType, inPlace, func, rank);
}
// To be called by each process/rank individually (see GroupCallsMultiProcess)
void SetUpPerProcess(int rank, std::vector<ncclFunc_t> const& func, ncclComm_t& comm, hipStream_t& stream, std::vector<Dataset*>& datasets)
{
SetUpPerProcessHelper(rank, comm, stream);
for (int i = 0; i < datasets.size(); i++)
{
datasets[i]->Initialize(numDevices, numElements, dataType, inPlace, func[i], rank);
}
}
// Clean up per process
void TearDownPerProcess(ncclComm_t& comm, hipStream_t& stream)
{
NCCL_CALL(ncclCommDestroy(comm));
HIP_CALL(hipStreamDestroy(stream));
}
void FillDatasetWithPattern(Dataset& dataset, int rank)
{
int8_t* arrayI1 = (int8_t *)malloc(dataset.NumBytes(ncclInputBuffer));
uint8_t* arrayU1 = (uint8_t *)arrayI1;
int32_t* arrayI4 = (int32_t *)arrayI1;
uint32_t* arrayU4 = (uint32_t *)arrayI1;
int64_t* arrayI8 = (int64_t *)arrayI1;
uint64_t* arrayU8 = (uint64_t *)arrayI1;
float* arrayF4 = (float *)arrayI1;
double* arrayF8 = (double *)arrayI1;
rccl_bfloat16* arrayB2 = (rccl_bfloat16 *)arrayI1;
// NOTE: Currently half-precision float tests are unsupported due to half being supported
// on GPU only and not host
// Fills input data[i][j] with (i + j) % 6
// - Keeping range small to reduce likelihood of overflow
// - Sticking with floating points values that are perfectly representable
for (int j = 0; j < dataset.NumBytes(ncclInputBuffer)/DataTypeToBytes(dataset.dataType); j++)
{
int valueI = (rank + j) % 6;
float valueF = (float)valueI;
switch (dataset.dataType)
{
case ncclInt8: arrayI1[j] = valueI; break;
case ncclUint8: arrayU1[j] = valueI; break;
case ncclInt32: arrayI4[j] = valueI; break;
case ncclUint32: arrayU4[j] = valueI; break;
case ncclInt64: arrayI8[j] = valueI; break;
case ncclUint64: arrayU8[j] = valueI; break;
case ncclFloat32: arrayF4[j] = valueF; break;
case ncclFloat64: arrayF8[j] = valueF; break;
case ncclBfloat16: arrayB2[j] = rccl_bfloat16(valueF); break;
default:
fprintf(stderr, "[ERROR] Unsupported datatype\n");
exit(0);
}
}
HIP_CALL(hipSetDevice(rank));
HIP_CALL(hipMemcpy(dataset.inputs[rank], arrayI1, dataset.NumBytes(ncclInputBuffer), hipMemcpyHostToDevice));
// Fills output data[i][j] with 0 (if not inplace)
if (!dataset.inPlace)
HIP_CALL(hipMemset(dataset.outputs[rank], 0, dataset.NumBytes(ncclOutputBuffer)));
free(arrayI1);
}
void ValidateResults(Dataset const& dataset, int rank, int root = 0) const
{
int8_t* outputI1 = (int8_t *)malloc(dataset.NumBytes(ncclOutputBuffer));
uint8_t* outputU1 = (uint8_t *)outputI1;
int32_t* outputI4 = (int32_t *)outputI1;
uint32_t* outputU4 = (uint32_t *)outputI1;
int64_t* outputI8 = (int64_t *)outputI1;
uint64_t* outputU8 = (uint64_t *)outputI1;
float* outputF4 = (float *)outputI1;
double* outputF8 = (double *)outputI1;
rccl_bfloat16* outputB2 = (rccl_bfloat16 *)outputI1;
bool isMatch = true;
// Loop over each device's output and compare it to the expected output
// (Each collective operation computes its own expected results)
// only output on root rank is valid for gather collective
if (dataset.function == ncclCollGather && rank != root)
return;
HIP_CALL(hipMemcpy(outputI1, dataset.outputs[rank], dataset.NumBytes(ncclOutputBuffer), hipMemcpyDeviceToHost));
int8_t* expectedI1 = (int8_t *)dataset.expected[rank];
uint8_t* expectedU1 = (uint8_t *)expectedI1;
int32_t* expectedI4 = (int32_t *)expectedI1;
uint32_t* expectedU4 = (uint32_t *)expectedI1;
int64_t* expectedI8 = (int64_t *)expectedI1;
uint64_t* expectedU8 = (uint64_t *)expectedI1;
float* expectedF4 = (float *)expectedI1;
double* expectedF8 = (double *)expectedI1;
rccl_bfloat16* expectedB2 = (rccl_bfloat16 *)expectedI1;
for (int j = 0; j < dataset.numElements && isMatch; j++)
{
switch (dataset.dataType)
{
case ncclInt8: isMatch &= (outputI1[j] == expectedI1[j]); break;
case ncclUint8: isMatch &= (outputU1[j] == expectedU1[j]); break;
case ncclInt32: isMatch &= (outputI4[j] == expectedI4[j]); break;
case ncclUint32: isMatch &= (outputU4[j] == expectedU4[j]); break;
case ncclInt64: isMatch &= (outputI8[j] == expectedI8[j]); break;
case ncclUint64: isMatch &= (outputU8[j] == expectedU8[j]); break;
case ncclFloat32: isMatch &= (outputF4[j] == expectedF4[j]); break;
case ncclFloat64: isMatch &= (outputF8[j] == expectedF8[j]); break;
case ncclBfloat16: isMatch &= (outputB2[j] == expectedB2[j]); break;
default:
fprintf(stderr, "[ERROR] Unsupported datatype\n");
exit(0);
}
if (!isMatch)
{
switch (dataset.dataType)
{
case ncclInt8:
printf("Output %d. Expected %d on device %d[%d]\n", outputI1[j], expectedI1[j], rank, j); break;
case ncclUint8:
printf("Output %u. Expected %u on device %d[%d]\n", outputU1[j], expectedU1[j], rank, j); break;
case ncclInt32:
printf("Output %d. Expected %d on device %d[%d]\n", outputI4[j], expectedI4[j], rank, j); break;
case ncclUint32:
printf("Output %u. Expected %u on device %d[%d]\n", outputU4[j], expectedU4[j], rank, j); break;
case ncclInt64:
printf("Output %ld. Expected %ld on device %d[%d]\n", outputI8[j], expectedI8[j], rank, j); break;
case ncclUint64:
printf("Output %lu. Expected %lu on device %d[%d]\n", outputU8[j], expectedU8[j], rank, j); break;
case ncclFloat32:
printf("Output %f. Expected %f on device %d[%d]\n", outputF4[j], expectedF4[j], rank, j); break;
case ncclFloat64:
printf("Output %lf. Expected %lf on device %d[%d]\n", outputF8[j], expectedF8[j], rank, j); break;
case ncclBfloat16:
printf("Output %f. Expected %f on device %d[%d]\n", (float)outputB2[j], (float)expectedB2[j], rank, j); break;
default:
fprintf(stderr, "[ERROR] Unsupported datatype\n");
exit(0);
}
}
}
ASSERT_EQ(isMatch, true);
}
};
std::string GenerateTestNameString(testing::TestParamInfo<MultiProcessCorrectnessTest::ParamType>& info);
}
#endif
projects/rccl/test/test_AllGatherMultiProcess.cpp
+95
View File
@@ -0,0 +1,95 @@
/*************************************************************************
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#include "test_AllGatherMultiProcess.hpp"
namespace CorrectnessTests
{
TEST_P(AllGatherMultiProcessCorrectnessTest, Correctness)
{
Dataset* dataset = (Dataset*)mmap(NULL, sizeof(Dataset), PROT_READ|PROT_WRITE, MAP_SHARED|MAP_ANONYMOUS, -1, 0);
dataset->InitializeRootProcess(numDevices, numElements, dataType, inPlace, ncclCollAllGather);
Barrier::ClearShmFiles(std::atoi(getenv("NCCL_COMM_ID")));
int pid1 = 0;
int pid2 = 0;
int pid3 = 0;
pid1 = fork();
// From this point on, ignore original process as we cannot have it create a HIP context
if (pid1 == 0)
{
pid2 = fork();
if (numDevices > 2)
{
pid3 = fork();
}
if ((pid2 > 0 && pid3 == 0 && numDevices == 2) || (pid2 > 0 && pid3 > 0 && numDevices > 2))
{
// Process 0
TestAllGather(0, *dataset);
if (pid3 > 0)
{
waitpid(pid3, NULL, 0);
}
}
else if ((pid2 == 0 && pid3 == 0 && numDevices == 2) || (pid2 == 0 && pid3 > 0 && numDevices > 2))
{
// Process 1
TestAllGather(1, *dataset);
if (numDevices > 2)
{
waitpid(pid3, NULL, 0);
}
exit(0);
}
else if (pid2 > 0 && pid3 == 0 && numDevices > 2)
{
// Process 2 (available when numDevices > 2)
TestAllGather(2, *dataset);
exit(0);
}
else if (pid2 == 0 && pid3 == 0 && numDevices == 4)
{
// Process 3 (available when numDevices == 4)
TestAllGather(3, *dataset);
exit(0);
}
else
{
exit(0);
}
waitpid(pid2, NULL, 0);
exit(0);
}
waitpid(pid1, NULL, 0);
munmap(dataset, sizeof(Dataset));
}
INSTANTIATE_TEST_SUITE_P(AllGatherMultiProcessCorrectnessSweep,
AllGatherMultiProcessCorrectnessTest,
testing::Combine(
// Reduction operator (not used)
testing::Values(ncclSum),
// Data types
testing::Values(ncclInt8,
ncclUint8,
ncclInt32,
ncclUint32,
ncclInt64,
ncclUint64,
//ncclFloat16,
ncclFloat32,
ncclFloat64,
ncclBfloat16),
// Number of elements
testing::Values(3072, 3145728),
// Number of devices
testing::Values(2,3,4),
// In-place or not
testing::Values(false, true),
testing::Values("")),
CorrectnessTest::PrintToStringParamName());
} // namespace
projects/rccl/test/test_AllGatherMultiProcess.hpp
+69
View File
@@ -0,0 +1,69 @@
/*************************************************************************
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#ifndef TEST_ALLGATHER_MULTI_PROCESS_HPP
#define TEST_ALLGATHER_MULTI_PROCESS_HPP
#include "CorrectnessTest.hpp"
namespace CorrectnessTests
{
class AllGatherMultiProcessCorrectnessTest : public MultiProcessCorrectnessTest
{
public:
static void ComputeExpectedResults(Dataset& dataset, Barrier& barrier, int const rank, int const numDevices)
{
size_t const byteCount = dataset.NumBytes() / dataset.numDevices;
HIP_CALL(hipMemcpy(static_cast<char*>(dataset.expected[0]) + rank * byteCount, (int8_t *)dataset.inputs[rank] + (rank * byteCount),
byteCount, hipMemcpyDeviceToHost));
barrier.Wait();
// Rank 0 sends answer to other ranks
if (rank == 0)
{
for (int i = 0; i < dataset.numDevices; i++)
{
if (i == rank) continue;
memcpy(dataset.expected[i], dataset.expected[0], dataset.NumBytes());
}
}
}
void TestAllGather(int rank, Dataset& dataset)
{
// Prepare input / output / expected results
SetUpPerProcess(rank, ncclCollAllGather, comms[rank], streams[rank], dataset);
if (numDevices > numDevicesAvailable) return;
if (numElements % numDevices != 0) return;
Barrier barrier(rank, numDevices, std::atoi(getenv("NCCL_COMM_ID")));
// Prepare input / output / expected results
FillDatasetWithPattern(dataset, rank);
ComputeExpectedResults(dataset, barrier, rank, numDevices);
size_t const byteCount = dataset.NumBytes() / numDevices;
size_t const sendCount = dataset.numElements / numDevices;
// Launch the reduction (1 process per GPU)
ncclAllGather((int8_t *)dataset.inputs[rank] + (rank * byteCount),
dataset.outputs[rank], sendCount,
dataType, comms[rank], streams[rank]);
// Wait for reduction to complete
HIP_CALL(hipStreamSynchronize(streams[rank]));
// Check results
ValidateResults(dataset, rank);
TearDownPerProcess(comms[rank], streams[rank]);
dataset.Release(rank);
}
};
}
#endif
projects/rccl/test/test_AllReduceMultiProcess.cpp
+96
View File
@@ -0,0 +1,96 @@
/*************************************************************************
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#include "test_AllReduceMultiProcess.hpp"
namespace CorrectnessTests
{
TEST_P(AllReduceMultiProcessCorrectnessTest, Correctness)
{
Dataset* dataset = (Dataset*)mmap(NULL, sizeof(Dataset), PROT_READ|PROT_WRITE, MAP_SHARED|MAP_ANONYMOUS, -1, 0);
dataset->InitializeRootProcess(numDevices, numElements, dataType, inPlace, ncclCollAllReduce);
Barrier::ClearShmFiles(std::atoi(getenv("NCCL_COMM_ID")));
int pid1 = 0;
int pid2 = 0;
int pid3 = 0;
pid1 = fork();
// From this point on, ignore original process as we cannot have it create a HIP context
if (pid1 == 0)
{
pid2 = fork();
if (numDevices > 2)
{
pid3 = fork();
}
if ((pid2 > 0 && pid3 == 0 && numDevices == 2) || (pid2 > 0 && pid3 > 0 && numDevices > 2))
{
// Process 0
TestAllReduce(0, *dataset);
if (pid3 > 0)
{
waitpid(pid3, NULL, 0);
}
}
else if ((pid2 == 0 && pid3 == 0 && numDevices == 2) || (pid2 == 0 && pid3 > 0 && numDevices > 2))
{
// Process 1
TestAllReduce(1, *dataset);
if (numDevices > 2)
{
waitpid(pid3, NULL, 0);
}
exit(0);
}
else if (pid2 > 0 && pid3 == 0 && numDevices > 2)
{
// Process 2 (available when numDevices > 2)
TestAllReduce(2, *dataset);
exit(0);
}
else if (pid2 == 0 && pid3 == 0 && numDevices == 4)
{
// Process 3 (available when numDevices == 4)
TestAllReduce(3, *dataset);
exit(0);
}
else
{
exit(0);
}
waitpid(pid2, NULL, 0);
exit(0);
}
waitpid(pid1, NULL, 0);
munmap(dataset, sizeof(Dataset));
}
INSTANTIATE_TEST_SUITE_P(AllReduceMultiProcessCorrectnessSweep,
AllReduceMultiProcessCorrectnessTest,
testing::Combine(
// Reduction operator
testing::Values(ncclSum, ncclProd, ncclMax, ncclMin),
// Data types
testing::Values(ncclInt8,
ncclUint8,
ncclInt32,
ncclUint32,
ncclInt64,
ncclUint64,
//ncclFloat16,
ncclFloat32,
ncclFloat64,
ncclBfloat16),
// Number of elements
testing::Values(1024, 1048576),
// Number of devices
testing::Values(2,3,4),
// In-place or not
testing::Values(false, true),
testing::Values("")),
CorrectnessTest::PrintToStringParamName());
} // namespace
projects/rccl/test/test_AllReduceMultiProcess.hpp
+105
View File
@@ -0,0 +1,105 @@
/*************************************************************************
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#ifndef TEST_ALLREDUCE_MULTI_PROCESS_HPP
#define TEST_ALLREDUCE_MULTI_PROCESS_HPP
#include "CorrectnessTest.hpp"
namespace CorrectnessTests
{
class AllReduceMultiProcessCorrectnessTest : public MultiProcessCorrectnessTest
{
public:
static void ComputeExpectedResults(Dataset& dataset, Barrier& barrier, ncclRedOp_t const op, int const rank)
{
// Copy all inputs to expected arrays temporarily to perform reduction on host
HIP_CALL(hipMemcpy(dataset.expected[rank], dataset.inputs[rank],
dataset.NumBytes(), hipMemcpyDeviceToHost));
barrier.Wait();
// Allocate temporary host array to accumulate results
int8_t* resultI1 = (int8_t *)malloc(dataset.NumBytes());
uint8_t* resultU1 = (uint8_t *)resultI1;
int32_t* resultI4 = (int32_t *)resultI1;
uint32_t* resultU4 = (uint32_t *)resultI1;
int64_t* resultI8 = (int64_t *)resultI1;
uint64_t* resultU8 = (uint64_t *)resultI1;
float* resultF4 = (float *)resultI1;
double* resultF8 = (double *)resultI1;
rccl_bfloat16* resultB2 = (rccl_bfloat16 *)resultI1;
// Initialize the result with the first device's array
memcpy(resultI1, dataset.expected[0], dataset.NumBytes());
barrier.Wait();
// Perform reduction
for (int i = 1; i < dataset.numDevices; i++)
{
int8_t* arrayI1 = (int8_t *)dataset.expected[i];
uint8_t* arrayU1 = (uint8_t *)arrayI1;
int32_t* arrayI4 = (int32_t *)arrayI1;
uint32_t* arrayU4 = (uint32_t *)arrayI1;
int64_t* arrayI8 = (int64_t *)arrayI1;
uint64_t* arrayU8 = (uint64_t *)arrayI1;
float* arrayF4 = (float *)arrayI1;
double* arrayF8 = (double *)arrayI1;
rccl_bfloat16* arrayB2 = (rccl_bfloat16 *)arrayI1;
for (int j = 0; j < dataset.numElements; j++)
{
switch (dataset.dataType)
{
case ncclInt8: resultI1[j] = ReduceOp(op, resultI1[j], arrayI1[j]); break;
case ncclUint8: resultU1[j] = ReduceOp(op, resultU1[j], arrayU1[j]); break;
case ncclInt32: resultI4[j] = ReduceOp(op, resultI4[j], arrayI4[j]); break;
case ncclUint32: resultU4[j] = ReduceOp(op, resultU4[j], arrayU4[j]); break;
case ncclInt64: resultI8[j] = ReduceOp(op, resultI8[j], arrayI8[j]); break;
case ncclUint64: resultU8[j] = ReduceOp(op, resultU8[j], arrayU8[j]); break;
case ncclFloat32: resultF4[j] = ReduceOp(op, resultF4[j], arrayF4[j]); break;
case ncclFloat64: resultF8[j] = ReduceOp(op, resultF8[j], arrayF8[j]); break;
case ncclBfloat16: resultB2[j] = ReduceOp(op, resultB2[j], arrayB2[j]); break;
default:
fprintf(stderr, "[ERROR] Unsupported datatype\n");
exit(0);
}
}
}
barrier.Wait();
// Copy results into expected array
memcpy(dataset.expected[rank], resultI1, dataset.NumBytes());
free(resultI1);
}
void TestAllReduce(int rank, Dataset& dataset)
{
SetUpPerProcess(rank, ncclCollAllReduce, comms[rank], streams[rank], dataset);
if (numDevices > numDevicesAvailable) return;
Barrier barrier(rank, numDevices, std::atoi(getenv("NCCL_COMM_ID")));
// Prepare input / output / expected results
FillDatasetWithPattern(dataset, rank);
ComputeExpectedResults(dataset, barrier, op, rank);
// Launch the reduction
ncclAllReduce(dataset.inputs[rank], dataset.outputs[rank],
numElements, dataType, op, comms[rank], streams[rank]);
// Wait for reduction to complete
HIP_CALL(hipStreamSynchronize(streams[rank]));
// Check results
ValidateResults(dataset, rank);
TearDownPerProcess(comms[rank], streams[rank]);
dataset.Release(rank);
}
};
}
#endif
projects/rccl/test/test_AllToAllMultiProcess.cpp
+95
View File
@@ -0,0 +1,95 @@
/*************************************************************************
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#include "test_AllToAllMultiProcess.hpp"
namespace CorrectnessTests
{
TEST_P(AllToAllMultiProcessCorrectnessTest, Correctness)
{
Dataset* dataset = (Dataset*)mmap(NULL, sizeof(Dataset), PROT_READ|PROT_WRITE, MAP_SHARED|MAP_ANONYMOUS, -1, 0);
dataset->InitializeRootProcess(numDevices, numElements, dataType, inPlace, ncclCollAllToAll);
int pid1 = 0;
int pid2 = 0;
int pid3 = 0;
pid1 = fork();
// From this point on, ignore original process as we cannot have it create a HIP context
if (pid1 == 0)
{
pid2 = fork();
if (numDevices > 2)
{
pid3 = fork();
}
if ((pid2 > 0 && pid3 == 0 && numDevices == 2) || (pid2 > 0 && pid3 > 0 && numDevices > 2))
{
// Process 0
TestAllToAll(0, *dataset);
if (pid3 > 0)
{
waitpid(pid3, NULL, 0);
}
}
else if ((pid2 == 0 && pid3 == 0 && numDevices == 2) || (pid2 == 0 && pid3 > 0 && numDevices > 2))
{
// Process 1
TestAllToAll(1, *dataset);
if (numDevices > 2)
{
waitpid(pid3, NULL, 0);
}
exit(0);
}
else if (pid2 > 0 && pid3 == 0 && numDevices > 2)
{
// Process 2 (available when numDevices > 2)
TestAllToAll(2, *dataset);
exit(0);
}
else if (pid2 == 0 && pid3 == 0 && numDevices == 4)
{
// Process 3 (available when numDevices == 4)
TestAllToAll(3, *dataset);
exit(0);
}
else
{
exit(0);
}
waitpid(pid2, NULL, 0);
exit(0);
}
waitpid(pid1, NULL, 0);
munmap(dataset, sizeof(Dataset));
}
INSTANTIATE_TEST_SUITE_P(AllToAllMultiProcessCorrectnessSweep,
AllToAllMultiProcessCorrectnessTest,
testing::Combine(
// Reduction operator is not used
testing::Values(ncclSum),
// Data types
testing::Values(ncclInt8,
ncclUint8,
ncclInt32,
ncclUint32,
ncclInt64,
ncclUint64,
//ncclFloat16,
ncclFloat32,
ncclFloat64,
ncclBfloat16),
// Number of elements
testing::Values(1024, 1048576),
// Number of devices
testing::Values(2,3,4),
// In-place or not
testing::Values(false),
testing::Values("RCCL_ALLTOALL_KERNEL_DISABLE=0", "RCCL_ALLTOALL_KERNEL_DISABLE=1")),
CorrectnessTest::PrintToStringParamName());
} // namespace
projects/rccl/test/test_AllToAllMultiProcess.hpp
+53
View File
@@ -0,0 +1,53 @@
/*************************************************************************
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#ifndef TEST_ALLTOALL_MULTI_PROCESS_HPP
#define TEST_ALLTOALL_MULTI_PROCESS_HPP
#include "CorrectnessTest.hpp"
namespace CorrectnessTests
{
class AllToAllMultiProcessCorrectnessTest : public MultiProcessCorrectnessTest
{
public:
static void ComputeExpectedResults(Dataset& dataset, int const rank)
{
for (int i = 0; i < dataset.numDevices; i++)
{
HIP_CALL(hipMemcpy((int8_t *)dataset.expected[i]+dataset.NumBytes()*rank, (int8_t *)dataset.inputs[rank]+dataset.NumBytes()*i,
dataset.NumBytes(), hipMemcpyDeviceToHost));
}
}
void TestAllToAll(int rank, Dataset& dataset)
{
SetUpPerProcess(rank, ncclCollAllToAll, comms[rank], streams[rank], dataset);
if (numDevices > numDevicesAvailable) return;
// Prepare input / output / expected results
FillDatasetWithPattern(dataset, rank);
ComputeExpectedResults(dataset, rank);
// Launch the reduction
ncclAllToAll(dataset.inputs[rank],
dataset.outputs[rank],
numElements, dataType,
comms[rank], streams[rank]);
// Wait for reduction to complete
HIP_CALL(hipStreamSynchronize(streams[rank]));
// Check results
ValidateResults(dataset, rank);
TearDownPerProcess(comms[rank], streams[rank]);
dataset.Release(rank);
}
};
}
#endif
projects/rccl/test/test_BroadcastMultiProcess.cpp
+103
View File
@@ -0,0 +1,103 @@
/*************************************************************************
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#include "test_BroadcastMultiProcess.hpp"
#include <stdio.h>
#include <unistd.h>
#include <sys/mman.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <iostream>
namespace CorrectnessTests
{
TEST_P(BroadcastMultiProcessCorrectnessTest, Correctness)
{
Dataset* dataset = (Dataset*)mmap(NULL, sizeof(Dataset), PROT_READ|PROT_WRITE, MAP_SHARED|MAP_ANONYMOUS, -1, 0);
dataset->InitializeRootProcess(numDevices, numElements, dataType, inPlace, ncclCollBroadcast);
Barrier::ClearShmFiles(std::atoi(getenv("NCCL_COMM_ID")));
int pid1 = 0;
int pid2 = 0;
int pid3 = 0;
pid1 = fork();
// From this point on, ignore original process as we cannot have it create a HIP context
if (pid1 == 0)
{
pid2 = fork();
if (numDevices > 2)
{
pid3 = fork();
}
if ((pid2 > 0 && pid3 == 0 && numDevices == 2) || (pid2 > 0 && pid3 > 0 && numDevices > 2))
{
// Process 0
TestBroadcast(0, *dataset);
if (pid3 > 0)
{
waitpid(pid3, NULL, 0);
}
}
else if ((pid2 == 0 && pid3 == 0 && numDevices == 2) || (pid2 == 0 && pid3 > 0 && numDevices > 2))
{
// Process 1
TestBroadcast(1, *dataset);
if (numDevices > 2)
{
waitpid(pid3, NULL, 0);
}
exit(0);
}
else if (pid2 > 0 && pid3 == 0 && numDevices > 2)
{
// Process 2 (available when numDevices > 2)
TestBroadcast(2, *dataset);
exit(0);
}
else if (pid2 == 0 && pid3 == 0 && numDevices == 4)
{
// Process 3 (available when numDevices == 4)
TestBroadcast(3, *dataset);
exit(0);
}
else
{
exit(0);
}
waitpid(pid2, NULL, 0);
exit(0);
}
waitpid(pid1, NULL, 0);
munmap(dataset, sizeof(Dataset));
}
INSTANTIATE_TEST_SUITE_P(BroadcastMultiProcessCorrectnessSweep,
BroadcastMultiProcessCorrectnessTest,
testing::Combine(
// Reduction operator is not used
testing::Values(ncclSum),
// Data types
testing::Values(ncclInt8,
ncclUint8,
ncclInt32,
ncclUint32,
ncclInt64,
ncclUint64,
//ncclFloat16,
ncclFloat32,
ncclFloat64,
ncclBfloat16),
// Number of elements
testing::Values(1024, 1048576),
// Number of devices
testing::Values(2,3,4),
// In-place or not
testing::Values(false, true),
testing::Values("")),
CorrectnessTest::PrintToStringParamName());
} // namespace
projects/rccl/test/test_BroadcastMultiProcess.hpp
+68
View File
@@ -0,0 +1,68 @@
/*************************************************************************
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#ifndef TEST_BROADCAST_MULTI_PROCESS_HPP
#define TEST_BROADCAST_MULTI_PROCESS_HPP
#include "CorrectnessTest.hpp"
namespace CorrectnessTests
{
class BroadcastMultiProcessCorrectnessTest : public MultiProcessCorrectnessTest
{
public:
static void ComputeExpectedResults(Dataset& dataset, int const root, int const rank)
{
// Root has the answer; share it via host memcpy's
if (rank == root)
{
HIP_CALL(hipMemcpy(dataset.expected[rank], dataset.inputs[rank],
dataset.NumBytes(), hipMemcpyDeviceToHost));
for (int i = 0; i < dataset.numDevices; i++)
{
if (i == rank) continue;
memcpy(dataset.expected[i], dataset.expected[root], dataset.NumBytes());
}
}
}
void TestBroadcast(int rank, Dataset& dataset)
{
SetUpPerProcess(rank, ncclCollBroadcast, comms[rank], streams[rank], dataset);
if (numDevices > numDevicesAvailable) return;
Barrier barrier(rank, numDevices, std::atoi(getenv("NCCL_COMM_ID")));
// Test each possible root
for (int root = 0; root < numDevices; root++)
{
// Prepare input / output / expected results
FillDatasetWithPattern(dataset, rank);
ComputeExpectedResults(dataset, root, rank);
// Launch the reduction (1 process per GPU)
ncclResult_t res = ncclBroadcast(dataset.inputs[rank],
dataset.outputs[rank],
numElements, dataType,
root, comms[rank], streams[rank]);
// Wait for reduction to complete
HIP_CALL(hipStreamSynchronize(streams[rank]));
// Check results
ValidateResults(dataset, rank);
// Ensure all processes have finished current iteration before proceeding
barrier.Wait();
}
TearDownPerProcess(comms[rank], streams[rank]);
dataset.Release(rank);
}
};
}
#endif
projects/rccl/test/test_CombinedCalls.cpp
+2 -2
View File
@@ -115,6 +115,6 @@ namespace CorrectnessTests
testing::Values(2,3,4,5,6,7,8),
// In-place or not
testing::Values(false),
testing::Values("RCCL_ALLTOALL_KERNEL_DISABLE=1", "RCCL_ENABLE_CLIQUE=0", "RCCL_ENABLE_CLIQUE=1")),
CorrectnessTest::PrintToStringParamName());
testing::Values("RCCL_ENABLE_CLIQUE=0", "RCCL_ENABLE_CLIQUE=1", "RCCL_ALLTOALL_KERNEL_DISABLE=0", "RCCL_ALLTOALL_KERNEL_DISABLE=1")),
CorrectnessTest::PrintToStringParamName());
} // namespace
projects/rccl/test/test_CombinedCallsMultiProcess.cpp
+112
View File
@@ -0,0 +1,112 @@
/*************************************************************************
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#include "test_CombinedCallsMultiProcess.hpp"
namespace CorrectnessTests
{
TEST_P(CombinedCallsMultiProcessCorrectnessTest, Correctness)
{
// Important: Make sure the order of ncclFunc_t's here match the order of ncclFunc_ts
// as they appear in TestCombinedCalls()
std::vector<ncclFunc_t> ncclFuncs;
ncclFuncs.push_back(ncclCollAllGather);
ncclFuncs.push_back(ncclCollAllReduce);
ncclFuncs.push_back(ncclCollBroadcast);
ncclFuncs.push_back(ncclCollReduce);
ncclFuncs.push_back(ncclCollReduceScatter);
// Create multiple datasets for combined operation
std::vector<Dataset*> datasets(ncclFuncs.size());
for (int i = 0; i < datasets.size(); i++)
{
datasets[i] = (Dataset*)mmap(NULL, sizeof(Dataset), PROT_READ|PROT_WRITE, MAP_SHARED|MAP_ANONYMOUS, -1, 0);
datasets[i]->InitializeRootProcess(numDevices, numElements, dataType, inPlace, ncclFuncs[i]);
}
Barrier::ClearShmFiles(std::atoi(getenv("NCCL_COMM_ID")));
int pid1 = 0;
int pid2 = 0;
int pid3 = 0;
pid1 = fork();
// From this point on, ignore original process as we cannot have it create a HIP context
if (pid1 == 0)
{
pid2 = fork();
if (numDevices > 2)
{
pid3 = fork();
}
if ((pid2 > 0 && pid3 == 0 && numDevices == 2) || (pid2 > 0 && pid3 > 0 && numDevices > 2))
{
// Process 0
TestCombinedCalls(0, datasets, ncclFuncs);
if (pid3 > 0)
{
waitpid(pid3, NULL, 0);
}
}
else if ((pid2 == 0 && pid3 == 0 && numDevices == 2) || (pid2 == 0 && pid3 > 0 && numDevices > 2))
{
// Process 1
TestCombinedCalls(1, datasets, ncclFuncs);
if (numDevices > 2)
{
waitpid(pid3, NULL, 0);
}
exit(0);
}
else if (pid2 > 0 && pid3 == 0 && numDevices > 2)
{
// Process 2 (available when numDevices > 2)
TestCombinedCalls(2, datasets, ncclFuncs);
exit(0);
}
else if (pid2 == 0 && pid3 == 0 && numDevices == 4)
{
// Process 3 (available when numDevices == 4)
TestCombinedCalls(3, datasets, ncclFuncs);
exit(0);
}
else
{
exit(0);
}
waitpid(pid2, NULL, 0);
exit(0);
}
waitpid(pid1, NULL, 0);
for (int i = 0; i < datasets.size(); i++)
{
munmap(datasets[i], sizeof(Dataset));
}
}
INSTANTIATE_TEST_SUITE_P(CombinedCallsMultiProcessCorrectnessSweep,
CombinedCallsMultiProcessCorrectnessTest,
testing::Combine(
// Reduction operator (not used)
testing::Values(ncclSum),
// Data types
testing::Values(ncclInt8,
ncclUint8,
ncclInt32,
ncclUint32,
ncclInt64,
ncclUint64,
//ncclFloat16,
ncclFloat32,
ncclFloat64,
ncclBfloat16),
// Number of elements
testing::Values(3072, 3145728),
// Number of devices
testing::Values(2,3,4),
// In-place or not
testing::Values(false, true),
testing::Values("")),
CorrectnessTest::PrintToStringParamName());
} // namespace
projects/rccl/test/test_CombinedCallsMultiProcess.hpp
+78
View File
@@ -0,0 +1,78 @@
/*************************************************************************
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#ifndef TEST_COMBINEDCALLS_MULTI_PROCESS_HPP
#define TEST_COMBINEDCALLS_MULTI_PROCESS_HPP
#include "CorrectnessTest.hpp"
#include "test_AllGatherMultiProcess.hpp"
#include "test_AllReduceMultiProcess.hpp"
#include "test_BroadcastMultiProcess.hpp"
#include "test_ReduceMultiProcess.hpp"
#include "test_ReduceScatterMultiProcess.hpp"
namespace CorrectnessTests
{
class CombinedCallsMultiProcessCorrectnessTest : public MultiProcessCorrectnessTest
{
public:
void TestCombinedCalls(int rank, std::vector<Dataset*>& datasets, std::vector<ncclFunc_t> const& funcs)
{
SetUpPerProcess(rank, funcs, comms[rank], streams[rank], datasets);
if (numDevices > numDevicesAvailable) return;
Barrier barrier(rank, numDevices, std::atoi(getenv("NCCL_COMM_ID")));
// Compute expected results for each dataset in combined
int const root = 0;
AllGatherMultiProcessCorrectnessTest::ComputeExpectedResults(*datasets[0], barrier, rank, numDevices);
AllReduceMultiProcessCorrectnessTest::ComputeExpectedResults(*datasets[1], barrier, op, rank);
BroadcastMultiProcessCorrectnessTest::ComputeExpectedResults(*datasets[2], root, rank);
ReduceMultiProcessCorrectnessTest::ComputeExpectedResults(*datasets[3], barrier, op, root, rank);
ReduceScatterMultiProcessCorrectnessTest::ComputeExpectedResults(*datasets[4], barrier, op, rank);
size_t const byteCount = datasets[0]->NumBytes() / numDevices;
size_t const elemCount = numElements / numDevices;
ncclAllGather((int8_t *)datasets[0]->inputs[rank] + (rank * byteCount),
datasets[0]->outputs[rank], elemCount,
dataType, comms[rank], streams[rank]);
ncclAllReduce(datasets[1]->inputs[rank], datasets[1]->outputs[rank],
numElements, dataType, op, comms[rank], streams[rank]);
ncclBroadcast(datasets[2]->inputs[rank],
datasets[2]->outputs[rank],
numElements, dataType,
root, comms[rank], streams[rank]);
ncclReduce(datasets[3]->inputs[rank],
datasets[3]->outputs[rank],
numElements, dataType, op,
root, comms[rank], streams[rank]);
ncclReduceScatter(datasets[4]->inputs[rank],
(int8_t *)datasets[4]->outputs[rank] + (rank * byteCount),
elemCount, dataType, op,
comms[rank], streams[rank]);
// Wait for reduction to complete
HIP_CALL(hipStreamSynchronize(streams[rank]));
// Check results for each collective in the combined
for (int i = 0; i < 5; i++)
{
ValidateResults(*datasets[i], rank);
barrier.Wait();
datasets[i]->Release(rank);
}
}
};
}
#endif
projects/rccl/test/test_GatherMultiProcess.cpp
+96
View File
@@ -0,0 +1,96 @@
/*************************************************************************
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#include "test_GatherMultiProcess.hpp"
namespace CorrectnessTests
{
TEST_P(GatherMultiProcessCorrectnessTest, Correctness)
{
Dataset* dataset = (Dataset*)mmap(NULL, sizeof(Dataset), PROT_READ|PROT_WRITE, MAP_SHARED|MAP_ANONYMOUS, -1, 0);
dataset->InitializeRootProcess(numDevices, numElements, dataType, inPlace, ncclCollGather);
Barrier::ClearShmFiles(std::atoi(getenv("NCCL_COMM_ID")));
int pid1 = 0;
int pid2 = 0;
int pid3 = 0;
pid1 = fork();
// From this point on, ignore original process as we cannot have it create a HIP context
if (pid1 == 0)
{
pid2 = fork();
if (numDevices > 2)
{
pid3 = fork();
}
if ((pid2 > 0 && pid3 == 0 && numDevices == 2) || (pid2 > 0 && pid3 > 0 && numDevices > 2))
{
// Process 0
TestGather(0, *dataset);
if (pid3 > 0)
{
waitpid(pid3, NULL, 0);
}
}
else if ((pid2 == 0 && pid3 == 0 && numDevices == 2) || (pid2 == 0 && pid3 > 0 && numDevices > 2))
{
// Process 1
TestGather(1, *dataset);
if (numDevices > 2)
{
waitpid(pid3, NULL, 0);
}
exit(0);
}
else if (pid2 > 0 && pid3 == 0 && numDevices > 2)
{
// Process 2 (available when numDevices > 2)
TestGather(2, *dataset);
exit(0);
}
else if (pid2 == 0 && pid3 == 0 && numDevices == 4)
{
// Process 3 (available when numDevices == 4)
TestGather(3, *dataset);
exit(0);
}
else
{
exit(0);
}
waitpid(pid2, NULL, 0);
exit(0);
}
waitpid(pid1, NULL, 0);
munmap(dataset, sizeof(Dataset));
}
INSTANTIATE_TEST_SUITE_P(GatherMultiProcessCorrectnessSweep,
GatherMultiProcessCorrectnessTest,
testing::Combine(
// Reduction operator is not used
testing::Values(ncclSum),
// Data types
testing::Values(ncclInt8,
ncclUint8,
ncclInt32,
ncclUint32,
ncclInt64,
ncclUint64,
//ncclFloat16,
ncclFloat32,
ncclFloat64,
ncclBfloat16),
// Number of elements
testing::Values(1024, 1048576),
// Number of devices
testing::Values(2,3,4),
// In-place or not
testing::Values(false),
testing::Values("RCCL_ALLTOALL_KERNEL_DISABLE=0", "RCCL_ALLTOALL_KERNEL_DISABLE=1")),
CorrectnessTest::PrintToStringParamName());
} // namespace
projects/rccl/test/test_GatherMultiProcess.hpp
+59
View File
@@ -0,0 +1,59 @@
/*************************************************************************
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#ifndef TEST_GATHER_MULTI_PROCESS_HPP
#define TEST_GATHER_MULTI_PROCESS_HPP
#include "CorrectnessTest.hpp"
namespace CorrectnessTests
{
class GatherMultiProcessCorrectnessTest : public MultiProcessCorrectnessTest
{
public:
static void ComputeExpectedResults(Dataset& dataset, int const root, int const rank)
{
HIP_CALL(hipMemcpy((int8_t *)dataset.expected[root]+dataset.NumBytes()*rank, dataset.inputs[rank],
dataset.NumBytes(), hipMemcpyDeviceToHost));
}
void TestGather(int rank, Dataset& dataset)
{
SetUpPerProcess(rank, ncclCollGather, comms[rank], streams[rank], dataset);
if (numDevices > numDevicesAvailable) return;
Barrier barrier(rank, numDevices, std::atoi(getenv("NCCL_COMM_ID")));
// Test each possible root
for (int root = 0; root < numDevices; root++)
{
// Prepare input / output / expected results
FillDatasetWithPattern(dataset, rank);
ComputeExpectedResults(dataset, root, rank);
// Launch the reduction (1 process per GPU)
ncclGather(dataset.inputs[rank],
dataset.outputs[rank],
numElements, dataType,
root, comms[rank], streams[rank]);
// Wait for reduction to complete
HIP_CALL(hipStreamSynchronize(streams[rank]));
// Check results
ValidateResults(dataset, rank, root);
// Ensure all processes have finished current iteration before proceeding
barrier.Wait();
}
TearDownPerProcess(comms[rank], streams[rank]);
dataset.Release(rank);
}
};
}
#endif
projects/rccl/test/test_GroupCallsMultiProcess.cpp
+127
View File
@@ -0,0 +1,127 @@
/*************************************************************************
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#include "test_GroupCallsMultiProcess.hpp"
namespace CorrectnessTests
{
TEST_P(GroupCallsMultiProcessCorrectnessTest, Correctness)
{
// Important: Make sure the order of ncclFunc_t's here match the order of ncclFunc_ts
// as they appear in TestGroupCalls()
std::vector<ncclFunc_t> ncclFuncs;
ncclFuncs.push_back(ncclCollAllGather);
ncclFuncs.push_back(ncclCollAllReduce);
ncclFuncs.push_back(ncclCollBroadcast);
ncclFuncs.push_back(ncclCollReduce);
ncclFuncs.push_back(ncclCollReduceScatter);
// Create multiple datasets for combined operation
std::vector<Dataset*> datasets(ncclFuncs.size());
for (int i = 0; i < datasets.size(); i++)
{
datasets[i] = (Dataset*)mmap(NULL, sizeof(Dataset), PROT_READ|PROT_WRITE, MAP_SHARED|MAP_ANONYMOUS, -1, 0);
datasets[i]->InitializeRootProcess(numDevices, numElements, dataType, inPlace, ncclFuncs[i]);
}
Barrier::ClearShmFiles(std::atoi(getenv("NCCL_COMM_ID")));
int pid1 = 0;
int pid2 = 0;
int pid3 = 0;
pid1 = fork();
// From this point on, ignore original process as we cannot have it create a HIP context
if (pid1 == 0)
{
pid2 = fork();
if (numDevices > 4)
{
pid3 = fork();
}
if ((pid2 > 0 && pid3 == 0 && numDevices == 4) || (pid2 > 0 && pid3 > 0 && numDevices > 4))
{
// Process 0
std::vector<int> ranks;
ranks.push_back(0);
ranks.push_back(1);
TestGroupCalls(0, ranks, datasets, ncclFuncs);
if (pid3 > 0)
{
waitpid(pid3, NULL, 0);
}
}
else if ((pid2 == 0 && pid3 == 0 && numDevices == 4) || (pid2 == 0 && pid3 > 0 && numDevices > 4))
{
// Process 1
std::vector<int> ranks;
ranks.push_back(2);
ranks.push_back(3);
TestGroupCalls(1, ranks, datasets, ncclFuncs);
if (pid3 > 0)
{
waitpid(pid3, NULL, 0);
}
exit(0);
}
else if (pid2 > 0 && pid3 == 0 && numDevices == 8)
{
// Process 2 (available when numDevices == 8)
std::vector<int> ranks;
ranks.push_back(4);
ranks.push_back(5);
TestGroupCalls(2, ranks, datasets, ncclFuncs);
exit(0);
}
else if (pid2 == 0 && pid3 == 0 && numDevices == 8)
{
// Process 3 (available when numDevices == 8)
std::vector<int> ranks;
ranks.push_back(6);
ranks.push_back(7);
TestGroupCalls(3, ranks, datasets, ncclFuncs);
exit(0);
}
else
{
exit(0);
}
waitpid(pid2, NULL, 0);
exit(0);
}
waitpid(pid1, NULL, 0);
for (int i = 0; i < datasets.size(); i++)
{
munmap(datasets[i], sizeof(Dataset));
}
}
INSTANTIATE_TEST_SUITE_P(GroupCallsMultiProcessCorrectnessSweep,
GroupCallsMultiProcessCorrectnessTest,
testing::Combine(
// Reduction operator (not used)
testing::Values(ncclSum),
// Data types
testing::Values(ncclInt8,
ncclUint8,
ncclInt32,
ncclUint32,
ncclInt64,
ncclUint64,
//ncclFloat16,
ncclFloat32,
ncclFloat64,
ncclBfloat16),
// Number of elements
testing::Values(3072, 3145728),
// Number of devices
testing::Values(4),
// In-place or not
testing::Values(false, true),
testing::Values("")),
CorrectnessTest::PrintToStringParamName());
} // namespace
projects/rccl/test/test_GroupCallsMultiProcess.hpp
+128
View File
@@ -0,0 +1,128 @@
/*************************************************************************
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#ifndef TEST_GROUPCALLS_MULTI_PROCESS_HPP
#define TEST_GROUPCALLS_MULTI_PROCESS_HPP
#include "CorrectnessTest.hpp"
#include "test_AllGatherMultiProcess.hpp"
#include "test_AllReduceMultiProcess.hpp"
#include "test_BroadcastMultiProcess.hpp"
#include "test_ReduceMultiProcess.hpp"
#include "test_ReduceScatterMultiProcess.hpp"
#include <string>
namespace CorrectnessTests
{
class GroupCallsMultiProcessCorrectnessTest : public MultiProcessCorrectnessTest
{
public:
void TestGroupCalls(int process, std::vector<int> const& ranks, std::vector<Dataset*>& datasets, std::vector<ncclFunc_t> const& funcs)
{
if (numDevices > numDevicesAvailable) return;
for (int i = 0; i < ranks.size(); i++)
{
SetUpPerProcess(ranks[i], funcs, comms[ranks[i]], streams[ranks[i]], datasets);
}
int numProcesses = numDevices / ranks.size();
Barrier barrier(process, numProcesses, std::atoi(getenv("NCCL_COMM_ID")));
int const root = 0;
for (int i = 0; i < ranks.size(); i++)
{
AllGatherMultiProcessCorrectnessTest::ComputeExpectedResults(*datasets[0], barrier, numDevices, ranks[i]);
AllReduceMultiProcessCorrectnessTest::ComputeExpectedResults(*datasets[1], barrier, op, ranks[i]);
BroadcastMultiProcessCorrectnessTest::ComputeExpectedResults(*datasets[2], root, ranks[i]);
ReduceMultiProcessCorrectnessTest::ComputeExpectedResults(*datasets[3], barrier, op, root, ranks[i]);
ReduceScatterMultiProcessCorrectnessTest::ComputeExpectedResults(*datasets[4], barrier, op, ranks[i]);
}
barrier.Wait();
ncclGroupStart();
// AllGather
size_t const byteCount = datasets[0]->NumBytes() / numDevices;
size_t const elemCount = numElements / numDevices;
for (int i = 0; i < ranks.size(); i++)
{
int rank = ranks[i];
ncclAllGather((int8_t *)datasets[0]->inputs[rank] + (rank * byteCount),
datasets[0]->outputs[rank], elemCount,
dataType, comms[rank], streams[rank]);
}
// AllReduce
for (int i = 0; i < ranks.size(); i++)
{
int rank = ranks[i];
ncclAllReduce(datasets[1]->inputs[rank], datasets[1]->outputs[rank],
numElements, dataType, op, comms[rank], streams[rank]);
}
// Broadcast
for (int i = 0; i < ranks.size(); i++)
{
int rank = ranks[i];
ncclBroadcast(datasets[2]->inputs[rank],
datasets[2]->outputs[rank],
numElements, dataType,
root, comms[rank], streams[rank]);
}
// Reduce
for (int i = 0; i < ranks.size(); i++)
{
int rank = ranks[i];
ncclReduce(datasets[3]->inputs[rank],
datasets[3]->outputs[rank],
numElements, dataType, op,
root, comms[rank], streams[rank]);
}
// ReduceScatter
for (int i = 0; i < ranks.size(); i++)
{
int rank = ranks[i];
ncclReduceScatter(datasets[4]->inputs[rank],
(int8_t *)datasets[4]->outputs[rank] + (i * byteCount),
elemCount, dataType, op,
comms[rank], streams[rank]);
}
// Signal end of group call
ncclGroupEnd();
for (int i = 0; i < ranks.size(); i++)
{
HIP_CALL(hipSetDevice(ranks[i]));
HIP_CALL(hipStreamSynchronize(streams[ranks[i]]));
}
for (int i = 0; i < funcs.size(); i++)
{
for (int j = 0; j < ranks.size(); j++)
{
ValidateResults(*datasets[i], ranks[j]);
}
barrier.Wait();
for (int j = 0; j < ranks.size(); j++)
{
datasets[i]->Release(ranks[j]);
}
}
for (int i = 0; i < ranks.size(); i++)
{
TearDownPerProcess(comms[ranks[i]], streams[ranks[i]]);
}
}
};
}
#endif
projects/rccl/test/test_ReduceMultiProcess.cpp
+96
View File
@@ -0,0 +1,96 @@
/*************************************************************************
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#include "test_ReduceMultiProcess.hpp"
namespace CorrectnessTests
{
TEST_P(ReduceMultiProcessCorrectnessTest, Correctness)
{
Dataset* dataset = (Dataset*)mmap(NULL, sizeof(Dataset), PROT_READ|PROT_WRITE, MAP_SHARED|MAP_ANONYMOUS, -1, 0);
dataset->InitializeRootProcess(numDevices, numElements, dataType, inPlace, ncclCollReduce);
Barrier::ClearShmFiles(std::atoi(getenv("NCCL_COMM_ID")));
int pid1 = 0;
int pid2 = 0;
int pid3 = 0;
pid1 = fork();
// From this point on, ignore original process as we cannot have it create a HIP context
if (pid1 == 0)
{
pid2 = fork();
if (numDevices > 2)
{
pid3 = fork();
}
if ((pid2 > 0 && pid3 == 0 && numDevices == 2) || (pid2 > 0 && pid3 > 0 && numDevices > 2))
{
// Process 0
TestReduce(0, *dataset);
if (pid3 > 0)
{
waitpid(pid3, NULL, 0);
}
}
else if ((pid2 == 0 && pid3 == 0 && numDevices == 2) || (pid2 == 0 && pid3 > 0 && numDevices > 2))
{
// Process 1
TestReduce(1, *dataset);
if (numDevices > 2)
{
waitpid(pid3, NULL, 0);
}
exit(0);
}
else if (pid2 > 0 && pid3 == 0 && numDevices > 2)
{
// Process 2 (available when numDevices > 2)
TestReduce(2, *dataset);
exit(0);
}
else if (pid2 == 0 && pid3 == 0 && numDevices == 4)
{
// Process 3 (available when numDevices == 4)
TestReduce(3, *dataset);
exit(0);
}
else
{
exit(0);
}
waitpid(pid2, NULL, 0);
exit(0);
}
waitpid(pid1, NULL, 0);
munmap(dataset, sizeof(Dataset));
}
INSTANTIATE_TEST_SUITE_P(ReduceMultiProcessCorrectnessSweep,
ReduceMultiProcessCorrectnessTest,
testing::Combine(
// Reduction operator
testing::Values(ncclSum, ncclProd, ncclMax, ncclMin),
// Data types
testing::Values(ncclInt8,
ncclUint8,
ncclInt32,
ncclUint32,
ncclInt64,
ncclUint64,
//ncclFloat16,
ncclFloat32,
ncclFloat64,
ncclBfloat16),
// Number of elements
testing::Values(1024, 1048576),
// Number of devices
testing::Values(2,3,4),
// In-place or not
testing::Values(false, true),
testing::Values("")),
CorrectnessTest::PrintToStringParamName());
} // namespace
projects/rccl/test/test_ReduceMultiProcess.hpp
+115
View File
@@ -0,0 +1,115 @@
/*************************************************************************
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#ifndef TEST_REDUCE_MULTI_PROCESS_HPP
#define TEST_REDUCE_MULTI_PROCESS_HPP
#include "CorrectnessTest.hpp"
namespace CorrectnessTests
{
class ReduceMultiProcessCorrectnessTest : public MultiProcessCorrectnessTest
{
public:
static void ComputeExpectedResults(Dataset& dataset, Barrier& barrier, ncclRedOp_t const op, int const root, int const rank)
{
// Copy all inputs to expected arrays temporarily to perform reduction on host
HIP_CALL(hipMemcpy(dataset.expected[rank], dataset.inputs[rank],
dataset.NumBytes(), hipMemcpyDeviceToHost));
barrier.Wait();
if (rank == root)
{
// Allocate temporary host array to accumulate results
int8_t* resultI1 = (int8_t *)malloc(dataset.NumBytes());
uint8_t* resultU1 = (uint8_t *)resultI1;
int32_t* resultI4 = (int32_t *)resultI1;
uint32_t* resultU4 = (uint32_t *)resultI1;
int64_t* resultI8 = (int64_t *)resultI1;
uint64_t* resultU8 = (uint64_t *)resultI1;
float* resultF4 = (float *)resultI1;
double* resultF8 = (double *)resultI1;
rccl_bfloat16* resultB2 = (rccl_bfloat16 *)resultI1;
// Initialize the result with the first device's array
memcpy(resultI1, dataset.expected[0], dataset.NumBytes());
// Perform reduction on the other device arrays
for (int i = 1; i < dataset.numDevices; i++)
{
int8_t* arrayI1 = (int8_t *)dataset.expected[i];
uint8_t* arrayU1 = (uint8_t *)arrayI1;
int32_t* arrayI4 = (int32_t *)arrayI1;
uint32_t* arrayU4 = (uint32_t *)arrayI1;
int64_t* arrayI8 = (int64_t *)arrayI1;
uint64_t* arrayU8 = (uint64_t *)arrayI1;
float* arrayF4 = (float *)arrayI1;
double* arrayF8 = (double *)arrayI1;
rccl_bfloat16* arrayB2 = (rccl_bfloat16 *)arrayI1;
for (int j = 0; j < dataset.numElements; j++)
{
switch (dataset.dataType)
{
case ncclInt8: resultI1[j] = ReduceOp(op, resultI1[j], arrayI1[j]); break;
case ncclUint8: resultU1[j] = ReduceOp(op, resultU1[j], arrayU1[j]); break;
case ncclInt32: resultI4[j] = ReduceOp(op, resultI4[j], arrayI4[j]); break;
case ncclUint32: resultU4[j] = ReduceOp(op, resultU4[j], arrayU4[j]); break;
case ncclInt64: resultI8[j] = ReduceOp(op, resultI8[j], arrayI8[j]); break;
case ncclUint64: resultU8[j] = ReduceOp(op, resultU8[j], arrayU8[j]); break;
case ncclFloat32: resultF4[j] = ReduceOp(op, resultF4[j], arrayF4[j]); break;
case ncclFloat64: resultF8[j] = ReduceOp(op, resultF8[j], arrayF8[j]); break;
case ncclBfloat16: resultB2[j] = ReduceOp(op, resultB2[j], arrayB2[j]); break;
default:
fprintf(stderr, "[ERROR] Unsupported datatype\n");
exit(0);
}
}
}
memcpy(dataset.expected[root], resultI1, dataset.NumBytes());
free(resultI1);
barrier.Wait();
}
else
{
barrier.Wait();
HIP_CALL(hipMemcpy(dataset.expected[rank], dataset.outputs[rank], dataset.NumBytes(), hipMemcpyDeviceToHost));
}
}
void TestReduce(int rank, Dataset& dataset)
{
SetUpPerProcess(rank, ncclCollReduce, comms[rank], streams[rank], dataset);
if (numDevices > numDevicesAvailable) return;
Barrier barrier(rank, numDevices, std::atoi(getenv("NCCL_COMM_ID")));
// Test each possible root
for (int root = 0; root < numDevices; root++)
{
// Prepare input / output / expected results
FillDatasetWithPattern(dataset, rank);
ComputeExpectedResults(dataset, barrier, op, root, rank);
// Launch the reduction (1 process per GPU)
ncclResult_t res = ncclReduce(dataset.inputs[rank],
dataset.outputs[rank],
numElements, dataType, op,
root, comms[rank], streams[rank]);
// Wait for reduction to complete
HIP_CALL(hipStreamSynchronize(streams[rank]));
// Check results
ValidateResults(dataset, rank);
// Ensure all processes have finished current iteration before proceeding
barrier.Wait();
}
TearDownPerProcess(comms[rank], streams[rank]);
dataset.Release(rank);
}
};
}
#endif
projects/rccl/test/test_ReduceScatterMultiProcess.cpp
+96
View File
@@ -0,0 +1,96 @@
/*************************************************************************
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#include "test_ReduceScatterMultiProcess.hpp"
namespace CorrectnessTests
{
TEST_P(ReduceScatterMultiProcessCorrectnessTest, Correctness)
{
Dataset* dataset = (Dataset*)mmap(NULL, sizeof(Dataset), PROT_READ|PROT_WRITE, MAP_SHARED|MAP_ANONYMOUS, -1, 0);
dataset->InitializeRootProcess(numDevices, numElements, dataType, inPlace, ncclCollReduceScatter);
Barrier::ClearShmFiles(std::atoi(getenv("NCCL_COMM_ID")));
int pid1 = 0;
int pid2 = 0;
int pid3 = 0;
pid1 = fork();
// From this point on, ignore original process as we cannot have it create a HIP context
if (pid1 == 0)
{
pid2 = fork();
if (numDevices > 2)
{
pid3 = fork();
}
if ((pid2 > 0 && pid3 == 0 && numDevices == 2) || (pid2 > 0 && pid3 > 0 && numDevices > 2))
{
// Process 0
TestReduceScatter(0, *dataset);
if (pid3 > 0)
{
waitpid(pid3, NULL, 0);
}
}
else if ((pid2 == 0 && pid3 == 0 && numDevices == 2) || (pid2 == 0 && pid3 > 0 && numDevices > 2))
{
// Process 1
TestReduceScatter(1, *dataset);
if (numDevices > 2)
{
waitpid(pid3, NULL, 0);
}
exit(0);
}
else if (pid2 > 0 && pid3 == 0 && numDevices > 2)
{
// Process 2 (available when numDevices > 2)
TestReduceScatter(2, *dataset);
exit(0);
}
else if (pid2 == 0 && pid3 == 0 && numDevices == 4)
{
// Process 3 (available when numDevices == 4)
TestReduceScatter(3, *dataset);
exit(0);
}
else
{
exit(0);
}
waitpid(pid2, NULL, 0);
exit(0);
}
waitpid(pid1, NULL, 0);
munmap(dataset, sizeof(Dataset));
}
INSTANTIATE_TEST_SUITE_P(ReduceScatterMultiProcessCorrectnessSweep,
ReduceScatterMultiProcessCorrectnessTest,
testing::Combine(
// Reduction operator
testing::Values(ncclSum, ncclProd, ncclMax, ncclMin),
// Data types
testing::Values(ncclInt8,
ncclUint8,
ncclInt32,
ncclUint32,
ncclInt64,
ncclUint64,
//ncclFloat16,
ncclFloat32,
ncclFloat64,
ncclBfloat16),
// Number of elements
testing::Values(3072, 3145728),
// Number of devices
testing::Values(2,3,4),
// In-place or not
testing::Values(false, true),
testing::Values("")),
CorrectnessTest::PrintToStringParamName());
} // namespace
projects/rccl/test/test_ReduceScatterMultiProcess.hpp
+128
View File
@@ -0,0 +1,128 @@
/*************************************************************************
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#ifndef TEST_REDUCE_SCATTER_MULTI_PROCESS_HPP
#define TEST_REDUCE_SCATTER_MULTI_PROCESS_HPP
#include "CorrectnessTest.hpp"
namespace CorrectnessTests
{
class ReduceScatterMultiProcessCorrectnessTest : public MultiProcessCorrectnessTest
{
public:
static void ComputeExpectedResults(Dataset& dataset, Barrier& barrier, ncclRedOp_t const op, int const rank)
{
// Copy all inputs to expected arrays temporarily to perform reduction on host
HIP_CALL(hipMemcpy(dataset.expected[rank], dataset.inputs[rank],
dataset.NumBytes(), hipMemcpyDeviceToHost));
barrier.Wait();
// Have rank 0 do the expected calculation, then send results to other processes
int8_t* resultI1;
if (rank == 0)
{
// Allocate temporary host array to accumulate results
resultI1 = (int8_t *)malloc(dataset.NumBytes());
uint8_t* resultU1 = (uint8_t *)resultI1;
int32_t* resultI4 = (int32_t *)resultI1;
uint32_t* resultU4 = (uint32_t *)resultI1;
int64_t* resultI8 = (int64_t *)resultI1;
uint64_t* resultU8 = (uint64_t *)resultI1;
float* resultF4 = (float *)resultI1;
double* resultF8 = (double *)resultI1;
rccl_bfloat16* resultB2 = (rccl_bfloat16 *)resultI1;
// Initialize the result with the first device's array
memcpy(resultI1, dataset.expected[0], dataset.NumBytes());
// Perform reduction on the other device arrays
for (int i = 1; i < dataset.numDevices; i++)
{
int8_t* arrayI1 = (int8_t *)dataset.expected[i];
uint8_t* arrayU1 = (uint8_t *)arrayI1;
int32_t* arrayI4 = (int32_t *)arrayI1;
uint32_t* arrayU4 = (uint32_t *)arrayI1;
int64_t* arrayI8 = (int64_t *)arrayI1;
uint64_t* arrayU8 = (uint64_t *)arrayI1;
float* arrayF4 = (float *)arrayI1;
double* arrayF8 = (double *)arrayI1;
rccl_bfloat16* arrayB2 = (rccl_bfloat16 *)arrayI1;
for (int j = 0; j < dataset.numElements; j++)
{
switch (dataset.dataType)
{
case ncclInt8: resultI1[j] = ReduceOp(op, resultI1[j], arrayI1[j]); break;
case ncclUint8: resultU1[j] = ReduceOp(op, resultU1[j], arrayU1[j]); break;
case ncclInt32: resultI4[j] = ReduceOp(op, resultI4[j], arrayI4[j]); break;
case ncclUint32: resultU4[j] = ReduceOp(op, resultU4[j], arrayU4[j]); break;
case ncclInt64: resultI8[j] = ReduceOp(op, resultI8[j], arrayI8[j]); break;
case ncclUint64: resultU8[j] = ReduceOp(op, resultU8[j], arrayU8[j]); break;
case ncclFloat32: resultF4[j] = ReduceOp(op, resultF4[j], arrayF4[j]); break;
case ncclFloat64: resultF8[j] = ReduceOp(op, resultF8[j], arrayF8[j]); break;
case ncclBfloat16: resultB2[j] = ReduceOp(op, resultB2[j], arrayB2[j]); break;
default:
fprintf(stderr, "[ERROR] Unsupported datatype\n");
exit(0);
}
}
}
}
barrier.Wait();
// Copy results into expected arrays
size_t const byteCount = dataset.NumBytes() / dataset.numDevices;
HIP_CALL(hipMemcpy(dataset.expected[rank], dataset.outputs[rank],
dataset.NumBytes(), hipMemcpyDeviceToHost));
barrier.Wait();
if (rank == 0)
{
for (int i = 0; i < dataset.numDevices; i++)
memcpy((int8_t *)dataset.expected[i] + (i * byteCount),
resultI1 + (i * byteCount), byteCount);
free(resultI1);
}
}
void TestReduceScatter(int rank, Dataset& dataset)
{
// Prepare input / output / expected results
SetUpPerProcess(rank, ncclCollAllGather, comms[rank], streams[rank], dataset);
if (numDevices > numDevicesAvailable) return;
if (numElements % numDevices != 0) return;
Barrier barrier(rank, numDevices, std::atoi(getenv("NCCL_COMM_ID")));
// Prepare input / output / expected results
FillDatasetWithPattern(dataset, rank);
ComputeExpectedResults(dataset, barrier, op, rank);
size_t const byteCount = dataset.NumBytes() / numDevices;
size_t const recvCount = dataset.numElements / numDevices;
// Launch the reduction (1 process per GPU)
ncclReduceScatter(dataset.inputs[rank],
(int8_t *)dataset.outputs[rank] + (rank * byteCount),
recvCount, dataType, op,
comms[rank], streams[rank]);
// Wait for reduction to complete
HIP_CALL(hipStreamSynchronize(streams[rank]));
// Check results
ValidateResults(dataset, rank);
TearDownPerProcess(comms[rank], streams[rank]);
dataset.Release(rank);
}
};
}
#endif
projects/rccl/test/test_ScatterMultiProcess.cpp
+96
View File
@@ -0,0 +1,96 @@
/*************************************************************************
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#include "test_ScatterMultiProcess.hpp"
namespace CorrectnessTests
{
TEST_P(ScatterMultiProcessCorrectnessTest, Correctness)
{
Dataset* dataset = (Dataset*)mmap(NULL, sizeof(Dataset), PROT_READ|PROT_WRITE, MAP_SHARED|MAP_ANONYMOUS, -1, 0);
dataset->InitializeRootProcess(numDevices, numElements, dataType, inPlace, ncclCollScatter);
Barrier::ClearShmFiles(std::atoi(getenv("NCCL_COMM_ID")));
int pid1 = 0;
int pid2 = 0;
int pid3 = 0;
pid1 = fork();
// From this point on, ignore original process as we cannot have it create a HIP context
if (pid1 == 0)
{
pid2 = fork();
if (numDevices > 2)
{
pid3 = fork();
}
if ((pid2 > 0 && pid3 == 0 && numDevices == 2) || (pid2 > 0 && pid3 > 0 && numDevices > 2))
{
// Process 0
TestScatter(0, *dataset);
if (pid3 > 0)
{
waitpid(pid3, NULL, 0);
}
}
else if ((pid2 == 0 && pid3 == 0 && numDevices == 2) || (pid2 == 0 && pid3 > 0 && numDevices > 2))
{
// Process 1
TestScatter(1, *dataset);
if (numDevices > 2)
{
waitpid(pid3, NULL, 0);
}
exit(0);
}
else if (pid2 > 0 && pid3 == 0 && numDevices > 2)
{
// Process 2 (available when numDevices > 2)
TestScatter(2, *dataset);
exit(0);
}
else if (pid2 == 0 && pid3 == 0 && numDevices == 4)
{
// Process 3 (available when numDevices == 4)
TestScatter(3, *dataset);
exit(0);
}
else
{
exit(0);
}
waitpid(pid2, NULL, 0);
exit(0);
}
waitpid(pid1, NULL, 0);
munmap(dataset, sizeof(Dataset));
}
INSTANTIATE_TEST_SUITE_P(ScatterMultiProcessCorrectnessSweep,
ScatterMultiProcessCorrectnessTest,
testing::Combine(
// Reduction operator is not used
testing::Values(ncclSum),
// Data types
testing::Values(ncclInt8,
ncclUint8,
ncclInt32,
ncclUint32,
ncclInt64,
ncclUint64,
//ncclFloat16,
ncclFloat32,
ncclFloat64,
ncclBfloat16),
// Number of elements
testing::Values(1024, 1048576),
// Number of devices
testing::Values(2,3,4),
// In-place or not
testing::Values(false),
testing::Values("RCCL_ALLTOALL_KERNEL_DISABLE=0", "RCCL_ALLTOALL_KERNEL_DISABLE=1")),
CorrectnessTest::PrintToStringParamName());
} // namespace
projects/rccl/test/test_ScatterMultiProcess.hpp
+64
View File
@@ -0,0 +1,64 @@
/*************************************************************************
* Copyright (c) 2019-2021 Advanced Micro Devices, Inc. All rights reserved.
*
* See LICENSE.txt for license information
************************************************************************/
#ifndef TEST_SCATTER_MULTI_PROCESS_HPP
#define TEST_SCATTER_MULTI_PROCESS_HPP
#include "CorrectnessTest.hpp"
namespace CorrectnessTests
{
class ScatterMultiProcessCorrectnessTest : public MultiProcessCorrectnessTest
{
public:
static void ComputeExpectedResults(Dataset& dataset, int const root, int const rank)
{
if (rank == root)
{
for (int i = 0; i < dataset.numDevices; i++)
HIP_CALL(hipMemcpy(dataset.expected[i], (int8_t *)dataset.inputs[root]+dataset.NumBytes()*i,
dataset.NumBytes(), hipMemcpyDeviceToHost));
}
}
void TestScatter(int rank, Dataset& dataset)
{
// Prepare input / output / expected results
SetUpPerProcess(rank, ncclCollScatter, comms[rank], streams[rank], dataset);
if (numDevices > numDevicesAvailable) return;
Barrier barrier(rank, numDevices, std::atoi(getenv("NCCL_COMM_ID")));
// Test each possible root
for (int root = 0; root < numDevices; root++)
{
// Prepare input / output / expected results
FillDatasetWithPattern(dataset, rank);
ComputeExpectedResults(dataset, root, rank);
// Launch the reduction (1 process per GPU)
ncclScatter(dataset.inputs[rank],
dataset.outputs[rank],
numElements, dataType,
root, comms[rank], streams[rank]);
// Wait for reduction to complete
HIP_CALL(hipStreamSynchronize(streams[rank]));
// Check results
ValidateResults(dataset, rank);
barrier.Wait();
}
TearDownPerProcess(comms[rank], streams[rank]);
dataset.Release(rank);
}
};
}
#endif
projects/rccl/tools/TransferBench/TransferBench.cpp
+26 -3
View File
@@ -166,12 +166,21 @@ int main(int argc, char **argv)
links[i].blockParam = (BlockParam*)malloc(ev.numCpuPerLink * sizeof(BlockParam));
}
}
<<<<<<< HEAD
// Loop over all the different number of bytes to use per Link
for (auto N : valuesOfN)
{
if (!ev.outputToCsv) printf("Test %d: [%lu bytes]\n", testNum, N * sizeof(float));
=======
// Loop over all the different number of bytes to use per Link
for (auto N : valuesOfN)
{
if (!ev.outputToCsv) printf("Test %d: [%lu bytes]\n", testNum, N * sizeof(float));
>>>>>>> origin/develop
// Prepare links based on current N
for (int i = 0; i < numLinks; i++)
{
@@ -331,6 +340,7 @@ int main(int argc, char **argv)
DeallocateMemory(links[i].dstMemType, links[i].dstIndex, links[i].dstMem);
if (links[i].exeMemType == MEM_GPU)
<<<<<<< HEAD
{
HIP_CALL(hipEventDestroy(links[i].startEvent));
HIP_CALL(hipEventDestroy(links[i].stopEvent));
@@ -339,6 +349,16 @@ int main(int argc, char **argv)
}
else if (links[i].exeMemType == MEM_CPU)
{
=======
{
HIP_CALL(hipEventDestroy(links[i].startEvent));
HIP_CALL(hipEventDestroy(links[i].stopEvent));
HIP_CALL(hipStreamDestroy(links[i].stream));
HIP_CALL(hipFree(links[i].blockParam));
}
else if (links[i].exeMemType == MEM_CPU)
{
>>>>>>> origin/develop
free(links[i].blockParam);
}
}
@@ -540,11 +560,12 @@ void DisplayTopology()
printf(" |");
for (int j = 0; j < numGpuDevices; j++)
printf(" GPU %02d |", j);
printf("\n");
printf(" PCIe Bus ID\n");
for (int j = 0; j <= numGpuDevices; j++)
printf("--------+");
printf("\n");
printf("-------------\n");
char pciBusId[20];
for (int i = 0; i < numGpuDevices; i++)
{
printf(" GPU %02d |", i);
@@ -565,7 +586,8 @@ void DisplayTopology()
hopCount);
}
}
printf("\n");
HIP_CALL(hipDeviceGetPCIBusId(pciBusId, 20, i));
printf(" %s\n", pciBusId);
}
}
@@ -865,6 +887,7 @@ void CheckOrFill(ModeType mode, int N, bool isMemset, bool isHipCall, float* ptr
exit(1);
}
}
free(hostBuffer);
}
free(refBuffer);
projects/rccl/tools/scripts/topo_val.sh
+1 -1
View File
@@ -21,7 +21,7 @@
DIR="$(cd -P "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
for i in {0..44}
for i in {0..47}
do
$DIR/../topo_expl/topo_expl -m $i > "topo_m$i.log"
$DIR/../TopoVisual/topo_visual.sh -i "topo_m$i.log"
projects/rccl/tools/topo_expl/models/topo_8p_rome_n4_1.xml
+93
View File
@@ -0,0 +1,93 @@
<system version="2">
<cpu numaid="1" affinity="00000000,00000000,00000000,ffff0000" arch="x86_64" vendor="AuthenticAMD" familyid="143" modelid="49">
<pci busid="0000:41:00.0" class="0x060400" link_speed="16 GT/s" link_width="16">
<pci busid="0000:43:00.0" class="0x038000" link_speed="16 GT/s" link_width="16">
<gpu dev="0" sm="90" gcn="908" arch="38911" rank="0" gdr="1">
<xgmi target="0000:23:00.0" count="1" tclass="0x038000"/>
<xgmi target="0000:26:00.0" count="1" tclass="0x038000"/>
<xgmi target="0000:03:00.0" count="1" tclass="0x038000"/>
</gpu>
</pci>
</pci>
</cpu>
<cpu numaid="2" affinity="00000000,00000000,0000ffff,00000000" arch="x86_64" vendor="AuthenticAMD" familyid="143" modelid="49">
<pci busid="0000:21:00.0" class="0x060400" link_speed="16 GT/s" link_width="16">
<pci busid="0000:23:00.0" class="0x038000" link_speed="16 GT/s" link_width="16">
<gpu dev="1" sm="90" gcn="908" arch="38911" rank="1" gdr="1">
<xgmi target="0000:43:00.0" count="1" tclass="0x038000"/>
<xgmi target="0000:26:00.0" count="1" tclass="0x038000"/>
<xgmi target="0000:03:00.0" count="1" tclass="0x038000"/>
</gpu>
</pci>
</pci>
<pci busid="0000:24:00.0" class="0x060400" link_speed="16 GT/s" link_width="16">
<pci busid="0000:26:00.0" class="0x038000" link_speed="16 GT/s" link_width="16">
<gpu dev="2" sm="90" gcn="908" arch="38911" rank="2" gdr="1">
<xgmi target="0000:43:00.0" count="1" tclass="0x038000"/>
<xgmi target="0000:23:00.0" count="1" tclass="0x038000"/>
<xgmi target="0000:03:00.0" count="1" tclass="0x038000"/>
</gpu>
</pci>
</pci>
</cpu>
<cpu numaid="3" affinity="00000000,00000000,ffff0000,00000000" arch="x86_64" vendor="AuthenticAMD" familyid="143" modelid="49">
<pci busid="0000:01:00.0" class="0x060400" link_speed="16 GT/s" link_width="16">
<pci busid="0000:03:00.0" class="0x038000" link_speed="16 GT/s" link_width="16">
<gpu dev="3" sm="90" gcn="908" arch="38911" rank="3" gdr="1">
<xgmi target="0000:43:00.0" count="1" tclass="0x038000"/>
<xgmi target="0000:23:00.0" count="1" tclass="0x038000"/>
<xgmi target="0000:26:00.0" count="1" tclass="0x038000"/>
</gpu>
</pci>
</pci>
</cpu>
<cpu numaid="5" affinity="00000000,ffff0000,00000000,00000000" arch="x86_64" vendor="AuthenticAMD" familyid="143" modelid="49">
<pci busid="0000:c1:00.0" class="0x060400" link_speed="16 GT/s" link_width="16">
<pci busid="0000:c3:00.0" class="0x038000" link_speed="16 GT/s" link_width="16">
<gpu dev="4" sm="90" gcn="908" arch="38911" rank="4" gdr="1">
<xgmi target="0000:c6:00.0" count="1" tclass="0x038000"/>
<xgmi target="0000:a3:00.0" count="1" tclass="0x038000"/>
<xgmi target="0000:83:00.0" count="1" tclass="0x038000"/>
</gpu>
</pci>
</pci>
<pci busid="0000:c4:00.0" class="0x060400" link_speed="16 GT/s" link_width="16">
<pci busid="0000:c6:00.0" class="0x038000" link_speed="16 GT/s" link_width="16">
<gpu dev="5" sm="90" gcn="908" arch="38911" rank="5" gdr="1">
<xgmi target="0000:c3:00.0" count="1" tclass="0x038000"/>
<xgmi target="0000:a3:00.0" count="1" tclass="0x038000"/>
<xgmi target="0000:83:00.0" count="1" tclass="0x038000"/>
</gpu>
</pci>
</pci>
</cpu>
<cpu numaid="6" affinity="0000ffff,00000000,00000000,00000000" arch="x86_64" vendor="AuthenticAMD" familyid="143" modelid="49">
<pci busid="0000:a1:00.0" class="0x060400" link_speed="16 GT/s" link_width="16">
<pci busid="0000:a3:00.0" class="0x038000" link_speed="16 GT/s" link_width="16">
<gpu dev="6" sm="90" gcn="908" arch="38911" rank="6" gdr="1">
<xgmi target="0000:c3:00.0" count="1" tclass="0x038000"/>
<xgmi target="0000:c6:00.0" count="1" tclass="0x038000"/>
<xgmi target="0000:83:00.0" count="1" tclass="0x038000"/>
</gpu>
</pci>
</pci>
</cpu>
<cpu numaid="7" affinity="ffff0000,00000000,00000000,00000000" arch="x86_64" vendor="AuthenticAMD" familyid="143" modelid="49">
<pci busid="0000:81:00.0" class="0x060400" link_speed="16 GT/s" link_width="16">
<pci busid="0000:83:00.0" class="0x038000" link_speed="16 GT/s" link_width="16">
<gpu dev="7" sm="90" gcn="908" arch="38911" rank="7" gdr="1">
<xgmi target="0000:c3:00.0" count="1" tclass="0x038000"/>
<xgmi target="0000:c6:00.0" count="1" tclass="0x038000"/>
<xgmi target="0000:a3:00.0" count="1" tclass="0x038000"/>
</gpu>
</pci>
</pci>
</cpu>
<cpu numaid="4" affinity="00000000,0000ffff,00000000,00000000" arch="x86_64" vendor="AuthenticAMD" familyid="143" modelid="49">
<pci busid="0000:e1:00.0" class="0x020700" link_speed="16 GT/s" link_width="16">
<nic>
<net name="mlx5_0" dev="0" speed="200000" port="1" guid="0x42fc9f00039b0398" maxconn="262144" gdr="1"/>
</nic>
</pci>
</cpu>
</system>
projects/rccl/tools/topo_expl/models/topo_8p_rome_pcie.xml
+52
View File
@@ -0,0 +1,52 @@
<system version="2">
<cpu numaid="0" affinity="00000000,00000000,ffffffff,ffffffff" arch="x86_64" vendor="AuthenticAMD" familyid="143" modelid="49">
<pci busid="0000:41:00.0" class="0x060400" link_speed="16 GT/s" link_width="16">
<pci busid="0000:43:00.0" class="0x038000" link_speed="16 GT/s" link_width="16">
<gpu dev="0" sm="90" gcn="906" arch="38911" rank="0" gdr="1"/>
</pci>
</pci>
<pci busid="0000:21:00.0" class="0x060400" link_speed="16 GT/s" link_width="16">
<pci busid="0000:23:00.0" class="0x038000" link_speed="16 GT/s" link_width="16">
<gpu dev="1" sm="90" gcn="906" arch="38911" rank="1" gdr="1"/>
</pci>
</pci>
<pci busid="0000:24:00.0" class="0x060400" link_speed="16 GT/s" link_width="16">
<pci busid="0000:26:00.0" class="0x038000" link_speed="16 GT/s" link_width="16">
<gpu dev="2" sm="90" gcn="906" arch="38911" rank="2" gdr="1"/>
</pci>
</pci>
<pci busid="0000:01:00.0" class="0x060400" link_speed="16 GT/s" link_width="16">
<pci busid="0000:03:00.0" class="0x038000" link_speed="16 GT/s" link_width="16">
<gpu dev="3" sm="90" gcn="906" arch="38911" rank="3" gdr="1"/>
</pci>
</pci>
</cpu>
<cpu numaid="1" affinity="ffffffff,ffffffff,00000000,00000000" arch="x86_64" vendor="AuthenticAMD" familyid="143" modelid="49">
<pci busid="0000:e1:00.0" class="0x060400" link_speed="16 GT/s" link_width="16">
<pci busid="0000:e3:00.0" class="0x038000" link_speed="16 GT/s" link_width="16">
<gpu dev="4" sm="90" gcn="906" arch="38911" rank="4" gdr="1"/>
</pci>
</pci>
<pci busid="0000:c1:00.0" class="0x060400" link_speed="16 GT/s" link_width="16">
<pci busid="0000:c3:00.0" class="0x038000" link_speed="16 GT/s" link_width="16">
<gpu dev="5" sm="90" gcn="906" arch="38911" rank="5" gdr="1"/>
</pci>
</pci>
<pci busid="0000:c4:00.0" class="0x060400" link_speed="16 GT/s" link_width="16">
<pci busid="0000:c6:00.0" class="0x038000" link_speed="16 GT/s" link_width="16">
<gpu dev="6" sm="90" gcn="906" arch="38911" rank="6" gdr="1"/>
</pci>
</pci>
<pci busid="0000:81:00.0" class="0x060400" link_speed="16 GT/s" link_width="16">
<pci busid="0000:83:00.0" class="0x038000" link_speed="16 GT/s" link_width="16">
<gpu dev="7" sm="90" gcn="906" arch="38911" rank="7" gdr="1"/>
</pci>
</pci>
<pci busid="0000:a1:00.0" class="0x020000" link_speed="8 GT/s" link_width="4">
<nic>
<net name="mlx5_0" dev="0" speed="100000" port="1" guid="0x7893510003a1420c" maxconn="262144" gdr="1"/>
<net name="mlx5_1" dev="1" speed="100000" port="2" guid="0x7893510003a1420c" maxconn="262144" gdr="1"/>
</nic>
</pci>
</cpu>
</system>
projects/rccl/tools/topo_expl/topo_expl.cpp
+3
View File
@@ -114,6 +114,9 @@ NodeModelDesc model_descs[] = {
{4, "topo_4p3l_n4.xml", "4 nodes 8 gfx908 Rome NPS=4"},
{1, "topo_4p3l_n2_1.xml", "single node 8 gfx908 Rome"},
{4, "topo_4p3l_n2_1.xml", "4 nodes 8 gfx908 Rome"},
{1, "topo_8p_rome_n4_1.xml", "single node 8 gfx908 Rome NPS=4"},
{4, "topo_8p_rome_n4_1.xml", "4 nodes node 8 gfx908 Rome NPS=4"},
{2, "topo_8p_rome_pcie.xml", "2 nodes node 8 VEGA20 PCIe"},
};
int main(int argc,char* argv[])