Fix permanent branch bug

  perm branches were getting duplicate messages in Hub and Trunk mode
  Prune mode wasn't affected, that's why the original test didn't pick it up.

Author: neilstephens

src/MiniPlex.cpp

@@ -29,8 +29,10 @@ MiniPlex::MiniPlex(const CmdArgs& Args, asio::io_context& IOC):
 	socket(IOC,local_ep),
 	socket_strand(IOC),
 	process_strand(IOC),
-	EndPointCache(process_strand,Args.CacheTimeout.getValue(),[](const asio::ip::udp::endpoint& ep)
+	ActiveBranches(process_strand,Args.CacheTimeout.getValue(),[this](const asio::ip::udp::endpoint& ep)
 	{
+		if(PermaBranches.contains(ep))
+			InactivePermaBranches.insert(ep);
 		auto ep_string = ep.address().to_string()+":"+std::to_string(ep.port());
 		spdlog::get("MiniPlex")->debug("Cache entry for {} timed out.",ep_string);
 	})
@@ -65,7 +67,8 @@ MiniPlex::MiniPlex(const CmdArgs& Args, asio::io_context& IOC):
 	for(size_t i=0; i<Args.BranchAddrs.getValue().size(); i++)
 	{
 		auto branch = asio::ip::udp::endpoint(asio::ip::address::from_string(Args.BranchAddrs.getValue()[i]),Args.BranchPorts.getValue()[i]);
-		PermaBranches.emplace_back(branch);
+		PermaBranches.insert(branch);
+		InactivePermaBranches.insert(branch);
 	}
 
 	socket_strand.post([this](){Rcv();});
@@ -117,17 +120,17 @@ void MiniPlex::RcvHandler(const asio::error_code err, const uint8_t* const buf,
 void MiniPlex::Hub(const std::list<asio::ip::udp::endpoint>& branches, const asio::ip::udp::endpoint& rcv_sender, const uint8_t* const buf, const size_t n)
 {
 	auto pForwardBuf = MakeSharedBuf(buf,n);
-	Forward(pForwardBuf,n,rcv_sender,PermaBranches,"fixed branches");
-	Forward(pForwardBuf,n,rcv_sender,branches,"cached branches");
+	Forward(pForwardBuf,n,rcv_sender,branches,"active branches");
+	Forward(pForwardBuf,n,rcv_sender,InactivePermaBranches,"inactive fixed branches");
 }
 
 void MiniPlex::Trunk(const std::list<asio::ip::udp::endpoint>& branches, const asio::ip::udp::endpoint& rcv_sender, const uint8_t* const buf, const size_t n)
 {
 	auto pForwardBuf = MakeSharedBuf(buf,n);
 	if(rcv_sender == trunk)
 	{
-		Forward(pForwardBuf,n,rcv_sender,PermaBranches,"fixed branches");
-		Forward(pForwardBuf,n,rcv_sender,branches,"cached branches");
+		Forward(pForwardBuf,n,rcv_sender,branches,"active branches");
+		Forward(pForwardBuf,n,rcv_sender,InactivePermaBranches,"inactive fixed branches");
 	}
 	else
 		Forward(pForwardBuf,n,rcv_sender,std::list{trunk},"trunk");
@@ -150,7 +153,7 @@ void MiniPlex::Prune(const std::list<asio::ip::udp::endpoint>& branches, const a
 			Forward(pForwardBuf,n,rcv_sender,std::list{*branches.begin()},"active prune branch");
 	}
 	else
-		Forward(pForwardBuf,n,rcv_sender,std::list{trunk},"pruned branch");
+		Forward(pForwardBuf,n,rcv_sender,std::list{trunk},"trunk");
 }
 
 inline std::shared_ptr<uint8_t> MiniPlex::MakeSharedBuf(const uint8_t* const buf, const size_t n)
@@ -163,11 +166,11 @@ inline std::shared_ptr<uint8_t> MiniPlex::MakeSharedBuf(const uint8_t* const buf
 	return pForwardBuf;
 }
 
-void MiniPlex::Forward(
+template<typename T> void MiniPlex::Forward(
 	const std::shared_ptr<uint8_t>& pBuf,
 	const size_t size,
 	const asio::ip::udp::endpoint& sender,
-	const std::list<asio::ip::udp::endpoint>& branches,
+	const T& branches,
 	const char* desc)
 {
 	spdlog::get("MiniPlex")->trace("Forward(): sending to {} {}",branches.size(),desc);
@@ -184,10 +187,12 @@ const std::list<asio::ip::udp::endpoint>& MiniPlex::Branches(const asio::ip::udp
 {
 	if(ep != trunk)
 	{
-		auto added = EndPointCache.Add(ep);
+		auto added = ActiveBranches.Add(ep);
 		spdlog::get("MiniPlex")->trace("RcvHandler(): {} cache entry for {}",added?"Inserted":"Refreshed",sender_string);
+		if(added)
+			InactivePermaBranches.erase(ep);
 	}
-	return EndPointCache.Keys();
+	return ActiveBranches.Keys();
 }
 
 void MiniPlex::Benchmark()

src/MiniPlex.h

@@ -23,6 +23,7 @@
 #include <atomic>
 #include <deque>
 #include <list>
+#include <set>
 #include <functional>
 
 struct CmdArgs;
@@ -39,11 +40,11 @@ class MiniPlex
 	void Rcv();
 	void RcvHandler(const asio::error_code err, const uint8_t* const buf, const asio::ip::udp::endpoint& rcv_sender, const size_t n);
 	inline std::shared_ptr<uint8_t> MakeSharedBuf(const uint8_t* const buf, const size_t n);
-	void Forward(
+	template<typename T> void Forward(
 		const std::shared_ptr<uint8_t>& pBuf,
 		const size_t size,
 		const asio::ip::udp::endpoint& sender,
-		const std::list<asio::ip::udp::endpoint>& branches,
+		const T& branches,
 		const char* desc);
 	const std::list<asio::ip::udp::endpoint>& Branches(const asio::ip::udp::endpoint& ep, const std::string& sender_string);
 
@@ -58,8 +59,9 @@ class MiniPlex
 	asio::ip::udp::socket socket;
 	asio::io_context::strand socket_strand;
 	asio::io_context::strand process_strand;
-	TimeoutCache<asio::ip::udp::endpoint> EndPointCache;
-	std::list<asio::ip::udp::endpoint> PermaBranches;
+	std::set<asio::ip::udp::endpoint> PermaBranches;
+	TimeoutCache<asio::ip::udp::endpoint> ActiveBranches;
+	std::set<asio::ip::udp::endpoint> InactivePermaBranches;
 	asio::ip::udp::endpoint trunk;
 
 	std::deque<std::shared_ptr<rbuf_t>> rcv_buf_q;