Working on project2c

Author: Junlong Gao

.dockerignore

@@ -1,3 +1,4 @@
 proto/_tools
 bin/*
 doc/*
+out

README.md

@@ -4,6 +4,7 @@
 | Project 1 | Done |
 | Project 2a| Done |
 | Project 2b| Done |
+| Project 2c| In Progress |
 
 # The TinyKV Course
 

kv/config/config.go

@@ -84,7 +84,7 @@ func NewDefaultConfig() *Config {
 func NewTestConfig() *Config {
 	return &Config{
 		Raft:                     true,
-		LogLevel:                 "fatal",
+		LogLevel:                 "info",
 		RaftBaseTickInterval:     50 * time.Millisecond,
 		RaftHeartbeatTicks:       2,
 		RaftElectionTimeoutTicks: 10,

kv/raftstore/peer_msg_handler.go

@@ -2,9 +2,9 @@ package raftstore
 
 import (
 	"fmt"
+	"github.com/pingcap-incubator/tinykv/kv/raftstore/meta"
 	"time"
 
-	"github.com/pingcap-incubator/tinykv/kv/raftstore/meta"
 	"github.com/pingcap-incubator/tinykv/kv/util/engine_util"
 
 	"github.com/Connor1996/badger/y"
@@ -52,7 +52,7 @@ func (d *peerMsgHandler) HandleRaftReady() {
 
 	rd := d.peer.RaftGroup.Ready()
 	// save state, this includes the hard state and flush logs to stable storage
-	d.peer.peerStorage.SaveReadyState(&rd)
+	d.peer.peerStorage.SaveReadyState(d.PeerId(), &rd)
 	// send messages here:
 	for _, m := range rd.Messages {
 		err := d.peer.sendRaftMessage(m, d.ctx.trans)
@@ -63,35 +63,52 @@ func (d *peerMsgHandler) HandleRaftReady() {
 	}
 
 	// apply the committed entries to the state machine
-	wb := engine_util.WriteBatch{}
+	kvWB := engine_util.WriteBatch{}
 	for _, m := range rd.CommittedEntries {
 		var r raft_cmdpb.RaftCmdRequest
 		err := r.Unmarshal(m.Data)
 		if err != nil {
 			log.Panic(err)
 		}
+		log.Infof("Node %v processing index %v, term %v", d.PeerId(), m.Index, m.Term)
+		if m.Index <= d.peerStorage.applyState.AppliedIndex {
+			log.Infof("Node %v see stale index %v <= %v", d.PeerId(), m.Index, d.peerStorage.applyState.AppliedIndex)
+			continue
+		}
 		for _, cmd := range r.Requests {
 			switch cmd.CmdType {
 			case raft_cmdpb.CmdType_Put:
-				wb.SetCF(cmd.Put.Cf, cmd.Put.Key, cmd.Put.Value)
+				log.Infof("Node %v applying %v->%v at index %v", d.PeerId(), cmd.Put.Key, cmd.Put.Value, m.Index)
+				kvWB.SetCF(cmd.Put.Cf, cmd.Put.Key, cmd.Put.Value)
 			case raft_cmdpb.CmdType_Delete:
-				wb.DeleteCF(cmd.Delete.Cf, cmd.Delete.Key)
+				kvWB.DeleteCF(cmd.Delete.Cf, cmd.Delete.Key)
 			}
 		}
-	}
-	if wb.Len() > 0 {
-		as, err := meta.GetApplyState(d.peerStorage.Engines.Kv, d.peer.Region().Id)
-		if err != nil {
-			panic(err)
-		}
 
-		as.AppliedIndex = rd.CommittedEntries[len(rd.CommittedEntries)-1].Index
-		wb.SetMeta(meta.ApplyStateKey(d.PeerId()), as)
-		err = d.peerStorage.Engines.WriteKV(&wb)
-		if err != nil {
-			log.Panicf("write error: %v", err)
+		d.peerStorage.applyState.AppliedIndex = m.Index
+		kvWB.SetMeta(meta.ApplyStateKey(d.PeerId()), d.peerStorage.applyState)
+
+		// finally, the admin request:
+		if r.AdminRequest != nil {
+			switch r.AdminRequest.CmdType {
+			case raft_cmdpb.AdminCmdType_CompactLog:
+				// This is done async. Updating truncate state does not need to wait on actual gc to complete.
+				d.ScheduleCompactLog(r.AdminRequest.CompactLog.GetCompactIndex())
+				d.peerStorage.applyState.TruncatedState = &rspb.RaftTruncatedState{
+					Index: r.AdminRequest.CompactLog.GetCompactIndex(),
+					Term:  r.AdminRequest.CompactLog.GetCompactTerm(),
+				}
+				kvWB.SetMeta(meta.ApplyStateKey(d.PeerId()), d.peerStorage.applyState)
+			default:
+				panic("not implemented")
+			}
+
 		}
 	}
+	err := d.peerStorage.Engines.WriteKV(&kvWB)
+	if err != nil {
+		log.Panicf("write error: %v", err)
+	}
 
 	// Now replay the entire command for read
 	for _, m := range rd.CommittedEntries {
@@ -137,6 +154,7 @@ func (d *peerMsgHandler) HandleRaftReady() {
 							},
 						},
 					})
+				log.Infof("Node %v executing snapshot read at index %v", d.PeerId(), m.Index)
 			case raft_cmdpb.CmdType_Get:
 				txn := d.peerStorage.Engines.Kv.NewTransaction(false)
 				val, err := txn.Get(engine_util.KeyWithCF(cmd.Get.Cf, cmd.Get.Key))
@@ -172,6 +190,7 @@ func (d *peerMsgHandler) HandleRaftReady() {
 				log.Infof("Node %v got error at %v(%v): %v", d.PeerId(), m.Index, m.Term, p.Header.Error)
 			}
 		}
+
 		cb.Done(&p)
 	}
 
@@ -255,6 +274,7 @@ func (d *peerMsgHandler) proposeRaftCommand(msg *raft_cmdpb.RaftCmdRequest, cb *
 		log.Panic(err)
 	}
 	err = d.peer.RaftGroup.Propose(data)
+	log.Infof("Node %v submit for cmd at %v %v", d.PeerId(), index, term)
 	d.peer.proposals[index] = proposal{
 		index: index,
 		term:  term,
@@ -296,7 +316,7 @@ func (d *peerMsgHandler) onRaftBaseTick() {
 	d.ticker.schedule(PeerTickRaft)
 }
 
-func (d *peerMsgHandler) ScheduleCompactLog(firstIndex uint64, truncatedIndex uint64) {
+func (d *peerMsgHandler) ScheduleCompactLog(truncatedIndex uint64) {
 	raftLogGCTask := &runner.RaftLogGCTask{
 		RaftEngine: d.ctx.engine.Raft,
 		RegionID:   d.regionId,
@@ -567,8 +587,7 @@ func (d *peerMsgHandler) onRaftGCLogTick() {
 
 	term, err := d.RaftGroup.Raft.RaftLog.Term(compactIdx)
 	if err != nil {
-		log.Fatalf("appliedIdx: %d, firstIdx: %d, compactIdx: %d", appliedIdx, firstIdx, compactIdx)
-		panic(err)
+		log.Panicf("appliedIdx: %d, firstIdx: %d, compactIdx: %d", appliedIdx, firstIdx, compactIdx)
 	}
 
 	// Create a compact log request and notify directly.

kv/raftstore/peer_storage.go

@@ -306,31 +306,35 @@ func ClearMeta(engines *engine_util.Engines, kvWB, raftWB *engine_util.WriteBatc
 
 // Append the given entries to the raft log and update ps.raftState also delete log entries that will
 // never be committed
-func (ps *PeerStorage) Append(entries []eraftpb.Entry, raftWB *engine_util.WriteBatch) error {
+func (ps *PeerStorage) Append(nodeId uint64, entries []eraftpb.Entry, raftWB *engine_util.WriteBatch) error {
 	// Your Code Here (2B).
 	if len(entries) == 0 {
 		return nil
 	}
 
 	for _, e := range entries {
+		log.Infof("Node %v appending log %v %v", nodeId, e.Index, e.Term)
 		raftWB.SetMeta(meta.RaftLogKey(ps.region.Id, e.Index), &eraftpb.Entry{
 			EntryType: eraftpb.EntryType_EntryNormal,
 			Index:     e.Index,
 			Term:      e.Term,
 			Data:      e.Data,
 		})
-	}
 
-	lastIndex := entries[len(entries)-1].Index
-	lastTerm := entries[len(entries)-1].Term
-	ps.raftState.LastIndex = lastIndex
-	ps.raftState.LastTerm = lastTerm
+		if e.Index < ps.raftState.LastIndex {
+			// This can happen if snapshot applied before calling this extended the last index state further
+			continue
+		}
+		ps.raftState.LastIndex = e.Index
+		ps.raftState.LastTerm = e.Term
+		raftWB.SetMeta(meta.RaftStateKey(ps.region.Id), ps.raftState)
+	}
 
 	// prune more entries after last index
 	txn := ps.Engines.Raft.NewTransaction(false)
 	iter := txn.NewIterator(badger.DefaultIteratorOptions)
 	defer iter.Close()
-	for iter.Seek(meta.RaftLogKey(ps.region.Id, lastIndex+1)); iter.Valid() && !meta.IsRaftStateKey(iter.Item().Key()); iter.Next() {
+	for iter.Seek(meta.RaftLogKey(ps.region.Id, ps.raftState.LastIndex+1)); iter.Valid() && !meta.IsRaftStateKey(iter.Item().Key()); iter.Next() {
 		logIdx, err := meta.RaftLogIndex(iter.Item().Key())
 		if err != nil {
 			log.Panic(err)
@@ -344,54 +348,93 @@ func (ps *PeerStorage) Append(entries []eraftpb.Entry, raftWB *engine_util.Write
 // Apply the peer with given snapshot
 func (ps *PeerStorage) ApplySnapshot(snapshot *eraftpb.Snapshot, kvWB *engine_util.WriteBatch, raftWB *engine_util.WriteBatch) (*ApplySnapResult, error) {
 	log.Infof("%v begin to apply snapshot", ps.Tag)
-	snapData := new(rspb.RaftSnapshotData)
-	if err := snapData.Unmarshal(snapshot.Data); err != nil {
-		return nil, err
-	}
 
 	// Hint: things need to do here including: update peer storage state like raftState and applyState, etc,
 	// and send RegionTaskApply task to region worker through ps.regionSched, also remember call ps.clearMeta
 	// and ps.clearExtraData to delete stale data
 	// Your Code Here (2C).
+	if snapshot.Metadata.Index <= ps.applyState.AppliedIndex {
+		log.Infof("%v stale snapshot %v <= %v", ps.Tag, snapshot.Metadata.Index, ps.applyState.AppliedIndex)
+		return nil, nil
+	}
+	done := make(chan bool)
+	ps.snapState = snap.SnapState{StateType: snap.SnapState_Applying}
+	ps.regionSched <- runner.RegionTaskApply{
+		RegionId: ps.Region().Id,
+		Notifier: done,
+		SnapMeta: snapshot.Metadata,
+		StartKey: ps.Region().StartKey,
+		EndKey:   ps.Region().EndKey,
+	}
+	<-done
+
+	ps.applyState.AppliedIndex = snapshot.Metadata.Index
+	ps.applyState.TruncatedState = &rspb.RaftTruncatedState{
+		Index:                snapshot.Metadata.Index,
+		Term:                 snapshot.Metadata.Term,
+		XXX_NoUnkeyedLiteral: struct{}{},
+	}
+	kvWB.SetMeta(meta.ApplyStateKey(ps.Region().Id), ps.applyState)
+
+	if snapshot.Metadata.Index > ps.raftState.LastIndex {
+		ps.raftState.LastIndex = snapshot.Metadata.Index
+		ps.raftState.LastTerm = snapshot.Metadata.Term
+		raftWB.SetMeta(meta.RaftStateKey(ps.Region().Id), ps.raftState)
+	}
+
+	err := ps.clearMeta(raftWB, kvWB)
+	if err != nil {
+		panic(err)
+	}
+
+	// ps.clearExtraData()
 	return nil, nil
 }
 
 // Save memory states to disk.
 // Do not modify ready in this function, this is a requirement to advance the ready object properly later.
-func (ps *PeerStorage) SaveReadyState(ready *raft.Ready) (*ApplySnapResult, error) {
+func (ps *PeerStorage) SaveReadyState(nodeId uint64, ready *raft.Ready) (*ApplySnapResult, error) {
 	// Hint: you may call `Append()` and `ApplySnapshot()` in this function
 	// Your Code Here (2B/2C).
 	rd := *ready
-	wb := engine_util.WriteBatch{}
-
-	err := ps.Append(rd.Entries, &wb)
-	if err != nil {
-		log.Panic(err)
+	raftWb := engine_util.WriteBatch{}
+	kvWB := engine_util.WriteBatch{}
+	/*
+		Apply snapshot first, then process the new entries to be persisted into the log.
+		The snapshot will replace the entire log state and the kv state, bring the last applied and committed state
+		forward.
+	*/
+	var ret *ApplySnapResult
+	if rd.Snapshot != nil {
+		status, err := ps.ApplySnapshot(rd.Snapshot, &kvWB, &raftWb)
+		if err != nil {
+			panic(err)
+		}
+		ret = status
 	}
 
