Add host API for *_on_stream operations (#340)

* Add functional test for barrier_all_on_stream

* Add rocshmem_barrier_all_on_stream support for GDA and RO backends

Implements rocshmem_barrier_all_on_stream operation for
GPU Direct Access and Reverse Offload backends.

Previously, rocshmem_barrier_all_on_stream was only supported for IPC backend.

* Add functional test for rocshmem_broadcastmem_on_stream

* Add host-side rocshmem_broadcastmem_on_stream API

Implement stream-based broadcast collective operation

- Add rocshmem_broadcastmem_on_stream host API and kernel implementation
- Add functional test TeamBroadcastmemOnStreamTester with multi-stream
  support and correctness verification
- Use per-workgroup contexts to avoid contention across parallel streams

API:
rocshmem_broadcastmem_on_stream(team, dest, source, nelems, pe_root, stream)

* Add functional test for rocshmem_getmem_on_stream

* Add host-side rocshmem_getmem_on_stream API

Implement stream-based point-to-point RMA get operation

- Add rocshmem_getmem_on_stream host API and kernel implementation
- Support for asynchronous getmem operations on HIP streams
- Add backend support for GDA, RO, and IPC contexts
- Use work-group collective getmem for efficient memory transfer

API:
rocshmem_getmem_on_stream(dest, source, nelems, pe, stream)

(AI Assist)

* Add host-side rocshmem_putmem_on_stream API

- Add rocshmem_putmem_on_stream for asynchronous remote writes
- Support for concurrent RMA operations on HIP streams
- Add backend support for GDA, RO, and IPC contexts
- Use work-group device collective operation

API:
rocshmem_putmem_on_stream(dest, source, bytes, pe, stream)

(AI Assist)

* Add functional test for rocshmem_putmem_on_stream

* Add host-side rocshmem_putmem_signal_on_stream API

Enables asynchronous putmem operations with signaling on HIP streams.

The implementation includes:
- Kernel wrapper rocshmem_putmem_signal_kernel
- Host interface putmem_signal_on_stream method
- Context layer support across all backends (IPC, GDA, RO)
- Public API

Function signature:
void rocshmem_putmem_signal_on_stream(void *dest, const void *source,
                                      size_t bytes, uint64_t *sig_addr,
                                      uint64_t signal, int sig_op,
                                      int pe, hipStream_t stream);

* Add functional test for rocshmem_putmem_signal_on_stream

* Add host-side rocshmem_signal_wait_until_on_stream API

Enables asynchronous signal wait operations on HIP streams.

The implementation includes:
- Kernel wrapper rocshmem_signal_wait_until_kernel
- Host interface signal_wait_until_on_stream method
- Context layer support across all backends (IPC, GDA, RO)
- Native uint64_t support in wait_until API (generated from P2P_SYNC.py)

Function signature:
void rocshmem_signal_wait_until_on_stream(uint64_t *sig_addr, int cmp,
                                          uint64_t cmp_value,
                                          hipStream_t stream);

(AI Assist)

* Add functional test for rocshmem_signal_wait_until_on_stream

* Add documentation for stream API functions

This commit adds API documentation for the following host-side
stream functions:

- rocshmem_barrier_all_on_stream (collective routines)
- rocshmem_broadcastmem_on_stream (collective routines)
- rocshmem_getmem_on_stream (RMA operations)
- rocshmem_putmem_on_stream (RMA operations)
- rocshmem_putmem_signal_on_stream (signaling operations)
- rocshmem_signal_wait_until_on_stream (point-to-point sync)

The documentation includes function signatures, parameter descriptions,
and detailed explanations of asynchronous behavior and stream handling.

(AI Assist)

* Rename "bytes" -> "nelems"

* Add "_TEST_" to the variables used in tests

* Remove incorrect hipStreamDefault usage

hipStreamDefault is not a default stream. This is a flag.

If stream == nullptr, then just pass it to kernel. It will launch the kernel on the default stream

[ROCm/rocshmem commit: d0c8380650]

projects/rocshmem/tests/functional_tests/tester_arguments.cpp

@@ -112,10 +112,12 @@ TesterArguments::TesterArguments(int argc, char *argv[]) {
     case TeamBarrierTestType:
     case TeamWAVEBarrierTestType:
     case TeamWGBarrierTestType:
+    case BarrierAllOnStreamTestType:
     case SyncAllTestType:
     case WAVESyncAllTestType:
     case WGSyncAllTestType:
     case TeamSyncTestType:
+    case SignalWaitUntilOnStreamTestType:
       min_msg_size = 8;
       max_msg_size = 8;
       break;
@@ -125,6 +127,8 @@ TesterArguments::TesterArguments(int argc, char *argv[]) {
       max_msg_size = 4;
       break;
     case RandomAccessTestType:
+    case TeamAlltoallmemOnStreamTestType:
+    case TeamBroadcastmemOnStreamTestType:
       min_msg_size = 4;
       break;
     case TeamFCollectTestType:
@@ -173,23 +177,49 @@ void TesterArguments::get_arguments() {
   myid = rocshmem_my_pe();
 
   TestType type = (TestType)algorithm;
-  if ((type != BarrierAllTestType) && (type != WAVEBarrierAllTestType) &&
-      (type != WGBarrierAllTestType) && (type != SyncAllTestType) &&
-      (type != WAVESyncAllTestType) && (type != WGSyncAllTestType) &&
-      (type != TeamSyncTestType) && (type != TeamWAVESyncTestType) &&
-      (type != TeamWGSyncTestType) && (type != TeamAllToAllTestType) &&
-      (type != TeamFCollectTestType) && (type != TeamReductionTestType) &&
-      (type != TeamBroadcastTestType) && (type != PingAllTestType) &&
-      (type != TeamBarrierTestType) && (type != TeamWAVEBarrierTestType) &&
-      (type != TeamWGBarrierTestType) && (type != TeamCtxInfraTestBlockType) &&
-      (type != TeamCtxInfraTestOddEvenType) &&
-      (type != TeamAlltoallmemOnStreamTestType)) {
-    if (numprocs != 2) {
-      if (myid == 0) {
-        std::cerr << "This test requires exactly two processes, we have "
-                  << numprocs << "\n";
-      }
-      exit(-1);
+  // Check if test requires exactly 2 PEs
+  // Tests that support arbitrary number of PEs are excluded
+  bool requires_two_pes = true;
+  switch (type) {
+    // Collective/barrier tests - support any number of PEs
+    case BarrierAllTestType:
+    case WAVEBarrierAllTestType:
+    case WGBarrierAllTestType:
+    case SyncAllTestType:
+    case WAVESyncAllTestType:
+    case WGSyncAllTestType:
+    case TeamSyncTestType:
+    case TeamWAVESyncTestType:
+    case TeamWGSyncTestType:
+    case TeamAllToAllTestType:
+    case TeamFCollectTestType:
+    case TeamReductionTestType:
+    case TeamBroadcastTestType:
+    case PingAllTestType:
+    case TeamBarrierTestType:
+    case TeamWAVEBarrierTestType:
+    case TeamWGBarrierTestType:
+    case TeamCtxInfraTestBlockType:
+    case TeamCtxInfraTestOddEvenType:
+    // On-stream tests - support any number of PEs
+    case TeamAlltoallmemOnStreamTestType:
+    case BarrierAllOnStreamTestType:
+    case TeamBroadcastmemOnStreamTestType:
+    case GetmemOnStreamTestType:
+    case PutmemOnStreamTestType:
+    case PutmemSignalOnStreamTestType:
+    case SignalWaitUntilOnStreamTestType:
+      requires_two_pes = false;
+      break;
+    default:
+      break;
+  }
+
+  if (requires_two_pes && numprocs != 2) {
+    if (myid == 0) {
+      std::cerr << "This test requires exactly two processes, we have "
+                << numprocs << "\n";
     }
+    exit(-1);
   }
 }