李 鑫，郭曉威，林宇斐

（1.國防科學(xué)技術(shù)大學(xué)高性能計(jì)算國家重點(diǎn)實(shí)驗(yàn)室，湖南 長沙410073；2.國防科學(xué)技術(shù)大學(xué)研究生院，湖南 長沙410073；3.總參第六十三研究所，江蘇 南京210007）

1 引言

在互聯(lián)網(wǎng)上提供面向大數(shù)據(jù)計(jì)算的運(yùn)行環(huán)境需要應(yīng)對資源異構(gòu)性、動態(tài)性、通信長延遲與帶寬有限等挑戰(zhàn)，現(xiàn)有的分布式計(jì)算模型尚存在一些不足，如云計(jì)算［1］可以實(shí)現(xiàn)對多種異構(gòu)物理資源的統(tǒng)一虛擬化資源池管理，僅僅為大數(shù)據(jù)平臺提供支撐運(yùn)行環(huán)境。目前主流的大數(shù)據(jù)技術(shù)，如MapReduce［2］、Spark Streaming［3］、Shark［4］、Hive［5］、Spark［6～7］等，主要是基于數(shù)據(jù)中心較穩(wěn)定的大規(guī)模同構(gòu)資源，對互聯(lián)網(wǎng)資源異構(gòu)性與節(jié)點(diǎn)動態(tài)性的支持還存在一定不足。網(wǎng)格計(jì)算［8］采用了類MPI編程模型，利用中間件屏蔽資源異構(gòu)性，但是其靜態(tài)綁定資源與數(shù)據(jù)的方式對資源動態(tài)性的支持還存在一定不足。P2P 計(jì)算模型［9］利用作業(yè)高度并行的特點(diǎn)進(jìn)行分布式計(jì)算，較難適應(yīng)流程復(fù)雜的應(yīng)用。

近年來，流計(jì)算模型已經(jīng)成功應(yīng)用在高性能計(jì)算等領(lǐng)域，如通用圖形處理器GPGPU 和Intel Xeon Phi［10］在“Tianhe－1A”［11］與“Tianhe－2”［12］的應(yīng)用，并應(yīng)用于石油勘探、動漫渲染、磁約束聚變數(shù)值模擬等大規(guī)模計(jì)算與數(shù)據(jù)處理應(yīng)用上，充分驗(yàn)證了流計(jì)算模型的實(shí)際性能與良好應(yīng)用特性。因此，針對上述挑戰(zhàn)與難題，作者基于流計(jì)算模型已提出了分布式流體系結(jié)構(gòu)作為其解決方案，可以有效適應(yīng)互聯(lián)網(wǎng)上不同的異構(gòu)計(jì)算資源、數(shù)據(jù)格式與多種執(zhí)行模式，為大數(shù)據(jù)應(yīng)用提供高效、低成本的互聯(lián)網(wǎng)計(jì)算環(huán)境。

雖然分布式流體系結(jié)構(gòu)已經(jīng)挖掘了相鄰計(jì)算核心函數(shù)之間的并行性，通過計(jì)算與通信等操作的重疊以隱藏通信長延時(shí)。但是，互聯(lián)網(wǎng)長通信開銷仍然是一個(gè)艱巨的挑戰(zhàn)。在分布式流體系結(jié)構(gòu)中，數(shù)據(jù)傳輸操作通常直到計(jì)算核心任務(wù)啟動操作之前才會請求執(zhí)行，由主程序通知數(shù)據(jù)所在節(jié)點(diǎn)將數(shù)據(jù)發(fā)送到后繼計(jì)算節(jié)點(diǎn)，從模擬實(shí)驗(yàn)結(jié)果數(shù)據(jù)分析看，在互聯(lián)網(wǎng)有限帶寬與長延遲的情況下，這種被動數(shù)據(jù)請求方式的通信時(shí)間會占用較長的數(shù)據(jù)等待時(shí)間（模擬實(shí)驗(yàn)顯示至少占40%以上執(zhí)行時(shí)間）。本文將這種被動響應(yīng)數(shù)據(jù)請求的方式稱之為數(shù)據(jù)流Lazy傳輸技術(shù)，顯然，它是一種保守的數(shù)據(jù)傳輸技術(shù)，嚴(yán)格按照程序原本的順序串行執(zhí)行并傳輸數(shù)據(jù)，以確保程序語義的正確性。本文提出了一種主動傳輸數(shù)據(jù)的方式，即數(shù)據(jù)流Eager傳輸技術(shù)。該技術(shù)可以將數(shù)據(jù)提前發(fā)送到目標(biāo)節(jié)點(diǎn)，挖掘非相鄰連續(xù)計(jì)算核心間之間潛在的并行性，執(zhí)行任務(wù)時(shí)不需要被動地等待數(shù)據(jù)傳輸，從而加快程序執(zhí)行性能。

2 分布式流體系結(jié)構(gòu)概述

2.1 基本概念

傳統(tǒng)的流計(jì)算模型以一種流（Stream）的觀點(diǎn)來組織程序結(jié)構(gòu)，數(shù)據(jù)被抽象為可并行操作的數(shù)據(jù)流，應(yīng)用被分解為并行執(zhí)行的若干計(jì)算核心函數(shù)程序（Kernel）。Kernel可以被映射到支持程序運(yùn)行的任何計(jì)算資源上，通過數(shù)據(jù)通道進(jìn)行數(shù)據(jù)交換，從而簡化計(jì)算流程，提高計(jì)算效率。這使得流計(jì)算模型具有計(jì)算資源普適性、高度數(shù)據(jù)并行性與延遲計(jì)算綁定特性、流水線并行性等特性。

作者基于流計(jì)算模型提出了一種新型的分布式流體系結(jié)構(gòu)DSA（Distributed Stream Architecture），在分布式環(huán)境下提供大數(shù)據(jù)運(yùn)行環(huán)境，從分布式計(jì)算模型的角度出發(fā)，將所有可用的軟硬件對象定義為計(jì)算核心（Kernel），所有計(jì)算數(shù)據(jù)與控制狀態(tài)數(shù)據(jù)定義為數(shù)據(jù)流（Stream），以描述分布式流體系結(jié)構(gòu)中的計(jì)算機(jī)制，刻畫其程序的執(zhí)行特點(diǎn)，其基本概念包括：

