Improve multicast in non-ideal cases with connectivity with some of peers

constants.xml

@@ -199,6 +199,7 @@
         <BLACKLIST_NUM_TO_POP>5</BLACKLIST_NUM_TO_POP>
         <MAX_PEER_CONNECTION>100</MAX_PEER_CONNECTION>
         <MAX_WHITELISTREQ_LIMIT>5</MAX_WHITELISTREQ_LIMIT>
+        <SENDJOBPEERS_TIMEOUT>5</SENDJOBPEERS_TIMEOUT>
     </p2pcomm>
     <pow>
         <CUDA_GPU_MINE>false</CUDA_GPU_MINE>

constants_local.xml

@@ -198,6 +198,7 @@
         <BLACKLIST_NUM_TO_POP>1</BLACKLIST_NUM_TO_POP>
         <MAX_PEER_CONNECTION>100</MAX_PEER_CONNECTION>
         <MAX_WHITELISTREQ_LIMIT>5</MAX_WHITELISTREQ_LIMIT>
+        <SENDJOBPEERS_TIMEOUT>5</SENDJOBPEERS_TIMEOUT>
     </p2pcomm>
     <pow>
         <CUDA_GPU_MINE>false</CUDA_GPU_MINE>

src/common/Constants.cpp

@@ -390,6 +390,8 @@ const unsigned int MAX_PEER_CONNECTION{
     ReadConstantNumeric("MAX_PEER_CONNECTION", "node.p2pcomm.")};
 const unsigned int MAX_WHITELISTREQ_LIMIT{
     ReadConstantNumeric("MAX_WHITELISTREQ_LIMIT", "node.p2pcomm.")};
+const unsigned int SENDJOBPEERS_TIMEOUT{
+    ReadConstantNumeric("SENDJOBPEERS_TIMEOUT", "node.p2pcomm.")};
 
 // PoW constants
 const bool CUDA_GPU_MINE{ReadConstantString("CUDA_GPU_MINE", "node.pow.") ==

src/common/Constants.h

@@ -304,6 +304,7 @@ extern const unsigned int MAX_READ_WATERMARK_IN_BYTES;
 extern const unsigned int BLACKLIST_NUM_TO_POP;
 extern const unsigned int MAX_PEER_CONNECTION;
 extern const unsigned int MAX_WHITELISTREQ_LIMIT;
+extern const unsigned int SENDJOBPEERS_TIMEOUT;
 
 // PoW constants
 extern const bool CUDA_GPU_MINE;

src/libNetwork/P2PComm.cpp

@@ -31,6 +31,7 @@
 #include <sys/socket.h>
 #include <unistd.h>
 #include <cstring>
+#include <future>
 #include <memory>
 #include <utility>
 
@@ -345,6 +346,7 @@ void SendJobPeer::DoSend() {
 
 template <class T>
 void SendJobPeers<T>::DoSend() {
+  LOG_MARKER();
   vector<unsigned int> indexes(m_peers.size());
 
   for (unsigned int i = 0; i < indexes.size(); i++) {
@@ -362,6 +364,7 @@ void SendJobPeers<T>::DoSend() {
                          << hashStr.substr(0, 6) << "] BEGN");
   }
 
+  vector<std::shared_ptr<std::future<void>>> runningSendTasks;
   for (vector<unsigned int>::const_iterator curr = indexes.begin();
        curr < indexes.end(); ++curr) {
     const Peer& peer = m_peers.at(*curr);
@@ -373,9 +376,24 @@ void SendJobPeers<T>::DoSend() {
       continue;
     }
 
-    SendMessageCore(peer, m_message, m_startbyte, m_hash);
+    // Launch the thread.
+    auto hThread = make_shared<std::future<void>>(
+        std::async(std::launch::async, SendMessageCore, peer, m_message,
+                   m_startbyte, m_hash));
+
+    // Park the thread if message sending taking more than expected time
+    auto status = hThread->wait_for(std::chrono::seconds(SENDJOBPEERS_TIMEOUT));
+
+    if (status == std::future_status::timeout) {
+      LOG_GENERAL(WARNING,
+                  "Sending delayed for " << peer.GetPrintableIPAddress());
+      runningSendTasks.push_back(hThread);
+    }
   }
 
+  // If any threads in runningSendTask exist, it will be blocked until it
+  // completes.
+
   if ((m_startbyte == START_BYTE_BROADCAST) && (m_selfPeer != Peer())) {
     LOG_STATE("[BROAD][" << std::setw(15) << std::left
                          << m_selfPeer.GetPrintableIPAddress() << "]["