Adding support for bonding

.gitignore

@@ -12,7 +12,7 @@
 
 # Build results
 _build*/
-build/
+build*/
 [Dd]ebug/
 [Dd]ebugPublic/
 [Rr]elease/

CMakeLists.txt

@@ -27,6 +27,7 @@ add_subdirectory(submodule/spdlog)
 SET(ENABLE_APPS ON)
 SET(ENABLE_SHARED OFF)
 SET(ENABLE_ENCRYPTION ON)
+SET(ENABLE_EXPERIMENTAL_BONDING ON)
 add_subdirectory(submodule/srt)
 
 #set_target_properties(srt-live-transmit

xtransmit/CMakeLists.txt

@@ -17,6 +17,15 @@ target_include_directories(srt-xtransmit PUBLIC
 	${PTHREAD_INCLUDE_DIR}
 	)
 
+if (ENABLE_EXPERIMENTAL_BONDING)
+	set (CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -DENABLE_EXPERIMENTAL_BONDING=1")
+	set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DENABLE_EXPERIMENTAL_BONDING=1")
+endif()
+
+if (ENABLE_STDCXX_SYNC)
+	set (CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -DENABLE_STDCXX_SYNC=1")
+	set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DENABLE_STDCXX_SYNC=1")
+endif()
 
 if (ENABLE_CXX17)
 	set(REQUIRE_CXX_VER 17)

xtransmit/generate.cpp

@@ -14,6 +14,7 @@
 // xtransmit
 #include "socket_stats.hpp"
 #include "srt_socket.hpp"
+#include "srt_socket_group.hpp"
 #include "udp_socket.hpp"
 #include "generate.hpp"
 #include "pacer.hpp"
@@ -36,7 +37,6 @@ using shared_sock = std::shared_ptr<socket::isocket>;
 void run_pipe(shared_sock dst, const config& cfg, const atomic_bool& force_break)
 {
 	vector<char> message_to_send(cfg.message_size);
-	iota(message_to_send.begin(), message_to_send.end(), (char)0);
 
 	const auto start_time   = steady_clock::now();
 	const int  num_messages = cfg.duration > 0 ? -1 : cfg.num_messages;
@@ -94,12 +94,19 @@ void run_pipe(shared_sock dst, const config& cfg, const atomic_bool& force_break
 	}
 }
 
-void xtransmit::generate::run(const string& dst_url, const config& cfg, const atomic_bool& force_break)
+void xtransmit::generate::run(const vector<string>& dst_urls, const config& cfg, const atomic_bool& force_break)
 {
-	const UriParser uri(dst_url);
+	if (dst_urls.empty())
+	{
+		spdlog::error(LOG_SC_GENERATE "No destination URI was provided");
+		return;
+	}
 
-	shared_sock sock;
-	shared_sock connection;
+	vector<UriParser> urls;
+	for (const string& url : dst_urls)
+	{
+		urls.emplace_back(UriParser(url));
+	}
 
 	const bool write_stats = cfg.stats_file != "" && cfg.stats_freq_ms > 0;
 	// make_unique is not supported by GCC 4.8, only starting from GCC 4.9 :(
@@ -121,39 +128,60 @@ void xtransmit::generate::run(const string& dst_url, const config& cfg, const at
 	do {
 		try
 		{
-			if (uri.proto() == "udp")
+			shared_sock sock;
+			shared_sock connection;
+
+			if (urls.size() == 1)
 			{
-				connection = make_shared<socket::udp>(uri);
+				if (urls[0].proto() == "udp")
+				{
+					connection = make_shared<socket::udp>(urls[0]);
+				}
+				else
+				{
+					sock = make_shared<socket::srt>(urls[0]);
+					socket::srt* s = dynamic_cast<socket::srt*>(sock.get());
+					const bool   accept = s->mode() == socket::srt::LISTENER;
+					if (accept)
+						s->listen();
+					connection = accept ? s->accept() : s->connect();
+				}
 			}
 			else
 			{
-				sock = make_shared<socket::srt>(uri);
-				socket::srt* s = static_cast<socket::srt*>(sock.get());
-				const bool   accept = s->mode() == socket::srt::LISTENER;
+				sock = make_shared<socket::srt_group>(urls);
+				socket::srt_group* s = dynamic_cast<socket::srt_group*>(sock.get());
+				const bool   accept = s->mode() == socket::srt_group::LISTENER;
 				if (accept)
+				{
 					s->listen();
+				}
 				connection = accept ? s->accept() : s->connect();
 			}
 
 			if (stats)
 				stats->add_socket(connection);
 			run_pipe(connection, cfg, force_break);
+			if (stats && cfg.reconnect)
+				stats->clear();
 		}
 		catch (const socket::exception& e)
 		{
 			spdlog::warn(LOG_SC_GENERATE "{}", e.what());
+			if (stats)
+				stats->clear();
 		}
 	} while (cfg.reconnect && !force_break);
 }
 
-CLI::App* xtransmit::generate::add_subcommand(CLI::App& app, config& cfg, string& dst_url)
+CLI::App* xtransmit::generate::add_subcommand(CLI::App& app, config& cfg, vector<string>& dst_urls)
 {
 	const map<string, int> to_bps{{"kbps", 1000}, {"Mbps", 1000000}, {"Gbps", 1000000000}};
 	const map<string, int> to_ms{{"s", 1000}, {"ms", 1}};
 	const map<string, int> to_sec{{"s", 1}, {"min", 60}, {"mins", 60}};
 
 	CLI::App* sc_generate = app.add_subcommand("generate", "Send generated data (SRT, UDP)")->fallthrough();
-	sc_generate->add_option("dst", dst_url, "Destination URI");
+	sc_generate->add_option("dst", dst_urls, "Destination URI")->expected(1, 10);
 	sc_generate->add_option("--msgsize", cfg.message_size, "Size of a message to send");
 	sc_generate->add_option("--sendrate", cfg.sendrate, "Bitrate to generate")
 		->transform(CLI::AsNumberWithUnit(to_bps, CLI::AsNumberWithUnit::CASE_SENSITIVE));

xtransmit/generate.hpp

@@ -24,8 +24,9 @@ struct config
 	std::string playback_csv;
 };
 
-void run(const std::string& dst_url, const config& cfg, const std::atomic_bool& force_break);
+void run(const std::vector<std::string>& dst_urls, const config& cfg, const std::atomic_bool& force_break);
+
+CLI::App* add_subcommand(CLI::App& app, config& cfg, std::vector<std::string>& dst_urls);
 
-CLI::App* add_subcommand(CLI::App& app, config& cfg, std::string& dst_url);
 } // namespace generate
 } // namespace xtransmit

xtransmit/metrics.hpp

@@ -63,6 +63,7 @@ namespace metrics
 		// TODO: latency measurements
 		inline void validate_packet(const vector<char>& payload)
 		{
+			// TODO: validate payload
 			const auto sys_time_now = system_clock::now();
 			const auto std_time_now = steady_clock::now();
 

xtransmit/metrics_reorder.hpp

@@ -40,16 +40,16 @@ class reorder
 		{
 			const uint64_t lost = pkt_seqno - m_stats.expected_seqno;
 			m_stats.pkts_lost += lost;
-			//spdlog::warn(LOG_SC_RFC4737 "Detected loss of {} packets", lost);
+			spdlog::warn("[METRICS] Detected loss of {} packets (seqno [{}; {}))", lost, m_stats.expected_seqno, pkt_seqno);
 			m_stats.expected_seqno = pkt_seqno + 1;
 		}
 		else // Packet reordering: pkt_seqno < m_seqno
 		{
 			++m_stats.pkts_reordered;
-			const uint64_t reorder_dist = pkt_seqno - m_stats.expected_seqno;
+			const uint64_t reorder_dist = m_stats.expected_seqno - pkt_seqno;
 			m_stats.reorder_dist = std::max(m_stats.reorder_dist, reorder_dist);
 
-			//spdlog::warn(LOG_SC_RFC4737 "Detected reordered packet, dist {}", reorder_dist);
+			spdlog::warn("[METRICS] Detected reordered packet, seqno {}, expected {} (dist {})", pkt_seqno, m_stats.expected_seqno, reorder_dist);
 		}
 	}
 

xtransmit/receive.cpp

@@ -13,6 +13,7 @@
 // xtransmit
 #include "socket_stats.hpp"
 #include "srt_socket.hpp"
+#include "srt_socket_group.hpp"
 #include "udp_socket.hpp"
 #include "receive.hpp"
 #include "metrics.hpp"
@@ -32,8 +33,6 @@ using shared_sock = std::shared_ptr<socket::isocket>;
 #define LOG_SC_RECEIVE "RECEIVE "
 
 
-
-
 void trace_message(const size_t bytes, const vector<char> &buffer, int conn_id)
 {
 	::cout << "RECEIVED MESSAGE length " << bytes << " on conn ID " << conn_id;
@@ -137,12 +136,19 @@ void run_pipe(shared_sock src, const config &cfg, const atomic_bool &force_break
 	}
 }
 
-void xtransmit::receive::run(const string &src_url, const config &cfg, const atomic_bool &force_break)
+void xtransmit::receive::run(const vector<string> &src_urls, const config &cfg, const atomic_bool &force_break)
 {
-	const UriParser uri(src_url);
+	if (src_urls.empty())
+	{
+		spdlog::error(LOG_SC_RECEIVE "No destination URI was provided");
+		return;
+	}
 
-	shared_sock sock;
-	shared_sock conn;
+	vector<UriParser> urls;
+	for (const string& url : src_urls)
+	{
+		urls.emplace_back(url);
+	}
 
 	unique_ptr<socket::stats_writer> stats;
 
@@ -164,17 +170,33 @@ void xtransmit::receive::run(const string &src_url, const config &cfg, const ato
 	do {
 		try
 		{
-			if (uri.proto() == "udp")
+			shared_sock sock;
+			shared_sock conn;
+
+			if (urls.size() == 1)
 			{
-				conn = make_shared<socket::udp>(uri);
+				if (urls[0].proto() == "udp")
+				{
+					conn = make_shared<socket::udp>(urls[0]);
+				}
+				else
+				{
+					sock = make_shared<socket::srt>(urls[0]);
+					socket::srt* s = dynamic_cast<socket::srt*>(sock.get());
+					const bool   accept = s->mode() == socket::srt::LISTENER;
+					if (accept)
+						s->listen();
+					conn = accept ? s->accept() : s->connect();
+				}
 			}
 			else
 			{
-				sock = make_shared<socket::srt>(uri);
-				socket::srt* s = static_cast<socket::srt*>(sock.get());
-				const bool  accept = s->mode() == socket::srt::LISTENER;
-				if (accept)
+				sock = make_shared<socket::srt_group>(urls);
+				socket::srt_group* s = dynamic_cast<socket::srt_group*>(sock.get());
+				const bool   accept = s->mode() == socket::srt_group::LISTENER;
+				if (accept) {
 					s->listen();
+				}
 				conn = accept ? s->accept() : s->connect();
 			}
 
@@ -187,16 +209,18 @@ void xtransmit::receive::run(const string &src_url, const config &cfg, const ato
 		catch (const socket::exception & e)
 		{
 			spdlog::warn(LOG_SC_RECEIVE "{}", e.what());
+			if (stats)
+				stats->clear();
 		}
 	} while (cfg.reconnect && !force_break);
 }
 
-CLI::App* xtransmit::receive::add_subcommand(CLI::App& app, config& cfg, string& src_url)
+CLI::App* xtransmit::receive::add_subcommand(CLI::App& app, config& cfg, vector<string>& src_urls)
 {
 	const map<string, int> to_ms{ {"s", 1000}, {"ms", 1} };
 
 	CLI::App* sc_receive = app.add_subcommand("receive", "Receive data (SRT, UDP)")->fallthrough();
-	sc_receive->add_option("src", src_url, "Source URI");
+	sc_receive->add_option("--input,-i,src", src_urls, "Source URI");
 	sc_receive->add_option("--msgsize", cfg.message_size, "Size of a buffer to receive message payload");
 	sc_receive->add_option("--statsfile", cfg.stats_file, "output stats report filename");
 	sc_receive->add_option("--statsfreq", cfg.stats_freq_ms, "output stats report frequency (ms)")
@@ -208,6 +232,7 @@ CLI::App* xtransmit::receive::add_subcommand(CLI::App& app, config& cfg, string&
 	sc_receive->add_option("--metricsfreq", cfg.metrics_freq_ms, "Metrics report frequency")
 		->transform(CLI::AsNumberWithUnit(to_ms, CLI::AsNumberWithUnit::CASE_SENSITIVE));
 	sc_receive->add_flag("--twoway", cfg.send_reply, "Both send and receive data");
+	sc_receive->add_option("--input-group", cfg.inputs, "More input group URLs for SRT bonding");
 
 	return sc_receive;
 }

xtransmit/receive.hpp

@@ -1,6 +1,7 @@
 #pragma once
 #include <atomic>
 #include <string>
+#include <vector>
 
 // Third party libraries
 #include "CLI/CLI.hpp"
@@ -21,14 +22,15 @@ namespace xtransmit {
 			int message_size			= 1316;
 			int stats_freq_ms			= 0;
 			std::string stats_file;
+			std::vector<std::string> inputs;
 		};
 
 
 
-		void run(const std::string& url, const config& cfg,
+		void run(const std::vector<std::string>& urls, const config& cfg,
 			const std::atomic_bool& force_break);
 
-		CLI::App* add_subcommand(CLI::App& app, config& cfg, std::string& src_url);
+		CLI::App* add_subcommand(CLI::App& app, config& cfg, std::vector<std::string>& src_urls);
 
 
 	} // namespace receive

xtransmit/route.cpp

@@ -78,7 +78,7 @@ namespace route
 		if(uri.proto() == "srt")
 		{
 			shared_sock socket = make_shared<socket::srt>(uri);
-			socket::srt* s = static_cast<socket::srt*>(socket.get());
+			socket::srt* s = dynamic_cast<socket::srt*>(socket.get());
 			const bool  accept = s->mode() == socket::srt::LISTENER;
 			if (accept)
 				s->listen();
@@ -95,6 +95,12 @@ namespace route
 void xtransmit::route::run(const string& src_url, const string& dst_url,
 	const config& cfg, const atomic_bool& force_break)
 {
+	if (src_url.empty() || dst_url.empty())
+	{
+		spdlog::error(LOG_SC_ROUTE "Empty source/destination was provided");
+		return;
+	}
+
 	try {
 		const bool write_stats = cfg.stats_file != "" && cfg.stats_freq_ms > 0;
 		// make_unique is not supported by GCC 4.8, only starting from GCC 4.9 :(
@@ -130,8 +136,8 @@ CLI::App* xtransmit::route::add_subcommand(CLI::App& app, config& cfg, string& s
 	const map<string, int> to_ms{ {"s", 1000}, {"ms", 1} };
 
 	CLI::App* sc_route = app.add_subcommand("route", "Route data (SRT, UDP)")->fallthrough();
-	sc_route->add_option("src", src_url, "Source URI");
-	sc_route->add_option("dst", dst_url, "Destination URI");
+	sc_route->add_option("-i,src", src_url, "Source URI")->expected(1);
+	sc_route->add_option("-o,dst", dst_url, "Destination URI")->expected(1);
 	sc_route->add_option("--msgsize", cfg.message_size, "Size of a buffer to receive message payload");
 	sc_route->add_flag("--bidir", cfg.bidir, "Enable bidirectional transmission");
 	sc_route->add_option("--statsfile", cfg.stats_file, "output stats report filename");