8.5. 併發的循環

本節中，我們會探索一些用來在並行時循環迭代的常見併發模型。我們會探究從全尺寸圖片生成一些縮略圖的問題。gopl.io/ch8/thumbnail包提供了ImageFile函數來幫我們拉伸圖片。我們不會說明這個函數的實現，只需要從gopl.io下載它。

gopl.io/ch8/thumbnail

package thumbnail

// ImageFile reads an image from infile and writes
// a thumbnail-size version of it in the same directory.
// It returns the generated file name, e.g., "foo.thumb.jpg".
func ImageFile(infile string) (string, error)

下面的程序會循環迭代一些圖片文件名，併為每一張圖片生成一個縮略圖：

gopl.io/ch8/thumbnail

// makeThumbnails makes thumbnails of the specified files.
func makeThumbnails(filenames []string) {
    for _, f := range filenames {
        if _, err := thumbnail.ImageFile(f); err != nil {
            log.Println(err)
        }
    }
}

顯然我們處理文件的順序無關緊要，因為每一個圖片的拉伸操作和其它圖片的處理操作都是彼此獨立的。像這種子問題都是完全彼此獨立的問題被叫做易並行問題（譯註：embarrassingly parallel，直譯的話更像是尷尬並行）。易並行問題是最容易被實現成並行的一類問題（廢話），並且最能夠享受到併發帶來的好處，能夠隨著並行的規模線性地擴展。

下面讓我們並行地執行這些操作，從而將文件IO的延遲隱藏掉，並用上多核cpu的計算能力來拉伸圖像。我們的第一個併發程序只是使用了一個go關鍵字。這裡我們先忽略掉錯誤，之後再進行處理。

// NOTE: incorrect!
func makeThumbnails2(filenames []string) {
    for _, f := range filenames {
        go thumbnail.ImageFile(f) // NOTE: ignoring errors
    }
}

這個版本運行的實在有點太快，實際上，由於它比最早的版本使用的時間要短得多，即使當文件名的slice中只包含有一個元素。這就有點奇怪了，如果程序沒有併發執行的話，那為什麼一個併發的版本還是要快呢？答案其實是makeThumbnails在它還沒有完成工作之前就已經返回了。它啟動了所有的goroutine，每一個文件名對應一個，但沒有等待它們一直到執行完畢。

沒有什麼直接的辦法能夠等待goroutine完成，但是我們可以改變goroutine裡的代碼讓其能夠將完成情況報告給外部的goroutine知曉，使用的方式是向一個共享的channel中發送事件。因為我們已經確切地知道有len(filenames)個內部goroutine，所以外部的goroutine只需要在返回之前對這些事件計數。

// makeThumbnails3 makes thumbnails of the specified files in parallel.
func makeThumbnails3(filenames []string) {
    ch := make(chan struct{})
    for _, f := range filenames {
        go func(f string) {
            thumbnail.ImageFile(f) // NOTE: ignoring errors
            ch <- struct{}{}
        }(f)
    }
    // Wait for goroutines to complete.
    for range filenames {
        <-ch
    }
}

注意我們將f的值作為一個顯式的變量傳給了函數，而不是在循環的閉包中聲明：

for _, f := range filenames {
    go func() {
        thumbnail.ImageFile(f) // NOTE: incorrect!
        // ...
    }()
}

回憶一下之前在5.6.1節中，匿名函數中的循環變量快照問題。上面這個單獨的變量f是被所有的匿名函數值所共享，且會被連續的循環迭代所更新的。當新的goroutine開始執行字面函數時，for循環可能已經更新了f並且開始了另一輪的迭代或者（更有可能的）已經結束了整個循環，所以當這些goroutine開始讀取f的值時，它們所看到的值已經是slice的最後一個元素了。顯式地添加這個參數，我們能夠確保使用的f是當go語句執行時的“當前”那個f。

如果我們想要從每一個worker goroutine往主goroutine中返回值時該怎麼辦呢？當我們調用thumbnail.ImageFile創建文件失敗的時候，它會返回一個錯誤。下一個版本的makeThumbnails會返回其在做拉伸操作時接收到的第一個錯誤：

// makeThumbnails4 makes thumbnails for the specified files in parallel.
// It returns an error if any step failed.
func makeThumbnails4(filenames []string) error {
    errors := make(chan error)

    for _, f := range filenames {
        go func(f string) {
            _, err := thumbnail.ImageFile(f)
            errors <- err
        }(f)
    }

    for range filenames {
        if err := <-errors; err != nil {
            return err // NOTE: incorrect: goroutine leak!
        }
    }

    return nil
}

這個程序有一個微妙的bug。當它遇到第一個非nil的error時會直接將error返回到調用方，使得沒有一個goroutine去排空errors channel。這樣剩下的worker goroutine在向這個channel中發送值時，都會永遠地阻塞下去，並且永遠都不會退出。這種情況叫做goroutine洩露（§8.4.4），可能會導致整個程序卡住或者跑出out of memory的錯誤。

