lte-ru.c

/*
 * Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The OpenAirInterface Software Alliance licenses this file to You under
 * the OAI Public License, Version 1.1  (the "License"); you may not use this file
 * except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.openairinterface.org/?page_id=698
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *-------------------------------------------------------------------------------
 * For more information about the OpenAirInterface (OAI) Software Alliance:
 *      contact@openairinterface.org
 */

/*! \file lte-ru.c
 * \brief Top-level threads for RU entity
 * \author R. Knopp, F. Kaltenberger, Navid Nikaein
 * \date 2019
 * \version 0.1
 * \company Eurecom
 * \email: {knopp, florian.kaltenberger, navid.nikaein}@eurecom.fr
 * \note
 * \warning
 */

#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <sys/ioctl.h>
#include <sys/types.h>
#include <sys/mman.h>
#include <sched.h>
#include <linux/sched.h>
#include <signal.h>
#include <execinfo.h>
#include <getopt.h>
#include <sys/sysinfo.h>

#undef MALLOC //there are two conflicting definitions, so we better make sure we don't use it at all

#include "assertions.h"
#include "PHY/defs_common.h"
#include "PHY/types.h"
#include "PHY/INIT/phy_init.h"
#include "PHY/phy_extern.h"
#include "PHY/LTE_ESTIMATION/lte_estimation.h"
#include "PHY/LTE_REFSIG/lte_refsig.h"
#include "PHY/LTE_TRANSPORT/if4_tools.h"
#include "PHY/LTE_TRANSPORT/transport_proto.h"
#include "SCHED/sched_common.h"
#include "common/utils/LOG/log.h"
#include "common/utils/LOG/vcd_signal_dumper.h"
#include "radio/COMMON/common_lib.h"
#include "radio/ETHERNET/ethernet_lib.h"

/* these variables have to be defined before including ENB_APP/enb_paramdef.h */
static int DEFBANDS[] = {7};
static int DEFENBS[] = {0};
static int DEFBFW[] = {0x00007fff};
static int DEFRUTPCORES[] = {2,4,6,8};


#include "ENB_APP/enb_paramdef.h"
#include "common/config/config_userapi.h"

#include "SIMULATION/ETH_TRANSPORT/proto.h"

#include "T.h"

#include "executables/softmodem-common.h"

#define MBMS_EXPERIMENTAL

extern int oai_exit;
extern clock_source_t clock_source;
#include "executables/thread-common.h"
//extern PARALLEL_CONF_t get_thread_parallel_conf(void);
//extern WORKER_CONF_t   get_thread_worker_conf(void);
extern void phy_init_RU(RU_t *);

void prach_procedures(PHY_VARS_eNB *eNB,int br_flag);

void stop_RU(RU_t **rup,int nb_ru);

static void do_ru_synch(RU_t *ru);

void configure_ru(int idx,
                  void *arg);

void configure_rru(int idx,
                   void *arg);

void reset_proc(RU_t *ru);
int connect_rau(RU_t *ru);

void wait_eNBs(void);

const char ru_states[6][9] = {"RU_IDLE","RU_CONFIG","RU_READY","RU_RUN","RU_ERROR","RU_SYNC"};

#if defined(PRE_SCD_THREAD)
#include "common/ran_context.h"
#include "nfapi/oai_integration/vendor_ext.h"
#include "openair2/LAYER2/MAC/mac_extern.h"
  extern uint8_t dlsch_ue_select_tbl_in_use;
  void init_ru_vnf(void);
  extern RAN_CONTEXT_t RC;
#endif

RU_t **RCconfig_RU(int nb_RU,int nb_L1_inst,PHY_VARS_eNB ***eNB,uint64_t *ru_mask,pthread_mutex_t *ru_mutex,pthread_cond_t *ru_cond);

/*************************************************************/
/* Functions to attach and configure RRU                     */


/*************************************************************/
/* Southbound Fronthaul functions, RCC/RAU                   */

// southbound IF5 fronthaul for 16-bit OAI format
static inline void fh_if5_south_out(RU_t *ru,int frame, int subframe, uint64_t timestamp) {
  if (ru->idx == 0) VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_TRX_TST, ru->proc.timestamp_tx&0xffffffff );

  ru->south_out_cnt++;
  int offset = subframe*ru->frame_parms->samples_per_tti;
  void *buffs[ru->nb_tx]; 
  for (int aid=0;aid<ru->nb_tx;aid++) buffs[aid] = (void*)&ru->common.txdata[aid][offset]; 

  ru->ifdevice.trx_write_func2(&ru->ifdevice,
  		               timestamp,
                               buffs,
			       0,
			       ru->frame_parms->samples_per_tti,
			       0,
			       ru->nb_tx); 

}


// southbound IF4p5 fronthaul
static inline void fh_if4p5_south_out(RU_t *ru,
                                      int frame,
                                      int subframe,
                                      uint64_t timestamp) {
  if (ru->idx == 0) VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_TRX_TST, ru->proc.timestamp_tx&0xffffffff );

  LOG_D(PHY,"ENTERED fh_if4p5_south_out   Sending IF4p5 for frame %d subframe %d ru %d\n",ru->proc.frame_tx,ru->proc.tti_tx,ru->idx);

  if (subframe_select(ru->frame_parms, subframe)!=SF_UL) {
    send_IF4p5(ru, frame, subframe, IF4p5_PDLFFT);
    ru->south_out_cnt++;
    LOG_D(PHY,"south_out_cnt %d\n",ru->south_out_cnt);
  }

  /*if (ru->idx == 0 || ru->idx == ru) {
      VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_FRAME_NUMBER_TX0_RU+ru->idx, ru->proc.frame_tx );
      VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_SUBFRAME_NUMBER_TX0_RU+ru->idx, ru->proc.subframe_tx );
    }*/
  VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME(VCD_SIGNAL_DUMPER_VARIABLES_FRAME_NUMBER_IF4P5_SOUTH_OUT_RU+ru->idx, ru->proc.frame_tx);
  VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME(VCD_SIGNAL_DUMPER_VARIABLES_SUBFRAME_NUMBER_IF4P5_SOUTH_OUT_RU+ru->idx, ru->proc.tti_tx);
}


