rdma: add multi qp experimental sample

app/meson.build

@@ -513,4 +513,16 @@ executable('RdmaVideoRxSample', rdma_video_rx_sample_sources,
   # asan should be always the first dep
   dependencies: [asan_dep, libpthread, mtl_rdma, libsdl2]
 )
+executable('RdmaVideoTxMultiSample', rdma_video_tx_multi_sample_sources,
+  c_args : app_c_args,
+  link_args: app_ld_args,
+  # asan should be always the first dep
+  dependencies: [asan_dep, libpthread, mtl_rdma]
+)
+executable('RdmaVideoRxMultiSample', rdma_video_rx_multi_sample_sources,
+  c_args : app_c_args,
+  link_args: app_ld_args,
+  # asan should be always the first dep
+  dependencies: [asan_dep, libpthread, mtl_rdma, libsdl2]
+)
 endif

app/sample/README.md

@@ -201,3 +201,15 @@ The dir include the simple sample code for how to develop application quickly ba
 ```bash
 ./build/app/RdmaVideoRxSample 192.168.75.11 192.168.75.10 20000
 ```
+
+[rdma_video_tx_multi.c](rdma/rdma_video_tx_multi.c): A tx video application based on rdma interface using multiple queue pair, just for experimental use. The default video format is 1920x1080 UYVY frame, and the default video frame rate is 30.
+
+```bash
+numactl -m 0 ./build/app/RdmaVideoTxMultiSample 192.168.75.10 20000 20001 test.yuv
+```
+
+[rdma_video_rx_multi.c](rdma/rdma_video_rx_multi.c): A rx video application based on rdma interface using multiple queue pair, just for experimental use. SDL display is enabled by default.
+
+```bash
+numactl -m 0 ./build/app/RdmaVideoRxMultiSample 192.168.75.11 192.168.75.10 20000 20001
+```

app/sample/meson.build

@@ -53,3 +53,5 @@ rdma_tx_sample_sources = files('rdma/rdma_tx.c')
 rdma_rx_sample_sources = files('rdma/rdma_rx.c')
 rdma_video_tx_sample_sources = files('rdma/rdma_video_tx.c')
 rdma_video_rx_sample_sources = files('rdma/rdma_video_rx.c')
+rdma_video_tx_multi_sample_sources = files('rdma/rdma_video_tx_multi.c')
+rdma_video_rx_multi_sample_sources = files('rdma/rdma_video_rx_multi.c')

app/sample/rdma/rdma_video_rx.c

@@ -6,6 +6,8 @@
 #include <mtl_rdma/mtl_rdma_api.h>
 #include <pthread.h>
 #include <signal.h>
+#include <stdatomic.h>
+#include <stdbool.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -21,7 +23,7 @@
 
 static pthread_mutex_t mtx = PTHREAD_MUTEX_INITIALIZER;
 static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
-static volatile int keep_running = 1;
+static atomic_bool keep_running = true;
 
 #ifdef APP_HAS_SDL2
 static SDL_Window* window = NULL;
