示例

本节中的示例提供了如何在各种环境中使用 MCCL 的总体视图，结合一种或多种技术：

• 每个线程/进程使用多个 GPU
• 使用多个线程
• 使用多个进程 - 多个进程的示例使用 MPI 作为并行运行时环境，但任何多进程系统应该都能类似地工作。

确保您始终检查 MCCL 函数的返回代码。 为了清晰起见，以下示例不包含错误检查。

示例索引补充

除原有集合通信和通信器示例外，这里还包含一批面向主机侧管理能力和扩展能力的示例：

• 通信器属性查询：examples/01_communicators/04_comm_query_properties
• 通信器动态扩容：examples/01_communicators/05_comm_grow
• 通信器派生 unique ID：examples/01_communicators/06_comm_get_unique_id
• 通信器内存管理：examples/01_communicators/07_comm_manage_memory
• LSA 设备指针查询：examples/05_symmetric_memory/02_get_lsa_device_pointer
• 窗口用户指针查询：examples/05_symmetric_memory/03_win_get_user_ptr
• RMA signal：examples/08_rma/01_signal

如果你的目标是验证 2.4 新接口，优先从这些示例开始，而不是只运行传统的 all_reduce_perf。

按能力选择示例

目标能力	推荐示例
传统通信器初始化/销毁	examples/01_communicators/01_* ~ 03_*
运行时探测设备 API / GIN / LSA	examples/01_communicators/04_comm_query_properties
动态加入新 rank	examples/01_communicators/05_comm_grow
通信器内存挂起/恢复	examples/01_communicators/07_comm_manage_memory
对称窗口与 LSA	examples/05_symmetric_memory/02_get_lsa_device_pointer、03_win_get_user_ptr
纯 device API / GIN kernel 模式	examples/06_device_api/*
RMA signal	examples/08_rma/01_signal

通信器创建和销毁示例

以下示例展示了 MCCL 初始化的常见用例。

示例 1：单进程，单线程，多个设备

在单进程的特定情况下，可以使用 mcclCommInitAll。 这里是一个创建 4 个设备的通信器的示例，因此， 有 4 个通信器对象：

mcclComm_t comms[4];
int devs[4] = { 0, 1, 2, 3 };
mcclCommInitAll(comms, 4, devs);

接下来，您可以使用单个线程和 组调用来调用 MCCL 集体操作， 或者使用多个线程，每个线程都提供一个 comm 对象。

在程序结束时，销毁所有通信器对象：

for (int i=0; i<4; i++)
  mcclCommDestroy(comms[i]);

相关链接：mcclCommDestroy

以下代码展示了一个完整的工作示例，其中单个进程管理多个设备：

#include <stdlib.h>
#include <stdio.h>
#include "musa_runtime.h"
#include "mccl.h"

#define MUSACHECK(cmd) do {                         \
  musaError_t err = cmd;                            \
  if (err != musaSuccess) {                         \
    printf("Failed: musa error %s:%d '%s'\n",       \
        __FILE__,__LINE__,musaGetErrorString(err)); \
    exit(EXIT_FAILURE);                             \
  }                                                 \
} while(0)


#define MCCLCHECK(cmd) do {                         \
  mcclResult_t res = cmd;                           \
  if (res != mcclSuccess) {                         \
    printf("Failed, MCCL error %s:%d '%s'\n",       \
        __FILE__,__LINE__,mcclGetErrorString(res)); \
    exit(EXIT_FAILURE);                             \
  }                                                 \
} while(0)


int main(int argc, char* argv[])
{
  mcclComm_t comms[4];


  //管理 4 个设备
  int nDev = 4;
  int size = 32*1024*1024;
  int devs[4] = { 0, 1, 2, 3 };


  //为设备缓冲区分配和初始化
  float** sendbuff = (float**)malloc(nDev * sizeof(float*));
  float** recvbuff = (float**)malloc(nDev * sizeof(float*));
  musaStream_t* s = (musaStream_t*)malloc(sizeof(musaStream_t)*nDev);


  for (int i = 0; i < nDev; ++i) {
    MUSACHECK(musaSetDevice(i));
    MUSACHECK(musaMalloc((void**)sendbuff + i, size * sizeof(float)));
    MUSACHECK(musaMalloc((void**)recvbuff + i, size * sizeof(float)));
    MUSACHECK(musaMemset(sendbuff[i], 1, size * sizeof(float)));
    MUSACHECK(musaMemset(recvbuff[i], 0, size * sizeof(float)));
    MUSACHECK(musaStreamCreate(s+i));
  }


  //初始化 MCCL
  MCCLCHECK(mcclCommInitAll(comms, nDev, devs));


    //调用 MCCL 通信 API。当使用
    //每个线程的多个设备时需要组 API
  MCCLCHECK(mcclGroupStart());
  for (int i = 0; i < nDev; ++i)
    MCCLCHECK(mcclAllReduce((const void*)sendbuff[i], (void*)recvbuff[i], size, mcclFloat, mcclSum,
        comms[i], s[i]));
  MCCLCHECK(mcclGroupEnd());


  //在 MUSA stream 上同步以等待 MCCL 操作完成
  for (int i = 0; i < nDev; ++i) {
    MUSACHECK(musaSetDevice(i));
    MUSACHECK(musaStreamSynchronize(s[i]));
  }


  //free 设备缓冲区
  for (int i = 0; i < nDev; ++i) {
    MUSACHECK(musaSetDevice(i));
    MUSACHECK(musaFree(sendbuff[i]));
    MUSACHECK(musaFree(recvbuff[i]));
  }


  //完成 MCCL
  for(int i = 0; i < nDev; ++i)
      mcclCommDestroy(comms[i]);



  printf("Success \n");
  return 0;
}

示例 2：每个进程或线程一个设备

当进程或主机线程最多负责一个 GPU 时， 可以使用 mcclCommInitRank 作为集体调用来创建通信器。每个线程或进程将获得自己的对象。

以下代码是使用 MPI 的通信器创建的示例，每个 MPI rank 使用一个设备。

首先，我们获取有关进程的 MPI 信息：

int myRank, nRanks;
MPI_Comm_rank(MPI_COMM_WORLD, &myRank);
MPI_Comm_size(MPI_COMM_WORLD, &nRanks);

接下来，单个 rank 将创建一个唯一 ID 并将其发送给所有其他 rank 以确保每个人都拥有它：

mcclUniqueId id;
if (myRank == 0) mcclGetUniqueId(&id);
MPI_Bcast(&id, sizeof(id), MPI_BYTE, 0, MPI_COMM_WORLD);

相关链接：mcclGetUniqueId

最后，我们创建通信器：

mcclComm_t comm;
mcclCommInitRank(&comm, nRanks, id, myRank);

相关链接：mcclCommInitRank

现在我们可以使用通信器调用 MCCL 集体操作。

mcclAllReduce( ... , comm);

相关链接：mcclAllReduce

最后，我们销毁通信器对象：

mcclCommDestroy(comm);

相关链接：mcclCommDestroy

以下代码展示了一个完整的工作示例，其中包含多个 MPI 进程和每个进程一个设备：

#include <stdio.h>
#include "musa_runtime.h"
#include "mccl.h"
#include "mpi.h"
#include <unistd.h>
#include <stdint.h>
#include <stdlib.h>


#define MPICHECK(cmd) do {                          \
  int e = cmd;                                      \
  if( e != MPI_SUCCESS ) {                          \
    printf("Failed: MPI error %s:%d '%d'\n",        \
        __FILE__,__LINE__, e);   \
    exit(EXIT_FAILURE);                             \
  }                                                 \
} while(0)


#define MUSACHECK(cmd) do {                         \
  musaError_t e = cmd;                              \
  if( e != musaSuccess ) {                          \
    printf("Failed: musa error %s:%d '%s'\n",             \
        __FILE__,__LINE__,musaGetErrorString(e));   \
    exit(EXIT_FAILURE);                             \
  }                                                 \
} while(0)


#define MCCLCHECK(cmd) do {                         \
  mcclResult_t r = cmd;                             \
  if (r!= mcclSuccess) {                            \
    printf("Failed, MCCL error %s:%d '%s'\n",             \
        __FILE__,__LINE__,mcclGetErrorString(r));   \
    exit(EXIT_FAILURE);                             \
  }                                                 \
} while(0)


static uint64_t getHash(const char* string) {
  // Based on DJB2a, result = result * 33 ^ char
  uint64_t result = 5381;
  for (int c = 0; string[c] != '\0'; c++){
    result = ((result << 5) + result) ^ string[c];
  }
  return result;
}

/* Generate a hash of the unique identifying string for this host
  *
  * that will be unique for both bare-metal and container instances
  * Equivalent of a hash of;
  *
  * $(hostname)$(cat /proc/sys/kernel/random/boot_id)
  *
  */
