jann/m2dev-client-src
1
0
Fork 0
forked from metin-server/m2dev-client-src
Code Pull Requests Activity

Consolidate file loading threading to CGameThreadPool

Browse Source
This commit is contained in:
savis
2026-01-05 17:42:13 +01:00
parent f19dfa1fe2
commit 4e68003fdd
6 changed files with 50 additions and 581 deletions
Download Patch File Download Diff File Expand all files Collapse all files

159
src/EterLib/FileLoaderThread.cpp
View File

@@ -3,171 +3,86 @@
#include "PackLib/PackManager.h" #include "PackLib/PackManager.h"
#include "FileLoaderThread.h" #include "FileLoaderThread.h"
#include "ResourceManager.h" #include "ResourceManager.h"
#include "GameThreadPool.h"
CFileLoaderThread::CFileLoaderThread() : m_bShutdowned(false), m_pArg(NULL), m_hThread(NULL), m_uThreadID(0) CFileLoaderThread::CFileLoaderThread() : m_bShutdowned(false)
{ {
} }
CFileLoaderThread::~CFileLoaderThread() CFileLoaderThread::~CFileLoaderThread()
{ {
Destroy(); Shutdown();
} }
int CFileLoaderThread::Create(void * arg) bool CFileLoaderThread::Create(void * arg)
{ {
Arg(arg); // Modern implementation doesn't need explicit thread creation
m_hThread = (HANDLE) _beginthreadex(NULL, 0, EntryPoint, this, 0, &m_uThreadID); // The global CGameThreadPool handles threading
m_bShutdowned = false;
if (!m_hThread)
return false;
SetThreadPriority(m_hThread, THREAD_PRIORITY_NORMAL);
return true; return true;
} }
UINT CFileLoaderThread::Run(void * arg)
{
if (!Setup())
return 0;
return (Execute(arg));
}
/* Static */
UINT CALLBACK CFileLoaderThread::EntryPoint(void * pThis)
{
CFileLoaderThread * pThread = (CFileLoaderThread *) pThis;
return pThread->Run(pThread->Arg());
}
//////////////////////////////////////////////////////////////////////////
void CFileLoaderThread::Destroy()
{
if (m_hSemaphore)
{
CloseHandle(m_hSemaphore);
m_hSemaphore = NULL;
}
stl_wipe(m_pRequestDeque);
stl_wipe(m_pCompleteDeque);
}
UINT CFileLoaderThread::Setup()
{
m_hSemaphore = CreateSemaphore(NULL, // no security attributes
0, // initial count
65535, // maximum count
NULL); // unnamed semaphore
if (!m_hSemaphore)
return 0;
return 1;
}
void CFileLoaderThread::Shutdown() void CFileLoaderThread::Shutdown()
{ {
if (!m_hSemaphore)
return;
BOOL bRet;
m_bShutdowned = true; m_bShutdowned = true;
do // Clear any pending completed items
{ {
bRet = ReleaseSemaphore(m_hSemaphore, 1, NULL); std::lock_guard<std::mutex> lock(m_CompleteMutex);
stl_wipe(m_pCompleteDeque);
} }
while (!bRet);
WaitForSingleObject(m_hThread, 10000); // 쓰레드가 종료 되기를 10초 기다림
} }
UINT CFileLoaderThread::Execute(void * /*pvArg*/) void CFileLoaderThread::Request(const std::string& c_rstFileName)
{ {
while (!m_bShutdowned) if (m_bShutdowned)
return;
// Enqueue file loading to the global thread pool
CGameThreadPool* pThreadPool = CGameThreadPool::InstancePtr();
if (pThreadPool)
{ {
DWORD dwWaitResult; pThreadPool->Enqueue([this, c_rstFileName]()
dwWaitResult = WaitForSingleObject(m_hSemaphore, INFINITE);
if (m_bShutdowned)
break;
switch (dwWaitResult)
{ {
case WAIT_OBJECT_0: ProcessFile(c_rstFileName);
{ });
Process(); }
} else
break; {
// Fallback to synchronous loading if thread pool not available
case WAIT_TIMEOUT: ProcessFile(c_rstFileName);
TraceError("CFileLoaderThread::Execute: Timeout occured while time-out interval is INIFITE");
break;
}
} }
Destroy();
return 1;
} }
void CFileLoaderThread::Request(std::string & c_rstFileName) // called in main thread bool CFileLoaderThread::Fetch(TData ** ppData)
{ {
TData * pData = new TData; std::lock_guard<std::mutex> lock(m_CompleteMutex);
pData->File.clear();
pData->stFileName = c_rstFileName;
m_RequestMutex.Lock();
m_pRequestDeque.push_back(pData);
m_RequestMutex.Unlock();
++m_iRestSemCount;
if (!ReleaseSemaphore(m_hSemaphore, m_iRestSemCount, NULL))
TraceError("CFileLoaderThread::Request: ReleaseSemaphore error");
--m_iRestSemCount;
}
bool CFileLoaderThread::Fetch(TData ** ppData) // called in main thread
{
m_CompleteMutex.Lock();
if (m_pCompleteDeque.empty()) if (m_pCompleteDeque.empty())
{
m_CompleteMutex.Unlock();
return false; return false;
}
*ppData = m_pCompleteDeque.front(); *ppData = m_pCompleteDeque.front();
m_pCompleteDeque.pop_front(); m_pCompleteDeque.pop_front();
m_CompleteMutex.Unlock();
return true; return true;
} }
void CFileLoaderThread::Process() // called in loader thread void CFileLoaderThread::ProcessFile(const std::string& fileName)
{ {
m_RequestMutex.Lock(); if (m_bShutdowned)
if (m_pRequestDeque.empty())
{
m_RequestMutex.Unlock();
return; return;
}
TData * pData = m_pRequestDeque.front(); TData * pData = new TData;
m_pRequestDeque.pop_front(); pData->File.clear();
pData->stFileName = fileName;
m_RequestMutex.Unlock();
CPackManager::instance().GetFile(pData->stFileName, pData->File); CPackManager::instance().GetFile(pData->stFileName, pData->File);
m_CompleteMutex.Lock(); // Add to completed queue
m_pCompleteDeque.push_back(pData); {
m_CompleteMutex.Unlock(); std::lock_guard<std::mutex> lock(m_CompleteMutex);
m_pCompleteDeque.push_back(pData);
}
Sleep(g_iLoadingDelayTime); Sleep(g_iLoadingDelayTime);
} }