/*************************************************************/
/* Input Fronthaul from south RCC/RAU                        */

// Synchronous if5 from south
void fh_if5_south_in(RU_t *ru,
                     int *frame,
                     int *subframe) {
  LTE_DL_FRAME_PARMS *fp = ru->frame_parms;
  RU_proc_t *proc = &ru->proc;
  ru->ifdevice.trx_read_func2(&ru->ifdevice,&proc->timestamp_rx,NULL,fp->samples_per_tti);
  proc->frame_rx    = (proc->timestamp_rx / (fp->samples_per_tti*10))&1023;
  proc->tti_rx = (proc->timestamp_rx / fp->samples_per_tti)%10;
  
  if (proc->first_rx == 0) {
    if (proc->tti_rx != *subframe) {
      LOG_E(PHY,"Received Timestamp doesn't correspond to the time we think it is (proc->tti_rx %d, subframe %d), resynching\n",proc->tti_rx,*subframe);
      *frame=proc->frame_rx;
      *subframe=proc->tti_rx;
    }

    if (proc->frame_rx != *frame) {
      LOG_E(PHY,"Received Timestamp doesn't correspond to the time we think it is (proc->frame_rx %d frame %d)\n",proc->frame_rx,*frame);
      exit_fun("Exiting");
    }
  } else {
    proc->first_rx = 0;
    *frame = proc->frame_rx;
    *subframe = proc->tti_rx;
  }

  VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_TRX_TS, proc->timestamp_rx&0xffffffff );
}


// Synchronous if4p5 from south
void fh_if4p5_south_in(RU_t *ru,
                       int *frame,
                       int *subframe) {
  LTE_DL_FRAME_PARMS *fp = ru->frame_parms;
  RU_proc_t *proc = &ru->proc;
  int f,sf;
  uint16_t packet_type;
  uint32_t symbol_number=0;
  uint32_t symbol_mask_full;

  if ((fp->frame_type == TDD) && (subframe_select(fp,*subframe)==SF_S))
    symbol_mask_full = (1<<fp->ul_symbols_in_S_subframe)-1;
  else
    symbol_mask_full = (1<<fp->symbols_per_tti)-1;

  LOG_D(PHY,"fh_if4p5_south_in: RU %d, frame %d, subframe %d, ru %d, mask %x\n",ru->idx,*frame,*subframe,ru->idx,proc->symbol_mask[*subframe]);
  //AssertFatal(proc->symbol_mask[*subframe]==0 || proc->symbol_mask[*subframe]>=symbol_mask_full,"rx_fh_if4p5: proc->symbol_mask[%d] = %x\n",*subframe,proc->symbol_mask[*subframe]); // >= because PULTICK for S-subframe could have been received during normal subframe

  if (proc->symbol_mask[*subframe]<symbol_mask_full) { // this is normal case, if not true then we received a PULTICK before the previous subframe was finished
    do {
      recv_IF4p5(ru, &f, &sf, &packet_type, &symbol_number);
      LOG_D(PHY,"fh_if4p5_south_in (%s/%d): RU %d, frame %d, subframe %d, f %d, sf %d, symbol %d\n",packet_type == IF4p5_PULFFT ? "PULFFT" : "PULTICK",packet_type,ru->idx,*frame,*subframe,f,sf,symbol_number);

      if (oai_exit == 1 || ru->cmd== STOP_RU) break;

      if (packet_type == IF4p5_PULFFT) proc->symbol_mask[sf] = proc->symbol_mask[sf] | (1<<symbol_number);
      else if (packet_type == IF4p5_PULTICK) {
        proc->symbol_mask[sf] = 0xffff;
        /*
                 if ((proc->first_rx==0) && (f!=*frame)) LOG_E(PHY,"rx_fh_if4p5: PULTICK received frame %d != expected %d (RU %d) \n",f,*frame, ru->idx);
                 else if ((proc->first_rx==0) && (sf!=*subframe)) LOG_E(PHY,"rx_fh_if4p5: PULTICK received subframe %d != expected %d (first_rx %d)\n",sf,*subframe,proc->first_rx);
                 else break; */
        //if (f==*frame || sf==*subframe) break;
      } else if (packet_type == IF4p5_PRACH) {
        // nothing in RU for RAU
      }

      LOG_D(PHY,"rx_fh_if4p5 for RU %d: subframe %d, sf %d, symbol %d, symbol mask %x\n",ru->idx,*subframe,sf,symbol_number,proc->symbol_mask[sf]);
    } while(proc->symbol_mask[*subframe] < symbol_mask_full);
  } else {
    f = *frame;
    sf = *subframe;
  }

  //calculate timestamp_rx, timestamp_tx based on frame and subframe
  proc->tti_rx       = sf;
  proc->frame_rx     = f;
  proc->timestamp_rx = ((proc->frame_rx * 10)  + proc->tti_rx ) * fp->samples_per_tti ;

  //  proc->timestamp_tx = proc->timestamp_rx +  (4*fp->samples_per_tti);
  if (get_thread_parallel_conf() == PARALLEL_SINGLE_THREAD) {
    proc->tti_tx   = (sf+ru->sf_ahead)%10;
    proc->frame_tx = (sf>(9-ru->sf_ahead)) ? (f+1)&1023 : f;
  }

  LOG_D(PHY,"Setting proc for (%d,%d)\n",sf,f);

  if (proc->first_rx == 0) {
    if (proc->tti_rx != *subframe) {
      LOG_E(PHY,"Received Timestamp (IF4p5) doesn't correspond to the time we think it is (proc->tti_rx %d, subframe %d, symbol_mask %x)\n", proc->tti_rx, *subframe, proc->symbol_mask[*subframe]);
      *subframe=sf;
      //exit_fun("Exiting");
    }

    if (ru->cmd != WAIT_RESYNCH && proc->frame_rx != *frame) {
      LOG_E(PHY,"Received Timestamp (IF4p5) doesn't correspond to the time we think it is (proc->frame_rx %d frame %d,symbol_mask %x\n",
            proc->frame_rx,*frame,proc->symbol_mask[*subframe]);
      //exit_fun("Exiting");
    } else if (ru->cmd == WAIT_RESYNCH && proc->frame_rx != *frame) {
      ru->cmd=EMPTY;
      *frame=proc->frame_rx;
    }
  } else {
    proc->first_rx = 0;
    *frame = proc->frame_rx;
    *subframe = proc->tti_rx;
  }

  /*if (ru->idx == 0 || ru->idx == 1) {
    VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_FRAME_NUMBER_RX0_RU+ru->idx, f );
    VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_SUBFRAME_NUMBER_RX0_RU+ru->idx, sf );
    VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_FRAME_NUMBER_RX0_RU, f );
    VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_TTI_NUMBER_RX0_RU, sf );
    VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_FRAME_NUMBER_TX0_RU, proc->frame_tx );
    VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_TTI_NUMBER_TX0_RU, proc->tti_tx );
  }*/
  VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME(VCD_SIGNAL_DUMPER_VARIABLES_FRAME_NUMBER_IF4P5_SOUTH_IN_RU+ru->idx,f);
  VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME(VCD_SIGNAL_DUMPER_VARIABLES_SUBFRAME_NUMBER_IF4P5_SOUTH_IN_RU+ru->idx,sf);
  proc->symbol_mask[sf] = 0;
  VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME(VCD_SIGNAL_DUMPER_VARIABLES_TRX_TS, proc->timestamp_rx&0xffffffff);
  LOG_D(PHY,"RU %d: fh_if4p5_south_in returning ...\n",ru->idx);
  //  usleep(100);
}


