raft prune wal

plugin/consensus/raft/chain33.test.toml

@@ -38,7 +38,7 @@ enableTxQuickIndex=true
 seeds=["127.0.0.1:13802"]
 enable=false
 isSeed=false
-serverStart=true
+serverStart=false
 innerSeedEnable=false
 useGithub=false
 innerBounds=300
@@ -108,12 +108,14 @@ peersURL="http://127.0.0.1:9021"
 # raft共识用到，指示raft集群中只读节点的IP（只同步日志，不参与raft共识）
 readOnlyPeersURL=""
 addPeersURL=""
-#raft共识用到，默认raft中多少条记录打包一个snapshot(这里为了测试调整小一点）
+#raft中多少条记录打包一个snapshot，默认为10000(这里为了测试调整小一点）
 defaultSnapCount=2
-#raft共识用到，默认raft中写区块时间间隔
+#raft中写区块时间间隔，默认为1秒
 writeBlockSeconds=1
-#raft共识用到，默认raft中leader发送心跳包时间间隔
+#raft中leader发送心跳包时间间隔，默认为1秒
 heartbeatTick=1
+#raft中leader打包空区块的时间间隔，默认为0，表示不打包空区块
+emptyBlockInterval=120
 # =============== raft共识配置参数 ===========================
 
 [store]
@@ -134,7 +136,7 @@ dbCache=16
 signType="secp256k1"
 
 [wallet.sub.ticket]
-minerdisable=false
+minerdisable=true
 minerwhitelist=["*"]
 minerWaitTime="1s"
 

plugin/consensus/raft/controller.go

