fix: mempool

Author: xiaokamikami

src/main/scala/fpga/Host.scala

@@ -67,6 +67,7 @@ class Difftest2AXIs(val difftest_width: Int, val axis_width: Int) extends Module
     val rd_occupancy = wr_cnt_sync - rd_cnt // Calculate occupancy in read clock domain
 
     val inTransfer = RegInit(false.B)
+    val packetBeats = RegInit(VecInit(Seq.fill(axis_send_len)(0.U(axis_width.W))))
     val sendCnt = RegInit(0.U(log2Ceil(axis_send_len).W))
     val sendLast = sendCnt === (axis_send_len - 1).U
     val counter = RegInit(0.U(3.W)) // 0 to 7
@@ -86,6 +87,7 @@ class Difftest2AXIs(val difftest_width: Int, val axis_width: Int) extends Module
 
     // Start transfer when we have data available
     when(loadNextPacket) {
+      packetBeats := packetPayloadBeats
       rd_ptr := rd_ptr + 1.U
       rd_cnt := rd_cnt + 1.U // Increment read counter
     }
@@ -113,7 +115,7 @@ class Difftest2AXIs(val difftest_width: Int, val axis_width: Int) extends Module
 
     // AXI output
     io.axis.valid := inTransfer
-    io.axis.bits.data := packetPayloadBeats(sendCnt)
+    io.axis.bits.data := packetBeats(sendCnt)
     io.axis.bits.last := inTransfer && sendLast && sendPacketEnd
   }
 }

src/test/csrc/fpga/xdma.cpp

