一個(gè)簡(jiǎn)單的pipeline是如何構(gòu)建起來(lái)的？

續(xù)接上文，這一次我們來(lái)詳細(xì)了解下一個(gè)簡(jiǎn)單的pipeline是如何構(gòu)建起來(lái)的

》最簡(jiǎn)單的流水線

書接上文，一個(gè)最簡(jiǎn)單的流水線例子，這里對(duì)data_in打兩拍做輸出：

我們逐行分析pipeline里每一行代碼都干了什么。

》Stage分析

在第八行我們聲明了一個(gè)Stageable變量，如前文所述，Stageable變量并不會(huì)立即產(chǎn)生電路對(duì)象。

代碼第9行，我們創(chuàng)建一個(gè)Stage，Stage中的下面的代碼會(huì)通過(guò)調(diào)用Pipeline中的addStage函數(shù)向Pipeline中的StageSet添加當(dāng)前Stage：

if(_pip != null) {
_pip.addStage(this)
}

在stage0中，第11行用io.data_in.valid驅(qū)動(dòng)stage0中的internals.input.valid。而在第12行，從左向右看，this(payload)函數(shù)會(huì)調(diào)用Stage的apply函數(shù)：

def apply[T <: Data](key : Stageable[T]) : T = {
????apply(StageableKey(key.asInstanceOf[Stageable[Data]], null)).asInstanceOf[T]
}

其進(jìn)一步調(diào)用：

defapply(key : StageableKey):Data = {
internals.stageableToData.getOrElseUpdate(key, ContextSwapper.outsideCondScope{
key.stageable()//.setCompositeName(this, s"${key}")
})
}

可以看到，這里的主要作用是一StageableKey作為Key查詢internals.stageableToData中是否包含該key，如果有則返回其value，否則創(chuàng)建該key-value，并將value返回。

最終這里會(huì)返回一個(gè)UInt(8 bits)電路對(duì)象。并將io.data_in.payload賦值給該電路對(duì)象。此時(shí)，stage0中的internals.stageableToData中包含一個(gè)元素。

在第14行，我們創(chuàng)建了stage1，其例化時(shí)傳入了Connection，其會(huì)調(diào)用：

 if(_pip != null) {
_pip.addStage(this)
}

def chainConnect(connection: ConnectionLogic): Unit ={
_pip.connect(_pip.stagesSet.takeWhile(_ != this).last, this)(connection)
}

def this(connection: ConnectionLogic)(implicit_pip: Pipeline) {
this()
chainConnect(connection)
}

這里的的調(diào)用關(guān)系是：

當(dāng)前Pipeline的StageSet添加當(dāng)前Stage元素

通過(guò)調(diào)用pipeline的connect函數(shù)和StageSet中的最后一個(gè)元素建立連接關(guān)系：

defconnect(m : Stage, s : Stage)(logics : ConnectionLogic*) = {
val c = new ConnectionModel
connections += c
c.m = m
c.s = s
c.logics ++= logics
c
}

在connection里，創(chuàng)建了一個(gè)連接關(guān)系ConnectionModel，用于連接stage0中的output接口和stage1的input接口，采用M2S()方式進(jìn)行連接。

同樣，15行同樣stage2的創(chuàng)建亦如此。

代碼第16行則是用stage2的output.valid驅(qū)動(dòng)data_out的valid輸出。而代碼第17行通過(guò)同樣的方式創(chuàng)建了一個(gè)UInt(8 bits)電路對(duì)象，用來(lái)驅(qū)動(dòng)data_out.payload。此時(shí)stage2中的internals.stageableToData中包含一個(gè)元素。

最后調(diào)用pipeline的build函數(shù)進(jìn)行流水線的搭建。

》build分析

我們按片段來(lái)分析build中的代碼：

