augurier
5 months ago
15 changed files with 614 additions and 142 deletions
Split View
Diff Options
-
+7 -10cmd/main.go
-
+19 -27internal/client/client_node.go
-
+140 -69internal/nodes/init.go
-
+5 -7internal/nodes/node.go
-
+111 -0internal/nodes/real_transport.go
-
+13 -13internal/nodes/replica.go
-
+2 -2internal/nodes/server_node.go
-
+101 -0internal/nodes/thread_transport.go
-
+10 -0internal/nodes/transport.go
-
+7 -8internal/nodes/vote.go
-
+6 -4test/restart_node_test.go
-
+5 -2test/server_client_test.go
-
+38 -0threadTest/common.go
-
+81 -0threadTest/restart_node_test.go
-
+69 -0threadTest/server_client_test.go
+ 7
- 10
cmd/main.go
View File
+ 19
- 27
internal/client/client_node.go
View File
+ 140
- 69
internal/nodes/init.go
View File
+ 5
- 7
internal/nodes/node.go
View File
+ 111
- 0
internal/nodes/real_transport.go
View File
| @ -0,0 +1,111 @@ | |||
| package nodes | |||
| import ( | |||
| "errors" | |||
| "fmt" | |||
| "net/rpc" | |||
| "time" | |||
| ) | |||
| // 真实rpc通讯的transport类型实现 | |||
| type HTTPTransport struct{ | |||
| NodeMap map[string]string // id到addr的映射 | |||
| } | |||
| // 封装有超时的dial | |||
| func (t *HTTPTransport) DialHTTPWithTimeout(network string, peerId string) (ClientInterface, error) { | |||
| done := make(chan struct{}) | |||
| var client *rpc.Client | |||
| var err error | |||
| go func() { | |||
| client, err = rpc.DialHTTP(network, t.NodeMap[peerId]) | |||
| close(done) | |||
| }() | |||
| select { | |||
| case <-done: | |||
| return &HTTPClient{rpcClient: client}, err | |||
| case <-time.After(50 * time.Millisecond): | |||
| return nil, fmt.Errorf("dial timeout: %s", t.NodeMap[peerId]) | |||
| } | |||
| } | |||
| func (t *HTTPTransport) CallWithTimeout(clientInterface ClientInterface, serviceMethod string, args interface{}, reply interface{}) error { | |||
| c, ok := clientInterface.(*HTTPClient) | |||
| client := c.rpcClient | |||
| if !ok { | |||
| return fmt.Errorf("invalid client type") | |||
| } | |||
| done := make(chan error, 1) | |||
| go func() { | |||
| switch serviceMethod { | |||
| case "Node.AppendEntries": | |||
| arg, ok := args.(*AppendEntriesArg) | |||
| resp, ok2 := reply.(*AppendEntriesReply) | |||
| if !ok || !ok2 { | |||
| done <- errors.New("type assertion failed for AppendEntries") | |||
| return | |||
| } | |||
| done <- client.Call(serviceMethod, arg, resp) | |||
| case "Node.RequestVote": | |||
| arg, ok := args.(*RequestVoteArgs) | |||
| resp, ok2 := reply.(*RequestVoteReply) | |||
| if !ok || !ok2 { | |||
| done <- errors.New("type assertion failed for RequestVote") | |||
| return | |||
| } | |||
| done <- client.Call(serviceMethod, arg, resp) | |||
| case "Node.WriteKV": | |||
| arg, ok := args.(*LogEntryCall) | |||
| resp, ok2 := reply.(*ServerReply) | |||
| if !ok || !ok2 { | |||
| done <- errors.New("type assertion failed for WriteKV") | |||
| return | |||
| } | |||
| done <- client.Call(serviceMethod, arg, resp) | |||
| case "Node.ReadKey": | |||
| arg, ok := args.(*string) | |||
| resp, ok2 := reply.(*ServerReply) | |||
| if !ok || !ok2 { | |||
| done <- errors.New("type assertion failed for ReadKey") | |||
| return | |||
| } | |||
| done <- client.Call(serviceMethod, arg, resp) | |||
| case "Node.FindLeader": | |||
| arg, ok := args.(struct{}) | |||
| resp, ok2 := reply.(*FindLeaderReply) | |||
| if !ok || !ok2 { | |||
| done <- errors.New("type assertion failed for FindLeader") | |||
| return | |||
| } | |||
| done <- client.Call(serviceMethod, arg, resp) | |||
| default: | |||
| done <- fmt.Errorf("unknown service method: %s", serviceMethod) | |||
| } | |||
| }() | |||
| select { | |||
| case err := <-done: | |||
| return err | |||
| case <-time.After(50 * time.Millisecond): // 设置超时时间 | |||
| return fmt.Errorf("call timeout: %s", serviceMethod) | |||
| } | |||
| } | |||
| type HTTPClient struct { | |||
| rpcClient *rpc.Client | |||
| } | |||
| func (h *HTTPClient) Close() error { | |||
| return h.rpcClient.Close() | |||
| } | |||
+ 13
- 13
internal/nodes/replica.go
View File
+ 2
- 2
internal/nodes/server_node.go
View File
+ 101
- 0
internal/nodes/thread_transport.go
View File
| @ -0,0 +1,101 @@ | |||
| package nodes | |||
| import ( | |||
| "fmt" | |||
| "sync" | |||
| "time" | |||
| ) | |||
| // RPC 请求结构 | |||
| type RPCRequest struct { | |||
| ServiceMethod string | |||
| Args interface{} | |||
| Reply interface{} | |||
| Done chan error // 用于返回响应 | |||
| } | |||
| // 线程版 Transport | |||
| type ThreadTransport struct { | |||
| mu sync.Mutex | |||
| nodeChans map[string]chan RPCRequest // 每个节点的消息通道 | |||
| } | |||
| // 线程版 dial的返回clientinterface | |||
| type ThreadClient struct { | |||
| targetId string | |||
| } | |||
| func (c *ThreadClient) Close() error { | |||
| return nil | |||
| } | |||
| // 初始化线程通信系统 | |||
| func NewThreadTransport() *ThreadTransport { | |||
| return &ThreadTransport{ | |||
| nodeChans: make(map[string]chan RPCRequest), | |||
| } | |||
| } | |||
| // 注册一个新节点chan | |||
| func (t *ThreadTransport) RegisterNodeChan(nodeId string, ch chan RPCRequest) { | |||
| t.mu.Lock() | |||
| defer t.mu.Unlock() | |||
| t.nodeChans[nodeId] = ch | |||
| } | |||
| // 获取节点的 channel | |||
| func (t *ThreadTransport) getNodeChan(nodeId string) (chan RPCRequest, bool) { | |||
| t.mu.Lock() | |||
| defer t.mu.Unlock() | |||
| ch, ok := t.nodeChans[nodeId] | |||
| return ch, ok | |||
| } | |||
| // 模拟 Dial 操作 | |||
| func (t *ThreadTransport) DialHTTPWithTimeout(network string, peerId string) (ClientInterface, error) { | |||
| t.mu.Lock() | |||
| defer t.mu.Unlock() | |||
| if _, exists := t.nodeChans[peerId]; !exists { | |||
| return nil, fmt.Errorf("节点 [%s] 不存在", peerId) | |||
| } | |||
| return &ThreadClient{targetId: peerId}, nil | |||
| } | |||
| // 模拟 Call 操作 | |||
| func (t *ThreadTransport) CallWithTimeout(client ClientInterface, serviceMethod string, args interface{}, reply interface{}) error { | |||
| threadClient, ok := client.(*ThreadClient) | |||
| if !ok { | |||
| return fmt.Errorf("无效的客户端") | |||
| } | |||
| // 获取目标节点的 channel | |||
| targetChan, exists := t.getNodeChan(threadClient.targetId) | |||
| if !exists { | |||
| return fmt.Errorf("目标节点 [%s] 不存在", threadClient.targetId) | |||
| } | |||
| // 创建响应通道（用于返回 RPC 结果） | |||
| done := make(chan error, 1) | |||
| // 发送请求 | |||
| request := RPCRequest{ | |||
| ServiceMethod: serviceMethod, | |||
| Args: args, | |||
| Reply: reply, | |||
| Done: done, | |||
| } | |||
| select { | |||
| case targetChan <- request: | |||
| // 等待响应或超时 | |||
| select { | |||
| case err := <-done: | |||
| return err | |||
| case <-time.After(100 * time.Millisecond): | |||
| return fmt.Errorf("RPC 调用超时: %s", serviceMethod) | |||
| } | |||
| default: | |||
| return fmt.Errorf("目标节点 [%s] 无法接收请求", threadClient.targetId) | |||
| } | |||
| } | |||
+ 10
- 0
internal/nodes/transport.go
View File
| @ -0,0 +1,10 @@ | |||
| package nodes | |||
| type ClientInterface interface{ | |||
| Close() error | |||
| } | |||
| type Transport interface { | |||
| DialHTTPWithTimeout(network string, peerId string) (ClientInterface, error) | |||
| CallWithTimeout(client ClientInterface, serviceMethod string, args interface{}, reply interface{}) error | |||
| } | |||
+ 7
- 8
internal/nodes/vote.go
View File
+ 6
- 4
test/restart_node_test.go
View File
+ 5
- 2
test/server_client_test.go
View File
+ 38
- 0
threadTest/common.go
View File
| @ -0,0 +1,38 @@ | |||
| package threadTest | |||
| import ( | |||
| "fmt" | |||
| "os" | |||
| "simple-kv-store/internal/nodes" | |||
| "github.com/syndtr/goleveldb/leveldb" | |||
| ) | |||
| func ExecuteNodeI(id string, isRestart bool, peerIds []string, threadTransport *nodes.ThreadTransport) (*nodes.Node, chan struct{}) { | |||
| if !isRestart { | |||
| os.RemoveAll("storage/node" + id + ".json") | |||
| } | |||
| os.RemoveAll("leveldb/simple-kv-store" + id) | |||
| db, err := leveldb.OpenFile("leveldb/simple-kv-store"+id, nil) | |||
| if err != nil { | |||
| fmt.Println("Failed to open database: ", err) | |||
| } | |||
| // 打开或创建节点数据持久化文件 | |||
| storage := nodes.NewRaftStorage("storage/node" + id + ".json") | |||
| var otherIds []string | |||
| for _, ids := range peerIds { | |||
| if ids != id { | |||
| otherIds = append(otherIds, ids) // 删除目标元素 | |||
| } | |||
| } | |||
| // 初始化 | |||
| node, quitChan := nodes.InitThreadNode(id, otherIds, db, storage, isRestart, threadTransport) | |||
| // 开启 raft | |||
| go nodes.Start(node, quitChan) | |||
| return node, quitChan | |||
| } | |||
+ 81
- 0
threadTest/restart_node_test.go
View File
| @ -0,0 +1,81 @@ | |||
| package threadTest | |||
| import ( | |||
| "simple-kv-store/internal/client" | |||
| "simple-kv-store/internal/nodes" | |||
| "strconv" | |||
| "testing" | |||
| "time" | |||
| ) | |||
| func TestNodeRestart(t *testing.T) { | |||
| // 登记结点信息 | |||
| n := 5 | |||
| var peerIds []string | |||
| for i := 0; i < n; i++ { | |||
| peerIds = append(peerIds, strconv.Itoa(i + 1)) | |||
| } | |||
| // 结点启动 | |||
| var quitCollections []chan struct{} | |||
| threadTransport := nodes.NewThreadTransport() | |||
| for i := 0; i < n; i++ { | |||
| _, quitChan := ExecuteNodeI(strconv.Itoa(i + 1), false, peerIds, threadTransport) | |||
| quitCollections = append(quitCollections, quitChan) | |||
| } | |||
| // 通知所有node结束 | |||
| defer func(){ | |||
| for _, quitChan := range quitCollections { | |||
| close(quitChan) | |||
| } | |||
| }() | |||
| time.Sleep(time.Second) // 等待启动完毕 | |||
| // client启动, 连接任意节点 | |||
| cWrite := clientPkg.Client{PeerIds: peerIds, Transport: threadTransport} | |||
| // 写入 | |||
| var s clientPkg.Status | |||
| for i := 0; i < 5; i++ { | |||
| key := strconv.Itoa(i) | |||
| newlog := nodes.LogEntry{Key: key, Value: "hello"} | |||
| s := cWrite.Write(nodes.LogEntryCall{LogE: newlog, CallState: nodes.Normal}) | |||
| if s != clientPkg.Ok { | |||
| t.Errorf("write test fail") | |||
| } | |||
| } | |||
| time.Sleep(time.Second) // 等待写入完毕 | |||
| // 模拟结点轮流崩溃 | |||
| for i := 0; i < n; i++ { | |||
| close(quitCollections[i]) | |||
| time.Sleep(time.Second) | |||
| _, quitChan := ExecuteNodeI(strconv.Itoa(i + 1), true, peerIds, threadTransport) | |||
| quitCollections[i] = quitChan | |||
| time.Sleep(time.Second) // 等待启动完毕 | |||
| } | |||
| // client启动 | |||
| cRead := clientPkg.Client{PeerIds: peerIds, Transport: threadTransport} | |||
| // 读写入数据 | |||
| for i := 0; i < 5; i++ { | |||
| key := strconv.Itoa(i) | |||
| var value string | |||
| s = cRead.Read(key, &value) | |||
| if s != clientPkg.Ok { | |||
| t.Errorf("Read test1 fail") | |||
| } | |||
| } | |||
| // 读未写入数据 | |||
| for i := 5; i < 15; i++ { | |||
| key := strconv.Itoa(i) | |||
| var value string | |||
| s = cRead.Read(key, &value) | |||
| if s != clientPkg.NotFound { | |||
| t.Errorf("Read test2 fail") | |||
| } | |||
| } | |||
| } | |||
+ 69
- 0
threadTest/server_client_test.go
View File
| @ -0,0 +1,69 @@ | |||
| package threadTest | |||
| import ( | |||
| "simple-kv-store/internal/client" | |||
| "simple-kv-store/internal/nodes" | |||
| "strconv" | |||
| "testing" | |||
| "time" | |||
| ) | |||
| func TestServerClient(t *testing.T) { | |||
| // 登记结点信息 | |||
| n := 5 | |||
| var peerIds []string | |||
| for i := 0; i < n; i++ { | |||
| peerIds = append(peerIds, strconv.Itoa(i + 1)) | |||
| } | |||
| // 结点启动 | |||
| var quitCollections []chan struct{} | |||
| threadTransport := nodes.NewThreadTransport() | |||
| for i := 0; i < n; i++ { | |||
| _, quitChan := ExecuteNodeI(strconv.Itoa(i + 1), false, peerIds, threadTransport) | |||
| quitCollections = append(quitCollections, quitChan) | |||
| } | |||
| // 通知所有node结束 | |||
| defer func(){ | |||
| for _, quitChan := range quitCollections { | |||
| close(quitChan) | |||
| } | |||
| }() | |||
| time.Sleep(time.Second) // 等待启动完毕 | |||
| // client启动 | |||
| c := clientPkg.Client{PeerIds: peerIds, Transport: threadTransport} | |||
| // 写入 | |||
| var s clientPkg.Status | |||
| for i := 0; i < 10; i++ { | |||
| key := strconv.Itoa(i) | |||
| newlog := nodes.LogEntry{Key: key, Value: "hello"} | |||
| s := c.Write(nodes.LogEntryCall{LogE: newlog, CallState: nodes.Normal}) | |||
| if s != clientPkg.Ok { | |||
| t.Errorf("write test fail") | |||
| } | |||
| } | |||
| time.Sleep(time.Second) // 等待写入完毕 | |||
| // 读写入数据 | |||
| for i := 0; i < 10; i++ { | |||
| key := strconv.Itoa(i) | |||
| var value string | |||
| s = c.Read(key, &value) | |||
| if s != clientPkg.Ok || value != "hello" { | |||
| t.Errorf("Read test1 fail") | |||
| } | |||
| } | |||
| // 读未写入数据 | |||
| for i := 10; i < 15; i++ { | |||
| key := strconv.Itoa(i) | |||
| var value string | |||
| s = c.Read(key, &value) | |||
| if s != clientPkg.NotFound { | |||
| t.Errorf("Read test2 fail") | |||
| } | |||
| } | |||
| } | |||