MIT6.824 lab2 實驗反思

0. 實驗文檔

6.5840 Lab 2: Key/Value Server

1. lab——，啟動——

1.1 需求描述

本實驗分為4個部分。

基于可靠信道的 Key/value server

• 需要實現一個基本的 Key/value 服務器和使用 Key/value 服務器的客戶端代理（clerk），不考慮網絡導致的信息丟失。
• 主要需求如下：
  1. 服務器（server）：
     • 維護一個內存中的映射（map），記錄每個鍵（key）對應的（值（value），版本號（version））元組。鍵和值都是字符串類型。
     • 處理兩個RPC請求：Put(key, value, version) 和 Get(key)。
     • Put(key, value, version)：只有當傳入的版本號與服務器上該鍵的當前版本號匹配時，才更新鍵的值并增加版本號（版本號加1）。如果版本號匹配，更新值并增加版本號，返回成功；如果不匹配，返回錯誤rpc.ErrVersion。如果要創建新鍵，客戶端傳入版本號0，服務器會將該鍵的版本號設為1（因為版本號從1開始）。如果版本號大于0且鍵不存在，返回rpc.ErrNoKey。
     • Get(key)：返回鍵的當前值和版本號。如果鍵不存在，返回錯誤rpc.ErrNoKey。
     • 確保操作的線性一致性（linearizability）：對于非并發操作，每個操作都能看到前一個操作完成后的狀態；對于并發操作，最終狀態和返回值必須與某個順序執行（一次一個操作）的結果相同。
  2. 客戶端代理（clerk）：
     • 提供Put和Get方法，用于向服務器發送RPC請求。
     • 在可靠信道下，不需要處理網絡丟包問題，即每個RPC請求都能到達服務器，并且每個回復都能到達客戶端。
     • 在客戶端代碼中，發送RPC請求并接收回復，根據回復返回相應結果（值或錯誤）。

基于 Key/value server 和可靠信道的分布式鎖

• 使用上面實現的Key/value服務器（在可靠信道下）來實現一個分布式鎖。
• 鎖提供兩個方法：Acquire和Release。
• 需求如下：
  • 同一時刻只有一個客戶端能夠成功獲取鎖（即Acquire返回成功），其他客戶端必須等待直到鎖被釋放。
  • 每個鎖客戶端需要一個唯一的標識符（可通過kvtest.RandValue(8)生成隨機字符串）。
  • 鎖服務使用一個特定的鍵（通過MakeLock的參數l指定）來存儲鎖的狀態。需要設計鎖狀態的具體表示（例如，可以存儲持有鎖的客戶端標識和當前鎖的版本號）。
  • 如果客戶端崩潰（但在本實驗中不考慮此情況），鎖可能永遠不會被釋放。但在本實驗中，假設客戶端不會崩潰。

基于不可靠信道的 Key/value server

• 在不可靠信道下擴展Key/value服務器客戶端代理（clerk）的功能，以處理網絡丟包、延遲和重排序等問題。
• 主要需求如下：
  • 客戶端必須重試RPC請求直到收到回復（因為網絡可能丟棄請求或回復）。
  • 對于Get請求，重試是安全的，因為它是只讀操作。
  • 對于Put請求，由于條件更新的特性（基于版本號），重試也是安全的，因為重復的Put請求（相同的版本號）在第一次成功執行后，第二次會因版本號不匹配而返回rpc.ErrVersion。
  • 處理重試時可能出現的特殊情況：當客戶端重試一個 Put 請求后，如果服務器返回 rpc.ErrVersion，客戶端必須區分是第一次請求就失敗（返回 rpc.ErrVersion）還是重試時失敗（返回 rpc.ErrMaybe，表示不確定是否執行）。具體來說：
    • 如果是第一次發送請求（未重試）就收到rpc.ErrVersion，則向應用層返回rpc.ErrVersion（表示未執行）。
    • 如果是重試后收到rpc.ErrVersion，則向應用層返回rpc.ErrMaybe（表示可能執行了，也可能沒執行）。
  • 在重試之間應等待一段時間（例如100毫秒），避免過度重試。