37
src/EterLib/FileLoaderThread.h
View File

@@ -2,8 +2,7 @@
#define __INC_YMIR_ETERLIB_FILELOADERTHREAD_H__ #define __INC_YMIR_ETERLIB_FILELOADERTHREAD_H__
#include <deque> #include <deque>
#include "Thread.h" #include <mutex>
#include "Mutex.h"
#include "PackLib/PackManager.h" #include "PackLib/PackManager.h"
class CFileLoaderThread class CFileLoaderThread
@@ -19,41 +18,19 @@ class CFileLoaderThread
CFileLoaderThread(); CFileLoaderThread();
~CFileLoaderThread(); ~CFileLoaderThread();
int Create(void * arg); bool Create(void * arg);
void Shutdown();
public: public:
void Request(std::string & c_rstFileName); void Request(const std::string& c_rstFileName);
bool Fetch(TData ** ppData); bool Fetch(TData ** ppData);
void Shutdown();
protected:
static UINT CALLBACK EntryPoint(void * pThis);
UINT Run(void * arg);
void * Arg() const { return m_pArg; }
void Arg(void * arg) { m_pArg = arg; }
HANDLE m_hThread;
private: private:
void * m_pArg; void ProcessFile(const std::string& fileName);
unsigned m_uThreadID;
protected:
UINT Setup();
UINT Execute(void * pvArg);
void Destroy();
void Process();
private: private:
std::deque<TData *> m_pRequestDeque; std::deque<TData*> m_pCompleteDeque;
Mutex m_RequestMutex; std::mutex m_CompleteMutex;
std::deque<TData *> m_pCompleteDeque;
Mutex m_CompleteMutex;
HANDLE m_hSemaphore;
int m_iRestSemCount;
bool m_bShutdowned; bool m_bShutdowned;
}; };