定義1 數(shù)據(jù)流包括兩種類型：

（1）控制數(shù)據(jù)流（ControlStream）：控制計(jì)算流程的數(shù)據(jù)或狀態(tài)數(shù)據(jù)；

（2）計(jì)算數(shù)據(jù)流（ComputeStream）：封裝計(jì)算核心并行處理的數(shù)據(jù)。

定義2 計(jì)算核心包括六種類型：

（1）軟計(jì)算核心SK（SoftKernel）：封裝計(jì)算核心程序信息的對象，其元信息包括軟件共享庫名稱、網(wǎng)絡(luò)位置等；

（2）硬計(jì)算核心HK（HardKernel）：封裝節(jié)點(diǎn)內(nèi)可用硬件資源信息的對象，其元信息包括網(wǎng)絡(luò)地址、處理器類型等；

（3）應(yīng)用計(jì)算核心AK（ApplicationKernel）：封裝應(yīng)用程序中主程序代碼相關(guān)信息的對象，是一種特殊的SK 代碼對象，負(fù)責(zé)申請獲取資源，監(jiān)控任務(wù)運(yùn)行狀態(tài)；

（4）客戶管理計(jì)算核心CMK（Client Management Kernel）：提供用戶查詢和請求服務(wù)的接口；

（5）資源管理計(jì)算核心RMK（Resource Management Kernel）：提供命令解釋器與執(zhí)行器的功能，負(fù)責(zé)向SMK 注冊本地資源信息；

（6）服務(wù)管理計(jì)算核心SMK（Service Management Kernel）：提供應(yīng)用服務(wù)等功能，負(fù)責(zé)維護(hù)服務(wù)（查詢、添加、刪除、更新等）與Kernel（HK、SK、AK、RMK 與CMK）的元信息，并調(diào)度軟硬件資源。

如圖1所示，以MPEG2編碼應(yīng)用為例說明分布式流體系結(jié)構(gòu)的運(yùn)行機(jī)制，如圖1a所示，N0節(jié)點(diǎn)上部署了SMK，N1～N9節(jié)點(diǎn)上部署了RMK，N10節(jié)點(diǎn)上部署了CMK，且CMK 有應(yīng)用程序MPEG2的主程序與各個(gè)計(jì)算核心程序。用戶運(yùn)行MPEG2編碼應(yīng)用的過程如下：

Figure 1 Execution flow diagram of the MPEG2encoding application in the distributed stream architecture圖1 在分布式流體系結(jié)構(gòu)上MPEG2編碼應(yīng)用執(zhí)行流程圖

（1）用戶通過CMK 向SMK 申請運(yùn)行應(yīng)用程序，SMK 返回資源節(jié)點(diǎn)N9等信息，CMK 接收后將主程序以及計(jì)算核心程序上傳到N9。同時(shí)，N1～N9節(jié)點(diǎn)上的RMK 啟動后主動將本地硬件資源注冊到SMK 上，如圖1b所示。

（2）CMK 向SMK 申 請 啟 動MPEG2 應(yīng) 用 程序，SMK 返回資源節(jié)點(diǎn)N1作為host主節(jié)點(diǎn)并啟動主程序（AK），如圖1c 的（2）所示。N1上的RMK 啟動線程從N9下載代碼及數(shù)據(jù)，并啟動AK 計(jì)算核心，如圖1c的（3）所示。

（3）當(dāng)AK 執(zhí)行到第一個(gè)計(jì)算核心（Color Conversion）時(shí)，AK 主動向SMK 申請計(jì)算資源，并分配得到計(jì)算節(jié)點(diǎn)N2作為device計(jì)算節(jié)點(diǎn)，AK 請求N2上的RMK 啟動線程運(yùn)行該計(jì)算核心，如圖1c的（4）所示，并從N1與N9下載代碼與數(shù)據(jù)，如圖1c的（5）所示。

（4）N2在獲取計(jì)算核心代碼與輸入數(shù)據(jù)后，啟動程序（Color Conversion）執(zhí)行計(jì)算任務(wù)，計(jì)算完畢后更新任務(wù)狀態(tài)與數(shù)據(jù)狀態(tài)給AK，如圖1d的（6）所示。AK 繼續(xù)執(zhí)行主程序，當(dāng)其執(zhí)行到計(jì)算核心（DCT）時(shí)根據(jù)編譯指導(dǎo)語句向SMK 申請兩個(gè)節(jié)點(diǎn)執(zhí)行代碼，SMK 分配N3與N4用于并行執(zhí)行計(jì)算任務(wù)，AK 劃分子任務(wù)后通知N3與N4上的RMK 啟動線程，并下載代碼與數(shù)據(jù)啟動子任務(wù)計(jì)算，如圖1d的（7）～（9）所示。

AK 如此推進(jìn)其他計(jì)算核心的執(zhí)行，直至所有任務(wù)計(jì)算完畢，最后通知將輸出數(shù)據(jù)上傳到資源節(jié)點(diǎn)N9上，如圖1e的（10）所示。AK 通知SMK 應(yīng)用程序計(jì)算完畢，請求釋放資源，SMK 主動釋放所有計(jì)算節(jié)點(diǎn)，重新添加到空閑資源池中，并通知CMK 從資源節(jié)點(diǎn)N9下載結(jié)果數(shù)據(jù)，至此，MPEG2應(yīng)用運(yùn)行完畢。

2.2 編程模型

分布式流體系結(jié)構(gòu)編程模型Brook＃提供了三種基本的編譯指導(dǎo)語句形式：parallel＿mode、distribute與barrier，采用C 和C＋＋標(biāo)準(zhǔn)提供的pragma機(jī)制，均以＃pragma brs開頭，允許程序員以顯式的方式指明代碼區(qū)域的程序執(zhí)行模式，使用時(shí)添加在代碼區(qū)域的起始與結(jié)束位置。表1展示了Brook＃核心編譯指導(dǎo)語句的所有語法細(xì)節(jié)。從計(jì)算執(zhí)行過程中數(shù)據(jù)流與計(jì)算核心的并行度看，Brook＃支持四種Kernel執(zhí)行模式：

（1）SKSS（Single Kernel Single Stream）：即在一個(gè)計(jì)算節(jié)點(diǎn)上執(zhí)行一個(gè)Kernel，處理單一數(shù)據(jù)流，這是最基本的執(zhí)行模型，主要依靠開發(fā)節(jié)點(diǎn)內(nèi)處理器的并行性來提升計(jì)算能力。

