Prepared changes for dispatch-to-acceptor (no build)

srtcore/core.cpp

@@ -5997,6 +5997,14 @@ void CUDT::acceptAndRespond(const sockaddr_any& agent, const sockaddr_any& peer,
     m_pRNode->m_bOnList = true;
     m_pRcvQueue->setNewEntry(this);
 
+    if (!createSendHSResponse(kmdata, kmdatasize, peer, (w_hs)))
+    {
+        throw CUDTException(MJ_SETUP, MN_REJECTED, 0);
+    }
+}
+
+bool CUDT::createSendHSResponse(uint32_t* kmdata, size_t kmdatasize, const sockaddr_any& peer, CHandShake& w_hs) ATR_NOTHROW
+{
     // send the response to the peer, see listen() for more discussions about this
     // XXX Here create CONCLUSION RESPONSE with:
     // - just the UDT handshake, if HS_VERSION_UDT4,
@@ -6010,11 +6018,11 @@ void CUDT::acceptAndRespond(const sockaddr_any& agent, const sockaddr_any& peer,
 
     // This will serialize the handshake according to its current form.
     HLOGC(cnlog.Debug,
-          log << "acceptAndRespond: creating CONCLUSION response (HSv5: with HSRSP/KMRSP) buffer size=" << size);
+          log << "createSendHSResponse: creating CONCLUSION response (HSv5: with HSRSP/KMRSP) buffer size=" << size);
     if (!createSrtHandshake(SRT_CMD_HSRSP, SRT_CMD_KMRSP, kmdata, kmdatasize, (response), (w_hs)))
     {
-        LOGC(cnlog.Error, log << "acceptAndRespond: error creating handshake response");
-        throw CUDTException(MJ_SETUP, MN_REJECTED, 0);
+        LOGC(cnlog.Error, log << "createSendHSResponse: error creating handshake response");
+        return false;
     }
 
     // Set target socket ID to the value from received handshake's source ID.
@@ -6027,7 +6035,7 @@ void CUDT::acceptAndRespond(const sockaddr_any& agent, const sockaddr_any& peer,
         CHandShake debughs;
         debughs.load_from(response.m_pcData, response.getLength());
         HLOGC(cnlog.Debug,
-              log << CONID() << "acceptAndRespond: sending HS from agent @"
+              log << CONID() << "createSendHSResponse: sending HS from agent @"
                 << debughs.m_iID << " to peer @" << response.m_iID
                 << "HS:" << debughs.show());
     }
@@ -6039,6 +6047,7 @@ void CUDT::acceptAndRespond(const sockaddr_any& agent, const sockaddr_any& peer,
     // coming as connected, but continue repeated handshake until finally
     // received the listener's handshake.
     m_pSndQueue->sendto(peer, response);
+    return true;
 }
 
 // This function is required to be called when a caller receives an INDUCTION

srtcore/core.h

@@ -365,6 +365,7 @@ class CUDT
     }
 
     SRTSOCKET socketID() const { return m_SocketID; }
+    SRTSOCKET peerID() const { return m_PeerID; }
 
     static CUDT* getUDTHandle(SRTSOCKET u);
     static std::vector<SRTSOCKET> existingSockets();
@@ -599,6 +600,7 @@ class CUDT
     /// @param hs [in/out] The handshake information sent by the peer side (in), negotiated value (out).
 
     void acceptAndRespond(const sockaddr_any& agent, const sockaddr_any& peer, const CPacket& hspkt, CHandShake& hs);
+    bool createSendHSResponse(uint32_t* kmdata, size_t kmdatasize, const sockaddr_any& peer, CHandShake& w_hs) ATR_NOTHROW;
     bool runAcceptHook(CUDT* acore, const sockaddr* peer, const CHandShake& hs, const CPacket& hspkt);
 
     /// Close the opened UDT entity.

srtcore/queue.cpp

@@ -773,16 +773,27 @@ CUDT *CHash::lookup(int32_t id)
     return NULL;
 }
 
