Team Fortress 2 Source Code as on 22/4/2020
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

815 lines
23 KiB

  1. //========= Copyright Valve Corporation, All rights reserved. ============//
  2. //
  3. // Purpose:
  4. //
  5. //=============================================================================//
  6. #include <winsock2.h>
  7. #include "vmpi_filesystem_internal.h"
  8. #include "threadhelpers.h"
  9. #include "zlib.h"
  10. #define NUM_BUFFERED_CHUNK_ACKS 512
  11. #define ACK_FLUSH_INTERVAL 500 // Flush the ack queue twice per second.
  12. static bool g_bReceivedMulticastIP = false;
  13. static CIPAddr g_MulticastIP;
  14. CCriticalSection g_FileResponsesCS;
  15. class CFileResponse
  16. {
  17. public:
  18. int m_RequestID;
  19. int m_Response;
  20. bool m_bZeroLength;
  21. };
  22. CUtlVector<CFileResponse> g_FileResponses;
  23. int g_RequestID = 0;
  24. class CFileChunkPacket
  25. {
  26. public:
  27. int m_Len;
  28. char m_Data[1];
  29. };
  30. CUtlLinkedList<CFileChunkPacket*, int> g_FileChunkPackets; // This is also protected by g_FileResponsesCS.
  31. // ------------------------------------------------------------------------------------------------------------------------ //
  32. // Classes.
  33. // ------------------------------------------------------------------------------------------------------------------------ //
  34. class CWorkerFile
  35. {
  36. public:
  37. const char* GetFilename() { return m_Filename.Base(); }
  38. const char* GetPathID() { return m_PathID.Base(); }
  39. bool IsReadyToRead() const { return m_nChunksToReceive == 0; }
  40. public:
  41. CFastTimer m_Timer; // To see how long it takes to download the file.
  42. // This has to be sent explicitly as part of the file info or else the protocol
  43. // breaks on empty files.
  44. bool m_bZeroLength;
  45. // This is false until we get any packets about the file. In the packets,
  46. // we find out what the size is supposed to be.
  47. bool m_bGotCompressedSize;
  48. // The ID the master uses to refer to this file.
  49. int m_FileID;
  50. CUtlVector<char> m_Filename;
  51. CUtlVector<char> m_PathID;
  52. // First data comes in here, then when it's all there, it is inflated into m_UncompressedData.
  53. CUtlVector<char> m_CompressedData;
  54. // 1 bit for each chunk.
  55. CUtlVector<unsigned char> m_ChunksReceived;
  56. // When this is zero, the file is done being received and m_UncompressedData is valid.
  57. int m_nChunksToReceive;
  58. CUtlVector<char> m_UncompressedData;
  59. };
  60. // ------------------------------------------------------------------------------------------------------------------------ //
  61. // Global helpers.
  62. // ------------------------------------------------------------------------------------------------------------------------ //
  63. static void RecvMulticastIP( CIPAddr *pAddr )
  64. {
  65. while ( !g_bReceivedMulticastIP )
  66. VMPI_DispatchNextMessage();
  67. *pAddr = g_MulticastIP;
  68. }
  69. static bool ZLibDecompress( const void *pInput, int inputLen, void *pOut, int outLen )
  70. {
  71. if ( inputLen == 0 )
  72. {
  73. // Zero-length file?
  74. return true;
  75. }
  76. z_stream decompressStream;
  77. // Initialize the decompression stream.
  78. memset( &decompressStream, 0, sizeof( decompressStream ) );
  79. if ( inflateInit( &decompressStream ) != Z_OK )
  80. return false;
  81. // Decompress all this stuff and write it to the file.
  82. decompressStream.next_in = (unsigned char*)pInput;
  83. decompressStream.avail_in = inputLen;
  84. char *pOutChar = (char*)pOut;
  85. while ( decompressStream.avail_in )
  86. {
  87. decompressStream.total_out = 0;
  88. decompressStream.next_out = (unsigned char*)pOutChar;
  89. decompressStream.avail_out = outLen - (pOutChar - (char*)pOut);
  90. int ret = inflate( &decompressStream, Z_NO_FLUSH );
  91. if ( ret != Z_OK && ret != Z_STREAM_END )
  92. return false;
  93. pOutChar += decompressStream.total_out;
  94. if ( ret == Z_STREAM_END )
  95. {
  96. if ( (pOutChar - (char*)pOut) == outLen )
  97. {
  98. return true;
  99. }
  100. else
  101. {
  102. Assert( false );
  103. return false;
  104. }
  105. }
  106. }
  107. Assert( false ); // Should have gotten to Z_STREAM_END.
  108. return false;
  109. }
  110. // ------------------------------------------------------------------------------------------------------------------------ //
  111. // CWorkerMulticastListener implementation.
  112. // ------------------------------------------------------------------------------------------------------------------------ //
  113. class CWorkerMulticastListener
  114. {
  115. public:
  116. CWorkerMulticastListener()
  117. {
  118. m_nUnfinishedFiles = 0;
  119. }
  120. ~CWorkerMulticastListener()
  121. {
  122. Term();
  123. }
  124. bool Init( const CIPAddr &mcAddr )
  125. {
  126. m_MulticastAddr = mcAddr;
  127. m_hMainThread = GetCurrentThread();
  128. return true;
  129. }
  130. void Term()
  131. {
  132. m_WorkerFiles.PurgeAndDeleteElements();
  133. }
  134. CWorkerFile* RequestFileFromServer( const char *pFilename, const char *pPathID )
  135. {
  136. Assert( pPathID );
  137. Assert( FindWorkerFile( pFilename, pPathID ) == NULL );
  138. // Send a request to the master to find out if this file even exists.
  139. CCriticalSectionLock csLock( &g_FileResponsesCS );
  140. csLock.Lock();
  141. int requestID = g_RequestID++;
  142. csLock.Unlock();
  143. unsigned char packetID[2] = { VMPI_PACKETID_FILESYSTEM, VMPI_FSPACKETID_FILE_REQUEST };
  144. const void *pChunks[4] = { packetID, &requestID, (void*)pFilename, pPathID };
  145. int chunkLengths[4] = { sizeof( packetID ), sizeof( requestID ), strlen( pFilename ) + 1, strlen( pPathID ) + 1 };
  146. VMPI_SendChunks( pChunks, chunkLengths, ARRAYSIZE( pChunks ), 0 );
  147. // Wait for the file ID to come back.
  148. CFileResponse response;
  149. response.m_Response = -1;
  150. response.m_bZeroLength = true;
  151. // We're in a worker thread.. the main thread should be dispatching all the messages, so let it
  152. // do that until we get our response.
  153. while ( 1 )
  154. {
  155. bool bGotIt = false;
  156. csLock.Lock();
  157. for ( int iResponse=0; iResponse < g_FileResponses.Count(); iResponse++ )
  158. {
  159. if ( g_FileResponses[iResponse].m_RequestID == requestID )
  160. {
  161. response = g_FileResponses[iResponse];
  162. g_FileResponses.Remove( iResponse );
  163. bGotIt = true;
  164. break;
  165. }
  166. }
  167. csLock.Unlock();
  168. if ( bGotIt )
  169. break;
  170. if ( GetCurrentThread() == m_hMainThread )
  171. VMPI_DispatchNextMessage( 20 );
  172. else
  173. Sleep( 20 );
  174. }
  175. // If we get -1 back, it means the file doesn't exist.
  176. int fileID = response.m_Response;
  177. if ( fileID == -1 )
  178. return NULL;
  179. CWorkerFile *pTestFile = new CWorkerFile;
  180. pTestFile->m_Filename.SetSize( strlen( pFilename ) + 1 );
  181. strcpy( pTestFile->m_Filename.Base(), pFilename );
  182. pTestFile->m_PathID.SetSize( strlen( pPathID ) + 1 );
  183. strcpy( pTestFile->m_PathID.Base(), pPathID );
  184. pTestFile->m_FileID = fileID;
  185. pTestFile->m_nChunksToReceive = 9999;
  186. pTestFile->m_Timer.Start();
  187. m_WorkerFiles.AddToTail( pTestFile );
  188. pTestFile->m_bGotCompressedSize = false;
  189. pTestFile->m_bZeroLength = response.m_bZeroLength;
  190. ++m_nUnfinishedFiles;
  191. return pTestFile;
  192. }
  193. void FlushAckChunks( unsigned short chunksToAck[NUM_BUFFERED_CHUNK_ACKS][2], int &nChunksToAck, DWORD &lastAckTime )
  194. {
  195. if ( nChunksToAck )
  196. {
  197. // Tell the master we received this chunk.
  198. unsigned char packetID[2] = { VMPI_PACKETID_FILESYSTEM, VMPI_FSPACKETID_CHUNK_RECEIVED };
  199. void *pChunks[2] = { packetID, chunksToAck };
  200. int chunkLengths[2] = { sizeof( packetID ), nChunksToAck * 4 };
  201. VMPI_SendChunks( pChunks, chunkLengths, 2, 0 );
  202. nChunksToAck = 0;
  203. }
  204. lastAckTime = GetTickCount();
  205. }
  206. void MaybeFlushAckChunks( unsigned short chunksToAck[NUM_BUFFERED_CHUNK_ACKS][2], int &nChunksToAck, DWORD &lastAckTime )
  207. {
  208. if ( nChunksToAck && GetTickCount() - lastAckTime > ACK_FLUSH_INTERVAL )
  209. FlushAckChunks( chunksToAck, nChunksToAck, lastAckTime );
  210. }
  211. void AddAckChunk(
  212. unsigned short chunksToAck[NUM_BUFFERED_CHUNK_ACKS][2],
  213. int &nChunksToAck,
  214. DWORD &lastAckTime,
  215. int fileID,
  216. int iChunk )
  217. {
  218. chunksToAck[nChunksToAck][0] = (unsigned short)fileID;
  219. chunksToAck[nChunksToAck][1] = (unsigned short)iChunk;
  220. // TCP filesystem acks all chunks immediately so it'll send more.
  221. ++nChunksToAck;
  222. if ( nChunksToAck == NUM_BUFFERED_CHUNK_ACKS || VMPI_GetFileSystemMode() == VMPI_FILESYSTEM_TCP )
  223. {
  224. FlushAckChunks( chunksToAck, nChunksToAck, lastAckTime );
  225. }
  226. }
  227. // Returns the length of the packet's data or -1 if there is nothing.
  228. int CheckFileChunkPackets( char *data, int dataSize )
  229. {
  230. // Using TCP.. pop the next received packet off the stack.
  231. CCriticalSectionLock csLock( &g_FileResponsesCS );
  232. csLock.Lock();
  233. if ( g_FileChunkPackets.Count() <= 0 )
  234. return -1;
  235. CFileChunkPacket *pPacket = g_FileChunkPackets[ g_FileChunkPackets.Head() ];
  236. g_FileChunkPackets.Remove( g_FileChunkPackets.Head() );
  237. // Yes, this is inefficient, but the amount of data we're handling here is tiny so the
  238. // effect is negligible.
  239. int len;
  240. if ( pPacket->m_Len > dataSize )
  241. {
  242. len = -1;
  243. Warning( "CWorkerMulticastListener::ListenFor: Got a section of data too long (%d bytes).", pPacket->m_Len );
  244. }
  245. else
  246. {
  247. memcpy( data, pPacket->m_Data, pPacket->m_Len );
  248. len = pPacket->m_Len;
  249. }
  250. free( pPacket );
  251. return len;
  252. }
  253. void ShowSDKWorkerMsg( const char *pMsg, ... )
  254. {
  255. if ( !g_bMPIMaster && VMPI_IsSDKMode() )
  256. {
  257. va_list marker;
  258. va_start( marker, pMsg );
  259. char str[4096];
  260. V_vsnprintf( str, sizeof( str ), pMsg, marker );
  261. va_end( marker );
  262. Msg( "%s", str );
  263. }
  264. }
  265. // This is the main function the workers use to pick files out of the multicast stream.
  266. // The app is waiting for a specific file, but we receive and ack any files we can until
  267. // we get the file they're looking for, then we return.
  268. //
  269. // NOTE: ideally, this would be in a thread, but it adds lots of complications and may
  270. // not be worth it.
  271. CWorkerFile* ListenFor( const char *pFilename, const char *pPathID )
  272. {
  273. CWorkerFile *pFile = FindWorkerFile( pFilename, pPathID );
  274. if ( !pFile )
  275. {
  276. // Ok, we haven't requested this file yet. Create an entry for it and
  277. // tell the master we'd like this file.
  278. pFile = RequestFileFromServer( pFilename, pPathID );
  279. if ( !pFile )
  280. return NULL;
  281. // If it's zero-length, we can return right now.
  282. if ( pFile->m_bZeroLength )
  283. {
  284. --m_nUnfinishedFiles;
  285. return pFile;
  286. }
  287. }
  288. // Setup a filename to print some debug spew with.
  289. char printableFilename[58];
  290. if ( V_strlen( pFilename ) > ARRAYSIZE( printableFilename ) - 1 )
  291. {
  292. V_strncpy( printableFilename, "[...]", sizeof( printableFilename ) );
  293. V_strncat( printableFilename, &pFilename[V_strlen(pFilename) - ARRAYSIZE(printableFilename) + 1 + V_strlen(printableFilename)], sizeof( printableFilename ) );
  294. }
  295. else
  296. {
  297. V_strncpy( printableFilename, pFilename, sizeof( printableFilename ) );
  298. }
  299. ShowSDKWorkerMsg( "\rRecv %s (0%%) ", printableFilename );
  300. int iChunkPayloadSize = VMPI_GetChunkPayloadSize();
  301. // Now start listening to the stream.
  302. // Note: no need to setup anything when in TCP mode - we just use the regular
  303. // VMPI dispatch stuff to handle that.
  304. ISocket *pSocket = NULL;
  305. if ( VMPI_GetFileSystemMode() == VMPI_FILESYSTEM_MULTICAST )
  306. {
  307. pSocket = CreateMulticastListenSocket( m_MulticastAddr );
  308. if ( !pSocket )
  309. {
  310. char str[512];
  311. IP_GetLastErrorString( str, sizeof( str ) );
  312. Warning( "CreateMulticastListenSocket (%d.%d.%d.%d:%d) failed\n%s\n", EXPAND_ADDR( m_MulticastAddr ), str );
  313. return NULL;
  314. }
  315. }
  316. else if ( VMPI_GetFileSystemMode() == VMPI_FILESYSTEM_BROADCAST )
  317. {
  318. pSocket = CreateIPSocket();
  319. if ( !pSocket->BindToAny( m_MulticastAddr.port ) )
  320. {
  321. pSocket->Release();
  322. pSocket = NULL;
  323. }
  324. }
  325. unsigned short chunksToAck[NUM_BUFFERED_CHUNK_ACKS][2];
  326. int nChunksToAck = 0;
  327. DWORD lastAckTime = GetTickCount();
  328. // Now just receive multicast data until this file has been received.
  329. while ( m_nUnfinishedFiles > 0 )
  330. {
  331. char data[MAX_CHUNK_PAYLOAD_SIZE+1024];
  332. int len = -1;
  333. if ( pSocket )
  334. {
  335. CIPAddr ipFrom;
  336. len = pSocket->RecvFrom( data, sizeof( data ), &ipFrom );
  337. }
  338. else
  339. {
  340. len = CheckFileChunkPackets( data, sizeof( data ) );
  341. }
  342. if ( len == -1 )
  343. {
  344. // Sleep for 10ms and also handle socket errors.
  345. Sleep( 0 );
  346. VMPI_DispatchNextMessage( 10 );
  347. continue;
  348. }
  349. g_nMulticastBytesReceived += len;
  350. // Alrighty. Figure out what the deal is with this file.
  351. CMulticastFileInfo *pInfo = (CMulticastFileInfo*)data;
  352. int *piChunk = (int*)( pInfo + 1 );
  353. const char *pTestFilename = (const char*)( piChunk + 1 );
  354. const char *pPayload = pTestFilename + strlen( pFilename ) + 1;
  355. int payloadLen = len - ( pPayload - data );
  356. if ( payloadLen < 0 )
  357. {
  358. Warning( "CWorkerMulticastListener::ListenFor: invalid packet received on multicast group\n" );
  359. continue;
  360. }
  361. if ( pInfo->m_FileID != pFile->m_FileID )
  362. continue;
  363. CWorkerFile *pTestFile = FindWorkerFile( pInfo->m_FileID );
  364. if ( !pTestFile )
  365. Error( "FindWorkerFile( %s ) failed\n", pTestFilename );
  366. // TODO: reenable this code and disable the if right above here.
  367. // We always get "invalid payload length" errors on the workers when using this, but
  368. // I haven't been able to figure out why yet.
  369. /*
  370. // Put the data into whatever file it belongs in.
  371. if ( !pTestFile )
  372. {
  373. pTestFile = RequestFileFromServer( pTestFilename );
  374. if ( !pTestFile )
  375. continue;
  376. }
  377. */
  378. // Is this the first packet about this file?
  379. if ( !pTestFile->m_bGotCompressedSize )
  380. {
  381. pTestFile->m_bGotCompressedSize = true;
  382. pTestFile->m_CompressedData.SetSize( pInfo->m_CompressedSize );
  383. pTestFile->m_UncompressedData.SetSize( pInfo->m_UncompressedSize );
  384. pTestFile->m_ChunksReceived.SetSize( PAD_NUMBER( pInfo->m_nChunks, 8 ) / 8 );
  385. pTestFile->m_nChunksToReceive = pInfo->m_nChunks;
  386. memset( pTestFile->m_ChunksReceived.Base(), 0, pTestFile->m_ChunksReceived.Count() );
  387. }
  388. // Validate the chunk index and uncompressed size.
  389. int iChunk = *piChunk;
  390. if ( iChunk < 0 || iChunk >= pInfo->m_nChunks )
  391. {
  392. Error( "ListenFor(): invalid chunk index (%d) for file '%s'\n", iChunk, pTestFilename );
  393. }
  394. // Only handle this if we didn't already received the chunk.
  395. if ( !(pTestFile->m_ChunksReceived[iChunk >> 3] & (1 << (iChunk & 7))) )
  396. {
  397. // Make sure the file is properly setup to receive the data into.
  398. if ( (int)pInfo->m_UncompressedSize != pTestFile->m_UncompressedData.Count() ||
  399. (int)pInfo->m_CompressedSize != pTestFile->m_CompressedData.Count() )
  400. {
  401. Error( "ListenFor(): invalid compressed or uncompressed size.\n"
  402. "pInfo = '%s', pTestFile = '%s'\n"
  403. "Compressed (pInfo = %d, pTestFile = %d)\n"
  404. "Uncompressed (pInfo = %d, pTestFile = %d)\n",
  405. pTestFilename,
  406. pTestFile->GetFilename(),
  407. pInfo->m_CompressedSize,
  408. pTestFile->m_CompressedData.Count(),
  409. pInfo->m_UncompressedSize,
  410. pTestFile->m_UncompressedData.Count()
  411. );
  412. }
  413. int iChunkStart = iChunk * iChunkPayloadSize;
  414. int iChunkEnd = min( iChunkStart + iChunkPayloadSize, pTestFile->m_CompressedData.Count() );
  415. int chunkLen = iChunkEnd - iChunkStart;
  416. if ( chunkLen != payloadLen )
  417. {
  418. Error( "ListenFor(): invalid payload length for '%s' (%d should be %d)\n"
  419. "pInfo = '%s', pTestFile = '%s'\n"
  420. "Chunk %d out of %d. Compressed size: %d\n",
  421. pTestFile->GetFilename(),
  422. payloadLen,
  423. chunkLen,
  424. pTestFilename,
  425. pTestFile->GetFilename(),
  426. iChunk,
  427. pInfo->m_nChunks,
  428. pInfo->m_CompressedSize
  429. );
  430. }
  431. memcpy( &pTestFile->m_CompressedData[iChunkStart], pPayload, chunkLen );
  432. pTestFile->m_ChunksReceived[iChunk >> 3] |= (1 << (iChunk & 7));
  433. --pTestFile->m_nChunksToReceive;
  434. if ( pTestFile == pFile )
  435. {
  436. int percent = 100 - (100 * pFile->m_nChunksToReceive) / pInfo->m_nChunks;
  437. ShowSDKWorkerMsg( "\rRecv %s (%d%%) [chunk %d/%d] ", printableFilename, percent, pInfo->m_nChunks - pFile->m_nChunksToReceive, pInfo->m_nChunks );
  438. }
  439. // Remember to ack what we received.
  440. AddAckChunk( chunksToAck, nChunksToAck, lastAckTime, pInfo->m_FileID, iChunk );
  441. // If we're done receiving the data, unpack it.
  442. if ( pTestFile->m_nChunksToReceive == 0 )
  443. {
  444. // Ack the file.
  445. FlushAckChunks( chunksToAck, nChunksToAck, lastAckTime );
  446. pTestFile->m_Timer.End();
  447. pTestFile->m_UncompressedData.SetSize( pInfo->m_UncompressedSize );
  448. --m_nUnfinishedFiles;
  449. if ( !ZLibDecompress(
  450. pTestFile->m_CompressedData.Base(),
  451. pTestFile->m_CompressedData.Count(),
  452. pTestFile->m_UncompressedData.Base(),
  453. pTestFile->m_UncompressedData.Count() ) )
  454. {
  455. if ( pSocket )
  456. pSocket->Release();
  457. FlushAckChunks( chunksToAck, nChunksToAck, lastAckTime );
  458. Error( "ZLibDecompress failed.\n" );
  459. return NULL;
  460. }
  461. char str[512];
  462. V_snprintf( str, sizeof( str ), "Got %s (%dk) in %.2fs",
  463. printableFilename,
  464. (pTestFile->m_UncompressedData.Count() + 511) / 1024,
  465. pTestFile->m_Timer.GetDuration().GetSeconds()
  466. );
  467. Msg( "\r%-79s\n", str );
  468. // Won't be needing this anymore.
  469. pTestFile->m_CompressedData.Purge();
  470. }
  471. }
  472. MaybeFlushAckChunks( chunksToAck, nChunksToAck, lastAckTime );
  473. }
  474. Assert( pFile->IsReadyToRead() );
  475. FlushAckChunks( chunksToAck, nChunksToAck, lastAckTime );
  476. if ( pSocket )
  477. pSocket->Release();
  478. return pFile;
  479. }
  480. CWorkerFile* FindWorkerFile( const char *pFilename, const char *pPathID )
  481. {
  482. FOR_EACH_LL( m_WorkerFiles, i )
  483. {
  484. CWorkerFile *pWorkerFile = m_WorkerFiles[i];
  485. if ( stricmp( pWorkerFile->GetFilename(), pFilename ) == 0 && stricmp( pWorkerFile->GetPathID(), pPathID ) == 0 )
  486. return pWorkerFile;
  487. }
  488. return NULL;
  489. }
  490. CWorkerFile* FindWorkerFile( int fileID )
  491. {
  492. FOR_EACH_LL( m_WorkerFiles, i )
  493. {
  494. if ( m_WorkerFiles[i]->m_FileID == fileID )
  495. return m_WorkerFiles[i];
  496. }
  497. return NULL;
  498. }
  499. private:
  500. CIPAddr m_MulticastAddr;
  501. CUtlLinkedList<CWorkerFile*, int> m_WorkerFiles;
  502. HANDLE m_hMainThread;
  503. // How many files do we have open that we haven't finished receiving from the server yet?
  504. // We always keep waiting for data until this is zero.
  505. int m_nUnfinishedFiles;
  506. };
  507. // ------------------------------------------------------------------------------------------------------------------------ //
  508. // CWorkerVMPIFileSystem implementation.
  509. // ------------------------------------------------------------------------------------------------------------------------ //
  510. class CWorkerVMPIFileSystem : public CBaseVMPIFileSystem
  511. {
  512. public:
  513. InitReturnVal_t Init();
  514. virtual void Term();
  515. virtual FileHandle_t Open( const char *pFilename, const char *pOptions, const char *pathID );
  516. virtual bool HandleFileSystemPacket( MessageBuffer *pBuf, int iSource, int iPacketID );
  517. virtual void CreateVirtualFile( const char *pFilename, const void *pData, int fileLength );
  518. virtual long GetFileTime( const char *pFileName, const char *pathID );
  519. virtual bool IsFileWritable( const char *pFileName, const char *pPathID );
  520. virtual bool SetFileWritable( char const *pFileName, bool writable, const char *pPathID );
  521. virtual CSysModule *LoadModule( const char *pFileName, const char *pPathID, bool bValidatedDllOnly );
  522. virtual void UnloadModule( CSysModule *pModule );
  523. private:
  524. CWorkerMulticastListener m_Listener;
  525. };
  526. CBaseVMPIFileSystem* CreateWorkerVMPIFileSystem()
  527. {
  528. CWorkerVMPIFileSystem *pRet = new CWorkerVMPIFileSystem;
  529. g_pBaseVMPIFileSystem = pRet;
  530. if ( pRet->Init() )
  531. {
  532. return pRet;
  533. }
  534. else
  535. {
  536. delete pRet;
  537. g_pBaseVMPIFileSystem = NULL;
  538. return NULL;
  539. }
  540. }
  541. InitReturnVal_t CWorkerVMPIFileSystem::Init()
  542. {
  543. // Get the multicast addr to listen on.
  544. CIPAddr mcAddr;
  545. RecvMulticastIP( &mcAddr );
  546. return m_Listener.Init( mcAddr ) ? INIT_OK : INIT_FAILED;
  547. }
  548. void CWorkerVMPIFileSystem::Term()
  549. {
  550. m_Listener.Term();
  551. }
  552. FileHandle_t CWorkerVMPIFileSystem::Open( const char *pFilename, const char *pOptions, const char *pathID )
  553. {
  554. Assert( g_bUseMPI );
  555. // When it finally asks the filesystem for a file, it'll pass NULL for pathID if it's "".
  556. if ( !pathID )
  557. pathID = "";
  558. if ( g_bDisableFileAccess )
  559. Error( "Open( %s, %s ) - file access has been disabled.", pFilename, pOptions );
  560. // Workers can't open anything for write access.
  561. bool bWriteAccess = (Q_stristr( pOptions, "w" ) != 0);
  562. if ( bWriteAccess )
  563. return FILESYSTEM_INVALID_HANDLE;
  564. // Do we have this file's data already?
  565. CWorkerFile *pFile = m_Listener.FindWorkerFile( pFilename, pathID );
  566. if ( !pFile || !pFile->IsReadyToRead() )
  567. {
  568. // Ok, start listening to the multicast stream until we get the file we want.
  569. // NOTE: it might make sense here to have the client ask for a list of ALL the files that
  570. // the master currently has and wait to receive all of them (so we don't come back a bunch
  571. // of times and listen
  572. // NOTE NOTE: really, the best way to do this is to have a thread on the workers that sits there
  573. // and listens to the multicast stream. Any time the master opens a new file up, it assumes
  574. // all the workers need the file, and it starts to send it on the multicast stream until
  575. // the worker threads respond that they all have it.
  576. //
  577. // (NOTE: this probably means that the clients would have to ack the chunks on a UDP socket that
  578. // the thread owns).
  579. //
  580. // This would simplify all the worries about a client missing half the stream and having to
  581. // wait for another cycle through it.
  582. pFile = m_Listener.ListenFor( pFilename, pathID );
  583. if ( !pFile )
  584. {
  585. return FILESYSTEM_INVALID_HANDLE;
  586. }
  587. }
  588. // Ok! Got the file. now setup a memory stream they can read out of it with.
  589. CVMPIFile_Memory *pOut = new CVMPIFile_Memory;
  590. pOut->Init( pFile->m_UncompressedData.Base(), pFile->m_UncompressedData.Count(), strchr( pOptions, 't' ) ? 't' : 'b' );
  591. return (FileHandle_t)pOut;
  592. }
  593. void CWorkerVMPIFileSystem::CreateVirtualFile( const char *pFilename, const void *pData, int fileLength )
  594. {
  595. Error( "CreateVirtualFile not supported in VMPI worker filesystem." );
  596. }
  597. long CWorkerVMPIFileSystem::GetFileTime( const char *pFileName, const char *pathID )
  598. {
  599. Error( "GetFileTime not supported in VMPI worker filesystem." );
  600. return 0;
  601. }
  602. bool CWorkerVMPIFileSystem::IsFileWritable( const char *pFileName, const char *pPathID )
  603. {
  604. Error( "GetFileTime not supported in VMPI worker filesystem." );
  605. return false;
  606. }
  607. bool CWorkerVMPIFileSystem::SetFileWritable( char const *pFileName, bool writable, const char *pPathID )
  608. {
  609. Error( "GetFileTime not supported in VMPI worker filesystem." );
  610. return false;
  611. }
  612. bool CWorkerVMPIFileSystem::HandleFileSystemPacket( MessageBuffer *pBuf, int iSource, int iPacketID )
  613. {
  614. // Handle this packet.
  615. int subPacketID = pBuf->data[1];
  616. switch( subPacketID )
  617. {
  618. case VMPI_FSPACKETID_MULTICAST_ADDR:
  619. {
  620. char *pInPos = &pBuf->data[2];
  621. g_MulticastIP = *((CIPAddr*)pInPos);
  622. pInPos += sizeof( g_MulticastIP );
  623. g_bReceivedMulticastIP = true;
  624. }
  625. return true;
  626. case VMPI_FSPACKETID_FILE_RESPONSE:
  627. {
  628. CCriticalSectionLock csLock( &g_FileResponsesCS );
  629. csLock.Lock();
  630. CFileResponse res;
  631. res.m_RequestID = *((int*)&pBuf->data[2]);
  632. res.m_Response = *((int*)&pBuf->data[6]);
  633. res.m_bZeroLength = *((bool*)&pBuf->data[10]);
  634. g_FileResponses.AddToTail( res );
  635. }
  636. return true;
  637. case VMPI_FSPACKETID_FILE_CHUNK:
  638. {
  639. int nDataBytes = pBuf->getLen() - 2;
  640. CFileChunkPacket *pPacket = (CFileChunkPacket*)malloc( sizeof( CFileChunkPacket ) + nDataBytes - 1 );
  641. memcpy( pPacket->m_Data, &pBuf->data[2], nDataBytes );
  642. pPacket->m_Len = nDataBytes;
  643. CCriticalSectionLock csLock( &g_FileResponsesCS );
  644. csLock.Lock();
  645. g_FileChunkPackets.AddToTail( pPacket );
  646. }
  647. return true;
  648. default:
  649. return false;
  650. }
  651. }
  652. CSysModule* CWorkerVMPIFileSystem::LoadModule( const char *pFileName, const char *pPathID, bool bValidatedDllOnly )
  653. {
  654. return Sys_LoadModule( pFileName );
  655. }
  656. void CWorkerVMPIFileSystem::UnloadModule( CSysModule *pModule )
  657. {
  658. Sys_UnloadModule( pModule );
  659. }