（2）SKMS（Single Kernel Multiple Streams）：即多個(gè)計(jì)算節(jié)點(diǎn)執(zhí)行相同的Kernel，但是處理不同的數(shù)據(jù)流，每個(gè)計(jì)算節(jié)點(diǎn)處理各自的數(shù)據(jù)流，通過空間并行方式或時(shí)間并行方式來提高單個(gè)Kernel處理性能，即SKMS－S與SKMS－T。該執(zhí)行模式利用數(shù)據(jù)并行性將同一任務(wù)盡可能平均劃分成多個(gè)子任務(wù)執(zhí)行，使其工作負(fù)載盡可能均衡。

（3）MKSS（Multiple Kernels Single Stream）：即多個(gè)計(jì)算節(jié)點(diǎn)上執(zhí)行多個(gè)Kernel以流水線方式處理同一個(gè)數(shù)據(jù)流，可以通過時(shí)間并行方式隱藏通信延遲，提高處理效率。

（4）MKMS （Multiple Kernels Multiple Streams）：即多個(gè)計(jì)算節(jié)點(diǎn)上同時(shí)執(zhí)行不同的Kernel處理不同的數(shù)據(jù)流，通過空間并行方式或時(shí)間并行方式同時(shí)執(zhí)行，用于開發(fā)多個(gè)計(jì)算核心之間的并行性，即MKMS－S與MKMS－T，每個(gè)Kernel只處理相關(guān)的數(shù)據(jù)流（任務(wù)級并行）。MKSS屬于MKMS的一個(gè)特例。

分布式流編程模型Brook＃可以充分利用互聯(lián)網(wǎng)分布式環(huán)境下的資源，能夠開發(fā)多個(gè)任務(wù)的任務(wù)級并行性與線程級并行性，挖掘程序間通信與計(jì)算的重疊操作，并提供多種性能優(yōu)化技術(shù)。

表1 中 的clause 指in/out｛streamName［（BLOCK/＊（n），…），BLOCK/CYCLE（n）］｝，表示數(shù)據(jù)流輸入或輸出方向、數(shù)據(jù)流名稱、子流的數(shù)據(jù)分布方式以及與子任務(wù)的映射方式，通過該語句實(shí)現(xiàn)數(shù)據(jù)流空間到子任務(wù)空間的映射。

Table 1 Brook＃compilation directives表1 Brook＃編譯指導(dǎo)語句列表

2.3 資源管理系統(tǒng)

在分布式流體系結(jié)構(gòu)中，資源管理系統(tǒng)一方面負(fù)責(zé)互聯(lián)網(wǎng)節(jié)點(diǎn)資源信息的維護(hù)，包括對硬件資源、軟件資源、服務(wù)以及用戶等元信息進(jìn)行查詢、添加、刪除、更新等；另一方面提供調(diào)度器對用戶的資源請求進(jìn)行資源調(diào)度，從資源池中選出符合請求的資源，同時(shí)實(shí)現(xiàn)對計(jì)算任務(wù)的監(jiān)控、啟動等任務(wù)管理功能。如圖2所示，資源管理系統(tǒng)主要由SMK、RMK、CMK、AK、SK、HK 等組件構(gòu)成。

SMK 負(fù)責(zé)維護(hù)節(jié)點(diǎn)資源元信息，其功能包括：

（1）負(fù)責(zé)注冊RMK 與CMK、RMK 的本地硬件信息、用戶作業(yè)的計(jì)算核心代碼等；

（2）負(fù)責(zé)管理作業(yè)的生命周期過程；

（3）負(fù)責(zé)對資源請求進(jìn)行資源調(diào)度，實(shí)現(xiàn)不同作業(yè)的安全隔離運(yùn)行。

RMK 資源管理計(jì)算核心是本地節(jié)點(diǎn)的命令解釋器、資源管理器與任務(wù)執(zhí)行器，其功能包括：

（1）管理本地硬件資源、作業(yè)文件資源與數(shù)據(jù)資源，并提供資源請求服務(wù)；

（2）管理與監(jiān)控本地計(jì)算任務(wù)，周期性發(fā)送心跳消息到SMK 更新狀態(tài)。

CMK 客戶管理計(jì)算核心類似于客戶端的功能，負(fù)責(zé)提交程序代碼（AK 或SK）以及數(shù)據(jù)到資源節(jié)點(diǎn)上，在SMK 上注冊作業(yè)信息，并接收結(jié)果數(shù)據(jù)。

AK應(yīng)用計(jì)算核心封裝了作業(yè)主程序代碼信息，負(fù)責(zé)具體每個(gè)應(yīng)用程序的執(zhí)行流程，其功能包括：

（1）負(fù)責(zé)向SMK 申請資源并分配給子任務(wù)；

（2）負(fù)責(zé)通知節(jié)點(diǎn)RMK 啟動子任務(wù)并監(jiān)控計(jì)算核心任務(wù)（SK）任務(wù)狀態(tài)；

（3）負(fù)責(zé)維護(hù)應(yīng)用程序數(shù)據(jù)的一致性。

Figure 2 Resource manager system framework of the distributed stream architecture圖2 分布式流體系結(jié)構(gòu)資源管理系統(tǒng)架構(gòu)

3 數(shù)據(jù)流Eager傳輸技術(shù)

3.1 基本概念

定義3 給定一個(gè)有向圖G＝〈V＊，E＊〉，程序每條語句都映射為圖中的節(jié)點(diǎn)V∈V＊，語句間的執(zhí)行次序關(guān)系映射為圖中的有向邊E∈E＊，有向邊的方向指明語句執(zhí)行順序，這樣形成的有向圖稱為程序控制依賴圖或控制流圖。

定義4 程序執(zhí)行層次是指程序執(zhí)行時(shí)進(jìn)入分支循環(huán)結(jié)構(gòu)的深度。當(dāng)進(jìn)入一個(gè)分支結(jié)構(gòu)或循環(huán)結(jié)構(gòu)時(shí)程序的執(zhí)行層次增加一層，當(dāng)退出同一個(gè)分支結(jié)構(gòu)或循環(huán)結(jié)構(gòu)時(shí)程序的執(zhí)行層次減少一層。程序啟動執(zhí)行時(shí)的執(zhí)行層次默認(rèn)為第0層。

