[core] Fixed problem with possibly conflicting peer group id

srtcore/api.cpp

@@ -432,7 +432,7 @@ SRTSOCKET CUDTUnited::generateSocketID(bool for_group)
     return sockval;
 }
 
-SRTSOCKET CUDTUnited::newSocket(CUDTSocket** pps)
+SRTSOCKET CUDTUnited::newSocket(CUDTSocket** pps, SRTSOCKET forceid)
 {
    // XXX consider using some replacement of std::unique_ptr
    // so that exceptions will clean up the object without the
@@ -450,14 +450,25 @@ SRTSOCKET CUDTUnited::newSocket(CUDTSocket** pps)
       throw CUDTException(MJ_SYSTEMRES, MN_MEMORY, 0);
    }
 
-   try
+   if (forceid == SRT_INVALID_SOCK)
    {
-      ns->m_SocketID = generateSocketID();
+       try
+       {
+           ns->m_SocketID = generateSocketID();
+       }
+       catch (...)
+       {
+           delete ns;
+           throw;
+       }
    }
-   catch (...)
+   else
    {
-       delete ns;
-       throw;
+#if ENABLE_EXPERIMENTAL_BONDING
+       if (IsSet(forceid, SRTGROUP_MASK))
+           throw CUDTException(MJ_NOTSUP, MN_INVAL, 0);
+#endif
+       ns->m_SocketID = forceid;
    }
    ns->m_Status = SRTS_INIT;
    ns->m_ListenSocket = 0;
@@ -3107,14 +3118,14 @@ int CUDT::cleanup()
    return s_UDTUnited.cleanup();
 }
 
