Add Batched and Streamed Functions for Scalability

This commit extends the CUDA API with new functions designed for high-throughput scalability testing and advanced concurrency. Key additions include: - Batched versions of the kernels (`..._batched`) that use the `blockIdx.z` grid dimension to process many independent channel simulations in parallel. - A C-callable wrapper, `run_channel_pipeline_cuda_batched`, to orchestrate the execution of these new kernels. - A "streamed" version of the pipeline, `run_channel_pipeline_cuda_streamed`, which accepts a `cudaStream_t` as an argument. This allows the calling application to manage the execution stream, enabling overlap with other GPU or CPU tasks.
2026-02-09 09:29:36 +00:00
parent d43d267626
commit affab00c90
4 changed files with 255 additions and 0 deletions
@@ -23,6 +23,23 @@
 #include <cuda_runtime.h>
 #include "oai_cuda.h"

+__global__ void multipath_channel_kernel_batched(const float2 *__restrict__ d_channel_coeffs,
+                                                 const float2 *__restrict__ tx_sig,
+                                                 float2 *__restrict__ rx_sig,
+                                                 int num_samples,
+                                                 int channel_length,
+                                                 int nb_tx,
+                                                 int nb_rx);
+
+__global__ void add_noise_and_phase_noise_kernel_batched(const float2 *__restrict__ r_sig,
+                                                         short2 *__restrict__ output_sig,
+                                                         curandState_t *states,
+                                                         int num_samples,
+                                                         int nb_rx,
+                                                         float sigma,
+                                                         float pn_std_dev,
+                                                         bool apply_phase_noise);
+
 #define CHECK_CUDA(val) checkCuda((val), #val, __FILE__, __LINE__)
 static void checkCuda(cudaError_t result, const char *const func, const char *const file, const int line)
 {
@@ -163,4 +180,117 @@ void sum_channel_outputs_cuda(void **d_individual_outputs, void *d_final_output,

  CHECK_CUDA(cudaFree(d_ptr_array));
 }
+
+void run_channel_pipeline_cuda_streamed(int nb_tx,
+                                        int nb_rx,
+                                        int channel_length,
+                                        uint32_t num_samples,
+                                        float *h_channel_coeffs,
+                                        float sigma2,
+                                        double ts,
+                                        uint16_t pdu_bit_map,
+                                        uint16_t ptrs_bit_map,
+                                        void *d_tx_sig_void,
+                                        void *d_intermediate_sig_void,
+                                        void *d_final_output_void,
+                                        void *d_curand_states_void,
+                                        void *h_tx_sig_pinned_void,
+                                        void *d_channel_coeffs_void,
+                                        void *stream_void)
+{
+  cudaStream_t stream = (cudaStream_t)stream_void;
+  float2 *d_intermediate_sig = (float2 *)d_intermediate_sig_void;
+  short2 *d_final_output = (short2 *)d_final_output_void;
+  curandState_t *d_curand_states = (curandState_t *)d_curand_states_void;
+  float2 *d_channel_coeffs = (float2 *)d_channel_coeffs_void;
+  float *kernel_input_ptr;
+
+#if defined(USE_UNIFIED_MEMORY) || defined(USE_ATS_MEMORY)
+  kernel_input_ptr = (float *)h_tx_sig_pinned_void;
+#else
+  float *d_tx_sig = (float *)d_tx_sig_void;
+  float *h_tx_sig_pinned = (float *)h_tx_sig_pinned_void;
+
+  const int padding_len = channel_length - 1;
+  const size_t total_padded_tx_bytes = nb_tx * (num_samples + padding_len) * 2 * sizeof(float);
+
+  CHECK_CUDA(cudaMemcpyAsync(d_tx_sig, h_tx_sig_pinned, total_padded_tx_bytes, cudaMemcpyHostToDevice, stream));
+  kernel_input_ptr = d_tx_sig;
+#endif
+
+  size_t channel_size_bytes = nb_tx * nb_rx * channel_length * sizeof(float2);
+  CHECK_CUDA(cudaMemcpyAsync(d_channel_coeffs, h_channel_coeffs, channel_size_bytes, cudaMemcpyHostToDevice, stream));
+
+  dim3 threads_multipath(512, 1);
+  dim3 blocks_multipath((num_samples + threads_multipath.x - 1) / threads_multipath.x, nb_rx);
+  size_t sharedMemSize = (threads_multipath.x + channel_length - 1) * sizeof(float2);
+  multipath_channel_kernel<<<blocks_multipath, threads_multipath, sharedMemSize, stream>>>(d_channel_coeffs,
+                                                                                           kernel_input_ptr,
+                                                                                           d_intermediate_sig,
+                                                                                           num_samples,
+                                                                                           channel_length,
+                                                                                           nb_tx,
+                                                                                           nb_rx);
+
+  dim3 threads_noise(256, 1);
+  dim3 blocks_noise((num_samples + threads_noise.x - 1) / threads_noise.x, nb_rx);
+  float pn_variance = 1e-5f * 2.0f * 3.1415926535f * 300.0f * (float)ts;
+  bool apply_phase_noise = (pdu_bit_map & ptrs_bit_map);
+  add_noise_and_phase_noise_kernel<<<blocks_noise, threads_noise, 0, stream>>>(d_intermediate_sig,
+                                                                               d_final_output,
+                                                                               d_curand_states,
+                                                                               num_samples,
+                                                                               sqrtf(sigma2 / 2.0f),
+                                                                               sqrtf(pn_variance),
+                                                                               apply_phase_noise);
+}
+
+void run_channel_pipeline_cuda_batched(int num_channels,
+                                       int nb_tx,
+                                       int nb_rx,
+                                       int channel_length,
+                                       uint32_t num_samples,
+                                       void *d_channel_coeffs_batch,
+                                       float sigma2,
+                                       double ts,
+                                       uint16_t pdu_bit_map,
+                                       uint16_t ptrs_bit_map,
+                                       void *d_tx_sig_batch,
+                                       void *d_intermediate_sig_batch,
+                                       void *d_final_output_batch,
+                                       void *d_curand_states)
+{
+  float2 *d_tx = (float2 *)d_tx_sig_batch;
+  float2 *d_intermediate = (float2 *)d_intermediate_sig_batch;
+  short2 *d_final = (short2 *)d_final_output_batch;
+  float2 *d_coeffs = (float2 *)d_channel_coeffs_batch;
+  curandState_t *d_states = (curandState_t *)d_curand_states;
+
+  dim3 threads_multipath(512, 1, 1);
+  dim3 blocks_multipath((num_samples + threads_multipath.x - 1) / threads_multipath.x, nb_rx, num_channels);
+  size_t sharedMemSize = (threads_multipath.x + channel_length - 1) * sizeof(float2);
+
+  multipath_channel_kernel_batched<<<blocks_multipath, threads_multipath, sharedMemSize>>>(d_coeffs,
+                                                                                           d_tx,
+                                                                                           d_intermediate,
+                                                                                           num_samples,
+                                                                                           channel_length,
+                                                                                           nb_tx,
+                                                                                           nb_rx);
+
+  dim3 threads_noise(256, 1, 1);
+  dim3 blocks_noise((num_samples + threads_noise.x - 1) / threads_noise.x, nb_rx, num_channels);
+  float pn_variance = 1e-5f * 2.0f * 3.1415926535f * 300.0f * (float)ts;
+  bool apply_phase_noise = (pdu_bit_map & ptrs_bit_map);
+  add_noise_and_phase_noise_kernel_batched<<<blocks_noise, threads_noise>>>(d_intermediate,
+                                                                            d_final,
+                                                                            d_states,
+                                                                            num_samples,
+                                                                            nb_rx,
+                                                                            sqrtf(sigma2 / 2.0f),
+                                                                            sqrtf(pn_variance),
+                                                                            apply_phase_noise);
+
+  // Synchronization happens in the benchmark (test_channel_scalability) after this call returns.
+}
 } // extern "C"