定義5 當(dāng)Kernel函數(shù)調(diào)用語句Pt0與Pt1存在控制依賴關(guān)系時(shí)，

（1）若Pt0與Pt1在同一個(gè)程序執(zhí)行層次上，兩者在同一個(gè)分支結(jié)構(gòu)上、循環(huán)體結(jié)構(gòu)或順序結(jié)構(gòu)的同一條執(zhí)行路徑上，不能跨越分支循環(huán)結(jié)構(gòu)層次，則稱Pt0與Pt1之間存在確定性控制依賴關(guān)系，記為［Pt0，Pt1］Dc；

（2）若Pt0與Pt1不存在確定性控制依賴關(guān)系，則稱兩者具有非確定性控制依賴關(guān)系，記為［Pt0，Pt1］UDc。

因此，分布式流程序中具有確定性控制依賴關(guān)系的計(jì)算核心需要遵循兩個(gè)基本約束規(guī)則：

（1）規(guī)則1：［Pt0，Pt1］Dc中的Pt0與Pt1是同一個(gè)層次執(zhí)行路徑的必經(jīng)節(jié)點(diǎn)，不能跨越分支循環(huán)結(jié)構(gòu)，若Pt0計(jì)算核心在分支循環(huán)結(jié)構(gòu)里，則Pt1不能在分支循環(huán)結(jié)構(gòu)外。

（2）規(guī)則2：［Pt0，Pt1］Dc中的Pt0或Pt1可以是單個(gè)計(jì)算核心、MKMS－T 或MKMS－S并行執(zhí)行模式程序塊中的計(jì)算核心，但兩者不能同時(shí)是同一個(gè)MKMS－S并行執(zhí)行模式程序塊里的計(jì)算核心。

3.2 數(shù)據(jù)流Eager傳輸技術(shù)的充分條件

定義6 給定一個(gè)有向圖GK＝〈V＊，E＊〉，程序Kernel函數(shù)調(diào)用語句都映射為圖中節(jié)點(diǎn)V∈V＊，Kernel之間的數(shù)據(jù)依賴關(guān)系映射為圖中有向邊E∈E＊，有向邊方向指明數(shù)據(jù)依賴方向，對于任意兩個(gè)Kernel函數(shù)調(diào)用語句Pt0與Pt1的映射節(jié)點(diǎn)Vt0與Vt1，若兩節(jié)點(diǎn)之間存在有向邊Et∈E＊，則稱Kernel函數(shù)調(diào)用語句Pt0與Pt1之間存在數(shù)據(jù)依賴關(guān)系，記為［Pt0，Pt1］d。

定義7 給定同一個(gè)程序的控制依賴圖Gc＝〈Vc，Ec〉與計(jì)算核心數(shù)據(jù)依賴圖Gd＝〈Vd，Ed〉，對于其中任意兩個(gè)計(jì)算核心Kernel調(diào)用語句Pt0與Pt1，若在控制依賴圖Gc中存在確定性控制依賴關(guān)系［Pt0，Pt1］Dc，并且在數(shù)據(jù)依賴圖Gd中對應(yīng)地存在數(shù)據(jù)依賴關(guān)系［Pt0，Pt1］d，則稱Pt0與Pt1之間存在控制與數(shù)據(jù)依賴關(guān)系對，記為［Pt0，Pt1］dc，簡稱Pt0與Pt1為計(jì)算核心對。

定理1 數(shù)據(jù)流Eager傳輸?shù)某浞謼l件是計(jì)算核心Kernel調(diào)用語句之間存在控制與數(shù)據(jù)依賴關(guān)系對。

證明 若計(jì)算核心Kernel調(diào)用語句Pt0與Pt1存在控制與數(shù)據(jù)依賴關(guān)系對［Pt0，Pt1］dc，則在控制依賴圖中一定存在確定性控制依賴關(guān)系［Pt0，Pt1］Dc，即程序執(zhí)行完P(guān)t0后才會執(zhí)行Pt1，而且一定會按照兩者的依賴關(guān)系順序執(zhí)行。因此，Pt0執(zhí)行完畢后才會有數(shù)據(jù)傳輸給Pt1，確保數(shù)據(jù)生成、傳輸與接收操作順序的正確性。

同時(shí)，Pt0與Pt1在數(shù)據(jù)依賴圖中存在數(shù)據(jù)依賴關(guān)系［Pt0，Pt1］d，則它們之間具有明確的數(shù)據(jù)流，Pt1的輸入數(shù)據(jù)依賴于Pt0的輸出數(shù)據(jù)，因此，當(dāng)Pt0計(jì)算完畢后，其輸出數(shù)據(jù)就是Pt1的輸入數(shù)據(jù)，確保了數(shù)據(jù)傳輸方向的正確性。

因此，若計(jì)算核心Kernel調(diào)用語句之間存在控制與數(shù)據(jù)依賴關(guān)系對，則可以采用數(shù)據(jù)流Eager傳輸技術(shù)提前傳輸數(shù)據(jù)到后繼目標(biāo)節(jié)點(diǎn)上。

3.3 識別控制與數(shù)據(jù)依賴關(guān)系對的靜態(tài)編譯分析方法

本文認(rèn)為分布式流計(jì)算程序是可歸約的或結(jié)構(gòu)良好的，即能夠通過一系列的變換將程序歸約為單個(gè)節(jié)點(diǎn)，而且不存在非正常區(qū)域或非結(jié)構(gòu)化區(qū)域的控制結(jié)構(gòu)。靜態(tài)編譯分析方法采用結(jié)構(gòu)分析的思路，將程序通過分析變換生成控制樹，找出所有符合條件的計(jì)算核心對集合。

Figure 3 Three program control structures of the distributed stream computing program圖3 分布式流計(jì)算程序的三類程序控制結(jié)構(gòu)

