MIT6.824 Lab1 預熱

假設有個字符串：

var str = "The MapReduce library in the user program first splits the input files into M pieces of typically 16 megabytes to 64 megabytes (MB) per piece (controllable by the user via an optional parameter). It then starts up many copies of the program on a cluster of machines."
複製代碼

須要統計裏面每一個字母出現的次數。最直觀簡單的作法就是利用一個 map，從開始到末尾讀這個字符串，並把字母做爲 key，出現的次數做爲 value。Map 中包含 key 的時候，value + 1，Map 中沒有 key 的時候默認 1。最後讀完這個字符串就 OK。

var m = make(map[string]int)
temp := strings.Split(str, "")

for _, c := range temp {
    if !unicode.IsLetter([]rune(c)[0]) {
        continue
    }
    if count, ok := m[c]; ok {
        m[c] = count + 1
    } else {
        m[c] = 1
    }
}
複製代碼

[M:3 R:1 y:7 o:13 v:1 e:26 h:7 l:10 i:14 r:15 T:1 p:13 d:1 u:6 c:8 b:5 s:14 g:4 a:17 f:5 m:7 t:20 B:1 I:1 n:10]
複製代碼

在現實世界中，這個 str 可能很是巨大，因此有時候咱們須要將源文本拆分紅多個小的字符串，而後多個線程同時處理，每一個線程計算獲得當前的中間結果，最後合併到一塊兒。

上述的過程在函數式編程中能夠被抽象爲 Map 和 Reduce 兩個函數。其中 Map 函數是把一個數組的每一個元素按照相同的邏輯處理以後返回的結果，Reduce 函數是把全部元素整合起來獲得結果。一般這個兩個函數的參數都是函數，Map 的返回值通常也是數組，Reduce 的返回值多是各類類型。

爲了在單機上實現出併發處理的效果，能夠用 Go 自帶的 goroutine 來實現。下面把拆分的工做省略，直接進入主題

接下來用 4 個 goroutine 同時處理這些 string，每一個作 goroutine 利用 單機串行版 的邏輯，生產出一個小規模的中間內容。隨後把每一箇中間內容都整合起來獲得最終值。接下來須要考慮

• Go 天生支持 CSP 編程模型，因此利用 channel 作通訊沒有問題
• 是否有 data race

package main

import (
	"strings"
	"sync"
	"unicode"
)

type ResultMap struct {
	sync.Mutex
	result map[string]int
}

func main()  {
	str1 := "The MapReduce library in the user program first"
	str2 := "splits the input files into M pieces of typically 16 megabytes to 64 megabytes (MB)"
	str3 := "per piece (controllable by the user via an optional parameter)."
	str4 := "It then starts up many copies of the program on a cluster of machines."

	strs := []string {str1, str2, str3, str4}

	// 主線程須要阻塞直到全部的 reduce 都結束
	var waitGroup sync.WaitGroup
	waitGroup.Add(len(strs))

	c := make(chan map[string]int)

	res := new(ResultMap)
	res.result = make(map[string]int)

	for _, str := range strs {
		go doMap(str, c)
		go doReduce(c, res, &waitGroup)
	}

	waitGroup.Wait()

	sortPrintMap(res.result)

}

// 生產出對應的 kv 傳遞給 channel
func doMap(str string, c chan map[string]int) {
	temp := strings.Split(str, "")
	m := make(map[string]int)

	for _, c := range temp {
		if !unicode.IsLetter([]rune(c)[0]) {
			continue
		}
		if count, ok := m[c]; ok {
			m[c] = count + 1
		} else {
			m[c] = 1
		}
	}
	c <- m
}

// 合併
func doReduce(c chan map[string]int, res *ResultMap, group *sync.WaitGroup) {
	res.Lock()
	defer res.Unlock()
	for k, v := range <- c {
		if count, ok := res.result[k]; ok {
			res.result[k] = count + v
		} else {
			res.result[k] = v
		}
	}
	group.Done()
}
複製代碼

檢查一下結果 (Map 的 key 自己是無序的，這裏是排好序以後的）

[M:3 R:1 y:7 o:13 v:1 e:26 h:7 l:10 i:14 r:15 T:1 p:13 d:1 u:6 c:8 b:5 s:14 g:4 a:17 f:5 m:7 t:20 B:1 I:1 n:10]
複製代碼

結果無誤以後，這個問題能夠再深刻

• 上述的 reduce 和 map 是單機上的，之間的數據共享用了 channel，若是是物理隔離的場景下，如何用別的東西作數據共享？
• 任何一個子任務都有可能由於各類緣由掛掉，如何在某個子任務掛掉的狀況下，系統的準確性不受影響，甚至能自愈？
• 上述的 goroutine 在執行結束以後就會被調度器回收，但實際上由於 map 老是會比 reduce 先結束，那麼後期的過程實際上能夠有更多的 goroutine 能夠參與到 reduce 任務中 r 如何實現這種調度讓資源能夠被更加充分的利用？