3 files changed, 55 insertions, 15 deletions

diff --git a/src/DabMod.cpp b/src/DabMod.cpp
index 3548f9d..2489797 100644
--- a/src/DabMod.cpp
+++ b/src/DabMod.cpp
@@ -866,6 +866,10 @@ run_modulator_state run_modulator(modulator_data& m)
             running = 0;
             ret = MOD_NORMAL_END;
         }
+    } catch (fct_discontinuity_error& e) {
+        // The OutputUHD saw a FCT discontinuity
+        fprintf(stderr, "Stream discontinuity\n");
+        ret = MOD_AGAIN;
     } catch (std::overflow_error& e) {
         // The ZeroMQ input has overflowed its buffer
         fprintf(stderr, "overflow error: %s\n", e.what());
diff --git a/src/OutputUHD.cpp b/src/OutputUHD.cpp
index 54acf40..a877161 100644
--- a/src/OutputUHD.cpp
+++ b/src/OutputUHD.cpp
@@ -230,8 +230,6 @@ OutputUHD::OutputUHD(
     mySyncBarrier = b;
     uwd.sync_barrier = b;
 
-    worker.start(&uwd);
-
     MDEBUG("OutputUHD:UHD ready.\n");
 }
 
@@ -288,6 +286,8 @@ int OutputUHD::process(Buffer* dataIn, Buffer* dataOut)
     if (first_run) {
         myLogger.level(debug) << "OutputUHD: UHD initialising...";
 
+        worker.start(&uwd);
+
         uwd.bufsize = dataIn->getLength();
         uwd.frame0.buf = malloc(uwd.bufsize);
         uwd.frame1.buf = malloc(uwd.bufsize);
@@ -332,6 +332,19 @@ int OutputUHD::process(Buffer* dataIn, Buffer* dataOut)
         }
         mySyncBarrier.get()->wait();
 
+        if (!uwd.running) {
+            worker.stop();
+            first_run = true;
+            if (uwd.failed_due_to_fct) {
+                throw fct_discontinuity_error();
+            }
+            else {
+                myLogger.level(error) <<
+                    "OutputUHD: Error, UHD worker failed";
+                throw std::runtime_error("UHD worker failed");
+            }
+        }
+
         // write into the our buffer while
         // the worker sends the other.
 
@@ -374,6 +387,21 @@ int OutputUHD::process(Buffer* dataIn, Buffer* dataOut)
 
 }
 
+void UHDWorker::process_errhandler()
+{
+    try {
+        process();
+    }
+    catch (fct_discontinuity_error& e) {
+        uwd->logger->level(warn) << e.what();
+        uwd->failed_due_to_fct = true;
+    }
+
+    uwd->running = false;
+    uwd->sync_barrier.get()->wait();
+    uwd->logger->level(warn) << "UHD worker terminated";
+}
+
 void UHDWorker::process()
 {
     int workerbuffer  = 0;
@@ -409,7 +437,7 @@ void UHDWorker::process()
 
     int expected_next_fct = -1;
 
-    while (running) {
+    while (uwd->running) {
         bool fct_discontinuity = false;
         md.has_time_spec = false;
         md.time_spec = uhd::time_spec_t(0.0);
@@ -449,6 +477,7 @@ void UHDWorker::process()
                     "OutputUHD: Incorrect expect fct " << frame->ts.fct;
 
                 fct_discontinuity = true;
+                throw fct_discontinuity_error();
             }
         }
 
@@ -553,7 +582,7 @@ void UHDWorker::process()
         PDEBUG("UHDWorker::process:max_num_samps: %zu.\n",
                 usrp_max_num_samps);
 
-        while (running && !uwd->muting && (num_acc_samps < sizeIn)) {
+        while (uwd->running && !uwd->muting && (num_acc_samps < sizeIn)) {
             size_t samps_to_send = std::min(sizeIn - num_acc_samps, usrp_max_num_samps);
 
             //ensure the the last packet has EOB set if the timestamps has been
@@ -665,8 +694,6 @@ loopend:
         // swap buffers
         workerbuffer = (workerbuffer + 1) % 2;
     }
-
-    uwd->logger->level(warn) << "UHD worker terminated";
 }
 
 
diff --git a/src/OutputUHD.h b/src/OutputUHD.h
index c5d561b..d92c7a4 100644
--- a/src/OutputUHD.h
+++ b/src/OutputUHD.h
@@ -83,9 +83,20 @@ struct UHDWorkerFrameData {
     struct frame_timestamp ts;
 };
 
+struct fct_discontinuity_error : public std::exception
+{
+  const char* what () const throw ()
+  {
+    return "FCT discontinuity detected";
+  }
+};
+
 enum refclk_lock_loss_behaviour_t { CRASH, IGNORE };
 
 struct UHDWorkerData {
+    bool running;
+    bool failed_due_to_fct;
+
 #if FAKE_UHD == 0
     uhd::usrp::multi_usrp::sptr myUsrp;
 #endif
@@ -130,28 +141,26 @@ struct UHDWorkerData {
 
 class UHDWorker {
     public:
-        UHDWorker () {
-            running = false;
-        }
-
         void start(struct UHDWorkerData *uhdworkerdata) {
-            running = true;
             uwd = uhdworkerdata;
-            uhd_thread = boost::thread(&UHDWorker::process, this);
+
+            uwd->running = true;
+            uwd->failed_due_to_fct = false;
+            uhd_thread = boost::thread(&UHDWorker::process_errhandler, this);
         }
 
         void stop() {
-            running = false;
+            uwd->running = false;
             uhd_thread.interrupt();
             uhd_thread.join();
         }
 
+    private:
         void process();
+        void process_errhandler();
 
 
-    private:
         struct UHDWorkerData *uwd;
-        bool running;
         boost::thread uhd_thread;
 
         uhd::tx_streamer::sptr myTxStream;