// Dummy FH from south for getting synchronization from master RU
void fh_slave_south_in(RU_t *ru,
                       int *frame,
                       int *subframe) {
  // This case is for synchronization to another thread
  // it just waits for an external event.  The actual rx_fh is handle by the asynchronous RX thread
  RU_proc_t *proc=&ru->proc;

  if (wait_on_condition(&proc->mutex_FH,&proc->cond_FH,&proc->instance_cnt_FH,"fh_slave_south_in") < 0)
    return;

  release_thread(&proc->mutex_FH,&proc->instance_cnt_FH,"rx_fh_slave_south_in");
}


// asynchronous inbound if4p5 fronthaul from south
void fh_if4p5_south_asynch_in(RU_t *ru,
                              int *frame,
                              int *subframe) {
  LTE_DL_FRAME_PARMS *fp = ru->frame_parms;
  RU_proc_t *proc        = &ru->proc;
  uint16_t packet_type;
  uint32_t symbol_number,symbol_mask,prach_rx;
  uint32_t got_prach_info=0;
  symbol_number = 0;
  symbol_mask   = (1<<fp->symbols_per_tti)-1;
  prach_rx      = 0;

  do {   // Blocking, we need a timeout on this !!!!!!!!!!!!!!!!!!!!!!!
    recv_IF4p5(ru, &proc->frame_rx, &proc->tti_rx, &packet_type, &symbol_number);

    if (ru->cmd == STOP_RU) break;

    // grab first prach information for this new subframe
    if (got_prach_info==0) {
      prach_rx       = is_prach_subframe(fp, proc->frame_rx, proc->tti_rx);
      got_prach_info = 1;
    }

    if (proc->first_rx != 0) {
      *frame = proc->frame_rx;
      *subframe = proc->tti_rx;
      proc->first_rx = 0;
    } else {
      if (proc->frame_rx != *frame) {
        LOG_E(PHY,"frame_rx %d is not what we expect %d\n",proc->frame_rx,*frame);
        exit_fun("Exiting");
      }

      if (proc->tti_rx != *subframe) {
        LOG_E(PHY,"tti_rx %d is not what we expect %d\n",proc->tti_rx,*subframe);
        exit_fun("Exiting");
      }
    }

    if      (packet_type == IF4p5_PULFFT)       symbol_mask &= (~(1<<symbol_number));
    else if (packet_type == IF4p5_PRACH)        prach_rx    &= (~0x1);
    else if (packet_type == IF4p5_PRACH_BR_CE0) prach_rx    &= (~0x2);
    else if (packet_type == IF4p5_PRACH_BR_CE1) prach_rx    &= (~0x4);
    else if (packet_type == IF4p5_PRACH_BR_CE2) prach_rx    &= (~0x8);
    else if (packet_type == IF4p5_PRACH_BR_CE3) prach_rx    &= (~0x10);
  } while( (symbol_mask > 0) || (prach_rx >0));   // haven't received all PUSCH symbols and PRACH information
}



/*************************************************************/
/* Input Fronthaul from North RRU                            */

// RRU IF4p5 TX fronthaul receiver. Assumes an if_device on input and if or rf device on output
// receives one subframe's worth of IF4p5 OFDM symbols and OFDM modulates
void fh_if4p5_north_in(RU_t *ru,
                       int *frame,
                       int *subframe) {
  uint32_t symbol_number=0;
  uint32_t symbol_mask, symbol_mask_full;
  uint16_t packet_type;
  /// **** incoming IF4p5 from remote RCC/RAU **** ///
  symbol_number = 0;
  symbol_mask = 0;
  symbol_mask_full = (1<<ru->frame_parms->symbols_per_tti)-1;
  LOG_D(PHY,"fh_if4p5_north_in: frame %d, subframe %d\n",*frame,*subframe);

  do {
    recv_IF4p5(ru, frame, subframe, &packet_type, &symbol_number);
    symbol_mask = symbol_mask | (1<<symbol_number);
  } while (symbol_mask != symbol_mask_full);

  ru->north_in_cnt++;

  // dump VCD output for first RU in list
  if (ru->idx == 0) {
    VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_FRAME_NUMBER_TX0_RU, *frame );
    VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_TTI_NUMBER_TX0_RU, *subframe );
  }
}