#define HOSTID_FILE "/proc/sys/kernel/random/boot_id"
static uint64_t getHostHash(const char* hostname) {
  char hostHash[1024];

  // Fall back is the hostname if something fails
  (void) strncpy(hostHash, hostname, sizeof(hostHash));
  int offset = strlen(hostHash);

  FILE *file = fopen(HOSTID_FILE, "r");
  if (file != NULL) {
    char *p;
    if (fscanf(file, "%ms", &p) == 1) {
        strncpy(hostHash+offset, p, sizeof(hostHash)-offset-1);
        free(p);
    }
  }
  fclose(file);

  // Make sure the string is terminated
  hostHash[sizeof(hostHash)-1]='\0';

  return getHash(hostHash, strlen(hostHash));
}

static void getHostName(char* hostname, int maxlen) {
  gethostname(hostname, maxlen);
  for (int i=0; i< maxlen; i++) {
    if (hostname[i] == '.') {
        hostname[i] = '\0';
        return;
    }
  }
}


int main(int argc, char* argv[])
{
  int size = 32*1024*1024;


  int myRank, nRanks, localRank = 0;


  //initializing MPI
  MPICHECK(MPI_Init(&argc, &argv));
  MPICHECK(MPI_Comm_rank(MPI_COMM_WORLD, &myRank));
  MPICHECK(MPI_Comm_size(MPI_COMM_WORLD, &nRanks));


  //calculating localRank based on hostname which is used in selecting a GPU
  uint64_t hostHashs[nRanks];
  char hostname[1024];
  getHostName(hostname, 1024);
  hostHashs[myRank] = getHostHash(hostname);
  MPICHECK(MPI_Allgather(MPI_IN_PLACE, 0, MPI_DATATYPE_NULL, hostHashs, sizeof(uint64_t), MPI_BYTE, MPI_COMM_WORLD));
  for (int p=0; p<nRanks; p++) {
      if (p == myRank) break;
      if (hostHashs[p] == hostHashs[myRank]) localRank++;
  }


  mcclUniqueId id;
  mcclComm_t comm;
  float *sendbuff, *recvbuff;
  musaStream_t s;


  //get MCCL unique ID at rank 0 and broadcast it to all others
  if (myRank == 0) mcclGetUniqueId(&id);
  MPICHECK(MPI_Bcast((void *)&id, sizeof(id), MPI_BYTE, 0, MPI_COMM_WORLD));


  //picking a GPU based on localRank, allocate device buffers
  MUSACHECK(musaSetDevice(localRank));
  MUSACHECK(musaMalloc(&sendbuff, size * sizeof(float)));
  MUSACHECK(musaMalloc(&recvbuff, size * sizeof(float)));
  MUSACHECK(musaStreamCreate(&s));


  //initializing MCCL
  MCCLCHECK(mcclCommInitRank(&comm, nRanks, id, myRank));


  //communicating using MCCL
  MCCLCHECK(mcclAllReduce((const void*)sendbuff, (void*)recvbuff, size, mcclFloat, mcclSum,
        comm, s));


  //completing MCCL operation by synchronizing on the MUSA stream
  MUSACHECK(musaStreamSynchronize(s));


  //free device buffers
  MUSACHECK(musaFree(sendbuff));
  MUSACHECK(musaFree(recvbuff));


  //finalizing MCCL
  mcclCommDestroy(comm);


  //finalizing MPI
  MPICHECK(MPI_Finalize());


  printf("[MPI Rank %d] Success \n", myRank);
  return 0;
}

