From 379475ca7c81dbe0c6ade68ac97518e78b10cbfc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miko=C5=82aj=20Ma=C5=82ecki?= Date: Tue, 9 Mar 2021 12:29:37 +0100 Subject: [PATCH 1/2] [core] Reformed peer group ID to avoid conflicts --- srtcore/api.cpp | 35 +++++++++++++++++++++++------------ srtcore/api.h | 4 ++-- srtcore/core.cpp | 22 ++++++++++++++-------- srtcore/core.h | 11 +++++++---- srtcore/group.cpp | 3 +-- srtcore/group.h | 40 ++++++++++++++++++++++++++++++++++++++-- srtcore/sync.h | 22 ++++++++++++++++++++++ testing/testmedia.cpp | 24 ++++++++++++++++++++++-- testing/testmedia.hpp | 1 + 9 files changed, 130 insertions(+), 32 deletions(-) diff --git a/srtcore/api.cpp b/srtcore/api.cpp index 5239693fc..df65265f7 100644 --- a/srtcore/api.cpp +++ b/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 diff --git a/srtcore/api.h b/srtcore/api.h index 478b0955c..f52243de9 100644 --- a/srtcore/api.h +++ b/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) diff --git a/srtcore/core.cpp b/srtcore/core.cpp index df40ef0d5..345c1956a 100644 --- a/srtcore/core.cpp +++ b/srtcore/core.cpp @@ -1244,7 +1244,7 @@ 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 }; + const uint32_t storedata [GRPD_E_SIZE] = { uint32_t(id), dataword, srt::sync::getProcessID() }; memcpy((space), storedata, sizeof storedata); const size_t ra_size = Size(storedata); @@ -2835,7 +2835,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 +2846,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 +2900,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,12 +2932,12 @@ 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 @@ -2939,7 +2945,7 @@ bool CUDT::interpretGroup(const int32_t groupdata[], size_t data_size SRT_ATR_UN // (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 +2964,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 +3004,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! diff --git a/srtcore/core.h b/srtcore/core.h index ef395fd2f..a63299e6d 100644 --- a/srtcore/core.h +++ b/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 diff --git a/srtcore/group.cpp b/srtcore/group.cpp index 4dbdc23fd..afe29861e 100644 --- a/srtcore/group.cpp +++ b/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 epollid; { diff --git a/srtcore/group.h b/srtcore/group.h index afb05f0f9..88874f4aa 100644 --- a/srtcore/group.h +++ b/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 +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 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); diff --git a/srtcore/sync.h b/srtcore/sync.h index aeb4e0663..e3951873c 100644 --- a/srtcore/sync.h +++ b/srtcore/sync.h @@ -24,6 +24,14 @@ #else #include #endif + +// For getProcessID() +#ifdef _WIN32 +#include +#else +#include +#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 diff --git a/testing/testmedia.cpp b/testing/testmedia.cpp index 625b91601..49d66dada 100755 --- a/testing/testmedia.cpp +++ b/testing/testmedia.cpp @@ -29,6 +29,7 @@ #include "netinet_any.h" #include "common.h" #include "api.h" +#include "core.h" #include "udt.h" #include "logging.h" #include "utilities.h" @@ -232,6 +233,10 @@ void SrtCommon::InitParameters(string host, string path, map par) Error("With //group, the group 'type' must be specified."); } + auto* is = map_getp(par, "forceid"); + if (is) + m_forced_id = stoi(*is); + vector parts; Split(m_group_type, '/', back_inserter(parts)); if (parts.size() == 0 || parts.size() > 2) @@ -355,6 +360,7 @@ void SrtCommon::InitParameters(string host, string path, map par) par.erase("type"); par.erase("nodes"); + par.erase("forceid"); // For a group-connect specification, it's // always the caller mode. @@ -465,6 +471,12 @@ void SrtCommon::InitParameters(string host, string path, map par) } } + if (par.count("forceid")) + { + m_forced_id = stoi(par["forceid"]); + par.erase("forceid"); + } + // Assign the others here. m_options = par; m_options["mode"] = m_mode; @@ -890,7 +902,11 @@ void SrtCommon::OpenClient(string host, int port) void SrtCommon::PrepareClient() { - m_sock = srt_create_socket(); + if (m_forced_id != SRT_INVALID_SOCK) + m_sock = CUDT::socket(m_forced_id); + else + m_sock = srt_create_socket(); + if (m_sock == SRT_ERROR) Error("srt_create_socket"); @@ -966,7 +982,11 @@ void SrtCommon::OpenGroupClient() Error("With //group, type='" + m_group_type + "' undefined"); } - m_sock = srt_create_group(type); + if (m_forced_id != SRT_INVALID_SOCK) + m_sock = CUDT::createGroup(type, m_forced_id); + else + m_sock = srt_create_group(type); + if (m_sock == -1) Error("srt_create_group"); diff --git a/testing/testmedia.hpp b/testing/testmedia.hpp index b251f140b..21d5f9451 100644 --- a/testing/testmedia.hpp +++ b/testing/testmedia.hpp @@ -95,6 +95,7 @@ class SrtCommon int srt_epoll = -1; SRT_EPOLL_T m_direction = SRT_EPOLL_OPT_NONE; //< Defines which of SND or RCV option variant should be used, also to set SRT_SENDER for output bool m_blocking_mode = true; //< enforces using SRTO_SNDSYN or SRTO_RCVSYN, depending on @a m_direction + SRTSOCKET m_forced_id = SRT_INVALID_SOCK; int m_timeout = 0; //< enforces using SRTO_SNDTIMEO or SRTO_RCVTIMEO, depending on @a m_direction bool m_tsbpdmode = true; int m_outgoing_port = 0; From 77e08c28bd563e86a6306afcaff2f904f28df13d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miko=C5=82aj=20Ma=C5=82ecki?= Date: Tue, 9 Mar 2021 17:12:07 +0100 Subject: [PATCH 2/2] Included possibility to use backward compat --- srtcore/core.cpp | 21 ++++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) diff --git a/srtcore/core.cpp b/srtcore/core.cpp index 345c1956a..556d21c00 100644 --- a/srtcore/core.cpp +++ b/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, srt::sync::getProcessID() }; - 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));