Add WIP for OutputUHDFeedback

7 files changed, 383 insertions, 23 deletions

diff --git a/Makefile.am b/Makefile.am
index d4365fd..12dfe6e 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -1,7 +1,7 @@
 # Copyright (C) 2007, 2008, 2009, 2010 Her Majesty the Queen in Right
 # of Canada (Communications Research Center Canada)
 #
-#  Copyright (C) 2016
+#  Copyright (C) 2017
 #  Matthias P. Braendli, matthias.braendli@mpb.li
 
 #   http://opendigitalradio.org
@@ -82,6 +82,8 @@ odr_dabmod_SOURCES  = src/DabMod.cpp \
 					  src/TimestampDecoder.cpp \
 					  src/OutputUHD.cpp \
 					  src/OutputUHD.h \
+					  src/OutputUHDFeedback.cpp \
+					  src/OutputUHDFeedback.h \
 					  src/OutputSoapy.cpp \
 					  src/OutputSoapy.h \
 					  src/InputMemory.cpp \
diff --git a/doc/example.ini b/doc/example.ini
index ffb20d7..fa03d26 100644
--- a/doc/example.ini
+++ b/doc/example.ini
@@ -255,6 +255,11 @@ behaviour_refclk_lock_lost=ignore
 ; default value: 0
 max_gps_holdover_time=600
 
+; Enable the TCP server to communicate TX and RX feedback for
+; digital predistortion.
+; Set to 0 to disable
+dpd_port=50055
+
 ; section defining ZeroMQ output properties
 [zmqoutput]
 
diff --git a/src/ConfigParser.cpp b/src/ConfigParser.cpp
index b7649df..8892642 100644
--- a/src/ConfigParser.cpp
+++ b/src/ConfigParser.cpp
@@ -255,6 +255,8 @@ static void parse_configfile(
 
         outputuhd_conf.maxGPSHoldoverTime = pt.get("uhdoutput.max_gps_holdover_time", 0);
 
+        outputuhd_conf.dpdFeedbackServerPort = pt.get<long>("uhdoutput.dpd_port", 0);
+
         mod_settings.outputuhd_conf = outputuhd_conf;
         mod_settings.useUHDOutput = 1;
     }
diff --git a/src/OutputUHD.cpp b/src/OutputUHD.cpp
index d78c4bf..6dc8878 100644
--- a/src/OutputUHD.cpp
+++ b/src/OutputUHD.cpp
@@ -169,8 +169,6 @@ OutputUHD::OutputUHD(
     RC_ADD_PARAMETER(muting, "Mute the output by stopping the transmitter");
     RC_ADD_PARAMETER(staticdelay, "Set static delay (uS) between 0 and 96000");
 
-    // TODO: find out how to use boost::bind to give the logger to the
-    // uhd_msg_handler
     uhd::msg::register_handler(uhd_msg_handler);
 
     uhd::set_thread_priority_safe();
@@ -286,13 +284,9 @@ OutputUHD::OutputUHD(
 
     SetDelayBuffer(myConf.dabMode);
 
-    MDEBUG("OutputUHD:UHD ready.\n");
-}
+    uhdFeedback.setup(myUsrp, myConf.dpdFeedbackServerPort);
 
-
-OutputUHD::~OutputUHD()
-{
-    MDEBUG("OutputUHD::~OutputUHD() @ %p\n", this);
+    MDEBUG("OutputUHD:UHD ready.\n");
 }
 
 
@@ -426,12 +420,11 @@ int OutputUHD::process(Buffer* dataIn)
                 "OutputUHD: dropping one frame with invalid FCT";
         }
         else {
-            while (true) {
-                size_t num_frames = uwd.frames.push_wait_if_full(frame,
-                        FRAMES_MAX_SIZE);
-                etiLog.log(trace, "UHD,push %zu", num_frames);
-                break;
-            }
+            uhdFeedback.set_tx_frame(frame.buf, frame.ts);
+
+            size_t num_frames = uwd.frames.push_wait_if_full(frame,
+                    FRAMES_MAX_SIZE);
+            etiLog.log(trace, "UHD,push %zu", num_frames);
         }
     }
 
diff --git a/src/OutputUHD.h b/src/OutputUHD.h
index d42245f..1246fc5 100644
--- a/src/OutputUHD.h
+++ b/src/OutputUHD.h
@@ -2,7 +2,7 @@
    Copyright (C) 2005, 2006, 2007, 2008, 2009, 2010 Her Majesty the
    Queen in Right of Canada (Communications Research Center Canada)
 