最簡單的解決辦法就是用一個具有合適大小的buffered channel，這樣這些worker goroutine向channel中發送錯誤時就不會被阻塞。（一個可選的解決辦法是創建一個另外的goroutine，當main goroutine返回第一個錯誤的同時去排空channel。）

下一個版本的makeThumbnails使用了一個buffered channel來返回生成的圖片文件的名字，附帶生成時的錯誤。

// makeThumbnails5 makes thumbnails for the specified files in parallel.
// It returns the generated file names in an arbitrary order,
// or an error if any step failed.
func makeThumbnails5(filenames []string) (thumbfiles []string, err error) {
    type item struct {
        thumbfile string
        err       error
    }

    ch := make(chan item, len(filenames))
    for _, f := range filenames {
        go func(f string) {
            var it item
            it.thumbfile, it.err = thumbnail.ImageFile(f)
            ch <- it
        }(f)
    }

    for range filenames {
        it := <-ch
        if it.err != nil {
            return nil, it.err
        }
        thumbfiles = append(thumbfiles, it.thumbfile)
    }

    return thumbfiles, nil
}

我們最後一個版本的makeThumbnails返回了新文件們的大小總計數（bytes）。和前面的版本都不一樣的一點是我們在這個版本里沒有把文件名放在slice裡，而是通過一個string的channel傳過來，所以我們無法對循環的次數進行預測。

為了知道最後一個goroutine什麼時候結束（最後一個結束並不一定是最後一個開始），我們需要一個遞增的計數器，在每一個goroutine啟動時加一，在goroutine退出時減一。這需要一種特殊的計數器，這個計數器需要在多個goroutine操作時做到安全並且提供在其減為零之前一直等待的一種方法。這種計數類型被稱為sync.WaitGroup，下面的代碼就用到了這種方法：

// makeThumbnails6 makes thumbnails for each file received from the channel.
// It returns the number of bytes occupied by the files it creates.
func makeThumbnails6(filenames <-chan string) int64 {
    sizes := make(chan int64)
    var wg sync.WaitGroup // number of working goroutines
    for f := range filenames {
        wg.Add(1)
        // worker
        go func(f string) {
            defer wg.Done()
            thumb, err := thumbnail.ImageFile(f)
            if err != nil {
                log.Println(err)
                return
            }
            info, _ := os.Stat(thumb) // OK to ignore error
            sizes <- info.Size()
        }(f)
    }

    // closer
    go func() {
        wg.Wait()
        close(sizes)
    }()

    var total int64
    for size := range sizes {
        total += size
    }
    return total
}

注意Add和Done方法的不對稱。Add是為計數器加一，必須在worker goroutine開始之前調用，而不是在goroutine中；否則的話我們沒辦法確定Add是在"closer" goroutine調用Wait之前被調用。並且Add還有一個參數，但Done卻沒有任何參數；其實它和Add(-1)是等價的。我們使用defer來確保計數器即使是在出錯的情況下依然能夠正確地被減掉。上面的程序代碼結構是當我們使用併發循環，但又不知道迭代次數時很通常而且很地道的寫法。

sizes channel攜帶了每一個文件的大小到main goroutine，在main goroutine中使用了range loop來計算總和。觀察一下我們是怎樣創建一個closer goroutine，並讓其在所有worker goroutine們結束之後再關閉sizes channel的。兩步操作：wait和close，必須是基於sizes的循環的併發。考慮一下另一種方案：如果等待操作被放在了main goroutine中，在循環之前，這樣的話就永遠都不會結束了，如果在循環之後，那麼又變成了不可達的部分，因為沒有任何東西去關閉這個channel，這個循環就永遠都不會終止。

圖8.5 表明了makethumbnails6函數中事件的序列。縱列表示goroutine。窄線段代表sleep，粗線段代表活動。斜線箭頭代表用來同步兩個goroutine的事件。時間向下流動。注意main goroutine是如何大部分的時間被喚醒執行其range循環，等待worker發送值或者closer來關閉channel的。

練習 8.4： 修改reverb2服務器，在每一個連接中使用sync.WaitGroup來計數活躍的echo goroutine。當計數減為零時，關閉TCP連接的寫入，像練習8.3中一樣。驗證一下你的修改版netcat3客戶端會一直等待所有的併發“喊叫”完成，即使是在標準輸入流已經關閉的情況下。

練習 8.5： 使用一個已有的CPU綁定的順序程序，比如在3.3節中我們寫的Mandelbrot程序或者3.2節中的3-D surface計算程序，並將他們的主循環改為併發形式，使用channel來進行通信。在多核計算機上這個程序得到了多少速度上的改進？使用多少個goroutine是最合適的呢？