基于 Key/value server 和不可靠信道的分布式鎖

• 使用上面擴展的Key/value服務器再次實現分布式鎖，確保在不可靠信道下也能正確工作。

• 需求與可靠信道下的分布式鎖相同，但需要處理網絡丟包和重試帶來的問題。

• 特別注意：鎖的Acquire和Release操作需要能夠處理Put返回的rpc.ErrMaybe錯誤，并確保鎖的正確性（即同一時刻只有一個客戶端持有鎖）。

1.2 詳細實現方案

1.2.1 基于可靠信道的 Key/value server 的實現方案

這部分相對來說比較簡單。首先需要明確的是，基于 Go 標準庫的 net/rpc 包實現的 RPC 服務對于每一個到來的 RPC 請求都會起一個 goroutine 進行處理。而這里需要同時實現一個 RPC 服務和一個客戶端代理 clerk，所以很容易想到可以在 server.go 中維護一個 map，記錄 <key, value, version> 三元組。同時考慮到要支持客戶端的并發，所以要對這個 map 加鎖，避免競態條件。

Put 方法

根據需求描述，Put 方法在版本號匹配的時候應該成功插入三元組，在不匹配的時候應該返回 rpc.ErrVersion，如果第一次插入而版本號不是 0 就要返回 rpc.ErrNoKey。顯然 Put 方法實現如下：

// Update the value for a key if args.Version matches the version of
// the key on the server. If versions don't match, return ErrVersion.
// If the key doesn't exist, Put installs the value if the
// args.Version is 0, and returns ErrNoKey otherwise.
func (kv *KVServer) Put(args *rpc.PutArgs, reply *rpc.PutReply) {
	// Your code here.
	key, val, ver := args.Key, args.Value, args.Version
	reply.Err = rpc.OK

	kv.mu.Lock()
	defer kv.mu.Unlock()

	v, ok := kv.table[key]
	if !ok {
		if ver != 0 {
			reply.Err = rpc.ErrNoKey
			return
		}

		kv.table[key] = &ValVerPair{
			value:   val,
			version: ver + 1,
		}

		return
	}

	if v.version != ver {
		reply.Err = rpc.ErrVersion
		return
	}

	v.value = val
	v.version++
}

因為這里需要對 map 做讀寫操作，所以需要加鎖。

Get 方法

對于 Get 方法，返回 value 和 version。如果 map 中沒有找到對應的 key，返回 rpc.ErrNoKey。

// Get returns the value and version for args.Key, if args.Key
// exists. Otherwise, Get returns ErrNoKey.
func (kv *KVServer) Get(args *rpc.GetArgs, reply *rpc.GetReply) {
	// Your code here.
	key := args.Key
	reply.Err = rpc.OK

	kv.mu.Lock()
	defer kv.mu.Unlock()

	v, ok := kv.table[key]
	if !ok {
		reply.Err = rpc.ErrNoKey
		return
	}

	reply.Value = v.value
	reply.Version = v.version
}

這里即使Get是只讀操作，也需要加鎖。以下是詳細原因：

1. 防止數據競爭（Data Race）

   • Go 的 map 在并發讀寫時會導致運行時 panic（fatal error: concurrent map read and map write）。

   • 即使只有多個 Get 并發讀，如果同時有一個 Put 在寫（修改或刪除鍵值對），也會觸發數據競爭。

   • 加鎖確保對 kv.table 的訪問是互斥的，避免程序崩潰。

2. 保證讀取的一致性

   • 假設 Get 不加鎖，而 Put加鎖：

     • 當 Put 修改 kv.table[key] 時（例如更新值或版本號），Get 可能讀到部分更新的臟數據（如舊值配新版本）。

     • 鎖確保 Get 讀取時，kv.table[key] 處于穩定狀態（例如值和版本號是原子更新的）。

