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

10 files changed, 544 insertions, 126 deletions

diff --git a/Makefile.am b/Makefile.am
index 12dfe6e..2d84ab6 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -44,6 +44,7 @@ odr_dabmod_CFLAGS   = -Wall -Isrc -Ilib \
 odr_dabmod_LDADD    = $(FFT_LDADD)
 odr_dabmod_SOURCES  = src/DabMod.cpp \
 					  src/PcDebug.h \
+					  src/Socket.h \
 					  src/porting.c \
 					  src/porting.h \
 					  src/DabModulator.cpp \
diff --git a/doc/example.ini b/doc/example.ini
index 425dfa4..21f15ea 100644
--- a/doc/example.ini
+++ b/doc/example.ini
@@ -127,6 +127,18 @@ dac_clk_rate=0
 ; and
 ;dac_clk_rate=128000000
 
+; Settings for crest factor reduction. Statistics for ratio of
+; samples that were clipped are available through the RC.
+[cfr]
+enable=0
+
+; At what amplitude the signal should be clipped
+clip=70.0
+
+; How much to clip the error signal used to compensate the effect
+; of clipping
+error_clip=0.05
+
 [firfilter]
 ; The FIR Filter can be used to create a better spectral quality.
 enabled=1
diff --git a/src/ConfigParser.cpp b/src/ConfigParser.cpp
index 9ac1280..1cc94c0 100644
--- a/src/ConfigParser.cpp
+++ b/src/ConfigParser.cpp
@@ -177,6 +177,13 @@ static void parse_configfile(
             pt.get<int>("poly.num_threads", 0);
     }
 
+    // Crest factor reduction
+    if (pt.get("cfr.enabled", 0) == 1) {
+        mod_settings.enableCfr = true;
+        mod_settings.cfrClip = pt.get<float>("cfr.clip");
+        mod_settings.cfrErrorClip = pt.get<float>("cfr.error_clip");
+    }
+
     // Output options
     std::string output_selected;
     try {
diff --git a/src/ConfigParser.h b/src/ConfigParser.h
index 89f0fb7..a8d7837 100644
--- a/src/ConfigParser.h
+++ b/src/ConfigParser.h
@@ -77,6 +77,12 @@ struct mod_settings_t {
     std::string polyCoefFilename = "";
     unsigned polyNumThreads = 0;
 
+    // Settings for crest factor reduction
+    bool enableCfr = false;
+    float cfrClip = 1.0f;
+    float cfrErrorClip = 1.0f;
+
+
 #if defined(HAVE_OUTPUT_UHD)
     OutputUHDConfig outputuhd_conf;
 #endif
diff --git a/src/DabModulator.cpp b/src/DabModulator.cpp
index cc2642a..0914469 100644
--- a/src/DabModulator.cpp
+++ b/src/DabModulator.cpp
@@ -183,7 +183,14 @@ int DabModulator::process(Buffer* dataOut)
         }
 
         auto cifOfdm = make_shared<OfdmGenerator>(
-                (1 + myNbSymbols), myNbCarriers, mySpacing);
+                (1 + myNbSymbols),
+                myNbCarriers,
+                mySpacing,
+                m_settings.enableCfr,
+                m_settings.cfrClip,
+                m_settings.cfrErrorClip);
+
+        rcs.enrol(cifOfdm.get());
 
         auto cifGain = make_shared<GainControl>(
                 mySpacing,
diff --git a/src/OfdmGenerator.cpp b/src/OfdmGenerator.cpp
index 20ba2b7..6a5044e 100644
--- a/src/OfdmGenerator.cpp
+++ b/src/OfdmGenerator.cpp
@@ -2,7 +2,7 @@
    Copyright (C) 2005, 2006, 2007, 2008, 2009, 2010, 2011 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
@@ -26,6 +26,8 @@
 
 #include "OfdmGenerator.h"
 #include "PcDebug.h"
+
+#include <complex>
 #include "fftw3.h"
 #define FFT_TYPE fftwf_complex
 
@@ -33,20 +35,28 @@
 #include <string.h>
 #include <stdexcept>
 #include <assert.h>
-#include <complex>
-typedef std::complex<float> complexf;
+#include <string>
+#include <numeric>
 
+static const size_t MAX_CLIP_STATS = 10;
 
 OfdmGenerator::OfdmGenerator(size_t nbSymbols,
-        size_t nbCarriers,
-        size_t spacing,
-        bool inverse) :
-    ModCodec(),
-    myFftPlan(NULL),
-    myFftIn(NULL), myFftOut(NULL),
+                             size_t nbCarriers,
+                             size_t spacing,
+                             bool enableCfr,
+                             float cfrClip,
+                             float cfrErrorClip,
+                             bool inverse) :
+    ModCodec(), RemoteControllable("ofdm"),
+    myFftPlan(nullptr),
+    myFftIn(nullptr), myFftOut(nullptr),
     myNbSymbols(nbSymbols),
     myNbCarriers(nbCarriers),
