wip

ring_buffer.c

@@ -73,11 +73,16 @@ static int rbuf_write_amount(ring_buffer_t * rbuf) {
 static volatile char * rbuf_write(ring_buffer_t * rbuf) {
   return rbuf->buffer + rbuf->write_index;
 }
-static void rbuf_increment_write(ring_buffer_t * rbuf, size_t size) {
+static void rbuf_increment_write(ring_buffer_t * rbuf, size_t size, int block) {
   rbuf->write_index = (rbuf->write_index + size) % rbuf->len;
+  if(!block)
+    return;
   rbuf->block_buffer[rbuf->block_write_index] = size;
   rbuf->block_write_index = (rbuf->block_write_index + 1) % rbuf->block_buf_len;
 }
+static void rbuf_increment_read(ring_buffer_t * rbuf, size_t size) {
+  rbuf->read_index = (rbuf->read_index + size) % rbuf->len;
+}
 static volatile char * rbuf_read(ring_buffer_t * rbuf, size_t * size) {
   volatile char * data = rbuf->buffer + rbuf->read_index;
   *size = rbuf->block_buffer[rbuf->block_read_index];

trx_ecpri.c

@@ -427,92 +427,36 @@ static void *encode_thread(void *p) {
     error(EXIT_FAILURE, errno, "Could not set CPU affinity to CPU %d\n", s->encode_affinity);
 
   for(int64_t i = 0;; i++) {
-    int n;
+    int64_t n;
 
     n = rbuf_write_amount(&tx_rbuf);
 
-    // Send empty frames until we receive something
-    if(s->start_sending) {
-      if(!sync_complete) {
-        if(i == 0)
-          clock_gettime(CLOCK_TAI, &next);
-        // Limit packets sent
-        if(encode_counter.counter > target_counter) {
-          int k = (encode_counter.counter - target_counter + (s->frame_frequency / 100) - 1) / (s->frame_frequency / 100);
-          add_ns(&next, k * 1000 * 1000 * 10); // 10ms to send 38400 packets
-          clock_nanosleep(CLOCK_TAI, TIMER_ABSTIME, &next, NULL);
-          target_counter += k * s->frame_frequency / 100;
-        }
-        n = (n > TX_SYNC_BURST_SIZE) ? n : TX_SYNC_BURST_SIZE;
-        n = (n < (s->frame_frequency / 100)) ? n : (s->frame_frequency / 100);
-        for(int j = 0; j < n; j++) {
-          *((uint16_t *) (RBUF_WRITE0(tx_rbuf, uint8_t) + 6)) = htons(seq_id++);
-          rbuf_update_write_index(&tx_rbuf);
-        }
-        update_counter(&encode_counter, n);
-      }
-      else if (reset_encode_counter) {
-        if(s->trace_tx)
-          encode_counter_prev = encode_counter.counter;
-        encode_counter.counter = 0;
-        reset_encode_counter = 0;
-        seq_id = 0;
-      }
-    }
-
     // If we have frames to encode (is there space in TX buffer)
     // If there are frames from trx_write callback to encode
     if(n && rbuf_read_amount(&trxw_rbuf[0])) {
-      sample_group_t * g; int nb_frames;
-
-      g = RBUF_READ0(trxw_group_rbuf, sample_group_t);
-      if(g->wait) {
-        g->wait = 0;
-        g->count -= encode_counter.counter;
-        g->zeroes = 1;
-      }
-      nb_frames = g->count > n ? n : g->count;
-      g->count -= nb_frames;
-      if(s->trace_tx) {
-        if(sync_complete && (encode_counter.counter + nb_frames) >= (tx_rbuf.buf_len + s->trace_offset)) {
-          tx_trace_ready = 1;
-          log_info("ENCODE_THREAD", "TX Trace ready");
-          pthread_exit(EXIT_SUCCESS);
-        } else if (tx_trace_ready) {
-          pthread_exit(EXIT_SUCCESS);
-        }
-      }
-      if(g->zeroes) {
-        for(int j = 0; j < nb_frames; j++) {
-          memset(RBUF_WRITE0(tx_rbuf, uint8_t) + 8, 0x00, 240);
-          *((uint16_t *) (RBUF_WRITE0(tx_rbuf, uint8_t) + 6)) = htons(seq_id++);
-          rbuf_update_write_index(&tx_rbuf);
-        }
-        trxw_rbuf[0].read_index = (trxw_rbuf[0].read_index + nb_frames) % trxw_rbuf[0].buf_len;
-      } else {
-
-        int nc;
-        int nf = nb_frames;
-        while((nc = rbuf_contiguous_copy(&trxw_rbuf[0], &tx_rbuf, nf))) {
-          Complex * iq_samples[4];
-          uint8_t * buf = RBUF_WRITE0(tx_rbuf, uint8_t) + 8;
+      // TODO
+      int nc;
+      int nf = nb_frames;
+      while((nc = rbuf_contiguous_copy(&trxw_rbuf[0], &tx_rbuf, nf))) {
+        Complex * iq_samples[4];
+        uint8_t * buf = RBUF_WRITE0(tx_rbuf, uint8_t) + 8;
+        for(int j = 0; j < s->tx_n_channel; j++)
+          iq_samples[j] = ((Complex *) trxw_rbuf[j].buffer) + (trxw_rbuf[0].read_index * trxw_rbuf[0].len);
+
+        for(int i = 0; i < nc; i++) {
+          for(int i = 0; i < s->tx_n_channel ; i++)
+            encode_s64_b60_2(buf + i * 60, (float *) iq_samples[i]);
+          *((uint16_t *)(buf - 2)) = htons(seq_id++);
           for(int j = 0; j < s->tx_n_channel; j++)
-            iq_samples[j] = ((Complex *) trxw_rbuf[j].buffer) + (trxw_rbuf[0].read_index * trxw_rbuf[0].len);
-
-          for(int i = 0; i < nc; i++) {
-            for(int i = 0; i < s->tx_n_channel ; i++)
-              encode_s64_b60_2(buf + i * 60, (float *) iq_samples[i]);
-            *((uint16_t *)(buf - 2)) = htons(seq_id++);
-            for(int j = 0; j < s->tx_n_channel; j++)
-              iq_samples[j] += trxw_rbuf[0].len;
-            buf += tx_rbuf.len;
-          }
-          tx_rbuf.write_index = (tx_rbuf.write_index + nc) % tx_rbuf.buf_len;
-          trxw_rbuf[0].read_index = (trxw_rbuf[0].read_index + nc) % trxw_rbuf[0].buf_len;
-          nf -= nc;
+            iq_samples[j] += trxw_rbuf[0].len;
+          buf += tx_rbuf.len;
         }
-        if(nf)
-          exit(EXIT_FAILURE);
+        tx_rbuf.write_index = (tx_rbuf.write_index + nc) % tx_rbuf.buf_len;
+        trxw_rbuf[0].read_index = (trxw_rbuf[0].read_index + nc) % trxw_rbuf[0].buf_len;
+        nf -= nc;
+      }
+      if(nf)
+        exit(EXIT_FAILURE);
 
       }
       update_counter(&encode_counter, nb_frames);
@@ -854,18 +798,21 @@ static void trx_ecpri_end(TRXState *s1)
 static int64_t prev_timestamp = 0;
 static int64_t prev_count = 0;
 
-/* count: how much IQ samples to write (multiple of 32)
-   timestamp: should be equal to the amount of written IQ samples
-   __samples: can be null if zeroes are to be written
+/*
+   Callback for processing TRX samples from Amarisoft
+   Writes to TRX write buffer
+ 
+   - count: how much IQ samples to write (multiple of 32)
+   - timestamp: should be equal to the amount of written IQ samples
+   - __samples: can be null if zeroes are to be written
 */
 static void trx_ecpri_write(TRXState *s1, trx_timestamp_t timestamp, const void **__samples, int count, int tx_port_index, TRXWriteMetadata *md)
 {
-  int64_t count_left, ts; sample_group_t * g;
-  size_t nc, nk;
+  int64_t count_left, ts, nc, offset;
   float ** _samples = (float **) __samples;
   TRXEcpriState *s = s1->opaque;
 
-  log_debug("TRX_ECPRI_WRITE", "trx_ecpri_write, count = %ld", count);
+  log_debug("TRX_ECPRI_WRITE", "trx_ecpri_write, count = %ld", count / N_SAMPLES);
   
   if(prev_count && ((timestamp - prev_timestamp) != prev_count)) {
     log_exit("TRX_ECPRI_WRITE",
@@ -881,13 +828,15 @@ static void trx_ecpri_write(TRXState *s1, trx_timestamp_t timestamp, const void
     return;
   }
 
-  count_left = count
-  while((nc = rbuf_contiguous_copy(NULL, &trxw_rbuf[0], count_left * sizeof(Complex)))) {
+  offset = 0;
+  count_left = count;
+  while((nc = rbuf_contiguous_copy(
+                NULL, &trxw_rbuf[0], count_left * sizeof(Complex)))) {
     for(int i = 0; i < s->tx_n_channel; i++) {
       if(__samples)
         memcpy(
           ((uint8_t *) trxw_rbuf[i].buffer) + trxw_rbuf[0].write_index,
-          ((uint8_t *) _samples[i]) + nk,
+          ((uint8_t *) _samples[i]) + offset,
           nc);
       else
         memset(
@@ -895,41 +844,50 @@ static void trx_ecpri_write(TRXState *s1, trx_timestamp_t timestamp, const void
           0,
           nc);
     }
-    rbuf_increment_write(&trxw_rbuf, nc);
+    rbuf_increment_write(&trxw_rbuf, nc, 0);
     count_left -= nc / sizeof(Complex);
-    nk += nc;
+    offset += nc;
   }
   update_counter(&write_counter, count / N_SAMPLES);
 }
 
+/*
+   Callback for sending TRX samples to Amarisoft
+   Reads from TRX read buffer
+ 
+   - count: how much IQ samples to read (multiple of 32)
+   - ptimestamp: number of total read IQ samples
+   - __samples: IQ samples returned
+*/
 static int trx_ecpri_read(TRXState *s1, trx_timestamp_t *ptimestamp, void **__samples, int count, int rx_port_index, TRXReadMetadata *md)
 {
-  int64_t nc; int64_t n, count_left;
+  int64_t nc, count_left, offset;
   float ** _samples = (float **) __samples;
-  int offset = 0;
   TRXEcpriState *s = s1->opaque;
 
-  log_debug("TRX_ECPRI_READ", "count = %ld (%li)", count, read_counter.counter);
+  log_debug("TRX_ECPRI_READ", "count = %ld (%li)",
+    count / N_SAMPLES, read_counter.counter);
 
   while((rbuf_read_amount(&trxr_rbuf[0]) / sizeof(Complex)) < count);
   
   sync_complete = 1;
 
+  offset = 0;
   count_left = count;
-  while((nc = rbuf_contiguous_copy(&trxr_rbuf[0], NULL, count_left * sizeof(Complex)))) {
+  while((nc = rbuf_contiguous_copy(
+                &trxr_rbuf[0], NULL, count_left * sizeof(Complex)))) {
     for(int i = 0; i < s->rx_n_channel; i++ ) {
       memcpy(
-        (uint8_t*) (_samples[i] + offset),
-        ((uint8_t *) trxr_rbuf[i].buffer) + trxr_rbuf[0].read_index * trxr_rbuf[0].len * sizeof(Complex);
-      , len);
+        ((uint8_t*) _samples[i]) + offset,
+        ((uint8_t *) trxr_rbuf[i].buffer) + trxr_rbuf[0].read_index,
+        nc);
     }
-    trxr_rbuf[0].read_index = (trxr_rbuf[0].read_index + nc) % trxr_rbuf[0].buf_len;
-    n -= nc;
-    offset += len;
+    rbuf_increment_read(&trxr_rbuf, nc);
+    count_left -= nc / sizeof(Complex);
+    offset += nc;
   }
-
-  *ptimestamp = (read_counter.counter) * M;
-  update_counter(&read_counter, read_count);
+  *ptimestamp = (read_counter.counter) * N_SAMPLES;
+  update_counter(&read_counter, count / N_SAMPLES);
 
   return count;
 }