一種基于集群的通用并行計算框架設計

王寧

（四川大學計算機學院，成都610065)

一種基于集群的通用并行計算框架設計

王寧

（四川大學計算機學院，成都610065)

近年來各領域應用的數據量和計算量需求都大幅增加，傳統單個計算設備往往無法勝任如此規模的計算量，因此越來越多的領域開始嘗試使用并行計算技術，分布式并行計算是進行并行計算的一種主要方式，常見的框架為基于MapReduce的Hadoop。提出一種基于集群的通用并行計算框架，參考“管道過濾器”模式，對三個模塊“任務劃分”、“控制器節點”和“計算節點”都進行詳細設計描述，相對于Hadoop，對有向無環圖型任務由更好支持，并且支持迭代型任務，另外增加緩存機制，減少系統耗時，一定程度支持實時性應用。

并行計算；集群；系統框架；有向無環圖；緩存

0 引言

并行計算[1]（Parallel Computing）是指同時使用多種計算資源解決計算問題的過程，是提高計算機系統計算速度和處理能力的一種有效手段。它的基本思想是用多個處理器來協同求解同一問題，即將被求解的問題分解成若干個部分，各部分均由一個獨立的處理機來并行計算。并行計算系統既可以是專門設計的、含有多個處理器的超級計算機，也可以是以某種方式互連的若干臺的獨立計算機構成的集群。通過并行計算集群完成數據的處理。

目前應用較為廣泛的并行計算模型為Jeffrey Dean等提出的MapReduce[2]，MapReduce的基本思想是將所有任務的執行看做兩個操作，分別是Map（映射）和Reduce（化簡），首先，Map會先對由很多獨立元素組成的邏輯列表中的每一個元素進行指定的操作，且原始列表不會被更改，會創建多個新的列表來保存Map的處理結果。也就意味著，Map操作是高度并行的。當Map工作完成之后，系統會接著對新生成的多個列表進行清理(Shuffle)和排序，之后，會這些新創建的列表進行Reduce操作，也就是對一個列表中的元素根據Key值進行適當的合并。

Hadoop作為最早的基于MapReduce的并行計算框架之一，目前得到了十分廣泛的使用，其以MapReduce為計算核心，添加了任務分配、負載均衡、網絡傳輸等模塊。Hadoop的優點在于對大數據問題的處理很方便，并且系統具有高可擴展性，即很容易將新的計算資源加入已有系統中，Hadoop對用戶隱藏了底層任務調度、負載均衡、網絡傳輸等細節，使用戶只需專心于MapReduce模塊的制定。

如果用戶定義了多個Job，并指定了它們之間的先后關系，則多個Job會依次按照MapReduce的方式進行處理，最終用戶需要的結果就存儲在最后一個Reduce節點上，整個的任務調度、網絡通信、數據存儲、負載平衡等工作都是MapReduce框架底層完成的，不需要用戶關心。

MapReduce框架的優點在于對大數據問題處理很方便，并且具有很高的可擴展性，即很容易將新的計算節點加入到已有的系統中。另外MapReduce和Hadoop也有一些很明顯的缺點：

①所有Reduce任務必須等前一步的Map任務全部完成才可以執行，這樣會大大降低可并行度；

②對DAG（有向無環圖）類型的任務支持不足，Hadoop雖然可以用多個Job來模擬出DAG圖，但是Job間的依賴需要開發者分別管理維護，并且不同層次的任務不能并行執行；

③很難支持迭代類型任務，迭代類型的任務通常無法預知迭代次數，所以無法預先生成定量的Job；

④MapReduce強制定義了Map和Reduce兩個階段，但有時用戶并不需要兩個階段的處理；

⑤無法支持實時性應用，Reduce操作生成的數據，會被HDFS存入硬盤中，由于沒有相應的緩存機制，所以存儲耗時導致系統時延過高，進而無法支持實時性應用。

本文提出的基于集群的并行計算框架在參考了MapReduce模型和Hadoop框架的基礎上，對于以上5點問題均得到了一定程度的解決，下面幾個小節將分別對此框架的整體架構和各模塊做相應說明。