@@ -142,6 +142,11 @@ void FpgaXdma::start_transmit_thread() {
 }
 
 void FpgaXdma::stop_thansmit_thread() {
+#ifdef FPGA_SIM
+  for (int i = 0; i < CONFIG_DMA_CHANNELS; i++) {
+    xdma_sim_interrupt(i);
+  }
+#endif
   for (int i = 0; i < CONFIG_DMA_CHANNELS; i++) {
     if (receive_thread[i].joinable())
       receive_thread[i].join();
@@ -163,18 +168,50 @@ void FpgaXdma::read_xdma_thread(int channel) {
   size_t mem_get_idx = 0;
   while (running) {
     char *mem = xdma_mempool.get_free_chunk(&mem_get_idx);
+    if (mem == nullptr) {
+      continue;
+    }
 #ifdef FPGA_SIM
     size_t size = xdma_sim_read(channel, mem, sizeof(FpgaPackgeHead));
 #else
     size_t size = read(xdma_c2h_fd[channel], mem, sizeof(FpgaPackgeHead));
 #endif // FPGA_SIM
+    if (!running || size == 0) {
+      break;
+    }
     if (xdma_mempool.write_free_chunk(mem[0], mem_get_idx) == false) {
       printf("It should not be the case that no available block can be found\n");
       assert(0);
     }
   }
 }
 
+void FpgaXdma::read_and_process_fallback() {
+  printf("start channel 0\n");
+  void *ptr = nullptr;
+  int ret = posix_memalign(&ptr, 4096, sizeof(FpgaPackgeHead));
+  if (ret != 0) {
+    perror("posix_memalign failed");
+    return;
+  }
+  FpgaPackgeHead *packge = (FpgaPackgeHead *)ptr;
+  memset(packge, 0, sizeof(FpgaPackgeHead));
+  while (running) {
+#ifdef FPGA_SIM
+    size_t size = xdma_sim_read(0, (char *)packge, sizeof(FpgaPackgeHead));
+#else
+    size_t size = read(xdma_c2h_fd[0], packge, sizeof(FpgaPackgeHead));
+#endif // FPGA_SIM
+    if (!running || size == 0) {
+      break;
+    }
+    for (size_t i = 0; i < DMA_PACKGE_NUM; i++) {
+      v_difftest_Batch(packge->diff_packge[i].diff_packge);
+    }
+  }
+  free(packge);
+}
+
 void FpgaXdma::write_difftest_thread() {
   FpgaPackgeHead *packge;
   uint8_t recv_count = 0;

src/test/csrc/fpga/xdma.h

@@ -77,12 +77,16 @@ class FpgaXdma {
       }
     } else {
 #ifdef USE_THREAD_MEMPOOL
+#if defined(FPGA_SIM) && (CONFIG_DMA_CHANNELS == 1)
+      read_and_process_fallback();
+#else
       std::unique_lock<std::mutex> lock(thread_mtx);
       start_transmit_thread();
       while (running) {
         thread_cv.wait(lock); // wait notify from stop
       }
       stop_thansmit_thread();
+#endif
 #else
       read_and_process();
 #endif // USE_THREAD_MEMPOOL
@@ -128,6 +132,7 @@ class FpgaXdma {
   void start_transmit_thread();
   void stop_thansmit_thread();
   void read_xdma_thread(int channel);
+  void read_and_process_fallback();
   void write_difftest_thread();
 #else
   void read_and_process();

src/test/csrc/fpga_sim/xdma_sim.cpp

@@ -31,6 +31,7 @@ typedef struct {
   pthread_mutex_t lock;
   pthread_cond_t read_cond;
   bool read_waiting;
+  bool closed;
   int read_size;
   int write_size;
   char buffer[BUFFER_SIZE];
@@ -66,25 +67,45 @@ class xdma_sim {
       pthread_condattr_init(&cattr);
       pthread_condattr_setpshared(&cattr, PTHREAD_PROCESS_SHARED);
       pthread_cond_init(&shm_ptr->read_cond, &cattr);
+      shm_ptr->closed = false;
     }
   }
   ~xdma_sim() {
+    interrupt();
     if (is_host) {
       shm_unlink(path);
     }
     munmap(shm_ptr, sizeof(xdma_shm));
     close(shm_fd);
   }
+  void interrupt() {
+    if (shm_ptr == nullptr) {
+      return;
+    }
+    pthread_mutex_lock(&shm_ptr->lock);
+    shm_ptr->closed = true;
+    shm_ptr->read_waiting = false;
+    pthread_cond_broadcast(&shm_ptr->read_cond);
+    pthread_mutex_unlock(&shm_ptr->lock);
+  }
   int read(char *buf, size_t size) {
     assert(size <= BUFFER_SIZE);
     pthread_mutex_lock(&shm_ptr->lock);
 
+    if (shm_ptr->closed) {
+      pthread_mutex_unlock(&shm_ptr->lock);
+      return 0;
+    }
     shm_ptr->read_waiting = true;
     shm_ptr->write_size = 0;
     shm_ptr->read_size = size;
-    while (shm_ptr->write_size < size) {
+    while (shm_ptr->write_size < size && !shm_ptr->closed) {
       pthread_cond_wait(&shm_ptr->read_cond, &shm_ptr->lock);
     }
+    if (shm_ptr->closed) {
+      pthread_mutex_unlock(&shm_ptr->lock);
+      return 0;
+    }
     size_t to_copy = size < shm_ptr->write_size ? size : shm_ptr->write_size;
     memcpy(buf, shm_ptr->buffer, to_copy);
 
@@ -94,10 +115,14 @@ class xdma_sim {
   }
   int write(const char *buf, unsigned char tlast, size_t size) {
     pthread_mutex_lock(&shm_ptr->lock);
-    while (!shm_ptr->read_waiting) {
+    while (!shm_ptr->read_waiting && !shm_ptr->closed) {
       pthread_mutex_unlock(&shm_ptr->lock);
       pthread_mutex_lock(&shm_ptr->lock);
     }
+    if (shm_ptr->closed) {
+      pthread_mutex_unlock(&shm_ptr->lock);
+      return 0;
+    }
     size_t space = shm_ptr->read_size - shm_ptr->write_size;
     size_t to_write = size < space ? size : space;
 
@@ -128,6 +153,12 @@ void xdma_sim_close(int channel) {
   xsim[channel] = nullptr;
 }
 
+void xdma_sim_interrupt(int channel) {
+  if (xsim[channel] != nullptr) {
+    xsim[channel]->interrupt();
+  }
+}
+
 int xdma_sim_read(int channel, char *buf, size_t size) {
   return xsim[channel]->read(buf, size);
 }

src/test/csrc/fpga_sim/xdma_sim.h

@@ -20,6 +20,7 @@
 
 void xdma_sim_open(int channel, bool is_host);
 void xdma_sim_close(int channel);
+void xdma_sim_interrupt(int channel);
 int xdma_sim_read(int channel, char *buf, size_t size);
 int xdma_sim_write(int channel, const char *buf, uint8_t tlast, size_t size);