void fh_if5_north_asynch_in(RU_t *ru,
                            int *frame,
                            int *subframe) {
  LTE_DL_FRAME_PARMS *fp = ru->frame_parms;
  RU_proc_t *proc        = &ru->proc;
  int tti_tx,frame_tx;
  openair0_timestamp timestamp_tx=0;
  //recv_IF5(ru, &timestamp_tx, *subframe, IF5_RRH_GW_DL);
  //      LOG_I(PHY,"Received subframe %d (TS %llu) from RCC\n",tti_tx,timestamp_tx);
  tti_tx = (timestamp_tx/fp->samples_per_tti)%10;
  frame_tx    = (timestamp_tx/(fp->samples_per_tti*10))&1023;
  VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_FRAME_NUMBER_TX0_RU, proc->frame_tx );
  VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_TTI_NUMBER_TX0_RU, proc->tti_tx );

  if (proc->first_tx != 0) {
    *subframe = tti_tx;
    *frame    = frame_tx;
    proc->first_tx = 0;
  } else {
    AssertFatal(tti_tx == *subframe, "tti_tx %d is not what we expect %d\n",tti_tx,*subframe);
    AssertFatal(frame_tx == *frame, "frame_tx %d is not what we expect %d\n",frame_tx,*frame);
  }

  ru->north_in_cnt++;
}


void fh_if4p5_north_asynch_in(RU_t *ru,
                              int *frame,
                              int *subframe) {
  LTE_DL_FRAME_PARMS *fp = ru->frame_parms;
  RU_proc_t *proc        = &ru->proc;
  uint16_t packet_type;
  uint32_t symbol_number = 0, symbol_mask = 0, symbol_mask_full;
  int tti_tx, frame_tx, ret;
  LOG_D(PHY, "%s(ru:%p frame, subframe)\n", __FUNCTION__, ru);
  symbol_mask_full = ((subframe_select(fp,*subframe) == SF_S) ? (1<<fp->dl_symbols_in_S_subframe) : (1<<fp->symbols_per_tti))-1;
  LOG_D(PHY,"fh_if4p5_north_asynch_in: RU %d, frame %d, subframe %d\n", ru->idx, *frame, *subframe);

  do {
    recv_IF4p5(ru, &frame_tx, &tti_tx, &packet_type, &symbol_number);
    LOG_D(PHY,"income frame.subframe %d.%d, our frame.subframe.symbol_number %d.%d.%d (symbol mask %x)\n",frame_tx,tti_tx,*frame,*subframe,symbol_number,symbol_mask);

    if (ru->cmd == STOP_RU) {
      LOG_E(PHY,"Got STOP_RU\n");
      AssertFatal((ret=pthread_mutex_lock(&proc->mutex_ru))==0,"mutex_lock returns %d\n",ret);
      proc->instance_cnt_ru = -1;
      AssertFatal((ret=pthread_mutex_unlock(&proc->mutex_ru))==0,"mutex_unlock returns %d\n",ret);
      ru->cmd=STOP_RU;
      return;
    }

    if ((subframe_select(fp,tti_tx) == SF_DL) && (symbol_number == 0)) start_meas(&ru->rx_fhaul);

    LOG_D(PHY,"subframe %d (%d): frame %d, subframe %d, symbol %d\n", *subframe, subframe_select(fp,*subframe), frame_tx, tti_tx, symbol_number);

    if (proc->first_tx != 0) {
      *frame    = frame_tx;
      *subframe = tti_tx;
      proc->first_tx = 0;
      symbol_mask_full = ((subframe_select(fp,*subframe) == SF_S) ? (1<<fp->dl_symbols_in_S_subframe) : (1<<fp->symbols_per_tti))-1;
    } else {
      /* AssertFatal(frame_tx == *frame, "frame_tx %d is not what we expect %d\n",frame_tx,*frame);
       AssertFatal(subframe_tx == *subframe, "In frame_tx %d : subframe_tx %d is not what we expect %d\n",frame_tx,subframe_tx,*subframe);
      */
      *frame    = frame_tx;
      *subframe = tti_tx;
    }

    if (packet_type == IF4p5_PDLFFT) {
      symbol_mask = symbol_mask | (1<<symbol_number);
    } else AssertFatal(1==0,"Illegal IF4p5 packet type (should only be IF4p5_PDLFFT got %d\n",packet_type);
  } while (symbol_mask != symbol_mask_full);

  if (subframe_select(fp,tti_tx) == SF_DL) stop_meas(&ru->rx_fhaul);

  ru->north_in_cnt++;
  proc->tti_tx   = tti_tx;
  proc->frame_tx = frame_tx;

  if ((frame_tx == 0)&&(tti_tx == 0)) proc->frame_tx_unwrap += 1024;

  proc->timestamp_tx = ((((uint64_t)frame_tx + (uint64_t)proc->frame_tx_unwrap) * 10) + (uint64_t)tti_tx) * (uint64_t)fp->samples_per_tti;
  LOG_D(PHY,"RU %d/%d TST %llu, frame %d, subframe %d\n",ru->idx,0,(long long unsigned int)proc->timestamp_tx,frame_tx,tti_tx);

  // dump VCD output for first RU in list
  if (ru->idx == 0) {
    /*VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_FRAME_NUMBER_TX0_RU, frame_tx );
    VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_TTI_NUMBER_TX0_RU, tti_tx );*/
    VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME(VCD_SIGNAL_DUMPER_VARIABLES_FRAME_NUMBER_IF4P5_NORTH_ASYNCH_IN, frame_tx);
    VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME(VCD_SIGNAL_DUMPER_VARIABLES_SUBFRAME_NUMBER_IF4P5_NORTH_ASYNCH_IN, tti_tx);
  }

  if (ru->feptx_ofdm) ru->feptx_ofdm(ru, frame_tx, tti_tx);

  if (ru->fh_south_out) ru->fh_south_out(ru, frame_tx, tti_tx, proc->timestamp_tx);
}


