From 4372d4e97d61ec86555d3a2113ac124f318890e5 Mon Sep 17 00:00:00 2001 From: Fletcher Dunn Date: Sun, 5 Sep 2021 17:40:34 -0700 Subject: [PATCH] Simplified code receiving messages on ad-hoc "messages" connections. P4:6753639 --- .../clientlib/csteamnetworkingmessages.cpp | 67 +++---------- .../clientlib/csteamnetworkingmessages.h | 7 +- .../steamnetworkingsockets_connections.cpp | 93 +++++++++++-------- .../steamnetworkingsockets_connections.h | 3 +- .../clientlib/steamnetworkingsockets_snp.cpp | 9 +- .../clientlib/steamnetworkingsockets_snp.h | 1 - 6 files changed, 80 insertions(+), 100 deletions(-) diff --git a/src/steamnetworkingsockets/clientlib/csteamnetworkingmessages.cpp b/src/steamnetworkingsockets/clientlib/csteamnetworkingmessages.cpp index f2ba2b7..12e5e31 100644 --- a/src/steamnetworkingsockets/clientlib/csteamnetworkingmessages.cpp +++ b/src/steamnetworkingsockets/clientlib/csteamnetworkingmessages.cpp @@ -77,8 +77,9 @@ CMessagesEndPoint::CMessagesEndPoint( CSteamNetworkingSockets &steamNetworkingSo CMessagesEndPoint::~CMessagesEndPoint() { - Assert( !m_pPollGroup ); + m_sharedConnectionLock.AssertHeldByCurrentThread(); Assert( !m_pListenSocket ); + m_sharedConnectionLock.unlock(); } bool CMessagesEndPoint::BInit() @@ -91,15 +92,6 @@ bool CMessagesEndPoint::BInit() return false; } - // Create poll group - PollGroupScopeLock pollGroupLock; - m_pPollGroup = m_steamNetworkingSockets.InternalCreatePollGroup( pollGroupLock ); - if ( !m_pPollGroup ) - { - AssertMsg( false, "Failed to create poll group" ); - return false; - } - m_steamNetworkingSockets.m_mapMessagesEndpointByVirtualPort.Insert( m_nLocalVirtualPort, this ); return true; } @@ -129,14 +121,7 @@ bool CMessagesEndPoint::BCreateListenSocket() void CMessagesEndPoint::FreeResources() { SteamNetworkingGlobalLock::AssertHeldByCurrentThread("CMessagesEndPoint::FreeResources"); - - // Destroy poll group, if any - if ( m_pPollGroup ) - { - m_pPollGroup->m_lock.lock(); // Only lock once, destructor unlocks! - delete m_pPollGroup; - m_pPollGroup = nullptr; - } + m_sharedConnectionLock.AssertHeldByCurrentThread(); // Destroy listen socket, if any if ( m_pListenSocket ) @@ -163,7 +148,7 @@ void CMessagesEndPoint::FreeResources() void CMessagesEndPoint::DestroyMessagesEndPoint() { - SteamNetworkingGlobalLock::AssertHeldByCurrentThread(); + m_sharedConnectionLock.lock(); FreeResources(); delete this; } @@ -321,8 +306,6 @@ void CMessagesEndPointSession::SetActiveConnection( CSteamNetworkConnectionBase m_bConnectionWasEverConnected = false; SetNextThinkTimeASAP(); MarkUsed( SteamNetworkingSockets_GetLocalTimestamp() ); - - pConn->SetPollGroup( m_messageEndPointOwner.m_pPollGroup ); } void CMessagesEndPointSession::ClearActiveConnection() @@ -396,6 +379,7 @@ bool CSteamNetworkingMessages::BInit() void CSteamNetworkingMessages::FreeResources() { SteamNetworkingGlobalLock::AssertHeldByCurrentThread( "CSteamNetworkingMessages::FreeResources" ); + m_sharedConnectionLock.AssertHeldByCurrentThread(); // Destroy all of our sessions. This will detach all of our connections FOR_EACH_HASHMAP( m_mapSessions, i ) @@ -538,30 +522,6 @@ int CSteamNetworkingMessages::ReceiveMessagesOnChannel( int nLocalChannel, Steam ShortDurationScopeLock lockMessageQueues( g_lockAllRecvMessageQueues ); - // Pull out all messages from the poll group into per-channel queues - if ( m_pPollGroup ) - { - for (;;) - { - CSteamNetworkingMessage *pMsg = m_pPollGroup->m_queueRecvMessages.m_pFirst; - if ( !pMsg ) - break; - pMsg->Unlink(); - - int idxSession = g_mapSessionsByConnection.Find( pMsg->m_conn ); - if ( idxSession == g_mapSessionsByConnection.InvalidIndex() ) - { - pMsg->Release(); - continue; - } - - SteamNetworkingMessagesSession *pSess = assert_cast( g_mapSessionsByConnection[ idxSession ] ); - Assert( pSess->m_pConnection ); - Assert( this == &pSess->MessagesOwner() ); - pSess->ReceivedMessage( pMsg ); - } - } - return pChan->m_queueRecvMessages.RemoveMessages( ppOutMessages, nMaxMessages ); } @@ -947,10 +907,9 @@ static void FreeMessageDataWithP2PMessageHeader( SteamNetworkingMessage_t *pMsg ::free( hdr ); } -void SteamNetworkingMessagesSession::ReceivedMessage( CSteamNetworkingMessage *pMsg ) +void SteamNetworkingMessagesSession::ReceivedMessage( CSteamNetworkingMessage *pMsg, CSteamNetworkConnectionBase *pConn ) { - // Caller locks this - g_lockAllRecvMessageQueues.AssertHeldByCurrentThread(); + m_pLock->AssertHeldByCurrentThread(); // Make sure the message is big enough to contain a header if ( pMsg->m_cbSize < sizeof(P2PMessageHeader) ) @@ -963,20 +922,24 @@ void SteamNetworkingMessagesSession::ReceivedMessage( CSteamNetworkingMessage *p // Process the header P2PMessageHeader *hdr = static_cast( pMsg->m_pData ); + pMsg->m_identityPeer = pConn->m_identityRemote; pMsg->m_nChannel = LittleDWord( hdr->m_nToChannel ); pMsg->m_cbSize -= sizeof(P2PMessageHeader); pMsg->m_pData = hdr+1; pMsg->m_conn = k_HSteamNetConnection_Invalid; // Invalidate this, we don't want app to think it's legit to access to the underlying connection pMsg->m_pfnFreeData = FreeMessageDataWithP2PMessageHeader; + // Mark channel as open + m_mapOpenChannels.Insert( pMsg->m_nChannel, true ); + CSteamNetworkingMessages::Channel *pChannel = MessagesOwner().FindOrCreateChannel( pMsg->m_nChannel ); + + // Grab the lock while we insert into the proper queues + ShortDurationScopeLock lockMessageQueues( g_lockAllRecvMessageQueues ); + // Add to the session pMsg->LinkToQueueTail( &CSteamNetworkingMessage::m_links, &m_queueRecvMessages ); - // Mark channel as open - m_mapOpenChannels.Insert( pMsg->m_nChannel, true ); - // Add to end of channel queue - CSteamNetworkingMessages::Channel *pChannel = MessagesOwner().FindOrCreateChannel( pMsg->m_nChannel ); pMsg->LinkToQueueTail( &CSteamNetworkingMessage::m_linksSecondaryQueue, &pChannel->m_queueRecvMessages ); } diff --git a/src/steamnetworkingsockets/clientlib/csteamnetworkingmessages.h b/src/steamnetworkingsockets/clientlib/csteamnetworkingmessages.h index 0306f86..86e1b7d 100644 --- a/src/steamnetworkingsockets/clientlib/csteamnetworkingmessages.h +++ b/src/steamnetworkingsockets/clientlib/csteamnetworkingmessages.h @@ -46,7 +46,6 @@ public: void DestroyMessagesEndPoint(); CSteamNetworkListenSocketP2P *m_pListenSocket = nullptr; // Might be NULL for "ephemeral" endpoints that cannot receive unsolicited traffic - CSteamNetworkPollGroup *m_pPollGroup = nullptr; // !SPEED! *All* of the sessions and connections share the same lock! // This could be improved, if we encounter a use case that needs it! @@ -104,6 +103,9 @@ public: virtual void SetActiveConnection( CSteamNetworkConnectionBase *pConn, ConnectionScopeLock &connectionLock ); virtual void ClearActiveConnection(); + /// Called when a message is received on one of our connections + virtual void ReceivedMessage( CSteamNetworkingMessage *pMsg, CSteamNetworkConnectionBase *pConn ) = 0; + /// Try to unlink from any old connections. The locking and /// object ownership is complicated here. This must be called /// from a safe place, when functions on the stack might have @@ -198,6 +200,7 @@ struct SteamNetworkingMessagesSession final : public CMessagesEndPointSession virtual void Think( SteamNetworkingMicroseconds usecNow ) override; virtual void SetActiveConnection( CSteamNetworkConnectionBase *pConn, ConnectionScopeLock &connectionLock ) override; virtual void ActiveConnectionStateChanged() override; + virtual void ReceivedMessage( CSteamNetworkingMessage *pMsg, CSteamNetworkConnectionBase *pConn ) override; /// Close the connection with the specified reason info void CloseConnection( int nReason, const char *pszDebug ); @@ -207,8 +210,6 @@ struct SteamNetworkingMessagesSession final : public CMessagesEndPointSession void UpdateConnectionInfo(); - void ReceivedMessage( CSteamNetworkingMessage *pMsg ); - #ifdef DBGFLAG_VALIDATE void Validate( CValidator &validator, const char *pchName ); #endif diff --git a/src/steamnetworkingsockets/clientlib/steamnetworkingsockets_connections.cpp b/src/steamnetworkingsockets/clientlib/steamnetworkingsockets_connections.cpp index 11c62a2..28530e3 100644 --- a/src/steamnetworkingsockets/clientlib/steamnetworkingsockets_connections.cpp +++ b/src/steamnetworkingsockets/clientlib/steamnetworkingsockets_connections.cpp @@ -137,6 +137,9 @@ CSteamNetworkingMessage *CSteamNetworkingMessage::New( uint32 cbSize ) pMsg->m_pfnRelease = ReleaseFunc; // Clear these fields + pMsg->m_nConnUserData = 0; + pMsg->m_usecTimeReceived = 0; + pMsg->m_nMessageNumber = 0; pMsg->m_nChannel = -1; pMsg->m_nFlags = 0; pMsg->m_links.Clear(); @@ -145,31 +148,6 @@ CSteamNetworkingMessage *CSteamNetworkingMessage::New( uint32 cbSize ) return pMsg; } -CSteamNetworkingMessage *CSteamNetworkingMessage::New( CSteamNetworkConnectionBase *pParent, uint32 cbSize, int64 nMsgNum, int nFlags, SteamNetworkingMicroseconds usecNow ) -{ - CSteamNetworkingMessage *pMsg = New( cbSize ); - if ( !pMsg ) - { - // Failed! if it's for a reliable message, then we must abort the connection. - // If unreliable message....well we've spewed, but let's try to keep on chugging. - if ( pParent && ( nFlags & k_nSteamNetworkingSend_Reliable ) ) - pParent->ConnectionState_ProblemDetectedLocally( k_ESteamNetConnectionEnd_Misc_InternalError, "Failed to allocate buffer to receive reliable message" ); - return nullptr; - } - - if ( pParent ) - { - pMsg->m_identityPeer = pParent->m_identityRemote; - pMsg->m_conn = pParent->m_hConnectionSelf; - pMsg->m_nConnUserData = pParent->GetUserData(); - } - pMsg->m_usecTimeReceived = usecNow; - pMsg->m_nMessageNumber = nMsgNum; - pMsg->m_nFlags = nFlags; - - return pMsg; -} - void CSteamNetworkingMessage::LinkBefore( CSteamNetworkingMessage *pSuccessor, Links CSteamNetworkingMessage::*pMbrLinks, SteamNetworkingMessageQueue *pQueue ) { // Make sure we're not already in a queue @@ -2627,23 +2605,28 @@ void CSteamNetworkConnectionBase::ConnectionStateChanged( ESteamNetworkingConnec m_pTransport->TransportConnectionStateChanged( eOldState ); } -bool CSteamNetworkConnectionBase::ReceivedMessage( const void *pData, int cbData, int64 nMsgNum, int nFlags, SteamNetworkingMicroseconds usecNow ) +CSteamNetworkingMessage *CSteamNetworkConnectionBase::AllocateNewRecvMessage( uint32 cbSize, int nFlags, SteamNetworkingMicroseconds usecNow ) +{ + CSteamNetworkingMessage *pMsg = CSteamNetworkingMessage::New( cbSize ); + if ( !pMsg ) + { + // Failed! if it's for a reliable message, then we must abort the connection. + // If unreliable message....well we've spewed, but let's try to keep on chugging. + if ( nFlags & k_nSteamNetworkingSend_Reliable ) + ConnectionState_ProblemDetectedLocally( k_ESteamNetConnectionEnd_Misc_InternalError, "Failed to allocate buffer to receive reliable message" ); + return nullptr; + } + + pMsg->m_usecTimeReceived = usecNow; + pMsg->m_nFlags = nFlags; + return pMsg; +} + +bool CSteamNetworkConnectionBase::ReceivedMessageData( const void *pData, int cbData, int64 nMsgNum, int nFlags, SteamNetworkingMicroseconds usecNow ) { -// // !TEST! Enable this during connection test to trap bogus messages earlier -// struct TestMsg -// { -// int64 m_nMsgNum; -// bool m_bReliable; -// int m_cbSize; -// uint8 m_data[ 20*1000 ]; -// }; -// const TestMsg *pTestMsg = (const TestMsg *)pData; -// -// // Size makes sense? -// Assert( sizeof(*pTestMsg) - sizeof(pTestMsg->m_data) + pTestMsg->m_cbSize == cbData ); // Create a message - CSteamNetworkingMessage *pMsg = CSteamNetworkingMessage::New( this, cbData, nMsgNum, nFlags, usecNow ); + CSteamNetworkingMessage *pMsg = AllocateNewRecvMessage( cbData, nFlags, usecNow ); if ( !pMsg ) { // Hm. this failure really is probably a sign that we are in a pretty bad state, @@ -2652,6 +2635,9 @@ bool CSteamNetworkConnectionBase::ReceivedMessage( const void *pData, int cbData return false; } + // Record message number + pMsg->m_nMessageNumber = nMsgNum; + // Copy the data memcpy( pMsg->m_pData, pData, cbData ); @@ -2670,6 +2656,20 @@ void CSteamNetworkConnectionBase::ReceivedMessage( CSteamNetworkingMessage *pMsg (long long)pMsg->m_nMessageNumber, pMsg->m_cbSize ); + // Check for redirecting it to the messages endpoint owner + #ifdef STEAMNETWORKINGSOCKETS_ENABLE_STEAMNETWORKINGMESSAGES + if ( m_pMessagesEndPointSessionOwner ) + { + m_pMessagesEndPointSessionOwner->ReceivedMessage( pMsg, this ); + return; + } + #endif + + // Fill in a few more details + pMsg->m_identityPeer = m_identityRemote; + pMsg->m_conn = m_hConnectionSelf; + pMsg->m_nConnUserData = GetUserData(); + // We use the same lock to protect *all* recv queues, for both connections and poll groups, // which keeps this really simple. g_lockAllRecvMessageQueues.lock(); @@ -2999,12 +2999,12 @@ void CSteamNetworkConnectionBase::ConnectionState_FinWait() // Check our state switch ( GetState() ) { - case k_ESteamNetworkingConnectionState_Dead: case k_ESteamNetworkingConnectionState_None: default: Assert( false ); return; + case k_ESteamNetworkingConnectionState_Dead: case k_ESteamNetworkingConnectionState_FinWait: break; @@ -3237,6 +3237,19 @@ void CSteamNetworkConnectionBase::Think( SteamNetworkingMicroseconds usecNow ) { SteamNetworkingMicroseconds usecTimeout = m_usecWhenEnteredConnectionState + k_usecFinWaitTimeout; + // If we're linked to a messages session, they need to unlink us! + #ifdef STEAMNETWORKINGSOCKETS_ENABLE_STEAMNETWORKINGMESSAGES + if ( m_pMessagesEndPointSessionOwner ) + { + m_pMessagesEndPointSessionOwner->SetNextThinkTimeASAP(); + + // And go ahead and schedule our own wakeup call + SetNextThinkTime( std::max( usecTimeout, usecNow + 10*1000 ) ); + return; + } + #endif + + // If we're linked to a messages session, they need to unlink us! #ifdef STEAMNETWORKINGSOCKETS_ENABLE_STEAMNETWORKINGMESSAGES if ( m_pMessagesEndPointSessionOwner ) diff --git a/src/steamnetworkingsockets/clientlib/steamnetworkingsockets_connections.h b/src/steamnetworkingsockets/clientlib/steamnetworkingsockets_connections.h index b133b95..7e2e123 100644 --- a/src/steamnetworkingsockets/clientlib/steamnetworkingsockets_connections.h +++ b/src/steamnetworkingsockets/clientlib/steamnetworkingsockets_connections.h @@ -712,8 +712,9 @@ protected: virtual void ConnectionGuessTimeoutReason( ESteamNetConnectionEnd &nReasonCode, ConnectionEndDebugMsg &msg, SteamNetworkingMicroseconds usecNow ); /// Called when we receive a complete message. Should allocate a message object and put it into the proper queues - bool ReceivedMessage( const void *pData, int cbData, int64 nMsgNum, int nFlags, SteamNetworkingMicroseconds usecNow ); + bool ReceivedMessageData( const void *pData, int cbData, int64 nMsgNum, int nFlags, SteamNetworkingMicroseconds usecNow ); void ReceivedMessage( CSteamNetworkingMessage *pMsg ); + CSteamNetworkingMessage *AllocateNewRecvMessage( uint32 cbSize, int nFlags, SteamNetworkingMicroseconds usecNow ); /// Timestamp when we last sent an end-to-end connection request packet SteamNetworkingMicroseconds m_usecWhenSentConnectRequest; diff --git a/src/steamnetworkingsockets/clientlib/steamnetworkingsockets_snp.cpp b/src/steamnetworkingsockets/clientlib/steamnetworkingsockets_snp.cpp index 973735b..b40c4e8 100644 --- a/src/steamnetworkingsockets/clientlib/steamnetworkingsockets_snp.cpp +++ b/src/steamnetworkingsockets/clientlib/steamnetworkingsockets_snp.cpp @@ -2426,7 +2426,7 @@ void CSteamNetworkConnectionBase::SNP_ReceiveUnreliableSegment( int64 nMsgNum, i // Deliver it immediately, don't go through the fragmentation assembly process below. // (Although that would work.) - ReceivedMessage( pSegmentData, cbSegmentSize, nMsgNum, k_nSteamNetworkingSend_Unreliable, usecNow ); + ReceivedMessageData( pSegmentData, cbSegmentSize, nMsgNum, k_nSteamNetworkingSend_Unreliable, usecNow ); return; } @@ -2509,10 +2509,13 @@ void CSteamNetworkConnectionBase::SNP_ReceiveUnreliableSegment( int64 nMsgNum, i return; } - CSteamNetworkingMessage *pMsg = CSteamNetworkingMessage::New( this, cbMessageSize, nMsgNum, k_nSteamNetworkingSend_Unreliable, usecNow ); + CSteamNetworkingMessage *pMsg = AllocateNewRecvMessage( cbMessageSize, k_nSteamNetworkingSend_Unreliable, usecNow ); if ( !pMsg ) return; + // Record the message number + pMsg->m_nMessageNumber = nMsgNum; + // OK, we have the complete message! Gather the // segments into a contiguous buffer for (;;) @@ -2903,7 +2906,7 @@ bool CSteamNetworkConnectionBase::SNP_ReceiveReliableSegment( int64 nPktNum, int } // We have a full message! Queue it - if ( !ReceivedMessage( pReliableDecode, cbMsgSize, nMsgNum, k_nSteamNetworkingSend_Reliable, usecNow ) ) + if ( !ReceivedMessageData( pReliableDecode, cbMsgSize, nMsgNum, k_nSteamNetworkingSend_Reliable, usecNow ) ) return false; // Weird failure. Most graceful response is to not ack this packet, and maybe we will work next on retry. pReliableDecode += cbMsgSize; int cbStreamConsumed = pReliableDecode-pReliableStart; diff --git a/src/steamnetworkingsockets/clientlib/steamnetworkingsockets_snp.h b/src/steamnetworkingsockets/clientlib/steamnetworkingsockets_snp.h index b44f5be..0e5b290 100644 --- a/src/steamnetworkingsockets/clientlib/steamnetworkingsockets_snp.h +++ b/src/steamnetworkingsockets/clientlib/steamnetworkingsockets_snp.h @@ -78,7 +78,6 @@ class CSteamNetworkingMessage : public SteamNetworkingMessage_t { public: STEAMNETWORKINGSOCKETS_DECLARE_CLASS_OPERATOR_NEW - static CSteamNetworkingMessage *New( CSteamNetworkConnectionBase *pParent, uint32 cbSize, int64 nMsgNum, int nFlags, SteamNetworkingMicroseconds usecNow ); static CSteamNetworkingMessage *New( uint32 cbSize ); static void DefaultFreeData( SteamNetworkingMessage_t *pMsg );