Project Overview

Building a fault-tolerant key/value storage service using Raft that supports Get(key), Put(key, value) and Append(key, value).

Implementation

Architecture

1

client - clerk - kvserver(leader) - raft

A client connects to a clerk. The clerk sends requests to the kvserver which has a leader Raft instance.

The files are client.go, server.go.

Client

A client does the following things:

1. Sending the request to the kvserver.
2. Handling the reply. When something goes wrong, such as requesting to a follower kvserver or timeout, it resends the request until succeeds.

The client could only receive an OK when it communicates to the kvserver who holds the leader raft instance. Otherwise it has to communicates to the next kvserver in its list until finding the leader kvserver.

From the behavior of a client, it’s easy to determine that a client(clerk) holds a list of servers (servers []*labrpc.ClientEnd) and the index of the leader (leader int).

Other properties? This will be discussed in the following part of this passage.

Server

First of all, a server holds a raft instance and its applyCh.

When a log entry is committed by raft cluster, a message will be sent to applyCh. So the kvserver need to read messages repeatedly from the channel (for applyMsg := range kv.applyCh) and handle it. This should be conducted in a goroutine.

When receiving a MODERATE message, the server should save the state to itself. A hash map (kv map[string]string) is okey. According to different types of commands, update the entries in the hash map.

Order Keeping

The client sends requests in serial.

In the client side, it keeps clientId and seq:

• clientId is generated randomly
• seq is an auto-increment sequence number starts from 0

In the server side, it keeps a lastApplied map[int64]int to remember the executed sequence number of each client. The key of lastApplied is the client id and the value is the last applied sequence number of this client.

When receving messages from the applyCh, the kvserver checks whether the seq is EXACTLY eligible: op.Seq == kv.lastApplied[op.ClientId] + 1. If the current sequence number is too small, it means this message is out-to-date, just do nothing to kv and notify. If the current sequence number is too big, return directly (after timeout the client will resend one).

Update lastApplied after the eligible execution:

1

kv.lastApplied[op.ClientId] = op.Seq

Notifying Mechanism

For each request from KVServer#Get or KVServer#PutAppend, a channel will be set in its lifecycle.

We use a map as kvserver’s property. The key is ClientId-Seq to unique every request. The value is a NotifyCh.

1
2
3
4
5

type NotifyCh struct {
    // just use one of them
	getCh chan string
	putAppendCh chan struct{}
}

In kvserver’s Get and PutAppend methods, creating the channel like this:

1
2
3
4
5

pendingKey := fmt.Sprintf("%d-%d", args.ClientId, args.Seq)
ch := &NotifyCh{getCh: make(chan string, 1)}
kv.mu.Lock()
kv.pendingCh[pendingKey] = ch
kv.mu.Unlock()

In kvserver’s apply method, notify like this:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

switch op.Type {
case "Get":
	value := kv.kv[op.Key]
	if ch, ok := kv.pendingCh[pendingKey]; ok {
		ch.getCh <- value
	}
	break
case "Put":
	kv.kv[op.Key] = op.Value
    if ch, ok := kv.pendingCh[pendingKey]; ok {
	ch.putAppendCh <- struct{}{}
	}
	break
case "Append":
	kv.kv[op.Key] += op.Value
   	if ch, ok := kv.pendingCh[pendingKey]; ok {
		ch.putAppendCh <- struct{}{}
	}
	break
}

And in kvserver’s Get and PutAppend method, waiting like this:

1
2
3
4
5
6
7

select {
case value := <-ch.getCh:
	reply.Err = OK
	reply.Value = value
case <-time.After(timeout):
	reply.Err = ErrTimeout
}

At last, the channel should be deleted to free memory:

1
2
3

kv.mu.Lock()
delete(kv.pendingCh, pendingKey)
kv.mu.Unlock()

Concurrency Control

The client does need concurrency control as it sends requests to the kvserver in single thread.

Lock is necessary when a kvserver modifies its properties. We try to lock as little as possible. In one hand, it will lead to better performance. In another hand, it reduces the risk of dead lock.

Note that any mutex must be unlocked before the thread is blocked, such as waiting a message from a channel.

Test

Passed.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

$ go test -run 3A
Test: one client (3A) ...
  ... Passed --  15.1  5 17220 1434
Test: many clients (3A) ...
  ... Passed --  15.2  5 50955 5038
Test: unreliable net, many clients (3A) ...
  ... Passed --  16.2  5 10606 1430
Test: concurrent append to same key, unreliable (3A) ...
  ... Passed --   0.9  3   266   52
Test: progress in majority (3A) ...
  ... Passed --   0.7  5    79    2
Test: no progress in minority (3A) ...
  ... Passed --   1.0  5   122    3
Test: completion after heal (3A) ...
  ... Passed --   1.0  5    63    3
Test: partitions, one client (3A) ...
  ... Passed --  22.5  5 23221 1144
Test: partitions, many clients (3A) ...
  ... Passed --  22.8  5 99257 3713
Test: restarts, one client (3A) ...
  ... Passed --  19.1  5 48949 1402
Test: restarts, many clients (3A) ...
  ... Passed --  19.2  5 143921 5135
Test: unreliable net, restarts, many clients (3A) ...
  ... Passed --  20.0  5 11545 1522
Test: restarts, partitions, many clients (3A) ...
  ... Passed --  26.8  5 165901 5252
Test: unreliable net, restarts, partitions, many clients (3A) ...
  ... Passed --  27.4  5  8408  923
Test: unreliable net, restarts, partitions, many clients, linearizability checks (3A) ...
  ... Passed --  25.4  7 31127 2618
PASS
ok  	github.com/alioth4j/6.824/src/kvraft	233.596s

Conclusion

In this part of lab we did several things:

• Communications between objects
• The usage of Raft
• Applying committed logs to the state machine
• Ordering keeping
• Notification mechanism
• Concurrency control