nccl 04 nvidia 官方小程序

1，代码重新编辑

为了地毯式地检查结果的正确性，这里修改了代码

主要步骤为

step1: data_p指向的空间中，分别生成随机数；

step2: 分别拷贝到gpu的sendbuff的显存中；

step3: 通过nccl_all_reduce sum；

step4: 取回 recvbuff中的数据；

step5: 将data_p中的数据 allreduce sum；

step6: 对比 recvbuff中的数据与 data_p中的数据的一致性；


#include <stdio.h>
#include <stdlib.h>
#include <iostream>
#include <omp.h>// seed_base
// void gen_512(seed, A, start_idx, N)
// gen_512(seed_base+0, A, 0*512, N);  gen_512(seed_base + 1, A+1*512); gen_512(seed_base + 2, A+2*512);     ...    gen_512(seed_base + n, A, , N); ...
#if 1//NN is tile length
#define NN 312
#define MM 156
#define MATRIX_A 0xB5026F5AA96619E9ULL
#define UM 0xFFFFFFFF80000000ULL /* Most significant 33 bits */
#define LM 0x7FFFFFFFULL /* Least significant 31 bits */class thread_mt199937{public:
void srand(unsigned long long seed)
{init_genrand64(seed);
}/* generates a random number on [0, 2^63-1]-interval */
long long genrand64_int63(void)
{return (long long)(genrand64_int64() >> 1);
}/* generates a random number on [0,1]-real-interval */
double genrand64_real1(void)
{return (genrand64_int64() >> 11) * (1.0/9007199254740991.0);
}/* generates a random number on [0,1)-real-interval */
double genrand64_real2(void)
{return (genrand64_int64() >> 11) * (1.0/9007199254740992.0);
}/* generates a random number on (0,1)-real-interval */
double genrand64_real3(void)
{return ((genrand64_int64() >> 12) + 0.5) * (1.0/4503599627370496.0);
}private:
/* The array for the state vector */
unsigned long long mt[NN];
/* mti==NN+1 means mt[NN] is not initialized */
int mti=NN+1;/* initializes mt[NN] with a seed */
void init_genrand64(unsigned long long seed)
{mt[0] = seed;for (mti=1; mti<NN; mti++)mt[mti] =  (6364136223846793005ULL * (mt[mti-1] ^ (mt[mti-1] >> 62)) + mti);
}/* initialize by an array with array-length */
/* init_key is the array for initializing keys */
/* key_length is its length */
void init_by_array64(unsigned long long init_key[],unsigned long long key_length)
{unsigned long long i, j, k;init_genrand64(19650218ULL);i=1; j=0;k = (NN>key_length ? NN : key_length);for (; k; k--) {mt[i] = (mt[i] ^ ((mt[i-1] ^ (mt[i-1] >> 62)) * 3935559000370003845ULL))+ init_key[j] + j; /* non linear */i++; j++;if (i>=NN) { mt[0] = mt[NN-1]; i=1; }if (j>=key_length) j=0;}for (k=NN-1; k; k--) {mt[i] = (mt[i] ^ ((mt[i-1] ^ (mt[i-1] >> 62)) * 2862933555777941757ULL))- i; /* non linear */i++;if (i>=NN) { mt[0] = mt[NN-1]; i=1; }}mt[0] = 1ULL << 63; /* MSB is 1; assuring non-zero initial array */
}/* generates a random number on [0, 2^64-1]-interval */
unsigned long long genrand64_int64(void)
{int i;unsigned long long x;static unsigned long long mag01[2]={0ULL, MATRIX_A};if (mti >= NN) { /* generate NN words at one time *//* if init_genrand64() has not been called, *//* a default initial seed is used     */if (mti == NN+1)init_genrand64(5489ULL);for (i=0;i<NN-MM;i++) {x = (mt[i]&UM)|(mt[i+1]&LM);mt[i] = mt[i+MM] ^ (x>>1) ^ mag01[(int)(x&1ULL)];}for (;i<NN-1;i++) {x = (mt[i]&UM)|(mt[i+1]&LM);mt[i] = mt[i+(MM-NN)] ^ (x>>1) ^ mag01[(int)(x&1ULL)];}x = (mt[NN-1]&UM)|(mt[0]&LM);mt[NN-1] = mt[MM-1] ^ (x>>1) ^ mag01[(int)(x&1ULL)];mti = 0;//printf("LL::\n");}x = mt[mti++];x ^= (x >> 29) & 0x5555555555555555ULL;x ^= (x << 17) & 0x71D67FFFEDA60000ULL;x ^= (x << 37) & 0xFFF7EEE000000000ULL;x ^= (x >> 43);return x;
}};#endif//#define NN 312// multi thread accelerating， to be singleton
class parallel_mt19937{
public:
//    int _base_seed;
void srand(unsigned long long seed)
{_base_seed = seed;
}
void rand_float(float* A, unsigned long len);private:unsigned long long _base_seed = 0;
};//int pmt19937::_base_seed = 0;void parallel_mt19937::rand_float(float* A, unsigned long len)
{#pragma omp parallel{unsigned long block_dim = omp_get_num_threads();unsigned long thid = omp_get_thread_num();if(thid == block_dim -1)std::cout << "Here are " << thid << " threads generating random number ..."<<std::endl;unsigned long tile_count = (len+NN-1)/NN;thread_mt199937 *t_mt_p = new thread_mt199937();// to be singletonfor(unsigned long tile_id=thid; tile_id<tile_count; tile_id+=block_dim){//each tile has a specific seed: (_base_seed + tile_id) to smt19937, to keep consistenceunsigned long tile_seed = _base_seed + tile_id;t_mt_p->srand(tile_seed);//if(thid == 35)                std::cout << "Hello from thread " << thid << std::endl;unsigned long tile_idx_start = tile_id*NN;unsigned long tile_idx_end = (((tile_id+1)*NN) <= len)? (tile_id+1)*NN: len;for(unsigned long idx = tile_idx_start; idx<tile_idx_end; idx++)A[idx] = float(t_mt_p->genrand64_real2());}delete t_mt_p;}
}////#define BUILD_MAIN//
#ifdef BUILD_MAINvoid print_matrix(unsigned long m, unsigned long n, float* A, unsigned long lda)
{for(unsigned long i=m-7; i<m; i++){for(unsigned long j=n-7; j<n; j++){printf("%7.4f ", A[i + j*lda]);}printf("\n");}
}#include <string.h>int main(){unsigned long m = 312*4*32*32*128;unsigned long n = 36;unsigned long lda = m;float* A = nullptr;
printf("LL: 00\n");A = (float*)malloc(lda * n * sizeof(float));//memset(A, 0x7F, lda*n*sizeof(float));
printf("LL:: 01\n");parallel_mt19937 gen_rand;gen_rand.srand(2024);gen_rand.rand_float(A, lda*n);
printf("LL:: 02\n");print_matrix(m, n, A, lda);free(A);return 0;
}#endifvoid init_data_float(float** data_p, int nDev, int size,  unsigned long seed)
{for(int idx=0; idx<nDev; idx++){*(data_p+idx) = nullptr;*(data_p+idx) = (float*)malloc(size*sizeof(float));}parallel_mt19937 gen_rand;for(int idx =0; idx<nDev; idx++){gen_rand.srand(seed+idx);gen_rand.rand_float(*(data_p + idx), size);}}void host_all_reduce_origin(float** sendbuff, int nDev, int size)
{for(int idx=1; idx<nDev; idx++){
#pragma omp paralell forfor(int i=0; i<size; i++)sendbuff[0][i] += sendbuff[idx][i];}
}void check_equal(float** result_recv_data_buff, float* sendbuff_0, int nDev, int size)
{for(int idx=0; idx<nDev; idx++){
#pragma omp paralell forfor(int i=0; i<size; i++){if(result_recv_data_buff[idx][i] != sendbuff_0[i])printf("ERROR: %7.4f != %7.4f  idx=%d, i=%d\n", result_recv_data_buff[idx][i], sendbuff_0[i], idx, i);return;}}
}void free_buff(float** buffArray, int n)
{for(int i=0; i<n; i++){if(buffArray[i] != nullptr){free(buffArray[i]);}}
}//#include <stdlib.h>
#include <stdio.h>
#include "cuda_runtime.h"
#include "nccl.h"#define CUDACHECK(cmd) do {                         \cudaError_t err = cmd;                            \if (err != cudaSuccess) {                         \printf("Failed: Cuda error %s:%d '%s'\n",       \__FILE__,__LINE__,cudaGetErrorString(err)); \exit(EXIT_FAILURE);                             \}                                                 \
} while(0)#define NCCLCHECK(cmd) do {                         \ncclResult_t res = cmd;                           \if (res != ncclSuccess) {                         \printf("Failed, NCCL error %s:%d '%s'\n",       \__FILE__,__LINE__,ncclGetErrorString(res)); \exit(EXIT_FAILURE);                             \}                                                 \
} while(0)int main(int argc, char* argv[])
{ncclComm_t comms[4];//managing 4 devices//int nDev = 4;int nDev = 2;int size = 32*1024*1024*16;int devs[4] = { 0, 1, 2, 3 };float** data_p = (float**)malloc(nDev*sizeof(float*));init_data_float(data_p, nDev, size, 2024);//allocating and initializing device buffersfloat** sendbuff = (float**)malloc(nDev * sizeof(float*));float** recvbuff = (float**)malloc(nDev * sizeof(float*));cudaStream_t* s = (cudaStream_t*)malloc(sizeof(cudaStream_t)*nDev);for (int i = 0; i < nDev; ++i) {CUDACHECK(cudaSetDevice(i));CUDACHECK(cudaMalloc((void**)sendbuff + i, size * sizeof(float)));CUDACHECK(cudaMalloc((void**)recvbuff + i, size * sizeof(float)));//CUDACHECK(cudaMemset(sendbuff[i], 1, size * sizeof(float)));CUDACHECK(cudaMemcpy(sendbuff[i], data_p[i], size*sizeof(float), cudaMemcpyHostToDevice));//CUDACHECK(cudaMemset(recvbuff[i], 0, size * sizeof(float)));CUDACHECK(cudaStreamCreate(s+i));}printf("LL:: 04\n");//initializing NCCLNCCLCHECK(ncclCommInitAll(comms, nDev, devs));
printf("LL:: 05\n");//calling NCCL communication API. Group API is required when using//multiple devices per threadNCCLCHECK(ncclGroupStart());for (int i = 0; i < nDev; ++i)NCCLCHECK(ncclAllReduce((const void*)sendbuff[i], (void*)recvbuff[i], size, ncclFloat, ncclSum,comms[i], s[i]));NCCLCHECK(ncclGroupEnd());
printf("LL:: 06\n");
/*LL::
In a sum allreduce operation between k ranks,
each rank will provide an array in of N values,
and receive identical results in array out of N values,
where out[i] = in0[i]+in1[i]+…+in(k-1)[i]
*///synchronizing on CUDA streams to wait for completion of NCCL operationfor (int i = 0; i < nDev; ++i) {CUDACHECK(cudaSetDevice(i));CUDACHECK(cudaStreamSynchronize(s[i]));}
printf("LL:: 07\n");
/*check:
each recvbuff[0] = sendbuff_0[0]+ sendbuff_1[0];
0x01010101 + 0x01010101 = 0x02020202;
*/float** result_recv_data_buff;result_recv_data_buff = (float**)malloc(nDev*sizeof(float*));init_data_float(result_recv_data_buff, nDev, size, 2024+nDev);printf("LL:: 08\n");for(int idx=0; idx<nDev; idx++){cudaMemcpy(result_recv_data_buff[idx], recvbuff[idx], size*sizeof(float), cudaMemcpyDeviceToHost);}printf("LL:: 09\n");host_all_reduce_origin(data_p, nDev, size);printf("LL:: 10\n");
// ditanshi checkcheck_equal(result_recv_data_buff, data_p[0], nDev, size);printf("LL:: 11\n");free_buff(data_p, nDev);free_buff(result_recv_data_buff, nDev);//free device buffersfor (int i = 0; i < nDev; ++i) {CUDACHECK(cudaSetDevice(i));CUDACHECK(cudaFree(sendbuff[i]));CUDACHECK(cudaFree(recvbuff[i]));}//finalizing NCCLfor(int i = 0; i < nDev; ++i)ncclCommDestroy(comms[i]);printf("Success \n");return 0;
}

Makefile

EXE := ex_1_1_SingleProcessSingleThreadMultipleDevicesall: $(EXE)INC := -I /usr/local/cuda/include -I /home/hipper/ex_nccl_20240701/local/include/
LD_FLAGS := -L /usr/local/cuda/lib64 -L /home/hipper/ex_nccl_20240701/local/lib -lcudart -lnccl -fopenmp%: %.cppg++ -g $< -o $@ $(INC) $(LD_FLAGS).PHONY: clean
clean:-rm -rf $(EXE)