tools: Added Balints kitchen sink debug util

3 files changed, 1880 insertions, 0 deletions

diff --git a/tools/kitchen_sink/.gitignore b/tools/kitchen_sink/.gitignore
new file mode 100644
index 000000000..567609b12
--- /dev/null
+++ b/tools/kitchen_sink/.gitignore
@@ -0,0 +1 @@
+build/
diff --git a/tools/kitchen_sink/CMakeLists.txt b/tools/kitchen_sink/CMakeLists.txt
new file mode 100644
index 000000000..19e7f99fc
--- /dev/null
+++ b/tools/kitchen_sink/CMakeLists.txt
@@ -0,0 +1,68 @@
+#
+# Copyright 2010-2012 Ettus Research LLC
+#
+# This program 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.
+#
+# This program 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 this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+
+cmake_minimum_required(VERSION 2.8)
+
+### Set up build environment ##################################################
+#### Set up Boost:
+set(BOOST_REQUIRED_COMPONENTS
+    date_time
+    filesystem
+    program_options
+    regex
+    system
+    thread
+    unit_test_framework
+)
+IF(UNIX AND EXISTS "/usr/lib64")
+    LIST(APPEND BOOST_LIBRARYDIR "/usr/lib64") #fedora 64-bit fix
+ENDIF(UNIX AND EXISTS "/usr/lib64")
+
+IF(MSVC)
+    SET(BOOST_ALL_DYN_LINK "${BOOST_ALL_DYN_LINK}" CACHE BOOL "boost enable dynamic linking")
+    IF(BOOST_ALL_DYN_LINK)
+        ADD_DEFINITIONS(-DBOOST_ALL_DYN_LINK) #setup boost auto-linking in msvc
+    ELSE(BOOST_ALL_DYN_LINK)
+        UNSET(BOOST_REQUIRED_COMPONENTS) #empty components list for static link
+    ENDIF(BOOST_ALL_DYN_LINK)
+ENDIF(MSVC)
+
+SET(Boost_ADDITIONAL_VERSIONS
+    "1.35.0" "1.35" "1.36.0" "1.36" "1.37.0" "1.37" "1.38.0" "1.38" "1.39.0" "1.39"
+    "1.40.0" "1.40" "1.41.0" "1.41" "1.42.0" "1.42" "1.43.0" "1.43" "1.44.0" "1.44"
+    "1.45.0" "1.45" "1.46.0" "1.46" "1.47.0" "1.47" "1.48.0" "1.48" "1.49.0" "1.49"
+    "1.50.0" "1.50" "1.51.0" "1.51" "1.52.0" "1.52" "1.53.0" "1.53" "1.54.0" "1.54"
+    "1.55.0" "1.55" "1.56.0" "1.56" "1.57.0" "1.57" "1.58.0" "1.58" "1.59.0" "1.59"
+    "1.60.0" "1.60" "1.61.0" "1.61" "1.62.0" "1.62" "1.63.0" "1.63" "1.64.0" "1.64"
+    "1.65.0" "1.65" "1.66.0" "1.66" "1.67.0" "1.67" "1.68.0" "1.68" "1.69.0" "1.69"
+)
+FIND_PACKAGE(Boost 1.36 COMPONENTS ${BOOST_REQUIRED_COMPONENTS})
+
+#### Set up UHD:
+find_package(UHD "3.7.3" REQUIRED)
+
+### Configure Compiler ########################################################
+include_directories(
+    ${Boost_INCLUDE_DIRS}
+    ${UHD_INCLUDE_DIRS}
+)
+
+### Make the executable #######################################################
+add_executable(kitchen_sink kitchen_sink.cpp)
+target_link_libraries(kitchen_sink ${UHD_LIBRARIES} ${Boost_LIBRARIES})
+
+# Skip installing.
diff --git a/tools/kitchen_sink/kitchen_sink.cpp b/tools/kitchen_sink/kitchen_sink.cpp
new file mode 100644
index 000000000..33fab4c86
--- /dev/null
+++ b/tools/kitchen_sink/kitchen_sink.cpp
@@ -0,0 +1,1811 @@
+//
+// Copyright 2011-2014 Ettus Research LLC
+//
+// This program 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.
+//
+// This program 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 this program.  If not, see <http://www.gnu.org/licenses/>.
+//
+
+#include <uhd/utils/thread_priority.hpp>
+#include <uhd/convert.hpp>
+#include <uhd/utils/safe_main.hpp>
+#include <uhd/usrp/multi_usrp.hpp>
+#include <boost/program_options.hpp>
+#include <boost/format.hpp>
+#include <boost/thread/thread.hpp>
+#include <boost/algorithm/string.hpp>
+#include <boost/lexical_cast.hpp>
+#include <iostream>
+#include <complex>
+#include <cstdlib>
+//#include <curses.h>
+#include <termios.h>
+#include <fstream>
+#include <stdint.h>
+
+#include <boost/thread/mutex.hpp>
+#include <boost/thread/condition_variable.hpp>
+#include <csignal>
+#include <uhd/utils/msg.hpp>
+
+namespace po = boost::program_options;
+
+//#define HAS_RX_METADATA_OUT_OF_SEQUENCE
+
+#define COLOUR_START        "\033["
+#define COLOUR_END          "m"
+#define COLOUR_RESET        COLOUR_START"0"COLOUR_END
+#define COLOUR_BLACK        "0"
+#define COLOUR_RED          "1"
+#define COLOUR_GREEN        "2"
+#define COLOUR_YELLOW       "3"
+#define COLOUR_BLUE         "4"
+#define COLOUR_MAGENTA      "5"
+#define COLOUR_CYAN         "6"
+#define COLOUR_WHITE        "7"
+#define COLOUR_LOW          "3"
+#define COLOUR_HIGH         "9"
+#define COLOUR_BACK         "4"
+#define COLOUR_BACK_HIGH    "10"
+#define COLOUR_BOLD         "1"
+#define COLOUR_UNDERLINE    "4"
+
+#define HEADER              "[  ] "
+#define HEADER_TX           "[" COLOUR_START COLOUR_HIGH COLOUR_RED   COLOUR_END "TX" COLOUR_RESET "] "
+#define HEADER_RX           "[" COLOUR_START COLOUR_HIGH COLOUR_GREEN COLOUR_END "RX" COLOUR_RESET "] "
+#define HEADER_AS           "[" COLOUR_START COLOUR_HIGH COLOUR_BLUE COLOUR_END "AS" COLOUR_RESET "] "
+#define HEADER_ERROR        "[" COLOUR_START COLOUR_BACK_HIGH COLOUR_RED COLOUR_END "!!" COLOUR_RESET "] "
+#define HEADER_WARN         "[" COLOUR_START COLOUR_BACK_HIGH COLOUR_YELLOW COLOUR_END "**" COLOUR_RESET "] "
+
+#define TICKS_PER_SEC       boost::posix_time::time_duration::ticks_per_second()
+
+/***********************************************************************
+ * Test result variables
+ **********************************************************************/
+unsigned long long num_overflows = 0;
+unsigned long long num_underflows = 0;
+unsigned long long num_underflows_in_packet = 0;
+unsigned long long num_rx_samps = 0;
+unsigned long long num_tx_samps = 0;
+unsigned long long num_dropped_samps = 0;
+unsigned long long num_seq_errors = 0;
+unsigned long long num_seq_errors_in_burst = 0;
+unsigned long long num_tx_acks = 0;
+unsigned long long num_late_packets = 0;
+unsigned long long num_late_packets_msg = 0;
+unsigned long long num_send_calls = 0;
+
+static boost::condition_variable rx_thread_complete, tx_thread_complete, tx_async_thread_complete;
+static boost::condition_variable begin, abort_event, rx_thread_begin, tx_thread_begin, recv_done/*, tx_async_begin*/;
+static uhd::rx_metadata_t last_rx_md;
+static size_t last_rx_samps;
+static boost::mutex last_rx_md_mutex, begin_rx_mutex, begin_tx_mutex, begin_tx_async_begin, stop_mutex;
+static volatile bool running = false;
+static volatile boost::this_thread::disable_interruption *rx_interrupt_disabler, *tx_interrupt_disabler, *tx_async_interrupt_disabler;
+static bool rx_thread_finished = false, tx_thread_finished = false, tx_async_thread_finished = false;
+
+static bool stop_signal_called = false;
+
+static void sig_int_handler(int signal)
+{
+    boost::mutex::scoped_lock l(stop_mutex);
+    running = false;
+    stop_signal_called = true;
+    abort_event.notify_all();
+}
+
+typedef std::map<char,size_t> msg_count_map_t;
+
+static boost::mutex msg_count_map_mutex;
+static msg_count_map_t msg_count_map;
+static boost::system_time start_time, last_msg_print_time;
+static double msg_print_interval = 0.0;
+
+// Derived from: http://cc.byexamples.com/2007/04/08/non-blocking-user-input-in-loop-without-ncurses/
+static bool kbhit(size_t timeout = 0/*ms*/)
+{
+    struct timeval tv;
+    fd_set fds;
+    tv.tv_sec = timeout / 1000;
+    tv.tv_usec = (timeout % 1000) * 1000;
+    FD_ZERO(&fds);
+    FD_SET(STDIN_FILENO, &fds);
+    select(STDIN_FILENO+1, &fds, NULL, NULL, &tv);
+    return (FD_ISSET(0, &fds));
+}
+
+static void set_nonblock(bool enable)
+{
+    struct termios ttystate;
+
+    //get the terminal state
+    tcgetattr(STDIN_FILENO, &ttystate);
+
+    if (enable)
+    {
+        //turn off canonical mode
+        ttystate.c_lflag &= ~ICANON;
+        //minimum of number input read.
+        ttystate.c_cc[VMIN] = 1;
+    }
+    else
+    {
+        //turn on canonical mode
+        ttystate.c_lflag |= ICANON;
+    }
+    //set the terminal attributes.
+    tcsetattr(STDIN_FILENO, TCSANOW, &ttystate);
+}
+
+static std::string get_stringified_time(struct timeval* tv = NULL)
+{
+    struct timeval _tv;
+    if (tv == NULL)
+    {
+        tv = &_tv;
+        gettimeofday(tv, NULL); // FIXME: Use boost::posix_time
+    }
+    time_t t = tv->tv_sec;
+    struct tm *lt = localtime(&t);
+    char s[20] = { 0 };
+    strftime(s, sizeof(s), "%Y/%m/%d %H:%M:%S", lt);
+    return
+        std::string(COLOUR_START COLOUR_UNDERLINE COLOUR_END) +
+        std::string(s) +
+        boost::str(boost::format(".%06ld") % tv->tv_usec) +
+        std::string(COLOUR_RESET)
+    ;
+}
+
+static void _select_msg_colour(char c, std::stringstream& ss)
+{
+    switch (c)
+    {
+        case 'U':
+        case 'L':
+            ss << COLOUR_RED;
+            break;
+        case 'S':
+        case 'O':
+        case 'D':
+            ss << COLOUR_GREEN;
+            break;
+        default:
+            ss << COLOUR_BLACK;
+    }
+}
+
+static void print_msgs(void)
+{
+    if (msg_print_interval <= 0.0)
+        return;
+
+    boost::system_time time_now = boost::get_system_time();
+    boost::posix_time::time_duration update_diff = time_now - last_msg_print_time;
+    if (((double)update_diff.ticks() / (double)TICKS_PER_SEC) >= msg_print_interval)
+    {
+        boost::mutex::scoped_lock l(msg_count_map_mutex);
+
+        if (msg_count_map.size() > 0)
+        {
+            std::stringstream ss;
+
+            ss << HEADER_WARN "(" << get_stringified_time() << ") ";
+
+            for (msg_count_map_t::iterator it = msg_count_map.begin(); it != msg_count_map.end(); ++it)
+            {
+                if (it != msg_count_map.begin())
+                    ss << ", ";
+                ss << COLOUR_START COLOUR_HIGH;
+                _select_msg_colour(it->first, ss);
+                ss << COLOUR_END;
+                ss << it->first;
+                ss << COLOUR_RESET;
+                ss << boost::str(boost::format(": %05d") % it->second);
+            }
+
+            std::cout << ss.str() << std::endl << std::flush;
+
+            last_msg_print_time = time_now;
+            msg_count_map.clear();
+        }
+    }
+}
+
+static void msg_handler(uhd::msg::type_t type, const std::string& msg)
+{
+    if ((type == uhd::msg::fastpath) && (msg.size() == 1))
+    {
+        char c = msg.c_str()[0];
+
+        if (c == 'L')
+            ++num_late_packets_msg;
+
+        if (msg_print_interval <= 0.0)
+        {
+            std::stringstream ss;
+
+            ss << COLOUR_START COLOUR_BACK;
+
+            _select_msg_colour(c, ss);
+
+            ss << ";" COLOUR_HIGH COLOUR_WHITE COLOUR_END;
+            ss << msg;
+            ss << COLOUR_RESET;
+
+            std::cout << ss.str() << std::flush;
+        }
+        else
+        {
+            {
+                boost::mutex::scoped_lock l(msg_count_map_mutex);
+
+                if (msg_count_map.find(c) == msg_count_map.end())
+                    msg_count_map[c] = 1;
+                else
+                    msg_count_map[c] += 1;
+            }
+
+            print_msgs();
+        }
+    }
+    else
+        std::cout << msg << std::flush;
+}
+
+/***********************************************************************
+ * Benchmark RX Rate
+ **********************************************************************/
+
+typedef struct RxParams {
+    size_t samps_per_packet;
+    size_t samps_per_buff;
+    uhd::time_spec_t start_time;
+    double start_time_delay;
+    double recv_timeout;
+    bool one_packet_at_a_time;
+    bool check_recv_time;
+    double progress_interval;
+    bool single_packets;
+    bool size_map;
+    size_t rx_sample_limit;
+    std::ofstream* capture_file;
+    bool set_rx_freq;
+    double rx_freq;
+    double rx_freq_delay;
+    double rx_lo_offset;
+    bool interleave_rx_file_samples;
+} RX_PARAMS;
+
+static uint64_t recv_samp_count_progress = 0;
+static boost::system_time recv_samp_count_progress_update;
+static size_t rx_sleep_delay_now = 0;
+
+void benchmark_rx_rate(
+    uhd::usrp::multi_usrp::sptr usrp,
+    const std::string &rx_cpu,
+    uhd::rx_streamer::sptr rx_stream,
+    RX_PARAMS& params)
+{
+    uhd::set_thread_priority_safe();
+
+    boost::mutex::scoped_lock l(begin_rx_mutex);
+
+    rx_interrupt_disabler = new boost::this_thread::disable_interruption();
+
+    //setup variables and allocate buffer
+    size_t bytes_per_samp = uhd::convert::get_bytes_per_item(rx_cpu);
+    std::vector<char> buff(params.samps_per_buff * bytes_per_samp * rx_stream->get_num_channels());
+    std::vector<void *> buffs;
+    for (size_t ch = 0; ch < rx_stream->get_num_channels(); ch++)
+        buffs.push_back(&buff.front() + (params.samps_per_buff * bytes_per_samp * ch)); //same buffer for each channel
+
+    bool had_an_overflow = false;
+    uhd::time_spec_t last_time;
+    const double rate = usrp->get_rx_rate();
+    const double master_clock_rate = usrp->get_master_clock_rate();
+    uhd::rx_metadata_t md;
+
+    if (params.set_rx_freq)
+    {
+        if (params.rx_freq_delay == 0)
+        {
+            std::cout << boost::format(HEADER_RX"Setting RX freq: %f (LO offset: %f Hz)") % params.rx_freq % params.rx_lo_offset << std::endl;
+
+            uhd::tune_request_t tune_request = uhd::tune_request_t(params.rx_freq, params.rx_lo_offset);
+            for (size_t ch = 0; ch < rx_stream->get_num_channels(); ch++)
+                usrp->set_rx_freq(tune_request, ch);
+        }
+    }
+
+    std::cout << HEADER_RX"Waiting..." << std::endl;
+    rx_thread_begin.notify_all();
+    begin.wait(l);
+    l.unlock();
+
+    {
+        std::stringstream ss;
+        ss << HEADER_RX"(" << get_stringified_time() << ") Running..." << std::endl;
+        std::cout << ss.str();
+    }
+
+    uhd::time_spec_t time_now = usrp->get_time_now();
+    uhd::time_spec_t diff = time_now - params.start_time;
+    std::cout << boost::format(HEADER_RX"USRP time difference between right now and start time: %ld ticks (%f seconds)") % diff.to_ticks(rate) % diff.get_real_secs() << std::endl;
+
+    uhd::time_spec_t actual_start_time = params.start_time + uhd::time_spec_t(params.start_time_delay);
+
+    if (params.start_time_delay >= 0.0)
+        std::cout << HEADER_RX"Will begin streaming at time " << boost::format("%.6f") % actual_start_time.get_real_secs() << std::endl;
+
+    uhd::stream_cmd_t cmd((params.rx_sample_limit == 0) ? uhd::stream_cmd_t::STREAM_MODE_START_CONTINUOUS : uhd::stream_cmd_t::STREAM_MODE_NUM_SAMPS_AND_DONE); // FIXME: The other streaming modes
+    cmd.num_samps = params.rx_sample_limit;
+    cmd.time_spec = actual_start_time;
+    cmd.stream_now = (params.start_time_delay == 0.0);
+    rx_stream->issue_stream_cmd(cmd);
+
+    if (params.set_rx_freq)
+    {
+        if (params.rx_freq_delay > 0)
+        {
+            std::cout << boost::format(HEADER_RX"Scheduling RX freq in %d seconds: %f (LO offset: %f Hz)") % params.rx_freq_delay % params.rx_freq % params.rx_lo_offset << std::endl;
+
+            uhd::time_spec_t tune_time = params.start_time + uhd::time_spec_t(params.rx_freq_delay);
+            usrp->set_command_time(tune_time);
+
+            uhd::tune_request_t tune_request = uhd::tune_request_t(params.rx_freq, params.rx_lo_offset);
+            for (size_t ch = 0; ch < rx_stream->get_num_channels(); ch++)
+                usrp->set_rx_freq(tune_request, ch);
+
+            usrp->clear_command_time();
+        }
+    }
+
+    double timeout = params.recv_timeout;
+    if (cmd.stream_now == false)
+        timeout += params.start_time_delay;
+
+    size_t num_recv_calls = 0;
+    int64_t cur_timestamp = 0;
+
+    boost::system_time time_last_progress;
+    uint64_t samps_last_progress = 0;
+
+    unsigned long long num_rx_samps_single_chan = 0;
+
+    //while (not boost::this_thread::interruption_requested()){
+    try {
+        while (running)
+        {
+            if (rx_sleep_delay_now > 0) // UNSYNC'd
+            {
+                size_t delay = rx_sleep_delay_now;
+                rx_sleep_delay_now = 0;
+                usleep(delay);
+            }
+
+            //try {
+                size_t recv_samps = rx_stream->recv(buffs, params.samps_per_buff, md, timeout, params.one_packet_at_a_time);
+                ++num_recv_calls;
+
+                if (params.progress_interval > 0.0)
+                {
+                    if (num_recv_calls == 1)
+                    {
+                        time_last_progress = boost::get_system_time();
+                        //samps_last_progress = recv_samps;
+                    }
+                    else
+                    {
+                        samps_last_progress += recv_samps;
+
+                        boost::system_time time_now = boost::get_system_time();
+                        boost::posix_time::time_duration update_diff = time_now - time_last_progress;
+                        if (((double)update_diff.ticks() / (double)TICKS_PER_SEC) >= params.progress_interval)
+                        {
+                            double d = ((double)samps_last_progress * (double)TICKS_PER_SEC) / ((double)update_diff.ticks());
+                            std::stringstream ss;
+                            ss << HEADER_RX"(" << get_stringified_time() << ") " << boost::format("%.6f Msps") % (d/1e6) << std::endl;
+                            std::cout << ss.str();
+                            time_last_progress = time_now;
+                            samps_last_progress = 0;
+                        }
+                    }
+                }
+
+                if (recv_samps > 0)
+                {
+                    if ((num_recv_calls == 1) && (cmd.stream_now == false))
+                        std::cout << HEADER_RX"(" << get_stringified_time() << ") Received first packet after delayed start with time " << boost::format("%.6f") % md.time_spec.get_real_secs() << std::endl;
+
+                    if (params.check_recv_time) {
+                        int64_t timestamp = md.time_spec.to_ticks(rate);
+                        if ((cur_timestamp != 0) && (cur_timestamp != timestamp)) {
+                            std::stringstream ss;
+                            ss << HEADER_RX"(" << get_stringified_time() << ") ";
+                            ss << boost::format("TS: %lld, expected: %lld (diff: %lld)") % timestamp % cur_timestamp % (timestamp - cur_timestamp) << std::endl;
+                            std::cout << ss.str();
+                        }
+                        cur_timestamp = timestamp + recv_samps;
+                    }
+
+                    if (params.size_map)
+                    {
+                        // FIXME
+                    }
+
+                    timeout = params.recv_timeout;
+                    num_rx_samps_single_chan += recv_samps;
+                    num_rx_samps += recv_samps * rx_stream->get_num_channels();
+
+                    if ((params.rx_sample_limit > 0) && (num_rx_samps_single_chan == params.rx_sample_limit))
+                    {
+                        std::stringstream ss;
+                        ss << HEADER_RX"(" << get_stringified_time() << ") ";
+                        ss << boost::format("Received all %lu requested samples") % params.rx_sample_limit << std::endl;
+                        std::cout << ss.str();
+                        break;
+                    }
+
+                    {
+                        boost::mutex::scoped_lock lock(last_rx_md_mutex);
+                        last_rx_md = md;
+                        last_rx_samps = recv_samps;
+                        recv_samp_count_progress += recv_samps;
+                        recv_samp_count_progress_update = boost::get_system_time();
+                        recv_done.notify_one();
+                    }
+
+                    if (params.capture_file != NULL)
+                    {
+                        if (params.interleave_rx_file_samples)
+                        {
+                            for (size_t i = 0; i < recv_samps; ++i)
+                            {
+                                size_t channel_count = rx_stream->get_num_channels();
+                                for (size_t j = 0; j < channel_count; ++j)
+                                {
+                                    params.capture_file->write((const char*)buffs[j] + (bytes_per_samp * i), bytes_per_samp);
+                                }
+                            }
+                        }
+                        else
+                        {
+                            for (size_t i = 0; i < rx_stream->get_num_channels(); ++i)
+                            {
+                                size_t num_bytes = recv_samps * bytes_per_samp;
+                                params.capture_file->write((const char*)buffs[i], num_bytes);
+                            }
+                        }
+                    }
+                }
+            //}
+            //catch (...) {
+              /* apparently, the boost thread interruption can sometimes result in
+                 throwing exceptions not of type boost::exception, this catch allows
+                 this thread to still attempt to issue the STREAM_MODE_STOP_CONTINUOUS
+              */
+            //  break;
+            //}
+
+            //handle the error codes
+            switch(md.error_code) {
+                case uhd::rx_metadata_t::ERROR_CODE_NONE:
+                    if (had_an_overflow) {
+                        had_an_overflow = false;
+                        num_dropped_samps += (md.time_spec - last_time).to_ticks(rate); // FIXME: Check this as 'num_dropped_samps' has come out -ve
+                    }
+                    break;
+
+                // ERROR_CODE_OVERFLOW can indicate overflow or sequence error
+                case uhd::rx_metadata_t::ERROR_CODE_OVERFLOW:   // 'recv_samps' should be 0
+                    last_time = md.time_spec;
+                    had_an_overflow = true;
+#if HAS_RX_METADATA_OUT_OF_SEQUENCE
+                    // check out_of_sequence flag to see if it was a sequence error or overflow
+                    if (!md.out_of_sequence)
+#endif // HAS_RX_METADATA_OUT_OF_SEQUENCE
+                        num_overflows++;
+                    break;
+
+                case uhd::rx_metadata_t::ERROR_CODE_TIMEOUT:
+                {
+                    std::stringstream ss;
+                    ss << HEADER_RX"(" << get_stringified_time() << ") ";
+                    ss << boost::format("Timeout") << std::endl;
+                    std::cout << ss.str();
+                    break;
+                }
+
+                default:
+                    std::cerr << HEADER_RX"Error code: " << md.error_code << std::endl;
+                    std::cerr << HEADER_RX"Unexpected error on recv, continuing..." << std::endl;
+                    break;
+            }
+
+            print_msgs();
+        }
+    }
+    catch (const std::runtime_error& e)
+    {
+        std::cout << HEADER_RX"Caught exception:" <<  e.what() << std::endl;
+    }
+    catch (...)
+    {
+        std::cout << HEADER_RX"Caught unhandled exception" << std::endl;
+    }
+
+    if (params.rx_sample_limit == 0)
+    {
+        std::cout << HEADER_RX"Stopping streaming..." << std::endl;
+        rx_stream->issue_stream_cmd(uhd::stream_cmd_t::STREAM_MODE_STOP_CONTINUOUS);
+    }
+
+    if (params.capture_file != NULL)
+    {
+        std::cout << HEADER_RX"Closing capture file..." << std::endl;
+        delete params.capture_file;
+        params.capture_file = NULL;
+    }
+
+    l.lock();
+    rx_thread_finished = true;
+
+    if (rx_interrupt_disabler)
+    {
+        delete rx_interrupt_disabler;
+        rx_interrupt_disabler = NULL;
+    }
+
+    std::cout << HEADER_RX"Exiting..." << std::endl;
+
+    rx_thread_complete.notify_all();
+}
+
+/***********************************************************************
+ * Benchmark TX Rate
+ **********************************************************************/
+
+static size_t tx_sleep_delay_now = 0;
+
+typedef struct TxParams {
+    uhd::time_spec_t start_time;
+    double send_timeout;
+    double send_start_delay;
+    bool use_tx_eob;
+    bool use_tx_timespec;
+    bool tx_rx_sync;
+    bool send_final_eob;
+    double progress_interval;
+    bool use_relative_timestamps;
+    bool follow_rx_timestamps;
+    double tx_time_offset;
+    double rx_rate;
+    size_t tx_burst_length;
+    size_t tx_flush_length;
+    double tx_full_scale;
+    double tx_time_between_bursts;
+    bool recover_late;
+    bool set_tx_freq;
+    double tx_freq;
+    double tx_freq_delay;
+    double tx_lo_offset;
+} TX_PARAMS;
+
+void benchmark_tx_rate(
+    uhd::usrp::multi_usrp::sptr usrp,
+    const std::string &tx_cpu,
+    uhd::tx_streamer::sptr tx_stream,
+    TX_PARAMS& params
+    )
+{
+    uhd::set_thread_priority_safe();
+
+    boost::mutex::scoped_lock l(begin_tx_mutex);
+
+    tx_interrupt_disabler = new boost::this_thread::disable_interruption();
+
+    //setup variables and allocate buffer
+    const double rate = usrp->get_tx_rate();
+    const size_t max_samps_per_packet = tx_stream->get_max_num_samps();
+
+    if (params.tx_burst_length == 0)
+    {
+        params.tx_burst_length = max_samps_per_packet - params.tx_flush_length;
+    }
+    size_t total_length = params.tx_burst_length + params.tx_flush_length;
+
+    uhd::time_spec_t packet_time = uhd::time_spec_t::from_ticks(total_length, rate);
+    size_t total_packet_count = (total_length / max_samps_per_packet) + ((total_length % max_samps_per_packet) ? 1 : 0);
+    if ((params.use_tx_eob) && (params.tx_time_between_bursts > 0))
+        packet_time += uhd::time_spec_t(params.tx_time_between_bursts);
+    size_t max_late_count = (size_t)(rate / (double)packet_time.to_ticks(rate)) * total_packet_count;
+
+    // Will be much higher L values (e.g. 31K) on e.g. B200 when entire TX pipeline is full of late packets (large size due to total TX buffering throughout transport & DSP)
+
+    std::cout << boost::format(HEADER_TX"TX burst length: %lu samples (flush: %lu samples), total: %lu (%f us)") % params.tx_burst_length % params.tx_flush_length % total_length % (1e6 * (double)total_length / rate) << std::endl;
+    std::cout << boost::format(HEADER_TX"Max late packet count: %lu") % max_late_count << std::endl;
+
+    std::vector<const void *> buffs;
+    std::vector<char> buff(max_samps_per_packet * uhd::convert::get_bytes_per_item(tx_cpu), 0);
+    float* pResponse = NULL;
+    if (tx_cpu == "fc32")
+    {
+        std::cout << HEADER_TX"Generating ramp" << std::endl;
+
+        pResponse = new float[total_length * 2];
+        for (int i = 0; i < (params.tx_burst_length * 2); i += 2)
+        {
+            pResponse[i+0] = (params.tx_full_scale) * ((double)i / (double)(params.tx_burst_length * 2));
+            pResponse[i+1] = 0.0f;
+        }
+
+        for (int i = (params.tx_burst_length * 2); i < (total_length * 2); ++i)
+        {
+            pResponse[i] = 0.0f;
+        }
+
+        for (size_t ch = 0; ch < tx_stream->get_num_channels(); ch++)
+        {
+            buffs.push_back(pResponse);
+        }
+    }
+    else
+    {
+        std::cout << HEADER_TX"Generating silence" << std::endl;
+
+        for (size_t ch = 0; ch < tx_stream->get_num_channels(); ch++)
+        {
+            buffs.push_back(&buff.front()); //same buffer for each channel
+        }
+    }
+
+    if (params.set_tx_freq)
+    {
+        if (params.tx_freq_delay > 0)   // FIXME: Experiment to see whether this will also experience head-of-line blocking due to SPI xact fill-up (as what happened with RX)
+        {
+            std::cout << boost::format(HEADER_TX"Scheduling TX freq in %d seconds: %f (LO offset: %f Hz)") % params.tx_freq_delay % params.tx_freq % params.tx_lo_offset << std::endl;
+
+            uhd::time_spec_t tune_time = params.start_time + uhd::time_spec_t(params.tx_freq_delay);
+            usrp->set_command_time(tune_time);
+        }
+        else
+            std::cout << boost::format(HEADER_TX"Setting TX freq: %f (LO offset: %f Hz)") % params.tx_freq % params.tx_lo_offset << std::endl;
+
+        uhd::tune_request_t tune_request = uhd::tune_request_t(params.tx_freq, params.tx_lo_offset);
+        for (size_t ch = 0; ch < tx_stream->get_num_channels(); ch++)
+            usrp->set_tx_freq(tune_request, ch);
+
+        if (params.tx_freq_delay > 0)
+        {
+            usrp->clear_command_time();
+        }
+    }
+
+    if ((params.use_tx_timespec) && (params.tx_rx_sync == false)) {
+        uhd::time_spec_t time_now  = usrp->get_time_now();
+        uhd::time_spec_t diff = time_now - params.start_time;
+        std::cout << boost::format(HEADER_TX"Now - start time: %ld ticks (%f seconds)") % diff.to_ticks(rate) % diff.get_real_secs() << std::endl;
+    }
+
+    uhd::tx_metadata_t md;
+    //md.start_of_burst;    // Currently not used on any HW
+    md.end_of_burst = params.use_tx_eob;
+
+    double timeout = params.send_timeout;
+
+    boost::system_time time_last_progress;
+    boost::system_time time_first_send;
+    uint64_t samps_last_progress = 0;
+
+    std::cout << HEADER_TX"Waiting..." << std::endl;
+    tx_thread_begin.notify_all();
+    begin.wait(l);
+    l.unlock();
+
+    uhd::time_spec_t last_recv_time;
+
+    if ((params.use_tx_timespec)/* && (params.send_start_delay > 0)*/) {
+        if ((params.tx_rx_sync) || (params.follow_rx_timestamps)) {
+            boost::mutex::scoped_lock lock(last_rx_md_mutex);
+            {
+                std::stringstream ss;
+                ss << HEADER_TX"(" << get_stringified_time() << ") Waiting for first RX packet..." << std::endl;
+                std::cout << ss.str();
+            }
+            recv_done.wait(lock);
+            {
+                std::stringstream ss;
+                ss << HEADER_TX"(" << get_stringified_time() << ") First RX packet arrived with time "<< boost::format("%.6f") % last_rx_md.time_spec.get_real_secs() << std::endl;
+                std::cout << ss.str();
+            }
+
+            last_recv_time = last_rx_md.time_spec;
+
+            if (params.tx_rx_sync)
+                params.start_time = last_rx_md.time_spec + uhd::time_spec_t(params.tx_time_offset);
+        }
+        else
+        {
+            std::stringstream ss;
+            ss << HEADER_TX"(" << get_stringified_time() << ") "<< boost::format("TX will start %lld ticks in the future") % uhd::time_spec_t(params.send_start_delay).to_ticks(rate) << std::endl;
+            std::cout << ss.str();
+
+            params.start_time += uhd::time_spec_t(params.send_start_delay);
+        }
+
+        md.time_spec = params.start_time;
+        md.has_time_spec = true;
+        //timeout += params.send_start_delay;    // In case we fill up HW buf with large initial send buffer
+    }
+
+    {
+        std::stringstream ss;
+        ss << HEADER_TX"(" << get_stringified_time() << ") Running..." << std::endl;
+        std::cout << ss.str();
+    }
+
+    // Important note:
+    //  When idle, HW will attempt to honour first packet's time_spec
+    //  If it's late, host will see L and HW will either drop it or send the next packet (depending on policy)
+    //  Once the first packet has been transmitted, and there is no EOB, the next packet will be sent
+    //      immediately following it, regardless of its time_spec
+    //  If there is an EOB, the HW is returned to idle
+    //  If there is an underrun, the HW is returned to idle
+
+    // o Free-run
+    // o Delayed start after common start
+    // o Delayed start after first RX
+    // o Calculate own TX time based on samples sent (will result in Ls)
+    //  - Why does this produce Ls with EOB enabled? Because (at least on B200) there is a finite state-switch time, so it won't be able to service the very next packet every time (e.g. burst separate of 1e-6 works)
+    // o Follow RX + N -> (really above) -> relative sync'd (calculated from samples count received)
+    // * Wait for next slot -> sync'd & not relative (use % (mod) for slot)
+    // o Detect Ls -> re-sync relative
+
+    size_t loops_to_send = 0;
+    uhd::time_spec_t next;
+    boost::system_time time_now = boost::get_system_time();
+    boost::system_time last_late_check_time = time_now;
+    unsigned long long last_num_late_packets = 0;
+    bool resync_time = false;
+    uhd::time_spec_t follow_time_target = md.time_spec;
+
+    //while (not boost::this_thread::interruption_requested()){
+    while (running)
+    {
+        if (tx_sleep_delay_now) // UNSYNC'd
+        {
+            size_t delay = tx_sleep_delay_now;
+            tx_sleep_delay_now = 0;
+            usleep(delay);
+        }
+
+        if (params.follow_rx_timestamps)
+        {
+            boost::mutex::scoped_lock lock(last_rx_md_mutex);
+
+            uhd::time_spec_t diff = last_rx_md.time_spec - last_recv_time;
+
+            if ((resync_time) || /*(num_send_calls == 0) || */((diff.get_real_secs() == 0) && (md.time_spec >= follow_time_target)))
+            {
+                recv_done.wait(lock);
+
+                if (resync_time)
+                {
+                    resync_time = false;
+
+                    last_recv_time = last_rx_md.time_spec;
+                    md.time_spec = last_rx_md.time_spec + uhd::time_spec_t(params.tx_time_offset);
+                    follow_time_target = md.time_spec;
+
+                    continue;
+                }
+
+                diff = last_rx_md.time_spec - last_recv_time;
+            }
+
+            if ((diff.get_real_secs() > 0) && (md.time_spec >= follow_time_target))
+            {
+                follow_time_target = md.time_spec + diff;
+
+                last_recv_time = last_rx_md.time_spec;
+            }
+        }
+
+        size_t nsent = tx_stream->send(buffs, total_length, md, timeout);
+        ++num_send_calls;
+
+        time_now = boost::get_system_time();
+
+        if (params.progress_interval > 0.0)
+        {
+            if (num_send_calls == 1)
+            {
+                time_first_send = time_last_progress = boost::get_system_time();
+                //samps_last_progress = nsent;
+            }
+            else
+            {
+                samps_last_progress += nsent;
+
+                //boost::system_time time_now = boost::get_system_time();
+                boost::posix_time::time_duration update_diff = time_now - time_last_progress;
+                if (((double)update_diff.ticks() / (double)TICKS_PER_SEC) >= params.progress_interval)
+                {
+                    double d = ((double)samps_last_progress * (double)TICKS_PER_SEC) / ((double)update_diff.ticks());
+                    std::stringstream ss;
+                    ss << HEADER_TX"(" << get_stringified_time() << ") " << boost::format("%.6f Msps") % (d/1e6) << std::endl;
+                    std::cout << ss.str();
+                    time_last_progress = time_now;
+                    samps_last_progress = 0;
+                }
+            }
+        }
+
+        if (nsent == 0)
+        {
+            if ((params.use_tx_timespec) && (params.send_start_delay > 0))
+            {
+                //boost::system_time time_now = boost::get_system_time();
+                boost::posix_time::time_duration diff = time_now - time_first_send;
+                if (((double)diff.ticks() / (double)TICKS_PER_SEC) >= (timeout + params.send_start_delay))
+                {
+                    // TX chain has filled after delayed start
+                }
+            }
+            else
+            {
+                // TX chain has filled after immediate start
+            }
+        }
+        else
+        {
+            if (nsent != total_length)
+            {
+                std::stringstream ss;
+                ss << HEADER_WARN"(" << get_stringified_time() << ") " << boost::format("Only sent %lu of %lu samples") % nsent % total_length << std::endl;
+                std::cout << ss.str();
+            }
+
+            if ((params.use_relative_timestamps) && (params.use_tx_timespec) && (md.has_time_spec))
+            {
+                md.time_spec += uhd::time_spec_t::from_ticks(nsent, rate);
+
+                if (params.tx_time_between_bursts)
+                    md.time_spec += uhd::time_spec_t(params.tx_time_between_bursts);
+            }
+        }
+
+        if ((num_tx_samps == 0) && (nsent > 0))
+        {
+            if (md.has_time_spec)
+            {
+                std::stringstream ss;
+                ss << HEADER_TX"(" << get_stringified_time() << ") ";
+                ss << boost::format("First send completed having sent %d samples") % nsent << std::endl;
+                std::cout << ss.str();
+            }
+        }
+
+        num_tx_samps += nsent * tx_stream->get_num_channels();
+
+        if ((params.use_tx_timespec == false) && (md.has_time_spec))
+            md.has_time_spec = false;
+
+        if (params.recover_late)
+        {
+            boost::posix_time::time_duration late_check_diff = time_now - last_late_check_time;
+            if (((double)late_check_diff.ticks() / (double)TICKS_PER_SEC) >= 1.0)   // MAGIC
+            {
+                // UNSYNC'd
+                // FIXME: Why doesn't num_late_packets go up with B200?
+                unsigned long long diff = /*num_late_packets*/num_late_packets_msg - last_num_late_packets;
+                if (diff >= max_late_count)
+                {
+                    std::stringstream ss;
+                    ss << HEADER_TX"(" << get_stringified_time() << ") ";
+                    ss << boost::format("Exceeded max late packet threshold: %llu >= %llu") % diff % max_late_count << std::endl;
+                    std::cout << ss.str();
+
+                    if ((params.tx_rx_sync) && (params.follow_rx_timestamps))
+                        resync_time = true;
+                    else
+                        md.time_spec = usrp->get_time_now() + uhd::time_spec_t(params.tx_time_offset);
+                }
+                last_num_late_packets = /*num_late_packets*/num_late_packets_msg;
+                last_late_check_time = time_now;
+            }
+        }
+
+        print_msgs();
+    }
+
+    if (params.send_final_eob)
+    {
+        std::cout << HEADER_TX"Sending final EOB..." << std::endl;
+
+        //send a mini EOB packet
+        md.has_time_spec = false;
+        md.end_of_burst = true;
+        tx_stream->send(buffs, 0, md);
+    }
+    else
+        std::cout << HEADER_TX"Not sending final EOB" << std::endl;
+
+    if (pResponse)
+    {
+        delete [] pResponse;
+        pResponse = NULL;
+    }
+
+    l.lock();
+    tx_thread_finished = true;
+
+    if (tx_interrupt_disabler)
+    {
+        delete tx_interrupt_disabler;
+        tx_interrupt_disabler = NULL;
+    }
+
+    std::cout << HEADER_TX"Exiting..." << std::endl;
+
+    tx_thread_complete.notify_all();
+}
+
+void benchmark_tx_rate_async_helper(
+    uhd::tx_streamer::sptr tx_stream,
+    double timeout)
+{
+    boost::mutex::scoped_lock l(begin_tx_async_begin);
+
+    tx_async_interrupt_disabler = new boost::this_thread::disable_interruption();
+
+    std::cout << HEADER_AS"Running..." << std::endl;
+
+    //setup variables and allocate buffer
+    uhd::async_metadata_t async_md;
+
+    l.unlock();
+
+    bool skip = false;
+
+    //while (not boost::this_thread::interruption_requested()){
+    while (running) {
+        if (not tx_stream->recv_async_msg(async_md, (skip ? 0 : timeout)))
+        {
+            //std::cout << "-" << std::endl;
+            skip = false;
+            continue;
+        }
+
+        skip = true;
+
+        //std::cout << "Async event code: " << async_md.event_code << std::endl;
+
+        //handle the error codes
+        switch(async_md.event_code)
+        {
+            case uhd::async_metadata_t::EVENT_CODE_BURST_ACK:
+                num_tx_acks++;
+                return;
+
+            case uhd::async_metadata_t::EVENT_CODE_UNDERFLOW:
+                num_underflows++;
+                break;
+
+            case uhd::async_metadata_t::EVENT_CODE_UNDERFLOW_IN_PACKET:
+                num_underflows_in_packet++;
+                break;
+
+            case uhd::async_metadata_t::EVENT_CODE_SEQ_ERROR:
+                num_seq_errors++;
+                break;
+
+            case uhd::async_metadata_t::EVENT_CODE_SEQ_ERROR_IN_BURST:
+                num_seq_errors_in_burst++;
+                break;
+
+            case uhd::async_metadata_t::EVENT_CODE_TIME_ERROR:
+                num_late_packets++;
+                break;
+
+            default:
+                std::cerr << HEADER_AS"Event code: " << async_md.event_code << std::endl;
+                std::cerr << HEADER_AS"Unexpected event on async recv, continuing..." << std::endl;
+                break;
+        }
+    }
+
+    l.lock();
+    tx_async_thread_finished = true;
+
+    if (tx_async_interrupt_disabler)
+    {
+        delete tx_async_interrupt_disabler;
+        tx_async_interrupt_disabler = NULL;
+    }
+
+    std::cout << HEADER_AS"Exiting..." << std::endl;
+
+    tx_async_thread_complete.notify_all();
+}
+
+std::vector<size_t> get_channels(const std::string& channel_list, size_t max = -1)
+{
+    std::vector<std::string> channel_strings;
+    std::vector<size_t> channel_nums;
+
+    if (channel_list.size() > 0)
+        boost::split(channel_strings, channel_list, boost::is_any_of("\"',"));
+
+    for (size_t ch = 0; ch < channel_strings.size(); ch++)
+    {
+        size_t chan = boost::lexical_cast<int>(channel_strings[ch]);
+        if ((max >= 0) && (chan >= max)) {
+            throw std::runtime_error("Invalid channel(s) specified.");
+        }
+        else {
+            channel_nums.push_back(boost::lexical_cast<int>(channel_strings[ch]));
+        }
+    }
+
+    return channel_nums;
+}
+
+/***********************************************************************
+ * Main code + dispatcher
+ **********************************************************************/
+int UHD_SAFE_MAIN(int argc, char *argv[]){
+    uhd::set_thread_priority_safe();
+
+    //variables to be set by po
+    std::string args;
+    double duration;
+    double rate = 1e6;
+    double rx_rate = rate, tx_rate = rate;
+    std::string rx_otw, tx_otw;
+    std::string rx_cpu, tx_cpu;
+    std::string mode;
+    std::string channel_list = "0";
+    std::string rx_channel_list/* = channel_list*/, tx_channel_list/* = channel_list*/;
+    size_t samps_per_buff;
+    size_t samps_per_packet;
+    double master_clock_rate;
+    double recv_timeout, send_timeout;
+    double recv_start_delay, send_start_delay;
+    double tx_async_timeout;
+    double interrupt_timeout;
+    double progress_interval = 0.0;
+    double rx_progress_interval = progress_interval, tx_progress_interval = progress_interval;
+    double tx_time_offset;
+    size_t tx_burst_length; // Optionally in time (not samples)
+    size_t tx_flush_length;
+    size_t interactive_sleep;
+    double tx_full_scale;
+    double tx_freq = 0, tx_freq_init = 0;
+    double rx_freq = 0, rx_freq_init = 0;
+    double rx_lo_offset, tx_lo_offset;
+    double tx_freq_delay = 0, rx_freq_delay = 0;
+    double tx_gain = 0, rx_gain = 0;
+    double tx_time_between_bursts;
+    size_t tx_sleep_delay;
+    size_t rx_sleep_delay;
+    size_t rx_sample_limit;
+    std::string rx_file;
+    std::string time_source, clock_source;
+    std::string tx_ant, rx_ant;
+
+    //setup the program options
+    po::options_description desc("Allowed options");
+    desc.add_options()
+        ("help", "help message")
+        ("args", po::value<std::string>(&args)->default_value(""), "single uhd device address args")
+        ("duration", po::value<double>(&duration)->default_value(0.0), "duration for the test in seconds (0 = forever)")
+        ("rate", po::value<double>(&rate)->default_value(rate), "specify to perform a TX & RX rate test (sps)")
+        ("rx-rate", po::value<double>(&rx_rate), "specify to perform a RX rate test (sps)")
+        ("tx-rate", po::value<double>(&tx_rate), "specify to perform a TX rate test (sps)")
+        ("rx-otw", po::value<std::string>(&rx_otw)->default_value("sc16"), "specify the over-the-wire sample mode for RX")
+        ("tx-otw", po::value<std::string>(&tx_otw)->default_value("sc16"), "specify the over-the-wire sample mode for TX")
+        ("rx-cpu", po::value<std::string>(&rx_cpu)->default_value("fc32"), "specify the host/cpu sample mode for RX")
+        ("tx-cpu", po::value<std::string>(&tx_cpu)->default_value("fc32"), "specify the host/cpu sample mode for TX")
+        ("mode", po::value<std::string>(&mode)->default_value("none"), "multi-channel sync mode option: none, mimo (mimo overrides time and clock source")
+        ("time", po::value<std::string>(&time_source), "time reference (external, mimo)")
+        ("clock", po::value<std::string>(&clock_source), "clock reference (internal, external, mimo)")
+        ("channels", po::value<std::string>(&channel_list)/*->default_value(channel_list)*/, "which channel(s) to use (specify \"0\", \"1\", \"0,1\", etc)")
+        ("rx-channels", po::value<std::string>(&rx_channel_list), "which RX channel(s) to use (specify \"0\", \"1\", \"0,1\", etc)")
+        ("tx-channels", po::value<std::string>(&tx_channel_list), "which TX channel(s) to use (specify \"0\", \"1\", \"0,1\", etc)")
+        ("spp", po::value<size_t>(&samps_per_packet)->default_value(0), "samples per packet (0: use default)")
+        ("spb", po::value<size_t>(&samps_per_buff)->default_value(0), "samples per buffer (0: use max samples per packet)")
+        ("mcr", po::value<double>(&master_clock_rate)->default_value(0.0), "master clock rate (0: use default)")
+        ("rx-timeout", po::value<double>(&recv_timeout)->default_value(0.1), "recv timeout")
+        ("tx-timeout", po::value<double>(&send_timeout)->default_value(0.1), "send timeout")
+        ("tx-async-timeout", po::value<double>(&tx_async_timeout)->default_value(0.1), "recv_async_msg timeout")
+        ("rx-start-delay", po::value<double>(&recv_start_delay)->default_value(0.0), "recv start delay (seconds)")
+        ("tx-start-delay", po::value<double>(&send_start_delay)->default_value(0.0), "send start delay (seconds)")
+        ("interrupt-timeout", po::value<double>(&interrupt_timeout)->default_value(0.0), "time before re-enabling boost thread interruption")
+        ("msg-interval", po::value<double>(&msg_print_interval)->default_value(0.0), "seconds between printing UHD fastpath status messages")
+        ("progress-interval", po::value<double>(&progress_interval)->default_value(progress_interval), "seconds between bandwidth updates (0 disables)")
+        ("rx-progress-interval", po::value<double>(&rx_progress_interval), "seconds between RX bandwidth updates (0 disables)")
+        ("tx-progress-interval", po::value<double>(&tx_progress_interval), "seconds between TX bandwidth updates (0 disables)")
+        ("tx-offset", po::value<double>(&tx_time_offset), "seconds that TX should be in front of RX when following")
+        ("tx-length", po::value<size_t>(&tx_burst_length)->default_value(0), "TX burst length in samples (0: maximum packet size)")
+        ("tx-flush", po::value<size_t>(&tx_flush_length)->default_value(0), "samples to flush TX with after burst")
+        ("tx-burst-separation", po::value<double>(&tx_time_between_bursts), "seconds between TX bursts")
+        ("interactive-sleep", po::value<size_t>(&interactive_sleep)->default_value(10), "interactive sleep period (ms)")
+        ("tx-full-scale", po::value<double>(&tx_full_scale)->default_value(0.7), "full-scale TX sample value")
+        ("tx-freq", po::value<double>(&tx_freq), "TX frequency (Hz)")
+        ("tx-lo-offset", po::value<double>(&tx_lo_offset)->default_value(0.0), "TX LO offset (Hz)")
+        ("tx-freq-init", po::value<double>(&tx_freq_init), "initial TX frequency before realising main TX frequency (Hz)")
+        ("tx-freq-delay", po::value<double>(&tx_freq_delay), "seconds after which to set main TX frequency (Hz)")
+        ("rx-freq", po::value<double>(&rx_freq), "RX frequency (Hz)")
+        ("rx-lo-offset", po::value<double>(&rx_lo_offset)->default_value(0.0), "RX LO offset (Hz)")
+        ("rx-freq-init", po::value<double>(&rx_freq_init), "initial RX frequency before realising main RX frequency (Hz)")
+        ("rx-freq-delay", po::value<double>(&rx_freq_delay), "seconds after which to set main RX frequency (Hz)")
+        ("tx-gain", po::value<double>(&tx_gain), "TX gain (Hz)")
+        ("rx-gain", po::value<double>(&rx_gain), "RX gain (Hz)")
+        ("tx-ant", po::value<std::string>(&tx_ant), "TX antenna")
+        ("rx-ant", po::value<std::string>(&rx_ant), "RX antenna")
+        ("tx-sleep-delay", po::value<size_t>(&tx_sleep_delay)->default_value(1000), "TX sleep delay (us)")
+        ("rx-sleep-delay", po::value<size_t>(&rx_sleep_delay)->default_value(1000), "RX sleep delay (us)")
+        ("rx-sample-limit", po::value<size_t>(&rx_sample_limit)->default_value(0), "total number of samples to receive (0 implies continuous streaming)")
+        ("rx-file", po::value<std::string>(&rx_file)->default_value(""), "RX capture file path")
+        //("allow-late", "allow late bursts")
+        ("drop-late", "drop late bursts")
+        ("still-set-rates", "still set rate on unused direction")
+        ("rx-single-packets", "receive one packet at a time")
+        ("check-rx-time", "check receive timespec rounding")
+        ("tx-eob", "use EOB")
+        ("tx-timespec", "use TX timespec")
+        ("tx-rx-sync", "sync TX timestamps to RX")
+        ("final-eob", "send final EOB")
+        ("relative-tx", "use relative TX timestamps")
+        ("tx-follows-rx", "TX timestamps follow RX")
+        ("rx-size-map", "collect size map of RX packets (implies receive one packet at a time)")
+        ("interactive", "interactive mode")
+        ("recover-late", "recover from excessive late TX packets")
+        ("disable-async", "disable the async message thread")
+        ("interleave-rx-file-samples", "interleave individual samples (default is interleaving buffers)")
+        // record TX/RX times
+        // Optional interruption
+        // simulate u / o at random / pulses
+        // exit on O / other error
+        // suppress msgs
+        // recv/send jitter
+        // capture each channel to separate files (if format string is spec'd)
+        // check sensors
+    ;
+    po::variables_map vm;
+    po::store(po::parse_command_line(argc, argv, desc), vm);
+    po::notify(vm);
+
+    if (vm.count("rx-rate") == 0)
+        rx_rate = rate;
+    if (vm.count("tx-rate") == 0)
+        tx_rate = rate;
+    //if (vm.count("rx-channels") == 0)
+    //    rx_channel_list = channel_list;
+    //if (vm.count("tx-channels") == 0)
+    //    tx_channel_list = channel_list;
+    if ((vm.count("rx-channels") == 0) && (vm.count("tx-channels") == 0))
+        rx_channel_list = tx_channel_list = channel_list;
+    if (vm.count("rx-progress-interval") == 0)
+        rx_progress_interval = progress_interval;
+    if (vm.count("tx-progress-interval") == 0)
+        tx_progress_interval = progress_interval;
+
+    //print the help message
+    if (vm.count("help") or ((rx_rate + tx_rate) == 0)){
+        std::cout << boost::format("UHD Benchmark Rate %s") % desc << std::endl;
+        std::cout <<
+        "    By default, performs single-channel full-duplex test at 1 Msps with continuous streaming.\n"
+        "    Specify --channels to set RX & TX,\n"
+        "        or just --rx-channels and/or --tx-channels.\n"
+        "    Specify --rate to set both RX & TX.\n"
+        "    Specify --rx-rate to set custom RX rate.\n"
+        "    Specify --tx-rate to set custom TX rate.\n"
+        << std::endl;
+        return ~0;
+    }
+
+    bool allow_late_bursts = (/*vm.count("allow-late") > 0*/vm.count("drop-late") == 0);
+    bool still_set_rates = (vm.count("still-set-rates") > 0);
+    bool recv_single_packets = (vm.count("rx-single-packets") > 0);
+    bool check_recv_time = (vm.count("check-rx-time") > 0);
+    bool use_tx_eob = (vm.count("tx-eob") > 0);
+    bool use_tx_timespec = (vm.count("tx-timespec") > 0);
+    bool tx_rx_sync = (vm.count("tx-rx-sync") > 0);
+    bool send_final_eob = (vm.count("final-eob") > 0);
+    bool use_relative_tx_timestamps = (vm.count("relative-tx") > 0);
+    bool tx_follows_rx = (vm.count("tx-follows-rx") > 0);
+    bool rx_size_map = (vm.count("rx-size-map") > 0);
+    bool interactive = (vm.count("interactive") > 0);
+    bool recover_late = (vm.count("recover-late") > 0);
+    bool enable_async = (vm.count("disable-async") == 0);
+    bool interleave_rx_file_samples = (vm.count("interleave-rx-file-samples") > 0);
+
+    boost::posix_time::time_duration interrupt_timeout_duration(boost::posix_time::seconds(long(interrupt_timeout)) + boost::posix_time::microseconds(long((interrupt_timeout - floor(interrupt_timeout))*1e6)));
+
+    if (interactive)
+    {
+        //WINDOW* window = initscr();
+        //newterm(NULL, stdin, NULL);
+        //cbreak();
+        //noecho();
+        //nonl();
+        //intrflush(window, FALSE);
+        //keypad(window, TRUE);   // Enable function keys, arrow keys, ...
+        //nodelay(window, 0);
+        //timeout(interactive_sleep);
+        set_nonblock(true);
+    }
+
+    try
+    {
+        //create a usrp device
+        uhd::device_addrs_t device_addrs = uhd::device::find(args);
+        if (not device_addrs.empty() and device_addrs.at(0).get("type", "") == "usrp1"){
+            std::cerr << HEADER_WARN"Benchmark results will be inaccurate on USRP1 due to insufficient hardware features.\n" << std::endl;
+        }
+        std::cout << boost::format(HEADER "Creating the usrp device with args \"%s\"") % args << std::endl;
+        uhd::usrp::multi_usrp::sptr usrp = uhd::usrp::multi_usrp::make(args);
+        std::cout << std::endl;
+        std::cout << boost::format(HEADER "Using Device: %s") % usrp->get_pp_string() << std::endl;
+
+        //std::vector<size_t> channel_nums = get_channels(channel_list);
+        std::vector<size_t> rx_channel_nums = get_channels((/*rx_channel_list.size() ? */rx_channel_list/* : channel_list*/), usrp->get_rx_num_channels());
+        std::vector<size_t> tx_channel_nums = get_channels((/*tx_channel_list.size() ? */tx_channel_list/* : channel_list*/), usrp->get_tx_num_channels());
+
+        if ((rx_channel_nums.size() == 0) && (tx_channel_nums.size() == 0))
+        {
+            std::cout << HEADER_ERROR "Need at least one RX or one TX channel to run" << std::endl;
+            return ~0;
+        }
+
+        if ((tx_rx_sync) || (tx_follows_rx))
+        {
+            if (tx_channel_nums.size() == 0)
+            {
+                std::cout << HEADER_ERROR "Cannot sync/follow TX to RX without any TX channels" << std::endl;
+                return ~0;
+            }
+            if (rx_channel_nums.size() == 0)
+            {
+                std::cout << HEADER_ERROR "Cannot sync/follow TX to RX without any RX channels" << std::endl;
+                return ~0;
+            }
+        }
+
+        if (master_clock_rate > 0)
+        {
+            std::cout << boost::format(HEADER "Requested master clock rate: %f") % master_clock_rate << std::endl;
+            usrp->set_master_clock_rate(master_clock_rate);
+        }
+
+        std::cout << boost::format(HEADER "Actual master clock rate: %f") % usrp->get_master_clock_rate() << std::endl;
+
+        if (mode == "mimo") // FIXME: Warn if time/clock sources manually set
+        {
+            usrp->set_clock_source("mimo", 0);  // FIXME: Check this (that it's specific to mboard 0)
+            usrp->set_time_source("mimo", 0);
+
+            std::cout << HEADER "Sleeping after setting clock & time sources" << std::endl;
+            boost::this_thread::sleep(boost::posix_time::seconds(1));
+        }
+        else
+        {
+            if (time_source.empty() == false)
+            {
+                usrp->set_time_source(time_source);
+                std::cout << boost::format(HEADER "Time source set to: %s") % time_source << std::endl;
+            }
+
+            if (clock_source.empty() == false)
+            {
+                usrp->set_clock_source(clock_source);
+                std::cout << boost::format(HEADER "Clock source set to: %s") % clock_source << std::endl;
+            }
+        }
+
+        if ((rx_channel_nums.size() > 0) || (still_set_rates))
+        {
+            usrp->set_rx_rate(rx_rate);
+            double actual_rx_rate = usrp->get_rx_rate();
+            std::cout << boost::format(HEADER_RX"Actual RX rate: %f") % actual_rx_rate << std::endl;
+
+            if (rx_ant.empty() == false)
+            {
+                std::cout << boost::format(HEADER_RX"Selecting RX antenna: %s") % rx_ant << std::endl;
+                for (size_t ch = 0; ch < rx_channel_nums.size(); ch++)
+                    usrp->set_rx_antenna(rx_ant, ch);
+            }
+
+            if (vm.count("rx-freq-init") > 0)
+            {
+                std::cout << boost::format(HEADER_RX"Setting initial RX freq: %f (LO offset: %f Hz)") % rx_freq_init % rx_lo_offset << std::endl;
+                uhd::tune_request_t tune_request = uhd::tune_request_t(rx_freq_init, rx_lo_offset);
+                for (size_t ch = 0; ch < rx_channel_nums.size(); ch++)
+                    usrp->set_rx_freq(tune_request, ch);
+            }
+
+            if (vm.count("rx-gain") > 0)
+            {
+                std::cout << boost::format(HEADER_RX"Setting RX gain: %f") % rx_gain << std::endl;
+                for (size_t ch = 0; ch < rx_channel_nums.size(); ch++)
+                    usrp->set_rx_gain(rx_gain, ch);
+            }
+        }
+
+        if ((tx_channel_nums.size() > 0) || (still_set_rates))
+        {
+            usrp->set_tx_rate(tx_rate);
+            double actual_tx_rate = usrp->get_tx_rate();
+            std::cout << boost::format(HEADER_TX"Actual TX rate: %f") % actual_tx_rate << std::endl;
+
+            if (tx_ant.empty() == false)
+            {
+                std::cout << boost::format(HEADER_TX"Selecting TX antenna: %s") % tx_ant << std::endl;
+                for (size_t ch = 0; ch < tx_channel_nums.size(); ch++)
+                    usrp->set_tx_antenna(tx_ant, ch);
+            }
+
+            if (vm.count("tx-freq-init") > 0)
+            {
+                std::cout << boost::format(HEADER_TX"Setting initial TX freq: %f Hz (LO offset: %f Hz)") % tx_freq_init % tx_lo_offset << std::endl;
+                uhd::tune_request_t tune_request = uhd::tune_request_t(tx_freq_init, tx_lo_offset);
+                for (size_t ch = 0; ch < tx_channel_nums.size(); ch++)
+                    usrp->set_tx_freq(tune_request, ch);
+            }
+
+            if (vm.count("tx-gain") > 0)
+            {
+                std::cout << boost::format(HEADER_TX"Setting TX gain: %f") % tx_gain << std::endl;
+                for (size_t ch = 0; ch < tx_channel_nums.size(); ch++)
+                    usrp->set_tx_gain(tx_gain, ch);
+            }
+        }
+
+        uhd::time_spec_t time_start = usrp->get_time_now();	// Usually DSP #0 on mboard #0
+
+        std::cout << boost::format(HEADER "Time now:  %f seconds (%llu ticks)") % time_start.get_real_secs() % time_start.to_ticks(usrp->get_master_clock_rate()) << std::endl;
+
+        boost::thread_group thread_group;
+
+        {
+            boost::mutex::scoped_lock l_tx(begin_tx_mutex);
+            boost::mutex::scoped_lock l_rx(begin_rx_mutex);
+
+            TX_PARAMS tx_params;
+            RX_PARAMS rx_params;
+
+            if (rx_channel_nums.size() > 0)
+            {
+                //create a receive streamer
+                size_t bytes_per_rx_sample = uhd::convert::get_bytes_per_item(rx_cpu);
+                std::cout << boost::format(HEADER_RX"CPU bytes per RX sample: %d for '%s'") % bytes_per_rx_sample % rx_cpu << std::endl;
+                size_t wire_bytes_per_rx_sample = uhd::convert::get_bytes_per_item(rx_otw);
+                std::cout << boost::format(HEADER_RX"OTW bytes per RX sample: %d for '%s'") % wire_bytes_per_rx_sample % rx_otw << std::endl;
+
+                uhd::stream_args_t rx_stream_args(rx_cpu, rx_otw);
+                rx_stream_args.channels = rx_channel_nums;
+                if (samps_per_packet > 0)
+                {
+                    std::cout << boost::format(HEADER_RX"Samples per RX packet requested: %d") % samps_per_packet << std::endl;
+                    rx_stream_args.args["spp"] = str(boost::format("%d") % samps_per_packet);
+                }
+                uhd::rx_streamer::sptr rx_stream = usrp->get_rx_stream(rx_stream_args);
+                samps_per_packet = rx_stream->get_max_num_samps();
+                std::cout << boost::format(HEADER_RX"Max samples per RX packet: %d") % samps_per_packet << std::endl;
+
+                if (samps_per_buff > 0)
+                {
+                    std::cout << boost::format(HEADER_RX"RX buffer size requested to be (samples): %d") % samps_per_buff << std::endl;
+                }
+                else
+                {
+                    std::cout << HEADER_RX"RX buffer size will accommodate one RX packet" << std::endl;
+                    samps_per_buff = samps_per_packet;
+                }
+
+                std::cout << boost::format(HEADER_RX"RX buffer size (samples): %d") % samps_per_buff << std::endl;
+
+                if (recv_start_delay > 0)
+                    std::cout << boost::format(HEADER_RX"RX streaming will begin in: %d seconds") % recv_start_delay << std::endl;
+                else
+                    std::cout << boost::format(HEADER_RX"RX streaming will begin immediately") << std::endl;
+
+                std::cout << boost::format(HEADER_RX"RX streamer timeout: %f seconds") % recv_timeout << std::endl;
+                if (recv_start_delay > 0)
+                    std::cout << boost::format(HEADER_RX"Initial RX streamer timeout: %f seconds") % (recv_timeout + recv_start_delay) << std::endl;
+
+                if ((recv_single_packets) && (samps_per_buff > samps_per_packet))
+                    std::cout << HEADER_RX"Will receive single packets" << std::endl;
+                else
+                    std::cout << HEADER_RX"Will receive as much as can fit in one host-side buffer" << std::endl;
+
+                if (rx_sample_limit > 0)
+                {
+                    std::cout << boost::format(HEADER_RX"Will receive a total of %ld samples") % rx_sample_limit << std::endl;
+                    const size_t upper_rx_sample_limit = 0x0FFFFFFF;
+                    if (rx_sample_limit > upper_rx_sample_limit)
+                        std::cout << boost::format(HEADER_WARN"Total number of requested samples (%ld) is greater than limit (%ld)") % rx_sample_limit % upper_rx_sample_limit << std::endl;
+                }
+
+                if (rx_size_map)
+                {
+                    if (recv_single_packets == false)
+                    {
+                        recv_single_packets = true;
+                    }
+                }
+
+                rx_params.capture_file = NULL;
+                if (rx_file.empty() == false)
+                {
+                    std::cout << boost::format(HEADER_RX"Capturing to \"%s\"") % rx_file << std::endl;
+                    rx_params.capture_file = new std::ofstream(rx_file.c_str(), std::ios::out);
+                }
+
+                std::cout << boost::format(
+                    HEADER_RX"Testing receive rate %f Msps on %u channels: %s"
+                ) % (usrp->get_rx_rate()/1e6) % rx_stream->get_num_channels() % rx_channel_list << std::endl;
+
+                rx_params.samps_per_packet = samps_per_packet;
+                rx_params.samps_per_buff = samps_per_buff;
+                rx_params.start_time = time_start;
+                rx_params.start_time_delay = recv_start_delay;
+                rx_params.recv_timeout = recv_timeout;
+                rx_params.one_packet_at_a_time = recv_single_packets;
+                rx_params.check_recv_time = check_recv_time;
+                rx_params.progress_interval = rx_progress_interval;
+                rx_params.size_map = rx_size_map;
+                rx_params.rx_sample_limit = rx_sample_limit;
+                rx_params.set_rx_freq = (vm.count("rx-freq") > 0);
+                rx_params.rx_freq = rx_freq;
+                rx_params.rx_freq_delay = rx_freq_delay;
+                rx_params.rx_lo_offset = rx_lo_offset;
+                rx_params.interleave_rx_file_samples = interleave_rx_file_samples;
+
+                thread_group.create_thread(boost::bind(
+                    &benchmark_rx_rate,
+                    usrp,
+                    rx_cpu,
+                    rx_stream,
+                    rx_params));
+            }
+
+            if (tx_channel_nums.size() > 0) {
+                //create a transmit streamer
+                size_t bytes_per_tx_sample = uhd::convert::get_bytes_per_item(tx_cpu);
+                std::cout << boost::format(HEADER_TX"CPU bytes per TX sample: %d for '%s'") % bytes_per_tx_sample % rx_cpu << std::endl;
+                size_t wire_bytes_per_tx_sample = uhd::convert::get_bytes_per_item(tx_otw);
+                std::cout << boost::format(HEADER_TX"OTW bytes per TX sample: %d for '%s'") % wire_bytes_per_tx_sample % tx_otw << std::endl;
+
+                uhd::stream_args_t tx_stream_args(tx_cpu, tx_otw);
+                tx_stream_args.channels = tx_channel_nums;
+                /*
+                 * In the "next_burst" mode, the DSP drops incoming packets until a new burst is started.
+                 * In the "next_packet" mode, the DSP starts transmitting again at the next packet.
+                 */
+                if (allow_late_bursts == false)
+                {
+                    std::cout << HEADER_TX"Underflow policy set to drop late bursts ('next_burst')" << std::endl;
+                    tx_stream_args.args["underflow_policy"] = "next_burst";
+                }
+                else
+                    std::cout << HEADER_TX"Default underflow policy: allow late bursts ('next_packet')" << std::endl;
+
+                uhd::tx_streamer::sptr tx_stream = usrp->get_tx_stream(tx_stream_args);
+
+                std::cout << boost::format(HEADER_TX"Max TX samples per packet: %d") % tx_stream->get_max_num_samps() << std::endl;
+
+                std::cout << boost::format(
+                    HEADER_TX"Testing transmit rate %f Msps on %u channels: %s"
+                ) % (usrp->get_tx_rate()/1e6) % tx_stream->get_num_channels() % tx_channel_list << std::endl;
+
+                if ((send_start_delay > 0) && (use_tx_timespec == false))   // FIXME: Don't display warnings if tx-follows-rx
+                {
+                    std::cout << HEADER_WARN"Send start delay ignored as not using TX timespec" << std::endl;
+                }
+                else if (((send_start_delay <= 0) && (tx_rx_sync == false) && (tx_follows_rx == false)) && (use_tx_timespec))
+                {
+                    std::cout << HEADER_WARN"Cannot use TX timespec without a TX start delay, TX-RX start sync, or TX following RX" << std::endl;
+                }
+                else if ((send_start_delay > 0) && (use_tx_timespec))
+                {
+                    if (use_relative_tx_timestamps)
+                        std::cout << HEADER_TX"Will relative timestamps" << std::endl;
+
+                    // FIXME: tx_follows_rx
+                }
+
+                if (tx_rx_sync)
+                {
+                    if (use_tx_timespec == false)
+                    {
+                        std::cout << HEADER_WARN"Cannot sync to RX when not using TX timespec" << std::endl;
+                    }
+                    else if (tx_time_offset <= 0)
+                    {
+                        std::cout << HEADER_WARN"Cannot sync to RX with no TX time offset" << std::endl;
+                    }
+                }
+
+                if (use_tx_eob)
+                    std::cout << HEADER_TX"Will use EOB" << std::endl;
+                else
+                    std::cout << HEADER_TX"Will not use EOB" << std::endl;
+
+                tx_params.start_time = time_start;
+                tx_params.send_timeout = send_timeout;
+                tx_params.send_start_delay = send_start_delay;
+                tx_params.use_tx_eob = use_tx_eob;
+                tx_params.use_tx_timespec = use_tx_timespec;
+                tx_params.tx_rx_sync = tx_rx_sync;
+                tx_params.send_final_eob = send_final_eob;
+                tx_params.progress_interval = tx_progress_interval;
+                tx_params.use_relative_timestamps = use_relative_tx_timestamps;
+                tx_params.follow_rx_timestamps = tx_follows_rx;
+                tx_params.tx_time_offset = tx_time_offset;
+                tx_params.rx_rate = rx_rate;    // FIXME: Check validity
+                tx_params.tx_burst_length = tx_burst_length;
+                tx_params.tx_flush_length = tx_flush_length;
+                tx_params.tx_full_scale = tx_full_scale;
+                tx_params.tx_time_between_bursts = tx_time_between_bursts;
+                tx_params.recover_late = recover_late;
+                tx_params.set_tx_freq = (vm.count("tx-freq") > 0);
+                tx_params.tx_freq = tx_freq;
+                tx_params.tx_freq_delay = tx_freq_delay;
+                tx_params.tx_lo_offset = tx_lo_offset;
+
+                thread_group.create_thread(boost::bind(&benchmark_tx_rate,
+                    usrp,
+                    tx_cpu,
+                    tx_stream,
+                    tx_params));
+
+                if (enable_async)
+                {
+                    thread_group.create_thread(boost::bind(&benchmark_tx_rate_async_helper,
+                        tx_stream,
+                        tx_async_timeout));
+                }
+            }
+
+            running = true;
+            std::cout << HEADER "Begin..." << std::endl;
+
+            if (tx_channel_nums.size() > 0)
+                tx_thread_begin.wait(l_tx);
+            if (rx_channel_nums.size() > 0)
+                rx_thread_begin.wait(l_rx);
+
+            std::signal(SIGINT, &sig_int_handler);
+
+            uhd::msg::register_handler(&msg_handler);
+
+            begin.notify_all();
+
+            // RTT is longer, so skip this
+            //time_start = usrp->get_time_now();	// Usually DSP #0 on mboard #0
+            //tx_params.start_time = rx_params.start_time = time_start;   // Update to ignore thread start-up time
+            //std::cout << boost::format("Time now:  %f seconds (%llu ticks)") % time_start.get_real_secs() % time_start.to_ticks(usrp->get_master_clock_rate()) << std::endl;
+        }   // TX/RX locks will be released
+
+        std::cout << HEADER << "(" << get_stringified_time() << ") Running..." << std::endl;
+
+        boost::mutex::scoped_lock l_stop(stop_mutex);
+        if (stop_signal_called == false)
+        {
+            if ((rx_sample_limit > 0) && (tx_channel_nums.size() == 0))
+            {
+                rx_thread_complete.wait(l_stop);
+            }
+            else if (interactive)
+            {
+                if (duration > 0)
+                {
+                    // FIXME: Stop time
+
+                    std::cout << HEADER "Waiting for Q to finish early..." << std::endl;
+                }
+                else
+                    std::cout << HEADER "Waiting for Q..." << std::endl;
+
+                while (stop_signal_called == false)
+                {
+                    // FIXME: Stop time
+
+                    if (kbhit(interactive_sleep))
+                    {
+                        char c = fgetc(stdin);
+                        if (c == EOF)
+                        {
+                            std::cout << HEADER "EOF" << std::endl;
+                            break;
+                        }
+
+                        if (tolower(c) == 'q')
+                            break;
+
+                        switch (c)
+                        {
+                            case 'L':
+                            case 'U':
+                                break;
+                            case 'l':
+                            case 'u':
+                                tx_sleep_delay_now = tx_sleep_delay;
+                                break;
+                            case 'O':
+                                break;
+                            case 'o':
+                                rx_sleep_delay_now = rx_sleep_delay;
+                                break;
+                        }
+                    }
+
+                    print_msgs();
+                }
+            }
+            else if (duration > 0)
+            {
+                //sleep for the required duration
+                std::cout << boost::format(HEADER "Main thread sleeping for: %f seconds (host wall clock)") % duration << std::endl;
+                std::cout << HEADER "Waiting for CTRL+C to finish early..." << std::endl;
+                const long secs = long(duration);
+                const long usecs = long((duration - secs)*1e6);
+
+                abort_event.timed_wait(l_stop, boost::posix_time::seconds(secs) + boost::posix_time::microseconds(usecs));
+                //boost::this_thread::sleep(boost::posix_time::seconds(secs) + boost::posix_time::microseconds(usecs));
+            }
+            else
+            {
+                std::cout << HEADER "Waiting for CTRL+C..." << std::endl;
+                abort_event.wait(l_stop);
+            }
+        }
+
+        running = false;
+
+        std::cout << HEADER "Stopping..." << std::endl;
+
+        // FIXME: Timed wait & re-enable interruptions
+
+        if (rx_channel_nums.size() > 0) {
+            std::cout << HEADER "Waiting for RX thread..." << std::endl;
+            boost::mutex::scoped_lock l(begin_rx_mutex);
+            while (rx_thread_finished == false)
+            {
+                if (interrupt_timeout == 0)
+                    rx_thread_complete.wait(l);
+                else
+                {
+                    if (rx_thread_complete.timed_wait(l, interrupt_timeout_duration) == false)
+                    {
+                        if (rx_interrupt_disabler)
+                        {
+                            std::cout << HEADER "Interrupting RX thread..." << std::endl;
+                            delete rx_interrupt_disabler;
+                            rx_interrupt_disabler = NULL;
+                            thread_group.interrupt_all();
+                        }
+                        else
+                        {
+                            std::cout << HEADER_WARN"Interrupting RX thread failed - giving up" << std::endl;
+                            break;
+                        }
+                    }
+                }
+            }
+        }
+
+        if (tx_channel_nums.size() > 0) {
+            std::cout << HEADER "Waiting for TX thread..." << std::endl;
+            boost::mutex::scoped_lock l(begin_tx_mutex);
+            while (tx_thread_finished == false)
+            {
+                if (interrupt_timeout == 0)
+                    tx_thread_complete.wait(l);
+                else
+                {
+                    if (tx_thread_complete.timed_wait(l, interrupt_timeout_duration) == false)
+                    {
+                        if (tx_interrupt_disabler)
+                        {
+                            std::cout << HEADER "Interrupting TX thread..." << std::endl;
+                            delete tx_interrupt_disabler;
+                            tx_interrupt_disabler = NULL;
+                            thread_group.interrupt_all();
+                        }
+                        else
+                        {
+                            std::cout << HEADER_WARN"Interrupting TX thread failed - giving up" << std::endl;
+                            break;
+                        }
+                    }
+                }
+            }
+
+            if (enable_async)
+            {
+                std::cout << HEADER "Waiting for TX async message thread..." << std::endl;
+                boost::mutex::scoped_lock l_async(begin_tx_async_begin);
+                while (tx_async_thread_finished == false)
+                {
+                    if (interrupt_timeout == 0)
+                        tx_async_thread_complete.wait(l_async);
+                    else
+                    {
+                        if (tx_async_thread_complete.timed_wait(l_async, interrupt_timeout_duration) == false)
+                        {
+                            if (tx_async_interrupt_disabler)
+                            {
+                                std::cout << HEADER "Interrupting TX async thread..." << std::endl;
+                                delete tx_async_interrupt_disabler;
+                                tx_async_interrupt_disabler = NULL;
+                                thread_group.interrupt_all();
+                            }
+                            else
+                            {
+                                std::cout << HEADER_WARN"Interrupting TX async thread failed - giving up" << std::endl;
+                                break;
+                            }
+                        }
+                    }
+                }
+            }
+        }
+
+        //interrupt and join the threads
+        thread_group.interrupt_all();
+        std::cout << HEADER "Waiting for threads to join..." << std::endl;
+        thread_group.join_all();
+    }
+    catch (const std::runtime_error& e)
+    {
+        std::cout << HEADER_ERROR "Unhandled exception: " << e.what() << std::endl;
+    }
+    catch (...)
+    {
+        std::cout << HEADER_ERROR "Caught an unknown exception" << std::endl;
+    }
+
+    //print summary
+    std::cout << std::endl << boost::format(
+        "Test summary:\n"
+        "  Num received samples:              %u\n"
+        "  Num dropped samples:               %u\n"
+        "  Num overflows detected:            %u\n"
+        "\n"
+        "  Num transmitted samples:           %u\n"
+        "  Num send calls:                    %u\n"
+        "  Num ACKs:                          %u\n"
+        "  Num sequence errors:               %u\n"
+        "  Num sequence in burst errors:      %u\n"
+        "  Num sequence errors (total):       %u\n"
+        "  Num underflows detected:           %u\n"
+        "  Num underflows in packet detected: %u\n"
+        "  Num underflows detected (total):   %u\n"
+        "  Num late packets:                  %u\n"
+    ) % num_rx_samps % num_dropped_samps % num_overflows % num_tx_samps % num_send_calls % num_tx_acks % num_seq_errors % num_seq_errors_in_burst % (num_seq_errors + num_seq_errors_in_burst) % num_underflows % num_underflows_in_packet % (num_underflows + num_underflows_in_packet) % num_late_packets;
+
+    //finished
+    std::cout << std::endl << "Done!" << std::endl;
+
+    if (interactive)
+    {
+        set_nonblock(false);
+        //endwin();
+    }
+
+    return EXIT_SUCCESS;
+}