Prepare for time-stamp IO with usrp b210

modules/modules.pro

@@ -10,7 +10,8 @@ SUBDIRS  += sources/source_plutosdr \
     network/network_p2p \
     sinks/sink_SQL \
     wrappers/wrapper_scripts \
-    uhd/uhd_usrp_continous
+    uhd/uhd_usrp_continous \
+        uhd/uhd_usrp_io
 
 
 qtHaveModule(charts){

modules/uhd/uhd_usrp_io/main.cpp

+/*
+ * Copyright 2015 Ettus Research LLC
+ * Copyright 2018 Ettus Research, a National Instruments Company
+ *
+ * SPDX-License-Identifier: GPL-3.0-or-later
+ */
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <string>
+#include <QCoreApplication>
+#include <QTextStream>
+#include "cmdlineparser.h"
+#include "tb_interface.h"
+#include <QFile>
+#include <QLocale>
+#ifdef Q_OS_LINUX
+#include <unistd.h>
+#endif
+using namespace TASKBUS;
+using namespace std;
+
+int do_iio(const cmdlineParser & args);
+
+int  main(int argc, char * argv[])
+{
+	QCoreApplication a(argc, argv);
+	init_client();
+#ifdef OFFLINEDEBUG
+	FILE * old_stdin, *old_stdout;
+	auto ars = debug("D:\\pid1008",&old_stdin,&old_stdout);
+	const  cmdlineParser args (ars);
+#else
+	const cmdlineParser args (argc,argv);
+#endif
+
+	int ret = 0;
+	QTextStream stmerr(stderr);
+	//每个模块要响应 --information参数,打印自己的功能定义字符串。或者提供一个json文件。
+	if (args.contains("information"))
+	{
+		QFile fp(":/uhd_usrp_continous."+QLocale::system().name()+".json");
+		if (!fp.open(QIODevice::ReadOnly))
+		{
+			fp.setFileName(":/uhd_usrp_continous.json");
+			fp.open(QIODevice::ReadOnly);
+		}
+		if (fp.isOpen())
+		{
+			QByteArray arr = fp.readAll();
+			arr.push_back('\0');
+			puts(arr.constData());
+			fflush(stdout);
+		}
+		ret = -1;
+	}
+	else if (args.contains("function"))//正常运行模式
+	{
+		fprintf(stderr,"Entering Loop...\n");
+		//用于接收消息的线程
+		ret = do_iio(args);
+	}
+	else
+	{
+		stmerr<<"Error:Function does not exits.\n";
+		ret = -1;
+	}
+	return ret;
+
+
+
+
+}

modules/uhd/uhd_usrp_io/resources.qrc

+<RCC>
+    <qresource prefix="/">
+        <file>uhd_usrp_io.json</file>
+        <file>uhd_usrp_io.zh_CN.json</file>
+    </qresource>
+</RCC>

modules/uhd/uhd_usrp_io/uhd_io.cpp

+#include <stdio.h>
+#include <string>
+#include <signal.h>
+#include <atomic>
+#include <vector>
+#include "uhd_thread.h"
+#include "cmdlineparser.h"
+#include "tb_interface.h"
+#include <uhd.h>
+using namespace TASKBUS;
+using namespace std;
+typedef short SPTYPE;
+static std::atomic<bool> stop_signal_called (false);
+#define UHD_DO(X) \
+{\
+    uhd_error e = (X);\
+    char buf_errs[512];\
+    if (e) { snprintf(buf_errs,sizeof(buf_errs),"Error in line %d, NO %d.",__LINE__,e);\
+    std::string ev = __FILE__;\
+    ev += ":";\
+    ev += buf_errs;\
+    return_code = 1;\
+    throw ev;\
+    }\
+    };
+void sigint_handler(int code){
+    (void)code;
+    stop_signal_called = true;
+}
+
+static char error_string[512];
+int do_iio(const cmdlineParser & args)
+{
+    int return_code = EXIT_SUCCESS;
+    std::string dev_args = args.toString("dev_args","");
+
+	const long long time_sync =2;//For multi channel-Sync;
+	//RX Channels, Default 0
+	std::vector<size_t> rx_channels = args.toUInt64Array("rx_channels");
+	adjuestUnit<size_t>(rx_channels,0);
+	int rx_channel_count = rx_channels.size();
+	if (rx_channel_count>2)
+		rx_channel_count = 2;
+	//RX Sample Rate in Hz
+	double rx_rate		=   args.toDouble("rx_rate",1)*1e6;
+	//RX Freq in MHz --> Hz
+	std::vector<double> rx_freq      =   args.toDoubleArray("rx_rf");
+	adjuestUnit<double> (rx_channel_count,rx_freq,1000,1e6);
+	//Rx Gain
+	std::vector<double> rx_gain      =   args.toDoubleArray("rx_gain");
+	adjuestUnit<double> (rx_channel_count,rx_gain,30);
+	//AGC
+	std::vector<int>    rx_agc		=   args.toIntArray("rx_agc");
+	adjuestUnit<int> (rx_channel_count,rx_agc,0);
+	//RX Bw in MHz --> Hz
+	std::vector<double> rx_bw		=   args.toDoubleArray("rx_bw");
+	adjuestUnit<double> (rx_channel_count,rx_bw ,0 ,1e6);
+	//RX Antenna
+	std::vector<std::string> rx_atn  =   args.toStringArray("rx_antenna");
+	adjustString (rx_channel_count,rx_atn,"RX2");
+	//RX switch
+    bool  rx_on         =   args.toInt("rx_on",0)?true:false;
+	//RX Buffer Len
+	quint64   rx_frame      =   args.toUInt64("rx_frame",10000);
+
+	//TX Channels, Default 0
+	std::vector<size_t> tx_channels = args.toUInt64Array("tx_channels");
+	adjuestUnit<size_t>(tx_channels,0);
+	int tx_channel_count = tx_channels.size();
+	if(tx_channel_count>2)
+		tx_channel_count = 2;
+	//TX Sample Rate in MHz -->Hz
+	double tx_rate		=   args.toDouble("tx_rate",1.0)*1e6;
+	//TX Freq in MHz -->Hz
+	std::vector<double> tx_freq		=   args.toDoubleArray("tx_rf");
+	adjuestUnit<double> (tx_channel_count,tx_freq,2000,1e6);
+	//Tx Gain in dB
+	std::vector<double> tx_gain      =   args.toDoubleArray("tx_gain");
+	adjuestUnit<double> (tx_channel_count,tx_gain,30);
+	//TX BW in MHz -->Hz
+	std::vector<double> tx_bw		=   args.toDoubleArray("tx_bw");
+	adjuestUnit<double> (tx_channel_count,tx_freq,0 ,1e6);
+	//TX Antenna
+	std::vector<std::string> tx_atn  =   args.toStringArray("tx_antenna");
+	adjustString (rx_channel_count,tx_atn,"TX/RX");
+	//TX switch
+    bool  tx_on         =   args.toInt("tx_on",0)?true:false;
+	//Tx Buffer Len
+	quint64   tx_frame      =   args.toUInt64("tx_frame",10000);
+
+	if (!rx_on && !tx_on)
+	{
+		fprintf(stderr,"Both RX and TX switch are OFF, nothing can do.\n");
+        return 0;
+	}
+	const quint32 instance	  = args.toUInt("instance",0);
+	const quint32 i_wav_rx[2]	  = {args.toUInt("wav_rx0",0),args.toUInt("wav_rx1",0)};
+	const quint32 i_wav_tx[2]	  = {args.toUInt("wav_tx0",0),args.toUInt("wav_tx1",0)};
+	const quint32 i_rx_tm = args.toInt("rx_timestamp",0);
+    //设备句柄
+	uhd_usrp_handle usrp = nullptr;
+	uhd_rx_streamer_handle rx_streamer = nullptr;
+	uhd_rx_metadata_handle rx_meta = nullptr;
+	uhd_tx_streamer_handle tx_streamer = nullptr;
+	uhd_tx_metadata_handle tx_meta = nullptr;
+
+
+    try{
+        fprintf(stderr, "Creating USRP with args \"%s\"...\n", dev_args.c_str());
+        UHD_DO(uhd_usrp_make(&usrp, dev_args.c_str()));
+
+        fflush(stderr);
+        // Ctrl+C will exit loop
+        signal(SIGINT, &sigint_handler);
+		if (rx_on)
+		{
+			// Create RX streamer
+			UHD_DO(uhd_rx_streamer_make(&rx_streamer));
+			// Create RX metadata
+			UHD_DO(uhd_rx_metadata_make(&rx_meta));
+			//RXPer-Channel Settings
+			for (int ch = 0; ch < rx_channel_count;++ch)
+			{
+				fprintf(stderr, "Setting RX Channel: %d=%lu...\n",ch, rx_channels[ch]);
+				UHD_DO(uhd_usrp_set_rx_antenna(usrp,rx_atn[ch].c_str(),ch));
+				fprintf(stderr, "Setting RX Rate: %lf...\n", rx_rate);
+				UHD_DO(uhd_usrp_set_rx_rate(usrp, rx_rate, rx_channels[ch]));
+				// See what rate actually is
+				UHD_DO(uhd_usrp_get_rx_rate(usrp, rx_channels[ch], &rx_rate));
+				fprintf(stderr, "Actual RX Rate: %lf...\n", rx_rate);
+				// Set gain
+				fprintf(stderr, "Setting RX Gain: %lf dB...\n", rx_gain[ch]);
+				UHD_DO(uhd_usrp_set_rx_gain(usrp, rx_gain[ch], rx_channels[ch], ""));
+				// See what gain actually is
+				UHD_DO(uhd_usrp_get_rx_gain(usrp, rx_channels[ch], "", &rx_gain[ch]));
+				fprintf(stderr, "Actual RX Gain: %lf...\n", rx_gain[ch]);
+				if (rx_agc[ch])
+				{
+					uhd_usrp_set_rx_agc(usrp,true,rx_channels[ch]);
+					uhd_usrp_set_rx_dc_offset_enabled(usrp,true,rx_channels[ch]);
+				}
+				// Set frequency
+				// Create other necessary structs
+				uhd_tune_request_t rx_tune_request =
+				{
+					/*.target_freq = */rx_freq[ch],
+					/*.rf_freq_policy = */UHD_TUNE_REQUEST_POLICY_AUTO,
+					/*.rf_freq = */0,
+					/*.dsp_freq_policy = */UHD_TUNE_REQUEST_POLICY_AUTO,
+					/*.dsp_freq = */0,
+					/*.args = */0
+				};
+				uhd_tune_result_t rx_tune_result;
+
+				fprintf(stderr, "Setting RX frequency: %lf MHz...\n", rx_freq[ch]/1e6);
+				UHD_DO(uhd_usrp_set_rx_freq(usrp, &rx_tune_request, rx_channels[ch], &rx_tune_result));
+				// See what frequency actually isrx_streamer
+				UHD_DO(uhd_usrp_get_rx_freq(usrp, rx_channels[ch], &rx_freq[ch]));
+				fprintf(stderr, "Actual RX frequency: %lf MHz...\n", rx_freq[ch] / 1e6);
+
+				fprintf(stderr, "Setting RX Bandwidth: %lf MHz...\n", rx_bw[ch]/1e6);
+				UHD_DO(uhd_usrp_set_rx_bandwidth(usrp, rx_bw[ch], rx_channels[ch]));
+				//Band
+				UHD_DO(uhd_usrp_get_rx_bandwidth(usrp, rx_channels[ch], &rx_bw[ch]));
+				fprintf(stderr, "Actual RX Bandwidth: %lf MHz...\n", rx_bw[ch] / 1e6);
+				fflush(stderr);
+
+
+			}
+		}
+		if (tx_on)
+		{
+			// Create TX streamer
+			UHD_DO(uhd_tx_streamer_make(&tx_streamer));
+			// Create TX metadata
+			UHD_DO(uhd_tx_metadata_make(&tx_meta, false, 0,0, false, false));
+
+			//TX Per-Channel Settings
+			for (int ch = 0;ch < tx_channel_count;++ch)
+			{
+				fprintf(stderr, "Setting TX Channel: %d=%lu...\n",ch, tx_channels[ch]);
+				UHD_DO(uhd_usrp_set_tx_antenna(usrp,tx_atn[ch].c_str(),ch));
+				// Set rate
+				fprintf(stderr, "Setting TX Rate: %lf...\n", tx_rate);
+				UHD_DO(uhd_usrp_set_tx_rate(usrp, tx_rate, tx_channels[ch]));
+				// See what rate actually is
+				UHD_DO(uhd_usrp_get_tx_rate(usrp, tx_channels[ch], &tx_rate));
+				fprintf(stderr, "Actual TX Rate: %lf...\n\n", tx_rate);
+				// Set gain
+				fprintf(stderr, "Setting TX Gain: %lf db...\n", tx_gain[ch]);
+				UHD_DO(uhd_usrp_set_tx_gain(usrp, tx_gain[ch],tx_channels[ch], ""));
+				// See what gain actually is
+				UHD_DO(uhd_usrp_get_tx_gain(usrp, tx_channels[ch], "", &tx_gain[ch]));
+				fprintf(stderr, "Actual TX Gain: %lf...\n", tx_gain[ch]);
+				// Create other necessary structs for TX
+				uhd_tune_request_t tx_tune_request = {
+					/*.target_freq = */tx_freq[ch],
+					/*.rf_freq_policy = */UHD_TUNE_REQUEST_POLICY_AUTO,
+					/*.rf_freq = */0,
+					/*.dsp_freq_policy = */UHD_TUNE_REQUEST_POLICY_AUTO,
+					/*.dsp_freq = */0,
+					/*.args = */0
+				};
+				uhd_tune_result_t tx_tune_result;
+				// Set frequency
+				fprintf(stderr, "Setting TX frequency: %f MHz...\n", tx_freq[ch] / 1e6);
+				UHD_DO(uhd_usrp_set_tx_freq(usrp, &tx_tune_request, tx_channels[ch], &tx_tune_result));
+
+				// See what frequency actually is
+				UHD_DO(uhd_usrp_get_tx_freq(usrp, tx_channels[ch], &tx_freq[ch]));
+				fprintf(stderr, "Actual TX frequency: %f MHz...\n", tx_freq[ch] / 1e6);
+
+				//Band
+				fprintf(stderr, "Setting TX Bandwidth: %f MHz...\n", tx_bw[ch]/1e6);
+				UHD_DO(uhd_usrp_set_tx_bandwidth(usrp, tx_bw[ch],tx_channels[ch]));
+
+				UHD_DO(uhd_usrp_get_tx_bandwidth(usrp, tx_channels[ch], &tx_bw[ch]));
+				fprintf(stderr, "Actual TX Bandwidth: %f MHz...\n", tx_bw[ch] / 1e6);
+
+
+			}
+		}
+		char rx_cpu_format[] = "sc16";
+        char rx_otw_format[] = "sc16";
+        char rx_args[] = "";
+
+		uhd_stream_args_t rx_stream_args = {
+		   /* .cpu_format = */rx_cpu_format,
+		   /* .otw_format = */rx_otw_format,
+		   /* .args = */rx_args,
+		   /* .channel_list = */rx_channels.data(),
+			/*.n_channels = */rx_channel_count
+        };
+		uhd_stream_cmd_t rx_stream_cmd = {
+			/*.stream_mode = */UHD_STREAM_MODE_START_CONTINUOUS,
+			/*.num_samps = */0,
+			/*.stream_now = */false,
+			/*.time_spec_full_secs = */time_sync,
+			/*.time_spec_frac_secs = */0
+        };
+
+        //char tx_cpu_format[] = "fc32";
+        char tx_cpu_format[] = "sc16";
+        char tx_otw_format[] = "sc16";
+        char tx_args[] = "";
+
+		uhd_stream_args_t tx_stream_args = {
+			/*.cpu_format = */tx_cpu_format,
+			/*.otw_format = */tx_otw_format,
+			/*.args = */tx_args,
+			/*.channel_list = */tx_channels.data(),
+		   /* .n_channels = */tx_channel_count
+        };
+        size_t rx_sps_buff = 0;
+        size_t tx_sps_buff = 0;
+		if (rx_on)
+		{
+			// Set up streamer
+			UHD_DO(uhd_usrp_get_rx_stream(usrp, &rx_stream_args, rx_streamer));
+			// Set up buffer
+			UHD_DO(uhd_rx_streamer_max_num_samps(rx_streamer, &rx_sps_buff));
+			fprintf(stderr, "RX Buffer size in samples: %zu\n", rx_sps_buff);
+		}
+        if (tx_on)
+        {
+            UHD_DO(uhd_usrp_get_tx_stream(usrp, &tx_stream_args, tx_streamer));
+            // Set up buffer
+            UHD_DO(uhd_tx_streamer_max_num_samps(tx_streamer, &tx_sps_buff));
+			fprintf(stderr, "TX Buffer size in samples: %zu\n", tx_sps_buff);
+            fflush(stderr);
+        }
+
+        //ring buffer in short inters
+		const quint64 sz_buffer_all = 8 * 1024 * 1024;
+		std::vector<std::vector<SPTYPE> > buf_rx_array;
+		std::vector<std::vector<SPTYPE> > buf_tx_array;
+		//In IQ Samples, RF io
+		std::vector<quint64> rx_pos, tx_pos;
+		//In IQ Samples, Std IO
+		std::vector<quint64> stdin_pos, stdout_pos;
+
+		std::vector<void *> rx_buff_ptr, tx_buff_ptr;
+		size_t num_rx_samps = 0,num_sps_sent = 0 ;
+
+		//RX timeStamp
+		int64_t rx_time_sec = 0;
+		double rx_time_frag = 0;
+		quint64 rx_time_sps = 0;
+
+		for (int ch = 0;ch<rx_channel_count;++ch)
+		{
+			buf_rx_array.push_back(std::vector<SPTYPE>(sz_buffer_all + 1024*1024,0));
+			rx_pos.push_back(0);
+			stdout_pos.push_back(0);
+			rx_buff_ptr.push_back(0);
+		}
+		for (int ch = 0;ch<tx_channel_count;++ch)
+		{
+			buf_tx_array.push_back(std::vector<SPTYPE>(sz_buffer_all + 1024*1024,0));
+			stdin_pos.push_back(0);
+			tx_pos.push_back(0);
+			tx_buff_ptr.push_back(0);
+		}
+		//reset time
+		uhd_usrp_set_time_now(usrp,0,0,0);
+        //Define RX Thread, Producer
+        std::function<void()> thread_rx = [&]()->void
+        {
+            try{
+                // Issue stream command
+                fprintf(stderr, "Issuing stream command.\n");
+                UHD_DO(uhd_rx_streamer_issue_stream_cmd(rx_streamer, &rx_stream_cmd));
+
+                //Read, RX
+                while (!stop_signal_called) {
+					//Init Ptr
+					for (int ch=0;ch<rx_channel_count;++ch)
+						rx_buff_ptr[ch] = (void *) &buf_rx_array[ch][(rx_pos[ch] * 2) % sz_buffer_all];
+                    // Handle data
+					UHD_DO(uhd_rx_streamer_recv(
+							   rx_streamer,
+							   rx_buff_ptr.data(),
+							   rx_sps_buff,
+							   &rx_meta, 5, true, &num_rx_samps));
+					quint64 rx_tmcurr = 0;
+					for (int ch=0;ch<rx_channel_count;++ch)
+					{
+						const quint64 bgnext = ((rx_pos[ch] + num_rx_samps) * 2) % sz_buffer_all;
+						if (bgnext < num_rx_samps * 2 && bgnext>0)
+							memcpy(&buf_rx_array[ch][0], (char *)(&buf_rx_array[ch][0]) + sz_buffer_all, bgnext * sizeof(SPTYPE) );
+						if (!ch)
+							rx_tmcurr = rx_pos[ch];
+						rx_pos[ch] += num_rx_samps;
+					}
+                    uhd_rx_metadata_error_code_t error_code;
+                    UHD_DO(uhd_rx_metadata_error_code(rx_meta, &error_code));
+                    if(error_code != UHD_RX_METADATA_ERROR_CODE_NONE){
+						fprintf(stderr, "Warning: Error code 0x%x 16was returned during streaming.\n", error_code);
+                        fputs(error_string,stderr);
+                        //UHD_DO(uhd_rx_streamer_issue_stream_cmd(rx_streamer, &rx_stream_cmd));
+						//stop_signal_called = true;
+                    }
+					else
+					{
+						bool hasTm;
+						UHD_DO(uhd_rx_metadata_has_time_spec(rx_meta, &hasTm));
+						if (hasTm)
+						{
+							uhd_rx_metadata_time_spec(rx_meta,&rx_time_sec, & rx_time_frag);
+							rx_time_sps = rx_tmcurr;
+						}
+					}
+                }
+             }
+             catch (std::string er)
+             {
+                fputs(er.c_str(),stderr);
+                stop_signal_called = true;
+             }
+        };
+
+
+        //Define StdOut Thread, Consumer
+        std::function<void()> thread_stdout = [&]()->void
+        {
+            try{
+                 //Read, RX
+				while (!stop_signal_called)
+				{
+					bool can_produce = true;
+					quint64 produce_clock = 0;
+					for (int ch = 0;ch<rx_channel_count && can_produce;++ch)
+					{						
+						if (stdout_pos[ch] + (long long)(rx_frame + rx_sps_buff )  >= rx_pos[ch] )
+							can_produce = false;
+						produce_clock = stdout_pos[ch];
+					}
+					if (!can_produce)
+					{
+						QThread::msleep(1);
+						continue;
+					}
+					if (i_rx_tm)
+					{
+						TASKBUS::push_subject(
+									i_rx_tm,
+									instance,
+									sizeof(produce_clock),
+									(unsigned char *) &produce_clock );
+					}
+
+					for (int ch = 0;ch<rx_channel_count && can_produce;++ch)
+					{
+						SPTYPE * buf_rx_ptr = buf_rx_array[ch].data();
+						const quint64 bgnext = ((stdout_pos[ch] + rx_frame) * 2) % sz_buffer_all;
+						//Dealing with t1.0he tail
+						if (bgnext < rx_frame * 2 && bgnext>0)
+						{
+							memcpy( buf_rx_ptr + sz_buffer_all,buf_rx_ptr, bgnext * sizeof(SPTYPE) );
+						}
+						if (i_wav_rx[ch]>0)
+							TASKBUS::push_subject(
+										i_wav_rx[ch],
+										instance,
+										rx_frame * 2 * sizeof(SPTYPE),
+										(unsigned char *) &buf_rx_ptr[(stdout_pos[ch] * 2)% sz_buffer_all] );
+						stdout_pos[ch] += rx_frame;
+					}
+
+
+                }
+             }
+             catch (std::string er)
+             {
+                fputs(er.c_str(),stderr);
+                stop_signal_called = true;
+             }
+        };
+
+        //Define StdIn Thread, Producer
+        std::function<void()> thread_stdin = [&]()->void
+        {
+            try{
+				 //Read STDIN
+                while (!stop_signal_called) {
+                    subject_package_header header;
+                    std::vector<unsigned char> packagedta = pull_subject(&header);
+                    if (!is_valid_header(header))
+                    {
+                        fprintf(stderr,"Recived BAD Command.");
+                        fflush(stderr);
+                        QThread::msleep(100);
+                        continue;
+                    }
+					if (packagedta.size())
+                    {
+                        if ( is_control_subject(header))
+                        {
+                            //收到命令进程退出的广播消息,退出
+                            if (strstr((const char *)packagedta.data(),"function=quit;")!=nullptr)
+                            {
+                                fprintf(stderr,"Recived Quit Command.");
+                                fflush(stderr);
+                                stop_signal_called = true;
+                            }
+                        }
+						else
+						{
+							for (int ch = 0; ch < tx_channel_count;++ch)
+							{
+								if (header.subject_id == i_wav_tx[ch])
+								{
+									SPTYPE * buf_tx_ptr = buf_tx_array[ch].data();
+									quint64 sps_rev = packagedta.size()/2/sizeof(SPTYPE);
+									short * iq = (short *)(packagedta.data());
+									memcpy(&buf_tx_ptr[(stdin_pos[ch] * 2)% sz_buffer_all],
+											iq,  sps_rev * sizeof(SPTYPE)*2
+											);
+									const quint64 bgnext = ((stdin_pos[ch] + sps_rev) * 2) % sz_buffer_all;
+									if (bgnext < sps_rev * 2 && bgnext>0)
+									{
+										memcpy(buf_tx_ptr, buf_tx_ptr + sz_buffer_all, bgnext * sizeof(SPTYPE) );
+									}
+									stdin_pos[ch] += sps_rev;
+								}
+							}
+						}
+					}
+                }
+             }
+             catch (std::string er)
+             {
+                fputs(er.c_str(),stderr);
+                stop_signal_called = true;
+             }
+        };
+
+        //Define thread_tx, Consumer
+        std::function<void()> thread_tx = [&]()->void
+        {
+            try{
+                while (!stop_signal_called)
+                {
+					bool can_send = true;
+					for (int ch=0;ch<tx_channel_count && can_send;++ch)
+					{
+						SPTYPE * buf_tx_ptr = buf_tx_array[ch].data();
+						if (stdin_pos[ch] < (tx_sps_buff + tx_frame)*10)
+						{
+							can_send = false;
+							continue;
+						}
+						if (tx_pos[ch] + (long long) (tx_sps_buff + tx_frame) >= stdin_pos[ch])
+						{
+							can_send = false;
+							continue;
+						}
+						const quint64 bgnext = ((tx_pos[ch] + tx_sps_buff) * 2) % sz_buffer_all;
+						//Dealing with the tail
+						if (bgnext < tx_sps_buff * 2 && bgnext>0)
+						{
+							memcpy( buf_tx_ptr + sz_buffer_all, buf_tx_ptr, bgnext * sizeof(SPTYPE) );
+						}
+						SPTYPE * tx_buff = &buf_tx_ptr[(tx_pos[ch] * 2) % sz_buffer_all];
+						tx_buff_ptr[ch] = (void *)tx_buff;
+					}
+					if (!can_send)
+					{
+						QThread::msleep(1);
+					}
+					else
+					{
+						UHD_DO(uhd_tx_streamer_send(tx_streamer,
+													(const void **)tx_buff_ptr.data(),
+													tx_sps_buff, &tx_meta,0.1, &num_sps_sent));
+						//tx_meta->tx_metadata_cpp.start_of_burst = false;
+						for(int ch = 0;ch<tx_channel_count;++ch)
+							tx_pos[ch] += num_sps_sent;
+					}
+                }
+            }
+            catch(std::string er)
+            {
+                fputs(er.c_str(),stderr);
+                stop_signal_called = true;
+            }
+        };
+
+        uhd_io_thread th_wav_rx(thread_rx,0);
+        uhd_io_thread th_wav_stdout(thread_stdout,0);
+        uhd_io_thread th_wav_stdin(thread_stdin,0);
+        uhd_io_thread th_wav_tx(thread_tx,0);
+
+		if (rx_on)
+        {
+            th_wav_rx.start(QThread::HighestPriority);
+            th_wav_stdout.start(QThread::HighestPriority);
+        }
+		if (tx_on)
+        {
+            th_wav_tx.start(QThread::HighestPriority);
+        }
+        th_wav_stdin.start(QThread::HighestPriority);
+
+        while (!stop_signal_called)
+        {
+			fprintf(stderr,"\nRX:");
+			for (int ch = 0;ch<rx_channel_count;++ch)
+				fprintf (stderr,"%d:%lld,%lld;",ch,rx_pos[ch],stdout_pos[ch]);
+			fprintf(stderr,"\n");
+			fprintf(stderr,"TX:");
+			for (int ch = 0;ch<tx_channel_count;++ch)
+				fprintf (stderr,"%d:%lld,%lld;",ch,stdin_pos[ch],tx_pos[ch]);
+			fprintf(stderr,"\n");
+            fflush(stderr);
+            QThread::msleep(1000);
+        }
+
+        th_wav_rx.wait();
+        th_wav_stdout.wait();
+        th_wav_tx.wait();
+        th_wav_stdin.wait();
+    }
+    catch(std::string s)
+    {
+        fputs(s.c_str(),stderr);
+    }
+
+    fflush(stderr);
+
+    if (tx_streamer) uhd_tx_streamer_free(&tx_streamer);
+    if (rx_streamer) uhd_rx_streamer_free(&rx_streamer);
+    if (tx_meta) uhd_tx_metadata_free(&tx_meta);
+    if (rx_meta) uhd_rx_metadata_free(&rx_meta);
+
+	if(return_code != EXIT_SUCCESS && usrp != nullptr){
+        uhd_usrp_last_error(usrp, error_string, 512);
+		fprintf(stderr, "USRP reported the following error: %s\n", error_string);
+    }
+    uhd_usrp_free(&usrp);
+
+    fprintf(stderr, (return_code ? "Failure\n" : "Success\n"));
+    fflush(stderr);
+    return return_code;
+}

modules/uhd/uhd_usrp_io/uhd_thread.cpp

+#include "uhd_thread.h"
+
+uhd_io_thread::uhd_io_thread (std::function<void (void)> runner,QObject * p)
+    :QThread(p)
+    ,m_runner(runner)
+
+{
+
+}
+uhd_io_thread::uhd_io_thread (QObject * p)
+    :QThread(p)
+
+{
+
+}
+
+void uhd_io_thread::run()
+{
+    if (m_runner)
+        m_runner();
+}

modules/uhd/uhd_usrp_io/uhd_thread.h

+#ifndef UHD_THREAD_H
+#define UHD_THREAD_H
+#include <QThread>
+#include <functional>
+class uhd_io_thread: public QThread{
+    Q_OBJECT
+public:
+    explicit uhd_io_thread (QObject * p);
+    explicit uhd_io_thread (std::function<void (void)> runner,QObject * p);
+    void setRunner(std::function<void (void)> r){ m_runner = r;}
+public:
+    void run() override;
+private:
+    std::function<void (void)> m_runner;
+};
+
+
+#endif // UHD_THREAD_H

modules/uhd/uhd_usrp_io/uhd_usrp_io.json

+{
+  "uhd_usrp_io": {
+    "name": "uhd_usrp_io",
+    "parameters": {
+      "dev_args": {
+        "type": "string",
+        "tooltip": "Additional Device Start Args.",
+        "default": ""
+      },
+      "rx_rate": {
+        "type": "int",
+        "tooltip": "RX rate(MHz)",
+        "default": 1.0
+      },
+      "rx_on": {
+        "type": "int",
+        "tooltip": "RX is ON",
+        "default": "0,0"
+      },
+      "rx_rf": {
+        "type": "double",
+        "tooltip": "RX Freq(MHz)",
+        "default": "1000.0,1100.0",
+        "range": "0-6000"
+      },
+      "rx_gain": {
+        "type": "double",
+        "tooltip": "RX Gain(dB)",
+        "default": "30.0,30.0",
+        "range": "0-100"
+      },
+      "rx_agc": {
+        "type": "int",
+        "tooltip": "RX AGC Switch",
+        "default": "0,0",
+        "range": "0=off,1=on"
+      },
+      "rx_bw": {
+        "type": "double",
+        "tooltip": "RX bandwidth(MHz)",
+        "default": "1.0,1.0"
+      },
+      "rx_atn": {
+        "type": "string",
+        "tooltip": "RX antenna",
+        "default": "RX2,RX2",
+        "range": "TX/RX,RX2"
+      },
+
+      "rx_channels": {
+        "type": "int",
+        "tooltip": "RX channels",
+        "default": "0,1",
+        "range": "0,1"
+      },
+      "tx_on": {
+        "type": "int",
+        "tooltip": "TX is ON",
+        "default": "0"
+      },
+      "tx_rate": {
+        "type": "int",
+        "tooltip": "TX rate(MHz)",
+        "default": 1.0
+      },
+      "tx_rf": {
+        "type": "double",
+        "tooltip": "TX freq(MHz)",
+        "default": "2000.0,2000.0"
+      },
+      "tx_gain": {
+        "type": "double",
+        "tooltip": "TX Gain(dB)",
+        "default": "0.0,0.0",
+        "range": "0-100"
+      },
+      "tx_bw": {
+        "type": "double",
+        "tooltip": "TX bandwidth(MHz)",
+        "default": "1.0,1.0"
+      },
+      "tx_atn": {
+        "type": "string",
+        "tooltip": "TX antenna",
+        "default": "TX/RX,TX/RX"
+      },
+
+      "tx_channels": {
+        "type": "int",
+        "tooltip": "TX channels",
+        "default": "0,1"
+      }
+    },
+    "input_subjects": {
+      "wav_tx0": {
+        "type": "short[2]",
+        "tooltip": "TX0IQ"
+      },
+      "wav_tx1": {
+        "type": "short[2]",
+        "tooltip": "TX1IQ"
+      }
+    },
+    "output_subjects": {
+      "rx_timestamp": {
+        "type": "uint64",
+        "tooltip": "RX_tm"
+      },
+      "wav_rx0": {
+        "type": "short[2]",
+        "tooltip": "RX0IQ"
+      },
+      "wav_rx1": {
+        "type": "short[2]",
+        "tooltip": "RX1IQ"
+      }
+    }
+  }
+}

modules/uhd/uhd_usrp_io/uhd_usrp_io.pro

+QT -= gui
+
+CONFIG += c++11 console
+CONFIG -= app_bundle
+# The following define makes your compiler emit warnings if you use
+# any Qt feature that has been marked deprecated (the exact warnings
+# depend on your compiler). Please consult the documentation of the
+# deprecated API in order to know how to port your code away from it.
+DEFINES += QT_DEPRECATED_WARNINGS
+INCLUDEPATH += ../../../tb_interface
+DESTDIR = $$OUT_PWD/../../../bin/modules
+SOURCES += \
+	main.cpp \
+	uhd_io.cpp \
+	uhd_thread.cpp
+
+
+RESOURCES += \
+	resources.qrc
+LIBS += -luhd
+win32{
+    INCLUDEPATH +="$(UHD_PKG_PATH)/include"
+    LIBS += -L"$(UHD_PKG_PATH)/lib"
+}
+
+HEADERS += \
+	uhd_thread.h

modules/uhd/uhd_usrp_io/uhd_usrp_io.zh_CN.json

+{
+    "uhd_usrp_io": {
+        "name": "UHD收发",
+        "parameters": {
+            "dev_args": {
+                "type": "string",
+                "tooltip": "设备初始化参数.",
+                "default": ""
+            },
+            "rx_rate": {
+                "type": "int",
+                "tooltip": "RX采样率(MHz)",
+                "default": 1.0
+            },
+            "rx_on": {
+                "type": "int",
+                "tooltip": "接收开关",
+                "default": "0"
+            },
+            "rx_frame": {
+                "type": "int",
+                "tooltip": "接收接口单次样点数",
+                "default": "10000",
+                "range": "200-10000"
+            },
+            "rx_rf": {
+                "type": "double",
+                "tooltip": "接收频率(MHz)",
+                "default": "99.0,101.1",
+                "range": "0-6000"
+            },
+            "rx_gain": {
+                "type": "double",
+                "tooltip": "接收增益(dB)",
+                "default": "30.0,30.0",
+                "range": "0-100"
+            },
+            "rx_agc": {
+                "type": "int",
+                "tooltip": "接收AGC开关",
+                "default": "0,0",
+                "range": "0=off,1=on"
+            },
+            "rx_bw": {
+                "type": "double",
+                "tooltip": "接收滤波器带宽(MHz)",
+                "default": "1.0,1.0"
+            },
+            "rx_atn": {
+                "type": "string",
+                "tooltip": "接收天线名称",
+                "default": "RX2,RX2",
+                "range": "TX/RX,RX2"
+            },
+            "rx_channels": {
+                "type": "int",
+                "tooltip": "接收各个通道号",
+                "default": "0,1",
+                "range": "0,1"
+            },
+            "tx_rate": {
+                "type": "int",
+                "tooltip": "TX采样率(MHz)",
+                "default": 1.0
+            },
+            "tx_on": {
+                "type": "int",
+                "tooltip": "发射开关",
+                "default": 0
+            },
+            "tx_frame": {
+                "type": "int",
+                "tooltip": "发射接口单次样点数",
+                "default": "10000",
+                "range": "200-10000"
+            },
+            "tx_rf": {
+                "type": "double",
+                "tooltip": "发射频率(MHz)",
+                "default": "200.0,203.0",
+                "range": "0-6000"
+            },
+            "tx_gain": {
+                "type": "double",
+                "tooltip": "发射增益(dB)",
+                "default": "30.0,30.0",
+                "range": "0-100"
+            },
+            "tx_bw": {
+                "type": "double",
+                "tooltip": "发射滤波器带宽(MHz)",
+                "default": "1.0,1.0"
+            },
+            "tx_atn": {
+                "type": "string",
+                "tooltip": "发射天线名称",
+                "default": "TX/RX,TX/RX"
+            },
+
+            "tx_channels": {
+                "type": "int",
+                "tooltip": "发射各个通道",
+                "default": "0,1"
+            }
+        },
+        "input_subjects": {
+            "wav_tx0": {
+                "type": "short[2]",
+                "tooltip": "TX0"
+            },
+            "wav_tx1": {
+                "type": "short[2]",
+                "tooltip": "TX1"
+            }
+        },
+        "output_subjects": {
+            "rx_timestamp": {
+                "type": "uint64",
+                "tooltip": "RX_tm"
+            },
+            "wav_rx0": {
+                "type": "short[2]",
+                "tooltip": "RX0"
+            },
+            "wav_rx1": {
+                "type": "short[2]",
+                "tooltip": "RX1"
+            }
+        }
+    }
+}