-    mySpacing(spacing)
+    mySpacing(spacing),
+    myCfr(enableCfr),
+    myCfrClip(cfrClip),
+    myCfrErrorClip(cfrErrorClip),
+    myCfrFft(nullptr)
 {
     PDEBUG("OfdmGenerator::OfdmGenerator(%zu, %zu, %zu, %s) @ %p\n",
             nbSymbols, nbCarriers, spacing, inverse ? "true" : "false", this);
@@ -56,6 +66,12 @@ OfdmGenerator::OfdmGenerator(size_t nbSymbols,
                 "OfdmGenerator::OfdmGenerator nbCarriers > spacing!");
     }
 
+    /* register the parameters that can be remote controlled */
+    RC_ADD_PARAMETER(cfr, "Enable crest factor reduction");
+    RC_ADD_PARAMETER(clip, "CFR: Clip to amplitude");
+    RC_ADD_PARAMETER(errorclip, "CFR: Limit error");
+    RC_ADD_PARAMETER(clip_stats, "CFR: statistics (clip ratio, errorclip ratio)");
+
     if (inverse) {
         myPosDst = (nbCarriers & 1 ? 0 : 1);
         myPosSrc = 0;
@@ -91,6 +107,12 @@ OfdmGenerator::OfdmGenerator(size_t nbSymbols,
             myFftIn, myFftOut,
             FFTW_BACKWARD, FFTW_MEASURE);
 
+    myCfrPostClip = (FFT_TYPE*)fftwf_malloc(sizeof(FFT_TYPE) * N);
+    myCfrPostFft = (FFT_TYPE*)fftwf_malloc(sizeof(FFT_TYPE) * N);
+    myCfrFft = fftwf_plan_dft_1d(N,
+            myCfrPostClip, myCfrPostFft,
+            FFTW_FORWARD, FFTW_MEASURE);
+
     if (sizeof(complexf) != sizeof(FFT_TYPE)) {
         printf("sizeof(complexf) %zu\n", sizeof(complexf));
         printf("sizeof(FFT_TYPE) %zu\n", sizeof(FFT_TYPE));
@@ -115,6 +137,10 @@ OfdmGenerator::~OfdmGenerator()
     if (myFftPlan) {
         fftwf_destroy_plan(myFftPlan);
     }
+
+    if (myCfrFft) {
+        fftwf_destroy_plan(myCfrFft);
+    }
 }
 
 int OfdmGenerator::process(Buffer* const dataIn, Buffer* dataOut)
@@ -147,6 +173,18 @@ int OfdmGenerator::process(Buffer* const dataIn, Buffer* dataOut)
                 "OfdmGenerator::process output size not valid!");
     }
 
