Merge branch 'next_memless' of github.com:Opendigitalradio/ODR-DabMod into next_memless

12 files changed, 121 insertions, 147 deletions

diff --git a/src/ConfigParser.cpp b/src/ConfigParser.cpp
index 459811f..9ac1280 100644
--- a/src/ConfigParser.cpp
+++ b/src/ConfigParser.cpp
@@ -171,7 +171,10 @@ static void parse_configfile(
     // Poly coefficients:
     if (pt.get("poly.enabled", 0) == 1) {
         mod_settings.polyCoefFilename =
-            pt.get<std::string>("poly.polycoeffile", "default");
+            pt.get<std::string>("poly.polycoeffile", "dpd/poly.coef");
+
+        mod_settings.polyNumThreads =
+            pt.get<int>("poly.num_threads", 0);
     }
 
     // Output options
diff --git a/src/ConfigParser.h b/src/ConfigParser.h
index 22a4fc5..89f0fb7 100644
--- a/src/ConfigParser.h
+++ b/src/ConfigParser.h
@@ -75,7 +75,7 @@ struct mod_settings_t {
     std::string filterTapsFilename = "";
 
     std::string polyCoefFilename = "";
-
+    unsigned polyNumThreads = 0;
 
 #if defined(HAVE_OUTPUT_UHD)
     OutputUHDConfig outputuhd_conf;
diff --git a/src/DabMod.cpp b/src/DabMod.cpp
index 7c342a2..15b3be2 100644
--- a/src/DabMod.cpp
+++ b/src/DabMod.cpp
@@ -317,18 +317,7 @@ int launch_modulator(int argc, char* argv[])
         }
         Flowgraph flowgraph;
 
-        auto modulator = make_shared<DabModulator>(
-                ediReader,
-                mod_settings.tiiConfig,
-                mod_settings.outputRate,
-                mod_settings.clockRate,
-                mod_settings.dabMode,
-                mod_settings.gainMode,
-                mod_settings.digitalgain,
-                mod_settings.normalise,
-                mod_settings.gainmodeVariance,
-                mod_settings.filterTapsFilename,
-                mod_settings.polyCoefFilename);
+        auto modulator = make_shared<DabModulator>(ediReader, mod_settings);
 
         if (format_converter) {
             flowgraph.connect(modulator, format_converter);
@@ -422,18 +411,7 @@ int launch_modulator(int argc, char* argv[])
             m.etiReader = &etiReader;
 
             auto input = make_shared<InputMemory>(&m.data);
-            auto modulator = make_shared<DabModulator>(
-                    etiReader,
-                    mod_settings.tiiConfig,
-                    mod_settings.outputRate,
-                    mod_settings.clockRate,
-                    mod_settings.dabMode,
-                    mod_settings.gainMode,
-                    mod_settings.digitalgain,
-                    mod_settings.normalise,
-                    mod_settings.gainmodeVariance,
-                    mod_settings.filterTapsFilename,
-                    mod_settings.polyCoefFilename);
+            auto modulator = make_shared<DabModulator>(etiReader, mod_settings);
 
             if (format_converter) {
                 flowgraph.connect(modulator, format_converter);
diff --git a/src/DabModulator.cpp b/src/DabModulator.cpp
index 5282a2d..cc2642a 100644
--- a/src/DabModulator.cpp
+++ b/src/DabModulator.cpp
@@ -55,46 +55,22 @@
 #include "RemoteControl.h"
 #include "Log.h"
 
-DabModulator::DabModulator(
-        EtiSource& etiSource,
-        tii_config_t& tiiConfig,
-        unsigned outputRate, unsigned clockRate,
-        unsigned dabMode, GainMode gainMode,
-        float& digGain, float normalise,
-        float gainmodeVariance,
-        const std::string& filterTapsFilename,
-        const std::string& polyCoefFilename
-        ) :
+DabModulator::DabModulator(EtiSource& etiSource,
+                           const mod_settings_t& settings) :
     ModInput(),
-    myOutputRate(outputRate),
-    myClockRate(clockRate),
-    myDabMode(dabMode),
-    myGainMode(gainMode),
-    myDigGain(digGain),
-    myNormalise(normalise),
-    myGainmodeVariance(gainmodeVariance),
+    m_settings(settings),
     myEtiSource(etiSource),
-    myFlowgraph(NULL),
-    myFilterTapsFilename(filterTapsFilename),
-    myPolyCoefFilename(polyCoefFilename),
-    myTiiConfig(tiiConfig)
+    myFlowgraph()
 {
     PDEBUG("DabModulator::DabModulator(%u, %u, %u, %zu) @ %p\n",
             outputRate, clockRate, dabMode, (size_t)gainMode, this);
 
-    if (myDabMode == 0) {
+    if (m_settings.dabMode == 0) {
         setMode(2);
-    } else {
-        setMode(myDabMode);
     }
-}
-
-
-DabModulator::~DabModulator()
-{
-    PDEBUG("DabModulator::~DabModulator() @ %p\n", this);
-
-    delete myFlowgraph;
+    else {
+        setMode(m_settings.dabMode);
+    }
 }
 
 
@@ -145,16 +121,16 @@ int DabModulator::process(Buffer* dataOut)
 
     PDEBUG("DabModulator::process(dataOut: %p)\n", dataOut);
 
-    if (myFlowgraph == NULL) {
+    if (not myFlowgraph) {
         unsigned mode = myEtiSource.getMode();
-        if (myDabMode != 0) {
-            mode = myDabMode;
+        if (m_settings.dabMode != 0) {
+            mode = m_settings.dabMode;
         } else if (mode == 0) {
             mode = 4;
         }
         setMode(mode);
 
-        myFlowgraph = new Flowgraph();
+        myFlowgraph = make_shared<Flowgraph>();
         ////////////////////////////////////////////////////////////////
         // CIF data initialisation
         ////////////////////////////////////////////////////////////////
@@ -174,10 +150,10 @@ int DabModulator::process(Buffer* dataOut)
         // TODO this needs a review
         bool useCicEq = false;
         unsigned cic_ratio = 1;
-        if (myClockRate) {
-            cic_ratio = myClockRate / myOutputRate;
+        if (m_settings.clockRate) {
+            cic_ratio = m_settings.clockRate / m_settings.outputRate;
             cic_ratio /= 4; // FPGA DUC
-            if (myClockRate == 400000000) { // USRP2
+            if (m_settings.clockRate == 400000000) { // USRP2
                 if (cic_ratio & 1) { // odd
                     useCicEq = true;
                 } // even, no filter
@@ -189,12 +165,16 @@ int DabModulator::process(Buffer* dataOut)
 
         auto cifCicEq = make_shared<CicEqualizer>(
                 myNbCarriers,
-                (float)mySpacing * (float)myOutputRate / 2048000.0f, cic_ratio);
+                (float)mySpacing * (float)m_settings.outputRate / 2048000.0f,
+                cic_ratio);
 
         shared_ptr<TII> tii;
         shared_ptr<PhaseReference> tiiRef;
         try {
-            tii = make_shared<TII>(myDabMode, myTiiConfig, myEtiSource.getFp());
+            tii = make_shared<TII>(
+                    m_settings.dabMode,
+                    m_settings.tiiConfig,
+                    myEtiSource.getFp());
             rcs.enrol(tii.get());
             tiiRef = make_shared<PhaseReference>(mode);
         }
@@ -206,8 +186,11 @@ int DabModulator::process(Buffer* dataOut)
                 (1 + myNbSymbols), myNbCarriers, mySpacing);
 
         auto cifGain = make_shared<GainControl>(
-                mySpacing, myGainMode, myDigGain, myNormalise, 
-                myGainmodeVariance);
+                mySpacing,
+                m_settings.gainMode,
+                m_settings.digitalgain,
+                m_settings.normalise,
+                m_settings.gainmodeVariance);
 
         rcs.enrol(cifGain.get());
 
@@ -215,23 +198,28 @@ int DabModulator::process(Buffer* dataOut)
                 myNbSymbols, mySpacing, myNullSize, mySymSize);
 
         shared_ptr<FIRFilter> cifFilter;
-        if (not myFilterTapsFilename.empty()) {
-            cifFilter = make_shared<FIRFilter>(myFilterTapsFilename);
+        if (not m_settings.filterTapsFilename.empty()) {
+            cifFilter = make_shared<FIRFilter>(m_settings.filterTapsFilename);
             rcs.enrol(cifFilter.get());
         }
 
         shared_ptr<MemlessPoly> cifPoly;
-        if (not myPolyCoefFilename.empty()) {
-            cifPoly = make_shared<MemlessPoly>(myPolyCoefFilename);
+        if (not m_settings.polyCoefFilename.empty()) {
+            cifPoly = make_shared<MemlessPoly>(m_settings.polyCoefFilename,
+                                               m_settings.polyNumThreads);
             rcs.enrol(cifPoly.get());
         }
 
         auto myOutput = make_shared<OutputMemory>(dataOut);
 
         shared_ptr<Resampler> cifRes;
-        if (myOutputRate != 2048000) {
-            cifRes = make_shared<Resampler>(2048000, myOutputRate, mySpacing);
-        } else {
+        if (m_settings.outputRate != 2048000) {
+            cifRes = make_shared<Resampler>(
+                    2048000,
+                    m_settings.outputRate,
+                    mySpacing);
+        }
+        else {
             fprintf(stderr, "No resampler\n");
         }
 
diff --git a/src/DabModulator.h b/src/DabModulator.h
index 0c691dd..6878853 100644
--- a/src/DabModulator.h
+++ b/src/DabModulator.h
@@ -36,6 +36,7 @@
 #include <memory>
 
 #include "ModPlugin.h"
+#include "ConfigParser.h"
 #include "EtiReader.h"
 #include "Flowgraph.h"
 #include "GainControl.h"
@@ -48,18 +49,8 @@
 class DabModulator : public ModInput
 {
 public:
-    DabModulator(
-            EtiSource& etiSource,
-            tii_config_t& tiiConfig,
-            unsigned outputRate, unsigned clockRate,
-            unsigned dabMode, GainMode gainMode,
-            float& digGain, float normalise,
-            float gainmodeVariance,
-            const std::string& filterTapsFilename,
-            const std::string& polyCoefFilename);
-    DabModulator(const DabModulator& other) = delete;
-    DabModulator& operator=(const DabModulator& other) = delete;
-    virtual ~DabModulator();
+    DabModulator(EtiSource& etiSource,
+                 const mod_settings_t& settings);
 
     int process(Buffer* dataOut);
     const char* name() { return "DabModulator"; }
@@ -70,19 +61,10 @@ public:
 protected:
     void setMode(unsigned mode);
 
-    unsigned myOutputRate;
-    unsigned myClockRate;
-    unsigned myDabMode;
-    GainMode myGainMode;
-    float& myDigGain;
-    float myNormalise;
-    float myGainmodeVariance;
+    const mod_settings_t& m_settings;
+
     EtiSource& myEtiSource;
-    Flowgraph* myFlowgraph;
-    OutputMemory* myOutput;
-    std::string myFilterTapsFilename;
-    std::string myPolyCoefFilename;
-    tii_config_t& myTiiConfig;
+    std::shared_ptr<Flowgraph> myFlowgraph;
 
     size_t myNbSymbols;
     size_t myNbCarriers;
diff --git a/src/GainControl.cpp b/src/GainControl.cpp
index f363d20..2a91b12 100644
--- a/src/GainControl.cpp
+++ b/src/GainControl.cpp
@@ -47,7 +47,7 @@ static float var_variance;
 
 GainControl::GainControl(size_t framesize,
                          GainMode mode,
-                         float& digGain,
+                         float digGain,
                          float normalise,
                          float varVariance) :
     PipelinedModCodec(),
diff --git a/src/GainControl.h b/src/GainControl.h
index e8f1be9..44c9fa9 100644
--- a/src/GainControl.h
+++ b/src/GainControl.h
@@ -52,7 +52,7 @@ class GainControl : public PipelinedModCodec, public RemoteControllable
     public:
         GainControl(size_t framesize,
                     GainMode mode,
-                    float& digGain,
+                    float digGain,
                     float normalise,
                     float varVariance);
 
@@ -76,7 +76,7 @@ class GainControl : public PipelinedModCodec, public RemoteControllable
                 Buffer* const dataIn, Buffer* dataOut) override;
 
         size_t m_frameSize;
-        float& m_digGain;
+        float m_digGain;
         float m_normalise;
 
         // The following variables are accessed from the RC thread
diff --git a/src/InputZeroMQReader.cpp b/src/InputZeroMQReader.cpp
index 1418db7..783f0f5 100644
--- a/src/InputZeroMQReader.cpp
+++ b/src/InputZeroMQReader.cpp
@@ -3,7 +3,7 @@
    Her Majesty the Queen in Right of Canada (Communications Research
    Center Canada)
 
-   Copyright (C) 2013, 2014, 2015
+   Copyright (C) 2017
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://opendigitalradio.org
@@ -42,6 +42,8 @@
 #include "PcDebug.h"
 #include "Utils.h"
 
+using namespace std;
+
 #define NUM_FRAMES_PER_ZMQ_MESSAGE 4
 /* A concatenation of four ETI frames,
  * whose maximal size is 6144.
@@ -63,7 +65,10 @@ struct zmq_dab_message_t
     uint8_t  buf[NUM_FRAMES_PER_ZMQ_MESSAGE*6144];
 };
 
-int InputZeroMQReader::Open(const std::string& uri, size_t max_queued_frames)
+#define ZMQ_DAB_MESSAGE_T_HEADERSIZE \
+    (sizeof(uint32_t) + NUM_FRAMES_PER_ZMQ_MESSAGE*sizeof(uint16_t))
+
+int InputZeroMQReader::Open(const string& uri, size_t max_queued_frames)
 {
     // The URL might start with zmq+tcp://
     if (uri.substr(0, 4) == "zmq+") {
@@ -89,7 +94,7 @@ int InputZeroMQReader::GetNextFrame(void* buffer)
         return 0;
     }
 
-    std::shared_ptr<std::vector<uint8_t> > incoming;
+    shared_ptr<vector<uint8_t> > incoming;
 
     /* Do some prebuffering because reads will happen in bursts
      * (4 ETI frames in TM1) and we should make sure that
@@ -167,41 +172,51 @@ void InputZeroMQWorker::RecvProcess(struct InputZeroMQThreadData* workerdata)
             }
             else if (queue_size < workerdata->max_queued_frames) {
                 if (buffer_full) {
-                    etiLog.level(info) << "ZeroMQ buffer recovered: " << queue_size << " elements";
+                    etiLog.level(info) << "ZeroMQ buffer recovered: " <<
+                        queue_size << " elements";
                     buffer_full = false;
                 }
 
-                zmq_dab_message_t* dab_msg = (zmq_dab_message_t*)incoming.data();
-
-                if (dab_msg->version != 1) {
-                    etiLog.level(error) << "ZeroMQ wrong packet version " << dab_msg->version;
+                if (incoming.size() < ZMQ_DAB_MESSAGE_T_HEADERSIZE) {
+                    throw runtime_error("ZeroMQ packet too small for header");
                 }
+                else {
+                    const zmq_dab_message_t* dab_msg = (zmq_dab_message_t*)incoming.data();
 
-                int offset = sizeof(dab_msg->version) +
-                    NUM_FRAMES_PER_ZMQ_MESSAGE * sizeof(*dab_msg->buflen);
-
-                for (int i = 0; i < NUM_FRAMES_PER_ZMQ_MESSAGE; i++)
-                {
-                    if (dab_msg->buflen[i] <= 0 ||
-                        dab_msg->buflen[i] > 6144)
-                    {
-                        etiLog.level(error) << "ZeroMQ buffer " << i << " has invalid length " <<
-                            dab_msg->buflen[i];
+                    if (dab_msg->version != 1) {
+                        etiLog.level(error) <<
+                            "ZeroMQ wrong packet version " <<
+                            dab_msg->version;
                     }
-                    else {
-                        std::shared_ptr<std::vector<uint8_t> > buf =
-                            std::make_shared<std::vector<uint8_t> >(6144, 0x55);
 
-                        const int framesize = dab_msg->buflen[i];
+                    int offset = sizeof(dab_msg->version) +
+                        NUM_FRAMES_PER_ZMQ_MESSAGE * sizeof(*dab_msg->buflen);
+
+                    for (int i = 0; i < NUM_FRAMES_PER_ZMQ_MESSAGE; i++) {
+                        if (dab_msg->buflen[i] > 6144) {
+                            stringstream ss;
+                            ss << "ZeroMQ buffer " << i <<
+                                " has invalid buflen " << dab_msg->buflen[i];
+                            throw runtime_error(ss.str());
+                        }
+                        else {
+                            auto buf = make_shared<vector<uint8_t> >(6144, 0x55);
+
+                            const int framesize = dab_msg->buflen[i];
+
+                            if ((ssize_t)incoming.size() < offset + framesize) {
+                                throw runtime_error("ZeroMQ packet too small");
+                            }
 
-                        memcpy(&buf->front(),
-                                ((uint8_t*)incoming.data()) + offset,
-                                framesize);
+                            memcpy(&buf->front(),
+                                    ((uint8_t*)incoming.data()) + offset,
+                                    framesize);
 
-                        offset += framesize;
+                            offset += framesize;
 
-                        queue_size = workerdata->in_messages->push(buf);
-                        etiLog.log(trace, "ZMQ,push %zu", queue_size);
+                            queue_size = workerdata->in_messages->push(buf);
+                            etiLog.log(trace, "ZMQ,push %zu", queue_size);
+                        }
                     }
                 }
             }
@@ -212,7 +227,7 @@ void InputZeroMQWorker::RecvProcess(struct InputZeroMQThreadData* workerdata)
                     etiLog.level(warn) << "ZeroMQ buffer overfull !";
 
                     buffer_full = true;
-                    throw std::runtime_error("ZMQ input full");
+                    throw runtime_error("ZMQ input full");
                 }
 
                 queue_size = workerdata->in_messages->size();
diff --git a/src/MemlessPoly.cpp b/src/MemlessPoly.cpp
index 71ceac3..b0d950c 100644
--- a/src/MemlessPoly.cpp
+++ b/src/MemlessPoly.cpp
@@ -53,9 +53,10 @@ static const std::array<complexf, 8> default_coefficients({{
         }});
 
 
-MemlessPoly::MemlessPoly(const std::string& coefs_file) :
+MemlessPoly::MemlessPoly(const std::string& coefs_file, unsigned int num_threads) :
     PipelinedModCodec(),
     RemoteControllable("memlesspoly"),
+    m_num_threads(num_threads),
     m_coefs(),
     m_coefs_file(coefs_file),
     m_coefs_mutex()
@@ -63,6 +64,16 @@ MemlessPoly::MemlessPoly(const std::string& coefs_file) :
     PDEBUG("MemlessPoly::MemlessPoly(%s) @ %p\n",
             coefs_file.c_str(), this);
 
+    if (m_num_threads == 0) {
+        const unsigned int hw_concurrency = std::thread::hardware_concurrency();
+        etiLog.level(info) << "Polynomial Predistorter will use " <<
+            hw_concurrency << " threads (auto detected)";
+    }
+    else {
+        etiLog.level(info) << "Polynomial Predistorter will use " <<
+            m_num_threads << " threads (set in config file)";
+    }
+
     RC_ADD_PARAMETER(ncoefs, "(Read-only) number of coefficients.");
     RC_ADD_PARAMETER(coeffile, "Filename containing coefficients. When written to, the new file gets automatically loaded.");
 
@@ -156,12 +167,15 @@ int MemlessPoly::internal_process(Buffer* const dataIn, Buffer* dataOut)
         std::lock_guard<std::mutex> lock(m_coefs_mutex);
         const unsigned int hw_concurrency = std::thread::hardware_concurrency();
 
-        if (hw_concurrency) {
-            const size_t step = sizeOut / hw_concurrency;
+        const unsigned int num_threads =
+            (m_num_threads > 0) ? m_num_threads : hw_concurrency;
+
+        if (num_threads) {
+            const size_t step = sizeOut / num_threads;
             vector<future<void> > flags;
 
             size_t start = 0;
-            for (size_t i = 0; i < hw_concurrency - 1; i++) {
+            for (size_t i = 0; i < num_threads - 1; i++) {
                 flags.push_back(async(launch::async, apply_coeff,
                             m_coefs, in, start, start + step, out));
 
@@ -177,13 +191,6 @@ int MemlessPoly::internal_process(Buffer* const dataIn, Buffer* dataOut)
             }
         }
         else {
-            static bool error_printed = false;
-            if (not error_printed) {
-                etiLog.level(warn) <<
-                    "Your platform doesn't seem to have hardware concurrency. "
-                    "MemlessPoly will run single-threaded";
-            }
-            // For some reason we don't have hw concurrency.
             apply_coeff(m_coefs, in, 0, sizeOut, out);
         }
     }
diff --git a/src/MemlessPoly.h b/src/MemlessPoly.h
index 9fe19d7..b7fd81e 100644
--- a/src/MemlessPoly.h
+++ b/src/MemlessPoly.h
@@ -52,7 +52,7 @@ typedef std::complex<float> complexf;
 class MemlessPoly : public PipelinedModCodec, public RemoteControllable
 {
 public:
-    MemlessPoly(const std::string& coefs_file);
+    MemlessPoly(const std::string& coefs_file, unsigned int num_threads);
 
     virtual const char* name() { return "MemlessPoly"; }
 
@@ -67,6 +67,7 @@ private:
     int internal_process(Buffer* const dataIn, Buffer* dataOut);
     void load_coefficients(const std::string &coefFile);
 
+    unsigned int m_num_threads;
     std::vector<complexf> m_coefs;
     std::string m_coefs_file;
     mutable std::mutex m_coefs_mutex;
diff --git a/src/TII.cpp b/src/TII.cpp
index 6bc3f0d..8a8bd86 100644
--- a/src/TII.cpp
+++ b/src/TII.cpp
@@ -106,7 +106,7 @@ const int pattern_tm1_2_4[][8] = { // {{{
     {1,1,1,0,1,0,0,0},
     {1,1,1,1,0,0,0,0} }; // }}}
 
-TII::TII(unsigned int dabmode, tii_config_t& tii_config, unsigned phase) :
+TII::TII(unsigned int dabmode, const tii_config_t& tii_config, unsigned phase) :
     ModCodec(),
     RemoteControllable("tii"),
     m_dabmode(dabmode),
diff --git a/src/TII.h b/src/TII.h
index 2bd2040..fe67978 100644
--- a/src/TII.h
+++ b/src/TII.h
@@ -83,7 +83,7 @@ class TIIError : public std::runtime_error {
 class TII : public ModCodec, public RemoteControllable
 {
     public:
-        TII(unsigned int dabmode, tii_config_t& tii_config, unsigned phase);
+        TII(unsigned int dabmode, const tii_config_t& tii_config, unsigned phase);
         virtual ~TII();
         TII(const TII&) = delete;
         TII& operator=(const TII&) = delete;
@@ -110,7 +110,7 @@ class TII : public ModCodec, public RemoteControllable
         unsigned int m_dabmode;
 
         // Remote-controllable settings
-        tii_config_t& m_conf;
+        tii_config_t m_conf;
 
         // Internal flag when to insert TII
         bool m_insert;