1.3 順序進程通訊 · golan基礎

[TOC] # 1.3 順序進程通訊 > 本節內容提供一個線上演講：[YouTube 在線](https://www.youtube.com/watch?v=Z8ZpWVuEx8c)，[Google Slides 講稿](https://changkun.de/s/csp/)。早在上個世紀七十年代，多核處理器還是一個科研主題，并沒有進入普通程序員的視野。 Tony Hoare 于 1977 年提出通信順序進程（CSP）理論，遙遙領先與他所在的時代。 CSP 的模型由并發執行的實體（線程或者進程）所組成，實體之間通過發送消息進行通信，這里發送消息時使用的就是通道（channel）。也就是我們常說的『 Don't communicate by sharing memory; share memory by communicating 』。多核處理器編程通常與操作系統研究、中斷處理、I/O 系統、消息傳遞息息相關當時涌現了很多不同的想法： * 信號量 semaphore \[Dijkstra, 1965\] * 監控 monitors \[Hoare, 1974\] * 鎖與互斥 locks/mutexes * 消息傳遞研究證明了消息傳遞與如今的線程與鎖是等價的 \[Lauer and Needham 1979\]。 ## 1.3.1 產生背景傳統程序設計中三種常見的構造結構：重復構造（for）、條件構造（if）、順序組合（;）除此之外，其他的一些結構： * Fortran: Subroutine * Algol 60: Procedure * PL/I: Entry * UNIX: Coroutine * SIMULA 64: Class * Concurrent Pascal: Process and monitor * CLU: Cluster * ALPHARD: Form * Hewitt: Actor 程序的演進史：https://spf13.com/presentation/the-legacy-of-go/ 并行性的引入： * 硬件：CDC 6600 并行單元 * 軟件：I/O 控制包、多編程操作系統處理器技術在多核處理器上提出一組自洽的處理器可以更加高效、可靠但為了使用機器的效能，必須在處理器之間進行通信與同步。為此也提出了很多方法 * 公共存儲的檢查與更新：Algol 68, PL/I, 各種不同的機器碼（開銷較大） * semaphore * events: PL/I * 條件臨界區 * monitors and queues: Concurrent Pascal * path expression 這么多方法并沒有一個統一的選擇標準。 C.A.R Hoare 1978 的萌芽論文，認為輸入輸出在一種模型編程語言中是基本要素。通信順序過程的并行組成。通信是 I/O。**簡潔設計、解決所有問題** * Dijkstra's guarded command * 條件分支 * 主要區別：若多個條件為真，則隨機選擇一個分支執行 * Dijkstra's parbegin parallel command * 啟動并發過程 * input and output command * 過程間通信手段 * 通信值存在復制 * 沒有緩存 * 指令推遲到其他過程能夠處理該消息 * input + guarded command * 等價于 Go 的 select 語句，條件分支僅當 input 源 ready 時開始執行 * 多個 input guards 同時就緒，只隨機選擇一個執行，其他 guards 沒有影響 * repetitive command * 循環 * 終止性取決于其所有 guards 是否已終止 * pattern-matching ## 1.3.2 設計細節 CSP 語言的結構非常簡單，極度的數學形式化、簡潔與優雅。將 Dijkstra 守護指令、`!`發送與`?`接受消息進行一般性的推廣： * `p!value`: 向過程`p`發送一個消息 * `p?var`: 從`p`接受一個值并存儲到`var` * `[A;B]`: A 運行后順序的運行 B * `[A||B]`: A 與 B 并行運行（組合） * `*[A]`: 循環運行 A * `[a -> A [] b -> B]`: 守護指令（如果`a`則`A`，否則如果`b`則`B`，但彼此并行）在這個語言中，通信是一種同步。每個命令可以成功或者失敗。整個語言包含： * 順序算符`;` * 并行算符`||` * 賦值算符`:=` * 輸入算符`?`（發送） * 輸出算符`!`（接受） * 選擇算符`□` * 守衛算符`→` * 重復算符`*` ### 程序結構 ~~~go <cmd> ::= <simple cmd> | <structured cmd> <simple cmd> ::= <assignment cmd> | <input cmd> | <output cmd> <structured cmd> ::= <alternative cmd> | <repetitive cmd> | <parallel cmd> <null cmd> ::= skip <cmd list> ::= {<declaration>; | <cmd>; } <cmd> ~~~ ### 并行指令 ~~~go <parallel cmd> ::= [<proc>{||<proc>}] <proc> ::= <proc label> <cmd list> <proc label> ::= <empty> | <identifier> :: | <identifier>(<label subscript>{,<label subscript>}) :: <label subscript> ::= <integer const> | <range> <integer constant> ::= <numeral> | <bound var> <bound var> ::= <identifier> <range> ::= <bound variable>:<lower bound>..<upper bound> <lower bound> ::= <integer const> <upper bound> ::= <integer const> ~~~ 舉例： ``` X (i : 1 .. n) :: CL ↑ ↑ ↑ ↑ ↑ identifier bound var lower bound upper bound cmd list ? X(1) :: CL1 || X(2) :: CL2 || … || X(n) :: CLn ``` ~~~go <parallel cmd> ::= [<proc>{||<proc>}] <proc> ::= <proc label> <cmd list> <proc label> ::= <empty> | <identifier> :: | <identifier>(<label subscript>{,<label subscript>}) :: <label subscript> ::= <integer const> | <range> <integer constant> ::= <numeral> | <bound var> <bound var> ::= <identifier> <range> ::= <bound variable>:<lower bound>..<upper bound> <lower bound> ::= <integer const> <upper bound> ::= <integer const> ~~~ 更多舉例： * `[cardreader ? cardimage || lineprinter ! lineimage]` * 兩個并行過程 * `[west::DISASSEMBLE || X::SQUASH || east::ASSEMBLE]` * 三個并行過程 * `[room::ROOM || fork(i:0..4)::FORK || phil(i:0..4)::PHIL]` * 十一個個并行過程，其中第二個和第三個并行過程分別包含五個并行的子過程 ### 賦值指令 ~~~go <assignment cmd> ::= <target var> := <expr> <expr> ::= <simple expr> | <structured expr> <structured expr> ::= <constructor> ( <expr list> ) <constructor> ::= <identifier> | <empty> <expr list> ::= <empty> | <expr> {, <expr> } <target var> ::= <simple var> | <structured target> <structured target> ::= <constructor> ( <target var list> ) <target var list> ::= <empty> | <target var> {, <target var> } ~~~ 舉例： ~~~go x := x + 1 // 普通的賦值 (x, y) := (y, x) // 普通的賦值 x := cons(left, right) // 結構體賦值 cons(left, right) := x // 如果 x 是一個結構體 cons(y, z)，則賦值成功 left:=y, right:=z，否則失敗 insert(n) := insert(2*x+1) // 等價于 n := 2*x + 1 c := P() // 空結構體, 或稱‘信號’ P() := c // 如果 c 同為信號，則無實際效果，否則失敗 insert(n) := has(n) // 不允許不匹配的結構體之間進行賦值 ~~~ ### 輸入輸出指令 ~~~go <input cmd> ::= <source> ? <target var> <output cmd> ::= <destination> ! <expr> <source> ::= <proc name> <destination> ::= <proc name> <proc name> ::= <identifier> | <identifier> ( <subscripts> ) <subscripts> ::= <integer expr> {, <integer expr> } ~~~ 舉例： ~~~go cardreader?cardimage // 類似于 cardimage <- cardreader lineprinter!lineimage // 類似于 lineprinter := <- lineimage X?(x,y) // 從過程 X 接受兩個值，并賦值給 x 和 y DIV!(3*a + b, 13) // 向過程 DIV 發送兩個值，分別為 3*a+b 和 13 console(i)?c // 向第 i 個 console 發送值 c console(j-1)!"A" // 向第 j-1 個 console 發送字符 "A" X(i)?V() // 從第 i 個 X 接受一個信號 V sem!P() // 向 sem 發送一個信號 P ~~~ 思考：根據例子 (3) 和 (4)，以下語句 ~~~ [X :: DIV!(3*a+b, 13) || DIV :: X?(x,y)] ~~~ 本質上是什么意思？ ### 選擇與重復指令 ~~~go <repetitive cmd> ::= * <alternative cmd> <alternative cmd> ::= [<guarded cmd> { □ <guarded cmd> }] <guarded cmd> ::= <guard> → <cmd list> | ( <range> {, <range> }) <guard> → <cmd list> <guard> ::= <guard list> | <guard list>;<input cmd> | <input cmd> <guard list> ::= <guard elem> {; <guard elem> } <guard elem> ::= <boolean expr> | <declaration> ~~~ 舉例： ~~~go // 如果某個條件成立，則執行 → 后的語句；若均成立則隨機選擇一個執行 [x ≥ y → m := x □ y ≥ x → m := y] // 依次檢查 content 中的值是否與 n 相同 i := 0; *[i < size; content(i) ≠ n → i := i+1] // 從 west 復制一個字符串并輸出到 east 中 *[c:character; west?c → east!c] ~~~~~~go <repetitive cmd> ::= * <alternative cmd> <alternative cmd> ::= [<guarded cmd> { □ <guarded cmd> }] <guarded cmd> ::= <guard> → <cmd list> | ( <range> {, <range> }) <guard> → <cmd list> <guard> ::= <guard list> | <guard list>;<input cmd> | <input cmd> <guard list> ::= <guard elem> {; <guard elem> } <guard elem> ::= <boolean expr> | <declaration> ~~~ 更多舉例： ~~~go // 監聽來自 10 個 console 的值，并將其發送給 X，當收到 sign off 時終止監聽 *[(i:1..10)continue(i); console(i)?c → X!(i,c); console(i)!ack(); continue(i) := (c ≠ sign off)] // insert 操作: 執行 INSERT // has 操作: 執行 SEARCH，并當 i < size 時，向 X 發送 true，否則發送 false *[n:integer; X?insert(n) → INSERT □ n:integer; X?has(n) → SEARCH; X!(i<size)] // V 操作: val++ // P 操作: val > 0 時, val--，否則失敗 *[X?V() → val := val+1 □ val > 0; Y?P() → val := val-1] ~~~ ## 1.3.3 協程 ### S31 COPY 編寫一個過程 X，復制從 west 過程輸出的字符到 east 過程 ~~~ COPY :: *[c:character; west?c → east!c] ~~~ ~~~go func S31_COPY(west, east chan rune) { for c := range west { east <- c } close(east) } ~~~ ### S32 SQUASH 調整前面的程序，替換成對出現的「\*\*」為「↑」，假設最后一個字符不是「\*」。 SQUASH :: \*\[c:character; west?c → \[ c != asterisk → east!c □ c = asterisk → west?c; \[ c != asterisk → east!asterisk; east!c □ c = asterisk → east!upward arrow \] \] \] 練習：調整上面的程序，處理最后以奇數個「\*」結尾的輸入數據。 ~~~ SQUASH_EX :: *[c:character; west?c → [ c != asterisk → east!c □ c = asterisk → west?c; [ c != asterisk → east!asterisk; east!c □ c = asterisk → east!upward arrow + ] □ east!asterisk ] ] ~~~ ``` func S32_SQUASH_EX(west, east chan rune) { for { c, ok := <-west if !ok { break } if c != '*' { east <- c } if c == '*' { c, ok = <-west if !ok { + east <- '*' break } if c != '*' { east <- '*' east <- c } if c == '*' { east <- '↑' } } } close(east) } ``` ### S33 DISASSEMBLE 從卡片盒中讀取卡片上的內容，并以流的形式將它們輸出到過程 X ，并在每個卡片的最后插入一個空格。 ~~~ DISASSEMBLE:: *[cardimage:(1..80)characters; cardfile?cardimage → i:integer; i := 1; *[i <= 80 → X!cardimage(i); i := i+1 ] X!space ] ~~~ ``` func S33_DISASSEMBLE(cardfile chan []rune, X chan rune) { cardimage := make([]rune, 0, 80) for tmp := range cardfile { if len(tmp) > 80 { cardimage = append(cardimage, tmp[:80]...) } else { cardimage = append(cardimage, tmp[:len(tmp)]...) } for i := 0; i < len(cardimage); i++ { X <- cardimage[i] } X <- ' ' cardimage = cardimage[:0] } close(X) } ``` ### S34 ASSEMBLE 將一串流式字符串從過程 X 打印到每行 125 個字符的 lineprinter 上。必要時將最后一行用空格填滿。 ~~~ ASSEMBLE:: lineimage:(1..125)character; i:integer, i:=1; *[c:character; X?c → lineimage(i) := c; [i <= 124 → i := i+1 □ i = 125 → lineprinter!lineimage; i:=1 ] ]; [ i = 1 → skip □ i > 1 → *[i <= 125 → lineimage(i) := space; i := i+1]; lineprinter!lineimage ] ~~~ ``` func S34_ASSEMBLE(X chan rune, lineprinter chan string) { lineimage := make([]rune, 125) i := 0 for c := range X { lineimage[i] = c if i <= 124 { i++ } if i == 125 { lineimage[i-1] = c lineprinter <- string(lineimage) i = 0 } } if i > 0 { for i <= 124 { lineimage[i] = ' ' i++ } lineprinter <- string(lineimage) } close(lineprinter) return } ``` ### S35 Reformat 從長度為 80 的卡片上進行讀取，并打印到長度為 125 個字符的 lineprinter 上。每個卡片必須跟隨一個額外的空格，最后一行須使用空格進行填充。 ~~~ REFORMAT:: [west::DISASSEMBLE || X::COPY || east::ASSEMBLE] ~~~ ``` func S35_Reformat(cardfile chan []rune, lineprinter chan string) { west, east := make(chan rune), make(chan rune) go S33_DISASSEMBLE(cardfile, west) go S31_COPY(west, east) S34_ASSEMBLE(east, lineprinter) } ``` ### S36 Conway's Problem 調整前面的程序，使用「↑」替換每個成對出現的「\*」。 ~~~ CONWAY:: [west::DISASSEMBLE || X::SQUASH || east::ASSEMBLE] ~~~ ``` func S35_ConwayProblem(cardfile chan []rune, lineprinter chan string) { west, east := make(chan rune), make(chan rune) go S33_DISASSEMBLE(cardfile, west) go S32_SQUASH_EX(west, east) S34_ASSEMBLE(east, lineprinter) } ``` ## 1.3.4 子程、數據表示與遞歸子程是一個與用戶過程并發執行的子過程： ~~~ [subr:SUBROUTINE || X::USER] SUBROUTINE::[X?(value params) → …; X!(result params)] USER::[ …; subr!(arguments); …; subr?(results)] ~~~ 數據表示可以視為一個具有多入口的子過程，根據 guarded command 進行分支選擇： ~~~ *[X? method1(value params) → … □ X? method2(value params) → … ] ~~~ 遞歸可以通過一個子程數組進行模擬，用戶過程（零號子程）向一號子程發送必要的參數，再從起接受遞歸后的結果： ~~~ [recsub(0)::USER || recsub(i:1..reclimit):: RECSUB] USER::recsub(1)!(arguments); … ; recsub(1)?(results); ~~~ ### S41 帶余除法編寫一個類型子程，接受一個正除數與被除數，返回其商和余數。 ~~~ [DIV::*[x,y:integer; X?(x,y)-> quot,rem:integer; quot := 0; rem := x; *[rem >= y -> rem := rem - y; quot := quot + 1;] X!(quot,rem) ] ||X::USER] ~~~ ``` func S41_DivisionWithRemainder(in chan S41_In, out chan S41_Out) { v := <-in x, y := v.X, v.Y quot, rem := 0, x for rem >= y { rem -= y quot++ } out <- S41_Out{quot, rem} } ``` ### S42 階乘給定一個上限，計算其階乘。 ~~~ [fac(i:1..limit):: *[n:integer;fac(i-1)?n -> [n=0 -> fac(i-1)!1 □ n>0 -> fac(i+1)!n-1; r:integer; fac(i+1)?r; fac(i-1)!(n*r) ]] || fac(0)::USER ] ~~~ ``` func S42_Factorial(fac []chan int, limit int) { for i := 1; i <= limit; i++ { go func(i int) { n := <-fac[i-1] if n == 0 { fac[i-1] <- 1 } else if n > 0 { // 注意，這里需要檢查 i 的上限 // 原始解法沒有對其進行檢查，如果用戶輸入等于或超過則無法終止程序 if i == limit { fac[i-1] <- n } else { fac[i] <- n - 1 r := <-fac[i] fac[i-1] <- n * r } } }(i) } } ``` ### S43 S44 S45 S46 集合實現一個集合的 insert 與 has 方法 ~~~ S:: content(0..99)integer; size:integer; size := 0; *[n:integer; X?has(n) -> SEARCH; X!(i<size) □ n:integer; X?insert(n) -> SEARCH; [i<size -> skip □i = size; size<100 -> content(size) := n; size := size+1 ]] SEARCH:: i:integer; i := 0; *[i<size; content(i) != n -> i:=i+1] ~~~ Go 實現：https://github.com/changkun/gobase/blob/f787593b4467793f8ee0b07583ea9ffde5adf2be/csp/csp.go#L392 ## 1.3.5 監控與調度監控可以被視為與多個用戶過程通信的單一過程，且總是能跟用戶過程進行通信。例如： ~~~ *[(i:1..100)X(i)?(value param) → …; X(i)!(results)] ~~~ 當兩個用戶過程同時選擇一個 X(i) 時，guarded cmd 保護了監控結果不會被錯誤的發送到錯誤的用戶過程中。 ### S51 Buffered Channel 構造一個帶緩沖的過程 X，用于平滑輸出的速度（即 buffered channel）。 ~~~ X:: buffer:(0..9)portion; in,out:integer; in:=0; out := 0; comment 0 <= out <= in <= out+10; *[in < out + 10; producer?buffer(in mod 10) -> in := in + 1 □ out < in; consumer?more() -> consumer!buffer(out mod 10); out := out + 1 ] ~~~ Go 實現：https://github.com/changkun/gobase/blob/f787593b4467793f8ee0b07583ea9ffde5adf2be/csp/csp.go#L609 ### S52 信號量實現證書信號量 S，在 100 個子過程間進行共享，每個過程可以通過 V 操作在信號量非正的情況下增加信號量。 ~~~ S::val:integer; val:=0; *[(i:1..100)X(i)?V()->val:=val+1 □ (i:1..100)val>0;X(i)?P()->val:=val-1] ~~~ Go 實現：https://github.com/changkun/gobase/blob/f787593b4467793f8ee0b07583ea9ffde5adf2be/csp/csp.go#L649 ### S53 哲學家進餐問題實現哲學家進餐問題。 ~~~ PHIL = *[...during i-th lifetime ... -> THINK; room!enter(); fork(i)!pickup();fork((i+1)mod5)!pickup(); EAT; fork(i)!putdown();fork((i+1)mod5)!putdown(); room!next()] FORK = *[phil(i)?pickup()->phil(i)?putdown() □ phil((i-1)mod5)?pickup()->phil((i-1)mod5)?putdown()] ROOM = occupancy:integer; occupancy := 0; *[(i:0..4)phil(i)?enter()->occupancy:=occupancy+1 □ (i:0..4)phil(i)?exit()->occupancy:=occupancy-1] [room::ROOM||fork(i:0..4)::FORK||phil(i:0..4)::PHIL] ~~~ Go 實現：https://github.com/changkun/gobase/blob/f787593b4467793f8ee0b07583ea9ffde5adf2be/csp/csp.go#L746 ## 1.3.6 算法 ### S61 Eratosthenes 素數篩法實現 Eratosthenes 素數篩法。 ~~~ [SIEVE(i:1..100):: p,mp:integer; SIEVE(i-1)?p; print!p; mp:=p; comment mp is a multiple of p; *[m:integer; SIEVE(i-1)?m-> *[m>mp->mp:=mp+p]; [m=mp->skip □ m<mp->SIEVE(i+1)!m ] ] || SIEVE(0)::print!2; n:integer; n:=3; *[n<10000->SIEVE(1)!n;n:=n+2] || SIEVE(101)::*[n:integer;SIEVE(100)?n->print!n] || print::*[(i:0..101)n:integer;SIEVE(i)?n->...] ] ~~~ Go 實現：https://github.com/changkun/gobase/blob/f787593b4467793f8ee0b07583ea9ffde5adf2be/csp/csp.go#L833 ### S62 矩陣乘法實現 3x3 矩陣乘法。 ~~~ [M(i:1..3,0)::WEST ||M(0,j:1..3)::NORTH ||M(i:1..3,4)::EAST ||M(4,j:1..3)::SOUTH ||M(i:1..3,j:1..3)::CENTER] NORTH = *[true -> M(1,j)!0] EAST = *[x:real; M(i,3)?x->skip] CENTER = *[x:real;M(i,j-1)?x-> M(i,j+1)!x;sum:real; M(i-1,j)?sum;M(i+1,j)!(A(i,j)*x+sum)] ~~~ Go 實現：https://github.com/changkun/gobase/blob/f787593b4467793f8ee0b07583ea9ffde5adf2be/csp/csp.go#L923 ## 1.3.7 總結 CSP 設計是 Tony Hoare 的早期提出的設計，與隨后將理論完整化后的 CSP（1985）存在兩大差異：缺陷1: 未對 channel 命名并行過程的構造具有唯一的名詞，并以一對冒號作為前綴：\[a::P || b::Q || … || c::R\] 在過程 P 中，命令 b!v 將 v 輸出到名為 b 的過程。該值由在過程 Q 中出現的命令 a?x 輸入。過程名稱對于引入它們的并行命令是局部的，并且組件過程間的通信是隱藏的。雖然其優點是不需要在語言中引入任何 channel 或者 channel 聲明的概念。缺點： 1. 子過程需要知道其使用過程的名稱，使得難以構建封裝性較好的子過程庫 2. 并行過程組合本質上是具有可變數量參數的運算，不能進行簡化（見 CSP 1985）缺陷2: 重復指令的終止性模糊重復指令默認當每個 guard 均已終止則指令中終止，這一假設過強。具體而言，對于 \*\[a?x → P □ b?x → Q □ …\] 要求當且僅當輸入的所有過程 a,b,… 均終止時整個過程才自動終止。缺點： 1. 定義和實現起來很復雜 2. 證明程序正確性的方法似乎比沒有簡單。一種可能的弱化條件為：直接假設子過程一定會終止。綜合來說，CSP 1978 中描述的編程語言（與 Go 所構建的基于通道的 channel/select 同步機制進行對比）： 1. channel 沒有被顯式命名 2. channel 沒有緩沖，對應 Go 中 unbuffered channel 3. buffered channel 不是一種基本的編程源語，并展示了一個使用 unbuffered channel 實現其作用的例子 4. guarded command 等價于 if 與 select 語句的組合，分支的隨機觸發性是為了提供公平性保障 5. guarded command 沒有對確定性分支選擇與非確定性（即多個分支有效時隨機選擇）分支選擇進行區分 6. repetitive command 的本質是一個無條件的 for 循環，但終止性所要求的條件太苛刻，不利于理論的進一步形式化 7. CSP 1978 中描述的編程語言對程序終止性的討論幾乎為零 8. 此時與 Actor 模型進行比較，CSP 與 Actor 均在實體間直接通信，區別在于 Actor 支持異步消息通信，而 CSP 1978 是同步通信 ## 進一步閱讀的參考文獻 * \[Hoare 78\] Hoare, C. A. R. (1978). Communicating sequential processes. Communications of the ACM, 21(8), 666–677. * \[Brookes 84\] S. D. Brookes, C. A. R. Hoare, and A. W. Roscoe. 1984. A Theory of Communicating Sequential Processes. J. ACM 31, 3 (June 1984), 560-599. * \[Hoare 85\] C. A. R. Hoare. 1985. Communicating Sequential Processes. Prentice-Hall, Inc., Upper Saddle River, NJ, USA. * \[Milner 82\] R. Milner. 1982. A Calculus of Communicating Systems. Springer-Verlag, Berlin, Heidelberg. * \[Fidge 94\] Fidge, C., 1994. A comparative introduction to CSP, CCS and LOTOS. Software Verification Research Centre, University of Queensland, Tech. Rep, pp.93-24. * \[Lauer and Needham 1979\] Hugh C. Lauer and Roger M. Needham. 1979. On the duality of operating system structures. SIGOPS Oper. Syst. Rev. 13, 2 (April 1979), 3-19.