首先，上面第一行代碼將connection中的stage添加到stageSet。在執(zhí)行該行之前，stageSet中包含了stage0、stage1、stage2三個(gè)元素，而connection按照描述進(jìn)行map后得到兩個(gè)元素:(stage0,stage1),(stage1,stage2)。由于stageSet為L(zhǎng)inkedHashSet，故執(zhí)行完后依然是(stage0,stage1,stage2)。

第二行代碼則是獲取帶有slave驅(qū)動(dòng)的stage，這里即stage0(驅(qū)動(dòng)stage1)、stage1(驅(qū)動(dòng)stage2)。

第三行則是獲取沒(méi)有slave驅(qū)動(dòng)的stage，這里只有stage2。

第7～8行則是分別建立每個(gè)stage都是由誰(shuí)驅(qū)動(dòng)的映射存儲(chǔ)至stageMaster、而stage input的驅(qū)動(dòng)邏輯則存儲(chǔ)至stageDriver。同時(shí)這里也限制了每個(gè)stage，最多只能由一個(gè)驅(qū)動(dòng)邏輯。處理完后，stageMaster、stageDriver中存儲(chǔ)的元素分別為：

stageMaster:
stage0->ArrayBuffer()
stage1->ArrayBuffer(stage0)
stage2->ArrayBuffer(stage2)
stageDriver：
stage1->Connection(m=stage0,s=stage1,logic=M2S)
stage2->Connection(m=stage1,s=stage2,logic=M2S)

代碼13～16行由于我們并沒(méi)有使用到stageableResultingToData,這里可以暫時(shí)忽略，等后續(xù)章節(jié)再進(jìn)行討論。

val clFlush = mutable.LinkedHashMap[ConnectionLogic, Bool]()
val clFlushNext = mutable.LinkedHashMap[ConnectionLogic, Bool]()
val clFlushNextHit = mutable.LinkedHashMap[ConnectionLogic, Bool]()
val clThrowOne = mutable.LinkedHashMap[ConnectionLogic, Bool]()
val clThrowOneHit = mutable.LinkedHashMap[ConnectionLogic, Bool]()

這里聲明的變量我們這里都是用不上，可以暫時(shí)先不進(jìn)行關(guān)注。

def propagateData(key : StageableKey, stage : Stage): Boolean ={
if(stage.internals.stageableTerminal.contains(key)) returnfalse
stage.stageableToData.get(key) match {
caseNone => {
val hits = ArrayBuffer[Stage]()
for(m <- stageMasters(stage)){
????????if(propagateData(key, m)){
??????????stage.apply(key) //Force creation
??????????hits += m
????????}
??????}
??????hits.size match {
????????case?0?=> false
case1=> true
case2=> PendingError(s"$key at $stage has multiple drivers : ${hits.mkString(",")}"); false
}
}
caseSome(x) => true
}
}

for(stage <- stagesSet){
??for(key <- stage.stageableToData.keys){
????for(m <- stageMasters(stage)) {
??????propagateData(key, m);
????}
??}
}

這里就有點(diǎn)兒意思了對(duì)于StageSet中的每個(gè)stage中stageableToData中的每個(gè)元素，都會(huì)調(diào)用propgateData函數(shù)進(jìn)行處理。我們不妨先來(lái)看看此時(shí)各stage中的stageableToData中的元素：

stage0:
StageableKey(payload,null)->UInt(8)
stage1:
null
stage2:
StageableKey(payload,null)->UInt(8)

由于stage0沒(méi)有master，無(wú)需考慮，stage1中stageableToData為空，也可以跳過(guò)。那么來(lái)看stage2，此時(shí)調(diào)用propagateData傳入的參數(shù)為：

key=StageableKey(payload,null)->UInt(8)
m=stage1

進(jìn)入propagateData函數(shù)，由于stageableTerminal我們并未使用，繼續(xù)往下看，由于stage1中的stageableToData為空，故這里match匹配為None，此時(shí)會(huì)看stage1的master端口，此時(shí)嵌套調(diào)用propagateData，傳入?yún)?shù)為：

key=StageableKey(payload,null)->UInt(8)
m=stage0