1 整體架構描述

系統整體架構如圖1所示，整個框架大體分為TaskSplitter（任務分割）、Master（主控節點）和ComputeNode（計算節點）三部分。TaskSplitter部分負責任務劃分，本文設計了一種腳本語言，用戶使用腳本描述自己的應用，TaskSplitter會根據腳本自動切分任務，并生成任務間的依賴關系，很容易構建DAG應用，隨后將任務填充到TaskManager中。Master部分主要負責任務調度，分別從TaskManager和ComputeNodeManager獲取任務信息和計算節點信息后，再由Schedule負責調度分配，最終Master下發消息給ComputeNode。ComputeNode部分主要負責接收Master發送過來的消息，執行具體的計算任務，并且對計算產生的數據進行存儲。

整個架構參考了“Pipe-Filter”架構模式[4]，”Pipe-Filter”總體思想如圖1所示，將系統看做一系列對原始數據的加工動作，首先將原始數據經過一次處理（即Filter一次），加工后的數據放到管道（Pipe）中，然后等待下一個Filter繼續對數據做加工處理，再放到一個Pipe里，如此持續進行下去直到所有Filter都完成了數據加工，那么最終數據就保存在最后一個Pipe里。

對于并行框架來說，所有的任務都可以抽象成先從某處取得對輸入數據，做一定的處理后輸出新的數據，我們很自然而然的用Filter類來表示一次計算，為了充分利用起計算資源，使用ComputeNode這樣一個邏輯上的節點來管理多個Filter，每個ComputeNode本身也是Pipe（這樣設計的原因后面再做詳述），Master則做為一個創建Filter、將Filter和Pipe連接起來的角色，Master和ComputeNode在邏輯上是一對多的，在這樣的設計下，整個架構流程為：首先TaskSplitter劃分得到任務（與此同時各個ComputeNode節點會通過線程自動向Master注冊信息以填充ComputeNodeManager），然后Master由調度模塊生成調度信息（調度信息包括計算任務信息和數據Pipe信息），發送給Com-puteNode，ComputeNode解析這些信息生成相應Filter去執行計算任務，計算完成后，由Pipe存儲數據，然后將任務完成的消息返回給Master，Master更新相應任務狀態后繼續執行下一次的任務調度，如此往復直到所有任務都執行完畢。

圖1 系統整體架構

圖2 Pipe-Filter

2 TaskSplitter部分

TaskSplitter部分的目的是給用戶提供一種方式（接口、腳本等）用以描述用戶的應用需求，根據用戶的描述進行分析，生成任務為后面的調度做好準備。因為此部分會直接和用戶交互，因此可理解性和可擴展性很重要，要盡可能讓用戶方便地去描述多類型的需求，為達到此目的，本文設計了一種簡便的腳本語言，腳本應用舉例如下：

以上所示是一個對圖片序列進行特征檢測的應用，腳本共包含四部分，INPUT：預定義一些變量，用以協助控制任務流程，支持INT、STR、BOOL、DOUBLE類型；KERNEL：預聲明計算處理功能，對應Filter的種類，用來創建不同Filter，CPU和MEMORY表示任務代價（1 CPU表示5%，1 MEMPRY表示100M）；DATA：預定義變量，可以是數組；PROCEDURE：描述應用流程，整個應用是從上到下串行處理的，對于同一層次可并行任務，使用FOR或WHILE循環來描述。根據此腳本，TaskSplitter會自動生成任務并設置好它們之間的依賴關系。

此腳本解決了2個問題：

①描述DAG任務很方便，用戶只需按照實際應用思路編寫PROCEDURE，系統會自動生成滿足依賴關系的任務列表，并且不同層次的子任務只要沒有前置依賴項便可以并行執行；

②支持迭代式任務，用戶使用BOOL類型變量和WHILE循環便可以描述迭代式任務，BOOL變量的值可以在用戶自定義的Filter中更改。