示例 3：每个线程多个设备

您可以结合使用多个进程或线程和每个进程或线程的多个设备。在这种情况下，我们需要使用组语义。

以下示例结合了 MPI 和每个进程的多个设备（=MPI rank）。

首先，我们获取有关进程的 MPI 信息：

int myRank, nRanks;
MPI_Comm_rank(MPI_COMM_WORLD, &myRank);
MPI_Comm_size(MPI_COMM_WORLD, &nRanks);

接下来，单个 rank 将创建一个唯一 ID 并将其发送给所有其他 rank 以确保每个人都拥有它：

mcclUniqueId id;
if (myRank == 0) mcclGetUniqueId(&id);
MPI_Bcast(id, sizeof(id), MPI_BYTE, 0, 0, MPI_COMM_WORLD);

然后，我们创建我们的 ngpus 通信器对象，它们是 更大的 ngpus*nRanks 组的一部分：

mcclComm_t comms[ngpus];
mcclGroupStart();
for (int i=0; i<ngpus; i++) {
  musaSetDevice(devs[i]);
  mcclCommInitRank(comms+i, ngpus*nRanks, id, myRank*ngpus+i);
}
mcclGroupEnd();

相关链接：mcclGroupStart、mcclGroupEnd、mcclCommInitRank

接下来，我们使用单个线程和组调用来调用 MCCL 集体操作， 或者使用多个线程，每个线程都提供一个 comm 对象。