void fh_if5_north_out(RU_t *ru) {
  /// **** send_IF5 of rxdata to BBU **** ///
  VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_SEND_IF5, 1 );
//  send_IF5(ru, proc->timestamp_rx, proc->tti_rx, &seqno, IF5_RRH_GW_UL);
  VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_SEND_IF5, 0 );
}


// RRU IF4p5 northbound interface (RX)
void fh_if4p5_north_out(RU_t *ru) {
  RU_proc_t *proc        = &ru->proc;
  LTE_DL_FRAME_PARMS *fp = ru->frame_parms;
  const int subframe     = proc->tti_rx;

  if (ru->idx==0) {
    VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME(VCD_SIGNAL_DUMPER_VARIABLES_TTI_NUMBER_RX0_RU, proc->tti_rx );
    VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME(VCD_SIGNAL_DUMPER_VARIABLES_SUBFRAME_NUMBER_IF4P5_NORTH_OUT, proc->tti_rx );
    VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME(VCD_SIGNAL_DUMPER_VARIABLES_FRAME_NUMBER_IF4P5_NORTH_OUT, proc->frame_rx );
  }

  LOG_D(PHY,"fh_if4p5_north_out: Sending IF4p5_PULFFT SFN.SF %d.%d\n", proc->frame_rx, proc->tti_rx);

  if ((fp->frame_type == TDD) && (subframe_select(fp,subframe)!=SF_UL)) {
    /// **** in TDD during DL send_IF4 of ULTICK to RCC **** ///
    send_IF4p5(ru, proc->frame_rx, proc->tti_rx, IF4p5_PULTICK);
    ru->north_out_cnt++;
    return;
  }

  start_meas(&ru->tx_fhaul);
  send_IF4p5(ru, proc->frame_rx, proc->tti_rx, IF4p5_PULFFT);
  ru->north_out_cnt++;
  stop_meas(&ru->tx_fhaul);
}

/* add fail safe for late command */
typedef enum {
  STATE_BURST_NORMAL    = 0,
  STATE_BURST_TERMINATE = 1,
  STATE_BURST_STOP_1    = 2,
  STATE_BURST_STOP_2    = 3,
  STATE_BURST_RESTART   = 4,
} late_control_e;

volatile late_control_e late_control=STATE_BURST_NORMAL;

/* add fail safe for late command end */

static void *emulatedRF_thread(void *param) {
  RU_proc_t *proc = (RU_proc_t *) param;
  int microsec = 500; // length of time to sleep, in miliseconds
  int numerology = 0 ;
  struct timespec req = {0};
  req.tv_sec = 0;
  req.tv_nsec = (numerology>0)? ((microsec * 1000L)/numerology):(microsec * 1000L)*2;
  cpu_set_t cpuset;
  CPU_ZERO(&cpuset);
  CPU_SET(1,&cpuset);
  pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);
  int policy;
  int ret;
  struct sched_param sparam;
  memset(&sparam, 0, sizeof(sparam));
  sparam.sched_priority = sched_get_priority_max(SCHED_FIFO);
  policy = SCHED_FIFO ;
  pthread_setschedparam(pthread_self(), policy, &sparam);
  wait_sync("emulatedRF_thread");

  while(!oai_exit) {
    nanosleep(&req, (struct timespec *)NULL);

    if(proc->emulate_rf_busy ) {
      LOG_E(PHY,"rf being delayed in emulated RF\n");
    }

    proc->emulate_rf_busy = 1;
    AssertFatal((ret=pthread_mutex_lock(&proc->mutex_emulateRF))==0,"mutex_lock returns %d\n",ret);
    ++proc->instance_cnt_emulateRF;
    pthread_cond_signal(&proc->cond_emulateRF);
    AssertFatal((ret=pthread_mutex_unlock(&proc->mutex_emulateRF))==0,"mutex_unlock returns %d\n",ret);
  }

  return 0;
}