+    // It is not guaranteed that fftw keeps the FFT input vector intact.
+    // That's why we copy it to the reference.
+    std::vector<complexf> reference;
+
+    std::vector<complexf> before_cfr;
+
+    size_t num_clip = 0;
+    size_t num_error_clip = 0;
+
+    // For performance reasons, do not calculate MER for every symbols.
+    myLastMERCalc = (myLastMERCalc + 1) % myNbSymbols;
+
     for (size_t i = 0; i < myNbSymbols; ++i) {
         myFftIn[0][0] = 0;
         myFftIn[0][1] = 0;
@@ -157,13 +195,217 @@ int OfdmGenerator::process(Buffer* const dataIn, Buffer* dataOut)
         memcpy(&myFftIn[myNegDst], &in[myNegSrc],
                 myNegSize * sizeof(FFT_TYPE));
 
-        fftwf_execute(myFftPlan);
+        if (myCfr) {
+            reference.resize(mySpacing);
+            memcpy(reference.data(), myFftIn, mySpacing * sizeof(FFT_TYPE));
+        }
+
+        fftwf_execute(myFftPlan); // IFFT from myFftIn to myFftOut
+
+        if (myCfr) {
+            if (myLastMERCalc == i) {
+                before_cfr.resize(mySpacing);
+                memcpy(before_cfr.data(), myFftOut, mySpacing * sizeof(FFT_TYPE));
+            }
+
+            complexf *symbol = reinterpret_cast<complexf*>(myFftOut);
+            /* cfr_one_iteration runs the myFftPlan again at the end, and
+             * therefore writes the output data to myFftOut.
+             */
+            const auto stat = cfr_one_iteration(symbol, reference.data());
+
+            if (myLastMERCalc == i) {
+                /* MER definition, ETSI ETR 290, Annex C
+                 *
+                 *                       \sum I^2 Q^2
+                 * MER[dB] = 10 log_10( -------------- )
+                 *                      \sum dI^2 dQ^2
+                 * Where I and Q are the ideal coordinates, and dI and dQ are the errors
+                 * in the received datapoints.
+                 *
+                 * In our case, we consider the constellation points given to the
+                 * OfdmGenerator as "ideal", and we compare the CFR output to it.
+                 */
+
+                double sum_iq = 0;
+                double sum_delta = 0;
+                for (size_t i = 0; i < mySpacing; i++) {
+                    sum_iq += std::norm(before_cfr[i]);
+                    sum_delta += std::norm(symbol[i] - before_cfr[i]);
+                }
+                const double mer = 10.0 * std::log10(sum_iq / sum_delta);
+                myMERs.push_back(mer);
+            }
+
+            num_clip += stat.clip_count;
+            num_error_clip += stat.errclip_count;
+        }
 
         memcpy(out, myFftOut, mySpacing * sizeof(FFT_TYPE));
 
         in += myNbCarriers;
         out += mySpacing;
     }
+
+    if (myCfr) {
+        std::lock_guard<std::mutex> lock(myCfrRcMutex);
+
+        const double num_samps = myNbSymbols * mySpacing;
+        const double clip_ratio = (double)num_clip / num_samps;
+
+        myClipRatios.push_back(clip_ratio);
+        while (myClipRatios.size() > MAX_CLIP_STATS) {
+            myClipRatios.pop_front();
+        }
+
+        const double errclip_ratio = (double)num_error_clip / num_samps;
+        myErrorClipRatios.push_back(errclip_ratio);
+        while (myErrorClipRatios.size() > MAX_CLIP_STATS) {
+            myErrorClipRatios.pop_front();
+        }
+
+        while (myMERs.size() > MAX_CLIP_STATS) {
+            myMERs.pop_front();
+        }
+    }
+
     return sizeOut;
 }
 