270
src/EterLib/FileLoaderThreadPool.cpp
View File

@@ -1,270 +0,0 @@
#include "StdAfx.h"
#include "FileLoaderThreadPool.h"
#include "BufferPool.h"
#include "ImageDecoder.h"
#include "PackLib/PackManager.h"
#include <algorithm>
static const bool USE_STAGED_TEXTURE_LOADING = true;
CFileLoaderThreadPool::CFileLoaderThreadPool()
: m_pCompletedQueue(nullptr)
, m_bShutdown(false)
, m_nextRequestID(0)
, m_activeTasks(0)
, m_threadCount(0)
{
}
CFileLoaderThreadPool::~CFileLoaderThreadPool()
{
Shutdown();
}
bool CFileLoaderThreadPool::Initialize(unsigned int threadCount)
{
if (!m_workers.empty())
{
TraceError("CFileLoaderThreadPool::Initialize: Already initialized");
return false;
}
if (threadCount == 0)
{
threadCount = std::thread::hardware_concurrency();
if (threadCount == 0)
threadCount = 4;
else
threadCount = std::max(4u, threadCount / 2);
}
threadCount = std::max(4u, std::min(16u, threadCount));
m_threadCount = threadCount;
Tracenf("CFileLoaderThreadPool: Initializing with %u worker threads", threadCount);
m_pCompletedQueue = new SPSCQueue<TLoadResult>(COMPLETED_QUEUE_SIZE);
m_workers.reserve(threadCount);
for (unsigned int i = 0; i < threadCount; ++i)
{
TWorkerThread worker;
worker.pRequestQueue = new SPSCQueue<TLoadRequest>(REQUEST_QUEUE_SIZE);
worker.bBusy.store(false, std::memory_order_relaxed);
try
{
worker.thread = std::thread(&CFileLoaderThreadPool::WorkerThreadFunction, this, i);
}
catch (const std::exception& e)
{
TraceError("CFileLoaderThreadPool::Initialize: Failed to create thread %u: %s", i, e.what());
delete worker.pRequestQueue;
worker.pRequestQueue = nullptr;
Shutdown();
return false;
}
m_workers.push_back(std::move(worker));
}
return true;
}
void CFileLoaderThreadPool::Shutdown()
{
if (m_workers.empty())
return;
// Signal shutdown
m_bShutdown.store(true, std::memory_order_release);
// Wait for all workers to finish
for (auto& worker : m_workers)
{
if (worker.thread.joinable())
worker.thread.join();
// Cleanup request queue
if (worker.pRequestQueue)
{
delete worker.pRequestQueue;
worker.pRequestQueue = nullptr;
}
}
m_workers.clear();
// Cleanup completed queue
if (m_pCompletedQueue)
{
delete m_pCompletedQueue;
m_pCompletedQueue = nullptr;
}
m_threadCount = 0;
}
bool CFileLoaderThreadPool::Request(const std::string& fileName)
{
if (m_workers.empty())
{
TraceError("CFileLoaderThreadPool::Request: Thread pool not initialized");
return false;
}
TLoadRequest request;
request.stFileName = fileName;
request.requestID = m_nextRequestID.fetch_add(1, std::memory_order_relaxed);
request.decodeImage = false;
if (USE_STAGED_TEXTURE_LOADING)
{
size_t dotPos = fileName.find_last_of('.');
if (dotPos != std::string::npos && dotPos + 1 < fileName.size())
{
const char* ext = fileName.c_str() + dotPos;
size_t extLen = fileName.size() - dotPos;
if ((extLen == 4 && (_stricmp(ext, ".dds") == 0 || _stricmp(ext, ".png") == 0 ||
_stricmp(ext, ".jpg") == 0 || _stricmp(ext, ".tga") == 0 || _stricmp(ext, ".bmp") == 0)) ||
(extLen == 5 && _stricmp(ext, ".jpeg") == 0))
{
request.decodeImage = true;
}
}
}
unsigned int targetWorker = SelectLeastBusyWorker();
if (!m_workers[targetWorker].pRequestQueue->Push(request))
{
for (unsigned int i = 0; i < m_threadCount; ++i)
{
unsigned int workerIdx = (targetWorker + i) % m_threadCount;
if (m_workers[workerIdx].pRequestQueue->Push(request))
{
m_activeTasks.fetch_add(1, std::memory_order_relaxed);
return true;
}
}
TraceError("CFileLoaderThreadPool::Request: All worker queues full for file: %s", fileName.c_str());
return false;
}
m_activeTasks.fetch_add(1, std::memory_order_relaxed);
return true;
}
bool CFileLoaderThreadPool::Fetch(TLoadResult& result)
{
if (!m_pCompletedQueue)
return false;
if (m_pCompletedQueue->Pop(result))
{
m_activeTasks.fetch_sub(1, std::memory_order_relaxed);
return true;
}
return false;
}
size_t CFileLoaderThreadPool::GetPendingCount() const
{
size_t total = 0;
for (const auto& worker : m_workers)
{
if (worker.pRequestQueue)
total += worker.pRequestQueue->Size();
}
return total;
}
bool CFileLoaderThreadPool::IsIdle() const
{
return m_activeTasks.load(std::memory_order_acquire) == 0;
}
unsigned int CFileLoaderThreadPool::SelectLeastBusyWorker() const
{
unsigned int leastBusyIdx = 0;
size_t minSize = m_workers[0].pRequestQueue->Size();
for (unsigned int i = 1; i < m_threadCount; ++i)
{
size_t queueSize = m_workers[i].pRequestQueue->Size();
if (queueSize < minSize)
{
minSize = queueSize;
leastBusyIdx = i;
}
}
return leastBusyIdx;
}
void CFileLoaderThreadPool::WorkerThreadFunction(unsigned int workerIndex)
{
TWorkerThread& worker = m_workers[workerIndex];
SPSCQueue<TLoadRequest>* pRequestQueue = worker.pRequestQueue;
CBufferPool* pBufferPool = CPackManager::instance().GetBufferPool();
Tracenf("CFileLoaderThreadPool: Worker thread %u started", workerIndex);
int idleCount = 0;
while (!m_bShutdown.load(std::memory_order_acquire))
{
TLoadRequest request;
if (pRequestQueue->Pop(request))
{
idleCount = 0;
worker.bBusy.store(true, std::memory_order_release);
TLoadResult result;
result.stFileName = request.stFileName;
result.requestID = request.requestID;
result.File.clear();
result.hasDecodedImage = false;
CPackManager::instance().GetFileWithPool(request.stFileName, result.File, pBufferPool);
if (request.decodeImage && !result.File.empty())
{
if (CImageDecoder::DecodeImage(result.File.data(), result.File.size(), result.decodedImage))
{
result.hasDecodedImage = true;
result.File.clear();
}
}
while (!m_pCompletedQueue->Push(result))
{
std::this_thread::yield();
if (m_bShutdown.load(std::memory_order_acquire))
break;
}
worker.bBusy.store(false, std::memory_order_release);
}
else
{
idleCount++;
if (idleCount > 1000)
{
Sleep(1);
idleCount = 0;
}
else if (idleCount > 10)
{
std::this_thread::yield();
}
}
}
Tracenf("CFileLoaderThreadPool: Worker thread %u stopped", workerIndex);
}