在程序结束时，我们销毁所有通信器对象：

for (int i=0; i<ngpus; i++)
  mcclCommDestroy(comms[i]);

相关链接：mcclCommDestroy

以下代码展示了一个完整的工作示例，其中包含多个 MPI 进程和每个进程多个设备：

#include <stdio.h>
#include "musa_runtime.h"
#include "mccl.h"
#include "mpi.h"
#include <unistd.h>
#include <stdint.h>


#define MPICHECK(cmd) do {                          \
  int e = cmd;                                      \
  if( e != MPI_SUCCESS ) {                          \
    printf("Failed: MPI error %s:%d '%d'\n",        \
        __FILE__,__LINE__, e);   \
    exit(EXIT_FAILURE);                             \
  }                                                 \
} while(0)


#define MUSACHECK(cmd) do {                         \
  musaError_t e = cmd;                              \
  if( e != musaSuccess ) {                          \
    printf("Failed: musa error %s:%d '%s'\n",             \
        __FILE__,__LINE__,musaGetErrorString(e));   \
    exit(EXIT_FAILURE);                             \
  }                                                 \
} while(0)


#define MCCLCHECK(cmd) do {                         \
  mcclResult_t r = cmd;                             \
  if (r!= mcclSuccess) {                            \
    printf("Failed, MCCL error %s:%d '%s'\n",             \
        __FILE__,__LINE__,mcclGetErrorString(r));   \
    exit(EXIT_FAILURE);                             \
  }                                                 \
} while(0)


static uint64_t getHash(const char* string) {
  // Based on DJB2a, result = result * 33 ^ char
  uint64_t result = 5381;
  for (int c = 0; string[c] != '\0'; c++){
    result = ((result << 5) + result) ^ string[c];
  }
  return result;
}

/* Generate a hash of the unique identifying string for this host
  *
  * that will be unique for both bare-metal and container instances
  * Equivalent of a hash of;
  *
  * $(hostname)$(cat /proc/sys/kernel/random/boot_id)
  *
  */