Brook＃編譯器基于LALR 算法對整個(gè)程序進(jìn)行分析并形成語句鏈表，根據(jù)語句相關(guān)屬性及其連接關(guān)系，將語句抽象為節(jié)點(diǎn)，將控制依賴關(guān)系抽象為邊，從而生成程序控制樹。其中，主程序包括六類語句：一般表達(dá)式語句與編譯指導(dǎo)語句、Kernel函數(shù)調(diào)用語句、SWITCH 分支語句、復(fù)合語句、分支循環(huán)語句與并行語句，且不允許出現(xiàn)復(fù)合語句與并行控制結(jié)構(gòu)存在相互嵌套的語法，如圖3 所示。同時(shí)，編譯器將語句變換為節(jié)點(diǎn)，并標(biāo)記節(jié)點(diǎn)類型：將一般表達(dá)式語句（S類型）、并行標(biāo)記語句（PS類型）與Kernel函數(shù)調(diào)用語句（K 類型）都變換為葉節(jié)點(diǎn)，將SWITCH 分支語句（W 類型）、復(fù)合語句（C類型）、分支循環(huán)語句（F類型）與并行語句（P類型）都變換為抽象節(jié)點(diǎn)。這樣，程序中的每條語句與表達(dá)式都一一映射到抽象節(jié)點(diǎn)與葉節(jié)點(diǎn)上，不存在二義性問題，其中，控制樹的根節(jié)點(diǎn)是原來的主程序，根節(jié)點(diǎn)和葉節(jié)點(diǎn)中間的節(jié)點(diǎn)是三種控制結(jié)構(gòu)的抽象節(jié)點(diǎn)，樹的邊表示每個(gè)控制結(jié)構(gòu)對應(yīng)抽象節(jié)點(diǎn)（即父節(jié)點(diǎn)）和那些構(gòu)成該控制結(jié)構(gòu)的語句（即后裔節(jié)點(diǎn)）之間的復(fù)合構(gòu)造關(guān)系。

假設(shè)控制樹一共m層，計(jì)算核心對集合NT初始化為空，其搜索方法如下：

（1）第一遍深度優(yōu)先后序遍歷控制樹：

①若當(dāng)前節(jié)點(diǎn)NC是P類型抽象節(jié)點(diǎn)，則將K類型子節(jié)點(diǎn)放入本節(jié)點(diǎn)的候選計(jì)算核心集合N中，置位搜索標(biāo)志位；

②若當(dāng)前節(jié)點(diǎn)NC是C類型抽象節(jié)點(diǎn)，則將K類型子節(jié)點(diǎn)、C類型子節(jié)點(diǎn)與P類型子節(jié)點(diǎn)候選計(jì)算核心集合加入到候選計(jì)算核心集合N中，置位NC搜索標(biāo)志位，并取消子節(jié)點(diǎn)搜索標(biāo)志位；

③若當(dāng)前節(jié)點(diǎn)NC是W類型或F類型抽象節(jié)點(diǎn)，則合并節(jié)點(diǎn)下所有子孫節(jié)點(diǎn)中的K類型節(jié)點(diǎn)的輸出流到該抽象節(jié)點(diǎn)輸出流集合OS。

（2）第二遍深度優(yōu)先后序遍歷控制樹：

①若當(dāng)前節(jié)點(diǎn)NC是P類型或C類型抽象節(jié)點(diǎn)，并且設(shè)置了搜索標(biāo)志位。首先，識別出本節(jié)點(diǎn)候選集合N中所有存在數(shù)據(jù)依賴關(guān)系的集合NBTC，其中，后繼節(jié)點(diǎn)集合記為NKTC。接著，假設(shè)其子節(jié)點(diǎn)中存在為W類型或F類型的抽象節(jié)點(diǎn)，其輸出流集合為OS，則在NKTC中含有OS的后繼節(jié)點(diǎn)集合為NKO，則記NK＝NKTC－NKO。最后，在NBTC中搜索出含有后繼節(jié)點(diǎn)集合NK的計(jì)算核心對集合NTC，則搜索樹的計(jì)算核心對集合NT＝NT∪NTC。

②其他類型的節(jié)點(diǎn)均沒有設(shè)置標(biāo)志位，不做任何操作。

至此，搜索存在控制與數(shù)據(jù)依賴關(guān)系對的計(jì)算核心對集合的靜態(tài)編譯方法識別過程結(jié)束，NT包含了控制樹所有存在控制與數(shù)據(jù)依賴關(guān)系對的計(jì)算核心對集合。

3.4 編譯指導(dǎo)語句與實(shí)現(xiàn)機(jī)制

在Brook＃語法中增加數(shù)據(jù)流Eager傳輸技術(shù)的編譯指導(dǎo)語句，即

該編譯指導(dǎo)語句使用在指定程序段的開始和結(jié)尾處，其語義是指由編譯器分析與標(biāo)記程序中的控制與數(shù)據(jù)依賴關(guān)系對集合，在計(jì)算核心執(zhí)行完畢時(shí)主動發(fā)送數(shù)據(jù)。

如圖4所示案例，兩個(gè)串行執(zhí)行的計(jì)算核心K1與K2均采用SKMS－S執(zhí)行模式，分別劃分為四個(gè)子任務(wù)與兩個(gè)子任務(wù)來并行執(zhí)行，并采用數(shù)據(jù)流Eager傳輸技術(shù)。host節(jié)點(diǎn)上為K1創(chuàng)建一個(gè)executor thread 線 程 執(zhí) 行 主 程 序（AK），一 個(gè)worker thread線程與四個(gè)subworker thread線程管理子任務(wù)的執(zhí)行過程，不同線程之間通過事件消息隊(duì)列來傳遞信息，其中，worker thread 與subworker thread的事件消息隊(duì)列分別簡記為wq與sq。同時(shí)，在四個(gè)device節(jié)點(diǎn)上創(chuàng)建了executor thread線程用于執(zhí)行計(jì)算核心代碼（SK）。其中，host節(jié)點(diǎn)上：

（1）executor thread：執(zhí)行AK 計(jì)算核心主 程序，管理整個(gè)程序的執(zhí)行流程，它為每個(gè)計(jì)算核心啟動一個(gè)worker thread維護(hù)其計(jì)算流程，并阻塞等待當(dāng)前Kernel計(jì)算完畢后才會繼續(xù)執(zhí)行下一個(gè)Kernel；

（2）worker thread：負(fù)責(zé)執(zhí)行對計(jì)算核心任務(wù)的所有相關(guān)操作與狀態(tài)監(jiān)控，在任務(wù)結(jié)束后負(fù)責(zé)更新數(shù)據(jù)流信息，以保證數(shù)據(jù)一致性；

（3）subworker thread：負(fù)責(zé)執(zhí)行對計(jì)算核心子任務(wù)的所有相關(guān)操作與狀態(tài)監(jiān)控；

