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

Update Barrier and Sync APIs (#73)

Browse Source 
* Add thread, wavefront, and workgroup-level `barrier` APIs in IPC and RO conduits; remove collectives on default context
 - Implemented `barrier` APIs for thread, wavefront, and workgroup scopes
 - Added support into both IPC and RO conduits
 - Added functional tests to cover all `barrier` APIs
 - Removed collective operations on default context

* Add thread, wavefront, and workgroup-level `sync` APIs in IPC and RO conduits.
  - Implemented `sync` APIs for thread, wavefront, and workgroup scopes
  - Added support into both IPC and RO conduits
  - Added functional tests to cover all `sync` APIs

* update naming convention for context-based `barrier` APIs
This commit is contained in:
Avinash Kethineedi
2025-04-08 11:25:31 -05:00
committed by GitHub
parent c652f58cef
commit dc61bca066
16 changed files with 347 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/functional_tests/sync_tester.cpp
+18 -11
View File
@@ -27,9 +27,12 @@
*****************************************************************************/
__global__ void SyncTest(int loop, int skip, long long int *start_time,
long long int *end_time, TestType type,
ShmemContextType ctx_type, rocshmem_team_t *teams) {
ShmemContextType ctx_type, int wf_size,
rocshmem_team_t *teams) {
__shared__ rocshmem_ctx_t ctx;
int t_id = get_flat_block_id();
int wg_id = get_flat_grid_id();
int wf_id = t_id / wf_size;
rocshmem_wg_init();
rocshmem_wg_ctx_create(ctx_type, &ctx);
@@ -39,16 +42,25 @@ __global__ void SyncTest(int loop, int skip, long long int *start_time,
start_time[wg_id] = wall_clock64();
}
__syncthreads();
switch (type) {
case SyncTestType:
if(t_id == 0) {
rocshmem_ctx_team_sync(ctx, teams[wg_id]);
}
break;
case WAVESyncTestType:
if(wf_id == 0) {
rocshmem_ctx_wave_team_sync(ctx, teams[wg_id]);
}
break;
case WGSyncTestType:
rocshmem_ctx_wg_team_sync(ctx, teams[wg_id]);
break;
default:
break;
}
__syncthreads();
}
__syncthreads();
if (hipThreadIdx_x == 0) {
end_time[wg_id] = wall_clock64();
@@ -100,15 +112,10 @@ void SyncTester::launchKernel(dim3 gridSize, dim3 blockSize, int loop,
hipLaunchKernelGGL(SyncTest, gridSize, blockSize, shared_bytes, stream, loop,
args.skip, start_time, end_time, _type, _shmem_context,
team_sync_world_dup);
wf_size, team_sync_world_dup);
num_msgs = loop + args.skip;
num_timed_msgs = loop;
if(_type == SyncTestType) {
num_msgs *= gridSize.x;
num_timed_msgs *= gridSize.x;
}
num_msgs = (loop + args.skip) * gridSize.x;
num_timed_msgs = loop * gridSize.x;
}
void SyncTester::postLaunchKernel() {
tests/functional_tests/team_barrier_tester.cpp
+26 -13
View File
@@ -28,28 +28,41 @@ rocshmem_team_t team_barrier_world_dup;
*****************************************************************************/
__global__ void TeamBarrierTest(int loop, int skip, long long int *start_time,
long long int *end_time,
ShmemContextType ctx_type,
rocshmem_team_t *teams) {
ShmemContextType ctx_type, TestType type,
int wf_size, rocshmem_team_t *teams) {
__shared__ rocshmem_ctx_t ctx;
int t_id = get_flat_block_id();
int wg_id = get_flat_grid_id();
int wf_id = t_id / wf_size;
rocshmem_wg_init();
rocshmem_wg_team_create_ctx(teams[wg_id], ctx_type, &ctx);
int n_pes = rocshmem_ctx_n_pes(ctx);
__syncthreads();
for (int i = 0; i < loop + skip; i++) {
if (i == skip && hipThreadIdx_x == 0) {
start_time[wg_id] = wall_clock64();
}
rocshmem_barrier(teams[wg_id]);
switch (type) {
case TeamBarrierTestType:
if(t_id == 0) {
rocshmem_ctx_barrier(ctx, teams[wg_id]);
}
break;
case TeamWAVEBarrierTestType:
if(wf_id == 0) {
rocshmem_ctx_wave_barrier(ctx, teams[wg_id]);
}
break;
case TeamWGBarrierTestType:
rocshmem_ctx_wg_barrier(ctx, teams[wg_id]);
break;
default:
break;
}
__syncthreads();
}
__syncthreads();
if (hipThreadIdx_x == 0) {
end_time[wg_id] = wall_clock64();
}
@@ -95,10 +108,10 @@ void TeamBarrierTester::launchKernel(dim3 gridSize, dim3 blockSize,
int loop, uint64_t size) {
size_t shared_bytes = 0;
hipLaunchKernelGGL(TeamBarrierTest, gridSize, blockSize,
shared_bytes, stream, loop, args.skip,
start_time, end_time, _shmem_context,
team_barrier_world_dup);
hipLaunchKernelGGL(TeamBarrierTest, gridSize, blockSize, shared_bytes,
stream, loop, args.skip, start_time, end_time,
_shmem_context, _type, wf_size,
team_barrier_world_dup);
num_msgs = (loop + args.skip) * gridSize.x;
num_timed_msgs = loop * gridSize.x;
tests/functional_tests/tester.cpp
+20 -2
View File
@@ -330,6 +330,14 @@ std::vector<Tester*> Tester::create(TesterArguments args) {
if (rank == 0) std::cout << "Team Barrier Test ###" << std::endl;
testers.push_back(new TeamBarrierTester(args));
return testers;
case TeamWAVEBarrierTestType:
if (rank == 0) std::cout << "Team WAVE Barrier Test ###" << std::endl;
testers.push_back(new TeamBarrierTester(args));
return testers;
case TeamWGBarrierTestType:
if (rank == 0) std::cout << "Team WG Barrier Test ###" << std::endl;
testers.push_back(new TeamBarrierTester(args));
return testers;
case SyncAllTestType:
if (rank == 0) std::cout << "SyncAll ###" << std::endl;
testers.push_back(new SyncTester(args));
@@ -346,6 +354,14 @@ std::vector<Tester*> Tester::create(TesterArguments args) {
if (rank == 0) std::cout << "Sync ###" << std::endl;
testers.push_back(new SyncTester(args));
return testers;
case WAVESyncTestType:
if (rank == 0) std::cout << "WAVE Sync ###" << std::endl;
testers.push_back(new SyncTester(args));
return testers;
case WGSyncTestType:
if (rank == 0) std::cout << "WG Sync ###" << std::endl;
testers.push_back(new SyncTester(args));
return testers;
case RandomAccessTestType:
if (rank == 0) std::cout << "Random_Access ###" << std::endl;
testers.push_back(new RandomAccessTester(args));
@@ -528,10 +544,12 @@ bool Tester::peLaunchesKernel() {
(_type == TeamAllToAllTestType) || (_type == TeamFCollectTestType) ||
(_type == PingPongTestType) || (_type == BarrierAllTestType) ||
(_type == WAVEBarrierAllTestType) || (_type == WGBarrierAllTestType) ||
(_type == SyncTestType) || (_type == SyncAllTestType) ||
(_type == SyncTestType) || (_type == WAVESyncTestType) ||
(_type == WGSyncTestType) || (_type == SyncAllTestType) ||
(_type == WAVESyncAllTestType) || (_type == WGSyncAllTestType) ||
(_type == RandomAccessTestType) || (_type == PingAllTestType) ||
(_type == TeamBarrierTestType);
(_type == TeamBarrierTestType) || (_type == TeamWAVEBarrierTestType) ||
(_type == TeamWGBarrierTestType);
return is_launcher;
}
tests/functional_tests/tester.hpp
+5 -1
View File
@@ -92,7 +92,7 @@ enum TestType {
SignalFetchTestType = 55,
WGSignalFetchTestType = 56,
WAVESignalFetchTestType = 57,
TeamBarrierTestType = 58,
TeamWGBarrierTestType = 58,
DefaultCTXGetTestType = 59,
DefaultCTXGetNBITestType = 60,
DefaultCTXPutTestType = 61,
@@ -103,6 +103,10 @@ enum TestType {
WGBarrierAllTestType = 66,
WAVESyncAllTestType = 67,
WGSyncAllTestType = 68,
TeamBarrierTestType = 69,
TeamWAVEBarrierTestType = 70,
WAVESyncTestType = 71,
WGSyncTestType = 72,
};
enum OpType { PutType = 0, GetType = 1 };
tests/functional_tests/tester_arguments.cpp
+6 -2
View File
@@ -88,6 +88,8 @@ TesterArguments::TesterArguments(int argc, char *argv[]) {
case WAVEBarrierAllTestType:
case WGBarrierAllTestType:
case TeamBarrierTestType:
case TeamWAVEBarrierTestType:
case TeamWGBarrierTestType:
case SyncAllTestType:
case WAVESyncAllTestType:
case WGSyncAllTestType:
@@ -140,10 +142,12 @@ void TesterArguments::get_rocshmem_arguments() {
if ((type != BarrierAllTestType) && (type != WAVEBarrierAllTestType) &&
(type != WGBarrierAllTestType) && (type != SyncAllTestType) &&
(type != WAVESyncAllTestType) && (type != WGSyncAllTestType) &&
(type != SyncTestType) && (type != TeamAllToAllTestType) &&
(type != SyncTestType) && (type != WAVESyncTestType) &&
(type != WGSyncTestType) && (type != TeamAllToAllTestType) &&
(type != TeamFCollectTestType) && (type != TeamReductionTestType) &&
(type != TeamBroadcastTestType) && (type != PingAllTestType) &&
(type != TeamBarrierTestType)) {
(type != TeamBarrierTestType) && (type != TeamWAVEBarrierTestType) &&
(type != TeamWGBarrierTestType)) {
if (numprocs != 2) {
if (myid == 0) {
std::cerr << "This test requires exactly two processes, we have "