-	currentState, err := meta.GetRaftLocalState(ps.Engines.Raft, ps.region.Id)
+	err := ps.Append(nodeId, rd.Entries, &raftWb)
 	if err != nil {
 		log.Panic(err)
 	}
+
 	if rd.HardState.Term != 0 {
-		currentState.HardState.Term = rd.HardState.Term
-		currentState.HardState.Vote = rd.HardState.Vote
-		currentState.HardState.Commit = rd.HardState.Commit
-	}
-	if len(rd.Entries) > 0 {
-		currentState.LastIndex = rd.Entries[len(rd.Entries)-1].Index
-		currentState.LastTerm = rd.Entries[len(rd.Entries)-1].Term
+		ps.raftState.HardState.Term = rd.HardState.Term
+		ps.raftState.HardState.Vote = rd.HardState.Vote
+		ps.raftState.HardState.Commit = rd.HardState.Commit
+		raftWb.SetMeta(meta.RaftStateKey(ps.region.Id), ps.raftState)
 	}
 
-	wb.SetMeta(meta.RaftStateKey(ps.region.Id), currentState)
-
-	err = ps.Engines.WriteRaft(&wb)
+	err = ps.Engines.WriteRaft(&raftWb)
+	if err != nil {
+		log.Panic(err)
+	}
+	err = ps.Engines.WriteKV(&kvWB)
 	if err != nil {
 		log.Panic(err)
 	}
 