90
src/EterLib/FileLoaderThreadPool.h
View File

@@ -1,90 +0,0 @@
#ifndef __INC_ETERLIB_FILELOADERTHREADPOOL_H__
#define __INC_ETERLIB_FILELOADERTHREADPOOL_H__
#include <vector>
#include <thread>
#include <atomic>
#include "SPSCQueue.h"
#include "PackLib/PackManager.h"
#include "DecodedImageData.h"
class CFileLoaderThreadPool
{
public:
struct TLoadRequest
{
std::string stFileName;
uint32_t requestID;
bool decodeImage;
};
struct TLoadResult
{
std::string stFileName;
TPackFile File;
uint32_t requestID;
TDecodedImageData decodedImage;
bool hasDecodedImage;
};
public:
CFileLoaderThreadPool();
~CFileLoaderThreadPool();
bool Initialize(unsigned int threadCount = 0);
void Shutdown();
bool Request(const std::string& fileName);
bool Fetch(TLoadResult& result);
size_t GetPendingCount() const;
bool IsIdle() const;
private:
struct TWorkerThread
{
std::thread thread;
SPSCQueue<TLoadRequest>* pRequestQueue;
std::atomic<bool> bBusy;
TWorkerThread() : pRequestQueue(nullptr), bBusy(false) {}
TWorkerThread(TWorkerThread&& other) noexcept
: thread(std::move(other.thread))
, pRequestQueue(other.pRequestQueue)
, bBusy(other.bBusy.load())
{
other.pRequestQueue = nullptr;
}
TWorkerThread& operator=(TWorkerThread&& other) noexcept
{
if (this != &other)
{
thread = std::move(other.thread);
pRequestQueue = other.pRequestQueue;
bBusy.store(other.bBusy.load());
other.pRequestQueue = nullptr;
}
return *this;
}
TWorkerThread(const TWorkerThread&) = delete;
TWorkerThread& operator=(const TWorkerThread&) = delete;
};
void WorkerThreadFunction(unsigned int workerIndex);
unsigned int SelectLeastBusyWorker() const;
private:
std::vector<TWorkerThread> m_workers;
SPSCQueue<TLoadResult>* m_pCompletedQueue;
std::atomic<bool> m_bShutdown;
std::atomic<uint32_t> m_nextRequestID;
std::atomic<int> m_activeTasks; // Fast IsIdle check
unsigned int m_threadCount;
static const size_t REQUEST_QUEUE_SIZE = 16384; // Doubled from 8192
static const size_t COMPLETED_QUEUE_SIZE = 32768; // Doubled from 16384
};
#endif // __INC_ETERLIB_FILELOADERTHREADPOOL_H__