#define HOSTID_FILE "/proc/sys/kernel/random/boot_id"
static uint64_t getHostHash(const char* hostname) {
  char hostHash[1024];

  // Fall back is the hostname if something fails
  (void) strncpy(hostHash, hostname, sizeof(hostHash));
  int offset = strlen(hostHash);

  FILE *file = fopen(HOSTID_FILE, "r");
  if (file != NULL) {
    char *p;
    if (fscanf(file, "%ms", &p) == 1) {
        strncpy(hostHash+offset, p, sizeof(hostHash)-offset-1);
        free(p);
    }
  }
  fclose(file);

  // Make sure the string is terminated
  hostHash[sizeof(hostHash)-1]='\0';

  return getHash(hostHash, strlen(hostHash));
}

static void getHostName(char* hostname, int maxlen) {
  gethostname(hostname, maxlen);
  for (int i=0; i< maxlen; i++) {
    if (hostname[i] == '.') {
        hostname[i] = '\0';
        return;
    }
  }
}


int main(int argc, char* argv[])
{
  int size = 32*1024*1024;


  int myRank, nRanks, localRank = 0;


  //initializing MPI
  MPICHECK(MPI_Init(&argc, &argv));
  MPICHECK(MPI_Comm_rank(MPI_COMM_WORLD, &myRank));
  MPICHECK(MPI_Comm_size(MPI_COMM_WORLD, &nRanks));


  //calculating localRank which is used in selecting a GPU
  uint64_t hostHashs[nRanks];
  char hostname[1024];
  getHostName(hostname, 1024);
  hostHashs[myRank] = getHostHash(hostname);
  MPICHECK(MPI_Allgather(MPI_IN_PLACE, 0, MPI_DATATYPE_NULL, hostHashs, sizeof(uint64_t), MPI_BYTE, MPI_COMM_WORLD));
  for (int p=0; p<nRanks; p++) {
      if (p == myRank) break;
      if (hostHashs[p] == hostHashs[myRank]) localRank++;
  }


  //each process is using two GPUs
  int nDev = 2;


  float** sendbuff = (float**)malloc(nDev * sizeof(float*));
  float** recvbuff = (float**)malloc(nDev * sizeof(float*));
  musaStream_t* s = (musaStream_t*)malloc(sizeof(musaStream_t)*nDev);


  //picking GPUs based on localRank
  for (int i = 0; i < nDev; ++i) {
    MUSACHECK(musaSetDevice(localRank*nDev + i));
    MUSACHECK(musaMalloc(sendbuff + i, size * sizeof(float)));
    MUSACHECK(musaMalloc(recvbuff + i, size * sizeof(float)));
    MUSACHECK(musaMemset(sendbuff[i], 1, size * sizeof(float)));
    MUSACHECK(musaMemset(recvbuff[i], 0, size * sizeof(float)));
    MUSACHECK(musaStreamCreate(s+i));
  }


  mcclUniqueId id;
  mcclComm_t comms[nDev];


  //generating MCCL unique ID at one process and broadcasting it to all
  if (myRank == 0) mcclGetUniqueId(&id);
  MPICHECK(MPI_Bcast((void *)&id, sizeof(id), MPI_BYTE, 0, MPI_COMM_WORLD));


  //initializing MCCL, group API is required around mcclCommInitRank as it is
  //called across multiple GPUs in each thread/process
  MCCLCHECK(mcclGroupStart());
  for (int i=0; i<nDev; i++) {
      MUSACHECK(musaSetDevice(localRank*nDev + i));
      MCCLCHECK(mcclCommInitRank(comms+i, nRanks*nDev, id, myRank*nDev + i));
  }
  MCCLCHECK(mcclGroupEnd());


  //calling MCCL communication API. Group API is required when using
  //multiple devices per thread/process
  MCCLCHECK(mcclGroupStart());
  for (int i=0; i<nDev; i++)
      MCCLCHECK(mcclAllReduce((const void*)sendbuff[i], (void*)recvbuff[i], size, mcclFloat, mcclSum,
            comms[i], s[i]));
  MCCLCHECK(mcclGroupEnd());


  //synchronizing on MUSA stream to complete MCCL communication
  for (int i=0; i<nDev; i++)
      MUSACHECK(musaStreamSynchronize(s[i]));


  //freeing device memory
  for (int i=0; i<nDev; i++) {
      MUSACHECK(musaFree(sendbuff[i]));
      MUSACHECK(musaFree(recvbuff[i]));
  }


  //finalizing MCCL
  for (int i=0; i<nDev; i++) {
      mcclCommDestroy(comms[i]);
  }


  //finalizing MPI
  MPICHECK(MPI_Finalize());


  printf("[MPI Rank %d] Success \n", myRank);
  return 0;
}

