Added 'blocking' option to pcps_acquisition_cc

Using C++ 11 threading routines to create a worker thread that can
perform the FFT for acquisition in a non-blocking manner. By default
'blocking' is set to true, which is identical to the previous behaviour
(with the added overhead of an extra thread). When 'blocking' is set to
false, then the main thread does not wait for the worker, thereby
letting the receiver continue even when the FFT thread is working. This
is very useful for using FFT acquisition at higher sampling rates, where
the FFT cannot be performed in real-time.

The aim is to replace the separate pcps_multithread_acquisition_cc.

TODO: implement the same thing for the pcps_acquisition_sc

src/algorithms/acquisition/adapters/galileo_e1_pcps_ambiguous_acquisition.cc

@@ -56,6 +56,7 @@ GalileoE1PcpsAmbiguousAcquisition::GalileoE1PcpsAmbiguousAcquisition(
     fs_in_ = configuration_->property("GNSS-SDR.internal_fs_sps", fs_in_deprecated);
     if_ = configuration_->property(role + ".if", 0);
     dump_ = configuration_->property(role + ".dump", false);
+    blocking_ = configuration_->property(role + ".blocking", true);
     doppler_max_ = configuration_->property(role + ".doppler_max", 5000);
     sampled_ms_ = configuration_->property(role + ".coherent_integration_time_ms", 4);
 
@@ -91,7 +92,8 @@ GalileoE1PcpsAmbiguousAcquisition::GalileoE1PcpsAmbiguousAcquisition(
             item_size_ = sizeof(lv_16sc_t);
             acquisition_sc_ = pcps_make_acquisition_sc(sampled_ms_, max_dwells_,
                     doppler_max_, if_, fs_in_, samples_per_ms, code_length_,
-                    bit_transition_flag_, use_CFAR_algorithm_flag_, dump_, dump_filename_);
+                    bit_transition_flag_, use_CFAR_algorithm_flag_, dump_,
+                    dump_filename_);
             DLOG(INFO) << "acquisition(" << acquisition_sc_->unique_id() << ")";
 
         }
@@ -100,7 +102,8 @@ GalileoE1PcpsAmbiguousAcquisition::GalileoE1PcpsAmbiguousAcquisition(
             item_size_ = sizeof(gr_complex);
             acquisition_cc_ = pcps_make_acquisition_cc(sampled_ms_, max_dwells_,
                     doppler_max_, if_, fs_in_, samples_per_ms, code_length_,
-                    bit_transition_flag_, use_CFAR_algorithm_flag_, dump_, dump_filename_);
+                    bit_transition_flag_, use_CFAR_algorithm_flag_, dump_, blocking_,
+                    dump_filename_);
             DLOG(INFO) << "acquisition(" << acquisition_cc_->unique_id() << ")";
         }
 

src/algorithms/acquisition/adapters/galileo_e1_pcps_ambiguous_acquisition.h

@@ -155,6 +155,7 @@ private:
     long fs_in_;
     long if_;
     bool dump_;
+    bool blocking_;
     std::string dump_filename_;
     std::complex<float> * code_;
     Gnss_Synchro * gnss_synchro_;

src/algorithms/acquisition/adapters/gps_l1_ca_pcps_acquisition.cc

@@ -59,6 +59,7 @@ GpsL1CaPcpsAcquisition::GpsL1CaPcpsAcquisition(
     fs_in_ = configuration_->property("GNSS-SDR.internal_fs_sps", fs_in_deprecated);
     if_ = configuration_->property(role + ".if", 0);
     dump_ = configuration_->property(role + ".dump", false);
+    blocking_ = configuration_->property(role + ".blocking", true);
     doppler_max_ = configuration_->property(role + ".doppler_max", 5000);
     sampled_ms_ = configuration_->property(role + ".coherent_integration_time_ms", 1);
 
@@ -94,7 +95,7 @@ GpsL1CaPcpsAcquisition::GpsL1CaPcpsAcquisition(
             item_size_ = sizeof(gr_complex);
             acquisition_cc_ = pcps_make_acquisition_cc(sampled_ms_, max_dwells_,
                     doppler_max_, if_, fs_in_, code_length_, code_length_,
-                    bit_transition_flag_, use_CFAR_algorithm_flag_, dump_, dump_filename_);
+                    bit_transition_flag_, use_CFAR_algorithm_flag_, dump_, blocking_, dump_filename_);
             DLOG(INFO) << "acquisition(" << acquisition_cc_->unique_id() << ")";
         }
 

src/algorithms/acquisition/adapters/gps_l1_ca_pcps_acquisition.h

@@ -160,6 +160,7 @@ private:
     long fs_in_;
     long if_;
     bool dump_;