void rx_rf(RU_t *ru,
           int *frame,
           int *subframe) {
  RU_proc_t *proc = &ru->proc;
  LTE_DL_FRAME_PARMS *fp = ru->frame_parms;
  void *rxp[ru->nb_rx];
  unsigned int rxs;
  int i;
  int resynch=0;
  openair0_timestamp ts=0,old_ts=0;

  for (i=0; i<ru->nb_rx; i++)
    rxp[i] = (void *)&ru->common.rxdata[i][*subframe*fp->samples_per_tti];

  VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_TRX_READ, 1 );
  old_ts = proc->timestamp_rx;

  if(ru->emulate_rf) {
    wait_on_condition(&proc->mutex_emulateRF,&proc->cond_emulateRF,&proc->instance_cnt_emulateRF,"emulatedRF_thread");
    release_thread(&proc->mutex_emulateRF,&proc->instance_cnt_emulateRF,"emulatedRF_thread");
    rxs = fp->samples_per_tti;
  } else {
    rxs = ru->rfdevice.trx_read_func(&ru->rfdevice,
                                     &ts,
                                     rxp,
                                     fp->samples_per_tti,
                                     ru->nb_rx);
  }

  VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_TRX_READ, 0 );
  ru->south_in_cnt++;
  LOG_D(PHY,"south_in_cnt %d\n",ru->south_in_cnt);

  if (ru->cmd==RU_FRAME_RESYNCH) {
    LOG_I(PHY,"Applying frame resynch %d => %d\n",*frame,ru->cmdval);

    if (proc->frame_rx>ru->cmdval) ru->ts_offset += (proc->frame_rx - ru->cmdval)*fp->samples_per_tti*10;
    else ru->ts_offset -= (-proc->frame_rx + ru->cmdval)*fp->samples_per_tti*10;

    *frame = ru->cmdval;
    ru->cmd=EMPTY;
    resynch=1;
  }

  if(get_softmodem_params()->emulate_rf) {
    proc->timestamp_rx = old_ts + fp->samples_per_tti;
  } else {
    proc->timestamp_rx = ts-ru->ts_offset;
  }

  //  AssertFatal(rxs == fp->samples_per_tti,
  //        "rx_rf: Asked for %d samples, got %d from SDR\n",fp->samples_per_tti,rxs);
  if(rxs != fp->samples_per_tti) {
    LOG_E(PHY,"rx_rf: Asked for %d samples, got %d from SDR\n",fp->samples_per_tti,rxs);
    late_control=STATE_BURST_TERMINATE;
  }

  if (proc->first_rx == 1) {
    ru->ts_offset = proc->timestamp_rx;
    proc->timestamp_rx = 0;
  } else if (resynch==0 && (proc->timestamp_rx - old_ts != fp->samples_per_tti)) {
    LOG_D(PHY,"rx_rf: rfdevice timing drift of %"PRId64" samples (ts_off %"PRId64")\n",proc->timestamp_rx - old_ts - fp->samples_per_tti,ru->ts_offset);
    ru->ts_offset += (proc->timestamp_rx - old_ts - fp->samples_per_tti);
    proc->timestamp_rx = ts-ru->ts_offset;
  }

  proc->frame_rx     = (proc->timestamp_rx / (fp->samples_per_tti*10))&1023;
  proc->tti_rx  = (proc->timestamp_rx / fp->samples_per_tti)%10;
  // synchronize first reception to frame 0 subframe 0

  if (get_thread_parallel_conf() == PARALLEL_SINGLE_THREAD && ru->fh_north_asynch_in == NULL) {
#ifdef PHY_TX_THREAD
    proc->timestamp_phy_tx = proc->timestamp_rx+((ru->sf_ahead-1)*fp->samples_per_tti);
    proc->subframe_phy_tx  = (proc->tti_rx+(ru->sf_ahead-1))%10;
    proc->frame_phy_tx     = (proc->tti_rx>(9-(ru->sf_ahead-1))) ? (proc->frame_rx+1)&1023 : proc->frame_rx;
#else
    proc->timestamp_tx = proc->timestamp_rx+(ru->sf_ahead*fp->samples_per_tti);
    proc->tti_tx       = (proc->tti_rx+ru->sf_ahead)%10;
    proc->frame_tx     = (proc->tti_rx>(9-ru->sf_ahead)) ? (proc->frame_rx+1)&1023 : proc->frame_rx;
#endif
    //proc->timestamp_tx = proc->timestamp_rx+(sf_ahead*fp->samples_per_tti);
    //proc->subframe_tx  = (proc->tti_rx+sf_ahead)%10;
    //proc->frame_tx     = (proc->tti_rx>(9-sf_ahead)) ? (proc->frame_rx+1)&1023 : proc->frame_rx;
    LOG_D(PHY,"RU %d/%d TS %llu (off %d), frame %d, subframe %d\n",
          ru->idx,
          0,
          (unsigned long long int)proc->timestamp_rx,
          (int)ru->ts_offset,
          proc->frame_rx,
          proc->tti_rx);
    LOG_D(PHY,"south_in/rx_rf: RU %d/%d TS %llu (off %d), frame %d, subframe %d\n",
          ru->idx,
          0,
          (unsigned long long int)proc->timestamp_rx,
          (int)ru->ts_offset,
          proc->frame_rx,
          proc->tti_rx);

  }

  // dump VCD output for first RU in list
  if (ru->idx == 0) {
    VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_FRAME_NUMBER_RX0_RU, proc->frame_rx );
    VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_TTI_NUMBER_RX0_RU, proc->tti_rx );
  }

  if (proc->first_rx == 0) {
    if (proc->tti_rx != *subframe) {
      LOG_E(PHY,"Received Timestamp (%llu) doesn't correspond to the time we think it is (proc->tti_rx %d, subframe %d)\n",(long long unsigned int)proc->timestamp_rx,proc->tti_rx,*subframe);
      exit_fun("Exiting");
    }

    if (proc->frame_rx != *frame) {
      LOG_E(PHY,"Received Timestamp (%llu) doesn't correspond to the time we think it is (proc->frame_rx %d frame %d)\n",(long long unsigned int)proc->timestamp_rx,proc->frame_rx,*frame);
      exit_fun("Exiting");
    }
  } else {
    proc->first_rx = 0;
    *frame = proc->frame_rx;
    *subframe = proc->tti_rx;
  }

  //LOG_I(PHY,"timestamp_rx %lu, frame %d(%d), subframe %d(%d)\n",ru->timestamp_rx,proc->frame_rx,frame,proc->tti_rx,subframe);
  VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_TRX_TS, proc->timestamp_rx&0xffffffff );

  if (rxs != fp->samples_per_tti) {
#if defined(USRP_REC_PLAY)
    exit_fun("Exiting IQ record/playback");
#else
    //exit_fun( "problem receiving samples" );
    LOG_E(PHY, "problem receiving samples");
#endif
  }
}