示例 4：每个设备多个通信器

MCCL 允许用户为每个设备创建多个通信器。以下代码展示了一个示例，其中包含多个 MPI 进程，每个进程一个设备，每个设备多个通信器：

    // blocking communicators
    MUSACHECK(musaSetDevice(localRank));
    for (int i = 0; i < commNum; ++i) {
      if (myRank == 0) mcclGetUniqueId(&id);
      MPICHECK(MPI_Bcast((void *)&id, sizeof(id), MPI_BYTE, 0, MPI_COMM_WORLD));
      MCCLCHECK(mcclCommInitRank(&blockingComms[i], nRanks, id, myRank));
    }

    // non-blocking communicators
    MUSACHECK(musaSetDevice(localRank));
    mcclConfig_t config = MCCL_CONFIG_INITIALIZER;
    config.blocking = 0;
    for (int i = 0; i < commNum; ++i) {
      if (myRank == 0) mcclGetUniqueId(&id);
      MPICHECK(MPI_Bcast((void *)&id, sizeof(id), MPI_BYTE, 0, MPI_COMM_WORLD));
      MCCLCHECK(mcclCommInitRankConfig(&nonblockingComms[i], nRanks, id, myRank, &config));
      do {
        MCCLCHECK(mcclCommGetAsyncError(nonblockingComms[i], &state));
      } while(state == mcclInProgress && checkTimeout() != true);
    }

相关链接：mcclConfig_t、mcclCommInitRankConfig、mcclCommGetAsyncError、mcclInProgress

checkTimeout() 应该是用户定义的函数。有关更多非阻塞通信器的使用，请查看 容错。此外，如果您想要拆分通信器而不是创建一个新的，请查看 mcclCommSplit()

通信示例

以下示例展示了执行 MCCL 集体操作的常见模式。

示例 1：每个进程或线程一个设备

如果您每个线程或进程都有一个设备，那么每个线程都为其设备调用集体操作，例如，AllReduce：

mcclAllReduce(sendbuff, recvbuff, count, datatype, op, comm, stream);

相关链接：mcclAllReduce

调用后，操作已排入 MUSA stream 队列。 因此，如果您想要等待操作完成，可以调用 musaStreamSynchronize：

musaStreamSynchronize(stream);

有关 MPI 和每个 MPI 进程一个设备的完整工作示例，请参阅"示例 2：每个进程或线程一个设备"。

示例 2：每个线程多个设备

当单个线程管理多个设备时，您需要使用组语义同时在多个设备上启动操作：

mcclGroupStart();
for (int i=0; i<ngpus; i++)
  mcclAllReduce(sendbuffs[i], recvbuff[i], count, datatype, op, comms[i], streams[i]);
mcclGroupEnd();

相关链接：mcclGroupStart、mcclGroupEnd、mcclAllReduce

mcclGroupEnd 之后，所有操作都已排入 MUSA stream 队列。 因此，您现在可以调用 musaStreamSynchronize，如果您想要等待操作完成：

for (int i=0; i<ngpus; i++)
  musaStreamSynchronize(streams[i]);

有关 MPI 和每个 MPI 进程多个设备的完整工作示例，请参阅 示例 3：每个线程多个设备。