+    bool blocking_;
     std::string dump_filename_;
     std::complex<float> * code_;
     Gnss_Synchro * gnss_synchro_;

src/algorithms/acquisition/adapters/gps_l2_m_pcps_acquisition.cc

@@ -59,6 +59,7 @@ GpsL2MPcpsAcquisition::GpsL2MPcpsAcquisition(
     fs_in_ = configuration_->property("GNSS-SDR.internal_fs_sps", fs_in_deprecated);
     if_ = configuration_->property(role + ".if", 0);
     dump_ = configuration_->property(role + ".dump", false);
+    blocking_ = configuration_->property(role + ".blocking", true);
     doppler_max_ = configuration->property(role + ".doppler_max", 5000);
 
     bit_transition_flag_ = configuration_->property(role + ".bit_transition_flag", false);
@@ -95,7 +96,8 @@ GpsL2MPcpsAcquisition::GpsL2MPcpsAcquisition(
             item_size_ = sizeof(gr_complex);
             acquisition_cc_ = pcps_make_acquisition_cc(1, max_dwells_,
                     doppler_max_, if_, fs_in_, code_length_, code_length_,
-                    bit_transition_flag_, use_CFAR_algorithm_flag_, dump_, dump_filename_);
+                    bit_transition_flag_, use_CFAR_algorithm_flag_, dump_, blocking_,
+                    dump_filename_);
             DLOG(INFO) << "acquisition(" << acquisition_cc_->unique_id() << ")";
         }
 

src/algorithms/acquisition/adapters/gps_l2_m_pcps_acquisition.h

@@ -157,6 +157,7 @@ private:
     long fs_in_;
     long if_;
     bool dump_;
+    bool blocking_;
     std::string dump_filename_;
     std::complex<float> * code_;
     Gnss_Synchro * gnss_synchro_;

src/algorithms/acquisition/gnuradio_blocks/pcps_acquisition_cc.cc

@@ -50,12 +50,12 @@ pcps_acquisition_cc_sptr pcps_make_acquisition_cc(
                                  unsigned int doppler_max, long freq, long fs_in,
                                  int samples_per_ms, int samples_per_code,
                                  bool bit_transition_flag, bool use_CFAR_algorithm_flag,
-                                 bool dump,
+                                 bool dump, bool blocking,
                                  std::string dump_filename)
 {
     return pcps_acquisition_cc_sptr(
             new pcps_acquisition_cc(sampled_ms, max_dwells, doppler_max, freq, fs_in, samples_per_ms,
-                    samples_per_code, bit_transition_flag, use_CFAR_algorithm_flag, dump, dump_filename));
+                    samples_per_code, bit_transition_flag, use_CFAR_algorithm_flag, dump, blocking, dump_filename));
 }
 
 
@@ -64,7 +64,7 @@ pcps_acquisition_cc::pcps_acquisition_cc(
                          unsigned int doppler_max, long freq, long fs_in,
                          int samples_per_ms, int samples_per_code,
                          bool bit_transition_flag, bool use_CFAR_algorithm_flag,
-                         bool dump,
+                         bool dump, bool blocking,
                          std::string dump_filename) :
     gr::block("pcps_acquisition_cc",
     gr::io_signature::make(1, 1, sizeof(gr_complex) * sampled_ms * samples_per_ms * ( bit_transition_flag ? 2 : 1 )),
@@ -129,6 +129,16 @@ pcps_acquisition_cc::pcps_acquisition_cc(
 
     d_gnss_synchro = 0;
     d_grid_doppler_wipeoffs = 0;
+
+    d_done = false;
+    d_blocking = blocking;
+    d_new_data_available = false;
+    d_worker_active = false;
+    d_data_buffer = static_cast<gr_complex*>(volk_gnsssdr_malloc(d_fft_size * sizeof(gr_complex), volk_gnsssdr_get_alignment()));
+
+    // Start the worker thread and wait for it to acknowledge:
+    std::thread t1( &pcps_acquisition_cc::acquisition_core, this );
+    std::swap( d_worker_thread, t1 );
 }
 
 
@@ -153,6 +163,16 @@ pcps_acquisition_cc::~pcps_acquisition_cc()
         {
             d_dump_file.close();
         }
+
+    // Let the worker thread know that we are done and then wait to join
+    {
+        std::lock_guard<std::mutex> lk( d_mutex );
+        d_done = true;
+        d_cond.notify_one();
+    }
+    
+    d_worker_thread.join();
+    volk_gnsssdr_free( d_data_buffer );
 }
 
 
