bladerf: migration to async api (WIP)

receive works, transmit locks up in work() after few seconds

6 files changed, 345 insertions, 294 deletions

diff --git a/lib/bladerf/bladerf_common.cc b/lib/bladerf/bladerf_common.cc
index 43b58ec..10884dd 100644
--- a/lib/bladerf/bladerf_common.cc
+++ b/lib/bladerf/bladerf_common.cc
@@ -43,18 +43,12 @@
 
 using namespace boost::assign;
 
-bladerf_common::bladerf_common() : running(true)
+bladerf_common::bladerf_common() :
+  _is_running(false)
 {
   const char *env_fifo_size;
   size_t fifo_size;
 
-  /* 1 Sample = i,q (2 int16_t's) */
-  this->raw_sample_buf = new int16_t[2 * BLADERF_SAMPLE_BLOCK_SIZE];
-  if (!raw_sample_buf) {
-    throw std::runtime_error( std::string(__FUNCTION__) +
-                              " has failed to allocate a raw sample buffer!" );
-  }
-
   env_fifo_size = getenv(BLADERF_FIFO_SIZE_ENV);
   fifo_size = BLADERF_SAMPLE_FIFO_SIZE;
 
@@ -75,16 +69,15 @@ bladerf_common::bladerf_common() : running(true)
     }
   }
 
-  this->sample_fifo = new boost::circular_buffer<gr_complex>(fifo_size);
-  if (!this->sample_fifo)
+  _fifo = new boost::circular_buffer<gr_complex>(fifo_size);
+  if (!_fifo)
     throw std::runtime_error( std::string(__FUNCTION__) +
                               " has failed to allocate a sample FIFO!" );
 }
 
 bladerf_common::~bladerf_common()
 {
-  delete[] this->raw_sample_buf;
-  delete this->sample_fifo;
+  delete _fifo;
 }
 
 osmosdr::freq_range_t bladerf_common::freq_range()
@@ -116,7 +109,7 @@ osmosdr::freq_range_t bladerf_common::filter_bandwidths()
       2.75, 3, 3.5, 4.375, 5, 6, 7, 10, 14;
 
   BOOST_FOREACH( double half_bw, half_bandwidths )
-    bandwidths += osmosdr::range_t( half_bw * 2.e6 );
+    bandwidths += osmosdr::range_t( half_bw * 2e6 );
 
   return bandwidths;
 }
@@ -129,9 +122,10 @@ std::vector< std::string > bladerf_common::devices()
 
   n_devices = bladerf_get_device_list(&devices);
 
