123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223 |
- #ifndef __CCSYNC_TASK_POOL_H_
- #define __CCSYNC_TASK_POOL_H_
- #include "platform/CCPlatformMacros.h"
- #include "base/CCDirector.h"
- #include "base/CCScheduler.h"
- #include <vector>
- #include <queue>
- #include <memory>
- #include <thread>
- #include <mutex>
- #include <condition_variable>
- #include <future>
- #include <functional>
- #include <stdexcept>
- NS_CC_BEGIN
- class CC_DLL AsyncTaskPool
- {
- public:
- typedef std::function<void(void*)> TaskCallBack;
-
- enum class TaskType
- {
- TASK_IO,
- TASK_NETWORK,
- TASK_OTHER,
- TASK_MAX_TYPE,
- };
-
- static AsyncTaskPool* getInstance();
-
- static void destroyInstance();
-
-
- CC_DEPRECATED_ATTRIBUTE static void destoryInstance() { return destroyInstance(); }
-
-
- void stopTasks(TaskType type);
-
-
- template<class F>
- inline void enqueue(TaskType type, const TaskCallBack& callback, void* callbackParam, F&& f);
-
- CC_CONSTRUCTOR_ACCESS:
- AsyncTaskPool();
- ~AsyncTaskPool();
-
- protected:
-
-
- class ThreadTasks {
- struct AsyncTaskCallBack
- {
- TaskCallBack callback;
- void* callbackParam;
- };
- public:
- ThreadTasks()
- : _stop(false)
- {
- _thread = std::thread(
- [this]
- {
- for(;;)
- {
- std::function<void()> task;
- AsyncTaskCallBack callback;
- {
- std::unique_lock<std::mutex> lock(this->_queueMutex);
- this->_condition.wait(lock,
- [this]{ return this->_stop || !this->_tasks.empty(); });
- if(this->_stop && this->_tasks.empty())
- return;
- task = std::move(this->_tasks.front());
- callback = std::move(this->_taskCallBacks.front());
- this->_tasks.pop();
- this->_taskCallBacks.pop();
- }
-
- task();
- Director::getInstance()->getScheduler()->performFunctionInCocosThread([&, callback]{ callback.callback(callback.callbackParam); });
- }
- }
- );
- }
- ~ThreadTasks()
- {
- {
- std::unique_lock<std::mutex> lock(_queueMutex);
- _stop = true;
-
- while(_tasks.size())
- _tasks.pop();
- while (_taskCallBacks.size())
- _taskCallBacks.pop();
- }
- _condition.notify_all();
- _thread.join();
- }
- void clear()
- {
- std::unique_lock<std::mutex> lock(_queueMutex);
- while(_tasks.size())
- _tasks.pop();
- while (_taskCallBacks.size())
- _taskCallBacks.pop();
- }
- template<class F>
- void enqueue(const TaskCallBack& callback, void* callbackParam, F&& f)
- {
- auto task = f;
-
- {
- std::unique_lock<std::mutex> lock(_queueMutex);
-
-
- if(_stop)
- {
- CC_ASSERT(0 && "already stop");
- return;
- }
-
- AsyncTaskCallBack taskCallBack;
- taskCallBack.callback = callback;
- taskCallBack.callbackParam = callbackParam;
- _tasks.emplace([task](){ task(); });
- _taskCallBacks.emplace(taskCallBack);
- }
- _condition.notify_one();
- }
- private:
-
-
- std::thread _thread;
-
- std::queue< std::function<void()> > _tasks;
- std::queue<AsyncTaskCallBack> _taskCallBacks;
-
-
- std::mutex _queueMutex;
- std::condition_variable _condition;
- bool _stop;
- };
-
-
- ThreadTasks _threadTasks[int(TaskType::TASK_MAX_TYPE)];
-
- static AsyncTaskPool* s_asyncTaskPool;
- };
- inline void AsyncTaskPool::stopTasks(TaskType type)
- {
- auto& threadTask = _threadTasks[(int)type];
- threadTask.clear();
- }
- template<class F>
- inline void AsyncTaskPool::enqueue(AsyncTaskPool::TaskType type, const TaskCallBack& callback, void* callbackParam, F&& f)
- {
- auto& threadTask = _threadTasks[(int)type];
-
- threadTask.enqueue(callback, callbackParam, f);
- }
- NS_CC_END
- #endif
|