@@ -212,6 +232,10 @@ void pcps_acquisition_cc::init()
             int doppler = -static_cast<int>(d_doppler_max) + d_doppler_step * doppler_index;
             update_local_carrier(d_grid_doppler_wipeoffs[doppler_index], d_fft_size, d_freq + doppler);
         }
+
+    d_new_data_available = false;
+    d_done = false;
+    d_worker_active = false;
 }
 
 
@@ -316,11 +340,98 @@ int pcps_acquisition_cc::general_work(int noutput_items,
 
     case 1:
         {
+            std::unique_lock<std::mutex> lk( d_mutex );
+
+            if( d_worker_active && !d_blocking )
+            {
+                d_sample_counter += d_fft_size * ninput_items[0];
+                consume_each( ninput_items[0] );
+            }
+            else
+            {
+                if( d_worker_active && d_blocking )
+                {
+                    d_cond.wait( lk, [&]{ return !this->d_worker_active; } );
+                }
+
+                d_sample_counter += d_fft_size;
+
+
+                if( d_well_count > 0 )
+                {
+
+                    if (!d_bit_transition_flag)
+                    {
+                        if (d_test_statistics > d_threshold)
+                        {
+                            d_state = 0; // Positive acquisition
+                            d_active = false;
+                            send_positive_acquisition();
+                        }
+                        else if (d_well_count == d_max_dwells)
+                        {
+                            d_state = 0;
+                            d_active = false;
+                            send_negative_acquisition();
+                        }
+                    }
+                    else
+                    {
+                        if (d_well_count == d_max_dwells) // d_max_dwells = 2
+                        {
+                            if (d_test_statistics > d_threshold)
+                            {
+                                d_state = 0; // Positive acquisition
+                                d_active = false;
+                                send_positive_acquisition();
+                            }
+                            else
+                            {
+                                d_state = 0; // Negative acquisition
+                                d_active = false;
+                                send_negative_acquisition();
+                            }
+                        }
+                    }
+                }
+
+                if( d_active )
+                {
+                    memcpy( d_data_buffer, input_items[0], d_fft_size * sizeof( gr_complex ) );
+                    d_new_data_available = true;
+                    d_cond.notify_one();
+                }
+
+                consume_each(1);
+            } // if worker_active (else)
+
+        } // case 1, switch d_state
+
+    } // switch d_state
+
+    return noutput_items;
+}
+
+void pcps_acquisition_cc::acquisition_core( void )
+{
+    d_worker_active = false;
+    while( 1 )
+    {
+        std::unique_lock<std::mutex> lk( d_mutex );
+        d_cond.wait( lk, [&]{ return this->d_new_data_available or this->d_done; } );
+        d_worker_active = !d_done;
+        lk.unlock();
+
+        if( d_done )
+        {
+            break;
+        }
+
         // initialize acquisition algorithm
         int doppler;
         uint32_t indext = 0;
         float magt = 0.0;
-            const gr_complex *in = reinterpret_cast<const gr_complex *>(input_items[0]);
+        const gr_complex *in = d_data_buffer; //Get the input samples pointer
 
         int effective_fft_size = ( d_bit_transition_flag ? d_fft_size/2 : d_fft_size );
 
@@ -328,14 +439,16 @@ int pcps_acquisition_cc::general_work(int noutput_items,
 
         d_input_power = 0.0;
         d_mag = 0.0;
-            d_sample_counter += d_fft_size; // sample counter
+        unsigned long int sample_counter = d_sample_counter; // sample counter
         d_well_count++;
 
         DLOG(INFO)<< "Channel: " << d_channel
             << " , doing acquisition of satellite: " << d_gnss_synchro->System << " " << d_gnss_synchro->PRN
-                    << " ,sample stamp: " << d_sample_counter << ", threshold: "
+            << " ,sample stamp: " << sample_counter << ", threshold: "
             << d_threshold << ", doppler_max: " << d_doppler_max
-                    << ", doppler_step: " << d_doppler_step<<std::endl;
+            << ", doppler_step: " << d_doppler_step
+            << ", use_CFAR_algorithm_flag: " << ( d_use_CFAR_algorithm_flag ? "true" : "false" )
+            <<std::endl;
 
         if (d_use_CFAR_algorithm_flag == true)
         {
@@ -400,7 +513,7 @@ int pcps_acquisition_cc::general_work(int noutput_items,
                 {
                     d_gnss_synchro->Acq_delay_samples = static_cast<double>(indext % d_samples_per_code);
                     d_gnss_synchro->Acq_doppler_hz = static_cast<double>(doppler);
-                                    d_gnss_synchro->Acq_samplestamp_samples = d_sample_counter;
+                    d_gnss_synchro->Acq_samplestamp_samples = sample_counter;
 
                     // 5- Compute the test statistics and compare to the threshold
                     //d_test_statistics = 2 * d_fft_size * d_mag / d_input_power;
@@ -428,64 +541,16 @@ int pcps_acquisition_cc::general_work(int noutput_items,
                 DLOG(INFO) << "Writing ACQ out to " << filename.str();
 
                 d_dump_file.open(filename.str().c_str(), std::ios::out | std::ios::binary);
-                            d_dump_file.write(reinterpret_cast<char*>(d_ifft->get_outbuf()), n); //write directly |abs(x)|^2 in this Doppler bin?
+                d_dump_file.write((char*)d_ifft->get_outbuf(), n); //write directly |abs(x)|^2 in this Doppler bin?
                 d_dump_file.close();
             }
         }
 
-            if (!d_bit_transition_flag)
+        lk.lock();
-                {
+        d_worker_active = false;
-                    if (d_test_statistics > d_threshold)
+        d_new_data_available = false;
-                        {
+        lk.unlock();
-                            d_state = 0; // Positive acquisition
+        d_cond.notify_one();
-                            d_active = false;
-                            send_positive_acquisition();
-                        }
-                    else if (d_well_count == d_max_dwells)
-                        {
-                            d_state = 0;
-                            d_active = false;
-                            send_negative_acquisition();
-                        }
-                }
-            else
-                {
-                    if (d_well_count == d_max_dwells) // d_max_dwells = 2
-                        {
-                            if (d_test_statistics > d_threshold)
-                                {
-                                    d_state = 0; // Positive acquisition
-                                    d_active = false;
-                                    send_positive_acquisition();
-                                }
-                            else
-                                {
-                                    d_state = 0; // Negative acquisition
-                                    d_active = false;
-                                    send_negative_acquisition();
-                                }
-                        }
-                }
 
-            consume_each(1);
-            break;
     }
-    }
-
-    return noutput_items;
 }
-
-
-//void pcps_acquisition_cc::forecast (int noutput_items, gr_vector_int &ninput_items_required)
-//{
-    //// COD:
-    //// For zero-padded case we need one extra code period
-    //if( d_bit_transition_flag )
-    //{
-        //ninput_items_required[0] = noutput_items*(d_samples_per_code * d_max_dwells + d_samples_per_code);
-    //}
-    //else
-    //{
-        //ninput_items_required[0] = noutput_items*d_fft_size*d_max_dwells;
-    //}
-//}

src/algorithms/acquisition/gnuradio_blocks/pcps_acquisition_cc.h

@@ -52,6 +52,9 @@
 
 #include <fstream>
 #include <string>
+#include <mutex>
+#include <thread>
+#include <condition_variable>
 #include <gnuradio/block.h>
 #include <gnuradio/gr_complex.h>
 #include <gnuradio/fft/fft.h>
@@ -66,7 +69,7 @@ pcps_make_acquisition_cc(unsigned int sampled_ms, unsigned int max_dwells,
                          unsigned int doppler_max, long freq, long fs_in,
                          int samples_per_ms, int samples_per_code,
                          bool bit_transition_flag, bool use_CFAR_algorithm_flag,
-                         bool dump,
+                         bool dump, bool blocking,
                          std::string dump_filename);
 
 /*!
@@ -83,18 +86,20 @@ private:
             unsigned int doppler_max, long freq, long fs_in,
             int samples_per_ms, int samples_per_code,
             bool bit_transition_flag, bool use_CFAR_algorithm_flag,
-            bool dump,
+            bool dump, bool blocking,
             std::string dump_filename);
 
     pcps_acquisition_cc(unsigned int sampled_ms, unsigned int max_dwells,
             unsigned int doppler_max, long freq, long fs_in,
             int samples_per_ms, int samples_per_code,
             bool bit_transition_flag, bool use_CFAR_algorithm_flag,
-            bool dump,
+            bool dump, bool blocking, 
             std::string dump_filename);
 
     void update_local_carrier(gr_complex* carrier_vector, int correlator_length_samples, float freq);
 
+    void acquisition_core( void );
+
     void send_negative_acquisition();
     void send_positive_acquisition();
     long d_fs_in;
@@ -132,6 +137,17 @@ private:
     unsigned int d_channel;
     std::string d_dump_filename;
 
+    std::thread d_worker_thread;
+    std::mutex  d_mutex;
+
+    std::condition_variable d_cond;
+    bool d_done;
+    bool d_new_data_available;
+    bool d_worker_active;
+    bool d_blocking;
+
+    gr_complex *d_data_buffer;
+
 public:
     /*!
      * \brief Default destructor.