-  if (n_devices > 0) {
-    for (ssize_t i = 0; i < n_devices; i++)  {
-
+  if (n_devices > 0)
+  {
+    for (ssize_t i = 0; i < n_devices; i++)
+    {
       std::stringstream s;
       std::string serial(devices[i].serial);
 
@@ -154,12 +148,14 @@ std::vector< std::string > bladerf_common::devices()
 
 bool bladerf_common::is_running()
 {
-  boost::shared_lock<boost::shared_mutex> lock(this->state_lock);
-  return this->running;
+  boost::shared_lock<boost::shared_mutex> lock(_state_lock);
+
+  return _is_running;
 }
 
-void bladerf_common::set_running(bool is_running)
+void bladerf_common::set_running( bool is_running )
 {
-  boost::unique_lock<boost::shared_mutex> lock(this->state_lock);
-  this->running = is_running;
+  boost::unique_lock<boost::shared_mutex> lock(_state_lock);
+
+  _is_running = is_running;
 }
diff --git a/lib/bladerf/bladerf_common.h b/lib/bladerf/bladerf_common.h
index a8480c0..ca95850 100644
--- a/lib/bladerf/bladerf_common.h
+++ b/lib/bladerf/bladerf_common.h
@@ -29,6 +29,7 @@
 #include <boost/thread/shared_mutex.hpp>
 #include <boost/thread/condition_variable.hpp>
 
+#include <gnuradio/thread/thread.h>
 #include <gnuradio/gr_complex.h>
 
 #include <libbladeRF.h>
@@ -52,7 +53,7 @@ class bladerf_common
 {
 public:
   bladerf_common();
-  ~bladerf_common();
+  virtual ~bladerf_common();
 
 protected:
   osmosdr::freq_range_t freq_range();
@@ -64,15 +65,24 @@ protected:
   bool is_running();
   void set_running(bool is_running);
 
-  bladerf *dev;
+  bladerf *_dev;
 
-  int16_t *raw_sample_buf;
-  boost::circular_buffer<gr_complex> *sample_fifo;
-  boost::mutex sample_fifo_lock;
-  boost::condition_variable samples_available;
+  void **_buffers;
+  struct bladerf_stream *_stream;
+  size_t _num_buffers;
+  size_t _buf_index;
+
+  gr::thread::thread _thread;
+
+  boost::circular_buffer<gr_complex> *_fifo;
+  boost::mutex _fifo_lock;
+  boost::condition_variable _samp_avail;
+
+  osmosdr::gain_range_t _vga1_range;
+  osmosdr::gain_range_t _vga2_range;
 private:
-  bool running;
-  boost::shared_mutex state_lock;
+  bool _is_running;
+  boost::shared_mutex _state_lock;
 };
 
 #endif
diff --git a/lib/bladerf/bladerf_sink_c.cc b/lib/bladerf/bladerf_sink_c.cc
index fe63ce8..2cd59f7 100644
--- a/lib/bladerf/bladerf_sink_c.cc
+++ b/lib/bladerf/bladerf_sink_c.cc
@@ -95,73 +95,83 @@ bladerf_sink_c::bladerf_sink_c (const std::string &args)
   device_name = boost::str(boost::format( "libusb:instance=%d" ) % device_number);
 
   /* Open a handle to the device */
-  ret = bladerf_open( &this->dev, device_name.c_str() );
+  ret = bladerf_open( &_dev, device_name.c_str() );
   if ( ret != 0 ) {
     throw std::runtime_error( std::string(__FUNCTION__) + " " +
                               "failed to open bladeRF device " + device_name );
   }
 
-  if (dict.count("fpga"))
+  if (dict.count("fw"))
   {
-    std::string fpga = dict["fpga"];
+    std::string fw = dict["fw"];
 
-    std::cerr << "Loading FPGA bitstream " << fpga << "..." << std::endl;
-    ret = bladerf_load_fpga( this->dev, fpga.c_str() );
+    std::cerr << "Flashing firmware image " << fw << "..., DO NOT INTERRUPT!"
+              << std::endl;
+    ret = bladerf_flash_firmware( _dev, fw.c_str() );
     if ( ret != 0 )
-      std::cerr << "bladerf_load_fpga has returned with " << ret << std::endl;
+      std::cerr << "bladerf_flash_firmware has failed with " << ret << std::endl;
     else
-      std::cerr << "The FPGA bitstream has been successfully loaded." << std::endl;
+      std::cerr << "The firmware has been successfully flashed." << std::endl;
   }
 
-  if (dict.count("fw"))
+  if (dict.count("fpga"))
   {
-    std::string fw = dict["fw"];
+    std::string fpga = dict["fpga"];
 
-    std::cerr << "Flashing firmware image " << fw << "..., "
-              << "DO NOT INTERRUPT!"
-              << std::endl;
-    ret = bladerf_flash_firmware( this->dev, fw.c_str() );
-    if ( ret != 0 )
-      std::cerr << "bladerf_flash_firmware has failed with " << ret << std::endl;
+    std::cerr << "Loading FPGA bitstream " << fpga << "..." << std::endl;
+    ret = bladerf_load_fpga( _dev, fpga.c_str() );
+    if ( ret != 0 && ret != 1 )
+      std::cerr << "bladerf_load_fpga has failed with " << ret << std::endl;
     else
-      std::cerr << "The firmare has been successfully flashed, "
-                << "please power cycle the bladeRF before using it."
-                << std::endl;
+      std::cerr << "The FPGA bitstream has been successfully loaded." << std::endl;
   }
 
   std::cerr << "Using nuand LLC bladeRF #" << device_number;
 
-  char serial[33];
-  if ( bladerf_get_serial( this->dev, serial ) == 0 )
+  char serial[BLADERF_SERIAL_LENGTH];
+  if ( bladerf_get_serial( _dev, serial ) == 0 )
     std::cerr << " SN " << serial;
 
   unsigned int major, minor;
-  if ( bladerf_get_fw_version( this->dev, &major, &minor) == 0 )
+  if ( bladerf_get_fw_version( _dev, &major, &minor) == 0 )
     std::cerr << " FW v" << major << "." << minor;
 
-  if ( bladerf_get_fpga_version( this->dev, &major, &minor) == 0 )
+  if ( bladerf_get_fpga_version( _dev, &major, &minor) == 0 )
     std::cerr << " FPGA v" << major << "." << minor;
 
   std::cerr << std::endl;
 
-  if ( bladerf_is_fpga_configured( this->dev ) != 1 )
+  if ( bladerf_is_fpga_configured( _dev ) != 1 )
   {
-    std::cerr << "ERROR: The FPGA is not configured! "
-              << "Use the device argument fpga=/path/to/the/bitstream.rbf to load it."
-              << std::endl;
+    std::ostringstream oss;
+    oss << "The FPGA is not configured! "
+        << "Provide device argument fpga=/path/to/the/bitstream.rbf to load it.";
+
+    throw std::runtime_error( oss.str() );
   }
 
   /* Set the range of VGA1, VGA1GAINT[7:0] */
-  this->vga1_range = osmosdr::gain_range_t( -35, -4, 1 );
+  _vga1_range = osmosdr::gain_range_t( -35, -4, 1 );
 
   /* Set the range of VGA2, VGA2GAIN[4:0] */
-  this->vga2_range = osmosdr::gain_range_t( 0, 25, 1 );
+  _vga2_range = osmosdr::gain_range_t( 0, 25, 1 );
+
+  _buf_index = 0;
+  _num_buffers = 8; /* TODO: make it an argument */
+  const size_t samp_per_buf = 1024 * 10; /* TODO: make it an argument */
+
+  /* Initialize the stream */
+  ret = bladerf_init_stream( &_stream, _dev, stream_callback,
+                             &_buffers, _num_buffers, BLADERF_FORMAT_SC16_Q12,
+                             samp_per_buf, _num_buffers, this );
+  if ( ret != 0 )
+    std::cerr << "bladerf_init_stream has failed with " << ret << std::endl;
 
-  ret = bladerf_enable_module(this->dev, BLADERF_MODULE_TX, true);
+  ret = bladerf_enable_module( _dev, BLADERF_MODULE_TX, true );
   if ( ret != 0 )
-    std::cerr << "bladerf_enable_module has returned with " << ret << std::endl;
+    std::cerr << "bladerf_enable_module has failed with " << ret << std::endl;
 
-	this->thread = gr::thread::thread(write_task_dispatch, this);
+  _thread = gr::thread::thread( boost::bind(&bladerf_sink_c::write_task, this) );
 }
 
 /*
@@ -171,78 +181,89 @@ bladerf_sink_c::~bladerf_sink_c ()
 {
   int ret;
 
-  this->set_running(false);
-  this->thread.join();
+  set_running(false);
+  _thread.join();
 
-  ret = bladerf_enable_module(this->dev, BLADERF_MODULE_TX, false);
+  ret = bladerf_enable_module( _dev, BLADERF_MODULE_TX, false );
   if ( ret != 0 )
-    std::cerr << "bladerf_enable_module has returned with " << ret << std::endl;
+    std::cerr << "bladerf_enable_module has failed with " << ret << std::endl;
+
+  /* Release stream resources */
+  bladerf_deinit_stream(_stream);
 
   /* Close the device */
-  bladerf_close( this->dev );
+  bladerf_close( _dev );
 }
 
-void bladerf_sink_c::write_task_dispatch(bladerf_sink_c *obj)
+void *bladerf_sink_c::stream_callback( struct bladerf *dev,
+                                       struct bladerf_stream *stream,
+                                       struct bladerf_metadata *metadata,
+                                       void *samples,
+                                       size_t num_samples,
+                                       void *user_data )
 {
-  obj->write_task();
+  bladerf_sink_c *obj = (bladerf_sink_c *) user_data;
+
+  if ( ! obj->is_running() )
+    return NULL;
+
+  return obj->stream_task( samples, num_samples );
 }
 
-void bladerf_sink_c::write_task()
+/* Convert & push samples to the sample fifo */
+void *bladerf_sink_c::stream_task( void *samples, size_t num_samples )
 {
-  int i, n_samples_avail, n_samples;
-  int16_t *p;
-  gr_complex sample;
+  size_t i, n_avail;
+  void *ret;
 
-  while ( this->is_running() )
-  {
+  ret = _buffers[_buf_index];
+  _buf_index = (_buf_index + 1) % _num_buffers;
 
+  while ( is_running() )
+  {
     {
       /* Lock the circular buffer */
-      boost::unique_lock<boost::mutex> lock(this->sample_fifo_lock);
+      boost::unique_lock<boost::mutex> lock(_fifo_lock);
 
       /* Check to make sure we have samples available */
-      n_samples_avail = this->sample_fifo->size();
-      while( n_samples_avail < BLADERF_SAMPLE_BLOCK_SIZE ) {
+      n_avail = _fifo->size();
+      while( n_avail < num_samples ) {
         /* Wait until there is at least a block size of samples ready */
-        this->samples_available.wait(lock);
-        n_samples_avail = this->sample_fifo->size();
+        _samp_avail.wait(lock);
+        n_avail = _fifo->size();
       }
 
       /* Pop samples from circular buffer, write samples to outgoing buffer */
-      int16_t *p = this->raw_sample_buf;
-      for( i = 0; i < BLADERF_SAMPLE_BLOCK_SIZE; ++i ) {
-        sample = this->sample_fifo->at(0);
-        this->sample_fifo->pop_front();
+      int16_t *p = (int16_t *) ret;
+      for( i = 0; i < num_samples; ++i ) {
+        gr_complex sample = _fifo->at(0);
+        _fifo->pop_front();
         *p++ = 0xa000 | (int16_t)(real(sample)*2000);
         *p++ = 0x5000 | (int16_t)(imag(sample)*2000);
       }
-    } /* Give up the lock by leaving the scope ...*/
+    } /* Give up the lock by leaving the scope ... */
 
     /* Notify that we've just popped some samples */
-    this->samples_available.notify_one();
-
-    /* Samples are available to write out */
-    n_samples = bladerf_tx(this->dev, BLADERF_FORMAT_SC16_Q12, this->raw_sample_buf,
-                                 BLADERF_SAMPLE_BLOCK_SIZE, NULL);
-
-    /* Check n_samples return value */
-    if( n_samples < 0 ) {
-      std::cerr << "Failed to write samples: "
-                << bladerf_strerror(n_samples) << std::endl;
-      this->set_running(false);
-    } else {
-      if(n_samples != BLADERF_SAMPLE_BLOCK_SIZE) {
-        if(n_samples > BLADERF_SAMPLE_BLOCK_SIZE) {
-          std::cerr << "Warning: sent bloated sample block of "
-                    << n_samples << " samples!" << std::endl;
-        } else {
-          std::cerr << "Warning: sent truncated sample block of "
-                    << n_samples << " samples!" << std::endl;
-        }
-      }
-    }
-
+    //std::cerr << "-" << std::flush;
+    _samp_avail.notify_one();
   }
+
+  return ret;
+}
+
+void bladerf_sink_c::write_task()
+{
+  int status;
+
+  set_running( true );
+
+  /* Start stream and stay there until we kill the stream */
+  status = bladerf_stream(_stream, BLADERF_MODULE_TX);
+
+  if (status < 0)
+      std::cerr << "Sink stream error: " << bladerf_strerror(status) << std::endl;
+
+  set_running( false );
 }
 
 int bladerf_sink_c::work( int noutput_items,
@@ -252,44 +273,43 @@ int bladerf_sink_c::work( int noutput_items,
   int n_space_avail, to_copy, limit, i;
   const gr_complex *in = (const gr_complex *) input_items[0];
 
-  if ( ! this->is_running() )
+  if ( ! is_running() )
     return WORK_DONE;
 
-  if( noutput_items >= 0 ) {
-    /* Total samples we want to process */
-    to_copy = noutput_items;
+  /* Total samples we want to process */
+  to_copy = noutput_items;
 
-    /* While there are still samples to copy out ... */
-    while( to_copy > 0 ) {
-      {
-        /* Acquire the circular buffer lock */
-        boost::unique_lock<boost::mutex> lock(this->sample_fifo_lock);
+  /* While there are still samples to copy out ... */
+  while( to_copy > 0 ) {
+    {
+      /* Acquire the circular buffer lock */
+      boost::unique_lock<boost::mutex> lock(_fifo_lock);
 
-        /* Check to see how much space is available */
-        n_space_avail = this->sample_fifo->capacity() - this->sample_fifo->size();
+      /* Check to see how much space is available */
+      n_space_avail = _fifo->capacity() - _fifo->size();
 
-        while (n_space_avail == 0) {
-          this->samples_available.wait(lock);
-          n_space_avail = this->sample_fifo->capacity() - this->sample_fifo->size();
-        }
+      while (n_space_avail == 0) {
+        _samp_avail.wait(lock);
+        n_space_avail = _fifo->capacity() - _fifo->size();
+      }
 
-        /* Limit ourselves to either the number of output items ...
-           ... or whatever space is available */
-        limit = (n_space_avail < noutput_items ? n_space_avail : noutput_items);
+      /* Limit ourselves to either the number of output items ...
+         ... or whatever space is available */
+      limit = (n_space_avail < noutput_items ? n_space_avail : noutput_items);
 
-        /* Consume! */
-        for( i = 0; i < limit; i++ ) {
-          this->sample_fifo->push_back(*in++);
-        }
+      /* Consume! */
+      for( i = 0; i < limit; i++ ) {
+        _fifo->push_back(*in++);
+      }
 
-        /* Decrement the amount we need to copy */
-        to_copy -= limit;
+      /* Decrement the amount we need to copy */
+      to_copy -= limit;
 
-      } /* Unlock by leaving the scope */
+    } /* Unlock by leaving the scope */
 
-      /* Notify that we've just added some samples */
-      this->samples_available.notify_one();
-    }
+    /* Notify that we've just added some samples */
+    //std::cerr << "+" << std::flush;
+    _samp_avail.notify_one();
   }
 
   return noutput_items;
@@ -308,7 +328,7 @@ size_t bladerf_sink_c::get_num_channels()
 
 osmosdr::meta_range_t bladerf_sink_c::get_sample_rates()
 {
-  return this->sample_rates();
+  return sample_rates();
 }
 
 double bladerf_sink_c::set_sample_rate(double rate)
@@ -320,7 +340,7 @@ double bladerf_sink_c::set_sample_rate(double rate)
   /* Check to see if the sample rate is an integer */
   if( (uint32_t)round(rate) == (uint32_t)rate )
   {
-    ret = bladerf_set_sample_rate( this->dev, BLADERF_MODULE_TX, (uint32_t)rate, &actual );
+    ret = bladerf_set_sample_rate( _dev, BLADERF_MODULE_TX, (uint32_t)rate, &actual );
     if( ret ) {
       throw std::runtime_error( std::string(__FUNCTION__) + " " +
                                 "has failed to set integer rate, error " +
@@ -328,7 +348,7 @@ double bladerf_sink_c::set_sample_rate(double rate)
     }
   } else {
     /* TODO: Fractional sample rate */
-    ret = bladerf_set_sample_rate( this->dev, BLADERF_MODULE_TX, (uint32_t)rate, &actual );
+    ret = bladerf_set_sample_rate( _dev, BLADERF_MODULE_TX, (uint32_t)rate, &actual );
     if( ret ) {
       throw std::runtime_error( std::string(__FUNCTION__) + " " +
                                 "has failed to set fractional rate, error " +
@@ -344,7 +364,7 @@ double bladerf_sink_c::get_sample_rate()
   int ret;
   unsigned int rate = 0;
 
-  ret = bladerf_get_sample_rate( this->dev, BLADERF_MODULE_TX, &rate );
+  ret = bladerf_get_sample_rate( _dev, BLADERF_MODULE_TX, &rate );
   if( ret ) {
     throw std::runtime_error( std::string(__FUNCTION__) + " " +
                               "has failed to get sample rate, error " +
@@ -356,7 +376,7 @@ double bladerf_sink_c::get_sample_rate()
 
 osmosdr::freq_range_t bladerf_sink_c::get_freq_range( size_t chan )
 {
-  return this->freq_range();
+  return freq_range();
 }
 
 double bladerf_sink_c::set_center_freq( double freq, size_t chan )
@@ -368,7 +388,7 @@ double bladerf_sink_c::set_center_freq( double freq, size_t chan )
       freq > get_freq_range( chan ).stop() ) {
     std::cerr << "Failed to set out of bound frequency: " << freq << std::endl;
   } else {
-    ret = bladerf_set_frequency( this->dev, BLADERF_MODULE_TX, (uint32_t)freq );
+    ret = bladerf_set_frequency( _dev, BLADERF_MODULE_TX, (uint32_t)freq );
     if( ret ) {
       throw std::runtime_error( std::string(__FUNCTION__) + " " +
                                 "failed to set center frequency " +
@@ -386,7 +406,7 @@ double bladerf_sink_c::get_center_freq( size_t chan )
   uint32_t freq;
   int ret;
 
-  ret = bladerf_get_frequency( this->dev, BLADERF_MODULE_TX, &freq );
+  ret = bladerf_get_frequency( _dev, BLADERF_MODULE_TX, &freq );
   if( ret ) {
     throw std::runtime_error( std::string(__FUNCTION__) + " " +
                               "failed to get center frequency, error " +
@@ -429,9 +449,9 @@ osmosdr::gain_range_t bladerf_sink_c::get_gain_range( const std::string & name,
   osmosdr::gain_range_t range;
 
   if( name == "VGA1" ) {
-    range = this->vga1_range;
+    range = _vga1_range;
   } else if( name == "VGA2" ) {
-    range = this->vga2_range;
+    range = _vga2_range;
   } else {
     throw std::runtime_error( std::string(__FUNCTION__) + " " +
                               "requested an invalid gain element " + name );
@@ -460,9 +480,9 @@ double bladerf_sink_c::set_gain( double gain, const std::string & name, size_t c
   int ret = 0;
 
   if( name == "VGA1" ) {
-    ret = bladerf_set_txvga1( this->dev, (int)gain );
+    ret = bladerf_set_txvga1( _dev, (int)gain );
   } else if( name == "VGA2" ) {
-    ret = bladerf_set_txvga2( this->dev, (int)gain );
+    ret = bladerf_set_txvga2( _dev, (int)gain );
   } else {
     throw std::runtime_error( std::string(__FUNCTION__) + " " +
                               "requested to set the gain "
@@ -490,9 +510,9 @@ double bladerf_sink_c::get_gain( const std::string & name, size_t chan )
   int ret = 0;
 
   if( name == "VGA1" ) {
-    ret = bladerf_get_txvga1( this->dev, &g );
+    ret = bladerf_get_txvga1( _dev, &g );
   } else if( name == "VGA2" ) {
-    ret = bladerf_get_txvga2( this->dev, &g );
+    ret = bladerf_get_txvga2( _dev, &g );
   } else {
     throw std::runtime_error( std::string(__FUNCTION__) + " " +
                               "requested to get the gain "
@@ -545,14 +565,14 @@ double bladerf_sink_c::set_bandwidth( double bandwidth, size_t chan )
   int ret;
   uint32_t actual;
 
-  ret = bladerf_set_bandwidth( this->dev, BLADERF_MODULE_TX, (uint32_t)bandwidth, &actual );
+  ret = bladerf_set_bandwidth( _dev, BLADERF_MODULE_TX, (uint32_t)bandwidth, &actual );
   if( ret ) {
     throw std::runtime_error( std::string(__FUNCTION__) + " " +
                               "could not set bandwidth, error " +
                               boost::lexical_cast<std::string>(ret) );
   }
 
-  return this->get_bandwidth();
+  return get_bandwidth();
 }
 
 double bladerf_sink_c::get_bandwidth( size_t chan )
@@ -560,7 +580,7 @@ double bladerf_sink_c::get_bandwidth( size_t chan )
   uint32_t bandwidth;
   int ret;
 
-  ret = bladerf_get_bandwidth( this->dev, BLADERF_MODULE_TX, &bandwidth );
+  ret = bladerf_get_bandwidth( _dev, BLADERF_MODULE_TX, &bandwidth );
   if( ret ) {
     throw std::runtime_error( std::string(__FUNCTION__) + " " +
                               "could not get bandwidth, error " +
@@ -572,5 +592,5 @@ double bladerf_sink_c::get_bandwidth( size_t chan )
 
 osmosdr::freq_range_t bladerf_sink_c::get_bandwidth_range( size_t chan )
 {
-  return this->filter_bandwidths();
+  return filter_bandwidths();
 }
diff --git a/lib/bladerf/bladerf_sink_c.h b/lib/bladerf/bladerf_sink_c.h
index 8db1a8e..8fa90f4 100644
--- a/lib/bladerf/bladerf_sink_c.h
+++ b/lib/bladerf/bladerf_sink_c.h
@@ -25,8 +25,6 @@
 #include <gnuradio/block.h>
 #include <gnuradio/sync_block.h>
 
-#include <libbladeRF.h>
-
 #include "osmosdr/ranges.h"
 #include "sink_iface.h"
 #include "bladerf_common.h"
@@ -108,13 +106,20 @@ public:
   double get_bandwidth( size_t chan = 0 );
   osmosdr::freq_range_t get_bandwidth_range( size_t chan = 0 );
 
-private:
-  static void write_task_dispatch(bladerf_sink_c *obj);
+private: /* functions */
+  static void *stream_callback( struct bladerf *_dev,
+                                struct bladerf_stream *stream,
+                                struct bladerf_metadata *metadata,
+                                void *samples,
+                                size_t num_samples,
+                                void *user_data );
+
+  void *stream_task(void *samples, size_t num_samples);
+
   void write_task();
 
-  gr::thread::thread thread;
-  osmosdr::gain_range_t vga1_range;
-  osmosdr::gain_range_t vga2_range;
+private: /* members */
+
 };
 
 #endif /* INCLUDED_BLADERF_SINK_C_H */
diff --git a/lib/bladerf/bladerf_source_c.cc b/lib/bladerf/bladerf_source_c.cc
index 99aa0e9..931e33b 100644
--- a/lib/bladerf/bladerf_source_c.cc
+++ b/lib/bladerf/bladerf_source_c.cc
@@ -95,70 +95,60 @@ bladerf_source_c::bladerf_source_c (const std::string &args)
   device_name = boost::str(boost::format( "libusb:instance=%d" ) % device_number);
 
   /* Open a handle to the device */
-  ret = bladerf_open( &this->dev, NULL );
+  ret = bladerf_open( &_dev, device_name.c_str() );
   if ( ret != 0 ) {
     throw std::runtime_error( std::string(__FUNCTION__) + " " +
                               "failed to open bladeRF device " + device_name );
   }
 
-  if (dict.count("fpga"))
+  if (dict.count("fw"))
   {
-    std::string fpga = dict["fpga"];
+    std::string fw = dict["fw"];
 
-    std::cerr << "Loading FPGA bitstream " << fpga << "..." << std::endl;
-    ret = bladerf_load_fpga( this->dev, fpga.c_str() );
+    std::cerr << "Flashing firmware image " << fw << "..., DO NOT INTERRUPT!"
+              << std::endl;
+    ret = bladerf_flash_firmware( _dev, fw.c_str() );
     if ( ret != 0 )
-      std::cerr << "bladerf_load_fpga has returned with " << ret << std::endl;
+      std::cerr << "bladerf_flash_firmware has failed with " << ret << std::endl;
     else
-      std::cerr << "The FPGA bitstream has been successfully loaded." << std::endl;
+      std::cerr << "The firmware has been successfully flashed." << std::endl;
   }
 
-  if (dict.count("fw"))
+  if (dict.count("fpga"))
   {
-    std::string fw = dict["fw"];
+    std::string fpga = dict["fpga"];
 
-    std::cerr << "Flashing firmware image " << fw << "..., "
-              << "DO NOT INTERRUPT!"
-              << std::endl;
-    ret = bladerf_flash_firmware( this->dev, fw.c_str() );
-    if ( ret != 0 )
-      std::cerr << "bladerf_flash_firmware has failed with " << ret << std::endl;
+    std::cerr << "Loading FPGA bitstream " << fpga << "..." << std::endl;
+    ret = bladerf_load_fpga( _dev, fpga.c_str() );
+    if ( ret != 0 && ret != 1 )
+      std::cerr << "bladerf_load_fpga has failed with " << ret << std::endl;
     else
-      std::cerr << "The firmare has been successfully flashed, "
-                << "please power cycle the bladeRF before using it."
-                << std::endl;
+      std::cerr << "The FPGA bitstream has been successfully loaded." << std::endl;
   }
 
   std::cerr << "Using nuand LLC bladeRF #" << device_number;
 
-  char serial[33];
-  if ( bladerf_get_serial( this->dev, serial ) == 0 )
+  char serial[BLADERF_SERIAL_LENGTH];
+  if ( bladerf_get_serial( _dev, serial ) == 0 )
     std::cerr << " SN " << serial;
 
   unsigned int major, minor;
-  if ( bladerf_get_fw_version( this->dev, &major, &minor) == 0 )
+  if ( bladerf_get_fw_version( _dev, &major, &minor) == 0 )
     std::cerr << " FW v" << major << "." << minor;
 
-  if ( bladerf_get_fpga_version( this->dev, &major, &minor) == 0 )
+  if ( bladerf_get_fpga_version( _dev, &major, &minor) == 0 )
     std::cerr << " FPGA v" << major << "." << minor;
 
   std::cerr << std::endl;
 
-  if ( bladerf_is_fpga_configured( this->dev ) != 1 )
+  if ( bladerf_is_fpga_configured( _dev ) != 1 )
   {
-    std::cerr << "ERROR: The FPGA is not configured! "
-              << "Use the device argument fpga=/path/to/the/bitstream.rbf to load it."
-              << std::endl;
-  }
+    std::ostringstream oss;
+    oss << "The FPGA is not configured! "
+        << "Provide device argument fpga=/path/to/the/bitstream.rbf to load it.";
 
-  /* Set the range of LNA, G_LNA_RXFE[1:0] */
-  this->lna_range = osmosdr::gain_range_t( 0, 6, 3 );
-
-  /* Set the range of VGA1, RFB_TIA_RXFE[6:0], nonlinear mapping done inside the lib */
-  this->vga1_range = osmosdr::gain_range_t( 5, 30, 1 );
-
-  /* Set the range of VGA2 VGA2GAIN[4:0], not recommended to be used above 30dB */
-  this->vga2_range = osmosdr::gain_range_t( 0, 60, 3 );
+    throw std::runtime_error( oss.str() );
+  }
 
   if (dict.count("sampling"))
   {
@@ -166,12 +156,11 @@ bladerf_source_c::bladerf_source_c (const std::string &args)
 
     std::cerr << "Setting bladerf sampling to " << sampling << std::endl;
     if( sampling == "internal") {
-      ret = bladerf_set_sampling( this->dev, BLADERF_SAMPLING_INTERNAL );
+      ret = bladerf_set_sampling( _dev, BLADERF_SAMPLING_INTERNAL );
       if ( ret != 0 )
         std::cerr << "Problem while setting sampling mode " << ret << std::endl;
-
     } else if( sampling == "external" ) {
-      ret = bladerf_set_sampling( this->dev, BLADERF_SAMPLING_EXTERNAL );
+      ret = bladerf_set_sampling( _dev, BLADERF_SAMPLING_EXTERNAL );
       if ( ret != 0 )
         std::cerr << "Problem while setting sampling mode " << ret << std::endl;
     } else {
@@ -179,12 +168,31 @@ bladerf_source_c::bladerf_source_c (const std::string &args)
     }
   }
 
+  /* Set the range of LNA, G_LNA_RXFE[1:0] */
+  _lna_range = osmosdr::gain_range_t( 0, 6, 3 );
+
+  /* Set the range of VGA1, RFB_TIA_RXFE[6:0], nonlinear mapping done inside the lib */
+  _vga1_range = osmosdr::gain_range_t( 5, 30, 1 );
 
-  ret = bladerf_enable_module(this->dev, BLADERF_MODULE_RX, true);
+  /* Set the range of VGA2 VGA2GAIN[4:0], not recommended to be used above 30dB */
+  _vga2_range = osmosdr::gain_range_t( 0, 60, 3 );
+
+  _buf_index = 0;
+  _num_buffers = 8; /* TODO: make it an argument */
+  const size_t samp_per_buf = 1024 * 10; /* TODO: make it an argument */
+
+  /* Initialize the stream */
+  ret = bladerf_init_stream( &_stream, _dev, stream_callback,
+                             &_buffers, _num_buffers, BLADERF_FORMAT_SC16_Q12,
+                             samp_per_buf, _num_buffers, this );
   if ( ret != 0 )
-    std::cerr << "bladerf_enable_module has returned with " << ret << std::endl;
+    std::cerr << "bladerf_init_stream has failed with " << ret << std::endl;
 
-  this->thread = gr::thread::thread(read_task_dispatch, this);
+  ret = bladerf_enable_module( _dev, BLADERF_MODULE_RX, true );
+  if ( ret != 0 )
+    std::cerr << "bladerf_enable_module has failed with " << ret << std::endl;
+
+  _thread = gr::thread::thread( boost::bind(&bladerf_source_c::read_task, this) );
 }
 
 /*
@@ -194,113 +202,121 @@ bladerf_source_c::~bladerf_source_c ()
 {
   int ret;
 
-  this->set_running(false);
-  this->thread.join();
+  set_running(false);
+  _thread.join();
 
-  ret = bladerf_enable_module(this->dev, BLADERF_MODULE_RX, false);
+  ret = bladerf_enable_module( _dev, BLADERF_MODULE_RX, false );
   if ( ret != 0 )
-    std::cerr << "bladerf_enable_module has returned with " << ret << std::endl;
+    std::cerr << "bladerf_enable_module has failed with " << ret << std::endl;
+
+  /* Release stream resources */
+  bladerf_deinit_stream(_stream);
 
   /* Close the device */
-  bladerf_close( this->dev );
+  bladerf_close( _dev );
 }
 
-void bladerf_source_c::read_task_dispatch(bladerf_source_c *obj)
+void *bladerf_source_c::stream_callback( struct bladerf *dev,
+                                         struct bladerf_stream *stream,
+                                         struct bladerf_metadata *metadata,
+                                         void *samples,
+                                         size_t num_samples,
+                                         void *user_data )
 {
-  obj->read_task();
+  bladerf_source_c *obj = (bladerf_source_c *) user_data;
+
+  if ( ! obj->is_running() )
+    return NULL;
+
+  return obj->stream_task( samples, num_samples );
 }
 
-void bladerf_source_c::read_task()
+/* Convert & push samples to the sample fifo */
+void *bladerf_source_c::stream_task( void *samples, size_t num_samples )
 {
-  int16_t si, sq, *next_val;
-  ssize_t n_samples;
-  size_t n_avail, to_copy;
+  size_t i, n_avail, to_copy;
+  int16_t *sample = (int16_t *)samples;
+  void *ret;
 
-  while ( this->is_running() )
-  {
+  ret = _buffers[_buf_index];
+  _buf_index = (_buf_index + 1) % _num_buffers;
 
-    n_samples = bladerf_rx(this->dev, BLADERF_FORMAT_SC16_Q12, this->raw_sample_buf,
-                                 BLADERF_SAMPLE_BLOCK_SIZE, NULL);
+  _fifo_lock.lock();
 
-    if (n_samples < 0) {
-      std::cerr << "Failed to read samples: "
-                << bladerf_strerror(n_samples) << std::endl;
-      this->set_running(false);
-    } else {
-      if (n_samples != BLADERF_SAMPLE_BLOCK_SIZE) {
-        if (n_samples > BLADERF_SAMPLE_BLOCK_SIZE) {
-          std::cerr << "Warning: received bloated sample block of "
-                    << n_samples << " bytes!"<< std::endl;
-        } else {
-          std::cerr << "Warning: received truncated sample block of "
-                    << n_samples << " bytes!"<< std::endl;
-        }
-      } else {
+  n_avail = _fifo->capacity() - _fifo->size();
+  to_copy = (n_avail < num_samples ? n_avail : num_samples);
 
-        //std::cerr << "+" << std::flush;
+  for(i = 0; i < to_copy; i++ ) {
+    /* Mask valid bits only */
+    *(sample) &= 0xfff;
+    *(sample+1) &= 0xfff;
 
-        next_val = this->raw_sample_buf;
+    /* Sign extend the 12-bit IQ values, if needed */
+    if( (*sample) & 0x800 ) *(sample) |= 0xf000;
+    if( *(sample+1) & 0x800 ) *(sample+1) |= 0xf000;
 
-        this->sample_fifo_lock.lock();
-        n_avail = this->sample_fifo->capacity() - this->sample_fifo->size();
-        to_copy = (n_avail < (size_t)n_samples ? n_avail : (size_t)n_samples);
+    /* Push sample to the fifo */
+    _fifo->push_back( gr_complex( *sample * (1.0f/2048.0f),
+                                  *(sample+1) * (1.0f/2048.0f) ) );
 
-        for (size_t i = 0; i < to_copy; ++i) {
-          si = *next_val++ & 0xfff;
-          sq = *next_val++ & 0xfff;
+    /* offset to the next I+Q sample */
+    sample += 2;
+  }
 
-          /* Sign extend the 12-bit IQ values, if needed */
-          if( si & 0x800 ) si |= 0xf000;
-          if( sq & 0x800 ) sq |= 0xf000;
+  _fifo_lock.unlock();
 
-          gr_complex sample((float)si * (1.0f/2048.0f),
-                            (float)sq * (1.0f/2048.0f));
+  /* We have made some new samples available to the consumer in work() */
+  if (to_copy) {
+    //std::cerr << "+" << std::flush;
+    _samp_avail.notify_one();
+  }
 
-          this->sample_fifo->push_back(sample);
-        }
+  /* Indicate overrun, if neccesary */
+  if (to_copy < num_samples)
+    std::cerr << "O" << std::flush;
 
-        this->sample_fifo_lock.unlock();
+  return ret;
+}
 
-        /* We have made some new samples available to the consumer in work() */
-        if (to_copy) {
-          this->samples_available.notify_one();
-        }
+void bladerf_source_c::read_task()
+{
+  int status;
 
-        /* Indicate overrun, if neccesary */
-        if (to_copy < (size_t)n_samples) {
-          std::cerr << "O" << std::flush;
-        }
-      }
-    }
+  set_running( true );
 
-  }
+  /* Start stream and stay there until we kill the stream */
+  status = bladerf_stream(_stream, BLADERF_MODULE_RX);
+
+  if (status < 0)
+      std::cerr << "Source stream error: " << bladerf_strerror(status) << std::endl;
+
+  set_running( false );
 }
 
-/* Main work function, pull samples from the driver */
+/* Main work function, pull samples from the sample fifo */
 int bladerf_source_c::work( int noutput_items,
                             gr_vector_const_void_star &input_items,
                             gr_vector_void_star &output_items )
 {
-  int n_samples_avail;
-
-  if ( ! this->is_running() )
+  if ( ! is_running() )
     return WORK_DONE;
 
-  if( noutput_items >= 0 ) {
+  if( noutput_items > 0 ) {
     gr_complex *out = (gr_complex *)output_items[0];
-    boost::unique_lock<boost::mutex> lock(this->sample_fifo_lock);
+
+    boost::unique_lock<boost::mutex> lock(_fifo_lock);
 
     /* Wait until we have the requested number of samples */
-    n_samples_avail = this->sample_fifo->size();
+    int n_samples_avail = _fifo->size();
 
     while (n_samples_avail < noutput_items) {
-      this->samples_available.wait(lock);
-      n_samples_avail = this->sample_fifo->size();
+      _samp_avail.wait(lock);
+      n_samples_avail = _fifo->size();
     }
 
     for(int i = 0; i < noutput_items; ++i) {
-      out[i] = this->sample_fifo->at(0);
-      this->sample_fifo->pop_front();
+      out[i] = _fifo->at(0);
+      _fifo->pop_front();
     }
 
     //std::cerr << "-" << std::flush;
@@ -322,7 +338,7 @@ size_t bladerf_source_c::get_num_channels()
 
 osmosdr::meta_range_t bladerf_source_c::get_sample_rates()
 {
-  return this->sample_rates();
+  return sample_rates();
 }
 
 double bladerf_source_c::set_sample_rate( double rate )
@@ -334,7 +350,7 @@ double bladerf_source_c::set_sample_rate( double rate )
   /* Check to see if the sample rate is an integer */
   if( (uint32_t)round(rate) == (uint32_t)rate )
   {
-    ret = bladerf_set_sample_rate( this->dev, BLADERF_MODULE_RX, (uint32_t)rate, &actual );
+    ret = bladerf_set_sample_rate( _dev, BLADERF_MODULE_RX, (uint32_t)rate, &actual );
     if( ret ) {
       throw std::runtime_error( std::string(__FUNCTION__) + " " +
                                 "has failed to set integer rate, error " +
@@ -342,7 +358,7 @@ double bladerf_source_c::set_sample_rate( double rate )
     }
   } else {
     /* TODO: Fractional sample rate */
-    ret = bladerf_set_sample_rate( this->dev, BLADERF_MODULE_RX, (uint32_t)rate, &actual );
+    ret = bladerf_set_sample_rate( _dev, BLADERF_MODULE_RX, (uint32_t)rate, &actual );
     if( ret ) {
       throw std::runtime_error( std::string(__FUNCTION__) + " " +
                                 "has failed to set fractional rate, error " +
@@ -358,7 +374,7 @@ double bladerf_source_c::get_sample_rate()
   int ret;
   unsigned int rate = 0;
 
-  ret = bladerf_get_sample_rate( this->dev, BLADERF_MODULE_RX, &rate );
+  ret = bladerf_get_sample_rate( _dev, BLADERF_MODULE_RX, &rate );
   if( ret ) {
     throw std::runtime_error( std::string(__FUNCTION__) + " " +
                               "has failed to get sample rate, error " +
@@ -370,7 +386,7 @@ double bladerf_source_c::get_sample_rate()
 
 osmosdr::freq_range_t bladerf_source_c::get_freq_range( size_t chan )
 {
-  return this->freq_range();
+  return freq_range();
 }
 
 double bladerf_source_c::set_center_freq( double freq, size_t chan )
@@ -382,7 +398,7 @@ double bladerf_source_c::set_center_freq( double freq, size_t chan )
       freq > get_freq_range( chan ).stop() ) {
     std::cerr << "Failed to set out of bound frequency: " << freq << std::endl;
   } else {
-    ret = bladerf_set_frequency( this->dev, BLADERF_MODULE_RX, (uint32_t)freq );
+    ret = bladerf_set_frequency( _dev, BLADERF_MODULE_RX, (uint32_t)freq );
     if( ret ) {
       throw std::runtime_error( std::string(__FUNCTION__) + " " +
                                 "failed to set center frequency " +
@@ -400,7 +416,7 @@ double bladerf_source_c::get_center_freq( size_t chan )
   uint32_t freq;
   int ret;
 
-  ret = bladerf_get_frequency( this->dev, BLADERF_MODULE_RX, &freq );
+  ret = bladerf_get_frequency( _dev, BLADERF_MODULE_RX, &freq );
   if( ret ) {
     throw std::runtime_error( std::string(__FUNCTION__) + " " +
                               "failed to get center frequency, error " +
@@ -443,11 +459,11 @@ osmosdr::gain_range_t bladerf_source_c::get_gain_range( const std::string & name
   osmosdr::gain_range_t range;
 
   if( name == "LNA" ) {
-    range = this->lna_range;
+    range = _lna_range;
   } else if( name == "VGA1" ) {
-    range = this->vga1_range;
+    range = _vga1_range;
   } else if( name == "VGA2" ) {
-    range = this->vga2_range;
+    range = _vga2_range;
   } else {
     throw std::runtime_error( std::string(__FUNCTION__) + " " +
                               "requested an invalid gain element " + name );
@@ -491,11 +507,11 @@ double bladerf_source_c::set_gain( double gain, const std::string & name, size_t
                 << "setting to LNA_MAX (6dB)" << std::endl;
       g = BLADERF_LNA_GAIN_MAX;
     }
-    ret = bladerf_set_lna_gain( this->dev, g );
+    ret = bladerf_set_lna_gain( _dev, g );
   } else if( name == "VGA1" ) {
-    ret = bladerf_set_rxvga1( this->dev, (int)gain );
+    ret = bladerf_set_rxvga1( _dev, (int)gain );
   } else if( name == "VGA2" ) {
-    ret = bladerf_set_rxvga2( this->dev, (int)gain );
+    ret = bladerf_set_rxvga2( _dev, (int)gain );
   } else {
     throw std::runtime_error( std::string(__FUNCTION__) + " " +
                               "requested to set the gain "
@@ -525,12 +541,12 @@ double bladerf_source_c::get_gain( const std::string & name, size_t chan )
 
   if( name == "LNA" ) {
     bladerf_lna_gain lna_g;
-    ret = bladerf_get_lna_gain( this->dev, &lna_g );
+    ret = bladerf_get_lna_gain( _dev, &lna_g );
     g = lna_g == BLADERF_LNA_GAIN_BYPASS ? 0 : lna_g == BLADERF_LNA_GAIN_MID ? 3 : 6;
   } else if( name == "VGA1" ) {
-    ret = bladerf_get_rxvga1( this->dev, &g );
+    ret = bladerf_get_rxvga1( _dev, &g );
   } else if( name == "VGA2" ) {
-    ret = bladerf_get_rxvga2( this->dev, &g );
+    ret = bladerf_get_rxvga2( _dev, &g );
   } else {
     throw std::runtime_error( std::string(__FUNCTION__) + " " +
                               "requested to get the gain "
@@ -583,14 +599,14 @@ double bladerf_source_c::set_bandwidth( double bandwidth, size_t chan )
   int ret;
   uint32_t actual;
 
-  ret = bladerf_set_bandwidth( this->dev, BLADERF_MODULE_RX, (uint32_t)bandwidth, &actual );
+  ret = bladerf_set_bandwidth( _dev, BLADERF_MODULE_RX, (uint32_t)bandwidth, &actual );
   if( ret ) {
     throw std::runtime_error( std::string(__FUNCTION__) + " " +
                               "could not set bandwidth, error " +
                               boost::lexical_cast<std::string>(ret) );
   }
 
-  return this->get_bandwidth();
+  return get_bandwidth();
 }
 
 double bladerf_source_c::get_bandwidth( size_t chan )
@@ -598,7 +614,7 @@ double bladerf_source_c::get_bandwidth( size_t chan )
   uint32_t bandwidth;
   int ret;
 
-  ret = bladerf_get_bandwidth( this->dev, BLADERF_MODULE_RX, &bandwidth );
+  ret = bladerf_get_bandwidth( _dev, BLADERF_MODULE_RX, &bandwidth );
   if( ret ) {
     throw std::runtime_error( std::string(__FUNCTION__) + " " +
                               "could not get bandwidth, error " +
@@ -610,5 +626,5 @@ double bladerf_source_c::get_bandwidth( size_t chan )
 
 osmosdr::freq_range_t bladerf_source_c::get_bandwidth_range( size_t chan )
 {
-  return this->filter_bandwidths();
+  return filter_bandwidths();
 }
diff --git a/lib/bladerf/bladerf_source_c.h b/lib/bladerf/bladerf_source_c.h
index 8074ccf..e2a43ee 100644
--- a/lib/bladerf/bladerf_source_c.h
+++ b/lib/bladerf/bladerf_source_c.h
@@ -25,8 +25,6 @@
 #include <gnuradio/block.h>
 #include <gnuradio/sync_block.h>
 
-#include <libbladeRF.h>
-
 #include "osmosdr/ranges.h"
 #include "source_iface.h"
 #include "bladerf_common.h"
@@ -108,14 +106,20 @@ public:
   double get_bandwidth( size_t chan = 0 );
   osmosdr::freq_range_t get_bandwidth_range( size_t chan = 0 );
 
-private:
-  static void read_task_dispatch(bladerf_source_c *obj);
+private: /* functions */
+  static void *stream_callback( struct bladerf *_dev,
+                                struct bladerf_stream *stream,
+                                struct bladerf_metadata *metadata,
+                                void *samples,
+                                size_t num_samples,
+                                void *user_data );
+
+  void *stream_task(void *samples, size_t num_samples);
+
   void read_task();
 
-  gr::thread::thread thread;
-  osmosdr::gain_range_t lna_range;
-  osmosdr::gain_range_t vga1_range;
-  osmosdr::gain_range_t vga2_range;
+private: /* members */
+  osmosdr::gain_range_t _lna_range;
 };
 
 #endif /* INCLUDED_BLADERF_SOURCE_C_H */