分布式流體系結構及其編程模型與資源管理*

李　鑫，楊學軍，徐新海

(1.國防科技大學 計算機學院， 湖南 長沙　410073；

2.國防科技大學 高性能計算國家重點實驗室， 湖南 長沙　410073；

3.中國人民解放軍總參謀部第六十三研究所， 江蘇 南京　210007)

?

分布式流體系結構及其編程模型與資源管理*

李鑫1,2,3，楊學軍1,2，徐新海1,2

(1.國防科技大學 計算機學院， 湖南 長沙410073；

2.國防科技大學 高性能計算國家重點實驗室， 湖南 長沙410073；

3.中國人民解放軍總參謀部第六十三研究所， 江蘇 南京210007)

摘要：利用互聯網資源提供大數據計算服務面臨著資源異構性、動態性與通信長延遲等方面的挑戰，現有分布式計算模型仍存在一些不足。運用流計算模型提出分布式流體系結構，包括分布式流編程模型與資源管理等，能夠高效支持多種并行執行模式。在10個CPU-GPU異構結點上實現了原型系統，仿真實驗驗證了7個不同的測試用例。實驗結果表明，與本地串行計算相比，分布式流體系結構可以平均提高39倍計算性能，具有較大的應用潛力。

關鍵詞：流體系結構；大數據；編程模型；分布式計算

近年來，“大數據”已經成為國際社會普遍關注的熱點，在金融、軍事、電信等領域引起了人們的高度重視。大數據具有數據量大、種類繁多、處理速度快與價值密度低等4V特征。利用互聯網資源構建面向大數據計算的運行環境具有較好的發展前景，這種方法不僅能降低計算成本與提高資源利用率，還能提供可擴展的計算能力，但是卻需要應對互聯網資源異構性、動態性、通信長延遲與有限帶寬等挑戰。

目前，主流的分布式計算模型仍然存在一些不足。云計算[1]主要基于虛擬化技術等提供彈性可擴展的快速服務部署能力，提供大數據架構基礎設施的運行環境。Hadoop等主流大數據技術能應對多種應用場景，如MapReduce[2]模型等特別適合松耦合的大規模數據處理應用的批處理過程；Spark[3-4]等適合于機器學習等內存迭代計算；Spark Streaming[5]等比較適合于數據規模龐大且不可預知或者實時動態產生的流式數據處理；Dremel[6]與Impala[7]等交互式處理技術主要是在數據量非常大的情況下提供實時或準實時的數據查詢分析能力；PowerGraph[8]等圖計算技術則是面向互聯網社交網絡等大規模圖相關的應用場景，它們主要支持非結構化數據與半結構化數據的大規模處理，通常運行在數據中心較穩定的大規模同構資源上，但是在支持互聯網資源異構性以及動態性上還存在一些不足。網格計算[9]采用類MPI編程模型，在支持動態負載均衡上還有待改進。P2P計算模型[10]提供的大數據處理方式較為簡單，計算任務并行性較好，通常處理流程較簡單。

近年來，流計算模型已經成功應用在高性能計算、媒體類應用等領域[11-13]，并取得了“天河”等廣泛而有影響力的應用成果[14-15]。流計算模型具有計算資源普適性、高度數據并行性、延遲計算綁定特性以及流水線并行性等特點，有很大的潛力應對上述互聯網多方面的挑戰。因此，李鑫等基于流計算模型首次提出了一種新型的分布式流體系結構(Distributed Stream Architecture，DSA)以試圖解決互聯網環境下大數據計算模型的科學問題，并設計實現了分布式流體系結構編程模型Brook#與資源管理。

1分布式流體系結構

分布式流體系結構首次將流處理思想引入分布式領域，擴展了傳統流計算模型的概念，將可用的軟硬件計算對象定義為Kernel(計算核心)，同時將計算數據與控制狀態數據定義為Stream(流或數據流)，其基本概念如下：

控制數據流(ControlStream)：控制計算流程的數據或狀態數據；