void tx_rf(RU_t *ru,
           int frame,
           int subframe,
           uint64_t timestamp) {
  RU_proc_t *proc = &ru->proc;
  LTE_DL_FRAME_PARMS *fp = ru->frame_parms;
  void *txp[ru->nb_tx];
  unsigned int txs;
  int i;
  T(T_ENB_PHY_OUTPUT_SIGNAL, T_INT(0), T_INT(0), T_INT(frame), T_INT(subframe),
    T_INT(0), T_BUFFER(&ru->common.txdata[0][subframe * fp->samples_per_tti], fp->samples_per_tti * 4));
  lte_subframe_t SF_type     = subframe_select(fp,subframe%10);
  lte_subframe_t prevSF_type = subframe_select(fp,(subframe+9)%10);
  //lte_subframe_t nextSF_type = subframe_select(fp,(subframe+1)%10);
  int sf_extension = 0;

  if ((SF_type == SF_DL) ||
      (SF_type == SF_S) ) {
    int siglen=fp->samples_per_tti;
    radio_tx_burst_flag_t flags = TX_BURST_MIDDLE;

    if (SF_type == SF_S) {
      int txsymb = fp->dl_symbols_in_S_subframe+(ru->is_slave==0 ? 1 : 0);
      AssertFatal(txsymb>0,"illegal txsymb %d\n",txsymb);
      /* end_of_burst_delay is used to stop TX only "after a while".
       * If we stop right after effective signal, with USRP B210 and
       * B200mini, we observe a high EVM on the S subframe (on the
       * PSS).
       * A value of 400 (for 30.72MHz) solves this issue. This is
       * the default.
       */
      siglen = (fp->ofdm_symbol_size + fp->nb_prefix_samples0)
               + (txsymb - 1) * (fp->ofdm_symbol_size + fp->nb_prefix_samples)
               + ru->end_of_burst_delay;
      flags = TX_BURST_END;
    }

    if (fp->frame_type == TDD &&
        SF_type == SF_DL &&
        prevSF_type == SF_UL) {
      flags = TX_BURST_START;
      sf_extension = ru->sf_extension;
    }

#if defined(__x86_64) || defined(__i386__)
    sf_extension = (sf_extension)&0xfffffff8;
#elif defined(__arm__) || defined(__aarch64__)
    sf_extension = (sf_extension)&0xfffffffc;
#endif

    for (i=0; i<ru->nb_tx; i++)
      txp[i] = (void *)&ru->common.txdata[i][(subframe*fp->samples_per_tti)-sf_extension];

    /* add fail safe for late command */
    if(late_control!=STATE_BURST_NORMAL) { //stop burst
      LOG_E(PHY,"%d.%d late_control : %d\n",frame,subframe,late_control);
      switch (late_control) {
        case STATE_BURST_TERMINATE:
          flags = TX_BURST_END_NO_TIME_SPEC;
          late_control=STATE_BURST_STOP_1;
          break;

        case STATE_BURST_STOP_1:
          flags = TX_BURST_INVALID;
          late_control=STATE_BURST_STOP_2;
          return;//no send
          break;

        case STATE_BURST_STOP_2:
          flags = TX_BURST_INVALID;
          late_control=STATE_BURST_RESTART;
          return;//no send
          break;

        case STATE_BURST_RESTART:
          flags = TX_BURST_START;
          late_control=STATE_BURST_NORMAL;
          break;

        default:
          LOG_D(PHY,"[TXPATH] RU %d late_control %d not implemented\n",ru->idx, late_control);
          break;
      }
    }
    VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_FRAME_NUMBER_TX0_RU, frame);
    VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_TTI_NUMBER_TX0_RU, subframe);
    VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_TRX_TST, (timestamp-ru->openair0_cfg.tx_sample_advance)&0xffffffff );
    VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_TRX_WRITE_FLAGS,flags);
 
    /* add fail safe for late command end */
   VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_TRX_WRITE, 1 );
    // prepare tx buffer pointers
    txs = ru->rfdevice.trx_write_func(&ru->rfdevice,
                                      timestamp+ru->ts_offset-ru->openair0_cfg.tx_sample_advance-sf_extension,
                                      txp,
                                      siglen+sf_extension,
                                      ru->nb_tx,
                                      flags);
    ru->south_out_cnt++;
    LOG_D(PHY,"south_out_cnt %d\n",ru->south_out_cnt);
    int se = dB_fixed(signal_energy(txp[0],siglen+sf_extension));

    if (SF_type == SF_S) LOG_D(PHY,"[TXPATH] RU %d tx_rf (en %d,len %d), writing to TS %llu, frame %d, unwrapped_frame %d, subframe %d\n",ru->idx, se,
                                 siglen+sf_extension, (long long unsigned int)timestamp, frame, proc->frame_tx_unwrap, subframe);

    VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_TRX_WRITE, 0 );

    //    AssertFatal(txs ==  siglen+sf_extension,"TX : Timeout (sent %d/%d)\n",txs, siglen);
    if( usrp_tx_thread == 0 && (txs !=  siglen+sf_extension) && (late_control==STATE_BURST_NORMAL) ) { /* add fail safe for late command */
      late_control=STATE_BURST_TERMINATE;
      LOG_E(PHY,"TX : Timeout (sent %d/%d) state =%d\n",txs, siglen,late_control);
    }
  } else if (IS_SOFTMODEM_RFSIM ) {
    // in case of rfsim, we always enable tx because we need to feed rx of the opposite side
    // we write 1 single I/Q sample to trigger Rx (rfsim will fill gaps with 0 I/Q)
    void *dummy_tx[ru->frame_parms->nb_antennas_tx];
    int16_t dummy_tx_data[ru->frame_parms->nb_antennas_tx][2]; // 2 because the function we call use pairs of int16_t implicitly as complex numbers
    memset(dummy_tx_data,0,sizeof(dummy_tx_data));
    for (int i=0; i<ru->frame_parms->nb_antennas_tx; i++)
      dummy_tx[i]= dummy_tx_data[i];
    
    AssertFatal( 1 ==
                 ru->rfdevice.trx_write_func(&ru->rfdevice,
                                             timestamp+ru->ts_offset-ru->openair0_cfg.tx_sample_advance-sf_extension,
                                             dummy_tx,
                                             1,
                                             ru->frame_parms->nb_antennas_tx,
                                             4),"");
    
  }
}


/*!
 * \brief The Asynchronous RX/TX FH thread of RAU/RCC/eNB/RRU.
 * This handles the RX FH for an asynchronous RRU/UE
 * \param param is a \ref L1_proc_t structure which contains the info what to process.
 * \returns a pointer to an int. The storage is not on the heap and must not be freed.
 */