+OfdmGenerator::cfr_iter_stat_t OfdmGenerator::cfr_one_iteration(
+        complexf *symbol, const complexf *reference)
+{
+    // use std::norm instead of std::abs to avoid calculating the
+    // square roots
+    const float clip_squared = myCfrClip * myCfrClip;
+
+    OfdmGenerator::cfr_iter_stat_t ret;
+
+    // Clip
+    for (size_t i = 0; i < mySpacing; i++) {
+        const float mag_squared = std::norm(symbol[i]);
+        if (mag_squared > clip_squared) {
+            // normalise absolute value to myCfrClip:
+            // x_clipped = x * clip / |x|
+            //           = x * sqrt(clip_squared) / sqrt(mag_squared)
+            //           = x * sqrt(clip_squared / mag_squared)
+            symbol[i] *= std::sqrt(clip_squared / mag_squared);
+            ret.clip_count++;
+        }
+    }
+
+    // Take FFT of our clipped signal
+    memcpy(myCfrPostClip, symbol, mySpacing * sizeof(FFT_TYPE));
+    fftwf_execute(myCfrFft); // FFT from myCfrPostClip to myCfrPostFft
+
+    // Calculate the error in frequency domain by subtracting our reference
+    // and clip it to myCfrErrorClip. By adding this clipped error signal
+    // to our FFT output, we compensate the introduced error to some
+    // extent.
+    const float err_clip_squared = myCfrErrorClip * myCfrErrorClip;
+
+    std::vector<float> error_norm(mySpacing);
+
+    for (size_t i = 0; i < mySpacing; i++) {
+        // FFTW computes an unnormalised transform, i.e. a FFT-IFFT pair
+        // or vice-versa gives back the original vector scaled by a factor
+        // FFT-size. Because we're comparing our constellation point
+        // (calculated with IFFT-clip-FFT) against reference (input to
+        // the IFFT), we need to divide by our FFT size.
+        const complexf constellation_point =
+            reinterpret_cast<complexf*>(myCfrPostFft)[i] / (float)mySpacing;
+
+        complexf error = reference[i] - constellation_point;
+
+        const float mag_squared = std::norm(error);
+        error_norm[i] = mag_squared;
+
+        if (mag_squared > err_clip_squared) {
+            error *= std::sqrt(err_clip_squared / mag_squared);
+            ret.errclip_count++;
+        }
+
+        // Update the input to the FFT directly to avoid another copy for the
+        // subsequence IFFT
+        complexf *fft_in = reinterpret_cast<complexf*>(myFftIn);
+        fft_in[i] = constellation_point + error;
+    }
+
+    // Run our error-compensated symbol through the IFFT again
+    fftwf_execute(myFftPlan); // IFFT from myFftIn to myFftOut
+
+    return ret;
+}
+
+
+void OfdmGenerator::set_parameter(const std::string& parameter,
+                                  const std::string& value)
+{
+    using namespace std;
+    stringstream ss(value);
+    ss.exceptions ( stringstream::failbit | stringstream::badbit );
+
+    if (parameter == "cfr") {
+        ss >> myCfr;
+    }
+    else if (parameter == "clip") {
+        ss >> myCfrClip;
+    }
+    else if (parameter == "errorclip") {
+        ss >> myCfrErrorClip;
+    }
+    else if (parameter == "clip_stats") {
+        throw ParameterError("Parameter 'clip_stats' is read-only");
+    }
+    else {
+        stringstream ss;
+        ss << "Parameter '" << parameter
+            << "' is not exported by controllable " << get_rc_name();
+        throw ParameterError(ss.str());
+    }
+}
+
+const std::string OfdmGenerator::get_parameter(const std::string& parameter) const
+{
+    using namespace std;
+    stringstream ss;
+    if (parameter == "cfr") {
+        ss << myCfr;
+    }
+    else if (parameter == "clip") {
+        ss << std::fixed << myCfrClip;
+    }
+    else if (parameter == "errorclip") {
+        ss << std::fixed << myCfrErrorClip;
+    }
+    else if (parameter == "clip_stats") {
+        std::lock_guard<std::mutex> lock(myCfrRcMutex);
+        if (myClipRatios.empty() or myErrorClipRatios.empty() or myMERs.empty()) {
+            ss << "No stats available";
+        }
+        else {
+            const double avg_clip_ratio =
+                std::accumulate(myClipRatios.begin(), myClipRatios.end(), 0.0) /
+                myClipRatios.size();
+
+            const double avg_errclip_ratio =
+                std::accumulate(myErrorClipRatios.begin(), myErrorClipRatios.end(), 0.0) /
+                myErrorClipRatios.size();
+
+            const double avg_mer =
+                std::accumulate(myMERs.begin(), myMERs.end(), 0.0) /
+                myMERs.size();
+
+            ss << "Statistics : " << std::fixed <<
+                avg_clip_ratio * 100 << "%"" samples clipped, " <<
+                avg_errclip_ratio * 100 << "%"" errors clipped. " <<
+                "MER after CFR: " << avg_mer << " dB";
+        }
+    }
+    else {
+        ss << "Parameter '" << parameter <<
+            "' is not exported by controllable " << get_rc_name();
+        throw ParameterError(ss.str());
+    }
+    return ss.str();
+}
diff --git a/src/OfdmGenerator.h b/src/OfdmGenerator.h
index a8c3c19..f357fa6 100644
--- a/src/OfdmGenerator.h
+++ b/src/OfdmGenerator.h
@@ -2,7 +2,7 @@
    Copyright (C) 2005, 2006, 2007, 2008, 2009, 2010, 2011 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
