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

Add HDP flush for gfx940 (#1434)

Browse Source 
* Fix collective trace

* Use nontemporal for st_global

* Fix previous commit

* Add HDP flush to data receive path

* Fix previous commit

* Control flushing by NCCL_NET_FORCE_FLUSH and RCCL_NET_HDP_FLUSH

* Introduce RCCL_NET_HDP_FLUSH and RCCL_NET_GDR_FLUSH

Both are on by default. Turn both off will skip all flush will likely
result in data error.

* Enable GDR copy by default

* Remove GDR flush env var because it is disabled by GDC flush

* Output kernel collective trace at comm destroy by default

* Limit kernel timeout messages to 100

* Use system relaxed atomic for loadInt

* Refine timeout messages and use atomic for setting offset from CPU

* Add kernel trace for barrier timeout

* Add backup barrier to avoid race in atomicAdd

* Use different counters for different warps

* Rework barrier implementation

* Fix for other GFX

* Use __hip_atomic_store and __hip_atomic_load

* Fix bug in previous commit

* Don't reset barrier values in running kernel

* Update trace format

* Fix typo

* Switch back to hip_atomic_fetch_add

* Use same barrier implementation for all GFX

* Remove extra threadfence

* Turn off HDP flush by default

Please use RCCL_NET_HDP_FLUSH=1 to switch on HDP flush

* Remove unnecessary changes from alterative barrier implementation

* Added back __threadfence_block

* Revert back to threadfence for gfx other than gfx94x
This commit is contained in:
Wenkai Du
2025-01-31 07:51:10 -08:00
committed by GitHub
parent ad8012f2fc
commit caba0bc049
8 changed files with 87 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/device/primitives.h
+24 -19
View File
@@ -16,37 +16,42 @@
#define NCCL_SPINS_BEFORE_CHECK_ABORT 1000000
#if defined(__gfx940__) || defined(__gfx941__) || defined(__gfx942__)
#define barrier_by_group() do { \
if (nthreads == NCCL_MAX_NTHREADS) { \
__builtin_amdgcn_s_barrier(); \
} else { \
const int w = threadIdx.x/WARP_SIZE; \
const int wid = threadIdx.x%WARP_SIZE; \
if (wid == 0) { \
barrier_next[w] += nthreads/WARP_SIZE; \
atomicAdd((unsigned long long *)barriers, 1); \
while (atomicAdd((unsigned long long *)barriers, 0) < barrier_next[w]) __builtin_amdgcn_s_sleep(1); \
__asm__ __volatile__("s_wakeup"); \
} \
} \
} while (0)
#define __THREAD_FENCE __threadfence_block()
#else
#define __THREAD_FENCE __threadfence()
#endif
#define barrier_by_group() do { \
if (nthreads == NCCL_MAX_NTHREADS) { \
__threadfence(); __builtin_amdgcn_s_barrier(); \
__THREAD_FENCE; __builtin_amdgcn_s_barrier(); \
} else { \
const int w = threadIdx.x/WARP_SIZE; \
const int wid = threadIdx.x%WARP_SIZE; \
__threadfence(); \
if (wid == 0) { \
barrier_next[w] += nthreads/WARP_SIZE; \
atomicAdd((unsigned long long *)barriers, 1); \
while (atomicAdd((unsigned long long *)barriers, 0) < barrier_next[w]) __builtin_amdgcn_s_sleep(1); \
__hip_atomic_fetch_add(barriers, 1, __ATOMIC_RELEASE, __HIP_MEMORY_SCOPE_WORKGROUP); \
int spins = 0; \
int rate_limit = 50; \
__THREAD_FENCE; \
while (__hip_atomic_load(barriers, __ATOMIC_ACQUIRE, __HIP_MEMORY_SCOPE_WORKGROUP) < barrier_next[w]) { \
spins++; \
if (spins == NCCL_SPINS_BEFORE_CHECK_ABORT) { \
if (__atomic_load_n(ncclShmem.comm.abortFlag, __ATOMIC_SEQ_CST)) { \
ncclShmem.aborted = 1; \
break; \
} \
spins = 0; \
} \
if (spins == 0 && rate_limit > 0) { \
rate_limit --; \
traceData(__LINE__, threadIdx.x, __hip_atomic_load(barriers, __ATOMIC_ACQUIRE, __HIP_MEMORY_SCOPE_WORKGROUP), barrier_next[w]); \
} \
__builtin_amdgcn_s_sleep(1); \
} \
__asm__ __volatile__("s_wakeup"); \
} \
} \
} while (0)
#endif
/* Protocol classes: ProtoSimple, ProtoLL, ProtoLL128
* We use these as template args to the Primtiives class instead of integral