AsyncLoader.h

#pragma once
#include "IAttributeFields.h"
#include "IDataLoader.h"
#include "util/LogService.h"
#include <vector>
#include <set>
#include <boost/thread.hpp>
#include <boost/asio.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/enable_shared_from_this.hpp>

#define E_TRYAGAIN  -123456

#define MAX_PROCESS_QUEUE 16

namespace ParaEngine
{
    class CDirectXEngine;
    class CGDIEngine;
    //--------------------------------------------------------------------------------------
    // Structures
    //--------------------------------------------------------------------------------------
    
    class CAsyncLoader : public IAttributeFields
    {
    public:
        enum AssetLogLevelEnum
        {
            Log_All = 0,
            Log_Debug,
            Log_Remote,
            Log_Warn,
            Log_Error,
        };

        typedef boost::shared_ptr<boost::thread> Boost_Thread_ptr_type;

        CAsyncLoader();
        virtual ~CAsyncLoader();

        ATTRIBUTE_DEFINE_CLASS(CAsyncLoader);

        virtual int InstallFields(CAttributeClass* pClass, bool bOverride);

        ATTRIBUTE_METHOD1(CAsyncLoader, GetEstimatedSizeInBytes_s, int*) { *p1 = cls->GetEstimatedSizeInBytes(); return S_OK; }
        ATTRIBUTE_METHOD1(CAsyncLoader, GetItemsLeft_s, int*) { *p1 = cls->GetItemsLeft(); return S_OK; }
        ATTRIBUTE_METHOD1(CAsyncLoader, GetBytesProcessed_s, int*) { *p1 = cls->GetBytesProcessed(); return S_OK; }
        
        ATTRIBUTE_METHOD1(CAsyncLoader, SetWorkerThreads_s, Vector2) { cls->CreateWorkerThreads((int)p1.x, (int)p1.y); return S_OK; }
        
        ATTRIBUTE_METHOD1(CAsyncLoader, SetProcessorQueueSize_s, Vector2) { cls->SetProcessorQueueSize((int)p1.x, (int)p1.y); return S_OK; }

        ATTRIBUTE_METHOD1(CAsyncLoader, GetLogLevel_s, int*) { *p1 = cls->GetLogLevel(); return S_OK; }
        ATTRIBUTE_METHOD1(CAsyncLoader, SetLogLevel_s, int) { cls->SetLogLevel(p1); return S_OK; }

        ATTRIBUTE_METHOD1(CAsyncLoader, log_s, const char*) { cls->log(p1); return S_OK; }
        ATTRIBUTE_METHOD(CAsyncLoader, WaitForAllItems_s) { cls->WaitForAllItems(); return S_OK; }
        
    private:
        struct ProcessorWorkerThread;
    public:
        static CAsyncLoader& GetSingleton();

        void CleanUp();

        int Start(int nWorkerCount = 5);

        int Stop();

        int AddWorkItem( IDataLoader* pDataLoader, IDataProcessor* pDataProcessor, HRESULT* pHResult, void** ppDeviceObject, int nProcessorThreadID=0);

        int AddWorkItem( ResourceRequest_ptr& request );

        HRESULT RunWorkItem( IDataLoader* pDataLoader, IDataProcessor* pDataProcessor, HRESULT* pHResult, void** ppDeviceObject);

        HRESULT RunWorkItem( ResourceRequest_ptr& request );
        
        void ProcessDeviceWorkItems( int CurrentNumResourcesToService=100, bool bRetryLoads = false );
        /* this function is usually called by the render thread, but it may also be called in IO or worker thread is IsDeviceObject() is false. */
        void ProcessDeviceWorkItemImp(ResourceRequest_ptr& pResourceRequest, bool bRetryLoads = false);

        bool CreateWorkerThreads(int nProcessorQueueID, int nMaxCount);
        int GetWorkerThreadsCount(int nProcessorQueueID);
        void SetProcessorQueueSize(int nProcessorQueueID, int nSize);
        int GetProcessorQueueSize(int nProcessorQueueID);

        void WaitForAllItems();

#ifdef PARAENGINE_CLIENT

        inline LPD3DXFILE GetFileParser() { return m_pXFileParser; };

        CDirectXEngine* GetEngine();

        CGDIEngine* GetGDIEngine();
#endif

        void AddPendingRequest(const char* sURL);

        bool HasPendingRequest(const char* sURL);

        void RemovePendingRequest(const char* sURL);

        void ClearAllPendingRequests();

        int GetLogLevel() const;
        void SetLogLevel(int val);

        void log(const string& msg);
        void log(int nLogLevel, const string& msg);

        inline bool interruption_requested(){return m_bInterruptSignal;}

        void Interrupt();

        int GetItemsLeft(int nItemType = -1);

        int GetEstimatedSizeInBytes();

        int GetBytesProcessed(int nItemType = -1);

    protected:

        int FileIOThreadProc();
        int FileIOThreadProc_HandleRequest(ResourceRequest_ptr& ResourceRequest);

        int ProcessingThreadProc(ProcessorWorkerThread* pThreadData);
        
    private:
        
        bool m_bDone;
        bool m_bProcessThreadDone;
        bool m_bIOThreadDone;

        volatile bool m_bInterruptSignal;

        int m_NumResourcesToService;
        int m_NumOutstandingResources;
        int m_nRemainingBytes;

        CResourceRequestQueue m_IOQueue;
        CResourceRequestQueue m_RenderThreadQueue;
        CResourceRequestQueue m_ProcessQueues[MAX_PROCESS_QUEUE];


        Boost_Thread_ptr_type m_io_thread;
        

        struct ProcessorWorkerThread : public IProcessorWorkerData
        {
        public:
            ProcessorWorkerThread();
            ProcessorWorkerThread(int nQueueID);

            ~ProcessorWorkerThread();
        public:
            void reset(boost::thread * pThread) {m_thread.reset(pThread);}
            void reset() {m_thread.reset();}
            void join() { if(m_thread.get()) m_thread->join();}

            bool timed_join(int nSeconds);
            
            virtual int GetProcessorQueueID() {return m_nQueueID;} 

            virtual void* GetCurlInterface(int nID = 0);

            virtual void AddBytesProcessed(int nBytesProcessed);

            virtual int GetBytesProcessed();
        public:
            //  thread ptr. 
            Boost_Thread_ptr_type m_thread;

            //  which queue to process. 
            int m_nQueueID;

            void* m_curl;

            // bytes processed
            volatile int m_nBytesProcessed;
        };

        struct DefaultWorkerThreadData  : public ProcessorWorkerThread
        {
        public:
            virtual void* GetCurlInterface(int nID = 0);
        };

        std::vector< ProcessorWorkerThread* > m_workers;

        DefaultWorkerThreadData* m_default_processor_worker_data;

        ParaEngine::mutex m_default_processor_mutex;

        CServiceLogger_ptr g_asset_logger;
        
        AssetLogLevelEnum m_nLogLevel;

#ifdef PARAENGINE_CLIENT

        LPD3DXFILE m_pXFileParser;

        CDirectXEngine* m_pEngine;

        CGDIEngine* m_pGDIEngine;
#endif
        // only for pending request. 
        ParaEngine::mutex m_pending_request_mutex;
        // for statistics of request
        ParaEngine::mutex m_request_stats;

        // all pending url, this could prevent the same url to be request multiple times. 
        std::set <std::string> m_pending_requests;
    };
}