比特幣源碼分析：任務調度器的使用

任務調度器

Bitcoin 進程啟動後，有一個專門的線程做任務調度， 這些任務根據指定的時刻，執行對應的函數:

bool AppInitMain() { ....... // Start the lightweight task scheduler thread CScheduler::Function serviceLoop = boost::bind(&CScheduler::serviceQueue, &scheduler); threadGroup.create_thread(boost::bind(&TraceThread, "scheduler", serviceLoop)); ....... }

調度器類主要是實現了一個生產者消費者的任務隊列，只是這個任務隊列是用 std::multimap 實現的，map 的key表達某一時刻，map的值表達：那一時刻要執行的函數，內部使用條件變數和鎖來保護multimap ，還有幾個bool 條件:

class CScheduler { public: CScheduler(); ~CScheduler(); typedef std::functionFunction; void schedule(Function f, boost::chrono::system_clock::time_point t=boost::chrono::system_clock::now()); void scheduleFromNow(Function f, int64_t deltaMilliSeconds); void scheduleEvery(Function f, int64_t deltaMilliSeconds); void serviceQueue(); void stop(bool drain=false); size_t getQueueInfo(boost::chrono::system_clock::time_point &first, boost::chrono::system_clock::time_point &last) const; bool AreThreadsServicingQueue() const; private: std::multimaptaskQueue; boost::condition_variable newTaskScheduled; mutable boost::mutex newTaskMutex; int nThreadsServicingQueue; bool stopRequested; bool stopWhenEmpty; bool shouldStop() const { return stopRequested || (stopWhenEmpty && taskQueue.empty()); } };

CScheduler的client 通過調用schedule 往內部multimap添加一個條目; scheduleFromNow 和scheduleEvery 內部都是調用schedule 方法實現; 這三個方法屬於生產者要生產任務的方法， 任務的消費者調用serviceQueue等待取走任務， 然後執行。

目前整個程序有一個全局的CScheduler實例:

static CScheduler scheduler; 這個實例對應只有一個消費者線程， 即唯一的後台調度器線程。

class SingleThreadedSchedulerClient 主要用途是，藉助CScheduler類型，保障被添加到內部鏈表的任務，被串列執行:

class SingleThreadedSchedulerClient { private: CScheduler *m_pscheduler; CCriticalSection m_cs_callbacks_pending; std::listm_callbacks_pending; bool m_are_callbacks_running = false; void MaybeScheduleProcessQueue(); void ProcessQueue(); public: explicit SingleThreadedSchedulerClient(CScheduler *pschedulerIn) : m_pscheduler(pschedulerIn) {} void AddToProcessQueue(std::function func); void EmptyQueue(); size_t CallbacksPending(); };

使用例子

基本的使用例子:

#include #includeind.hpp> #includehread.hpp> #includeest/unit_test.hpp> #include static void doN(){ std::coutre>

進程啟動後， 全局對象連接管理器connman初始化後， connman 的Start 方法最後，通過scheduler 線程安排了一個定時任務: 每隔15分鐘， 把connman 對象內部成員，banmap_t 類型的 setBanned, CAddrMan 類型的addrman 序列化到本地文件banlist.dat 和 peers.dat。

//init.cpp if (!connman.Start(scheduler, connOptions)) { return false; } //net.cpp bool CConnman::Start(CScheduler& scheduler, const Options& connOptions) { ............... scheduler.scheduleEvery(std::bind(&CConnman::DumpData, this), DUMP_ADDRESSES_INTERVAL * 1000); }

如果錢包功能編譯使能, 會讓scheduler 線程安排每隔500毫秒刷新錢包狀態。

//init.cpp #ifdef ENABLE_WALLET StartWallets(scheduler); #endif //wallet/init.cpp void StartWallets(CScheduler& scheduler) { for (CWalletRef pwallet : vpwallets) { pwallet->postInitProcess(scheduler); } } //wallet/wallet.cpp void CWallet::postInitProcess(CScheduler& scheduler) { ReacceptWalletTransactions(); if (!CWallet::fFlushScheduled.exchange(true)) { scheduler.scheduleEvery(MaybeCompactWalletDB, 500); } }

PeerLogicValidation 對象的構造函數內部, scheduler 線程安排每45秒執行CheckForStaleTipAndEvictPeer函數主要做兩件事:

關掉多餘的外出tcp 連接

根據當前時間，檢查當前節點的blockchain 的tip 是否有可能過時了，建立額外的連接同步跟上

PeerLogicValidation::PeerLogicValidation(CConnman* connmanIn, CScheduler &scheduler) : connman(connmanIn), m_stale_tip_check_time(0) { // Initialize global variables that cannot be constructed at startup. recentRejects.reset(new CRollingBloomFilter(120000, 0.000001)); const Consensus::Params& consensusParams = Params().GetConsensus(); // Stale tip checking and peer eviction are on two different timers, but we // don"t want them to get out of sync due to drift in the scheduler, so we // combine them in one function and schedule at the quicker (peer-eviction) // timer. static_assert(EXTRA_PEER_CHECK_INTERVALm_stale_tip_check_time) { LOCK(cs_main); // Check whether our tip is stale, and if so, allow using an extra // outbound peer if (TipMayBeStale(consensusParams)) { LogPrintf("Potential stale tip detected, will try using extra outbound peer (last tip update: %d seconds ago)
", time_in_seconds - g_last_tip_update); connman->SetTryNewOutboundPeer(true); } else if (connman->GetTryNewOutboundPeer()) { connman->SetTryNewOutboundPeer(false); } m_stale_tip_check_time = time_in_seconds + STALE_CHECK_INTERVAL; } }

以上就是bitoin 裡面CScheduler類的主要使用場景。

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！