@@ -40,7 +42,7 @@ static int rx_notify_buffer_ready(void* priv, struct mtl_rdma_buffer* buffer) {
 
 void int_handler(int dummy) {
   (void)(dummy);
-  keep_running = 0;
+  keep_running = false;
   pthread_mutex_lock(&mtx);
   pthread_cond_signal(&cond);
   pthread_mutex_unlock(&mtx);

app/sample/rdma/rdma_video_rx_multi.c

@@ -0,0 +1,265 @@
+/* SPDX-License-Identifier: BSD-3-Clause
+ * Copyright(c) 2024 Intel Corporation
+ */
+
+#include <inttypes.h>
+#include <mtl_rdma/mtl_rdma_api.h>
+#include <pthread.h>
+#include <signal.h>
+#include <stdatomic.h>
+#include <stdbool.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/mman.h>
+#include <time.h>
+#include <unistd.h>
+
+#ifdef APP_HAS_SDL2
+#include <SDL2/SDL.h>
+#endif
+
+#define NANOSECONDS_IN_SECOND 1000000000
+
+static pthread_mutex_t mtx = PTHREAD_MUTEX_INITIALIZER;
+static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
+static atomic_bool keep_running = true;
+
+#ifdef APP_HAS_SDL2
+static SDL_Window* window = NULL;
+static SDL_Renderer* renderer = NULL;
+static SDL_Texture* texture = NULL;
+#endif
+
+static int rx_notify_buffer_ready(void* priv, struct mtl_rdma_buffer* buffer) {
+  (void)(priv);
+  (void)(buffer);
+  pthread_mutex_lock(&mtx);
+  pthread_cond_signal(&cond);
+  pthread_mutex_unlock(&mtx);
+  return 0;
+}
+
+void int_handler(int dummy) {
+  (void)(dummy);
+  keep_running = false;
+  pthread_mutex_lock(&mtx);
+  pthread_cond_signal(&cond);
+  pthread_mutex_unlock(&mtx);
+}
+
+#ifdef APP_HAS_SDL2
+int sdl_init(size_t width, size_t height) {
+  if (SDL_Init(SDL_INIT_VIDEO) < 0) {
+    printf("SDL could not initialize! SDL_Error: %s\n", SDL_GetError());
+    return -1;
+  }
+
+  window = SDL_CreateWindow("RDMA Frame Display", SDL_WINDOWPOS_UNDEFINED,
+                            SDL_WINDOWPOS_UNDEFINED, 640, 360, SDL_WINDOW_SHOWN);
+  if (!window) {
+    printf("Window could not be created! SDL_Error: %s\n", SDL_GetError());
+    return -1;
+  }
+
+  renderer = SDL_CreateRenderer(window, -1, SDL_RENDERER_ACCELERATED);
+  if (!renderer) {
+    printf("Renderer could not be created! SDL_Error: %s\n", SDL_GetError());
+    return -1;
+  }
+
+  texture = SDL_CreateTexture(renderer, SDL_PIXELFORMAT_UYVY, SDL_TEXTUREACCESS_STREAMING,
+                              width, height);
+  if (!texture) {
+    printf("Texture could not be created! SDL_Error: %s\n", SDL_GetError());
+    return -1;
+  }
+
+  return 0;
+}
+
+void sdl_display_frame(void* frame, size_t width, size_t height) {
+  (void)(height);
+  SDL_UpdateTexture(texture, NULL, frame,
+                    width * 2);  // Assuming UYVY (2 bytes per pixel)
+  SDL_RenderClear(renderer);
+  SDL_RenderCopy(renderer, texture, NULL, NULL);
+  SDL_RenderPresent(renderer);
+}
+
+void sdl_cleanup() {
+  if (texture) SDL_DestroyTexture(texture);
+  if (renderer) SDL_DestroyRenderer(renderer);
+  if (window) SDL_DestroyWindow(window);
+  SDL_Quit();
+}
+#endif
+
+int main(int argc, char** argv) {
+#ifdef APP_HAS_SDL2
+  if (sdl_init(1920, 1080) != 0) {
+    fprintf(stderr, "Failed to initialize SDL.\n");
+    return -1;
+  }
+#endif
+
+  if (argc != 5) {
+    printf("Usage: %s <local_ip> <ip> <port> <port1>\n", argv[0]);
+    return -1;
+  }
+  signal(SIGINT, int_handler);
+
+  int ret = 0;
+  void* buffers[3] = {};
+  void* buffers1[3] = {};
+  mtl_rdma_handle mrh = NULL;
+  mtl_rdma_rx_handle rx0 = NULL;
+  mtl_rdma_rx_handle rx1 = NULL;
+  struct mtl_rdma_init_params p = {
+      .log_level = MTL_RDMA_LOG_LEVEL_INFO,
+      //.flags = MTL_RDMA_FLAG_LOW_LATENCY,
+  };
+  mrh = mtl_rdma_init(&p);
+  if (!mrh) {
+    printf("Failed to initialize RDMA\n");
+    ret = -1;
+    goto out;
+  }
+
+  size_t frame_size = 1920 * 1080 * 2; /* UYVY */
+  for (int i = 0; i < 3; i++) {
+    buffers[i] = mmap(NULL, frame_size, PROT_READ | PROT_WRITE,
+                      MAP_PRIVATE | MAP_ANONYMOUS | MAP_HUGETLB, -1, 0);
+    if (buffers[i] == MAP_FAILED) {
+      printf("Failed to allocate buffer\n");
+      ret = -1;
+      goto out;
+    }
+    buffers1[i] = buffers[i] + frame_size / 2;
+  }
+
+  struct mtl_rdma_rx_ops rx_ops = {
+      .name = "rdma_rx",
+      .local_ip = argv[1],
+      .ip = argv[2],
+      .port = argv[3],
+      .num_buffers = 3,
+      .buffers = buffers,
+      .buffer_capacity = frame_size / 2,
+      .notify_buffer_ready = rx_notify_buffer_ready,
+  };
+
+  rx0 = mtl_rdma_rx_create(mrh, &rx_ops);
+  if (!rx0) {
+    printf("Failed to create RDMA RX0\n");
+    ret = -1;
+    goto out;
+  }
+
+  rx_ops.name = "rdma_rx1";
+  rx_ops.buffers = buffers1;
+  rx_ops.port = argv[4];
+
+  rx1 = mtl_rdma_rx_create(mrh, &rx_ops);
+  if (!rx1) {
+    printf("Failed to create RDMA RX1\n");
+    ret = -1;
+    goto out;
+  }
+
+  struct timespec start_time, current_time;
+  clock_gettime(CLOCK_MONOTONIC, &start_time);
+  double elapsed_time;
+  int frame_count = 0;
+  double fps = 0.0;
+
+  printf("Starting to receive frames\n");
+
+  int frames_consumed = 0;
+  struct mtl_rdma_buffer* buffer = NULL;
+  struct mtl_rdma_buffer* buffer1 = NULL;
+  while (keep_running) {
+    if (!buffer) buffer = mtl_rdma_rx_get_buffer(rx0);
+    if (!buffer) {
+      /* wait for buffer ready */
+      pthread_mutex_lock(&mtx);
+      pthread_cond_wait(&cond, &mtx);
+      pthread_mutex_unlock(&mtx);
+      continue;
+    }
+    if (!buffer1) buffer1 = mtl_rdma_rx_get_buffer(rx1);
+    if (!buffer1) {
+      /* wait for buffer ready */
+      pthread_mutex_lock(&mtx);
+      pthread_cond_wait(&cond, &mtx);
+      pthread_mutex_unlock(&mtx);
+      continue;
+    }
+
+    if (buffer->user_meta && buffer->user_meta_size) {
+      struct timespec now;
+      clock_gettime(CLOCK_REALTIME, &now);
+      uint64_t recv_time_ns =
+          ((uint64_t)now.tv_sec * NANOSECONDS_IN_SECOND) + now.tv_nsec;
+      uint64_t send_time_ns = *(uint64_t*)buffer->user_meta;
+      printf("Latency: %.2f us\n", (recv_time_ns - send_time_ns) / 1000.0);
+    }
+
+    if (buffer1->user_meta && buffer1->user_meta_size) {
+      int sent = *(int*)buffer1->user_meta;
+      printf("Buffer sent: %d\n", sent);
+    }
+
+#ifdef APP_HAS_SDL2
+    /* display frame */
+    sdl_display_frame(buffer->addr, 1920, 1080);
+#endif
+
+    ret = mtl_rdma_rx_put_buffer(rx0, buffer);
+    if (ret < 0) {
+      printf("Failed to put buffer\n");
+      ret = -1;
+      break;
+    }
+    buffer = NULL;
+
+    ret = mtl_rdma_rx_put_buffer(rx1, buffer1);
+    if (ret < 0) {
+      printf("Failed to put buffer\n");
+      ret = -1;
+      break;
+    }
+    buffer1 = NULL;
+
+    frames_consumed++;
+    frame_count++;
+    clock_gettime(CLOCK_MONOTONIC, &current_time);
+    elapsed_time = current_time.tv_sec - start_time.tv_sec;
+    elapsed_time += (current_time.tv_nsec - start_time.tv_nsec) / 1000000000.0;
+
+    if (elapsed_time >= 5.0) {
+      fps = frame_count / elapsed_time;
+      printf("FPS: %.2f\n", fps);
+      frame_count = 0;
+      clock_gettime(CLOCK_MONOTONIC, &start_time);
+    }
+  }
+
+  printf("Received %d frames\n", frames_consumed);
+
+out:
+  if (rx0) mtl_rdma_rx_free(rx0);
+  if (rx1) mtl_rdma_rx_free(rx1);
+
+  for (int i = 0; i < 3; i++) {
+    if (buffers[i] && buffers[i] != MAP_FAILED) munmap(buffers[i], frame_size);
+  }
+
+  if (mrh) mtl_rdma_uinit(mrh);
+
+#ifdef APP_HAS_SDL2
+  sdl_cleanup();
+#endif
+
+  return 0;
+}

app/sample/rdma/rdma_video_tx.c

@@ -6,6 +6,8 @@
 #include <mtl_rdma/mtl_rdma_api.h>
 #include <pthread.h>
 #include <signal.h>
+#include <stdatomic.h>
+#include <stdbool.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -19,7 +21,7 @@
 
 static pthread_mutex_t mtx = PTHREAD_MUTEX_INITIALIZER;
 static pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
-static volatile int keep_running = 1;
+static atomic_bool keep_running = true;
 
 static int frames_sent = 0;
 static int frames_acked = -3;
@@ -63,7 +65,7 @@ static int tx_notify_buffer_done(void* priv, struct mtl_rdma_buffer* buffer) {
 
 void int_handler(int dummy) {
   (void)(dummy);
-  keep_running = 0;
+  keep_running = false;
   pthread_mutex_lock(&mtx);
   pthread_cond_signal(&cond);
   pthread_mutex_unlock(&mtx);