（4）device節(jié)點(diǎn)上executor thread：負(fù)責(zé)直接執(zhí)行子任務(wù)的相關(guān)操作，包括請求下載數(shù)據(jù)與代碼、啟動計(jì)算執(zhí)行等，并將計(jì)算完成狀態(tài)發(fā)送給主節(jié)點(diǎn)。

為了支持?jǐn)?shù)據(jù)流Eager傳輸技術(shù)，分布式流體系結(jié)構(gòu)引入了code thread線程與cq事件消息隊(duì)列，其中：

（1）code thread：負(fù)責(zé)更新采用Eager技術(shù)傳輸?shù)臄?shù)據(jù)狀態(tài)信息，以維護(hù)數(shù)據(jù)一致性；

Figure 4 Organization structure of the runtime that supports the stream Eager transmission technique圖4 支持?jǐn)?shù)據(jù)流Eager傳輸技術(shù)的運(yùn)行時(shí)組織結(jié)構(gòu)示意圖

（2）cq事件消息隊(duì)列：當(dāng)Kernel子任務(wù)采用Eager傳輸技術(shù)發(fā)送完數(shù)據(jù)后，后繼Kernel的cq隊(duì)列會接收到更新的數(shù)據(jù)狀態(tài)消息。

本節(jié)采用一種線程操作表的偽代碼方法來描述分布式流體系結(jié)構(gòu)技術(shù)實(shí)現(xiàn)的相關(guān)細(xì)節(jié)。所使用到的操作符號如表2 所示。假設(shè)計(jì)算核心K2劃分為兩個(gè)子任務(wù)并行執(zhí)行，則分別記為K21與K22，使用K21.OP表示計(jì)算核心子任務(wù)K21執(zhí)行相關(guān)操作OP，用ex（OP）表示當(dāng)前線程執(zhí)行操作OP，wq.p（OP）表示wq事件消息隊(duì)列壓入相應(yīng)工作線程執(zhí)行的操作OP，并交給相應(yīng)的工作線程處理。

Table 2 Operation list of Kernel and Stream表2 計(jì)算核心與數(shù)據(jù)流相關(guān)操作列表

如表3所示為K1與K2計(jì)算核心在SKMS－S執(zhí)行模式下各個(gè)線程的處理操作序列。表3 中executor thread列中代碼的執(zhí)行順序代表了應(yīng)用主程序（AK 代碼）的執(zhí)行順序，同一行的操作代碼表示對應(yīng)操作在不同線程中的執(zhí)行流程，不同線程執(zhí)行操作的先后順序是由該操作決定的，圖3最后一列指明了事件消息在各線程執(zhí)行順序的大致方向。圖4中的關(guān)鍵操作都標(biāo)記在表3相應(yīng)的位置，且1≤i≤4，1≤j≤2，i與j均為整數(shù)，其中，host節(jié)點(diǎn)上：

（1）executor thread：負(fù)責(zé)執(zhí)行主程序。

①ex（K1.OPT）：創(chuàng)建work thread線程用于管理計(jì)算核心任務(wù)的執(zhí)行流程；

②wq.p（K1.OPST）：壓入創(chuàng)建子任務(wù)工作線程操作OPST，請求worker thread創(chuàng)建子任務(wù)工作線程；

③wq.p（K1.OPCT）：壓入創(chuàng)建code thread線程操作OPCT，該操作只適用于存在控制與數(shù)據(jù)依賴關(guān)系對的后繼計(jì)算核心，如本例中的K2；

④ex.p（K1.OPR）：請求執(zhí)行申請計(jì)算資源的操作OPR；

⑤wq.p（K1.OPL）：壓入wq事件消息隊(duì)列中請求worker thread執(zhí)行計(jì)算任務(wù)；

⑥ex（K1.OPW）：阻塞等待worker thread線程執(zhí)行完計(jì)算任務(wù)；

⑦ex（K1.OPW）與ex（K2.OPW）：K1與K2都需要生成等待工作線程結(jié)束的代碼，以確保當(dāng)前已啟動的計(jì)算任務(wù)執(zhí)行完成，由于采用了SKMS－S執(zhí)行模式，則直接在當(dāng)前Kernel執(zhí)行處的最后位置生成K.OPW語句。

（2）worker thread：負(fù)責(zé)管理計(jì)算核心任務(wù)執(zhí)行過程，通過讀取并解析wq事件消息隊(duì)列中的事件消息來執(zhí)行相關(guān)操作。

①ex（K1i.OPST）：根據(jù)當(dāng)前計(jì)算核心子任務(wù)數(shù)目創(chuàng)建相應(yīng)數(shù)目的subworker thread用于管理子任務(wù)的實(shí)際執(zhí)行過程；

②sq.p（K1i.OPL）：向sq事件消息隊(duì)列中壓入子任務(wù)請求計(jì)算任務(wù)的操作OPL，如圖4與表3中標(biāo)記②所示；

③ex（K1.OPUD）：當(dāng)發(fā)現(xiàn)子任務(wù)執(zhí)行完畢時(shí)，主動更新當(dāng)前計(jì)算核心的輸出流節(jié)點(diǎn)信息與已完成的子任務(wù)信息；

④ex（K1.OPET）：當(dāng)更新完數(shù)據(jù)流信息時(shí)，worker thread主動退出工作線程，以響應(yīng)executor thread等待工作線程結(jié)束的阻塞操作ex（K1.OPW），使其可以正常繼續(xù)執(zhí)行。

（3）subworker thread：負(fù)責(zé)管理計(jì)算核心子任務(wù)的執(zhí)行過程，通過讀取解析sq事件消息隊(duì)列中的消息來執(zhí)行相關(guān)操作。

Table 3 Thread operations in the stream Eager transform technique表3 數(shù)據(jù)流Eager傳輸技術(shù)中各線程操作表

①ex（K1i.OPL）：根據(jù)分配的資源信息請求遠(yuǎn)程RMK執(zhí)行子任務(wù)；

②wq.p（K1i.OPF）：子任務(wù)計(jì)算完畢后將狀態(tài)反饋給workerthread，將OPF操作壓入wq隊(duì)列，該操作如圖4與表3中標(biāo)記③所示；