66
src/EterLib/ResourceManager.cpp
View File

@@ -71,14 +71,7 @@ void CResourceManager::ProcessBackgroundLoading()
//printf("REQ %s\n", stFileName.c_str()); //printf("REQ %s\n", stFileName.c_str());
if (m_pLoaderThreadPool) ms_loadingThread.Request(stFileName);
{
m_pLoaderThreadPool->Request(stFileName);
}
else
{
ms_loadingThread.Request(stFileName);
}
m_WaitingMap.insert(TResourceRequestMap::value_type(dwFileCRC, stFileName)); m_WaitingMap.insert(TResourceRequestMap::value_type(dwFileCRC, stFileName));
itor = m_RequestMap.erase(itor); itor = m_RequestMap.erase(itor);
@@ -87,44 +80,7 @@ void CResourceManager::ProcessBackgroundLoading()
DWORD dwCurrentTime = ELTimer_GetMSec(); DWORD dwCurrentTime = ELTimer_GetMSec();
if (m_pLoaderThreadPool) // Process thread results
{
CFileLoaderThreadPool::TLoadResult result;
while (m_pLoaderThreadPool->Fetch(result))
{
CResource * pResource = GetResourcePointer(result.stFileName.c_str());
if (pResource)
{
if (pResource->IsEmpty())
{
if (result.hasDecodedImage)
{
CGraphicImage* pImage = dynamic_cast<CGraphicImage*>(pResource);
if (pImage)
{
pImage->OnLoadFromDecodedData(result.decodedImage);
}
else
{
pResource->OnLoad(result.File.size(), result.File.data());
}
}
else
{
pResource->OnLoad(result.File.size(), result.File.data());
}
pResource->AddReferenceOnly();
m_pResRefDecreaseWaitingMap.insert(TResourceRefDecreaseWaitingMap::value_type(dwCurrentTime, pResource));
}
}
m_WaitingMap.erase(GetCRC32(result.stFileName.c_str(), result.stFileName.size()));
}
}
// Process old thread results
CFileLoaderThread::TData * pData; CFileLoaderThread::TData * pData;
while (ms_loadingThread.Fetch(&pData)) while (ms_loadingThread.Fetch(&pData))
{ {
@@ -580,19 +536,9 @@ void CResourceManager::ReserveDeletingResource(CResource * pResource)
} }
CResourceManager::CResourceManager() CResourceManager::CResourceManager()
: m_pLoaderThreadPool(nullptr) : m_pTextureCache(nullptr)
, m_pTextureCache(nullptr)
{ {
ms_loadingThread.Create(0); ms_loadingThread.Create(0);
m_pLoaderThreadPool = new CFileLoaderThreadPool();
if (!m_pLoaderThreadPool->Initialize())
{
TraceError("CResourceManager: Failed to initialize FileLoaderThreadPool");
delete m_pLoaderThreadPool;
m_pLoaderThreadPool = nullptr;
}
m_pTextureCache = new CTextureCache(512); m_pTextureCache = new CTextureCache(512);
} }
@@ -601,12 +547,6 @@ CResourceManager::~CResourceManager()
Destroy(); Destroy();
ms_loadingThread.Shutdown(); ms_loadingThread.Shutdown();
if (m_pLoaderThreadPool)
{
delete m_pLoaderThreadPool;
m_pLoaderThreadPool = nullptr;
}
if (m_pTextureCache) if (m_pTextureCache)
{ {
delete m_pTextureCache; delete m_pTextureCache;

3
src/EterLib/ResourceManager.h
View File

@@ -2,7 +2,6 @@
#include "Resource.h" #include "Resource.h"
#include "FileLoaderThread.h" #include "FileLoaderThread.h"
#include "FileLoaderThreadPool.h"
#include <set> #include <set>
#include <map> #include <map>
@@ -47,7 +46,6 @@ class CResourceManager : public CSingleton<CResourceManager>
void PushBackgroundLoadingSet(std::set<std::string> & LoadingSet); void PushBackgroundLoadingSet(std::set<std::string> & LoadingSet);
CTextureCache* GetTextureCache() { return m_pTextureCache; } CTextureCache* GetTextureCache() { return m_pTextureCache; }
CFileLoaderThreadPool* GetLoaderThreadPool() { return m_pLoaderThreadPool; }
protected: protected:
void __DestroyDeletingResourceMap(); void __DestroyDeletingResourceMap();
@@ -75,7 +73,6 @@ class CResourceManager : public CSingleton<CResourceManager>
TResourceRefDecreaseWaitingMap m_pResRefDecreaseWaitingMap; TResourceRefDecreaseWaitingMap m_pResRefDecreaseWaitingMap;
static CFileLoaderThread ms_loadingThread; static CFileLoaderThread ms_loadingThread;
CFileLoaderThreadPool* m_pLoaderThreadPool;
CTextureCache* m_pTextureCache; CTextureCache* m_pTextureCache;
mutable std::mutex m_ResourceMapMutex; // Thread-safe resource map access mutable std::mutex m_ResourceMapMutex; // Thread-safe resource map access