計算數據流(ComputeStream)：封裝計算核心并行處理的數據；

軟計算核心(SoftKernel，SK)：封裝計算核心程序信息的對象，其元信息包括軟件共享庫名稱、網絡位置等；

硬計算核心(HardKernel，HK)：封裝結點內可用計算結點硬件資源信息的對象，其元信息包括網絡地址、處理器類型、線程數目等；

應用計算核心(ApplicationKernel，AK)：封裝應用程序中主程序代碼相關信息的對象，負責申請獲取資源，管理與監控計算任務運行；

客戶管理計算核心(Client Management Kernel，CMK)：提供用戶查詢和請求服務的接口；

資源管理計算核心(Resource Management Kernel，RMK)：提供命令解釋器與執行器的功能，負責向SMK注冊本地資源信息；

服務管理計算核心(Service Management Kernel，SMK)：提供應用服務等功能，負責維護服務(查詢、添加、刪除、更新等)、Kernel(HK，SK，AK與RMK)與用戶(CMK)的元信息，并負責調度軟硬件資源。

如圖1所示，分布式流體系結構的資源管理采用了主從架構，SMK負責統一維護資源元信息與資源調度，RMK負責命令解釋與執行，CMK用于提交應用程序請求。運行一個完整的分布式流應用程序通常采用1個host結點用于管理執行環境與多個device結點用于執行計算核心程序。其中：host結點上運行主程序(AK)，負責管理整個應用執行流程；device結點上運行普通計算核心程序(SK)。AK與SK均由RMK負責啟動執行。MPEG2編碼應用的復雜處理流程共包括1個AK與7個SK計算核心。

圖1　分布式流體系結構基本概念與MPEG2編碼應用Fig.1　Basic concepts of DSA and MPEG2 encoder application

host結點上的RMK負責啟動一個線程來管理應用主程序(AK)。當該線程執行到一個計算核心時，它會根據編譯指導命令劃分原計算任務為若干子任務并行執行，創建管理該任務及其子任務的線程，并向SMK申請執行該計算核心的軟硬件資源，通知計算結點RMK下載代碼與數據，計算完畢后由主程序(AK)更新維護數據一致性。主程序(AK)如此推進計算過程直至完成整個任務。

2分布式流體系結構編程模型Brook#

2.1　基本概念

Brook#在計算執行過程中流與計算核心的并行度劃分為四種Kernel執行模式，如圖2所示。

圖2　Kernel并行執行模式劃分圖Fig.2　Parallel execution modes of Kernel

單個計算核心單個流(Single Kernel Single Stream，SKSS)：即在一個計算結點上執行一個Kernel計算核心任務，處理單一流，依靠開發結點內處理器的并行性來提升計算能力；

單個計算核心多個流(Single Kernel Multiple Streams，SKMS)：即多個計算結點執行相同的Kernel代碼以完成一個計算核心任務，并處理不同的流，通過空間并行(SKMS-S)或時間并行(SKMS-T)的方式提高性能；

多個計算核心單個流(Multiple Kernels Single Stream，MKSS)：即多個計算結點上執行多個Kernel以流水線方式處理同一個流，通過時間并行的方式隱藏通信延遲，從而提高性能；

多個計算核心多個流(Multiple Kernels Multiple Streams，MKMS)：即多個計算結點上同時執行不同的Kernel代碼且處理不同的流，包括空間并行性(MKMS-S)與時間并行性(MKMS-T)兩種并行方式。MKSS是MKMS的一種特例。

2.2　編譯指導語句

Brook#繼承了傳統流編程模型AMD ATI Stream SDK 1.4 Beta的Brook+編程模型全部語法規范，并開發擴充了具有分布式語義的語法元素。見表1，Brook#允許程序員使用編譯指導語句顯式的指明代碼區域Kernel執行模式及相關屬性，采用C和C++標準提供的pragma機制，并提供三類形式：parallel_mode，distribute與barrier，同時無須關心資源異構性、動態負載均衡等底層細節。