@@ -32,38 +32,78 @@
 
 #include "porting.h"
 #include "ModPlugin.h"
-
+#include "RemoteControl.h"
 #include "fftw3.h"
-
 #include <sys/types.h>
+#include <vector>
+#include <complex>
 
+typedef std::complex<float> complexf;
 
-class OfdmGenerator : public ModCodec
+class OfdmGenerator : public ModCodec, public RemoteControllable
 {
-public:
-    OfdmGenerator(size_t nbSymbols, size_t nbCarriers, size_t spacing, bool inverse = true);
-    virtual ~OfdmGenerator();
-    OfdmGenerator(const OfdmGenerator&);
-    OfdmGenerator& operator=(const OfdmGenerator&);
-
-
-    int process(Buffer* const dataIn, Buffer* dataOut);
-    const char* name() { return "OfdmGenerator"; }
-
-protected:
-    fftwf_plan myFftPlan;
-    fftwf_complex *myFftIn, *myFftOut;
-    size_t myNbSymbols;
-    size_t myNbCarriers;
-    size_t mySpacing;
-    unsigned myPosSrc;
-    unsigned myPosDst;
-    unsigned myPosSize;
-    unsigned myNegSrc;
-    unsigned myNegDst;
-    unsigned myNegSize;
-    unsigned myZeroDst;
-    unsigned myZeroSize;
+    public:
+        OfdmGenerator(size_t nbSymbols,
+                      size_t nbCarriers,
+                      size_t spacing,
+                      bool enableCfr,
+                      float cfrClip,
+                      float cfrErrorClip,
+                      bool inverse = true);
+        virtual ~OfdmGenerator();
+        OfdmGenerator(const OfdmGenerator&) = delete;
+        OfdmGenerator& operator=(const OfdmGenerator&) = delete;
+
+        int process(Buffer* const dataIn, Buffer* dataOut) override;
+        const char* name() override { return "OfdmGenerator"; }
+
+        /* Functions for the remote control */
+        /* Base function to set parameters. */
+        virtual void set_parameter(
+                const std::string& parameter,
+                const std::string& value) override;
+
+        /* Getting a parameter always returns a string. */
+        virtual const std::string get_parameter(
+                const std::string& parameter) const override;
+
+    protected:
+        struct cfr_iter_stat_t {
+            size_t clip_count = 0;
+            size_t errclip_count = 0;
+        };
+
+        cfr_iter_stat_t cfr_one_iteration(
+                complexf *symbol, const complexf *reference);
+
+        fftwf_plan myFftPlan;
+        fftwf_complex *myFftIn, *myFftOut;
+        const size_t myNbSymbols;
+        const size_t myNbCarriers;
+        const size_t mySpacing;
+        unsigned myPosSrc;
+        unsigned myPosDst;
+        unsigned myPosSize;
+        unsigned myNegSrc;
+        unsigned myNegDst;
+        unsigned myNegSize;
+        unsigned myZeroDst;
+        unsigned myZeroSize;
+
+        bool myCfr; // Whether to enable crest factor reduction
+        mutable std::mutex myCfrRcMutex;
+        float myCfrClip;
+        float myCfrErrorClip;
+        fftwf_plan myCfrFft;
+        fftwf_complex *myCfrPostClip;
+        fftwf_complex *myCfrPostFft;
+
+        // Statistics for CFR
+        std::deque<double> myClipRatios;
+        std::deque<double> myErrorClipRatios;
+
+        size_t myLastMERCalc = 0;
+        std::deque<double> myMERs;
 };
 
 