③ex（K1i.OPU）：當(dāng)子任務(wù)計(jì)算完畢后，其工作線程主動將計(jì)算完畢后的數(shù)據(jù)流狀態(tài)更新到主執(zhí)行線程（AK代碼）；

④ex（K1i.OPES）：若當(dāng)前計(jì)算核心K1采用數(shù)據(jù)流Eager傳輸技術(shù)，則對所有后繼Kernel（K2）執(zhí)行數(shù)據(jù)傳輸操作，發(fā)送數(shù)據(jù)到K2綁定的節(jié)點(diǎn)，完成后通過K2.cq.p（K1i.OPES）操作通知后繼Kernel的code thread進(jìn)一步處理。

（4）code thread：接收處理前驅(qū)Kernel發(fā)送來的數(shù)據(jù)狀態(tài)消息，若是OPES操作，說明前驅(qū)Kernel相應(yīng)的子任務(wù)計(jì)算完畢，且相應(yīng)數(shù)據(jù)已經(jīng)傳輸?shù)侥康墓?jié)點(diǎn)，則codethread更新相應(yīng)數(shù)據(jù)流狀態(tài)到全局信息中。

如上所述，分布式流體系結(jié)構(gòu)基于改進(jìn)的組織結(jié)構(gòu)可以有效支持?jǐn)?shù)據(jù)流Eager傳輸技術(shù)，盡可能使得通信與計(jì)算相互重疊，以減少程序等待數(shù)據(jù)傳輸?shù)臅r(shí)間。

4 實(shí)驗(yàn)驗(yàn)證

4.1 實(shí)驗(yàn)方法

本文在由10個(gè)節(jié)點(diǎn)組成的互連網(wǎng)絡(luò)上完成整個(gè)實(shí)驗(yàn)評估，其中，每個(gè)節(jié)點(diǎn)都是由一個(gè)多核CPU組成，實(shí)驗(yàn)平臺參數(shù)如表4所示。實(shí)驗(yàn)測試用例采用NAS Grid Benchmarks（NGB）3.1中的Visualization Pipe（VP）與Mixed Bag（MB）兩個(gè)用例，都是由單個(gè)NPB 實(shí)例求解器（BT、SP、LU、MG 或FT）組成的串行數(shù)據(jù)流處理模式，涉及到多個(gè)任務(wù)（計(jì)算核心函數(shù)）的交互與數(shù)據(jù)通信，代表了兩種典型的程序執(zhí)行模型。同時(shí)，它們在不同實(shí)例求解器之間使用MF 過濾器來轉(zhuǎn)換數(shù)據(jù)，其中每一個(gè)NPB實(shí)例求解器都必須按照順序依次執(zhí)行。本文選取兩個(gè)用例各自反復(fù)執(zhí)行兩次的數(shù)據(jù)流流程作為實(shí)驗(yàn)測試用例，每個(gè)實(shí)例求解器與過濾器都封裝為一個(gè)計(jì)算核心，計(jì)算核心均采用CPU 代碼，使用Brook＃語言移植到分布式流體系結(jié)構(gòu)上，在主程序中所有計(jì)算核心都是順序執(zhí)行的，并通過模擬的方法產(chǎn)生原始數(shù)據(jù)到達(dá)CMK1。每個(gè)測試用例執(zhí)行流程如圖5所示，計(jì)算核心之間的連線表示存在數(shù)據(jù)依賴關(guān)系。

Table 4 Parameter list of the experimental platform表4 實(shí)驗(yàn)平臺參數(shù)列表

Figure 5 Schematic diagrams of the computing process in test case VP and MB圖5 實(shí)驗(yàn)測試用例VP與MB的計(jì)算流程示意圖

本文在實(shí)驗(yàn)中通過測量wall－clock執(zhí)行時(shí)間記錄實(shí)驗(yàn)結(jié)果，其中，國際互聯(lián)網(wǎng)延時(shí)采用Internet Traffic Report網(wǎng)站統(tǒng)計(jì)中2015年五大洲延遲時(shí)間平均值100ms，國際互聯(lián)網(wǎng)帶寬Speedtest在2013年180多個(gè)國家與地區(qū)測量帶寬的30 天移動平均值13.98 Mbps進(jìn)行模擬，其余時(shí)間都采用實(shí)際測試的時(shí)間統(tǒng)計(jì)數(shù)據(jù)，包括任務(wù)計(jì)算執(zhí)行時(shí)間、軟件控制開銷等。

本文采用的實(shí)驗(yàn)基準(zhǔn)程序是未采用數(shù)據(jù)傳輸優(yōu)化技術(shù)的程序版本，即只使用了Lazy被動傳輸技術(shù)，只有當(dāng)任務(wù)執(zhí)行時(shí)才會請求數(shù)據(jù)傳輸。本文實(shí)驗(yàn)同時(shí)測試采用了數(shù)據(jù)流Eager傳輸優(yōu)化技術(shù)的程序版本，通過比較這兩個(gè)程序版本的執(zhí)行時(shí)間，以減少的執(zhí)行時(shí)間開銷來評估數(shù)據(jù)流Eager傳輸技術(shù)的有效性。在實(shí)驗(yàn)中本文采用了A與B兩個(gè)規(guī)模級別的測試集，以評估其在不同規(guī)模下的效果，如VP.A就表示采用了A級別測試集的VP測試用例。

4.2 實(shí)驗(yàn)結(jié)果

如圖6 所示，在采用數(shù)據(jù)流Eager傳輸技術(shù)后，在分布式流體系結(jié)構(gòu)下的各個(gè)測試用例在性能上都獲得了不同程度的提升，執(zhí)行時(shí)間開銷平均下降了19.58%。

Figure 6 Reduction percentage of the execution time of test benchmarks with the stream Eager transmission technique圖6 采用數(shù)據(jù)流Eager傳輸技術(shù)后測試用例執(zhí)行時(shí)間開銷下降的百分比