-	// applied is for snapshot?
-	return &ApplySnapResult{}, nil
+	return ret, nil
 }
 
 func (ps *PeerStorage) ClearData() {

proto/pkg/raft_serverpb/raft_serverpb.pb.go

@@ -81,7 +81,7 @@ message Message {
     bool reject = 10;
 }
 
-// HardState contains the state of a node need to be peristed, including the current term, commit index 
+// HardState contains the state of a node need to be persisted, including the current term, commit index
 // and the vote record
 message HardState {
     uint64 term = 1;

proto/proto/eraftpb.proto

@@ -215,14 +215,6 @@ func (l *RaftLog) nextEnts() (ents []pb.Entry) {
 	// Your Code Here (2A).
 
 	checkInv(l)
-	firstIdx, err := l.storage.FirstIndex()
-	if err != nil {
-		log.Panic(err)
-	}
-	if l.applied+1 < firstIdx {
-		log.Panicf("Index %v < %v", l.applied+1, firstIdx)
-	}
-
 	if l.applied+1 >= l.committed+1 {
 		return make([]pb.Entry, 0)
 	}
@@ -383,6 +375,7 @@ func (l *RaftLog) Slice(startIdx, length uint64) []*pb.Entry {
 			if ents[i-startIdx].Index != i {
 				log.Panic(ents[i-startIdx])
 			}
+			log.Infof("Got index from storage %v %v", i, ents[i-startIdx].Term)
 			ret = append(ret, &ents[i-startIdx])
 		}
 	}
@@ -406,3 +399,11 @@ func (l *RaftLog) allEntries() []pb.Entry {
 
 	return ret
 }
+
+func (l *RaftLog) FirstIndex() uint64 {
+	idx, err := l.storage.FirstIndex()
+	if err != nil {
+		panic(err)
+	}
+	return idx
+}