Skip to content

Commit 94a680d

Browse files
MC952-archMC952-arch
authored
Support cluster split functionality and add gpu check in CI (#169)
* support cluster split functionality * abort if nclusters > n-inter-ranks * add gpu check in CI * update README file
1 parent b662853 commit 94a680d

7 files changed

Lines changed: 132 additions & 19 deletions

File tree

‎.github/workflows/unit-test.yml‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -81,4 +81,3 @@ jobs:
8181
run: |
8282
cd /__w/FlagCX/FlagCX/test/unittest
8383
mpirun -np 8 ./build/bin/main
84-

‎README.md‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
[<img src="flagopen.png">](https://flagopen.baai.ac.cn/)
22

33
## Latest News
4-
- **[2025/04]** Released [v0.1](https://github.com/FlagOpen/FlagCX/tree/release/v0.1):
4+
- **[2025/04]** Released [v0.1](https://github.com/FlagOpen/FlagCX/tree/release/v0.1):
55
- Supports five native communications libraries with automatic topology detection.
66
- Delivers 11 heterogeneous collective communication algorithms, including both P2P and collective ops.
77
- Provides a full-stack open-source solution, FlagScale + FlagCX, for efficient heterogeneous training.
@@ -20,9 +20,9 @@ FlagCX leverages native collective communications libraries to provide the full
2020
| Mode | Homo | Homo | Homo | Homo | Homo | Homo | Hetero | Hetero | Hetero |
2121
| send | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ |✓ | ✓ | ✓ |
2222
| recv | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ |✓ | ✓ | ✓ |
23-
| broadcast | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ |✘ | ✘ | ✓ |
24-
| gather | ✓ | ✓ | ✓ | ✓ | ✘ | ✓ |✘ | ✘ | ✓ |
25-
| scatter | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ |✘ | ✘ | ✓ |
23+
| broadcast | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ |✓ | ✘ | ✓ |
24+
| gather | ✓ | ✓ | ✓ | ✓ | ✘ | ✓ |✓ | ✘ | ✓ |
25+
| scatter | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ |✓ | ✘ | ✓ |
2626
| reduce | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ |✓ | ✘ | ✓ |
2727
| allreduce | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ |✓ | ✓ | ✓ |
2828
| allgather | ✓ | ✓ | ✓ | ✓ | ✓ | ✓ |✓ | ✓ | ✓ |

‎flagcx/core/c2c_algo.cc‎

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1443,7 +1443,7 @@ flagcxResult_t flagcxC2cPlanner::execute(const void *sendbuff, void *recvbuff,
14431443
if (redOp_ != flagcxRedNoOp) {
14441444
if (redOp_ != flagcxSum && redOp_ != flagcxMax && redOp_ != flagcxMin) {
14451445
WARN("Unsupported reduction operation %d", redOp_);
1446-
return flagcxInvalidArgument;
1446+
return flagcxNotSupported;
14471447
}
14481448
}
14491449

@@ -1454,6 +1454,24 @@ flagcxResult_t flagcxC2cPlanner::execute(const void *sendbuff, void *recvbuff,
14541454
return flagcxInvalidArgument;
14551455
}
14561456

1457+
// commOp validation
1458+
// abort if nclusters > n-inter-ranks
1459+
if (commOp_ == flagcxCommOpAllReduce ||
1460+
commOp_ == flagcxCommOpReduceScatter) {
1461+
int clusterCountValid_ = 1;
1462+
for (int i = 0; i < comm_->nclusters; ++i) {
1463+
if (comm_->nclusters > int(clusterInterRankList_[i].size())) {
1464+
clusterCountValid_ = 0;
1465+
break;
1466+
}
1467+
}
1468+
if (!clusterCountValid_) {
1469+
WARN("Unsupported communication operation %d since cluster count is "
1470+
"larger than inter-rank count",
1471+
commOp_);
1472+
return flagcxNotSupported;
1473+
}
1474+
}
14571475
// sendrecv counts and displs validation and initialization
14581476
if (commOp_ == flagcxCommOpAlltoAllv) {
14591477
if (sendCounts == nullptr || sDispls == nullptr || recvCounts == nullptr ||

‎flagcx/core/cluster.cc‎

Lines changed: 61 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,25 @@
11
#include "cluster.h"
22
#include <cstring>
33

4+
flagcxResult_t parseClusterSplitList(const char *input,
5+
std::vector<int> &output) {
6+
output.clear();
7+
std::stringstream ss(input);
8+
std::string token;
9+
10+
while (std::getline(ss, token, ',')) {
11+
try {
12+
int value = std::stoi(token);
13+
output.push_back(value);
14+
} catch (const std::exception &e) {
15+
WARN("Invalid cluster split info, its format should be like '2,4,8,...'");
16+
return flagcxSystemError;
17+
}
18+
}
19+
20+
return flagcxSuccess;
21+
}
22+
423
flagcxResult_t flagcxCollectClusterInfos(const flagcxVendor *allData,
524
flagcxCommunicatorType_t *type,
625
int *homo_rank, int *homo_root_rank,
@@ -11,7 +30,7 @@ flagcxResult_t flagcxCollectClusterInfos(const flagcxVendor *allData,
1130
*homo_root_rank = 0;
1231
*homo_ranks = 1;
1332
*cluster_id = 0;
14-
*cluster_inter_rank = -1;
33+
*cluster_inter_rank = -1; // deprecated, to be removed
1534
*ncluster = 1;
1635
*type = flagcxCommunicatorHomo;
1736

@@ -55,21 +74,50 @@ flagcxResult_t flagcxCollectClusterInfos(const flagcxVendor *allData,
5574
}
5675

5776
if (*type == flagcxCommunicatorHybrid) {
58-
const char *useDev = flagcxGetEnv("FLAGCX_USEDEV");
59-
int useDev_;
60-
if (useDev == NULL) {
61-
useDev_ = -1;
62-
} else {
63-
useDev_ = std::stoi(useDev);
77+
*cluster_id = currCluster;
78+
*ncluster = numClusters;
79+
}
80+
81+
// split and obtain sub-clusters
82+
const char *clusterSplitInfo = flagcxGetEnv("FLAGCX_CLUSTER_SPLIT_LIST");
83+
if (clusterSplitInfo != NULL) {
84+
std::vector<int> clusterSplitList;
85+
FLAGCXCHECK(parseClusterSplitList(clusterSplitInfo, clusterSplitList));
86+
if (*ncluster != int(clusterSplitList.size())) {
87+
WARN("Invalid cluster split info, its length should be equal to the "
88+
"number of homogeneous cluster");
89+
return flagcxSystemError;
6490
}
65-
if (*homo_rank == useDev_) {
66-
*cluster_inter_rank = rank;
91+
92+
int subClusterId = 0;
93+
for (int i = 0; i < currCluster; ++i) {
94+
subClusterId += clusterSplitList[i];
6795
}
68-
if (*homo_ranks <= useDev_ && *homo_rank == *homo_ranks - 1) {
69-
*cluster_inter_rank = rank;
96+
int subHomoRanks = (*homo_ranks) / clusterSplitList[currCluster];
97+
int hasRes =
98+
(((*homo_rank) / subHomoRanks) >= clusterSplitList[currCluster]) ? 1
99+
: 0;
100+
subClusterId += (hasRes == 1) ? ((*homo_rank) / subHomoRanks) - 1
101+
: ((*homo_rank) / subHomoRanks);
102+
int subHomoRank = (hasRes == 1)
103+
? subHomoRanks + ((*homo_rank) % subHomoRanks)
104+
: ((*homo_rank) % subHomoRanks);
105+
int subHomoRootRank = rank - subHomoRank;
106+
if (hasRes == 1 ||
107+
((*homo_rank) / subHomoRanks) == clusterSplitList[currCluster] - 1) {
108+
subHomoRanks += (*homo_ranks) % clusterSplitList[currCluster];
70109
}
71-
*cluster_id = currCluster;
72-
*ncluster = numClusters;
110+
int subNClusters = 0;
111+
for (int i = 0; i < (*ncluster); ++i) {
112+
subNClusters += clusterSplitList[i];
113+
}
114+
*homo_rank = subHomoRank;
115+
*homo_root_rank = subHomoRootRank;
116+
*homo_ranks = subHomoRanks;
117+
*cluster_id = subClusterId;
118+
*ncluster = subNClusters;
119+
*type =
120+
(subNClusters > 1) ? flagcxCommunicatorHybrid : flagcxCommunicatorHomo;
73121
}
74122

75123
return flagcxSuccess;

‎flagcx/core/cluster.h‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,14 @@
44
#include "adaptor.h"
55
#include "flagcx.h"
66
#include "param.h"
7+
#include <iostream>
78
#include <map>
9+
#include <sstream>
810
#include <string>
11+
#include <vector>
12+
13+
flagcxResult_t parseClusterSplitList(const char *input,
14+
std::vector<int> &output);
915

1016
flagcxResult_t flagcxCollectClusterInfos(const flagcxVendor *allData,
1117
flagcxCommunicatorType_t *type,

‎test/script/_gpu_check.sh‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
#!/bin/bash
2+
3+
wait_for_gpu() {
4+
gpu_count=$(nvidia-smi --query-gpu=name --format=csv,noheader | wc -l)
5+
6+
memory_usage_max=30000
7+
8+
while true; do
9+
10+
IFS=$'\n' read -d '' -r -a memory_usage_array <<< "$(nvidia-smi --query-gpu=memory.used --format=csv,noheader,nounits)"
11+
IFS=$'\n' read -d '' -r -a memory_total_array <<< "$(nvidia-smi --query-gpu=memory.total --format=csv,noheader,nounits)"
12+
13+
need_wait=false
14+
15+
for ((i=0; i<$gpu_count; i++)); do
16+
17+
memory_usage_i=$((${memory_usage_array[$i]}))
18+
memory_total_i=$((${memory_total_array[$i]}))
19+
memory_remin_i=$(($memory_total_i-$memory_usage_i))
20+
21+
if [ $memory_remin_i -lt $memory_usage_max ]; then
22+
need_wait=true
23+
fi
24+
25+
done
26+
27+
if [ "$need_wait" = false ]; then
28+
break
29+
fi
30+
31+
echo "wait for gpu free"
32+
sleep 1m
33+
34+
unset memory_usage_array
35+
unset memory_total_array
36+
37+
done
38+
39+
echo "All gpu is free"
40+
}

‎test/script/auto_script.sh‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@ if [ $? -ne 0 ]; then
1919
exit 1
2020
fi
2121

22+
source ../script/_gpu_check.sh
23+
wait_for_gpu
2224

2325
mpirun -np 8 ./test_alltoall -b 128M -e 1G -f 2 -p 1
2426
if [ $? -ne 0 ]; then

0 commit comments

Comments
 (0)