-   Copyright (C) 2016
+   Copyright (C) 2017
    Matthias P. Braendli, matthias.braendli@mpb.li
 
     http://opendigitalradio.org
@@ -56,6 +56,7 @@ DESCRIPTION:
 #include "TimestampDecoder.h"
 #include "RemoteControl.h"
 #include "ThreadsafeQueue.h"
+#include "OutputUHDFeedback.h"
 
 #include <stdio.h>
 #include <sys/types.h>
@@ -210,14 +211,15 @@ struct OutputUHDConfig {
 
     // static delay in microseconds
     int staticDelayUs = 0;
-};
 
+    // TCP port on which to serve TX and RX samples for the
+    // digital pre distortion learning tool
+    uint16_t dpdFeedbackServerPort = 0;
+};
 
 class OutputUHD: public ModOutput, public RemoteControllable {
     public:
-
         OutputUHD(OutputUHDConfig& config);
-        ~OutputUHD();
 
         int process(Buffer* dataIn);
 
@@ -235,11 +237,7 @@ class OutputUHD: public ModOutput, public RemoteControllable {
         virtual const std::string get_parameter(
                 const std::string& parameter) const;
 
-
     protected:
-        OutputUHD(const OutputUHD& other) = delete;
-        OutputUHD& operator=(const OutputUHD& other) = delete;
-
         EtiSource *myEtiSource;
         OutputUHDConfig& myConf;
         uhd::usrp::multi_usrp::sptr myUsrp;
@@ -248,6 +246,7 @@ class OutputUHD: public ModOutput, public RemoteControllable {
         bool gps_fix_verified;
         struct UHDWorkerData uwd;
         UHDWorker worker;
+        OutputUHDFeedback uhdFeedback;
 
     private:
         // Resize the internal delay buffer according to the dabMode and
diff --git a/src/OutputUHDFeedback.cpp b/src/OutputUHDFeedback.cpp
new file mode 100644
index 0000000..dfe0f74
--- /dev/null
+++ b/src/OutputUHDFeedback.cpp
@@ -0,0 +1,245 @@
+/*
+   Copyright (C) 2005, 2006, 2007, 2008, 2009, 2010 Her Majesty the
+   Queen in Right of Canada (Communications Research Center Canada)
+
+   Copyright (C) 2017
+   Matthias P. Braendli, matthias.braendli@mpb.li
+
+    http://opendigitalradio.org
+
+DESCRIPTION:
+   This presents a TCP socket to an external tool which calculates
+   a Digital Predistortion model from a short sequence of transmit
+   samples and corresponding receive samples.
+*/
+
+/*
+   This file is part of ODR-DabMod.
+
+   ODR-DabMod is free software: you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as
+   published by the Free Software Foundation, either version 3 of the
+   License, or (at your option) any later version.
+
+   ODR-DabMod is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with ODR-DabMod.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include <vector>
+#include <uhd/types/stream_cmd.hpp>
+#include <sys/socket.h>
+#include "OutputUHDFeedback.h"
+#include "Utils.h"
+
+using namespace std;
+typedef std::complex<float> complexf;
+
+OutputUHDFeedback::OutputUHDFeedback()
+{
+    running = false;
+}
+
+void OutputUHDFeedback::setup(uhd::usrp::multi_usrp::sptr usrp, uint16_t port)
+{
+    myUsrp = usrp;
+    burstRequest.state = BurstRequestState::None;
+
+    if (port) {
+        m_port = port;
+        running = true;
+
+        rx_burst_thread = boost::thread(&OutputUHDFeedback::ReceiveBurstThread, this);
+        burst_tcp_thread = boost::thread(&OutputUHDFeedback::ServeFeedbackThread, this);
+    }
+}
+
+OutputUHDFeedback::~OutputUHDFeedback()
+{
+    running = false;
+    rx_burst_thread.join();
+    burst_tcp_thread.join();
+}
+
+void OutputUHDFeedback::set_tx_frame(
+        const std::vector<uint8_t> &buf,
+        const struct frame_timestamp& ts)
+{
+    boost::mutex::scoped_lock lock(burstRequest.mutex);
+
+    if (burstRequest.state == BurstRequestState::SaveTransmitFrame) {
+        const size_t n = std::min(
+                burstRequest.frame_length * sizeof(complexf), buf.size());
+
+        burstRequest.tx_samples.clear();
+        burstRequest.tx_samples.resize(n);
+        copy(buf.begin(), buf.begin() + n, burstRequest.tx_samples.begin());
+
+        burstRequest.tx_second = ts.timestamp_sec;
+        burstRequest.tx_pps = ts.timestamp_pps;
+
+        // Prepare the next state
+        burstRequest.rx_second = ts.timestamp_sec;
+        burstRequest.rx_pps = ts.timestamp_pps;
+        burstRequest.state = BurstRequestState::SaveReceiveFrame;
+
+        lock.unlock();
+        burstRequest.mutex_notification.notify_one();
+    }
+    else {
+        lock.unlock();
+    }
+}
+
+void OutputUHDFeedback::ReceiveBurstThread()
+{
+    set_thread_name("uhdreceiveburst");
+
+    uhd::stream_args_t stream_args("fc32"); //complex floats
+    auto rxStream = myUsrp->get_rx_stream(stream_args);
+
+    while (running) {
+        boost::mutex::scoped_lock lock(burstRequest.mutex);
+        while (burstRequest.state != BurstRequestState::SaveReceiveFrame) {
+            if (not running) break;
+            burstRequest.mutex_notification.wait(lock);
+        }
+
+        if (not running) break;
+
+        uhd::stream_cmd_t cmd(
+                uhd::stream_cmd_t::stream_mode_t::STREAM_MODE_NUM_SAMPS_AND_DONE);
+        cmd.num_samps = burstRequest.frame_length;
+        cmd.stream_now = false;
+
+        double pps = burstRequest.rx_pps / 16384000.0;
+        cmd.time_spec = uhd::time_spec_t(burstRequest.rx_second, pps);
+
+        rxStream->issue_stream_cmd(cmd);
+
+        uhd::rx_metadata_t md;
+        burstRequest.rx_samples.resize(burstRequest.frame_length * sizeof(complexf));
+        rxStream->recv(&burstRequest.rx_samples[0], burstRequest.frame_length, md);
+
+        burstRequest.rx_second = md.time_spec.get_full_secs();
+        burstRequest.rx_pps = md.time_spec.get_frac_secs() * 16384000.0;
+
+        burstRequest.state = BurstRequestState::Acquired;
+
+        lock.unlock();
+        burstRequest.mutex_notification.notify_one();
+    }
+}
+
+void OutputUHDFeedback::ServeFeedbackThread()
+{
+    set_thread_name("uhdservefeedback");
+
+    int server_sock = -1;
+    try {
+        if ((server_sock = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
+            throw std::runtime_error("Can't create TCP socket");
+        }
+
+        struct sockaddr_in addr;
+        addr.sin_family = AF_INET;
+        addr.sin_port = htons(m_port);
+        addr.sin_addr.s_addr = htonl(INADDR_ANY);
+
+        if (bind(server_sock, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
+            throw std::runtime_error("Can't bind TCP socket");
+        }
+
+        if (listen(server_sock, 1) < 0) {
+            throw std::runtime_error("Can't listen TCP socket");
+        }
+
+        while (running) {
+            struct sockaddr_in client;
+            socklen_t client_len = sizeof(client);
+            int client_sock = accept(server_sock,
+                    (struct sockaddr*)&client, &client_len);
+
+            if (client_sock < 0) {
+                throw runtime_error("Could not establish new connection");
+            }
+
+            while (running) {
+                uint8_t request_version = 0;
+                int read = recv(client_sock, &request_version, 1, 0);
+                if (!read) break; // done reading
+                if (read < 0) {
+                    etiLog.level(info) <<
+                        "DPD Feedback Server Client read request verson failed";
+                }
+
+                if (request_version != 1) {
+                    etiLog.level(info) << "DPD Feedback Server wrong request version";
+                    break;
+                }
+
+                uint32_t num_samples = 0;
+                read = recv(client_sock, &num_samples, 4, 0);
+                if (!read) break; // done reading
+                if (read < 0) {
+                    etiLog.level(info) <<
+                        "DPD Feedback Server Client read num samples failed";
+                }
+
+                // We are ready to issue the request now
+                {
+                    boost::mutex::scoped_lock lock(burstRequest.mutex);
+                    burstRequest.frame_length = num_samples;
+                    burstRequest.state = BurstRequestState::SaveTransmitFrame;
+
+                    lock.unlock();
+                }
+
+                // Wait for the result to be ready
+                boost::mutex::scoped_lock lock(burstRequest.mutex);
+                while (burstRequest.state != BurstRequestState::Acquired) {
+                    if (not running) break;
+                    burstRequest.mutex_notification.wait(lock);
+                }
+
+                burstRequest.state = BurstRequestState::None;
+                lock.unlock();
+
+                if (send(client_sock,
+                            &burstRequest.tx_second,
+                            sizeof(burstRequest.tx_second),
+                            0) < 0) {
+                    etiLog.level(info) <<
+                        "DPD Feedback Server Client send tx_second failed";
+                    break;
+                }
+
+                if (send(client_sock,
+                            &burstRequest.tx_pps,
+                            sizeof(burstRequest.tx_pps),
+                            0) < 0) {
+                    etiLog.level(info) <<
+                        "DPD Feedback Server Client send tx_pps failed";
+                    break;
+                }
+
+#warning "Send buf"
+            }
+
+            close(client_sock);
+        }
+    }
+    catch (runtime_error &e) {
+        etiLog.level(error) << "DPD Feedback Server fault: " << e.what();
+    }
+
+    running = false;
+
+    if (server_sock != -1) {
+        close(server_sock);
+    }
+}
diff --git a/src/OutputUHDFeedback.h b/src/OutputUHDFeedback.h
new file mode 100644
index 0000000..31f7547
--- /dev/null
+++ b/src/OutputUHDFeedback.h
@@ -0,0 +1,114 @@
+/*
+   Copyright (C) 2005, 2006, 2007, 2008, 2009, 2010 Her Majesty the
+   Queen in Right of Canada (Communications Research Center Canada)
+
+   Copyright (C) 2017
+   Matthias P. Braendli, matthias.braendli@mpb.li
+
+    http://opendigitalradio.org
+
+DESCRIPTION:
+   This presents a TCP socket to an external tool which calculates
+   a Digital Predistortion model from a short sequence of transmit
+   samples and corresponding receive samples.
+*/
+
+/*
+   This file is part of ODR-DabMod.
+
+   ODR-DabMod is free software: you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as
+   published by the Free Software Foundation, either version 3 of the
+   License, or (at your option) any later version.
+
+   ODR-DabMod is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with ODR-DabMod.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#ifdef HAVE_CONFIG_H
+#   include <config.h>
+#endif
+
+#ifdef HAVE_OUTPUT_UHD
+
+#include <uhd/utils/thread_priority.hpp>
+#include <uhd/utils/safe_main.hpp>
+#include <uhd/usrp/multi_usrp.hpp>
+#include <boost/thread.hpp>
+#include <memory>
+#include <string>
+
+#include "Log.h"
+#include "TimestampDecoder.h"
+
+enum class BurstRequestState {
+    None, // To pending request
+    SaveTransmitFrame, // The TX thread has to save an outgoing frame
+    SaveReceiveFrame, // The RX thread has to save an incoming frame
+    Acquired, // Both TX and RX frames are ready
+};
+
+struct UHDReceiveBurstRequest {
+    // All fields in this struct are protected
+    mutable boost::mutex mutex;
+    boost::condition_variable mutex_notification;
+
+    BurstRequestState state;
+
+    // In the SaveTransmit states, frame_length samples are saved into
+    // the vectors
+    size_t frame_length;
+
+    // The timestamp of the first sample of the TX buffers
+    uint32_t tx_second;
+    uint32_t tx_pps; // in units of 1/16384000s
+
+    std::vector<uint8_t> tx_samples;
+
+    // The timestamp of the first sample of the RX buffers
+    uint32_t rx_second;
+    uint32_t rx_pps;
+
+    std::vector<uint8_t> rx_samples;
+};
+
+
+class OutputUHDFeedback {
+    public:
+        OutputUHDFeedback();
+        OutputUHDFeedback(const OutputUHDFeedback& other) = delete;
+        OutputUHDFeedback& operator=(const OutputUHDFeedback& other) = delete;
+        ~OutputUHDFeedback();
+
+        void setup(uhd::usrp::multi_usrp::sptr usrp, uint16_t port);
+
+        void set_tx_frame(const std::vector<uint8_t> &buf,
+                const struct frame_timestamp& ts);
+
+
+    private:
+        // Thread that reacts to burstRequests and receives from the USRP
+        void ReceiveBurstThread(void);
+
+        // Thread that listens for requests over TCP to get TX and RX feedback
+        void ServeFeedbackThread(void);
+
+        boost::thread rx_burst_thread;
+        boost::thread burst_tcp_thread;
+
+        UHDReceiveBurstRequest burstRequest;
+
+        bool running = false;
+        uint16_t m_port = 0;
+        uhd::usrp::multi_usrp::sptr myUsrp;
+};
+
+
+#endif // HAVE_OUTPUT_UHD