+CUDT* CHash::lookupPeer(int32_t peerid)
+{
+    // Decode back the socket ID if it has that peer
+    int32_t id = map_get(m_RevPeerMap, peerid, -1);
+    if (id == -1)
+        return NULL; // no such peer id
+    return lookup(id);
+}
+
 void CHash::insert(int32_t id, CUDT *u)
 {
     CBucket *b = m_pBucket[id % m_iHashSize];
 
     CBucket *n = new CBucket;
     n->m_iID   = id;
+    n->m_iPeerID = u->peerID();
     n->m_pUDT  = u;
     n->m_pNext = b;
 
     m_pBucket[id % m_iHashSize] = n;
+    m_RevPeerMap[u->peerID()] = id;
 }
 
 void CHash::remove(int32_t id)
@@ -799,6 +810,7 @@ void CHash::remove(int32_t id)
             else
                 p->m_pNext = b->m_pNext;
 
+            m_RevPeerMap.erase(b->m_iPeerID);
             delete b;
 
             return;
@@ -1402,11 +1414,44 @@ EConnectStatus CRcvQueue::worker_ProcessConnectionRequest(CUnit* unit, const soc
                  << " result:" << RequestTypeStr(UDTRequestType(listener_ret)));
         return listener_ret == SRT_REJ_UNKNOWN ? CONN_CONTINUE : CONN_REJECT;
     }
+    else
+    {
+        if (worker_TryAcceptedSocket(unit, addr))
+            return CONN_CONTINUE;
+    }
 
     // If there's no listener waiting for the packet, just store it into the queue.
     return worker_TryAsyncRend_OrStore(0, unit, addr); // 0 id because the packet came in with that very ID.
 }
 
+bool CRcvQueue::worker_TryAcceptedSocket(CUnit* unit, const sockaddr_any& addr)
+{
+    // We are working with a possibly HS packet... check that.
+    CPacket& pkt = unit->m_Packet;
+
+    if (pkt.getLength() < CHandShake::m_iContentSize || !pkt.isControl(UMSG_HANDSHAKE))
+        return false;
+
+    CHandShake hs;
+    if (0 != hs.load_from(pkt.data(), pkt.size()))
+        return false;
+
+    if (hs.m_iReqType != URQ_CONCLUSION)
+        return false;
+
+    // Ok, at last we have a peer ID info
+    int32_t peerid = hs.m_iID;
+
+    // Now search for a socket that has this peer ID
+    CUDT* u = m_pHash->lookupPeer(peerid);
+    if (!u)
+        return false; // no socket has that peer in this multiplexer
+
+    // CRAFT KMX DATA FOR RESPONSE
+
+    return u->createSendHSResponse(kmdata, kmdatasize, (hs));
+}
+
 EConnectStatus CRcvQueue::worker_ProcessAddressedPacket(int32_t id, CUnit* unit, const sockaddr_any& addr)
 {
     CUDT *u = m_pHash->lookup(id);

srtcore/queue.h

@@ -292,6 +292,12 @@ class CHash
 
    CUDT* lookup(int32_t id);
 
+     /// Look for a UDT instance from the hash table by source ID
+     /// @param [in] peerid socket ID of the peer reported as source ID
+     /// @return Pointer to a UDT instance where m_PeerID == peerid, or NULL if not found
+
+   CUDT* lookupPeer(int32_t peerid);
+
       /// Insert an entry to the hash table.
       /// @param [in] id socket ID
       /// @param [in] u pointer to the UDT instance
@@ -307,13 +313,16 @@ class CHash
    struct CBucket
    {
       int32_t m_iID;		// Socket ID
+      int32_t m_iPeerID;    // Peer ID
       CUDT* m_pUDT;		// Socket instance
 
       CBucket* m_pNext;		// next bucket
    } **m_pBucket;		// list of buckets (the hash table)
 
    int m_iHashSize;		// size of hash table
 
+   std::map<int32_t, int32_t> m_RevPeerMap;
+
 private:
    CHash(const CHash&);
    CHash& operator=(const CHash&);
@@ -489,6 +498,7 @@ friend class CUDTUnited;
    EConnectStatus worker_ProcessConnectionRequest(CUnit* unit, const sockaddr_any& sa);
    EConnectStatus worker_TryAsyncRend_OrStore(int32_t id, CUnit* unit, const sockaddr_any& sa);
    EConnectStatus worker_ProcessAddressedPacket(int32_t id, CUnit* unit, const sockaddr_any& sa);
+   bool worker_TryAcceptedSocket(CUnit* unit, const sockaddr_any& addr);
 
 private:
    CUnitQueue m_UnitQueue;      // The received packet queue