@@ -105,6 +105,55 @@ __global__ void multipath_channel_kernel(const float2 *__restrict__ d_channel_co
  rx_sig[rx_ant_idx * num_samples + i].y = rx_tmp.y;
 }

+__global__ void multipath_channel_kernel_batched(const float2 *__restrict__ d_channel_coeffs,
+                                                 const float2 *__restrict__ tx_sig,
+                                                 float2 *__restrict__ rx_sig,
+                                                 int num_samples,
+                                                 int channel_length,
+                                                 int nb_tx,
+                                                 int nb_rx)
+{
+  extern __shared__ float2 tx_shared[];
+
+  const int i = blockIdx.x * blockDim.x + threadIdx.x;
+  const int rx_ant_idx = blockIdx.y;
+  const int c = blockIdx.z;
+
+  if (i >= num_samples)
+    return;
+
+  float2 rx_tmp = make_float2(0.0f, 0.0f);
+
+  const int padding_len = channel_length - 1;
+  const int padded_num_samples = num_samples + padding_len;
+
+  const int channel_tx_offset = c * nb_tx * padded_num_samples;
+  const int channel_rx_offset = c * nb_rx * num_samples;
+
+  for (int j = 0; j < nb_tx; j++) {
+    const int tid = threadIdx.x;
+    const int block_start_idx = blockIdx.x * blockDim.x;
+    const int shared_mem_size = blockDim.x + channel_length - 1;
+
+    for (int k = tid; k < shared_mem_size; k += blockDim.x) {
+      int load_idx = block_start_idx + k;
+      tx_shared[k] = tx_sig[channel_tx_offset + j * padded_num_samples + load_idx];
+    }
+    __syncthreads();
+
+    for (int l = 0; l < channel_length; l++) {
+      float2 tx_sample = tx_shared[tid + (channel_length - 1) - l];
+      int chan_link_idx = (c * nb_tx * nb_rx) + (rx_ant_idx + j * nb_rx);
+      float2 chan_weight = d_channel_coeffs[chan_link_idx * channel_length + l];
+      rx_tmp = complex_add(rx_tmp, complex_mul(tx_sample, chan_weight));
+    }
+    __syncthreads();
+  }
+
+  rx_sig[channel_rx_offset + rx_ant_idx * num_samples + i].x = rx_tmp.x;
+  rx_sig[channel_rx_offset + rx_ant_idx * num_samples + i].y = rx_tmp.y;
+}
+
 extern "C" {

 void multipath_channel_cuda(float **rx_sig_re,
@@ -109,6 +109,38 @@ void run_channel_pipeline_cuda(c16_t **output_signal,
                               void *h_final_output_pinned,
                               void *d_channel_coeffs);

+void run_channel_pipeline_cuda_batched(int num_channels,
+                                       int nb_tx,
+                                       int nb_rx,
+                                       int channel_length,
+                                       uint32_t num_samples,
+                                       void *d_channel_coeffs_batch,
+                                       float sigma2,
+                                       double ts,
+                                       uint16_t pdu_bit_map,
+                                       uint16_t ptrs_bit_map,
+                                       void *d_tx_sig_batch,
+                                       void *d_intermediate_sig_batch,
+                                       void *d_final_output_batch,
+                                       void *d_curand_states);
+
+void run_channel_pipeline_cuda_streamed(int nb_tx,
+                                        int nb_rx,
+                                        int channel_length,
+                                        uint32_t num_samples,
+                                        float *h_channel_coeffs,
+                                        float sigma2,
+                                        double ts,
+                                        uint16_t pdu_bit_map,
+                                        uint16_t ptrs_bit_map,
+                                        void *d_tx_sig_void,
+                                        void *d_intermediate_sig_void,
+                                        void *d_final_output_void,
+                                        void *d_curand_states_void,
+                                        void *h_tx_sig_pinned_void,
+                                        void *d_channel_coeffs_void,
+                                        void *stream_void);
+
 void sum_channel_outputs_cuda(void **d_individual_outputs, void *d_final_output, int num_channels, int nb_rx, int num_samples);

 /**
@@ -86,6 +86,50 @@ __global__ void add_noise_and_phase_noise_kernel(const float2 *__restrict__ r_si
                                                 (short)fmaxf(-32768.0f, fminf(32767.0f, final_signal.y)));
 }

+__global__ void add_noise_and_phase_noise_kernel_batched(const float2 *__restrict__ r_sig,
+                                                         short2 *__restrict__ output_sig,
+                                                         curandState_t *states,
+                                                         int num_samples,
+                                                         int nb_rx,
+                                                         float sigma,
+                                                         float pn_std_dev,
+                                                         bool apply_phase_noise)
+{
+  const int i = blockIdx.x * blockDim.x + threadIdx.x; // Sample index
+  const int ii = blockIdx.y; // RX antenna index
+  const int c = blockIdx.z; // Channel index
+
+  if (i >= num_samples)
+    return;
+
+  const int flat_output_idx = (c * nb_rx * num_samples) + (ii * num_samples) + i;
+  // Index for the smaller, reused pool of cuRAND states (does NOT include channel)
+  const int state_pool_idx = (ii * num_samples) + i;
+
+  curandState_t local_state = states[state_pool_idx];
+
+  float2 noisy_signal = r_sig[flat_output_idx];
+
+  float2 awgn = curand_normal2(&local_state);
+  noisy_signal.x += awgn.x * sigma;
+  noisy_signal.y += awgn.y * sigma;
+
+  float2 final_signal = noisy_signal;
+
+  if (apply_phase_noise) {
+    float phase_error = curand_normal(&local_state) * pn_std_dev;
+    float cos_phi, sin_phi;
+    __sincosf(phase_error, &sin_phi, &cos_phi);
+    float2 phase_rot = make_float2(cos_phi, sin_phi);
+    final_signal = complex_mul(noisy_signal, phase_rot);
+  }
+
+  states[state_pool_idx] = local_state;
+
+  output_sig[flat_output_idx] = make_short2((short)fmaxf(-32768.0f, fminf(32767.0f, final_signal.x)),
+                                            (short)fmaxf(-32768.0f, fminf(32767.0f, final_signal.y)));
+}
+
 extern "C" {

 void add_noise_cuda(const float **r_re,