表1　Brook#編譯指導語句表

①clause語法形式是in/out{streamName[(BLOCK/*(n),…), BLOCK/CYCLE(n)]}。

2.2.1parallel_mode語句

程序員可以使用兩個parallel_mode編譯指導語句包圍住代碼區域，并指明其單個或多個Kernel的并性執行模式parallel_mode，即

#pragma brs parallel_mode(n) start

Kernel/Multiple Kernels

#pragma brs parallel_mode finish

其中，n表示子任務數。

2.2.2distribute語句

由于子任務映射到輸入流與輸出流上的數據在維度上可能是不同的，所以，每個子任務映射的數據劃分方式也可能不一樣。程序員需要根據經驗與實際情況靈活配置流的劃分方式與任務映射方式，達到高效計算的目的，即

#pragma brs distribute [clause, …]

Kernel

其中，clause的語法形式如表1中注釋①所示，它表示輸入輸出流的基本屬性，包括流輸入或輸出方向、流名稱、流劃分方式以及與子任務的映射方式。

Brook#中流劃分方式包括塊分布(BLOCK)和缺省不劃分(*)兩種方式，其中，塊分布(BLOCK)是指將流按照指定的維度均勻劃分為n塊。子任務映射方式包括塊分布(BLOCK)與循環分布(CYCLE)，其中塊分布(BLOCK)是指同一個數據塊映射到n個連續的子任務上，循環分布(CYCLE)則是將數據塊依次循環映射到下標遞增的子任務上。如圖3所示，假設輸入流的劃分方式為BLOCK(4)，程序員指定啟動8個子任務并行執行該計算核心。若流映射任務的方式為BLOCK(2)，則第1個數據塊映射到子任務1與子任務2，說明這兩個子任務的計算都需要輸入流的第1個數據塊，依次類推。若流映射任務的方式為CYCLE，則說明子任務1與子任務5的計算需要第1個數據塊。數據塊映射子任務的方式是根據不同的程序執行特點而靈活指定的，需要程序員了解應用程序的執行特點來配置完成。

圖3　流劃分方式與子任務映射方式示例Fig.3　Example of data division method and sub-tasks mapping method for analyzing input streams

2.2.3barrier語句

程序員可以使用barrier指定程序執行的同步點，實現對單個或多個Kernel的同步操作，以確保該同步點之前所有Kernel或指定Kernel必須執行完畢后才能繼續執行程序，即

ProgramCode

2.3　Brook#編譯器運行時

分布式流體系結構編譯器運行時的整體組織結構示意圖如圖4所示。當RMK接收到啟動應用程序AK計算核心(host主程序)的請求后，RMK會創建執行線程(executor thread)負責執行AK程序，并調用編譯器運行時庫執行已經被翻譯成具有指定Brook#語義的多線程代碼序列。執行線程負責創建任務工作線程(worker thread)與子任務工作線程(subworker thread)用于管理每個計算核心任務及其子任務的執行流程，與遠程計算結點RMK進行交互，如流與代碼的下載與發送、任務啟動等。遠程device結點上RMK接收到啟動任務請求后，創建執行線程(SK代碼)啟動子任務，調用本地設備(CPU或GPU等)的Kernel函數代碼進行計算，一旦結束就通知host結點上本任務工作線程(AK程序)更新全局列表信息，以保證數據一致性。

圖4　編譯器Brook#運行時整體組織結構示例Fig.4　Example of entire organization 　　struture of Brook# runtime

圖4中分布式流程序包括matsum與matmul兩個計算核心，并分別被劃分為4個子任務與2個子任務，并采用MKMS-T執行模式以流水線并行的方式執行。因此，編譯器運行時在host結點上創建了1個executor thread、2個worker thread與6個subworker thread以及在device結點上創建了6個subworker thread。

3分布式流體系結構資源管理

分布式流體系結構資源管理主要包括SMK，RMK，CMK，AK，SK等組件，如圖5所示，其采用了主從架構以適應大規模資源的管理，具有良好的可擴展性，支持多任務獨立運行。

SMK負責維護結點資源元信息，包括對硬件、軟件、服務以及用戶等元信息的查詢、添加、刪除、更新等操作。如圖5所示，RMK與CMK啟動后主動注冊到SMK上，包括本地可用硬件信息，用戶則將作業的計算核心代碼(AK與SK)以及數據上傳到資源結點并注冊到SMK。此外，SMK還負責管理作業的生命周期過程，對資源請求進行合理的資源調度，以實現不同作業的安全隔離運行。

RMK資源管理計算核心負責解釋與執行請求的消息命令，是本地結點的資源管理器與任務執行器，管理本地可用硬件資源、作業文件資源與數據資源，并提供資源請求服務，同時管理與監控本地計算任務，周期性地向SMK匯報并更新本結點運行狀態。

CMK客戶管理計算核心提供客戶端的功能，一般部署在用戶結點上，負責將程序代碼(AK或SK)以及數據提交到資源結點上，并將應用程序或作業注冊到SMK，請求查詢作業運行狀態，并從資源結點上接收結果數據。

AK應用計算核心是封裝了作業主程序代碼信息的對象，負責每個應用程序的具體任務執行過程，并采用了一種中間列表法的方式來維護數據一致性。AK會維護記錄輸入輸出流結點信息的KernelList與記錄最近更新流的Kernel名的StreamList。當主程序執行線程(AK)執行到一個計算核心時，AK會主動向SMK申請資源分配給子任務，通過StreamList查詢更新輸入流的Kernel名，并在KernelList中查找其所在的結點信息，將這些信息發送給計算結點，通知RMK啟動計算并監控任務狀態。當任務完畢后，AK 更新KernelList中輸出流的結點信息和StreamList中對應流的Kernel名，從而維護數據一致性。

圖5　分布式流體系結構資源管理架構Fig.5　Framework of resource management of distributed stream architecture

4實驗驗證

整個實驗評估過程是在10個結點組成的互連網絡(千兆以太網)上完成的，每個結點由1個六核Intel Xeon X5670與1個AMD Radeon HD 4870×2 GPU組成，操作系統為64位Red Hat Enterprise Linux 5.4，內存容量為24GB。

選取7個典型測試用例(輸入規模，用例來源)，包括MatrixMul(16 384×16 384矩陣規模，AMD)、Blackscholes(30 000 000個期權，PARSEC)、BinomialOption(8 192 000個期權，AMD)、N-Body(499 968個體，AMD)、MRI-Q(64組Large數據，Parboil)、CP(100 000個原子，Parboil)與MatrixMulAdd(16 384×16 384矩陣規模，AMD)，并采用Brook#將其移植到分布式流體系結構原型系統上，分別簡記為MM，BL，BO，NB，MQ，CP與MMA，均采用單精度浮點數。

實驗采用的基準時間是本地1個CPU程序版本的串行執行時間，實驗的對比程序設置了三組測試。第一組測試使用本地1個GPU的程序版本(LOCAL-1G)，第二組與第三組測試都使用移植到分布式流計算體系結構上遠程運行的程序版本，其中，第二組測試采用多GPU程序版本(DSA-G)，第三組測試使用多CPU程序版本(DSA-C)或混合使用多CPU與多GPU的程序版本(DSA-C/G)。通過對比各組程序相對基準程序時間的加速比來評估在互聯網模擬通信延遲與有限帶寬等環境下分布式流體系結構的有效性。結點內的通信方法主要是基于PCIE協議與GPU等設備進行數據傳輸，結點間的通信方法是基于TCP/IP協議進行互聯網模擬通信，其中，國際互聯網延時采用Internet Traffic Report網站統計的2015年五大洲延遲時間平均值100ms，國際互聯網帶寬采用Speedtest網站在2013年186個國家與地區測量帶寬的30天移動平均值13.98Mbps。

實驗結果如圖6與圖7所示， 圖6顯示了三組測試相對本地1個CPU的加速比實驗結果，圖7顯示了第二組與第三組測試中測試用例通信時間、計算時間和其他時間開銷與通信開銷的歸一化時間統計情況。

第二組測試中的DSA-C或DSA-C/G版本平均加速比達到39倍，第三組測試中的DSA-G版本平均加速比達到58倍，分布式流體系結構可以支持模擬互聯網環境下測試用例的運行，能夠利用異構資源與并行計算來提升性能，其中，第三組測試中的MMA采用了MKMS-T與SKMS-S執行模式的DSA-C/G版本，其余測試用例都采用了SKMS-S執行模式的DSA-C或DSA-C/G版本，以挖掘程序潛在的線程級并行性與任務級并行性。

圖6　相對本地1個CPU執行時間的加速比實驗結果Fig.6　Speedups of experimental results over execution time of a local CPU

圖7　測試用例通信時間、計算時間以及其他時間的歸一化實驗結果Fig.7　Normalized experimental results of communication time, computing time and other time of benchmarks

基于多GPU的DSA-G或DSA-C/G版本的加速效果并不比LOCAL-1G版本顯著，這是由互聯網中通信問題的復雜性等造成的。從實驗結果看，DSA-G或DSA-C/G由于是遠程運行的應用程序，在互聯網模擬環境有限帶寬和長延遲等情況下，計算過程必然帶來不可忽視的通信開銷，使得通信開銷成為影響性能的因素。因此，這使得加速效果不如本地LOCAL-1G版本。

如圖6所示，7個典型測試用例在三組測試上的結果表現出三類特點：

1)一般情況下，應用加速比呈現下降趨勢，如MM，NB，CP與MMA。其中，MM，NB與MMA加速比下降得較快。這是由于分布式環境引入了大量不可忽視的通信與控制開銷，其中通信開銷平均達到82%，顯著增加了執行時間，這使得并行計算方式帶來的性能提升不足以補償通信開銷造成的性能損失，從而造成加速比下降。

2)加速比呈上升趨勢，如BL與BO。其中，BL的DSA-G與DSA-C/G加速比分別增長了3.2倍與4.8倍，BO的DSA版本與本地GPU版本具有相當的加速比。雖然通信帶來較大開銷，但在這些應用中引入的通信開銷與計算開銷相當，圖7中BL的通信開銷比例平均達到50%，異構資源強大的并行計算能力使得加速比保持上升。雖然BO通信開銷比例較高，但是相對其他測試組并不大，因此，使得加速比保持緩慢上升。

3)MQ加速比在DSA-G版本中上升，但在DSA-C版本中下降。這是由于前者利用GPU并行計算能力帶來的性能優勢能夠補償通信開銷造成的損失，而DSA-C版本則不足以補償這些性能損失，增大的程序控制開銷也影響了性能，造成加速比下降。

由此可見，在分布式流體系結構中應用程序的計算開銷、通信開銷與控制開銷之間的優化對于程序運行性能有至關重要的影響。

5結論

分布式流體系結構能夠較好地適應互聯網資源特點，提供高效的分布式編程模型與資源管理，支持多種計算核心并行執行模式，而無須關心資源異構性、動態資源綁定等細節，具有較大的大數據計算應用潛力。

參考文獻(References)

[1]Mell P M, Grance T. The NIST definition of cloud computing[R]. NIST, 2011.

[2]Dean J, Ghemawat S. MapReduce: simplified data processing on large clusters [C]// Proceedings of the 6th Conference on Symposium on Operating Systems Design and Implementation, 2004: 147-152.

[3]Zaharia M, Chowdhury M, Das T, et al. Resilient distributed datasets: a fault-tolerant abstraction for in-memory cluster computing [C]// Proceedings of the 9th USENIX Conference on Networked Systems Design and Implementation, USENIX Association, 2012:2.

[4]Zaharia M, Chowdhury M, Franklin M J, et al. Spark: cluster computing with working sets [C]// Proceedings of the 2nd USENIX Conference on Hot Topics in Cloud Computing. USENIX Association, 2010，10:10.

[5]Zaharia M, Das T, Li H, et al. Discretized streams: fault-tolerant streaming computation at scale [C]// Proceedings of the 24th ACM Symposium on Operating Systems Principles,ACM, 2013: 423-438.

[6]Melnik S, Gubarev A, Long J J, et al. Dremel: interactive analysis of web-scale datasets [C]// Proceedings of the VLDB Endowment, 2010, 3(1): 330-339.

[7]Erickson J. Impala: a modern SQL engine for Hadoop [R]. Tech Report, 2013.

[8]Gonzalez J, Low Y, Gu H. Power graph: distributed graph-parallel computation on natural graphs [C]// Proceedings of the 10th USENIX Symposium on Operating Systems Design and Implementation,USENIX Association, 2012: 17-30.

[9]Foster I, Kesselman C. The grid 2: blueprint for a new computing infrastructure [M]. 2rd ed.USA:Morgan Kaufmann Publishers Inc., 2003.

[10]Kamvar S D, Schlosser M T, Garcia-Molina H. The EigenTrust algorithm for reputation management in P2P networks [C]// Proceedings of the 12th International World Wide Web Conference,ACM, 2003: 640-651.

[11]Yang X J, Yan X B, Xing Z C, et al. A 64-bit stream processor architecture for scientific applications [C]// Proceedings of the 34th Annual International Symposium on Computer Architecture,San Diego, USA, ACM, 2007: 210-219.

[12]Kapasi U J, Dally W J, Rixner S, et al. The imagine stream processor [C]// Proceedings of the 20th IEEE International Conference on Computer Design: VLSI in Computers and Processors,IEEE, 2002: 282-288.

[13]Ye Y, Li K L, Wang Y, et al. Parallel computation of Entropic Lattice Boltzmann method on hybrid CPU-GPU accelerated system [J]. Computers & Fluids, 2015, 110:114-121.

[14]Xue W, Yang C, Fu H H, et al. Ultra-scalable CPU-MIC acceleration of mesoscale atmospheric modeling on Tianhe-2[J]. IEEE Transactions on Computers, 2015, 64(8):2382-2393.

[15]Liao X K, Pang Z B, Wang K F, et al. High performance interconnect network for Tianhe System [J]. Journal of Computer Science and Technology, 2015, 30(2):259-272.

http://journal.nudt.edu.cn

Programming model and resource management of distributed stream architecture

LIXin1,2,3,YANGXuejun1,2,XUXinhai1,2

(1. College of Computer, National University of Defense Technology, Changsha 410073, China；

2. State Key Laboratory of High Performance Computing, National University of Defense Technology, Changsha 410073, China；

3. The 63rd Research Institute of PLA General Staff Headquarters, Nanjing 210007, China)

Abstract：While providing big data computing services using Internet resources, there remains a big challenge to researchers, including heterogeneity of Internet resources, dynamics of Internet resources and long latency of Internet communication. Current influent distributed computing models still have some shortage. A novel distributed stream computing model was proposed based on the traditional stream computing model, including the distributed stream programming model and resource management can efficiently support multiple parallel execution modes. The prototype system implemented on the 10 CPU-GPU heterogeneous nodes. Seven different benchmarks used in the simulation experiment. The experimental result shows that the distributed stream architecture can achieve the speedup of at least on average over the local serial computing, with significant potential for applications.

Key words：stream architecture; big data; programming model; distributed computing

中圖分類號：TP338.8

文獻標志碼：A

文章編號：1001-2486(2015)06-110-06

作者簡介：李鑫(1984—)，男，安徽安慶人，博士研究生，E-mail：xinli@nudt.edu.cn；楊學軍(通信作者)，男，教授，博士，博士生導師，E-mail：xjyang@nudt.edu.cn

基金項目：國家自然科學基金資助項目(61221491，61303071)

收稿日期：*2015-09-06

doi:10.11887/j.cn.201506021