-SRTSOCKET CUDT::socket()
+SRTSOCKET CUDT::socket(SRTSOCKET forceid)
 {
    if (!s_UDTUnited.m_bGCStatus)
       s_UDTUnited.startup();
 
    try
    {
-      return s_UDTUnited.newSocket();
+      return s_UDTUnited.newSocket(NULL, forceid);
    }
    catch (const CUDTException& e)
    {
@@ -3151,15 +3162,15 @@ CUDT::APIError::APIError(CodeMajor mj, CodeMinor mn, int syserr)
 // This doesn't have argument of GroupType due to header file conflicts.
 
 // [[using locked(s_UDTUnited.m_GlobControlLock)]]
-CUDTGroup& CUDT::newGroup(const int type)
+CUDTGroup& CUDT::newGroup(const int type, const SRTSOCKET forceid)
 {
-    const SRTSOCKET id = s_UDTUnited.generateSocketID(true);
+    const SRTSOCKET id = (forceid == SRT_INVALID_SOCK) ? s_UDTUnited.generateSocketID(true): forceid;
 
     // Now map the group
     return s_UDTUnited.addGroup(id, SRT_GROUP_TYPE(type)).set_id(id);
 }
 
-SRTSOCKET CUDT::createGroup(SRT_GROUP_TYPE gt)
+SRTSOCKET CUDT::createGroup(SRT_GROUP_TYPE gt, const SRTSOCKET forceid)
 {
     // Doing the same lazy-startup as with srt_create_socket()
     if (!s_UDTUnited.m_bGCStatus)
@@ -3168,7 +3179,7 @@ SRTSOCKET CUDT::createGroup(SRT_GROUP_TYPE gt)
     try
     {
         srt::sync::ScopedLock globlock (s_UDTUnited.m_GlobControlLock);
-        return newGroup(gt).id();
+        return newGroup(gt, forceid).id();
         // Note: potentially, after this function exits, the group
         // could be deleted, immediately, from a separate thread (tho
         // unlikely because the other thread would need some handle to

srtcore/api.h

@@ -224,7 +224,7 @@ friend class CRendezvousQueue;
       /// @param [out] pps Variable (optional) to which the new socket will be written, if succeeded
       /// @return The new UDT socket ID, or INVALID_SOCK.
 
-   SRTSOCKET newSocket(CUDTSocket** pps = NULL);
+   SRTSOCKET newSocket(CUDTSocket** pps = NULL, SRTSOCKET forceid = SRT_INVALID_SOCK);
 
       /// Create a new UDT connection.
       /// @param [in] listen the listening UDT socket;
@@ -307,7 +307,7 @@ friend class CRendezvousQueue;
    void deleteGroup(CUDTGroup* g);
 
    // [[using locked(m_GlobControlLock)]]
-   CUDTGroup* findPeerGroup_LOCKED(SRTSOCKET peergroup)
+   CUDTGroup* findPeerGroup_LOCKED(PeerGroupType peergroup)
    {
        for (groups_t::iterator i = m_Groups.begin();
                i != m_Groups.end(); ++i)

srtcore/core.cpp

@@ -1244,10 +1244,21 @@ size_t CUDT::fillHsExtGroup(uint32_t* pcmdspec)
         | SrtHSRequest::HS_GROUP_FLAGS::wrap(flags)
         | SrtHSRequest::HS_GROUP_WEIGHT::wrap(m_parent->m_GroupMemberData->weight);
 
-    const uint32_t storedata [GRPD_E_SIZE] = { uint32_t(id), dataword };
-    memcpy((space), storedata, sizeof storedata);
+    uint32_t peerapp = m_parent->m_GroupOf->peerid().appid;
+    uint32_t storedata [GRPD_E_SIZE] = { uint32_t(id), dataword, srt::sync::getProcessID() };
+
+    bool respond_to_old = m_SrtHsSide == HSD_RESPONDER && peerapp == 0;
+
+    size_t ra_size =
+        respond_to_old
+        ? GRPD_E_SIZE_V1  // backward-compatible
+        : GRPD_E_SIZE;
+
+    size_t memsize = ra_size * GRPD_FIELD_SIZE;
+    memcpy((space), storedata, memsize);
+
+    HLOGC(cnlog.Debug, log << "fillHsExtGroup: size=" << ra_size << " peerappid=" << peerapp);
 
-    const size_t ra_size = Size(storedata);
     *pcmdspec = HS_CMDSPEC_CMD::wrap(SRT_CMD_GROUP) | HS_CMDSPEC_SIZE::wrap(ra_size);
 
     return ra_size;
@@ -2654,7 +2665,7 @@ bool CUDT::interpretSrtHandshake(const CHandShake& hs,
                 }
             }
 #if ENABLE_EXPERIMENTAL_BONDING
-            else if ( cmd == SRT_CMD_GROUP )
+            else if (cmd == SRT_CMD_GROUP)
             {
                 // Note that this will fire in both cases:
                 // - When receiving HS request from the Initiator, which belongs to a group, and agent must
@@ -2663,7 +2674,7 @@ bool CUDT::interpretSrtHandshake(const CHandShake& hs,
                 // - When receiving HS response from the Responder, with its mirror group ID, so the agent
                 //   must put the group into his peer group data
                 int32_t groupdata[GRPD_E_SIZE] = {};
-                if (bytelen < GRPD_MIN_SIZE * GRPD_FIELD_SIZE || bytelen % GRPD_FIELD_SIZE || blocklen > GRPD_E_SIZE)
+                if (bytelen < GRPD_MIN_SIZE * GRPD_FIELD_SIZE || bytelen % GRPD_FIELD_SIZE)
                 {
                     m_RejectReason = SRT_REJ_ROGUE;
                     LOGC(cnlog.Error, log << "PEER'S GROUP wrong size: " << (bytelen/GRPD_FIELD_SIZE));
@@ -2835,7 +2846,7 @@ bool CUDT::checkApplyFilterConfig(const std::string &confstr)
 }
 
 #if ENABLE_EXPERIMENTAL_BONDING
-bool CUDT::interpretGroup(const int32_t groupdata[], size_t data_size SRT_ATR_UNUSED, int hsreq_type_cmd SRT_ATR_UNUSED)
+bool CUDT::interpretGroup(const int32_t groupdata[], size_t data_size, int hsreq_type_cmd SRT_ATR_UNUSED)
 {
     // `data_size` isn't checked because we believe it's checked earlier.
     // Also this code doesn't predict to get any other format than the official one,
@@ -2846,6 +2857,10 @@ bool CUDT::interpretGroup(const int32_t groupdata[], size_t data_size SRT_ATR_UN
     SRTSOCKET grpid = groupdata[GRPD_GROUPID];
     uint32_t gd = groupdata[GRPD_GROUPDATA];
 
+    uint32_t appid = 0;
+    if (data_size > GRPD_E_SIZE_V1)
+        appid = groupdata[GRPD_APPID];
+
     SRT_GROUP_TYPE gtp = SRT_GROUP_TYPE(SrtHSRequest::HS_GROUP_TYPE::unwrap(gd));
     int link_weight = SrtHSRequest::HS_GROUP_WEIGHT::unwrap(gd);
     uint32_t link_flags = SrtHSRequest::HS_GROUP_FLAGS::unwrap(gd);
@@ -2896,6 +2911,8 @@ bool CUDT::interpretGroup(const int32_t groupdata[], size_t data_size SRT_ATR_UN
         return false;
     }
 
+    PeerGroupType incoming_peer (grpid, appid, m_PeerAddr);
+
     ScopedLock guard_group_existence (s_UDTUnited.m_GlobControlLock);
 
     if (m_SrtHsSide == HSD_INITIATOR)
@@ -2926,20 +2943,20 @@ bool CUDT::interpretGroup(const int32_t groupdata[], size_t data_size SRT_ATR_UN
             return false;
         }
 
-        SRTSOCKET peer = pg->peerid();
-        if (peer == -1)
+        PeerGroupType peer = pg->peerid();
+        if (peer.empty())
         {
             // This is the first connection within this group, so this group
             // has just been informed about the peer membership. Accept it.
-            pg->set_peerid(grpid);
+            pg->set_peerid(incoming_peer);
             HLOGC(cnlog.Debug, log << "HS/RSP: group $" << pg->id() << " mapped to peer mirror $" << pg->peerid());
         }
         // Otherwise the peer id must be the same as existing, otherwise
         // this group is considered already bound to another peer group.
         // (Note that the peer group is peer-specific, and peer id numbers
         // may repeat among sockets connected to groups established on
         // different peers).
-        else if (pg->peerid() != grpid)
+        else if (pg->peerid() != incoming_peer)
         {
             LOGC(cnlog.Error, log << "IPE: HS/RSP: group membership responded for peer $" << grpid
                     << " but the current socket's group $" << pg->id() << " has already a peer $" << peer);
@@ -2958,7 +2975,7 @@ bool CUDT::interpretGroup(const int32_t groupdata[], size_t data_size SRT_ATR_UN
         // and its group ID will be added to the HS extensions as mirror group
         // ID to the peer.
 
-        SRTSOCKET lgid = makeMePeerOf(grpid, gtp, link_flags);
+        SRTSOCKET lgid = makeMePeerOf(incoming_peer, gtp, link_flags);
         if (!lgid)
             return true; // already done
 
@@ -2998,7 +3015,7 @@ bool CUDT::interpretGroup(const int32_t groupdata[], size_t data_size SRT_ATR_UN
 // exclusively on the listener side (HSD_RESPONDER, HSv5+).
 
 // [[using locked(s_UDTUnited.m_GlobControlLock)]]
-SRTSOCKET CUDT::makeMePeerOf(SRTSOCKET peergroup, SRT_GROUP_TYPE gtp, uint32_t link_flags)
+SRTSOCKET CUDT::makeMePeerOf(PeerGroupType peergroup, SRT_GROUP_TYPE gtp, uint32_t link_flags)
 {
     // Note: This function will lock pg->m_GroupLock!
 

srtcore/core.h

@@ -124,6 +124,8 @@ enum GroupDataItem
     GRPD_GROUPID,
     GRPD_GROUPDATA,
 
+    GRPD_APPID, GRPD_E_SIZE_V1  = GRPD_APPID, // size in previous version
+
     GRPD_E_SIZE
 };
 
@@ -141,6 +143,7 @@ enum SeqPairItems
 
 #if ENABLE_EXPERIMENTAL_BONDING
 class CUDTGroup;
+struct PeerGroupType;
 #endif
 
 // Extended SRT Congestion control class - only an incomplete definition required
@@ -185,9 +188,9 @@ class CUDT
 public: //API
     static int startup();
     static int cleanup();
-    static SRTSOCKET socket();
+    static SRTSOCKET socket(SRTSOCKET forceid = SRT_INVALID_SOCK);
 #if ENABLE_EXPERIMENTAL_BONDING
-    static SRTSOCKET createGroup(SRT_GROUP_TYPE);
+    static SRTSOCKET createGroup(SRT_GROUP_TYPE, const SRTSOCKET forceid = SRT_INVALID_SOCK);
     static int addSocketToGroup(SRTSOCKET socket, SRTSOCKET group);
     static int removeSocketFromGroup(SRTSOCKET socket);
     static SRTSOCKET getGroupOfSocket(SRTSOCKET socket);
@@ -512,11 +515,11 @@ class CUDT
     SRT_ATR_NODISCARD bool checkApplyFilterConfig(const std::string& cs);
 
 #if ENABLE_EXPERIMENTAL_BONDING
-    static CUDTGroup& newGroup(const int); // defined EXCEPTIONALLY in api.cpp for convenience reasons
+    static CUDTGroup& newGroup(const int, SRTSOCKET forceid = SRT_INVALID_SOCK); // defined EXCEPTIONALLY in api.cpp for convenience reasons
     // Note: This is an "interpret" function, which should treat the tp as
     // "possibly group type" that might be out of the existing values.
     SRT_ATR_NODISCARD bool interpretGroup(const int32_t grpdata[], size_t data_size, int hsreq_type_cmd);
-    SRT_ATR_NODISCARD SRTSOCKET makeMePeerOf(SRTSOCKET peergroup, SRT_GROUP_TYPE tp, uint32_t link_flags);
+    SRT_ATR_NODISCARD SRTSOCKET makeMePeerOf(PeerGroupType peergroup, SRT_GROUP_TYPE tp, uint32_t link_flags);
     void synchronizeWithGroup(CUDTGroup* grp);
 #endif
 

srtcore/group.cpp

@@ -282,7 +282,6 @@ CUDTGroup::SocketData CUDTGroup::prepareData(CUDTSocket* s)
 CUDTGroup::CUDTGroup(SRT_GROUP_TYPE gtype)
     : m_pGlobal(&CUDT::s_UDTUnited)
     , m_GroupID(-1)
-    , m_PeerGroupID(-1)
     , m_selfManaged(true)
     , m_bSyncOnMsgNo(false)
     , m_type(gtype)
@@ -989,7 +988,7 @@ void CUDTGroup::close()
         // removing themselves from the group when closing because they
         // are unaware of being group members.
         m_Group.clear();
-        m_PeerGroupID = -1;
+        m_PeerGroupID = PeerGroupType();
 
         set<int> epollid;
         {

srtcore/group.h

@@ -24,6 +24,42 @@ Written by
 const char* const srt_log_grp_state[] = {"PENDING", "IDLE", "RUNNING", "BROKEN"};
 #endif
 
+struct PeerGroupType
+{
+    SRTSOCKET groupid;
+    uint32_t  appid;
+    sockaddr_any addr;
+
+    PeerGroupType(uint32_t gid, uint32_t aid, const sockaddr_any& a)
+        : groupid(gid)
+        , appid(aid)
+        , addr(a)
+    {
+    }
+
+    PeerGroupType(): groupid(SRT_INVALID_SOCK), appid(0) {}
+
+    bool empty() { return groupid == SRT_INVALID_SOCK; }
+
+    bool operator==(const PeerGroupType& oth) const
+    {
+        return
+            groupid == oth.groupid &&
+            appid == oth.appid &&
+            addr.equal_address(oth.addr);
+    }
+    bool operator!=(const PeerGroupType& oth) const { return !(*this == oth); }
+};
+
+template <class T_STREAM>
+inline T_STREAM& operator<<(T_STREAM& sout, const PeerGroupType& in)
+{
+    return (
+            sout << in.groupid << '.' << in.appid << '/'
+                 << in.addr.str()
+           );
+}
+
 class CUDTGroup
 {
     friend class CUDTUnited;
@@ -412,7 +448,7 @@ class CUDTGroup
     srt::sync::Mutex  m_GroupLock;
 
     SRTSOCKET m_GroupID;
-    SRTSOCKET m_PeerGroupID;
+    PeerGroupType m_PeerGroupID;
     struct GroupContainer
     {
         std::list<SocketData>        m_List;
@@ -817,7 +853,7 @@ class CUDTGroup
 
     // Property accessors
     SRTU_PROPERTY_RW_CHAIN(CUDTGroup, SRTSOCKET, id, m_GroupID);
-    SRTU_PROPERTY_RW_CHAIN(CUDTGroup, SRTSOCKET, peerid, m_PeerGroupID);
+    SRTU_PROPERTY_RW_CHAIN(CUDTGroup, PeerGroupType, peerid, m_PeerGroupID);
     SRTU_PROPERTY_RW_CHAIN(CUDTGroup, bool, managed, m_selfManaged);
     SRTU_PROPERTY_RW_CHAIN(CUDTGroup, SRT_GROUP_TYPE, type, m_type);
     SRTU_PROPERTY_RW_CHAIN(CUDTGroup, int32_t, currentSchedSequence, m_iLastSchedSeqNo);

srtcore/sync.h

@@ -24,6 +24,14 @@
 #else
 #include <pthread.h>
 #endif
+
+// For getProcessID()
+#ifdef _WIN32
+#include <processthreadsapi.h>
+#else
+#include <unistd.h>
+#endif
+
 #include "utilities.h"
 
 class CUDTException;    // defined in common.h
@@ -810,6 +818,20 @@ void SetThreadLocalError(const CUDTException& e);
 /// @returns CUDTException pointer
 CUDTException& GetThreadLocalError();
 
+// This is something that can't be really portable, but
+// actually the version should be split into Windows and POSIX-compliant.
+inline uint32_t getProcessID()
+{
+#ifdef _WIN32
+    // This returns DWORD, which by definition
+    // is unsigned 32-bit integer.
+    return GetCurrentProcessId();
+#else
+    // Assume non-Windows platforms are POSIX-compliant.
+    return getpid();
+#endif
+}
+
 } // namespace sync
 } // namespace srt