1607 lines
44 KiB
C++
1607 lines
44 KiB
C++
//========= Copyright Valve Corporation, All rights reserved. ============//
|
|
//
|
|
// Purpose:
|
|
//
|
|
//=============================================================================//
|
|
|
|
#include <winsock2.h>
|
|
#include "vmpi_filesystem_internal.h"
|
|
#include "zlib.h"
|
|
#include "vstdlib/random.h"
|
|
|
|
|
|
#define MINIMUM_SLEEP_MS 1
|
|
|
|
// NOTE: This number comes from measurements on our network to find out how fast
|
|
// we can broadcast without the network freaking out.
|
|
//
|
|
// This number can be changed on the command line with the -mpi_FileTransmitRate parameter.
|
|
int MULTICAST_TRANSMIT_RATE = (1024*1000); // N megs per second
|
|
|
|
// Defines when we'll stop transmitting a file to a client.
|
|
// (After we've transmitted the file to the client N times and we haven't heard an ack back for M seconds).
|
|
#define MIN_FILE_CYCLE_COUNT 5
|
|
#define CLIENT_FILE_ACK_TIMEOUT 20
|
|
|
|
|
|
|
|
// ------------------------------------------------------------------------------------------------------------------------ //
|
|
// Global helpers.
|
|
// ------------------------------------------------------------------------------------------------------------------------ //
|
|
|
|
static void SendMulticastIP( const CIPAddr *pAddr )
|
|
{
|
|
unsigned char packetID[2] = { VMPI_PACKETID_FILESYSTEM, VMPI_FSPACKETID_MULTICAST_ADDR };
|
|
VMPI_Send2Chunks(
|
|
packetID, sizeof( packetID ),
|
|
pAddr, sizeof( *pAddr ),
|
|
VMPI_PERSISTENT );
|
|
}
|
|
|
|
|
|
static bool IsOpeningForWriteAccess( const char *pOptions )
|
|
{
|
|
return strchr( pOptions, 'w' ) || strchr( pOptions, 'a' ) || strchr( pOptions, '+' );
|
|
}
|
|
|
|
|
|
// This does a fast zlib compression of the source data into the 'out' buffer.
|
|
static bool ZLibCompress( const void *pData, int len, CUtlVector<char> &out )
|
|
{
|
|
if ( len == 0 )
|
|
{
|
|
out.Purge();
|
|
return true;
|
|
}
|
|
|
|
int outStartLen = len;
|
|
RETRY:;
|
|
|
|
// Prepare the compression stream.
|
|
z_stream zs;
|
|
memset( &zs, 0, sizeof( zs ) );
|
|
|
|
if ( deflateInit( &zs, 1 ) != Z_OK )
|
|
return false;
|
|
|
|
|
|
// Now compress it into the output buffer.
|
|
out.SetSize( outStartLen );
|
|
|
|
zs.next_in = (unsigned char*)pData;
|
|
zs.avail_in = len;
|
|
|
|
zs.next_out = (unsigned char*)out.Base();
|
|
zs.avail_out = out.Count();
|
|
|
|
int ret = deflate( &zs, Z_FINISH );
|
|
deflateEnd( &zs );
|
|
|
|
if ( ret == Z_STREAM_END )
|
|
{
|
|
// Get rid of whatever was left over.
|
|
out.RemoveMultiple( zs.total_out, out.Count() - zs.total_out );
|
|
return true;
|
|
}
|
|
else if ( ret == Z_OK )
|
|
{
|
|
// Need more space in the output buffer.
|
|
outStartLen += 1024 * 128;
|
|
goto RETRY;
|
|
}
|
|
else
|
|
{
|
|
return false;
|
|
}
|
|
}
|
|
|
|
|
|
// ------------------------------------------------------------------------------------------------------------------------ //
|
|
// CVMPIFile_PassThru
|
|
// ------------------------------------------------------------------------------------------------------------------------ //
|
|
|
|
class CVMPIFile_PassThru : public IVMPIFile
|
|
{
|
|
public:
|
|
void Init( IBaseFileSystem *pPassThru, FileHandle_t fp )
|
|
{
|
|
m_pPassThru = pPassThru;
|
|
m_fp = fp;
|
|
}
|
|
|
|
virtual void Close()
|
|
{
|
|
m_pPassThru->Close( m_fp );
|
|
delete this;
|
|
}
|
|
|
|
virtual void Seek( int pos, FileSystemSeek_t seekType )
|
|
{
|
|
m_pPassThru->Seek( m_fp, pos, seekType );
|
|
}
|
|
|
|
virtual unsigned int Tell()
|
|
{
|
|
return m_pPassThru->Tell( m_fp );
|
|
}
|
|
|
|
virtual unsigned int Size()
|
|
{
|
|
return m_pPassThru->Size( m_fp );
|
|
}
|
|
|
|
virtual void Flush()
|
|
{
|
|
m_pPassThru->Flush( m_fp );
|
|
}
|
|
|
|
virtual int Read( void* pOutput, int size )
|
|
{
|
|
return m_pPassThru->Read( pOutput, size, m_fp );
|
|
}
|
|
|
|
virtual int Write( void const* pInput, int size )
|
|
{
|
|
return m_pPassThru->Write( pInput, size, m_fp );
|
|
}
|
|
|
|
|
|
private:
|
|
IBaseFileSystem *m_pPassThru;
|
|
FileHandle_t m_fp;
|
|
};
|
|
|
|
|
|
|
|
// ---------------------------------------------------------------------------------------------------- //
|
|
// CTransmitRateMgr coordinates with any other currently-running VMPI jobs, and they all will cut
|
|
// down their transmission rate to stay within MULTICAST_TRANSMIT_RATE.
|
|
// ---------------------------------------------------------------------------------------------------- //
|
|
|
|
#define TRANSMITRATEMGR_BROADCAST_INVERVAL (1.0 / 3.0) // How many times per second we broadcast our presence.
|
|
#define TRANSMITRATEMGR_EXPIRE_TIME 0.7 // How long it'll go before deciding a guy is not transmitting anymore.
|
|
|
|
static char s_cTransmitRateMgrPacket[] = {2,6,-3,2,1,-66};
|
|
|
|
class CTransmitRateMgr
|
|
{
|
|
public:
|
|
CTransmitRateMgr();
|
|
|
|
void ReadPackets();
|
|
void BroadcastPresence();
|
|
|
|
double GetMicrosecondsPerByte() const;
|
|
|
|
private:
|
|
class CMachineRecord
|
|
{
|
|
public:
|
|
unsigned long m_UniqueID;
|
|
float m_flLastTime;
|
|
};
|
|
CUtlVector<CMachineRecord> m_MachineRecords;
|
|
|
|
unsigned long m_UniqueID;
|
|
float m_flLastBroadcastTime;
|
|
double m_nMicrosecondsPerByte;
|
|
ISocket *m_pSocket;
|
|
};
|
|
|
|
CTransmitRateMgr::CTransmitRateMgr()
|
|
{
|
|
m_nMicrosecondsPerByte = 1000000.0 / (double)MULTICAST_TRANSMIT_RATE;
|
|
m_flLastBroadcastTime = 0;
|
|
|
|
// Build a (hopefully) unique ID.
|
|
m_UniqueID = (unsigned long)this;
|
|
CCycleCount cnt;
|
|
cnt.Sample();
|
|
m_UniqueID += cnt.GetMicroseconds();
|
|
Sleep( 1 );
|
|
m_UniqueID += cnt.GetMicroseconds();
|
|
|
|
m_pSocket = CreateIPSocket();
|
|
if ( m_pSocket )
|
|
{
|
|
m_pSocket->BindToAny( VMPI_MASTER_FILESYSTEM_BROADCAST_PORT );
|
|
}
|
|
}
|
|
|
|
void CTransmitRateMgr::ReadPackets()
|
|
{
|
|
if ( !m_pSocket )
|
|
return;
|
|
|
|
float flCurTime = Plat_FloatTime();
|
|
|
|
// First, update/add records.
|
|
while ( 1 )
|
|
{
|
|
char data[512];
|
|
CIPAddr ipFrom;
|
|
int len = m_pSocket->RecvFrom( data, sizeof( data ), &ipFrom );
|
|
if ( len == -1 )
|
|
break;
|
|
|
|
if ( len == sizeof( s_cTransmitRateMgrPacket ) + sizeof( unsigned long ) &&
|
|
memcmp( data, s_cTransmitRateMgrPacket, sizeof( s_cTransmitRateMgrPacket ) ) == 0 )
|
|
{
|
|
unsigned long id = *((unsigned long*)&data[sizeof(s_cTransmitRateMgrPacket)]);
|
|
if ( id == m_UniqueID )
|
|
continue;
|
|
|
|
int i;
|
|
for ( i=0; i < m_MachineRecords.Count(); i++ )
|
|
{
|
|
if ( m_MachineRecords[i].m_UniqueID == id )
|
|
{
|
|
m_MachineRecords[i].m_flLastTime = flCurTime;
|
|
break;
|
|
}
|
|
}
|
|
|
|
if ( i == m_MachineRecords.Count() )
|
|
{
|
|
int index = m_MachineRecords.AddToTail();
|
|
m_MachineRecords[index].m_UniqueID = id;
|
|
m_MachineRecords[index].m_flLastTime = flCurTime;
|
|
}
|
|
}
|
|
}
|
|
|
|
// Now, expire any old records.
|
|
for ( int i=0; i < m_MachineRecords.Count(); i++ )
|
|
{
|
|
if ( (flCurTime - m_MachineRecords[i].m_flLastTime) > TRANSMITRATEMGR_EXPIRE_TIME )
|
|
{
|
|
m_MachineRecords.Remove( i );
|
|
--i;
|
|
}
|
|
}
|
|
|
|
// Recalculate our transmit rate (assuming we're receiving our own broadcast packets).
|
|
m_nMicrosecondsPerByte = 1000000.0 / (double)(MULTICAST_TRANSMIT_RATE / (m_MachineRecords.Count() + 1));
|
|
}
|
|
|
|
void CTransmitRateMgr::BroadcastPresence()
|
|
{
|
|
if ( !m_pSocket )
|
|
return;
|
|
|
|
float flCurTime = Plat_FloatTime();
|
|
if ( (flCurTime - m_flLastBroadcastTime) < TRANSMITRATEMGR_BROADCAST_INVERVAL )
|
|
return;
|
|
|
|
m_flLastBroadcastTime = flCurTime;
|
|
|
|
char cData[sizeof( s_cTransmitRateMgrPacket ) + sizeof( unsigned long )];
|
|
memcpy( cData, s_cTransmitRateMgrPacket, sizeof( s_cTransmitRateMgrPacket ) );
|
|
*((unsigned long*)&cData[ sizeof( s_cTransmitRateMgrPacket ) ] ) = m_UniqueID;
|
|
|
|
m_pSocket->Broadcast( cData, sizeof( cData ), VMPI_MASTER_FILESYSTEM_BROADCAST_PORT );
|
|
}
|
|
|
|
inline double CTransmitRateMgr::GetMicrosecondsPerByte() const
|
|
{
|
|
return m_nMicrosecondsPerByte;
|
|
}
|
|
|
|
|
|
// ---------------------------------------------------------------------------------------------------- //
|
|
// CRateLimiter manages waiting for small periods of time between packets so the rate is
|
|
// whatever we want it to be.
|
|
//
|
|
// It also will give up some CPU time to other processes every 50 milliseconds.
|
|
// ---------------------------------------------------------------------------------------------------- //
|
|
|
|
class CRateLimiter
|
|
{
|
|
public:
|
|
|
|
CRateLimiter();
|
|
|
|
void GiveUpTimeSlice();
|
|
void NoteExcessTimeTaken( unsigned long excessTimeInMicroseconds );
|
|
|
|
|
|
public:
|
|
|
|
DWORD m_SleepIntervalMS; // Give up a timeslice every N milliseconds.
|
|
|
|
// Since we sleep once in a while, we time how long the sleep took and we beef
|
|
// up the transmit rate until we've accounted for the time lost during the sleep.
|
|
DWORD m_AccumulatedSleepMicroseconds;
|
|
|
|
// When was the last time we gave up a little bit of CPU to other programs.
|
|
CCycleCount m_LastSleepTime;
|
|
};
|
|
|
|
CRateLimiter::CRateLimiter()
|
|
{
|
|
m_SleepIntervalMS = 50;
|
|
m_AccumulatedSleepMicroseconds = 0;
|
|
m_LastSleepTime.Sample();
|
|
}
|
|
|
|
void CRateLimiter::GiveUpTimeSlice()
|
|
{
|
|
// Sleep again?
|
|
CCycleCount currentTime, dtSinceLastSleep;
|
|
currentTime.Sample();
|
|
CCycleCount::Sub( currentTime, m_LastSleepTime, dtSinceLastSleep );
|
|
|
|
if ( dtSinceLastSleep.GetMilliseconds() >= m_SleepIntervalMS )
|
|
{
|
|
CFastTimer sleepTimer;
|
|
|
|
sleepTimer.Start();
|
|
Sleep( 10 );
|
|
sleepTimer.End();
|
|
|
|
m_AccumulatedSleepMicroseconds += sleepTimer.GetDuration().GetMicroseconds();
|
|
m_LastSleepTime.Sample();
|
|
}
|
|
}
|
|
|
|
|
|
void CRateLimiter::NoteExcessTimeTaken( unsigned long excessTimeInMicroseconds )
|
|
{
|
|
// Note: we give up time slices above.
|
|
if ( excessTimeInMicroseconds > m_AccumulatedSleepMicroseconds )
|
|
{
|
|
excessTimeInMicroseconds -= m_AccumulatedSleepMicroseconds;
|
|
m_AccumulatedSleepMicroseconds = 0;
|
|
|
|
CCycleCount startCount;
|
|
startCount.Sample();
|
|
while ( 1 )
|
|
{
|
|
CCycleCount curCount, diff;
|
|
curCount.Sample();
|
|
|
|
CCycleCount::Sub( curCount, startCount, diff );
|
|
if ( diff.GetMicroseconds() >= excessTimeInMicroseconds )
|
|
break;
|
|
}
|
|
}
|
|
else
|
|
{
|
|
m_AccumulatedSleepMicroseconds -= excessTimeInMicroseconds;
|
|
excessTimeInMicroseconds = 0;
|
|
}
|
|
}
|
|
|
|
|
|
// ------------------------------------------------------------------------------------------------------------------------ //
|
|
// CMasterMulticastThread.
|
|
// ------------------------------------------------------------------------------------------------------------------------ //
|
|
|
|
class CMasterMulticastThread
|
|
{
|
|
public:
|
|
|
|
CMasterMulticastThread();
|
|
~CMasterMulticastThread();
|
|
|
|
// This creates the socket and starts the thread (initially in an idle state since it doesn't
|
|
// know of any files anyone wants).
|
|
bool Init( IBaseFileSystem *pPassThru, unsigned short localPort, const CIPAddr *pAddr, int maxFileSystemMemoryUsage );
|
|
void Term();
|
|
|
|
// Returns -1 if there is an error.
|
|
int FindOrAddFile( const char *pFilename, const char *pPathID );
|
|
const CUtlVector<char>& GetFileData( int iFile ) const;
|
|
|
|
// When a client requests a files, this is called to tell the thread to start
|
|
// adding chunks from the specified file into the queue it's multicasting.
|
|
//
|
|
// Returns -1 if the file isn't there. Otherwise, it returns the file ID
|
|
// that will be sent along with the file's chunks in the multicast packets.
|
|
int AddFileRequest( const char *pFilename, const char *pPathID, int clientID, bool *bZeroLength );
|
|
|
|
// As each client receives multicasted chunks, they ack them so the master can
|
|
// stop transmitting any chunks it knows nobody wants.
|
|
void OnChunkReceived( int fileID, int clientID, int iChunk );
|
|
void OnFileReceived( int fileID, int clientID );
|
|
|
|
// Call this if a client disconnects so it can stop sending anything this client wants.
|
|
void OnClientDisconnect( int clientID, bool bGrabCriticalSection=true );
|
|
|
|
void CreateVirtualFile( const char *pFilename, const void *pData, unsigned long fileLength );
|
|
|
|
private:
|
|
|
|
class CChunkInfo
|
|
{
|
|
public:
|
|
unsigned short m_iChunk;
|
|
unsigned short m_RefCount; // How many clients want this chunk.
|
|
unsigned short m_iActiveChunksIndex; // Index into m_ActiveChunks.
|
|
};
|
|
|
|
|
|
// This stores a client's reference to a file so it knows which pieces of the file the client needs.
|
|
class CClientFileInfo
|
|
{
|
|
public:
|
|
bool NeedsChunk( int i ) const { return (m_ChunksToSend[i>>3] & (1 << (i&7))) != 0; }
|
|
|
|
public:
|
|
int m_ClientID;
|
|
CUtlVector<unsigned char> m_ChunksToSend; // One bit for each chunk that this client still wants.
|
|
int m_nChunksLeft;
|
|
|
|
// TCP transmission only.
|
|
int m_TCP_LastChunkAcked;
|
|
int m_TCP_LastChunkSent;
|
|
|
|
float m_flTransmitStartTime;
|
|
|
|
float m_flLastAckTime; // Last time we heard an ack back from this client about this file.
|
|
// If this goes for too long, then we assume that the client is
|
|
// in a screwed state, and we stop sending the file to him.
|
|
int m_nTimesFileCycled; // How many times has the master multicast thread cycled over this file?
|
|
// We won't kick the client until we've cycled over the file a few times
|
|
// after the client asked for it.
|
|
};
|
|
|
|
|
|
class CMulticastFile
|
|
{
|
|
public:
|
|
~CMulticastFile()
|
|
{
|
|
m_Clients.PurgeAndDeleteElements();
|
|
}
|
|
|
|
const char* GetFilename() { return m_Filename.Base(); }
|
|
const char* GetPathID() { return m_PathID.Base(); }
|
|
|
|
|
|
public:
|
|
int m_nCycles; // How many times has the multicast thread visited this file?
|
|
|
|
// This is sent along with every packet. If a client gets a chunk and doesn't have that file's
|
|
// info, the client will receive that file too.
|
|
CUtlVector<char> m_Filename;
|
|
CUtlVector<char> m_PathID;
|
|
|
|
CMulticastFileInfo m_Info;
|
|
|
|
// This is stored so the app can read out the uncompressed data.
|
|
CUtlVector<char> m_UncompressedData;
|
|
|
|
// zlib-compressed file data
|
|
CUtlVector<char> m_Data;
|
|
|
|
// This gets set to false if we run over our memory limit and start caching file data out.
|
|
// Then it'll reload the data if a client requests the file.
|
|
bool m_bDataLoaded;
|
|
|
|
// m_Chunks holds the chunks by index.
|
|
// m_ActiveChunks holds them sorted by whether they're active or not.
|
|
//
|
|
// Each chunk has a refcount. While the refcount is > 0, the chunk is in the first
|
|
// half of m_ActiveChunks. When the refcount gets to 0, the chunk is moved to the end of
|
|
// m_ActiveChunks. That way, we can iterate through the chunks that need to be sent and
|
|
// stop iterating the first time we hit one with a refcount of 0.
|
|
CUtlVector<CChunkInfo> m_Chunks;
|
|
CUtlLinkedList<CChunkInfo*,int> m_ActiveChunks;
|
|
|
|
// This tells which clients want pieces of this file.
|
|
CUtlLinkedList<CClientFileInfo*,int> m_Clients;
|
|
};
|
|
|
|
|
|
private:
|
|
|
|
static DWORD WINAPI StaticMulticastThread( LPVOID pParameter );
|
|
DWORD MulticastThread();
|
|
|
|
bool CheckClientTimeouts();
|
|
bool Thread_SendFileChunk_Multicast( int *pnBytesSent );
|
|
void Thread_SeekToNextActiveChunk();
|
|
|
|
// In TCP mode, we send new chunks as they are acked.
|
|
void TCP_SendNextChunk( CMulticastFile *pFile, CClientFileInfo *pClient );
|
|
|
|
void EnsureMemoryLimit( CMulticastFile *pIgnore );
|
|
|
|
// Called after pFile->m_UncompressedData has been setup. This compresses the data, prepares the header,
|
|
// copies the filename, and adds it into the queue for the multicast thread.
|
|
int FinishFileSetup( CMulticastFile *pFile, const char *pFilename, const char *pPathID, bool bFileAlreadyExisted );
|
|
|
|
void IncrementChunkRefCount( CMasterMulticastThread::CMulticastFile *pFile, int iChunk );
|
|
void DecrementChunkRefCount( int iFile, int iChunk );
|
|
|
|
int FindFile( const char *pFilename, const char *pPathID );
|
|
|
|
bool FindWarningSuppression( const char *pFilename );
|
|
void AddWarningSuppression( const char *pFilename );
|
|
|
|
private:
|
|
|
|
CUtlLinkedList<CMulticastFile*,int> m_Files;
|
|
|
|
unsigned long m_nCurMemoryUsage; // Total of all the file data we have loaded.
|
|
unsigned long m_nMaxMemoryUsage; // 0 means that there is no limit.
|
|
|
|
// This tracks how many chunks we have that want to be sent.
|
|
int m_nTotalActiveChunks;
|
|
|
|
SOCKET m_Socket;
|
|
sockaddr_in m_MulticastAddr;
|
|
|
|
HANDLE m_hMainThread;
|
|
IBaseFileSystem *m_pPassThru;
|
|
|
|
HANDLE m_hThread;
|
|
CRITICAL_SECTION m_CS;
|
|
|
|
// Events used to communicate with our thread.
|
|
HANDLE m_hTermEvent;
|
|
|
|
// The thread walks through this as it spews chunks of data.
|
|
volatile int m_iCurFile; // Index into m_Files.
|
|
volatile int m_iCurActiveChunk; // Current index into CMulticastFile::m_ActiveChunks.
|
|
|
|
CUtlLinkedList<char*,int> m_WarningSuppressions;
|
|
};
|
|
|
|
|
|
CMasterMulticastThread::CMasterMulticastThread()
|
|
{
|
|
m_hThread = m_hMainThread = NULL;
|
|
m_Socket = INVALID_SOCKET;
|
|
m_nTotalActiveChunks = 0;
|
|
m_iCurFile = m_iCurActiveChunk = -1;
|
|
m_pPassThru = NULL;
|
|
|
|
m_hTermEvent = CreateEvent( NULL, FALSE, FALSE, NULL );
|
|
InitializeCriticalSection( &m_CS );
|
|
m_nCurMemoryUsage = m_nMaxMemoryUsage = 0;
|
|
}
|
|
|
|
|
|
CMasterMulticastThread::~CMasterMulticastThread()
|
|
{
|
|
Term();
|
|
|
|
CloseHandle( m_hTermEvent );
|
|
|
|
DeleteCriticalSection( &m_CS );
|
|
}
|
|
|
|
|
|
bool CMasterMulticastThread::Init( IBaseFileSystem *pPassThru, unsigned short localPort, const CIPAddr *pAddr, int maxMemoryUsage )
|
|
{
|
|
Term();
|
|
|
|
m_nMaxMemoryUsage = maxMemoryUsage;
|
|
Assert( m_nCurMemoryUsage == 0 );
|
|
m_nCurMemoryUsage = 0;
|
|
|
|
if ( VMPI_GetFileSystemMode() == VMPI_FILESYSTEM_TCP )
|
|
{
|
|
// No need for an extra socket in this mode.
|
|
m_Socket = INVALID_SOCKET;
|
|
}
|
|
else
|
|
{
|
|
// First, create our socket.
|
|
m_Socket = socket( AF_INET, SOCK_DGRAM, IPPROTO_IP );
|
|
if ( m_Socket == INVALID_SOCKET )
|
|
{
|
|
Warning( "CMasterMulticastThread::Init - socket() failed\n" );
|
|
return false;
|
|
}
|
|
|
|
// Bind to INADDR_ANY.
|
|
CIPAddr localAddr( 0, 0, 0, 0, localPort );
|
|
|
|
sockaddr_in addr;
|
|
IPAddrToSockAddr( &localAddr, &addr );
|
|
|
|
int status = bind( m_Socket, (sockaddr*)&addr, sizeof(addr) );
|
|
if ( status != 0 )
|
|
{
|
|
Term();
|
|
Warning( "CMasterMulticastThread::Init - bind( %d.%d.%d.%d:%d ) failed\n", EXPAND_ADDR( *pAddr ) );
|
|
return false;
|
|
}
|
|
|
|
if ( VMPI_GetFileSystemMode() == VMPI_FILESYSTEM_BROADCAST )
|
|
{
|
|
// Set up for broadcast
|
|
BOOL bBroadcast = TRUE;
|
|
if ( setsockopt( m_Socket, SOL_SOCKET, SO_BROADCAST, (char*)&bBroadcast, bBroadcast ) == SOCKET_ERROR )
|
|
{
|
|
Term();
|
|
Warning( "CMasterMulticastThread::Init - setsockopt() failed to set broadcast mode\n" );
|
|
return false;
|
|
}
|
|
}
|
|
|
|
// Remember the address we want to send to.
|
|
IPAddrToSockAddr( pAddr, &m_MulticastAddr );
|
|
|
|
// Now create our thread.
|
|
DWORD dwThreadID = 0;
|
|
m_hThread = CreateThread( NULL, 0, &CMasterMulticastThread::StaticMulticastThread, this, 0, &dwThreadID );
|
|
if ( !m_hThread )
|
|
{
|
|
Term();
|
|
Warning( "CMasterMulticastThread::Init - CreateThread failed\n" );
|
|
return false;
|
|
}
|
|
|
|
SetThreadPriority( m_hThread, THREAD_PRIORITY_LOWEST );
|
|
}
|
|
|
|
// For debug mode to verify that we don't try to open files while in another thread.
|
|
m_hMainThread = GetCurrentThread();
|
|
|
|
m_pPassThru = pPassThru;
|
|
return true;
|
|
}
|
|
|
|
|
|
void CMasterMulticastThread::Term()
|
|
{
|
|
// Stop the thread if it is running.
|
|
if ( m_hThread )
|
|
{
|
|
SetEvent( m_hTermEvent );
|
|
WaitForSingleObject( m_hThread, INFINITE );
|
|
CloseHandle( m_hThread );
|
|
|
|
m_hThread = NULL;
|
|
}
|
|
|
|
// Close the socket.
|
|
if ( m_Socket != INVALID_SOCKET )
|
|
{
|
|
closesocket( m_Socket );
|
|
m_Socket = INVALID_SOCKET;
|
|
}
|
|
|
|
// Free up other data.
|
|
m_Files.PurgeAndDeleteElements();
|
|
m_nCurMemoryUsage = m_nMaxMemoryUsage = 0;
|
|
}
|
|
|
|
|
|
void CMasterMulticastThread::TCP_SendNextChunk( CMulticastFile *pFile, CClientFileInfo *pClient )
|
|
{
|
|
// No more chunks to send?
|
|
if ( (pClient->m_TCP_LastChunkSent+1) >= pFile->m_Info.m_nChunks )
|
|
return;
|
|
|
|
// Figure out what data we'd be sending.
|
|
int iChunkToSend = pClient->m_TCP_LastChunkSent + 1;
|
|
int iStartByte = iChunkToSend * TCP_CHUNK_PAYLOAD_SIZE;
|
|
int iEndByte = min( iStartByte + TCP_CHUNK_PAYLOAD_SIZE, pFile->m_Data.Count() );
|
|
|
|
// If the start point is past the end, then we're done sending the file to this client.
|
|
if ( iStartByte >= pFile->m_Data.Count() )
|
|
return;
|
|
|
|
// Record that we sent this data.
|
|
pClient->m_TCP_LastChunkSent = iChunkToSend;
|
|
|
|
// Assemble the packet.
|
|
unsigned char cPacket[2] = { VMPI_PACKETID_FILESYSTEM, VMPI_FSPACKETID_FILE_CHUNK };
|
|
|
|
const void *chunks[5] =
|
|
{
|
|
cPacket,
|
|
&pFile->m_Info,
|
|
&iChunkToSend,
|
|
pFile->GetFilename(),
|
|
&pFile->m_Data[iStartByte]
|
|
};
|
|
|
|
int chunkLengths[5] =
|
|
{
|
|
sizeof( cPacket ),
|
|
sizeof( pFile->m_Info ),
|
|
sizeof( m_iCurActiveChunk ),
|
|
strlen( pFile->GetFilename() ) + 1,
|
|
iEndByte - iStartByte
|
|
};
|
|
|
|
VMPI_SendChunks( chunks, chunkLengths, 5, pClient->m_ClientID );
|
|
}
|
|
|
|
|
|
int CMasterMulticastThread::AddFileRequest( const char *pFilename, const char *pPathID, int clientID, bool *bZeroLength )
|
|
{
|
|
// Firstly, do we already have this file?
|
|
int iFile = FindOrAddFile( pFilename, pPathID );
|
|
if ( iFile == -1 )
|
|
return -1;
|
|
|
|
CMulticastFile *pFile = m_Files[iFile];
|
|
|
|
// Now that we have a file setup, merge in this client's info.
|
|
EnterCriticalSection( &m_CS );
|
|
|
|
CClientFileInfo *pClient = new CClientFileInfo;
|
|
pClient->m_TCP_LastChunkAcked = -1;
|
|
pClient->m_TCP_LastChunkSent = -1;
|
|
pClient->m_ClientID = clientID;
|
|
pClient->m_flLastAckTime = Plat_FloatTime();
|
|
pClient->m_flTransmitStartTime = pClient->m_flLastAckTime;
|
|
pClient->m_nTimesFileCycled = 0;
|
|
pClient->m_nChunksLeft = pFile->m_Info.m_nChunks;
|
|
pClient->m_ChunksToSend.SetSize( PAD_NUMBER( pFile->m_Info.m_nChunks, 8 ) / 8 );
|
|
memset( pClient->m_ChunksToSend.Base(), 0xFF, pClient->m_ChunksToSend.Count() );
|
|
pFile->m_Clients.AddToTail( pClient );
|
|
|
|
for ( int i=0; i < pFile->m_Chunks.Count(); i++ )
|
|
{
|
|
IncrementChunkRefCount( pFile, i );
|
|
}
|
|
|
|
// In TCP mode, let's get the sliding window started..
|
|
if ( VMPI_GetFileSystemMode() == VMPI_FILESYSTEM_TCP )
|
|
{
|
|
for ( int iDepth=0; iDepth < TCP_CHUNK_QUEUE_LEN; iDepth++ )
|
|
TCP_SendNextChunk( pFile, pClient );
|
|
}
|
|
|
|
LeaveCriticalSection( &m_CS );
|
|
|
|
*bZeroLength = (pFile->m_Info.m_UncompressedSize == 0);
|
|
|
|
return iFile;
|
|
}
|
|
|
|
|
|
void CMasterMulticastThread::OnChunkReceived( int fileID, int clientID, int iChunk )
|
|
{
|
|
if ( !m_Files.IsValidIndex( fileID ) )
|
|
{
|
|
Warning( "CMasterMulticastThread::OnChunkReceived: invalid file (%d) from client %d\n", fileID, clientID );
|
|
return;
|
|
}
|
|
|
|
CMulticastFile *pFile = m_Files[fileID];
|
|
CClientFileInfo *pClient = NULL;
|
|
FOR_EACH_LL( pFile->m_Clients, iClient )
|
|
{
|
|
if ( pFile->m_Clients[iClient]->m_ClientID == clientID )
|
|
{
|
|
pClient = pFile->m_Clients[iClient];
|
|
break;
|
|
}
|
|
}
|
|
if ( !pClient )
|
|
{
|
|
// This will spam sometimes if a worker stops responding and we timeout on it,
|
|
// but then it comes back alive and starts responding. So let's ignore its packets silently.
|
|
//Warning( "CMasterMulticastThread::OnChunkReceived: invalid client ID (%d) for file %s\n", clientID, pFile->GetFilename() );
|
|
return;
|
|
}
|
|
|
|
if ( VMPI_GetFileSystemMode() == VMPI_FILESYSTEM_TCP )
|
|
{
|
|
// Send the next chunk, if there is one.
|
|
EnterCriticalSection( &m_CS );
|
|
TCP_SendNextChunk( pFile, pClient );
|
|
LeaveCriticalSection( &m_CS );
|
|
}
|
|
else
|
|
{
|
|
if ( !pFile->m_Chunks.IsValidIndex( iChunk ) )
|
|
{
|
|
Warning( "CMasterMulticastThread::OnChunkReceived: invalid chunk index (%d) for file %s\n", iChunk, pFile->GetFilename() );
|
|
return;
|
|
}
|
|
|
|
// Mark that this client doesn't need this chunk anymore.
|
|
pClient->m_ChunksToSend[iChunk >> 3] &= ~(1 << (iChunk & 7));
|
|
pClient->m_nChunksLeft--;
|
|
|
|
pClient->m_flLastAckTime = Plat_FloatTime();
|
|
if ( pClient->m_nChunksLeft == 0 && g_iVMPIVerboseLevel >= 2 )
|
|
Warning( "Client %d got file %s\n", clientID, pFile->GetFilename() );
|
|
|
|
EnterCriticalSection( &m_CS );
|
|
DecrementChunkRefCount( fileID, iChunk );
|
|
LeaveCriticalSection( &m_CS );
|
|
}
|
|
}
|
|
|
|
|
|
void CMasterMulticastThread::OnFileReceived( int fileID, int clientID )
|
|
{
|
|
if ( !m_Files.IsValidIndex( fileID ) )
|
|
{
|
|
Warning( "CMasterMulticastThread::OnChunkReceived: invalid file (%d) from client %d\n", fileID, clientID );
|
|
return;
|
|
}
|
|
|
|
CMulticastFile *pFile = m_Files[fileID];
|
|
for ( int i=0; i < pFile->m_Info.m_nChunks; i++ )
|
|
OnChunkReceived( fileID, clientID, i );
|
|
}
|
|
|
|
|
|
void CMasterMulticastThread::OnClientDisconnect( int clientID, bool bGrabCriticalSection )
|
|
{
|
|
if ( bGrabCriticalSection )
|
|
EnterCriticalSection( &m_CS );
|
|
|
|
// Remove all references from this client.
|
|
FOR_EACH_LL( m_Files, iFile )
|
|
{
|
|
CMulticastFile *pFile = m_Files[iFile];
|
|
|
|
FOR_EACH_LL( pFile->m_Clients, iClient )
|
|
{
|
|
CClientFileInfo *pClient = pFile->m_Clients[iClient];
|
|
|
|
if ( pClient->m_ClientID != clientID )
|
|
continue;
|
|
|
|
// Ok, this is our man. Decrement the refcount of any chunks this client wanted.
|
|
for ( int iChunk=0; iChunk < pFile->m_Info.m_nChunks; iChunk++ )
|
|
{
|
|
if ( pClient->NeedsChunk( iChunk ) )
|
|
{
|
|
DecrementChunkRefCount( iFile, iChunk );
|
|
}
|
|
}
|
|
|
|
delete pClient;
|
|
pFile->m_Clients.Remove( iClient );
|
|
|
|
break;
|
|
}
|
|
}
|
|
|
|
if ( bGrabCriticalSection )
|
|
LeaveCriticalSection( &m_CS );
|
|
}
|
|
|
|
|
|
void CMasterMulticastThread::CreateVirtualFile( const char *pFilename, const void *pData, unsigned long fileLength )
|
|
{
|
|
const char *pPathID = VMPI_VIRTUAL_FILES_PATH_ID;
|
|
|
|
int iFile = FindFile( pFilename, pPathID );
|
|
if ( iFile != -1 )
|
|
Error( "CMasterMulticastThread::CreateVirtualFile( %s ) - file already exists!", pFilename );
|
|
|
|
CMulticastFile *pFile = new CMulticastFile;
|
|
pFile->m_UncompressedData.CopyArray( (const char*)pData, fileLength );
|
|
|
|
FinishFileSetup( pFile, pFilename, pPathID, false );
|
|
}
|
|
|
|
|
|
DWORD WINAPI CMasterMulticastThread::StaticMulticastThread( LPVOID pParameter )
|
|
{
|
|
return ((CMasterMulticastThread*)pParameter)->MulticastThread();
|
|
}
|
|
|
|
|
|
bool CMasterMulticastThread::CheckClientTimeouts()
|
|
{
|
|
bool bRet = false;
|
|
CMulticastFile *pFile = m_Files[m_iCurFile];
|
|
|
|
float flCurTime = Plat_FloatTime();
|
|
|
|
int iNext;
|
|
for( int iCur=pFile->m_Clients.Head(); iCur != pFile->m_Clients.InvalidIndex(); iCur=iNext )
|
|
{
|
|
iNext = pFile->m_Clients.Next( iCur );
|
|
CClientFileInfo *pInfo = pFile->m_Clients[iCur];
|
|
|
|
// If the client has already fully received this file, don't bother timing out on it.
|
|
if ( pInfo->m_nChunksLeft == 0 )
|
|
continue;
|
|
|
|
++pInfo->m_nTimesFileCycled;
|
|
if ( pInfo->m_nTimesFileCycled >= MIN_FILE_CYCLE_COUNT && (flCurTime - pInfo->m_flLastAckTime) > CLIENT_FILE_ACK_TIMEOUT )
|
|
{
|
|
// For debug output, get the most recent time we heard any ack from this client at all.
|
|
float flMostRecentTime = pInfo->m_flLastAckTime;
|
|
FOR_EACH_LL( m_Files, iTestFile )
|
|
{
|
|
CMulticastFile *pTestFile = m_Files[iTestFile];
|
|
FOR_EACH_LL( pTestFile->m_Clients, iTestClient )
|
|
{
|
|
if ( pTestFile->m_Clients[iTestClient]->m_ClientID == pInfo->m_ClientID )
|
|
{
|
|
flMostRecentTime = max( flMostRecentTime, pTestFile->m_Clients[iTestClient]->m_flLastAckTime );
|
|
}
|
|
}
|
|
}
|
|
|
|
Warning( "\nClient %s timed out on file %s (latest: %.2f / cur: %.2f).\n",
|
|
VMPI_GetMachineName( pInfo->m_ClientID ), pFile->GetFilename(), flMostRecentTime, flCurTime );
|
|
OnClientDisconnect( pInfo->m_ClientID, false );
|
|
bRet = true; // yes, we booted a client.
|
|
}
|
|
}
|
|
|
|
return bRet;
|
|
}
|
|
|
|
inline bool CMasterMulticastThread::Thread_SendFileChunk_Multicast( int *pnBytesSent )
|
|
{
|
|
// Send the next chunk (file, size, time, chunk data).
|
|
CMulticastFile *pFile = m_Files[m_iCurFile];
|
|
|
|
int iStartByte = m_iCurActiveChunk * MULTICAST_CHUNK_PAYLOAD_SIZE;
|
|
int iEndByte = min( iStartByte + MULTICAST_CHUNK_PAYLOAD_SIZE, pFile->m_Data.Count() );
|
|
|
|
WSABUF bufs[4];
|
|
bufs[0].buf = (char*)&pFile->m_Info;
|
|
bufs[0].len = sizeof( pFile->m_Info );
|
|
|
|
bufs[1].buf = (char*)&m_iCurActiveChunk;
|
|
bufs[1].len = sizeof( m_iCurActiveChunk );
|
|
|
|
bufs[2].buf = (char*)pFile->GetFilename();
|
|
bufs[2].len = strlen( pFile->GetFilename() ) + 1;
|
|
|
|
bufs[3].buf = &pFile->m_Data[iStartByte];
|
|
bufs[3].len = iEndByte - iStartByte;
|
|
|
|
DWORD nBytesSent = 0;
|
|
DWORD nWantedBytes = ( bufs[0].len + bufs[1].len + bufs[2].len + bufs[3].len );
|
|
bool bSuccess;
|
|
|
|
if ( m_MulticastAddr.sin_addr.S_un.S_un_b.s_b1 == 127 &&
|
|
m_MulticastAddr.sin_addr.S_un.S_un_b.s_b2 == 0 &&
|
|
m_MulticastAddr.sin_addr.S_un.S_un_b.s_b3 == 0 &&
|
|
m_MulticastAddr.sin_addr.S_un.S_un_b.s_b4 == 1 )
|
|
{
|
|
// For some mysterious reason, WSASendTo only sends the first buffer
|
|
// if we're sending to 127.0.0.1 (ie: in local mode).
|
|
char allData[1024*8];
|
|
if ( nWantedBytes > sizeof( allData ) )
|
|
Error( "nWantedBytes > sizeof( allData )" );
|
|
|
|
memcpy( &allData[0], bufs[0].buf, bufs[0].len );
|
|
memcpy( &allData[bufs[0].len], bufs[1].buf, bufs[1].len );
|
|
memcpy( &allData[bufs[0].len+bufs[1].len], bufs[2].buf, bufs[2].len );
|
|
memcpy( &allData[bufs[0].len+bufs[1].len+bufs[2].len], bufs[3].buf, bufs[3].len );
|
|
int ret = sendto( m_Socket, allData, nWantedBytes, 0, (sockaddr*)&m_MulticastAddr, sizeof( m_MulticastAddr ) );
|
|
bSuccess = (ret == (int)nWantedBytes);
|
|
}
|
|
else
|
|
{
|
|
WSASendTo(
|
|
m_Socket,
|
|
bufs,
|
|
ARRAYSIZE( bufs ),
|
|
&nBytesSent,
|
|
0,
|
|
(sockaddr*)&m_MulticastAddr,
|
|
sizeof( m_MulticastAddr ),
|
|
NULL,
|
|
NULL );
|
|
bSuccess = (nBytesSent == nWantedBytes);
|
|
}
|
|
|
|
// Handle errors.. let it get a few errors, then quit.
|
|
if ( bSuccess )
|
|
{
|
|
*pnBytesSent = (int)nBytesSent;
|
|
}
|
|
else
|
|
{
|
|
static int nWarnings = 0;
|
|
++nWarnings;
|
|
if ( nWarnings < 10 )
|
|
{
|
|
Warning( "\nMulticastThread: WSASendTo with %d bytes sent %d bytes.\n", nWantedBytes, nBytesSent );
|
|
|
|
char *lpMsgBuf;
|
|
if ( FormatMessage(
|
|
FORMAT_MESSAGE_ALLOCATE_BUFFER |
|
|
FORMAT_MESSAGE_FROM_SYSTEM |
|
|
FORMAT_MESSAGE_IGNORE_INSERTS,
|
|
NULL,
|
|
GetLastError(),
|
|
MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT), // Default language
|
|
(char*)&lpMsgBuf,
|
|
0,
|
|
NULL
|
|
) )
|
|
{
|
|
Warning( "%s", lpMsgBuf );
|
|
LocalFree( lpMsgBuf );
|
|
}
|
|
}
|
|
else if ( nWarnings == 10 )
|
|
{
|
|
Warning( "\nThis machine's ability to multicast may be broken. Please reboot and try again.\n" );
|
|
}
|
|
}
|
|
|
|
return bSuccess;
|
|
}
|
|
|
|
|
|
void CMasterMulticastThread::Thread_SeekToNextActiveChunk()
|
|
{
|
|
// Make sure we're on a valid chunk.
|
|
if ( m_iCurFile == -1 )
|
|
{
|
|
Assert( m_Files.Count() > 0 );
|
|
m_iCurFile = m_Files.Head();
|
|
m_iCurActiveChunk = m_Files[m_iCurFile]->m_ActiveChunks.Head();
|
|
}
|
|
|
|
while ( 1 )
|
|
{
|
|
if ( m_iCurActiveChunk == m_Files[m_iCurFile]->m_ActiveChunks.InvalidIndex() ||
|
|
m_Files[m_iCurFile]->m_ActiveChunks[m_iCurActiveChunk]->m_RefCount == 0 )
|
|
{
|
|
// Now check for client timeouts.
|
|
// (This is kicking clients unjustly for some reason.. need to debug).
|
|
if ( CheckClientTimeouts() && m_nTotalActiveChunks == 0 )
|
|
break;
|
|
|
|
// Finished with that file. Send the next one.
|
|
m_iCurFile = m_Files.Next( m_iCurFile );
|
|
if ( m_iCurFile == m_Files.InvalidIndex() )
|
|
m_iCurFile = m_Files.Head();
|
|
|
|
m_iCurActiveChunk = m_Files[m_iCurFile]->m_ActiveChunks.Head();
|
|
}
|
|
|
|
if ( m_iCurActiveChunk != m_Files[m_iCurFile]->m_ActiveChunks.InvalidIndex() )
|
|
{
|
|
// Only break if we're on an active chunk.
|
|
if ( m_Files[m_iCurFile]->m_ActiveChunks[m_iCurActiveChunk]->m_RefCount != 0 )
|
|
{
|
|
break;
|
|
}
|
|
|
|
m_iCurActiveChunk = m_Files[m_iCurFile]->m_ActiveChunks.Next( m_iCurActiveChunk );
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
DWORD CMasterMulticastThread::MulticastThread()
|
|
{
|
|
CTransmitRateMgr transmitRateMgr;
|
|
CRateLimiter rateLimiter;
|
|
|
|
|
|
DWORD msToWait = 0; // Only temporarily used if we don't have any data to send.
|
|
|
|
while ( WaitForSingleObject( m_hTermEvent, msToWait ) != WAIT_OBJECT_0 )
|
|
{
|
|
rateLimiter.GiveUpTimeSlice();
|
|
msToWait = 0;
|
|
|
|
EnterCriticalSection( &m_CS );
|
|
|
|
transmitRateMgr.ReadPackets();
|
|
|
|
// If we have nothing to send then kick back for a while.
|
|
if ( m_nTotalActiveChunks == 0 )
|
|
{
|
|
LeaveCriticalSection( &m_CS );
|
|
msToWait = 50;
|
|
continue;
|
|
}
|
|
|
|
// Ok, now we're active, so send out our presence to other CTransmitRateMgrs on the network.
|
|
transmitRateMgr.BroadcastPresence();
|
|
|
|
|
|
// We're going to time how long this chunk took to send.
|
|
CFastTimer timer;
|
|
timer.Start();
|
|
|
|
Thread_SeekToNextActiveChunk();
|
|
|
|
// We have to do this check a second time here because the CheckClientTimeouts() call may have
|
|
// booted our last client. If we don't check it here, we might be transmitting
|
|
// something we don't want to transmit. Also, if we don't break out of the loop above,
|
|
// it can prevent the process from ever exiting because it'll never exit that while() loop.
|
|
if ( m_nTotalActiveChunks == 0 )
|
|
{
|
|
LeaveCriticalSection( &m_CS );
|
|
msToWait = 50;
|
|
continue;
|
|
}
|
|
|
|
int nBytesSent = 0;
|
|
|
|
bool bSuccess;
|
|
bSuccess = Thread_SendFileChunk_Multicast( &nBytesSent );
|
|
|
|
g_nMulticastBytesSent += (int)nBytesSent;
|
|
|
|
// Move to the next chunk.
|
|
m_iCurActiveChunk = m_Files[m_iCurFile]->m_ActiveChunks.Next( m_iCurActiveChunk );
|
|
|
|
LeaveCriticalSection( &m_CS );
|
|
|
|
|
|
// Measure how long it took to send this.
|
|
timer.End();
|
|
unsigned long timeTaken = timer.GetDuration().GetMicroseconds();
|
|
|
|
|
|
// Measure how long it should have taken.
|
|
unsigned long estimatedPacketHeaderSize = 32;
|
|
unsigned long optimalTimeTaken = (unsigned long)( transmitRateMgr.GetMicrosecondsPerByte() * (nBytesSent + estimatedPacketHeaderSize) );
|
|
|
|
|
|
// If we went faster than we should have, then wait for the difference in time.
|
|
if ( timeTaken < optimalTimeTaken )
|
|
{
|
|
rateLimiter.NoteExcessTimeTaken( optimalTimeTaken - timeTaken );
|
|
}
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
|
|
void CMasterMulticastThread::IncrementChunkRefCount( CMasterMulticastThread::CMulticastFile *pFile, int iChunk )
|
|
{
|
|
CChunkInfo *pChunk = &pFile->m_Chunks[iChunk];
|
|
|
|
if ( pChunk->m_RefCount == 0 )
|
|
{
|
|
++m_nTotalActiveChunks;
|
|
|
|
// Move the chunk to the head of the list since it is now active.
|
|
pFile->m_ActiveChunks.Remove( pChunk->m_iActiveChunksIndex );
|
|
pChunk->m_iActiveChunksIndex = pFile->m_ActiveChunks.AddToHead( pChunk );
|
|
}
|
|
|
|
pChunk->m_RefCount++;
|
|
}
|
|
|
|
|
|
void CMasterMulticastThread::DecrementChunkRefCount( int iFile, int iChunk )
|
|
{
|
|
CMulticastFile *pFile = m_Files[iFile];
|
|
CChunkInfo *pChunk = &pFile->m_Chunks[iChunk];
|
|
|
|
if ( pChunk->m_RefCount == 0 )
|
|
{
|
|
Error( "CMasterMulticastThread::DecrementChunkRefCount - refcount already zero!\n" );
|
|
}
|
|
|
|
pChunk->m_RefCount--;
|
|
if ( pChunk->m_RefCount == 0 )
|
|
{
|
|
--m_nTotalActiveChunks;
|
|
|
|
// If this is the current chunk the thread is reading on, seek up to the next chunk so
|
|
// the thread doesn't spin off into the next file and skip its current file's contents.
|
|
if ( iFile == m_iCurFile && pChunk->m_iActiveChunksIndex == m_iCurActiveChunk )
|
|
{
|
|
m_iCurActiveChunk = pFile->m_ActiveChunks.Next( pChunk->m_iActiveChunksIndex );
|
|
}
|
|
|
|
// Move the chunk to the end of the list since it is now inactive.
|
|
pFile->m_ActiveChunks.Remove( pChunk->m_iActiveChunksIndex );
|
|
pChunk->m_iActiveChunksIndex = pFile->m_ActiveChunks.AddToTail( pChunk );
|
|
}
|
|
}
|
|
|
|
|
|
int CMasterMulticastThread::FindFile( const char *pName, const char *pPathID )
|
|
{
|
|
FOR_EACH_LL( m_Files, i )
|
|
{
|
|
CMulticastFile *pFile = m_Files[i];
|
|
if ( stricmp( pFile->GetFilename(), pName ) == 0 && stricmp( pFile->GetPathID(), pPathID ) == 0 )
|
|
return i;
|
|
}
|
|
return -1;
|
|
}
|
|
|
|
|
|
bool CMasterMulticastThread::FindWarningSuppression( const char *pFilename )
|
|
{
|
|
FOR_EACH_LL( m_WarningSuppressions, i )
|
|
{
|
|
if ( Q_stricmp( m_WarningSuppressions[i], pFilename ) == 0 )
|
|
return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
|
|
void CMasterMulticastThread::AddWarningSuppression( const char *pFilename )
|
|
{
|
|
char *pBlah = new char[ strlen( pFilename ) + 1 ];
|
|
strcpy( pBlah, pFilename );
|
|
m_WarningSuppressions.AddToTail( pBlah );
|
|
}
|
|
|
|
|
|
int CMasterMulticastThread::FindOrAddFile( const char *pFilename, const char *pPathID )
|
|
{
|
|
CMulticastFile *pFile = NULL;
|
|
bool bFileAlreadyExisted = false;
|
|
|
|
// See if we've already opened this file.
|
|
int iFile = FindFile( pFilename, pPathID );
|
|
if ( iFile != -1 )
|
|
{
|
|
pFile = m_Files[iFile];
|
|
if ( pFile->m_bDataLoaded )
|
|
{
|
|
return iFile;
|
|
}
|
|
else
|
|
{
|
|
// Ok, we have the file entry, but its data has been freed, so we need to reload it.
|
|
EnterCriticalSection( &m_CS );
|
|
bFileAlreadyExisted = true;
|
|
}
|
|
}
|
|
|
|
// Can't open a file outside our main thread, because we have to talk to the filesystem
|
|
// and the filesystem doesn't support that.
|
|
Assert( GetCurrentThread() == m_hMainThread );
|
|
|
|
// When the worker originally asked for the path ID, they could pass NULL and it would come through as "".
|
|
// Now set it back to null for the filesystem we're passing the call to.
|
|
FileHandle_t fp = m_pPassThru->Open( pFilename, "rb", pPathID[0] == 0 ? NULL : pPathID );
|
|
if ( !fp )
|
|
{
|
|
if ( bFileAlreadyExisted )
|
|
LeaveCriticalSection( &m_CS );
|
|
|
|
return -1;
|
|
}
|
|
|
|
if ( !bFileAlreadyExisted )
|
|
pFile = new CMulticastFile;
|
|
|
|
pFile->m_UncompressedData.SetSize( m_pPassThru->Size( fp ) );
|
|
m_pPassThru->Read( pFile->m_UncompressedData.Base(), pFile->m_UncompressedData.Count(), fp );
|
|
m_pPassThru->Close( fp );
|
|
|
|
int iRet = FinishFileSetup( pFile, pFilename, pPathID, bFileAlreadyExisted );
|
|
if ( bFileAlreadyExisted )
|
|
LeaveCriticalSection( &m_CS );
|
|
|
|
return iRet;
|
|
}
|
|
|
|
|
|
int CMasterMulticastThread::FinishFileSetup( CMulticastFile *pFile, const char *pFilename, const char *pPathID, bool bFileAlreadyExisted )
|
|
{
|
|
// Compress the file's contents.
|
|
if ( !ZLibCompress( pFile->m_UncompressedData.Base(), pFile->m_UncompressedData.Count(), pFile->m_Data ) )
|
|
{
|
|
delete pFile;
|
|
return -1;
|
|
}
|
|
|
|
pFile->m_bDataLoaded = true;
|
|
int chunkSize = VMPI_GetChunkPayloadSize();
|
|
|
|
// Get this file in the queue.
|
|
if ( !bFileAlreadyExisted )
|
|
{
|
|
pFile->m_Filename.SetSize( strlen( pFilename ) + 1 );
|
|
strcpy( pFile->m_Filename.Base(), pFilename );
|
|
|
|
pFile->m_PathID.SetSize( strlen( pPathID ) + 1 );
|
|
strcpy( pFile->m_PathID.Base(), pPathID );
|
|
|
|
pFile->m_nCycles = 0;
|
|
|
|
pFile->m_Info.m_CompressedSize = pFile->m_Data.Count();
|
|
pFile->m_Info.m_UncompressedSize = pFile->m_UncompressedData.Count();
|
|
|
|
pFile->m_Info.m_nChunks = PAD_NUMBER( pFile->m_Info.m_CompressedSize, chunkSize ) / chunkSize;
|
|
|
|
// Initialize the chunks.
|
|
pFile->m_Chunks.SetSize( pFile->m_Info.m_nChunks );
|
|
for ( int i=0; i < pFile->m_Chunks.Count(); i++ )
|
|
{
|
|
CChunkInfo *pChunk = &pFile->m_Chunks[i];
|
|
|
|
pChunk->m_iChunk = (unsigned short)i;
|
|
pChunk->m_RefCount = 0;
|
|
pChunk->m_iActiveChunksIndex = pFile->m_ActiveChunks.AddToTail( pChunk );
|
|
}
|
|
|
|
EnterCriticalSection( &m_CS );
|
|
}
|
|
|
|
// Boot some other file out of memory if we're out of space.
|
|
m_nCurMemoryUsage += ( pFile->m_Info.m_CompressedSize + pFile->m_Info.m_UncompressedSize );
|
|
EnsureMemoryLimit( pFile );
|
|
|
|
if ( !bFileAlreadyExisted )
|
|
{
|
|
pFile->m_Info.m_FileID = m_Files.AddToTail( pFile );
|
|
LeaveCriticalSection( &m_CS );
|
|
}
|
|
|
|
return pFile->m_Info.m_FileID;
|
|
}
|
|
|
|
|
|
void CMasterMulticastThread::EnsureMemoryLimit( CMulticastFile *pIgnore )
|
|
{
|
|
if ( m_nMaxMemoryUsage != 0 && m_nCurMemoryUsage > m_nMaxMemoryUsage )
|
|
{
|
|
// Free all the files that we can.
|
|
FOR_EACH_LL( m_Files, iFile )
|
|
{
|
|
CMulticastFile *pFile = m_Files[iFile];
|
|
if ( pFile == pIgnore || !pFile->m_bDataLoaded )
|
|
continue;
|
|
|
|
if ( pFile->m_ActiveChunks.Count() == 0 )
|
|
{
|
|
m_nCurMemoryUsage -= ( pFile->m_Info.m_CompressedSize + pFile->m_Info.m_UncompressedSize );
|
|
|
|
pFile->m_Data.Purge();
|
|
pFile->m_UncompressedData.Purge();
|
|
pFile->m_bDataLoaded = false;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
const CUtlVector<char>& CMasterMulticastThread::GetFileData( int iFile ) const
|
|
{
|
|
return m_Files[iFile]->m_UncompressedData;
|
|
}
|
|
|
|
|
|
// ------------------------------------------------------------------------------------------------------------------------ //
|
|
// CMasterVMPIFileSystem implementation.
|
|
// ------------------------------------------------------------------------------------------------------------------------ //
|
|
|
|
class CMasterVMPIFileSystem : public CBaseVMPIFileSystem
|
|
{
|
|
public:
|
|
CMasterVMPIFileSystem();
|
|
virtual ~CMasterVMPIFileSystem();
|
|
|
|
bool Init( int maxMemoryUsage, IFileSystem *pPassThru );
|
|
virtual void Term();
|
|
|
|
virtual FileHandle_t Open( const char *pFilename, const char *pOptions, const char *pathID );
|
|
virtual bool HandleFileSystemPacket( MessageBuffer *pBuf, int iSource, int iPacketID );
|
|
|
|
virtual void CreateVirtualFile( const char *pFilename, const void *pData, int fileLength );
|
|
|
|
virtual CSysModule *LoadModule( const char *pFileName, const char *pPathID, bool bValidatedDllOnly );
|
|
virtual void UnloadModule( CSysModule *pModule );
|
|
|
|
|
|
private:
|
|
|
|
static void OnClientDisconnect( int procID, const char *pReason );
|
|
|
|
|
|
private:
|
|
CMasterMulticastThread m_MasterThread;
|
|
IFileSystem *m_pMasterVMPIFileSystemPassThru;
|
|
|
|
static CMasterVMPIFileSystem *s_pMasterVMPIFileSystem;
|
|
};
|
|
|
|
|
|
CMasterVMPIFileSystem *CMasterVMPIFileSystem::s_pMasterVMPIFileSystem = NULL;
|
|
|
|
|
|
CBaseVMPIFileSystem* CreateMasterVMPIFileSystem( int maxMemoryUsage, IFileSystem *pPassThru )
|
|
{
|
|
CMasterVMPIFileSystem *pRet = new CMasterVMPIFileSystem;
|
|
g_pBaseVMPIFileSystem = pRet;
|
|
if ( pRet->Init( maxMemoryUsage, pPassThru ) )
|
|
{
|
|
return pRet;
|
|
}
|
|
else
|
|
{
|
|
delete pRet;
|
|
g_pBaseVMPIFileSystem = NULL;
|
|
return NULL;
|
|
}
|
|
}
|
|
|
|
|
|
CMasterVMPIFileSystem::CMasterVMPIFileSystem()
|
|
{
|
|
Assert( !s_pMasterVMPIFileSystem );
|
|
s_pMasterVMPIFileSystem = this;
|
|
}
|
|
|
|
|
|
CMasterVMPIFileSystem::~CMasterVMPIFileSystem()
|
|
{
|
|
Assert( s_pMasterVMPIFileSystem == this );
|
|
s_pMasterVMPIFileSystem = NULL;
|
|
}
|
|
|
|
|
|
bool CMasterVMPIFileSystem::Init( int maxMemoryUsage, IFileSystem *pPassThru )
|
|
{
|
|
// Only init the BASE filesystem passthru. Leave the IFileSystem passthru using NULL so it'll crash
|
|
// immediately if they try to use a function we don't support.
|
|
InitPassThru( pPassThru, true );
|
|
m_pMasterVMPIFileSystemPassThru = pPassThru;
|
|
|
|
// Pick a random IP in the multicast range (224.0.0.2 to 239.255.255.255);
|
|
CCycleCount cnt;
|
|
cnt.Sample();
|
|
RandomSeed( (int)cnt.GetMicroseconds() );
|
|
|
|
int localPort = 23412; // This can be anything.
|
|
|
|
unsigned short port = RandomInt( 22000, 25000 );
|
|
if ( VMPI_GetRunMode() == VMPI_RUN_NETWORKED )
|
|
{
|
|
if ( VMPI_GetFileSystemMode() == VMPI_FILESYSTEM_MULTICAST )
|
|
{
|
|
m_MulticastIP.port = port;
|
|
m_MulticastIP.ip[0] = (unsigned char)RandomInt( 225, 238 );
|
|
m_MulticastIP.ip[1] = (unsigned char)RandomInt( 0, 255 );
|
|
m_MulticastIP.ip[2] = (unsigned char)RandomInt( 0, 255 );
|
|
m_MulticastIP.ip[3] = (unsigned char)RandomInt( 3, 255 );
|
|
}
|
|
else if ( VMPI_GetFileSystemMode() == VMPI_FILESYSTEM_BROADCAST )
|
|
{
|
|
m_MulticastIP.Init( 0xFF, 0xFF, 0xFF, 0xFF, port );
|
|
}
|
|
}
|
|
else
|
|
{
|
|
// Doesn't matter.. we don't use the multicast IP in TCP mode.
|
|
m_MulticastIP.Init( 0, 0, 0, 0, 0 );
|
|
}
|
|
|
|
if ( !m_MasterThread.Init( pPassThru, localPort, &m_MulticastIP, maxMemoryUsage ) )
|
|
return false;
|
|
|
|
// Send out the multicast addr to all the clients.
|
|
SendMulticastIP( &m_MulticastIP );
|
|
|
|
// Make sure we're notified when a client disconnects so we can unlink them from the
|
|
// multicast thread's structures.
|
|
VMPI_AddDisconnectHandler( &CMasterVMPIFileSystem::OnClientDisconnect );
|
|
return true;
|
|
}
|
|
|
|
|
|
void CMasterVMPIFileSystem::Term()
|
|
{
|
|
m_MasterThread.Term();
|
|
}
|
|
|
|
|
|
FileHandle_t CMasterVMPIFileSystem::Open( const char *pFilename, const char *pOptions, const char *pPathID )
|
|
{
|
|
Assert( g_bUseMPI );
|
|
|
|
if ( g_bDisableFileAccess )
|
|
Error( "Open( %s, %s ) - file access has been disabled.", pFilename, pOptions );
|
|
|
|
// Use a stdio file if they want to write to it.
|
|
bool bWriteAccess = IsOpeningForWriteAccess( pOptions );
|
|
if ( bWriteAccess )
|
|
{
|
|
FileHandle_t fp = m_pBaseFileSystemPassThru->Open( pFilename, pOptions, pPathID );
|
|
if ( fp == FILESYSTEM_INVALID_HANDLE )
|
|
return FILESYSTEM_INVALID_HANDLE;
|
|
|
|
CVMPIFile_PassThru *pFile = new CVMPIFile_PassThru;
|
|
pFile->Init( m_pBaseFileSystemPassThru, fp );
|
|
return (FileHandle_t)pFile;
|
|
}
|
|
|
|
// Internally, we require path IDs to be non-null. We'll convert it back to null whenever we make filesystem calls though.
|
|
if ( !pPathID )
|
|
pPathID = "";
|
|
|
|
// Have our multicast thread load all the data so it's there when workers want it.
|
|
int iFile = m_MasterThread.FindOrAddFile( pFilename, pPathID );
|
|
if ( iFile == -1 )
|
|
return FILESYSTEM_INVALID_HANDLE;
|
|
|
|
const CUtlVector<char> &data = m_MasterThread.GetFileData( iFile );
|
|
|
|
CVMPIFile_Memory *pFile = new CVMPIFile_Memory;
|
|
pFile->Init( data.Base(), data.Count(), strchr( pOptions, 't' ) ? 't' : 'b' );
|
|
return (FileHandle_t)pFile;
|
|
}
|
|
|
|
|
|
void CMasterVMPIFileSystem::OnClientDisconnect( int procID, const char *pReason )
|
|
{
|
|
s_pMasterVMPIFileSystem->m_MasterThread.OnClientDisconnect( procID );
|
|
}
|
|
|
|
|
|
void CMasterVMPIFileSystem::CreateVirtualFile( const char *pFilename, const void *pData, int fileLength )
|
|
{
|
|
m_MasterThread.CreateVirtualFile( pFilename, pData, fileLength );
|
|
}
|
|
|
|
bool CMasterVMPIFileSystem::HandleFileSystemPacket( MessageBuffer *pBuf, int iSource, int iPacketID )
|
|
{
|
|
// Handle this packet.
|
|
int subPacketID = pBuf->data[1];
|
|
switch( subPacketID )
|
|
{
|
|
case VMPI_FSPACKETID_FILE_REQUEST:
|
|
{
|
|
int requestID = *((int*)&pBuf->data[2]);
|
|
const char *pFilename = (const char*)&pBuf->data[6];
|
|
const char *pPathID = (const char*)pFilename + strlen( pFilename ) + 1;
|
|
|
|
if ( g_iVMPIVerboseLevel >= 2 )
|
|
Msg( "Client %d requested '%s'\n", iSource, pFilename );
|
|
|
|
bool bZeroLength;
|
|
int fileID = m_MasterThread.AddFileRequest( pFilename, pPathID, iSource, &bZeroLength );
|
|
|
|
// Send back the file ID.
|
|
unsigned char cPacket[2] = { VMPI_PACKETID_FILESYSTEM, VMPI_FSPACKETID_FILE_RESPONSE };
|
|
void *pChunks[4] = { cPacket, &requestID, &fileID, &bZeroLength };
|
|
int chunkLen[4] = { sizeof( cPacket ), sizeof( requestID ), sizeof( fileID ), sizeof( bZeroLength ) };
|
|
|
|
VMPI_SendChunks( pChunks, chunkLen, ARRAYSIZE( pChunks ), iSource );
|
|
}
|
|
return true;
|
|
|
|
case VMPI_FSPACKETID_CHUNK_RECEIVED:
|
|
{
|
|
unsigned short *pFileID = (unsigned short*)&pBuf->data[2];
|
|
unsigned short *pChunkID = pFileID+1;
|
|
|
|
int nChunks = (pBuf->getLen() - 2) / 4;
|
|
for ( int i=0; i < nChunks; i++ )
|
|
{
|
|
m_MasterThread.OnChunkReceived( *pFileID, iSource, *pChunkID );
|
|
pFileID += 2;
|
|
pChunkID += 2;
|
|
}
|
|
}
|
|
return true;
|
|
|
|
case VMPI_FSPACKETID_FILE_RECEIVED:
|
|
{
|
|
unsigned short *pFileID = (unsigned short*)&pBuf->data[2];
|
|
m_MasterThread.OnFileReceived( *pFileID, iSource );
|
|
}
|
|
return true;
|
|
|
|
default:
|
|
return false;
|
|
}
|
|
}
|
|
|
|
|
|
CSysModule* CMasterVMPIFileSystem::LoadModule( const char *pFileName, const char *pPathID, bool bValidatedDllOnly )
|
|
{
|
|
return m_pMasterVMPIFileSystemPassThru->LoadModule( pFileName, pPathID, bValidatedDllOnly );
|
|
}
|
|
|
|
void CMasterVMPIFileSystem::UnloadModule( CSysModule *pModule )
|
|
{
|
|
m_pMasterVMPIFileSystemPassThru->UnloadModule( pModule );
|
|
}
|
|
|