由于stage0中的stageableToData包含該key，那么此時(shí)返回true退出，再次回到頭次調(diào)用的處理上。

由于嵌套返回true，那么此時(shí)會(huì)在stage1上調(diào)用apply函數(shù)，為stage1插入一個(gè)stageableKey。最終，各stage中stageableToData的結(jié)果為：

stage0:
StageableKey(payload,null)->UInt(8)
stage1:
StageableKey(payload,null)->UInt(8)
stage2:
StageableKey(payload,null)->UInt(8)

看到這里，是不是明白了一些門道了呢？pipeline自動(dòng)幫我們補(bǔ)齊了stage1中所依賴的元素，完成了payload從stage0到stage2中的傳輸元素補(bǔ)齊。

對(duì)于propagateRequirements函數(shù)，這里我們并用不上，這里我們先無(wú)需關(guān)注。

接下來(lái)看時(shí)internal connections:

這里我們近會(huì)用到s.output.valid:=s.input.valid，即將每個(gè)stage中的internal.output.valid由internal.input.valid進(jìn)行驅(qū)動(dòng)。

剩下的就是stage之間的連接了：

for(c<- connections){
??val stageables = (c.m.stageableToData.keys).filter(key => c.s.stageableToData.contains(key) && !c.m.stageableTerminal.contains(key))
var m= ConnectionPoint(c.m.output.valid, c.m.output.ready, stageables.map(c.m.outputOf(_)).toList)
for((l, id) <- c.logics.zipWithIndex){
????val s = if(l?== c.logics.last)
??????ConnectionPoint(c.s.input.valid, c.s.input.ready, stageables.map(c.s.stageableToData(_)).toList)
????else?{
??????ConnectionPoint(Bool(), (m.ready != null) generate Bool(), stageables.map(_.stageable.craft()).toList)
????}
????val area = l.on(m, s, clFlush(l), clFlushNext(l), clFlushNextHit(l), clThrowOne(l), clThrowOneHit(l))
????if(c.logics.size != 1)
??????area.setCompositeName(c, s"level_$id", true)
????else
??????area.setCompositeName(c, true)
????m?= s
??}
}

對(duì)于每個(gè)connection，首先是將master端和slave端stage共有的stageableToData給篩選到stageables中去，這里對(duì)應(yīng)的為：

Connection(m=stage0,s=stage1,logic=M2S):
StageableKey(payload,null)
Connection(m=stage1,s=stage2,logic=M2S):
StageableKey(payload,null)

接著創(chuàng)建ConnectionPoint，對(duì)應(yīng)的paylaod即為StageabelKey所對(duì)應(yīng)的電路對(duì)象，也就意味著:

Connection(m=stage0,s=stage1,logic=M2S): 
m=ConenctionPoint(stage0.out.valid,stage0.out.ready,stage0.stageableToData(StageableKey(payload,null)))
s=ConenctionPoint(stage1.in.valid,stage1.in.ready,stage1.stageableToData(StageableKey(payload,null)))
Connection(m=stage1,s=stage2,logic=M2S):
m=ConenctionPoint(stage1.out.valid,stage1.out.ready,stage1.stageableToData(StageableKey(payload,null)))
s=ConenctionPoint(stage2.in.valid,stage2.in.ready,stage2.stageableToData(StageableKey(payload,null)))

最終，調(diào)用M2S.on創(chuàng)建stage之間的連接關(guān)系：

這里我們只用到6～7行，建立起各stage之間的連接關(guān)系。

至此！完成整個(gè)流水線的創(chuàng)建。

》寫在最后

通過(guò)本篇，分析了一個(gè)簡(jiǎn)單的流水線在Pipeline中的創(chuàng)建實(shí)現(xiàn)，后續(xù)將陸續(xù)進(jìn)行更加復(fù)雜流水線的Demo及其背后自動(dòng)實(shí)現(xiàn)原理。

審核編輯：劉清