Solved performance issue.

src/algorithms/signal_source/adapters/rtl_tcp_signal_source.cc

@@ -81,7 +81,7 @@ RtlTcpSignalSource::RtlTcpSignalSource(ConfigurationInterface* configuration,
             {
 	       std::cout << "Connecting to " << address_ << ":" << port_ << std::endl;
 	       LOG (INFO) << "Connecting to " << address_ << ":" << port_;
-	       signal_source_ = rtl_tcp_make_signal_source_cc (address_, port_);
+	       signal_source_ = rtl_tcp_make_signal_source_c (address_, port_);
             }
             catch( boost::exception & e )
             {
@@ -128,9 +128,6 @@ RtlTcpSignalSource::RtlTcpSignalSource(ConfigurationInterface* configuration,
             file_sink_ = gr::blocks::file_sink::make(item_size_, dump_filename_.c_str());
             DLOG(INFO) << "file_sink(" << file_sink_->unique_id() << ")";
         }
-
-    deinterleave_ = gr::blocks::deinterleave::make (sizeof (float));
-    ftoc_ = gr::blocks::float_to_complex::make ( 1 );
 }
 
 
@@ -139,13 +136,8 @@ RtlTcpSignalSource::~RtlTcpSignalSource()
 
 
 void RtlTcpSignalSource::connect(gr::top_block_sptr top_block) {
-   // deinterleave and convert to complex
-   top_block->connect (signal_source_, 0, deinterleave_, 0);
-   top_block->connect (deinterleave_, 0, ftoc_, 0);
-   top_block->connect (deinterleave_, 1, ftoc_, 1);
-
    if ( samples_ ) {
-      top_block->connect (ftoc_, 0, valve_, 0);
+      top_block->connect (signal_source_, 0, valve_, 0);
       DLOG(INFO) << "connected rtl tcp source to valve";
       if ( dump_ ) {
 	 top_block->connect(valve_, 0, file_sink_, 0);
@@ -153,23 +145,20 @@ void RtlTcpSignalSource::connect(gr::top_block_sptr top_block) {
       }
    }
    else if ( dump_ ) {
-	 top_block->connect(ftoc_, 0, file_sink_, 0);
+	 top_block->connect(signal_source_, 0, file_sink_, 0);
 	 DLOG(INFO) << "connected rtl tcp source to file sink";
    }
 }
 
 void RtlTcpSignalSource::disconnect(gr::top_block_sptr top_block) {
-   top_block->disconnect (signal_source_, 0, deinterleave_, 0);
-   top_block->disconnect (deinterleave_, 0, ftoc_, 0);
-   top_block->disconnect (deinterleave_, 1, ftoc_, 1);
    if ( samples_ ) {
-      top_block->disconnect (ftoc_, 0, valve_, 0);
+      top_block->disconnect (signal_source_, 0, valve_, 0);
       if ( dump_ ) {
 	 top_block->disconnect(valve_, 0, file_sink_, 0);
       }
    }
    else if ( dump_ ) {
-	 top_block->disconnect(ftoc_, 0, file_sink_, 0);
+	 top_block->disconnect(signal_source_, 0, file_sink_, 0);
    }
 }
 
@@ -183,6 +172,6 @@ gr::basic_block_sptr RtlTcpSignalSource::get_right_block() {
       return valve_;
    }
    else {
-      return ftoc_;
+     return signal_source_;
    }
 }

src/algorithms/signal_source/adapters/rtl_tcp_signal_source.h

@@ -38,7 +38,7 @@
 #include <gnuradio/blocks/file_sink.h>
 #include <gnuradio/blocks/deinterleave.h>
 #include <gnuradio/blocks/float_to_complex.h>
-#include "rtl_tcp_signal_source_cc.h"
+#include "rtl_tcp_signal_source_c.h"
 #include "gnss_block_interface.h"
 
 class ConfigurationInterface;
@@ -102,9 +102,7 @@ private:
     bool dump_;
     std::string dump_filename_;
 
-    rtl_tcp_signal_source_cc_sptr signal_source_;
+    rtl_tcp_signal_source_c_sptr signal_source_;
-    gr::blocks::deinterleave::sptr deinterleave_;
-    gr::blocks::float_to_complex::sptr ftoc_;
 
     boost::shared_ptr<gr::block> valve_;
     gr::blocks::file_sink::sptr file_sink_;

src/algorithms/signal_source/gnuradio_blocks/CMakeLists.txt

@@ -20,7 +20,7 @@
 set(SIGNAL_SOURCE_GR_BLOCKS_SOURCES
      unpack_byte_2bit_samples.cc
      unpack_intspir_1bit_samples.cc
-     rtl_tcp_signal_source_cc.cc
+     rtl_tcp_signal_source_c.cc
 )
 
 include_directories(

src/algorithms/signal_source/gnuradio_blocks/rtl_tcp_signal_source_c.cc

@@ -1,5 +1,5 @@
 /*!
- * \file rtl_tcp_signal_source_cc.cc
+ * \file rtl_tcp_signal_source_c.cc
  * \brief An rtl_tcp signal source reader.
  * \author Anthony Arnold, 2015. anthony.arnold(at)uqconnect.edu.au
  *
@@ -28,15 +28,21 @@
  * -------------------------------------------------------------------------
  */
 
-#include "rtl_tcp_signal_source_cc.h"
+#include "rtl_tcp_signal_source_c.h"
 #include <glog/logging.h>
-#include <boost/thread.hpp>
+#include <boost/thread/thread.hpp>
 
 using google::LogMessage;
 
 namespace ip = boost::asio::ip;
 using boost::asio::ip::tcp;
 
+// Buffer constants
+enum {
+  RTL_TCP_BUFFER_SIZE = 1024 * 16, // 16 KB
+  RTL_TCP_PAYLOAD_SIZE = 1024 * 4  //  4 KB
+};
+
 // command ids
 enum {
    CMD_ID_SET_FREQUENCY = 1,
@@ -81,21 +87,22 @@ struct set_sample_rate_command : command {
    }
 };
 
-rtl_tcp_signal_source_cc_sptr
+rtl_tcp_signal_source_c_sptr
-rtl_tcp_make_signal_source_cc(const std::string &address,
+rtl_tcp_make_signal_source_c(const std::string &address,
 			      short port)
 {
-   return gnuradio::get_initial_sptr (new rtl_tcp_signal_source_cc (address, port));
+   return gnuradio::get_initial_sptr (new rtl_tcp_signal_source_c (address, port));
 }
 
 
-rtl_tcp_signal_source_cc::rtl_tcp_signal_source_cc(const std::string &address,
+rtl_tcp_signal_source_c::rtl_tcp_signal_source_c(const std::string &address,
 						   short port)
-   : gr::sync_block ("rtl_tcp_signal_source_cc",
+   : gr::sync_block ("rtl_tcp_signal_source_c",
                    gr::io_signature::make(0, 0, 0),
-                   gr::io_signature::make(1, 1, sizeof(float))),
+                   gr::io_signature::make(1, 1, sizeof(gr_complex))),
      socket_ (io_service_),
-     buffer_ (1048576),
+     data_ (RTL_TCP_PAYLOAD_SIZE),
+     buffer_ (RTL_TCP_BUFFER_SIZE),
      unread_ (0)
 {
    boost::system::error_code ec;
@@ -123,31 +130,32 @@ rtl_tcp_signal_source_cc::rtl_tcp_signal_source_cc(const std::string &address,
    LOG (WARNING)  << "Connected to " << addr << ":" << port;
 
    boost::asio::async_read (socket_, boost::asio::buffer (data_),
-			    boost::bind (&rtl_tcp_signal_source_cc::handle_read,
+			    boost::bind (&rtl_tcp_signal_source_c::handle_read,
 					 this, _1, _2));
    boost::thread (boost::bind (&boost::asio::io_service::run, &io_service_));
-   //io_service_.poll ();
 }
 
-rtl_tcp_signal_source_cc::~rtl_tcp_signal_source_cc()
+rtl_tcp_signal_source_c::~rtl_tcp_signal_source_c()
 {
    io_service_.stop ();
 }
 
-int rtl_tcp_signal_source_cc::work (int noutput_items,
+int rtl_tcp_signal_source_c::work (int noutput_items,
-				    gr_vector_const_void_star &input_items,
+                                   gr_vector_const_void_star &/*input_items*/,
 				    gr_vector_void_star &output_items)
 {
-   float *out = reinterpret_cast <float *>( output_items[0] );
+  gr_complex *out = reinterpret_cast <gr_complex *>( output_items[0] );
   int i = 0;
 
   {
     boost::mutex::scoped_lock lock (mutex_);
-      not_empty_.wait (lock, boost::bind (&rtl_tcp_signal_source_cc::not_empty,
+    not_empty_.wait (lock, boost::bind (&rtl_tcp_signal_source_c::not_empty,
                                         this));
 
-      for ( ; i < noutput_items && unread_ > 0; i++ ) {
+    for ( ; i < noutput_items && unread_ > 1; i++ ) {
-	 out[i] = buffer_[--unread_];
+      float re = buffer_[--unread_];
+      float im = buffer_[--unread_];
+      out[i] = gr_complex (re, im);
     }
   }
   not_full_.notify_one ();
@@ -155,7 +163,7 @@ int rtl_tcp_signal_source_cc::work (int noutput_items,
 }
 
 
-void rtl_tcp_signal_source_cc::set_frequency (int frequency) {
+void rtl_tcp_signal_source_c::set_frequency (int frequency) {
    boost::system::error_code ec =
       set_frequency_command (frequency).send(socket_);
    if (ec) {
@@ -164,7 +172,7 @@ void rtl_tcp_signal_source_cc::set_frequency (int frequency) {
    }
 }
 
-void rtl_tcp_signal_source_cc::set_sample_rate (int sample_rate) {
+void rtl_tcp_signal_source_c::set_sample_rate (int sample_rate) {
    boost::system::error_code ec =
       set_sample_rate_command (sample_rate).send(socket_);
    if (ec) {
@@ -173,35 +181,45 @@ void rtl_tcp_signal_source_cc::set_sample_rate (int sample_rate) {
    }
 }
 
-void rtl_tcp_signal_source_cc::handle_read  (const boost::system::error_code &ec,
+void
+rtl_tcp_signal_source_c::handle_read (const boost::system::error_code &ec,
                                        size_t bytes_transferred)
 {
    if (ec) {
       std::cout << "Error during read: " << ec << std::endl;
       LOG (WARNING) << "Error during read: " << ec;
+      boost::mutex::scoped_lock lock (mutex_);
+      buffer_.clear ();
       not_empty_.notify_one ();
    }
    else {
      {
+        // Unpack read data
         boost::mutex::scoped_lock lock (mutex_);
-	 not_full_.wait (lock, boost::bind (&rtl_tcp_signal_source_cc::not_full,
+        not_full_.wait (lock,
+                        boost::bind (&rtl_tcp_signal_source_c::not_full,
                                      this));
 
         for (size_t i = 0; i < bytes_transferred; i++) {
           while (!not_full( )) {
+            // uh-oh, buffer overflow
+            // wait until there's space for more
             not_empty_.notify_one ();
-	       not_full_.wait (lock, boost::bind (&rtl_tcp_signal_source_cc::not_full,
+            not_full_.wait (lock,
+                            boost::bind (&rtl_tcp_signal_source_c::not_full,
                                          this));
           }
 
-	    buffer_.push_front (lookup_ [data_[i]]);
+          buffer_.push_front (lookup_[data_[i]]);
           unread_++;
         }
       }
+      // let woker know that more data is available 
       not_empty_.notify_one ();
-
+      // Read some more
-      boost::asio::async_read (socket_, boost::asio::buffer (data_),
+      boost::asio::async_read (socket_,
-			       boost::bind (&rtl_tcp_signal_source_cc::handle_read,
+                               boost::asio::buffer (data_),
+			       boost::bind (&rtl_tcp_signal_source_c::handle_read,
 					    this, _1, _2));
    }
 }

src/algorithms/signal_source/gnuradio_blocks/rtl_tcp_signal_source_c.h

@@ -1,8 +1,15 @@
 /*!
- * \file rtl_tcp_signal_source_cc.h
+ * \file rtl_tcp_signal_source_c.h
  * \brief Interface of an rtl_tcp signal source reader.
  * \author Anthony Arnold, 2015. anthony.arnold(at)uqconnect.edu.au
  *
+ * The implementation of this block is a combination of various helpful
+ * sources. The data format and command structure is taken from the
+ * original Osmocom rtl_tcp_source_f (http://git.osmocom.org/gr-osmosdr).
+ * The aynchronous reading code comes from the examples provides
+ * by Boost.Asio and the bounded buffer producer-consumer solution is
+ * taken from the Boost.CircularBuffer examples (http://boost.org/).
+ *
  * -------------------------------------------------------------------------
  *
  * Copyright (C) 2010-2015  (see AUTHORS file for a list of contributors)
@@ -28,8 +35,8 @@
  * -------------------------------------------------------------------------
  */
 
-#ifndef GNSS_SDR_RTL_TCP_SIGNAL_SOURCE_CC_H
+#ifndef GNSS_SDR_RTL_TCP_SIGNAL_SOURCE_C_H
-#define	GNSS_SDR_RTL_TCP_SIGNAL_SOURCE_CC_H
+#define	GNSS_SDR_RTL_TCP_SIGNAL_SOURCE_C_H
 
 #include <boost/asio.hpp>
 #include <gnuradio/sync_block.h>
@@ -39,23 +46,23 @@
 #include <boost/array.hpp>
 #include <boost/circular_buffer.hpp>
 
-class rtl_tcp_signal_source_cc;
+class rtl_tcp_signal_source_c;
 
-typedef boost::shared_ptr<rtl_tcp_signal_source_cc>
+typedef boost::shared_ptr<rtl_tcp_signal_source_c>
-        rtl_tcp_signal_source_cc_sptr;
+        rtl_tcp_signal_source_c_sptr;
 
-rtl_tcp_signal_source_cc_sptr
+rtl_tcp_signal_source_c_sptr
-rtl_tcp_make_signal_source_cc(const std::string &address,
+rtl_tcp_make_signal_source_c(const std::string &address,
 			      short port);
 
 /*!
  * \brief This class reads interleaved I/Q samples
- * from an rtl_tcp server.
+ * from an rtl_tcp server and outputs complex types.
  */
-class rtl_tcp_signal_source_cc : public gr::sync_block
+class rtl_tcp_signal_source_c : public gr::sync_block
 {
 public:
-    ~rtl_tcp_signal_source_cc();
+    ~rtl_tcp_signal_source_c();
 
     int work (int noutput_items,
 	      gr_vector_const_void_star &input_items,
@@ -65,25 +72,31 @@ public:
     void set_sample_rate (int sample_rate);
 
 private:
-    friend rtl_tcp_signal_source_cc_sptr
+    typedef boost::circular_buffer_space_optimized<float> buffer_type;
-       rtl_tcp_make_signal_source_cc(const std::string &address,
+    
+    friend rtl_tcp_signal_source_c_sptr
+       rtl_tcp_make_signal_source_c(const std::string &address,
 				     short port);
 
-    rtl_tcp_signal_source_cc(const std::string &address,
+    rtl_tcp_signal_source_c(const std::string &address,
 			     short port);
 
+    // IO members
     boost::asio::io_service io_service_;
     boost::asio::ip::tcp::socket socket_;
+    std::vector<unsigned char> data_;
     
+    // producer-consumer helpers
     boost::mutex mutex_;
     boost::condition not_full_;
     boost::condition not_empty_;
-    boost::circular_buffer<float> buffer_;
+    buffer_type buffer_;
     size_t unread_;
 
+    // lookup for scaling bytes
     boost::array<float, 256> lookup_;
-    boost::array<unsigned char, 256> data_;
 
+    // async read callback
     void handle_read (const boost::system::error_code &ec,
 		      size_t bytes_transferred);
     
@@ -97,4 +110,4 @@ private:
 
 };
 
-#endif //GNSS_SDR_RTL_TCP_SIGNAL_SOURCE_CC_H
+#endif //GNSS_SDR_RTL_TCP_SIGNAL_SOURCE_C_H