[ao] 読み進めメモ

[aokomoriuta]

allreduceを上から追っておきます

[aokomoriuta]

すごく初歩から

• allreducの宣言

  nccl/src/nccl.h.in

  Lines 185 to 186 in 7c72dee

  	ncclResult_t ncclAllReduce(const void* sendbuff, void* recvbuff, size_t count,
  	ncclDataType_t datatype, ncclRedOp_t op, ncclComm_t comm, cudaStream_t stream);

• 実体

  nccl/src/collectives/all_reduce.cc

  Lines 12 to 18 in 7c72dee

  	ncclResult_t ncclAllReduce(const void* sendbuff, void* recvbuff, size_t count,
  	ncclDataType_t datatype, ncclRedOp_t op, ncclComm* comm, cudaStream_t stream) {
  	struct ncclInfo info = { ncclCollAllReduce, "AllReduce",
  	sendbuff, recvbuff, count, datatype, op, 0, comm, stream, /* Args */
  	ALLREDUCE_CHUNKSTEPS, ALLREDUCE_SLICESTEPS };
  	return ncclEnqueueCheck(&info);
  	}

[aokomoriuta]

順に、NCCL_APIについて

nccl/src/include/core.h

Lines 25 to 39 in 9db4b1d

	#ifdef PROFAPI
	#define NCCL_API(ret, func, args...) \
	__attribute__ ((visibility("default"))) \
	__attribute__ ((alias(#func))) \
	ret p##func (args); \
	extern "C" \
	__attribute__ ((visibility("default"))) \
	__attribute__ ((weak)) \
	ret func(args)
	#else
	#define NCCL_API(ret, func, args...) \
	extern "C" \
	__attribute__ ((visibility("default"))) \
	ret func(args)
	#endif // end PROFAPI

PROFAPIコンパイルスイッチは

nccl/makefiles/common.mk

Line 13 in 9db4b1d

PROFAPI ?= 0

にあるが、よくわからないので飛ばす（普通は使われないっぽい）。

通常版の

nccl/src/include/core.h

Lines 36 to 38 in 9db4b1d

	extern "C" \
	__attribute__ ((visibility("default"))) \
	ret func(args)

は

• extern C
• シンボルエクスポート

しているだけ。後者はgcc拡張で、実はNCCLはattributeで明示的に指定しない関数は全部ライブラリの外からは見えなくなっている

nccl/makefiles/common.mk

Line 46 in 9db4b1d

CXXFLAGS := -DCUDA_MAJOR=$(CUDA_MAJOR) -DCUDA_MINOR=$(CUDA_MINOR) -fPIC -fvisibility=hidden

[aokomoriuta]

ncclInfoは単に引数をまとめただけのもの

nccl/src/include/info.h

Lines 23 to 43 in 9db4b1d

	struct ncclInfo {
	ncclColl_t coll;
	const char* opName;
	// NCCL Coll Args
	const void* sendbuff;
	void* recvbuff;
	size_t count;
	ncclDataType_t datatype;
	ncclRedOp_t op;
	int root;
	ncclComm_t comm;
	cudaStream_t stream;
	// Algorithm details
	int chunkSteps;
	int sliceSteps;
	// Computed later
	ncclPattern_t pattern;
	size_t nBytes;
	int nstepsPerLoop;
	int nchunksPerLoop;
	};

大体はncclAllreduceの引数をそのまま転送しているが違うのは以下

• ncclCollAllReduce：

  nccl/src/include/devcomm.h

  Line 16 in 9db4b1d

  typedef enum { ncclCollBroadcast, ncclCollReduce, ncclCollAllGather, ncclCollReduceScatter, ncclCollAllReduce, ncclCollCount } ncclColl_t;

  にあるけどなんだか分からん（あとで出てきそう）
• opName：なにかに使うんだろうか（エラーメッセージかな）
• op, 0, commの0はルートプロセス番号っぽい（allreduceにルートもなにもなくないか？）

[aokomoriuta]

ncclEnqueueCheck

nccl/src/enqueue.cc

Lines 409 to 442 in 9db4b1d

	ncclResult_t ncclEnqueueCheck(struct ncclInfo* info) {
	if (info->comm == NULL) return ncclInvalidArgument;
	INFO(NCCL_COLL,"%s: opCount %lx sendbuff %p recvbuff %p count %zi datatype %d op %d root %d comm %p [nranks=%d] stream %p",
	info->opName, info->comm->opCount, info->sendbuff, info->recvbuff, info->count,
	info->datatype, info->op, info->root, info->comm, info->comm->nRanks, info->stream);
	// Launch asynchronously if needed
	if (ncclAsyncMode()) {
	ncclResult_t ret = ncclSuccess;
	int savedDev = -1;
	if (info->comm->checkPointers) {
	CUDACHECKGOTO(cudaGetDevice(&savedDev), ret, end);
	CUDACHECKGOTO(cudaSetDevice(info->comm->cudaDev), ret, end);
	}
	// Check arguments
	NCCLCHECKGOTO(ArgsCheck(info), ret, end);
	// Always register comm even in case of error to make sure ncclGroupEnd
	// cleans it up.
	NCCLCHECKGOTO(ncclAsyncColl(info->comm), ret, end);
	NCCLCHECKGOTO(saveKernel(info), ret, end);
	end:
	if (savedDev != -1) CUDACHECK(cudaSetDevice(savedDev));
	ncclAsyncErrCheck(ret);
	return ret;
	} else {
	NCCLCHECK(ArgsCheck(info));
	NCCLCHECK(saveKernel(info));
	NCCLCHECK(ncclBarrierEnqueue(info->comm));
	NCCLCHECK(ncclBarrierEnqueueWait(info->comm));
	NCCLCHECK(ncclEnqueueEvents(info->comm));
	return ncclSuccess;
	}
	}

Asyncモードかどうかで別れている。asyncモードとは、1プロセスで複数GPUを使う時に、ncclAllreduceを非同期にして、全GPUでncclAllreduceを呼び出すためのもの。複雑になるすぎるので割愛する。

ということで、ここでの処理は

nccl/src/enqueue.cc

Lines 435 to 439 in 9db4b1d

	NCCLCHECK(ArgsCheck(info));
	NCCLCHECK(saveKernel(info));
	NCCLCHECK(ncclBarrierEnqueue(info->comm));
	NCCLCHECK(ncclBarrierEnqueueWait(info->comm));
	NCCLCHECK(ncclEnqueueEvents(info->comm));

[aokomoriuta]

ArgsCheck

nccl/src/misc/argcheck.cc

Lines 35 to 68 in 9db4b1d

	ncclResult_t ArgsCheck(struct ncclInfo* info) {
	NCCLCHECK(PtrCheck(info->comm, info->opName, "comm"));
	// First, the easy ones
	if (info->root < 0 || info->root >= info->comm->nRanks) {
	WARN("%s : invalid root %d (root should be in the 0..%d range)", info->opName, info->root, info->comm->nRanks);
	return ncclInvalidArgument;
	}
	if (info->datatype < 0 || info->datatype >= ncclNumTypes) {
	WARN("%s : invalid type %d", info->opName, info->datatype);
	return ncclInvalidArgument;
	}
	// Type is OK, compute nbytes. Convert Allgather/Broadcast calls to chars.
	info->nBytes = info->count * ncclTypeSize(info->datatype);
	if (info->coll == ncclCollAllGather || info->coll == ncclCollBroadcast) {
	info->count = info->nBytes;
	info->datatype = ncclInt8;
	}
	if (info->coll == ncclCollAllGather || info->coll == ncclCollReduceScatter) info->nBytes *= info->comm->nRanks; // count is per rank
	if (info->op < 0 || info->op >= ncclNumOps) {
	WARN("%s : invalid reduction operation %d", info->opName, info->op);
	return ncclInvalidArgument;
	}
	if (info->comm->checkPointers) {
	// Check CUDA device pointers
	if (info->coll != ncclCollBroadcast || info->comm->rank == info->root) {
	NCCLCHECK(CudaPtrCheck(info->sendbuff, info->comm, "sendbuff", info->opName));
	}
	if (info->coll != ncclCollReduce || info->comm->rank == info->root) {
	NCCLCHECK(CudaPtrCheck(info->recvbuff, info->comm, "recvbuff", info->opName));
	}
	}
	return ncclSuccess;

1. ルートプロセス番号が0-プロセス数の間か
2. datatypeが定義されているものか
3. 通信長を決定
4. 演算子が定義されているものか
5. send/recvがCUDAのデバイスポインタかどうか

[aokomoriuta]

チャンネルという概念が出てくるが、これは何を指しているのだろう・・・？

[aokomoriuta]

ncclChannelの定義

nccl/src/include/devcomm.h

Lines 147 to 171 in 9db4b1d

	struct ncclChannel {
	union {
	struct {
	struct ncclRing ring;
	struct ncclTree tree;
	int id;
	int nthreads;
	int buffSize;
	// Communication structures
	struct ncclPeer* peers;
	struct ncclPeer* devPeers;
	// Operation list for aggregation
	struct ncclColl* collectives;
	struct ncclColl* devCollectives;
	int collStart;
	int collCount;
	int collFifoHead; // Only used by GPU
	int collFifoTail; // Only used by CPU
	};
	int data[0x80];
	};
	};

通信方法を規定している何かには見えるが・・・？

初期化は

nccl/src/channel.cc

Lines 12 to 34 in 9db4b1d

	ncclResult_t initChannel(struct ncclComm* comm, int channelid) {
	struct ncclChannel* channel = comm->channels+channelid;
	channel->id = channelid;
	// Setup intermediate buffering
	channel->buffSize = ncclParamBuffsize();
	// Ring index to user rank table.
	NCCLCHECK(ncclCudaCalloc(&channel->ring.devUserRanks, comm->nRanks));
	NCCLCHECK(ncclCalloc(&channel->ring.userRanks, comm->nRanks));
	// Communication structures with peers.
	NCCLCHECK(ncclCudaCalloc(&channel->devPeers, comm->nRanks));
	NCCLCHECK(ncclCalloc(&channel->peers, comm->nRanks));
	for (size_t i=0; i<comm->nRanks; ++i) {
	channel->peers[i].send.comm = comm;
	channel->peers[i].recv.comm = comm;
	}
	// Per-channel operation list.
	NCCLCHECK(ncclCudaHostAlloc((void**)&channel->collectives, (void**)&channel->devCollectives, sizeof(struct ncclColl)*NCCL_MAX_OPS));
	return ncclSuccess;
	}

[aokomoriuta]

nccl/src/init.cc

Lines 812 to 818 in 8e04d80

	for (int r=0; r<nrings; r++) {
	struct ncclChannel* channel = comm->channels+r;
	NCCLCHECK(setupChannel(comm, r, rank, nranks, rings+r*nranks, treeIn+r*nranks));
	NCCLCHECK(p2pSetup(comm, channel, 1, &channel->ring.prev, 1, &channel->ring.next));
	NCCLCHECK(p2pSetup(comm, channel, NCCL_MAX_TREE_ARITY, channel->tree.down, 1, &channel->tree.up));
	NCCLCHECK(p2pSetup(comm, channel, 1, &channel->tree.up, NCCL_MAX_TREE_ARITY, channel->tree.down));
	}

で、channelIdにリング番号が入っているのを発見した。もしかして、チャンネル＝リング（ひとつ）か？

[aokomoriuta]

なおリングというのは https://research.preferred.jp/2018/07/prototype-allreduce-library/ にあるやつ。

となると、リングが複数あるのってどういう状況なんだろう？

[aokomoriuta]

チャンネル＝リングと仮定して話を進める。

チャンネル数は上限16なので、それほど多く作るものではないらしい。

そしてSlack等で話した限り、リングは、NVLink等で各GPUが全結合しているのを使い切るために複数作られる模様。

16個しかないのは、NVLinkが16GPUまでしかサポートしてないからか？実際のnringsを決めているのは

nccl/src/misc/rings.cc

Line 204 in 0ceaec9

ncclResult_t ncclGetRings(int* nrings, int* nthreads, int rank, int nranks, int* transports, ncclTvalue_t* values, int* prev, int* next, int* treeIn, int* treeOut) {

だが割と複雑そう。ちょっと脱線するがどうやって決めてるのか興味あるので読み解いてみよう。

[aokomoriuta]

nccl/src/misc/rings.cc

Lines 209 to 225 in 0ceaec9

	char* str = getenv("NCCL_RINGS");
	if (str && strlen(str)>0) {
	int ret = parseRings(str, nrings, nranks, prev, next);
	if (ret == ncclSuccess && *nrings > 0) {
	if (rank == 0) INFO(NCCL_INIT,"%d ring(s) set by environment", *nrings);
	NCCLCHECK(getEnvThreads(nthreads));
	for (int r = 0; r<*nrings; r++) {
	for (int i = 0; i<nranks; i++) {
	if (transports[i*nranks+prev[r*nranks+i]] == 2) treeIn[r*nranks+i] = 1;
	if (transports[i*nranks+next[r*nranks+i]] == 2) treeOut[r*nranks+i] = 1;
	}
	}
	return ncclSuccess;
	}
	if (rank == 0) INFO(NCCL_INIT,"No valid ring found in environment, ignoring");
	*nrings = 0;
	}

の部分。NCCL_RINGS環境変数から設定を読み込んでいる。文字列を解析してリングが作れたら使うし、使えなかった場合はエラーではなくそのまま内部生成に進む。中身はあんまり面白くなさそうなので割愛

[aokomoriuta]

どうやら #1 にあるリング生成アルゴリズムを理解しておく必要がありそう。誰か知ってそうだが、練習（？）も兼ねて自力でやる。

[aokomoriuta]

新しくtransportsという概念が出てきたので整理。transportには3種類ある

• P2P通信
• 共有メモリ(shm)経由通信
• ネットワーク通信

nccl/src/transport.cc

Lines 13 to 17 in 9db4b1d

	struct ncclTransport ncclTransports[NTRANSPORTS] = {
	p2pTransport,
	shmTransport,
	netTransport,
	};

[aokomoriuta]

リングを作る前に、各デバイス（というかランク）間の接続状況を

nccl/src/misc/rings.cc

Lines 131 to 171 in 0ceaec9

	static ncclResult_t fillCoords(int nranks, int* matrix, int* coords, int* rankToIdx, int* idxToRank) {
	int current[NTRANSPORTS];
	int* p2pConnected;
	NCCLCHECK(ncclCalloc(&p2pConnected, nranks));
	for (int i=0; i<NTRANSPORTS; i++) current[i] = 0;
	int curRank = 0, idx = 0;
	while (1) {
	// P2P is handled separately as there is no level below it and we need to
	// cover the case of being connected to another GPU indirectly.
	// So we detect all GPUs in the same P2P domain once and add them all at
	// once.
	isConnected(curRank, p2pConnected, nranks, matrix, 0);
	for (int r=0; r<nranks; r++) {
	if (p2pConnected[r]) {
	NEW_IDX(r);
	curRank = r;
	current[0]++;
	}
	}
	current[0] = 0;
	if (idx == nranks) {
	free(p2pConnected);
	return ncclSuccess;
	}
	// Find next group, either connected through SHM or NET.
	int rank;
	int transport = 1;
	while ((rank = findConnected(curRank, matrix, nranks, transport, coords)) == -1) {
	current[transport] = 0;
	transport++;
	if (transport == NTRANSPORTS) {
	WARN("Error : Could not find transport to connect next group\n");
	free(p2pConnected);
	return ncclInternalError; }
	}
	curRank = rank;
	current[transport]++;
	}
	}

で調べている。