[Discuss-gnuradio] Threaded c++ only USRP "cfile" dump

[Chris Stankevitz]

Recently I have been asking questions about overruns, threaded blocks, 
and c++ only USRP reading.  My application requires uninterrupted data, 
which I was unable to get from the USRP with my laptop, even just 
recording data to a file_sink.  Attached is a program that records data 
from the USRP.  It is entirely in c++ and uses threads to prevent losing 
data by writing to the hard drive while it reads from the USRP.

Thanks to Ian Larsen for a C++-only USRP example and to Greg Heckler for 
the DBSRX c++ driver.

In addition to the attached, you also need db_dbs_rx.cpp and db_dbs_rx.h 
from http://lists.gnu.org/archive/html/patch-gnuradio/2007-08/msg00000.html

Compile with -lusrp and -lpthread

Chris

#include "db_dbs_rx.h"
#include <usrp_standard.h>
#include <fstream>
#include <iostream>
//#include <pthread.h>

using namespace std;

#define NUMBUFFERS 16
#define NUMBYTES (4096*sizeof(short)*2/512*512)

void* Producer(void* Args);
void* Consumer(void* Args);

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
struct TSQueue
{
  TSQueue();
  ~TSQueue();

  char Buffers[NUMBUFFERS][NUMBYTES];
  unsigned NumBytes[NUMBUFFERS];
  unsigned Head, Tail;
  bool Full;
  bool Empty;

  void Add(const char* in, unsigned n);
  void Delete(char* out, unsigned& n);

  pthread_mutex_t* pMutex;
  pthread_cond_t* pNotFullCond;
  pthread_cond_t* pNotEmptyCond;
};
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
TSQueue::TSQueue()
{
  memset(NumBytes, 0, sizeof(NumBytes));

  Empty = true;
  Full = false;
  Head = 0;
  Tail = 0;
  pMutex = new pthread_mutex_t;
  pthread_mutex_init (pMutex, NULL);
  pNotFullCond = new pthread_cond_t;
  pthread_cond_init (pNotFullCond, NULL);
  pNotEmptyCond = new pthread_cond_t;
  pthread_cond_init (pNotEmptyCond, NULL);
}
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
TSQueue::~TSQueue()
{
  pthread_mutex_destroy (pMutex);
  delete pMutex;  
  pthread_cond_destroy (pNotFullCond);
  delete pNotFullCond;
  pthread_cond_destroy (pNotEmptyCond);
  delete pNotEmptyCond;
}
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

void TSQueue::Add (const char* in, unsigned n)
{
  if(Full)
  {
    pthread_cond_wait(pNotFullCond, pMutex);
  }

  NumBytes[Tail] = n;
  memcpy(Buffers[Tail], in, n);
  ++Tail;
  if (Tail == NUMBUFFERS) Tail = 0;

  if (Tail == Head) Full = true;
  Empty = false;

  pthread_cond_signal(pNotEmptyCond);
}
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

void TSQueue::Delete(char *out, unsigned& n)
{
  if(Empty)
  {
    pthread_cond_wait(pNotEmptyCond, pMutex);
  }

  n = NumBytes[Head];
  memcpy(out, Buffers[Head], n);

  ++Head;
  if (Head == NUMBUFFERS) Head = 0;
  if (Head == Tail) Empty = true;
  Full = false;

  pthread_cond_signal(pNotFullCond);
}
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
struct TSProducerArg
{
  TSQueue* pQueue;
  usrp_standard_rx* pRx;
};
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

struct TSConsumerArg
{
  TSQueue* pQueue;
  ofstream* pStream;
};
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
void* Producer(void* q)
{
  TSProducerArg* pArg = reinterpret_cast<TSProducerArg*>(q);
  TSQueue *pQueue = pArg->pQueue;
  usrp_standard_rx* rx = pArg->pRx;

  rx->start();

  while(true)
  {
    bool Overrun;
    char Buffer[NUMBYTES];
    int Size = rx->read(Buffer, NUMBYTES, &Overrun);

    if(Size == -1)
    {
      cerr << "Problem" << endl;
    }
    else
    {
      if(Overrun)
      { 
        cerr << "overrun" << endl;
      }

      pQueue->Add(Buffer, Size);
    }
  }

  rx->stop();

  return 0;
}
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
void* Consumer(void* q)
{
  TSConsumerArg* pArg = reinterpret_cast<TSConsumerArg*>(q);
  TSQueue *pQueue = pArg->pQueue;
  ofstream* pStream = pArg->pStream;

  while(true)
  {
    unsigned n = 0;
    char Buffer[NUMBYTES];
    pQueue->Delete(Buffer, n);

    pStream->write(Buffer, n);
  }

  return 0;
}
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
int main(int argc, char** argv)
{
  if(argc < 2)
  {
    cout << "Usage: " << argv[0] << " outfilename" << endl;
    return 1;
  }

  usrp_standard_rx* rx = usrp_standard_rx::make(0, 16);

  db_dbs_rx dbsrx(rx, 0);
  static const double Freq = 1.57542e9;
  //  Add 600kHz to remove birdie
  double TunedFreq = dbsrx.tune(Freq+600e3);
  dbsrx.gain(25);
  double DdcFreq = TunedFreq-Freq;
  static const int Mux = 0x00000010;
  static const unsigned int Format = 768;
  static const unsigned DecimRate = 16;
  rx->set_decim_rate(DecimRate);
  rx->set_rx_freq(0, DdcFreq);
  rx->set_mux(Mux);
  rx->set_format(Format);

  ofstream s(argv[1], ios::binary);

  TSQueue* pQueue = new TSQueue;
  pthread_t pro, con;

  TSProducerArg ProdArg = {pQueue, rx};
  TSConsumerArg ConsArg = {pQueue, &s};

  pthread_create (&pro, NULL, Producer, &ProdArg);
  pthread_create (&con, NULL, Consumer, &ConsArg);
  pthread_join (pro, NULL);
  pthread_join (con, NULL);

  delete pQueue;

  return 0;
}