3. 避免過時數據（Stale Read）

   • 如果 Get 不加鎖，可能讀到已被修改但未完成寫入的數據（例如 Put 只更新了值，尚未更新版本號）。

   • 加鎖后，Get 會等待 Put 完成整個操作，確保讀取到最新提交的數據。

而客戶端代理 clerk 的操作就相對簡單了，通過 rpc 調用 server 這邊提供的 Put 方法和 Get 方法執行對應的功能就行了。

1.2.2 基于 Key/value server 和可靠信道的分布式鎖的實現方案

所謂分布式鎖，是指當某個共享資源被集群中的多臺機器共享的時候，我們需要對它加一個全局的鎖，以保證不會在機器層面出現競態條件。

我們可以依賴于 value + version 兩個字段實現一個分布式鎖。實現思路如下：

• 當某個機器拿到這個共享資源的時候，要對其加鎖。加鎖就是往 Key/value server 中寫入一個鍵值對，key 是鎖的名稱，value 是持有鎖的機器ID，version 是當前鎖的版本號，用于比對加鎖釋放鎖操作是否合法。
• 當機器對這個資源的操作結束以后，就要及時釋放鎖。釋放鎖可以把 value 字段記錄的 ID 改成空字符串，同時更新 version 字段。這樣當另外一個機器要對其加鎖的時候，首先驗證鎖是否被其他機器持有，沒有的話再根據獲得的版本號寫入自己的 ID 到 value，這樣就成功加鎖了。

理清楚兩個操作具體做的事情以后，不難寫出下面的代碼：

func (lk *Lock) Acquire() {
	// Your code here
	delay := lk.backoff

	for {
		val, ver, err := lk.ck.Get(lk.key)
		if err != rpc.ErrNoKey && val != "" {
			delay = min(lk.maxBackoff, 2*delay)
			time.Sleep(delay)

			continue
		}

		err = lk.ck.Put(lk.key, lk.id, ver)
		if err == rpc.ErrVersion || err == rpc.ErrNoKey {
			delay = min(lk.maxBackoff, 2*delay)
			time.Sleep(delay)

			continue
		}

		break
	}
}

func (lk *Lock) Release() {
	// Your code here
	delay := lk.backoff

	for {
		val, ver, err := lk.ck.Get(lk.key)
		if err == rpc.ErrNoKey || val == "" {
			delay = min(lk.maxBackoff, 2*delay)
			time.Sleep(delay)

			continue
		}

		err = lk.ck.Put(lk.key, "", ver)
		if err == rpc.ErrVersion || err == rpc.ErrNoKey {
			delay = min(lk.maxBackoff, 2*delay)
			time.Sleep(delay)

			continue
		}

		break
	}
}

1.2.3 基于不可靠信道的 Key/value server 的實現方案

不可靠信道下，可能出現網絡丟包、延遲和重排序等問題，但是我們要保證 Key/value server 不受這些問題的影響。其實核心就是一個點——在沒有得到確切回復的情況下重試。

對于 Get 方法，這相當好實現，只要寫一個循環，在得到確切結果之前重復請求，設置好間隔時間就行了。但是對于 Put 方法來說，就要稍微復雜一些。Put 方法在調用過程中可能出現的情況是：鍵值對已經存放到服務器里面了，并且服務器也正確返回了消息，但是這個消息丟包了。那下一次重試過程中，Put 方法仍然使用的是舊的版本號，這個時候就會返回版本號不匹配的錯誤。但是實際上我們是正確存放了結果的。那么針對這個情景，實驗文檔給出的方案是，在不能確定是否插入鍵值對的時候，返回 ErrMaybe，交由應用層進行處理。

我們可以在Put 方法中維護一個計數器，計數重試次數。如果是第一次嘗試或者返回的結果是OK，就正常返回；否則返回 ErrMaybe。所以 Put 方法的實現如下：