static void *ru_thread_asynch_rxtx( void *param ) {
  static int ru_thread_asynch_rxtx_status;
  RU_t *ru         = (RU_t *)param;
  RU_proc_t *proc  = &ru->proc;
  int subframe=0, frame=0;
  thread_top_init("ru_thread_asynch_rxtx",1,870000,1000000,1000000);
  // wait for top-level synchronization and do one acquisition to get timestamp for setting frame/subframe
  wait_sync("ru_thread_asynch_rxtx");
  // wait for top-level synchronization and do one acquisition to get timestamp for setting frame/subframe
  LOG_I(PHY, "waiting for devices (ru_thread_asynch_rxtx)\n");
  wait_on_condition(&proc->mutex_asynch_rxtx,&proc->cond_asynch_rxtx,&proc->instance_cnt_asynch_rxtx,"thread_asynch");
  LOG_I(PHY, "devices ok (ru_thread_asynch_rxtx)\n");

  while (!oai_exit) {

    if (ru->state != RU_RUN) {
      subframe=0;
      frame=0;
      usleep(1000);
    } else {
      if (subframe==9) {
        subframe=0;
        frame++;
        frame&=1023;
      } else {
        subframe++;
      }

      LOG_D(PHY,"ru_thread_asynch_rxtx: Waiting on incoming fronthaul\n");

      // asynchronous receive from north (RRU IF4/IF5)
      if (ru->fh_north_asynch_in) {
        if (subframe_select(ru->frame_parms,subframe)!=SF_UL)
          ru->fh_north_asynch_in(ru, &frame, &subframe);
      } else
        AssertFatal(1==0,"Unknown function in ru_thread_asynch_rxtx\n");
    }
  }

  ru_thread_asynch_rxtx_status=0;
  return(&ru_thread_asynch_rxtx_status);
}


void wakeup_slaves(RU_proc_t *proc) {
  int ret;
  struct timespec wait;
  int time_ns = 5000000L;

  for (int i=0; i<proc->num_slaves; i++) {
    RU_proc_t *slave_proc = proc->slave_proc[i];
    // wake up slave FH thread
    // lock the FH mutex and make sure the thread is ready
    clock_gettime(CLOCK_REALTIME,&wait);
    wait.tv_nsec += time_ns;

    if(wait.tv_nsec >= 1000*1000*1000) {
      wait.tv_nsec -= 1000*1000*1000;
      wait.tv_sec  += 1;
    }

    AssertFatal((ret=pthread_mutex_timedlock(&slave_proc->mutex_FH,&wait))==0,"ERROR pthread_mutex_lock for RU %d slave %d (IC %d)\n",proc->ru->idx,slave_proc->ru->idx,slave_proc->instance_cnt_FH);
    int cnt_slave            = ++slave_proc->instance_cnt_FH;
    slave_proc->frame_rx     = proc->frame_rx;
    slave_proc->tti_rx  = proc->tti_rx;
    slave_proc->timestamp_rx = proc->timestamp_rx;
    slave_proc->timestamp_tx = proc->timestamp_tx;
    AssertFatal((ret=pthread_mutex_unlock( &slave_proc->mutex_FH ))==0,"mutex_unlock returns %d\n",ret);

    if (cnt_slave == 0) {
      // the thread was presumably waiting where it should and can now be woken up
      if (pthread_cond_signal(&slave_proc->cond_FH) != 0) {
        LOG_E( PHY, "ERROR pthread_cond_signal for RU %d, slave RU %d\n",proc->ru->idx,slave_proc->ru->idx);
        exit_fun( "ERROR pthread_cond_signal" );
        break;
      }
    } else {
      LOG_W( PHY,"[RU] Frame %d, slave %d thread busy!! (cnt_FH %i)\n",slave_proc->frame_rx,slave_proc->ru->idx, cnt_slave);
      exit_fun( "FH thread busy" );
      break;
    }
  }
}


/*!
 * \brief The prach receive thread of RU.
 * \param param is a \ref RU_proc_t structure which contains the info what to process.
 * \returns a pointer to an int. The storage is not on the heap and must not be freed.
 */
void *ru_thread_prach( void *param ) {
  static int ru_thread_prach_status;
  RU_t *ru        = (RU_t *)param;
  RU_proc_t *proc = (RU_proc_t *)&ru->proc;
  // set default return value
  ru_thread_prach_status = 0;
  thread_top_init("ru_thread_prach",1,500000,1000000,20000000);
  //wait_sync("ru_thread_prach");

  while (*ru->ru_mask>0 && ru->function!=eNodeB_3GPP) {
    usleep(1e6);
    LOG_D(PHY,"%s() RACH waiting for RU to be configured\n", __FUNCTION__);
  }

  LOG_I(PHY,"%s() RU configured - RACH processing thread running\n", __FUNCTION__);

  while (!oai_exit) {
    if (wait_on_condition(&proc->mutex_prach,&proc->cond_prach,&proc->instance_cnt_prach,"ru_prach_thread") < 0) break;

    if (oai_exit) break;

    VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_RU_PRACH_RX, 1 );

    if (ru->eNB_list[0]) {
      prach_procedures(
        ru->eNB_list[0],0
      );
    } else {
      rx_prach(NULL,
               ru,
               NULL,
               NULL,
               NULL,
               NULL,
               proc->frame_prach,
               0,0
              );
    }

    VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_RU_PRACH_RX, 0 );

    if (release_thread(&proc->mutex_prach,&proc->instance_cnt_prach,"ru_prach_thread") < 0) break;
  }

  LOG_I(PHY, "Exiting RU thread PRACH\n");
  ru_thread_prach_status = 0;
  return &ru_thread_prach_status;
}


void *ru_thread_prach_br( void *param ) {
  static int ru_thread_prach_status;
  RU_t *ru        = (RU_t *)param;