@@ -19,10 +19,11 @@ var (
 	rlog                    = log.New("module", "raft")
 	genesis                 string
 	genesisBlockTime        int64
-	defaultSnapCount        uint64 = 1000
-	snapshotCatchUpEntriesN uint64 = 1000
+	defaultSnapCount        uint64 = 10000
+	snapshotCatchUpEntriesN uint64 = 10000
 	writeBlockSeconds       int64  = 1
 	heartbeatTick                  = 1
+	emptyBlockInterval      int64
 	isLeader                = false
 	mux                     atomic.Value
 	confChangeC             chan raftpb.ConfChange
@@ -40,6 +41,7 @@ type subConfig struct {
 	DefaultSnapCount   int64  `json:"defaultSnapCount"`
 	WriteBlockSeconds  int64  `json:"writeBlockSeconds"`
 	HeartbeatTick      int32  `json:"heartbeatTick"`
+	EmptyBlockInterval int64  `json:"emptyBlockInterval"`
 }
 
 func init() {
@@ -64,7 +66,7 @@ func NewRaftCluster(cfg *types.Consensus, sub []byte) queue.Module {
 		//TODO 当传入的参数异常时，返回给主函数的是个nil,这时候需要做异常处理
 		return nil
 	}
-	// 默认1000个Entry打一个snapshot
+	// 默认10000个Entry打一个snapshot
 	if subcfg.DefaultSnapCount > 0 {
 		defaultSnapCount = uint64(subcfg.DefaultSnapCount)
 		snapshotCatchUpEntriesN = uint64(subcfg.DefaultSnapCount)
@@ -77,6 +79,11 @@ func NewRaftCluster(cfg *types.Consensus, sub []byte) queue.Module {
 	if subcfg.HeartbeatTick > 0 {
 		heartbeatTick = int(subcfg.HeartbeatTick)
 	}
+	// write empty block interval in second
+	if subcfg.EmptyBlockInterval > 0 {
+		emptyBlockInterval = subcfg.EmptyBlockInterval
+	}
+
 	var b *Client
 	getSnapshot := func() ([]byte, error) { return b.getSnapshot() }
 	// raft集群的建立,1. 初始化两条channel： propose channel用于客户端和raft底层交互, commit channel用于获取commit消息
@@ -96,12 +103,13 @@ func NewRaftCluster(cfg *types.Consensus, sub []byte) queue.Module {
 	//采用context来统一管理所有服务
 	ctx, stop := context.WithCancel(context.Background())
 	// propose channel
-	proposeC := make(chan *types.Block)
+	proposeC := make(chan BlockInfo)
 	confChangeC = make(chan raftpb.ConfChange)
-	commitC, errorC, snapshotterReady, validatorC := NewRaftNode(ctx, int(subcfg.NodeID), subcfg.IsNewJoinNode, peers, readOnlyPeers, addPeers, getSnapshot, proposeC, confChangeC)
+	node, commitC, errorC, snapshotterReady, validatorC := NewRaftNode(ctx, int(subcfg.NodeID), subcfg.IsNewJoinNode, peers, readOnlyPeers, addPeers, getSnapshot, proposeC, confChangeC)
 	//启动raft删除节点操作监听
 	go serveHTTPRaftAPI(ctx, int(subcfg.RaftAPIPort), confChangeC, errorC)
 	// 监听commit channel,取block
 	b = NewBlockstore(ctx, cfg, <-snapshotterReady, proposeC, commitC, errorC, validatorC, stop)
+	node.SetClient(b)
 	return b
 }

plugin/consensus/raft/raft.go

@@ -6,6 +6,7 @@ package raft
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"net/http"
 	"net/url"
@@ -14,7 +15,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/33cn/chain33/types"
 	"github.com/coreos/etcd/etcdserver/stats"
 	"github.com/coreos/etcd/pkg/fileutil"
 	typec "github.com/coreos/etcd/pkg/types"
@@ -24,7 +24,6 @@ import (
 	"github.com/coreos/etcd/snap"
 	"github.com/coreos/etcd/wal"
 	"github.com/coreos/etcd/wal/walpb"
-	"github.com/golang/protobuf/proto"
 )
 
 var (
@@ -32,9 +31,10 @@ var (
 )
 
 type raftNode struct {
-	proposeC         <-chan *types.Block
+	client           *Client
+	proposeC         <-chan BlockInfo
 	confChangeC      <-chan raftpb.ConfChange
-	commitC          chan<- *types.Block
+	commitC          chan<- *BlockInfo
 	errorC           chan<- error
 	id               int
 	bootstrapPeers   []string
@@ -65,13 +65,21 @@ type raftNode struct {
 	restartC chan struct{}
 }
 
+type Node struct {
+	*raftNode
+}
+
+func (node *Node) SetClient(client *Client) {
+	node.client = client
+}
+
 // NewRaftNode create raft node
-func NewRaftNode(ctx context.Context, id int, join bool, peers []string, readOnlyPeers []string, addPeers []string, getSnapshot func() ([]byte, error), proposeC <-chan *types.Block,
-	confChangeC <-chan raftpb.ConfChange) (<-chan *types.Block, <-chan error, <-chan *snap.Snapshotter, <-chan bool) {
+func NewRaftNode(ctx context.Context, id int, join bool, peers []string, readOnlyPeers []string, addPeers []string, getSnapshot func() ([]byte, error), proposeC <-chan BlockInfo,
+	confChangeC <-chan raftpb.ConfChange) (*Node, <-chan *BlockInfo, <-chan error, <-chan *snap.Snapshotter, <-chan bool) {
 
 	rlog.Info("Enter consensus raft")
 	// commit channel
-	commitC := make(chan *types.Block)
+	commitC := make(chan *BlockInfo)
 	errorC := make(chan error)
 	rc := &raftNode{
 		proposeC:         proposeC,
@@ -94,7 +102,7 @@ func NewRaftNode(ctx context.Context, id int, join bool, peers []string, readOnl
 	}
 	go rc.startRaft()
 
-	return commitC, errorC, rc.snapshotterReady, rc.validatorC
+	return &Node{rc}, commitC, errorC, rc.snapshotterReady, rc.validatorC
 }
 
 //  启动raft节点
@@ -224,7 +232,7 @@ func (rc *raftNode) serveChannels() {
 				if !ok {
 					rc.proposeC = nil
 				} else {
-					out, err := proto.Marshal(prop)
+					out, err := json.Marshal(prop)
 					if err != nil {
 						rlog.Error(fmt.Sprintf("failed to marshal block:%v ", err.Error()))
 					}
@@ -257,6 +265,10 @@ func (rc *raftNode) serveChannels() {
 		case <-ticker.C:
 			rc.node.Tick()
 		case rd := <-rc.node.Ready():
+			if !rc.checkEntries(rd.Entries) || !rc.checkEntries(rd.CommittedEntries) {
+				rc.stop()
+				return
+			}
 			rc.wal.Save(rd.HardState, rd.Entries)
 			if !raft.IsEmptySnap(rd.Snapshot) {
 				rc.saveSnap(rd.Snapshot)
@@ -283,6 +295,25 @@ func (rc *raftNode) serveChannels() {
 	}
 }
 
+func (rc *raftNode) checkEntries(ents []raftpb.Entry) bool {
+	for i := range ents {
+		if ents[i].Type == raftpb.EntryNormal && len(ents[i].Data) != 0 {
+			info := &BlockInfo{}
+			if err := json.Unmarshal(ents[i].Data, info); err != nil {
+				rlog.Error("checkEntries Unmarshal BlockInfo fail", "err", err)
+				return false
+			}
+			if rc.client != nil {
+				if !rc.client.CheckBlockInfo(info) {
+					rlog.Error("checkEntries CheckBlockInfo fail")
+					return false
+				}
+			}
+		}
+	}
+	return true
+}
+
 func (rc *raftNode) updateValidator() {
 
 	//TODO 这块监听后期需要根据场景进行优化?
@@ -325,6 +356,7 @@ func (rc *raftNode) updateValidator() {
 
 	}
 }
+
 func (rc *raftNode) Status() raft.Status {
 	rc.stopMu.RLock()
 	defer rc.stopMu.RUnlock()
@@ -383,15 +415,13 @@ func (rc *raftNode) maybeTriggerSnapshot() {
 		return
 	}
 
-	appliedIndex := rc.appliedIndex
-	snapshotIndex := rc.snapshotIndex
-	confState := rc.confState
-	rlog.Info(fmt.Sprintf("start snapshot [applied index: %d | last snapshot index: %d]", appliedIndex, snapshotIndex))
-	ents, err := rc.raftStorage.Entries(appliedIndex, appliedIndex+1, 2)
+	rlog.Info(fmt.Sprintf("start snapshot [applied index: %d | last snapshot index: %d]", rc.appliedIndex, rc.snapshotIndex))
+	data, err := rc.getSnapshot()
 	if err != nil {
-		rlog.Error(fmt.Sprintf("Err happened when get snapshot:%v", err.Error()))
+		rlog.Error("getSnapshot fail", "err", err)
+		panic(err)
 	}
-	snapShot, err := rc.raftStorage.CreateSnapshot(appliedIndex, &confState, ents[0].Data)
+	snapShot, err := rc.raftStorage.CreateSnapshot(rc.appliedIndex, &rc.confState, data)
 	if err != nil {
 		panic(err)
 	}
@@ -400,15 +430,15 @@ func (rc *raftNode) maybeTriggerSnapshot() {
 	}
 
 	compactIndex := uint64(1)
-	if appliedIndex > snapshotCatchUpEntriesN {
-		compactIndex = appliedIndex - snapshotCatchUpEntriesN
+	if rc.appliedIndex > snapshotCatchUpEntriesN {
+		compactIndex = rc.appliedIndex - snapshotCatchUpEntriesN
 	}
 	if err := rc.raftStorage.Compact(compactIndex); err != nil {
 		panic(err)
 	}
 
 	rlog.Info(fmt.Sprintf("compacted log at index %d", compactIndex))
-	rc.snapshotIndex = appliedIndex
+	rc.snapshotIndex = rc.appliedIndex
 }
 
 func (rc *raftNode) publishSnapshot(snapshotToSave raftpb.Snapshot) {
@@ -487,12 +517,13 @@ func (rc *raftNode) publishEntries(ents []raftpb.Entry) bool {
 				break
 			}
 			// 解码
-			block := &types.Block{}
-			if err := proto.Unmarshal(ents[i].Data, block); err != nil {
-				rlog.Error(fmt.Sprintf("failed to unmarshal: %v", err.Error()))
+			info := &BlockInfo{}
+			if err := json.Unmarshal(ents[i].Data, info); err != nil {
+				rlog.Error("Unmarshal BlockInfo fail", "err", err)
+				break
 			}
 			select {
-			case rc.commitC <- block:
+			case rc.commitC <- info:
 			case <-rc.ctx.Done():
 				return false
 			}

plugin/consensus/raft/tools/chain33-dump-logs/main.go

@@ -5,6 +5,7 @@
 package main
 
 import (
+	"encoding/json"
 	"flag"
 	"fmt"
 	"log"
@@ -18,7 +19,6 @@ import (
 	raftsnap "github.com/coreos/etcd/snap"
 	"github.com/coreos/etcd/wal"
 	"github.com/coreos/etcd/wal/walpb"
-	"github.com/golang/protobuf/proto"
 )
 
 func main() {
@@ -94,12 +94,12 @@ func main() {
 				break
 			}
 			// 解码
-			block := &Block{}
-			if err := proto.Unmarshal(e.Data, block); err != nil {
+			info := &BlockInfo{}
+			if err := json.Unmarshal(e.Data, info); err != nil {
 				log.Printf("failed to unmarshal: %v", err)
 				break
 			}
-			msg = fmt.Sprintf("%s\t BlockHeight:%d", msg, block.Height)
+			msg = fmt.Sprintf("%s\tHeight=%d\tHash=%s", msg, info.Height, info.Hash)
 		case raftpb.EntryConfChange:
 			msg = fmt.Sprintf("%s\tconf", msg)
 			var r raftpb.ConfChange
@@ -133,21 +133,7 @@ func genIDSlice(a []uint64) []types.ID {
 	return ids
 }
 
-// Block struct
-type Block struct {
-	Version    int64  `protobuf:"varint,1,opt,name=version" json:"version,omitempty"`
-	ParentHash []byte `protobuf:"bytes,2,opt,name=parentHash,proto3" json:"parentHash,omitempty"`
-	TxHash     []byte `protobuf:"bytes,3,opt,name=txHash,proto3" json:"txHash,omitempty"`
-	StateHash  []byte `protobuf:"bytes,4,opt,name=stateHash,proto3" json:"stateHash,omitempty"`
-	Height     int64  `protobuf:"varint,5,opt,name=height" json:"height,omitempty"`
-	BlockTime  int64  `protobuf:"varint,6,opt,name=blockTime" json:"blockTime,omitempty"`
-	//Signature  *Signature     `protobuf:"bytes,8,opt,name=signature" json:"signature,omitempty"`
-	//Txs        []*Transaction `protobuf:"bytes,7,rep,name=txs" json:"txs,omitempty"`
+type BlockInfo struct {
+	Height int64  `json:"height"`
+	Hash   string `json:"hash"`
 }
-
-// Reset method
-func (m *Block) Reset()         { *m = Block{} }
-func (m *Block) String() string { return proto.CompactTextString(m) }
-
-// ProtoMessage method
-func (*Block) ProtoMessage() {}