在進行解析時，本文使用一個語句池（Statement-Pool）來存放解析得到的單條語句，TaskSplitter不斷從語句池中取得語句去生成新的任務（Task）并更新與其他任務間的依賴關系，生成的任務被不斷填充到任務管理器中（TaskManager）。使用語句池的目的在于，避免TaskSplitter解析大量循環時導致Master阻塞，Master沒有必要等待全部任務解析完成，任務解析和任務分配可以同時進行。

3 Master部分

Master負責任務管理、任務調度、計算節點管理等工作。Master的設計使用了組合模式，任務管理功能由TaskManager完成，計算節點管理由ComputeNodeManager完成，任務調度由Schedule完成，而Master本身只負責協調各方工作，這樣可以避免Master變得臃腫，并且各組件可以二次開發，提高了系統可擴展性，用戶甚至可以自行定制各組件以滿足特殊需求。以下對消息機制、計算節點管理、任務調度和容錯機制做進一步的介紹。

3.1 消息機制

網絡中所有傳輸的數據都以Packet為基類，由Packet派生得到Message類，作為各種消息的基類，這些消息按照Master To ComputeNode、ComputeNode To Master和ComputeNode To ComputeNode分類如圖3所示：

圖3 消息類型

Master和ComputeNode均使用MessagePool來做消息緩沖池，收到的消息先壓入消息池，然后每次從中取一條消息進行處理，在消息處理方面使用了命令模式，即為每類消息創建一個Handle類，在處理消息時，根據消息類型由工廠模式[5]創建相應的MessageHan dle，調用execute方法執行處理，這樣方便對消息類型進行擴展。

3.2 計算節點管理

要對計算節點（ComputeNode）進行管理，首先要獲得計算節點的信息，而計算節點信息的獲取又分為兩個階段，第一個階段是初始化，當一個計算節點啟動時，會自動開啟一個線程，向配置文件里配置的Master的IP和PORT發送注冊請求，Master收到注冊請求后便將此節點信息添加到ComputeNodeManager中，計算節點只有當收到Master的SimpleMessage::RegisterConfirmed消息后，才會停止注冊請求；第二個階段是更新信息，當注冊成功后，計算節點會開啟一個Keep-Alive線程，每隔固定時間K便向Master發送NodeStatusMessage消息更新ComputeNodeManager中相應的計算節點信息，需注意K設置的太大會使任務調度時信息不準確，K設置過小又會使得此線程開銷過大影響系統效率。

3.3 任務調度

任務調度有三個關鍵問題，分別是調度對象，調度時機和調度策略。

對于本系統，調度對象指的是任務和計算節點，根據前面的描述，我們知道當執行任務調度時，TaskManger和ComputeNodeManager中已有了用來調度的任務和計算資源。

調度時機有兩個，首先當有新的計算節點注冊時，在執行完注冊相關操作后會進行任務調度，其次當Master收到TaskStatusMessage后，會接著進行一次任務調度，因為Task的執行狀態只有兩種，分部是成功和失敗，執行成功意味著有計算資源處于空閑，可以進行新的任務分配，而執行失敗表示有空閑的任務可被調度。

在進行調度時，Schedule首先從TaskManager和ComputeNodeManager獲取所有可分配的任務和計算資源，先對可用計算節點分別按CPU和內存排序，然后對每一個任務，若CPU權值＞=內存權值，則優先選CPU最大的節點分配此任務，否則優先按內存大小尋找節點，然后此節點減去消耗值，將此任務加入此節點的待分配隊列，然后對下一個任務繼續如此分配，直到所有待分配任務分配完成或所有節點不滿足任何一個任務的分配，最后下發消息將待分配隊列的內容分別發送給各個計算節點。

3.4 容錯機制

