rfsim: process received packet when its received completely

The simplifies some code in packet reception and is a prerequisite
for processing received without relying on circular buffer

radio/rfsimulator/simulator.cpp

@@ -27,6 +27,7 @@
  * When the opposite side switch from passive reading to active R+Write, the synchro is not fully deterministic
  */
 
+#include "utils.h"
 #include <sys/socket.h>
 #include <sys/types.h>
 #include <netinet/in.h>
@@ -165,8 +166,8 @@ typedef struct buffer_s {
   char *circularBufEnd;
   sample_t *circularBuf;
   channel_desc_t *channel_model;
-  char *beamPtr;
-  size_t beam_ptr_sz;
+  char *packet_ptr;
+  size_t payload_sz;
   size_t remainToTransferBeam;
 } buffer_t;
 
@@ -993,10 +994,6 @@ static bool add_client(rfsimulator_state_t *t)
 
 static void process_recv_header(rfsimulator_state_t *t, buffer_t *b, bool first_time)
 {
-  // check the header and start block transfer
-  if (b->remainToTransfer != 0)
-    return;
-
   b->headerMode = false; // We got the header
 
   if (first_time) {
@@ -1031,109 +1028,27 @@ static void process_recv_header(rfsimulator_state_t *t, buffer_t *b, bool first_
       LOG_W(HW, "UEsock(%d) Tx/Rx shift too large Tx:%lu, Rx:%lu\n", b->conn_sock, t->lastWroteTS, b->lastReceivedTS);
     mutexunlock(t->Sockmutex);
   }
-  // move back the pointer to overwrite the header, we don't need it anymore
-  b->transferPtr = (char *)&b->circularBuf[(b->lastReceivedTS * b->th.nbAnt) % CirSize];
-  // we now need to read the samples
-  b->remainToTransfer = sampleToByte(b->th.size, b->th.nbAnt);
 
-  if (t->beam_ctrl->enable_beams) {
   int num_beams = __builtin_popcountll(b->th.beam_map);
-    AssertFatal(num_beams > 0, "Needs at least one beam\n");
-    int num_ant = b->th.nbAnt;
-    int num_samples = b->th.size;
-    size_t beam_packet_size = sizeof(sample_t) * num_ant * num_beams * num_samples;
-    if (b->beamPtr == NULL || beam_packet_size > b->beam_ptr_sz) {
-      free(b->beamPtr);
-      b->beamPtr = static_cast<char *>(malloc(beam_packet_size));
-    }
-    b->remainToTransferBeam = beam_packet_size;
-  }
+  AssertFatal(b->th.beam_map == 1ULL || t->beam_ctrl->enable_beams == 1,
+              "The transmitter has enabled beam simulation while this receiver has not\n");
+  size_t payload_sz = sampleToByte(b->th.size, b->th.nbAnt) * num_beams;
+  if (b->packet_ptr == NULL || payload_sz > b->payload_sz) {
+    free(b->packet_ptr);
+    b->packet_ptr = static_cast<char *>(calloc_or_fail(1, payload_sz));
+    b->payload_sz = payload_sz;
+  }
+  b->transferPtr = b->packet_ptr;
+  b->remainToTransfer = payload_sz;
   return;
 }
 
 static void process_recv(rfsimulator_state_t *t, buffer_t *b, bool first_time)
 {
-  if (b->transferPtr == b->circularBufEnd)
-    b->transferPtr = (char *)b->circularBuf;
-
-  if (!b->trashingPacket) {
-    b->lastReceivedTS = b->th.timestamp + b->th.size - byteToSample(b->remainToTransfer, b->th.nbAnt);
-    LOG_D(HW, "UEsock: %d Set b->lastReceivedTS %ld\n", b->conn_sock, b->lastReceivedTS);
-  }
-
-  if (b->remainToTransfer == 0) {
-    LOG_D(HW, "UEsock: %d Completed block reception: %ld\n", b->conn_sock, b->lastReceivedTS);
-    b->headerMode = true;
-    b->transferPtr = (char *)&b->th;
-    b->remainToTransfer = sizeof(samplesBlockHeader_t);
-    b->trashingPacket = false;
-  }
-}
-
-static bool flushInput(rfsimulator_state_t *t, int timeout, bool first_time)
-{
-  // Process all incoming events on sockets
-  // store the data in lists
-  struct epoll_event events[MAX_FD_RFSIMU] = {{0}};
-  int nfds = epoll_wait(t->epollfd, events, MAX_FD_RFSIMU, timeout);
-
-  if (nfds == -1) {
-    if (!(errno == EINTR || errno == EAGAIN))
-      LOG_W(HW, "epoll_wait() failed, errno(%d)\n", errno);
-    return false;
-  }
-
-  for (int nbEv = 0; nbEv < nfds; ++nbEv) {
-    buffer_t *b = static_cast<buffer_t *>(events[nbEv].data.ptr);
-
-    if (events[nbEv].events & EPOLLIN && b == NULL) {
-      bool ret = add_client(t);
-      if (!ret)
-        return ret;
-      continue;
-    } else {
-      if (events[nbEv].events & (EPOLLHUP | EPOLLERR | EPOLLRDHUP)) {
-        socketError(t, b);
-        continue;
-      }
-    }
-
-    if (b->circularBuf == NULL) {
-      LOG_E(HW, "Received data on not connected socket %d\n", events[nbEv].data.fd);
-      continue;
-    }
-
-    ssize_t blockSz;
-    if (b->headerMode)
-      blockSz = b->remainToTransfer;
-    else {
-      blockSz =
-          b->transferPtr + b->remainToTransfer <= b->circularBufEnd ? b->remainToTransfer : b->circularBufEnd - b->transferPtr;
-    }
-
-    ssize_t sz = 0;
-    if (!b->headerMode && t->beam_ctrl->enable_beams) {
   int num_beams = __builtin_popcountll(b->th.beam_map);
   AssertFatal(num_beams > 0, "Needs at least one beam\n");
   int num_ant = b->th.nbAnt;
   int num_samples = b->th.size;
-      size_t beam_packet_size = sizeof(sample_t) * num_beams * num_ant * num_samples;
-      size_t bytes_received = beam_packet_size - b->remainToTransferBeam;
-      ssize_t recv_size = recv(b->conn_sock, &b->beamPtr[bytes_received], b->remainToTransferBeam, MSG_DONTWAIT);
-      if (recv_size <= 0) {
-        if (recv_size < 0 && errno != EAGAIN)
-          LOG_E(HW, "recv() failed, errno(%d)\n", errno);
-        continue;
-      }
-      b->remainToTransferBeam -= recv_size;
-
-      int beam_sample_sz = num_beams * num_ant * sizeof(sample_t);
-      int previously_written_sample = bytes_received / beam_sample_sz;
-      int last_full_sample = (bytes_received + recv_size) / beam_sample_sz;
-      int num_samples_received = last_full_sample - previously_written_sample;
-      int first_sample = previously_written_sample + 1;
-      openair0_timestamp sample_timestamp = b->th.timestamp + first_sample;
-
   int tx_beam_ids[MAX_BEAMS];
   int *tx_beam_ids_p = tx_beam_ids;
   for (int i = 0; i < MAX_BEAMS; i++) {
@@ -1141,9 +1056,9 @@ static bool flushInput(rfsimulator_state_t *t, int timeout, bool first_time)
       *tx_beam_ids_p++ = i;
     }
   }
-      c16_t *in = (c16_t *)&b->beamPtr[first_sample * beam_sample_sz];
-      int num_samples_to_process = num_samples_received;
-      openair0_timestamp first_sample_timestamp = sample_timestamp;
+  c16_t *in = (c16_t *)b->packet_ptr;
+  int num_samples_to_process = num_samples;
+  openair0_timestamp first_sample_timestamp = b->th.timestamp;
   while (num_samples_to_process > 0) {
     uint32_t nsamps_out;
     uint64_t rx_beam_map = get_beam_map(&t->beam_ctrl->rx, first_sample_timestamp, num_samples_to_process, &nsamps_out);
@@ -1173,24 +1088,67 @@ static bool flushInput(rfsimulator_state_t *t, int timeout, bool first_time)
     first_sample_timestamp += nsamps_out;
     num_samples_to_process -= nsamps_out;
   }
-      int antenna_samples_sz = num_ant * sizeof(sample_t);
-      // pretend we received some samples without beams
-      sz = num_samples_received * antenna_samples_sz;
+}
+
+static bool flushInput(rfsimulator_state_t *t, int timeout, bool first_time)
+{
+  // Process all incoming events on sockets
+  // store the data in lists
+  struct epoll_event events[MAX_FD_RFSIMU] = {{0}};
+  int nfds = epoll_wait(t->epollfd, events, MAX_FD_RFSIMU, timeout);
+
+  if (nfds == -1) {
+    if (!(errno == EINTR || errno == EAGAIN))
+      LOG_W(HW, "epoll_wait() failed, errno(%d)\n", errno);
+    return false;
+  }
+
+  for (int nbEv = 0; nbEv < nfds; ++nbEv) {
+    buffer_t *b = static_cast<buffer_t *>(events[nbEv].data.ptr);
+
+    if (events[nbEv].events & EPOLLIN && b == NULL) {
+      bool ret = add_client(t);
+      if (!ret)
+        return ret;
+      continue;
     } else {
-      sz = recv(b->conn_sock, b->transferPtr, blockSz, MSG_DONTWAIT);
+      if (events[nbEv].events & (EPOLLHUP | EPOLLERR | EPOLLRDHUP)) {
+        socketError(t, b);
+        continue;
+      }
+    }
+
+    if (b->circularBuf == NULL) {
+      LOG_E(HW, "Received data on not connected socket %d\n", events[nbEv].data.fd);
+      continue;
+    }
+
+    ssize_t sz = recv(b->conn_sock, b->transferPtr, b->remainToTransfer, MSG_DONTWAIT);
     if (sz <= 0) {
       if (sz < 0 && errno != EAGAIN)
         LOG_E(HW, "recv() failed, errno(%d)\n", errno);
       continue;
     }
-    }
     LOG_D(HW, "Socket rcv %zd bytes\n", sz);
     b->remainToTransfer -= sz;
     b->transferPtr += sz;
+    if (b->remainToTransfer == 0) {
       if (b->headerMode)
         process_recv_header(t, b, first_time);
-    else
+      else {
+        LOG_D(HW, "UEsock: %d Completed block reception: %ld\n", b->conn_sock, b->lastReceivedTS);
+        b->headerMode = true;
+        b->transferPtr = (char *)&b->th;
+        b->remainToTransfer = sizeof(samplesBlockHeader_t);
+
+        if (!b->trashingPacket) {
           process_recv(t, b, first_time);
+          b->lastReceivedTS = b->th.timestamp + b->th.size;
+          LOG_D(HW, "UEsock: %d Set b->lastReceivedTS %ld\n", b->conn_sock, b->lastReceivedTS);
+        }
+        b->trashingPacket = false;
+      }
+    }
   }
   return nfds > 0;
 }