// Put updates key with value only if the version in the
// request matches the version of the key at the server.  If the
// versions numbers don't match, the server should return
// ErrVersion.  If Put receives an ErrVersion on its first RPC, Put
// should return ErrVersion, since the Put was definitely not
// performed at the server. If the server returns ErrVersion on a
// resend RPC, then Put must return ErrMaybe to the application, since
// its earlier RPC might have been processed by the server successfully
// but the response was lost, and the Clerk doesn't know if
// the Put was performed or not.
//
// You can send an RPC with code like this:
// ok := ck.clnt.Call(ck.server, "KVServer.Put", &args, &reply)
//
// The types of args and reply (including whether they are pointers)
// must match the declared types of the RPC handler function's
// arguments. Additionally, reply must be passed as a pointer.
func (ck *Clerk) Put(key, value string, version rpc.Tversion) rpc.Err {
	// You will have to modify this function.
	delay := ck.backoff
	var attempt int

	for {
		attempt++
		args := rpc.PutArgs{
			Key:     key,
			Value:   value,
			Version: version,
		}

		reply := rpc.PutReply{}

		ok := ck.clnt.Call(ck.server, "KVServer.Put", &args, &reply)
		if ok {
			if attempt == 1 || reply.Err == rpc.OK { // first call or clearly success
				return reply.Err
			}
			return rpc.ErrMaybe
		}

		time.Sleep(delay)
		delay = min(delay*2, ck.maxBackoff)
	}
}

1.2.4 基于 Key/value server 和不可靠信道的分布式鎖的實現方案

這里的分布式鎖其實就對應上文中提到的應用層。在加鎖釋放鎖的時候，Put 方法可能會返回 ErrMaybe 這種錯誤。對于這種錯誤，我們采取最直接的解決方案——重試。實現代碼如下：

func (lk *Lock) Acquire() {
	// Your code here
	delay := lk.backoff

	for {
		val, ver, err := lk.ck.Get(lk.key)
		if err != rpc.ErrNoKey && val != "" {
			time.Sleep(delay)
			delay = min(lk.maxBackoff, 2*delay)

			continue
		}

		err = lk.ck.Put(lk.key, lk.id, ver)
		if err == rpc.ErrVersion || err == rpc.ErrNoKey {
			time.Sleep(delay)
			delay = min(lk.maxBackoff, 2*delay)

			continue
		}

		if err == rpc.ErrMaybe {
			value, _, e := lk.ck.Get(lk.key)
			if e != rpc.OK || value != lk.id {
				time.Sleep(delay)
				delay = min(lk.maxBackoff, 2*delay)

				continue
			}
		}

		break
	}
}

func (lk *Lock) Release() {
	// Your code here
	delay := lk.backoff

	for {
		val, ver, err := lk.ck.Get(lk.key)
		if err == rpc.ErrNoKey || val == "" {
			time.Sleep(delay)
			delay = min(lk.maxBackoff, 2*delay)

			continue
		}

		if val != lk.id { // if current goroutine doesn't get the lock, exit straightly
			return
		}

		err = lk.ck.Put(lk.key, "", ver)
		if err == rpc.ErrVersion || err == rpc.ErrNoKey {
			time.Sleep(delay)
			delay = min(lk.maxBackoff, 2*delay)

			continue
		}

		if err == rpc.ErrMaybe {
			value, _, e := lk.ck.Get(lk.key)
			if e != rpc.OK || value != "" {
				time.Sleep(delay)
				delay = min(lk.maxBackoff, 2*delay)

				continue
			}
		}

		break
	}
}

2. 總結

本實驗旨在實現一個簡單的 Key/value server 和一個簡單的分布式鎖。

對于本實驗來說，有以下兩個難點：

• 在 Key/value server 不支持刪除操作的情況下，給出分布式鎖的實現方案
• 基于不可靠信道的改進方案

對于絕大多數問題，它都不是難，而是復雜。難是無從下手，復雜是好像有那么一點思路，但是細枝末節的地方不好搞定。那么面對復雜問題的時候，我們可以把問題拆解成一個一個的簡單問題，就像這個實驗一樣，逐個擊破，就一定能夠成功。那么對于剩下的那些難題，就忘了它罷，難道這世上所有的問題都能夠解決嗎？