容錯機制指的是當計算任務出現問題，或者計算節點出現問題時整個系統的應對機制。首先當任務執行錯誤時，會由ComputeNode主動向Master報告此錯誤，然后Master會回滾此任務狀態重新分配執行；而當計算節點因為斷電、機械故障等原因無法和整個系統通信時，Master若超過3個Keep-Alive線程周期都沒有收到某ComputeNode的節點狀態信息，則認為此節點故障，此時Master會先檢查任務列表，查看此節點上哪些已完成任務還有后置任務（即有其余的尚未完成的任務依賴于此節點上已完成的任務），將這些任務和上一次分配分配給此節點的任務一起重新調度執行，這樣便可使得整個系統的任務依賴和數據依賴重新修復，但會因為任務重做造成一定的時間損耗。

4 ComputeNode部分

ComputeNode部分主要是完成Master分配的計算任務，任務完成或失敗后返回消息給Master，并且通過Pipe對計算得到的數據進行存儲，接收到其他節點的數據請求時發送數據給對方。此部分的消息處理機制

和Master部分是一樣的。當收到Master發來的Filter-Message時，便對任務隊列的每一個任務單獨開啟一個線程，由工廠創建對應的Filter執行計算任務（如果所需數據不在本地，需要進行數據請求），執行完成后，通過Pipe進行數據存儲。下面對數據存儲和請求數據兩部分做進一步說明。

4.1 數據請求

FilterMessage消息攜帶了每個任務所需數據存放的節點IP、名稱等信息，當Filter執行時，對每一個所需參數先檢查數據IP是否和本地IP一致，若一致則檢查下一個，若不一致則向此IP發送數據請求消息（RequsetDataMessage），目的ComputeNode接收此消息后，通過Pipe查找到數據，發送給請求方的PipePool（使用Pool是為了避免多個數據傳輸阻塞），請求方獲取到數據后，再對下一個參數做同樣檢查或請求，直到所有參數都準備就緒，便開始執行計算。

4.2 數據存儲

當Filter執行完計算任務后，需要通過Pipe將數據存儲下來，Pipe本身不執行存儲操作，它只負責對數據名稱、路徑等信息做記錄，真正的存儲操作交給更底層的Storage處理，本文在設計Storage時使用了緩存技術，即以一定大小的內存來緩存數據對象，避免寫到硬盤帶來的序列化和逆序列化時間消耗，使用的內存大小由配置文件進行設置。內存緩存策略為：新生成的數據會優先存到內存中，如果可用內存超過了閾值的話，就將一部分數據置換到硬盤上，置換算法使用LRU[6]算法，即存儲中的每個數據對應一個int，剛存入時置為0，每次存儲新數據時，以前的都加1，被訪問的數據重新置0，這樣當置換時，優先置換int值最大的數據。緩存技術加上底層Infiniband高速網絡[7-8]使得本文并框架對實時性應用也有很好的支持。

5 實驗結果

本文通過一個實際的應用來驗證并行計算框架的實用性和有效性，選取的應用是“全景圖拼接”[9]，其處理流程如圖4所示，包含多個步驟，該應用是一個典型的DAG型應用，如圖5所示，有的步驟可以切分成多個子任務，有的步驟又需要全局的數據，各步驟之間的任務存在較復雜的依賴關系，可以全面地檢驗本文并行系統的有效性。

5.1 實驗平臺

實驗環境中，每臺計算機配置相同，配置均為：處理器：Intel Xeon CPU E3-1230 v3@3.30GHz四核；內存：16.0GB；操作系統：Windows 7(64位)。

5.2 結果分析

為了證明本文并行計算框架的有效性，分別對不同計算節點數量、不同規模輸入數據情況下的計算耗時做測試，各情況下耗時如表1所示。

可以看出，在數據量較小時，任務并行計算帶來的效率提升被數據傳輸的損耗所抵消，在2臺計算節點、108張輸入圖片的情況下，多機并行的執行速度已經超過了單機串行的速度，并且隨著計算節點、輸入數據規模的增大，效率提升越來越明顯。當計算節點為4個，輸入圖像數量為288張時，并行框架效率比串行時高了一倍。

圖4 景圖拼接流程

圖5 全景圖拼接子任務依賴

表1 不同輸入圖像（張）和不同計算節點（個）下運算耗時

