GitHub/GameNetworkingSockets
1
0
Fork 0
mirror of https://github.com/ValveSoftware/GameNetworkingSockets.git synced 2026-05-29 16:20:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Simplified code receiving messages on ad-hoc "messages" connections.

Browse Source 
P4:6753639
This commit is contained in:
Fletcher Dunn
2021-09-05 17:40:34 -07:00
parent 6e771147a2
commit 4372d4e97d
6 changed files with 80 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/steamnetworkingsockets/clientlib/csteamnetworkingmessages.cpp
+15 -52
View File
@@ -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<SteamNetworkingMessagesSession*>( 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<P2PMessageHeader *>( 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 );
}
src/steamnetworkingsockets/clientlib/csteamnetworkingmessages.h
+4 -3
View File
@@ -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
src/steamnetworkingsockets/clientlib/steamnetworkingsockets_connections.cpp
+53 -40
View File
@@ -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 )
src/steamnetworkingsockets/clientlib/steamnetworkingsockets_connections.h
+2 -1
View File
@@ -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;
src/steamnetworkingsockets/clientlib/steamnetworkingsockets_snp.cpp
+6 -3
View File
@@ -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;
src/steamnetworkingsockets/clientlib/steamnetworkingsockets_snp.h
-1
View File
@@ -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 );