當(dāng)采用數(shù)據(jù)流Eager傳輸技術(shù)時(shí)，VP測試用例中的各個(gè)計(jì)算核心（BT、MG與FT）之間是以流水線方式形成計(jì)算流程的，其中，計(jì)算核心BT 與FT 可以在計(jì)算完畢后主動將數(shù)據(jù)傳輸給下一個(gè)循環(huán)執(zhí)行過程對應(yīng)的BT與FT，并與其他計(jì)算通信過程重疊起來，而不需要按照串行執(zhí)行順序等到執(zhí)行該計(jì)算核心時(shí)才啟動數(shù)據(jù)傳輸。在VP.A測試用例中，BT的執(zhí)行時(shí)間較長而FT 通信時(shí)間較短，而在VP.B測試用例中，BT的執(zhí)行時(shí)間較短而FT 通信時(shí)間較長，這使得兩者執(zhí)行時(shí)的性能關(guān)鍵路徑是不同的，VP.B中在FT通信未結(jié)束前就完成其他執(zhí)行路徑上的計(jì)算過程。這樣，VP.B在關(guān)鍵路徑上就不需要等待BT較長的數(shù)據(jù)傳輸時(shí)間，而在原來采用Lazy被動傳輸技術(shù)的版本中，第二個(gè)BT 的數(shù)據(jù)傳輸必須等待計(jì)算任務(wù)執(zhí)行時(shí)才能開始，從而需要的執(zhí)行時(shí)間相對較長。因此，VP.A與VP.B兩者的優(yōu)化效果差別較大，VP.A的時(shí)間開銷降低了11.18%，而VP.B降低了21.75%的時(shí)間開銷。

MB測試用例中的各個(gè)計(jì)算核心（LU、MG 與FT，其中MB.B還包含BT 計(jì)算核心）是以混合交叉方式傳輸數(shù)據(jù)形成計(jì)算過程的，其中，前一階段的計(jì)算核心結(jié)束后可以直接向后面依賴的兩個(gè)計(jì)算核心傳輸數(shù)據(jù)，同樣不需要等待后面兩個(gè)計(jì)算核心啟動時(shí)才被動傳輸數(shù)據(jù)。BT 與LU 的執(zhí)行時(shí)間較長，它們的通信時(shí)間可以被有效隱藏，而MG 與FT 的通信時(shí)間較長而執(zhí)行時(shí)間相對較短，因此，MB 的兩個(gè)循環(huán)執(zhí)行過程對應(yīng)的通信過程可以相互重疊，使得MB.A與MB.B的優(yōu)化效果相差不大，分別達(dá)到21.81%與23.58%。

通過以上分析可見，數(shù)據(jù)流Eager傳輸技術(shù)可以開發(fā)潛在的通信與其他操作的并行性，降低應(yīng)用程序執(zhí)行時(shí)間，提升計(jì)算性能。

5 結(jié)束語

互聯(lián)網(wǎng)較長的通信開銷等制約了應(yīng)用在分布式流體系結(jié)構(gòu)下的計(jì)算性能，為了充分挖掘計(jì)算與通信之間的并行性，本文提出了一種面向分布式流體系結(jié)構(gòu)的性能優(yōu)化技術(shù)，即數(shù)據(jù)流Eager傳輸技術(shù)，并在分布式體系結(jié)構(gòu)原型系統(tǒng)中實(shí)現(xiàn)了該技術(shù)。實(shí)驗(yàn)結(jié)果驗(yàn)證了該優(yōu)化技術(shù)的有效性，表明該優(yōu)化技術(shù)能夠顯著提高應(yīng)用的性能，具有良好的應(yīng)用前景。

下一步研究工作將針對通信帶寬等特點(diǎn)定制自適應(yīng)優(yōu)化策略，通過被動感知通信環(huán)境的方式改進(jìn)資源調(diào)度策略，引入人工智能算法，基于主動分析與被動感知策略相結(jié)合的方法來提高系統(tǒng)通信的效率。

本文的主要?jiǎng)?chuàng)新工作在第3節(jié)，描述了數(shù)據(jù)流Eager傳輸優(yōu)化技術(shù)的基本原理和設(shè)計(jì)實(shí)現(xiàn)機(jī)制。第2節(jié)概述了作者已提出的分布式流體系結(jié)構(gòu)，第4節(jié)和第5節(jié)分別是實(shí)驗(yàn)結(jié)果與結(jié)束語。

［1］ Mell P，Grance T.The NIST definition of cloud computing［R］.NIST，2011.

［2］ Dean J，Ghemawat S.MapReduce：Simplified data processing on large clusters ［J］.Communications of the ACM，2008，51（1）：107－113.

［3］ Zaharia M，Das T，Li H，et al.Discretized streams：Faulttolerant streaming computation at scale［C］∥Proc of the 24th ACM Symposium on Operating Systems Principles，2013：423－438.

［4］ Xin R S，Rosen J，Zaharia M，et al.Shark：SQL and rich analytics at scale［C］∥Proc of the 2013ACM SIGMOD International Conference on Management of DataACM，2013：13－24.

［5］ Thusoo A，Sarma J S，Jain N，et al.Hive－apetabyte scale data warehouse using Hadoop［C］∥Proc of IEEE 26th International Conference on Data Engineering，2010：996－1005.

［6］ Zaharia M，Chowdhury M，Das T，et al.Resilient distributed datasets：A fault－tolerant abstraction for in－Memory cluster computing［C］∥Proc of the 9th USENIX Conference on Networked Systems Design and Implementation，2012：2.

［7］ Zaharia M，Chowdhury M，F(xiàn)ranklin M J，et al.Spark：Cluster computing with working sets［C］∥Proc of the 2nd USENIX Conference on Hot Topics in Cloud Computing，2010：10.

［8］ Foster I，Kesselman C.The Grid 2：Blueprint for a new computing infrastructure［M］.New York：Elsevier，2003.

［9］ Chawathe Y，Ratnasamy S，Breslau L，et al.Making gnutella－like P2Psystems scalable［C］∥Proc of the 2003Conference on Applications，Technologies，Architectures，and Protocols for Computer Communications，2003：407－418.

［10］ Jeffers J，Reinders J.Intel Xeon Phi coprocessor high－performance programming［J］.San Francisco：Morgan Kaufmann，2013.

［11］ Xie M，Lu Y，Liu L，et al.Implementation and evaluation of network interface and message passing services for Tian－He－1Asupercomputer［C］∥Proc of the 2011 19th Annual IEEE Symposium on High Performance Interconnects，2011：78－86.

［12］ Xue W，Yang C，F(xiàn)u H，et al.Enabling and scaling aglobal shallow－water atmospheric model on Tianhe－2［C］∥Proc of the 2014IEEE 28th International Parallel and Distributed Processing Symposium，2014：745－754.