通過此實驗可以驗證本文并行框架的有效性，但是效率提升并未達到1:1（即效率提升倍數等于同等配置計算節點個數），分析其原因主要有以下3點：（1）并行框架底層的消息傳輸、數據傳輸以及任務分配算法都會有一定耗時；（2）對于DAG型應用，各任務間有依賴，有時需要等待任務同步；（3）全景圖拼接應用中有一些“瓶頸”任務（例如相機標定），這類任務依賴于上一步的所有數據，因此無法進行并行。

6 結語

本文提出一種基于集群的通用并行計算框架，參考“管道過濾器”模型，由TaskSplitter解析自定義的任務描述腳本，自動劃分子任務和生成依賴關系，由Master進行計算資源管理和任務調度，由ComputeNode完成任務計算和數據存儲?？蚣苤С秩魏蜠AG型應用，并且支持迭代式應用和實時性應用，另外對于ComputeNode具有一定程度容錯備災能力。

本文框架目前無法解決Master節點單點故障，下一步計劃采用分布式共享存儲系統[9]重新對Master信息和計算數據做冗余備災。

[1]王磊.并行計算技術綜述[J].信息技術,2012,10.

[2]J.Dean,S.Ghemawat.MapReduce:Simplified Data Processing on Large Clusters[J].Communications of the ACM-50th Anniversary Issue:1958-2008.2008.51(1):107-113.

[3]T.White.Hadoop:The Definitive Guide,Third Edtion[M].United States of America.O'Reilly Media,Inc.2012:3-12.

[4]V.Ambriola.G.Tortora.Advances in Software Engineering and Knowledge Engineering[M].Farrer Road,Singapore 9128.World Scientific Publishing Co.Pte.Ltd.1993:95-109.

[5]W.Pree.Design Patterns For Object-Oriented Software Development[M].Wokingham,England.Computing Machines(ACM)and Addison-Wesley Publishing Company，1994

[6]E.J.O'Neil,P.E.O'Neil and G.Weikum.The LRU-K Page Replacement Algorithm for Database Disk Buffering.Proc.of the 1993 ACM SIGMOD International Conference on Management of Data,pp.297-306.

[7][2]Pentakalos O I.An Introduction to the InfiniBand Architecture[M].High Performance Mass Storage and Parallel I/O:Technologies and Applications.Wiley-IEEE Press,2002:616.

[8][4]Vivek D.Deshmukh.InfiniBand:A New Era in Networkint[C].Proceedings of National Conference on Innovative Paradigms in Engineering&Technology.New York,USA:Foundation of Computer Science,2012.

[9][8]Pandey A,Pati U C.Panoramic Image Mosaicing:An Optimized Graph-Cut Approach[M].Proceedings of 3rd International Conference on Advanced Computing,Networking and Informatics.Springer India,2016:299.

[10]K.Shvachko,H.Kuang,S.Radia,R.Chansler.The Hadoop Distributed File System[J].Mass Storage Systems and Technologies (MSST),2010 IEEE 26th Symposium on.IEEE,2010：1-10.

Design of a General Parallel Computing Framework Based on Cluster

WANG Ning

（College of Computer Science,Sichuan University，Chengdu 610065）

In recent years,the amount of data and computation requirements in various fields has increased significantly.Traditional single computing devices are often incapable of performing such computational tasks.More and more fields are beginning to use parallel computing. Distributed computation is one of the main parallel computing methods,Hadoop is a most common framework based on Map-Reduce in distributed computation.Proposes a general parallel computing framework based on cluster.Describes the"task splitter","master node" and"computing node"in details with reference to"pipeline filter"mode.Compared with Hadoop,directed acyclic graph task is better supported,iteration task is supported.In addition,caching mechanism is added to reduce system time-consuming and support real-time application to a certain extent.

Parallel Computing;Cluster;System Framework;DAG;Cache

1007-1423（2016）35-0020-07

10.3969/j.issn.1007-1423.2016.35.004

王寧（1992-），男，陜西咸陽人，碩士研究生，研究方向為計算機圖形學、并行計算

2016-10-18

2016-11-30