diff --git a/src/OutputUHDFeedback.cpp b/src/OutputUHDFeedback.cpp
index b370885..056be29 100644
--- a/src/OutputUHDFeedback.cpp
+++ b/src/OutputUHDFeedback.cpp
@@ -46,6 +46,7 @@ DESCRIPTION:
 #include <boost/date_time/posix_time/posix_time.hpp>
 #include "OutputUHDFeedback.h"
 #include "Utils.h"
+#include "Socket.h"
 
 using namespace std;
 typedef std::complex<float> complexf;
@@ -87,7 +88,9 @@ void OutputUHDFeedback::set_tx_frame(
 {
     boost::mutex::scoped_lock lock(burstRequest.mutex);
 
-    assert(buf.size() % sizeof(complexf) == 0);
+    if (buf.size() % sizeof(complexf) != 0) {
+        throw std::logic_error("Buffer for tx frame has incorrect size");
+    }
 
     if (burstRequest.state == BurstRequestState::SaveTransmitFrame) {
         const size_t n = std::min(
@@ -183,85 +186,23 @@ void OutputUHDFeedback::ReceiveBurstThread()
     }
 }
 
-static int accept_with_timeout(int server_socket, int timeout_ms, struct sockaddr_in *client)
-{
-    struct pollfd fds[1];
-    fds[0].fd = server_socket;
-    fds[0].events = POLLIN | POLLOUT;
-
-    int retval = poll(fds, 1, timeout_ms);
-
-    if (retval == -1) {
-        throw std::runtime_error("TCP Socket accept error: " + to_string(errno));
-    }
-    else if (retval) {
-        socklen_t client_len = sizeof(struct sockaddr_in);
-        return accept(server_socket, (struct sockaddr*)&client, &client_len);
-    }
-    else {
-        return -2;
-    }
-}
-
-static ssize_t sendall(int socket, const void *buffer, size_t buflen)
-{
-    uint8_t *buf = (uint8_t*)buffer;
-    while (buflen > 0) {
-        ssize_t sent = send(socket, buf, buflen, 0);
-        if (sent < 0) {
-            return -1;
-        }
-        else {
-            buf += sent;
-            buflen -= sent;
-        }
-    }
-    return buflen;
-}
-
 void OutputUHDFeedback::ServeFeedback()
 {
-    if ((m_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);
-
-    const int reuse = 1;
-    if (setsockopt(m_server_sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse))
-            < 0) {
-        throw std::runtime_error("Can't reuse address for TCP socket");
-    }
-
-    if (bind(m_server_sock, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
-        close(m_server_sock);
-        throw std::runtime_error("Can't bind TCP socket");
-    }
-
-    if (listen(m_server_sock, 1) < 0) {
-        close(m_server_sock);
-        throw std::runtime_error("Can't listen TCP socket");
-    }
+    TCPSocket m_server_sock;
+    m_server_sock.listen(m_port);
 
     etiLog.level(info) << "DPD Feedback server listening on port " << m_port;
 
     while (m_running) {
         struct sockaddr_in client;
-        int client_sock = accept_with_timeout(m_server_sock, 1000, &client);
+        TCPSocket client_sock = m_server_sock.accept_with_timeout(1000, &client);
 
-        if (client_sock == -1) {
-            close(m_server_sock);
+        if (not client_sock.valid()) {
             throw runtime_error("Could not establish new connection");
         }
-        else if (client_sock == -2) {
-            continue;
-        }
 
         uint8_t request_version = 0;
-        ssize_t read = recv(client_sock, &request_version, 1, 0);
+        ssize_t read = client_sock.recv(&request_version, 1, 0);
         if (!read) break; // done reading
         if (read < 0) {
             etiLog.level(info) <<
@@ -275,7 +216,7 @@ void OutputUHDFeedback::ServeFeedback()
         }
 
         uint32_t num_samples = 0;
-        read = recv(client_sock, &num_samples, 4, 0);
+        read = client_sock.recv(&num_samples, 4, 0);
         if (!read) break; // done reading
         if (read < 0) {
             etiLog.level(info) <<
@@ -308,13 +249,13 @@ void OutputUHDFeedback::ServeFeedback()
                     burstRequest.rx_samples.size() / sizeof(complexf)));
 
         uint32_t num_samples_32 = burstRequest.num_samples;
-        if (sendall(client_sock, &num_samples_32, sizeof(num_samples_32)) < 0) {
+        if (client_sock.sendall(&num_samples_32, sizeof(num_samples_32)) < 0) {
             etiLog.level(info) <<
                 "DPD Feedback Server Client send num_samples failed";
             break;
         }
 
-        if (sendall(client_sock,
+        if (client_sock.sendall(
                     &burstRequest.tx_second,
                     sizeof(burstRequest.tx_second)) < 0) {
             etiLog.level(info) <<
@@ -322,7 +263,7 @@ void OutputUHDFeedback::ServeFeedback()
             break;
         }
 
-        if (sendall(client_sock,
+        if (client_sock.sendall(
                     &burstRequest.tx_pps,
                     sizeof(burstRequest.tx_pps)) < 0) {
             etiLog.level(info) <<
@@ -332,8 +273,11 @@ void OutputUHDFeedback::ServeFeedback()
 
         const size_t frame_bytes = burstRequest.num_samples * sizeof(complexf);
 
-        assert(burstRequest.tx_samples.size() >= frame_bytes);
-        if (sendall(client_sock,
+        if (burstRequest.tx_samples.size() < frame_bytes) {
+            throw logic_error("DPD Feedback burstRequest invalid: not enough TX samples");
+        }
+
+        if (client_sock.sendall(
                     &burstRequest.tx_samples[0],
                     frame_bytes) < 0) {
             etiLog.level(info) <<
@@ -341,7 +285,7 @@ void OutputUHDFeedback::ServeFeedback()
             break;
         }
 
-        if (sendall(client_sock,
+        if (client_sock.sendall(
                     &burstRequest.rx_second,
                     sizeof(burstRequest.rx_second)) < 0) {
             etiLog.level(info) <<
@@ -349,7 +293,7 @@ void OutputUHDFeedback::ServeFeedback()
             break;
         }
 
-        if (sendall(client_sock,
+        if (client_sock.sendall(
                     &burstRequest.rx_pps,
                     sizeof(burstRequest.rx_pps)) < 0) {
             etiLog.level(info) <<
@@ -357,16 +301,17 @@ void OutputUHDFeedback::ServeFeedback()
             break;
         }
 
-        assert(burstRequest.rx_samples.size() >= frame_bytes);
-        if (sendall(client_sock,
+        if (burstRequest.rx_samples.size() < frame_bytes) {
+            throw logic_error("DPD Feedback burstRequest invalid: not enough RX samples");
+        }
+
+        if (client_sock.sendall(
                     &burstRequest.rx_samples[0],
                     frame_bytes) < 0) {
             etiLog.level(info) <<
                 "DPD Feedback Server Client send rx_frame failed";
             break;
         }
-
-        close(client_sock);
     }
 }
 
@@ -392,11 +337,6 @@ void OutputUHDFeedback::ServeFeedbackThread()
     }
 
     m_running = false;
-
-    if (m_server_sock != -1) {
-        close(m_server_sock);
-        m_server_sock = -1;
-    }
 }
 
 #endif
diff --git a/src/OutputUHDFeedback.h b/src/OutputUHDFeedback.h
index c68f4c2..da0d487 100644
--- a/src/OutputUHDFeedback.h
+++ b/src/OutputUHDFeedback.h
@@ -109,7 +109,6 @@ class OutputUHDFeedback {
         UHDReceiveBurstRequest burstRequest;
 
         std::atomic_bool m_running;
-        int m_server_sock = -1;
         uint16_t m_port = 0;
         uint32_t m_sampleRate = 0;
         uhd::usrp::multi_usrp::sptr m_usrp;
diff --git a/src/Socket.h b/src/Socket.h
new file mode 100644
index 0000000..1d9c252
--- /dev/null
+++ b/src/Socket.h
@@ -0,0 +1,164 @@
+/*
+   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:
+   Abstraction for sockets.
+*/
+
+/*
+   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
+
+#include <unistd.h>
+#include <cstdint>
+#include <stdexcept>
+#include <sys/socket.h>
+#include <netinet/ip.h>
+#include <errno.h>
+#include <poll.h>
+
+class TCPSocket {
+    public:
+        TCPSocket() {
+            if ((m_sock = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
+                throw std::runtime_error("Can't create TCP socket");
+            }
+        }
+
+        ~TCPSocket() {
+            if (m_sock != -1) {
+                ::close(m_sock);
+            }
+        }
+
+        TCPSocket(const TCPSocket& other) = delete;
+        TCPSocket& operator=(const TCPSocket& other) = delete;
+        TCPSocket(TCPSocket&& other) {
+            m_sock = other.m_sock;
+
+            if (other.m_sock != -1) {
+                other.m_sock = -1;
+            }
+        }
+
+        TCPSocket& operator=(TCPSocket&& other)
+        {
+            m_sock = other.m_sock;
+
+            if (other.m_sock != -1) {
+                other.m_sock = -1;
+            }
+
+            return *this;
+        }
+
+        bool valid(void) const {
+            return m_sock != -1;
+        }
+
+        void listen(int port) {
+            struct sockaddr_in addr;
+            addr.sin_family = AF_INET;
+            addr.sin_port = htons(port);
+            addr.sin_addr.s_addr = htonl(INADDR_ANY);
+
+            const int reuse = 1;
+            if (setsockopt(m_sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)) < 0) {
+                throw std::runtime_error("Can't reuse address for TCP socket");
+            }
+
+            if (bind(m_sock, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
+                close();
+                throw std::runtime_error("Can't bind TCP socket");
+            }
+
+            if (::listen(m_sock, 1) < 0) {
+                close();
+                m_sock = -1;
+                throw std::runtime_error("Can't listen TCP socket");
+            }
+
+        }
+
+        void close(void) {
+            ::close(m_sock);
+            m_sock = -1;
+        }
+
+        TCPSocket accept_with_timeout(int timeout_ms, struct sockaddr_in *client)
+        {
+            struct pollfd fds[1];
+            fds[0].fd = m_sock;
+            fds[0].events = POLLIN | POLLOUT;
+
+            int retval = poll(fds, 1, timeout_ms);
+
+            if (retval == -1) {
+                throw std::runtime_error("TCP Socket accept error: " + std::to_string(errno));
+            }
+            else if (retval) {
+                socklen_t client_len = sizeof(struct sockaddr_in);
+                int sockfd = accept(m_sock, (struct sockaddr*)&client, &client_len);
+                TCPSocket s(sockfd);
+                return s;
+            }
+            else {
+                TCPSocket s(-1);
+                return s;
+            }
+        }
+
+        ssize_t sendall(const void *buffer, size_t buflen)
+        {
+            uint8_t *buf = (uint8_t*)buffer;
+            while (buflen > 0) {
+                ssize_t sent = send(m_sock, buf, buflen, 0);
+                if (sent < 0) {
+                    return -1;
+                }
+                else {
+                    buf += sent;
+                    buflen -= sent;
+                }
+            }
+            return buflen;
+        }
+
+        ssize_t recv(void *buffer, size_t length, int flags)
+        {
+            return ::recv(m_sock, buffer, length, flags);
+        }
+
+    private:
+        explicit TCPSocket(int sockfd) {
+            m_sock = sockfd;
+        }
+
+        int m_sock = -1;
+};
+