func (l *raftLog) WaitCommitted() <-chan replication.LogEntry {
	return l.waitCommitted
}
WaitCommitted implements replication.LogReplicator
package raftlog
import (
	"encoding/binary"
	"fmt"
	"log"
	"math"
	"sync"
	"time"
	"google.golang.org/grpc"
	"github.com/andres-erbsen/clock"
	"github.com/coreos/etcd/raft"
	"github.com/coreos/etcd/raft/raftpb"
	"github.com/yahoo/coname/keyserver/kv"
	"github.com/yahoo/coname/keyserver/replication"
	"github.com/yahoo/coname/keyserver/replication/raftlog/proto"
	"golang.org/x/net/context"
)
const (
	HARDSTATE_KEY    = "HS"
	CONFSTATE_KEY    = "CS"
	ENTRY_KEY_PREFIX = "E"
	COMMITTED_BUFFER = 10  
)
type raftLog struct {
	config raft.Config
	node   raft.Node  
	clk          clock.Clock
	tickInterval time.Duration
	waitCommitted chan replication.LogEntry
	leaderHintSet chan bool
	leaderHint    bool
	grpcServer      *grpc.Server
	dial            func(uint64) proto.RaftClient
	grpcClientCache map[uint64]proto.RaftClient
	grpcDropClient  chan uint64
	stopOnce sync.Once
	stop     chan struct{}
	stopped  chan struct{}
}
var _ replication.LogReplicator = (*raftLog)(nil)
var _ proto.RaftServer = (*raftLog)(nil)
func (l *raftLog) Step(ctx context.Context, msg *raftpb.Message) (*proto.Nothing, error) {
	return &proto.Nothing{}, l.node.Step(ctx, *msg)
}
func New(
	thisReplica uint64, initialReplicas []uint64,
	db kv.DB, prefix []byte,
	clk clock.Clock, tickInterval time.Duration,
	server *grpc.Server, dial func(id uint64) proto.RaftClient,
) replication.LogReplicator {
	confState := raftpb.ConfState{}
	for _, id := range initialReplicas {
		confState.Nodes = append(confState.Nodes, id)
	}
	storage := mkRaftStorage(db, prefix, confState)
	l := &raftLog{
		config: raft.Config{
			ID:              thisReplica,
			ElectionTick:    10,
			HeartbeatTick:   1,
			MaxSizePerMsg:   4 * 1024,
			MaxInflightMsgs: 256,
			Storage:         storage,
		},
		node:         nil,
		clk:          clk,
		tickInterval: tickInterval,
		grpcServer:   server,
		dial:         dial,
	}
	proto.RegisterRaftServer(l.grpcServer, l)
	return l
}
func (l *raftLog) Start(lo uint64) error {
	inited, err := l.config.Storage.(*raftStorage).IsInitialized()
	if err != nil {
		return err
	}
	if inited {
		l.config.Applied = lo
		l.node = raft.RestartNode(&l.config)
	} else {
		if lo != 0 {
			log.Panicf("storage uninitialized but state machine not fresh: lo = %d", lo)
		}
		hardState, confState, err := l.config.Storage.InitialState()
		if err != nil {
			return err
		}
		confNodes := make([]raft.Peer, 0, len(confState.Nodes))
		for _, id := range confState.Nodes {
			confNodes = append(confNodes, raft.Peer{ID: id})
		}
		l.config.Storage.(*raftStorage).save(hardState, make([]raftpb.Entry, 1))
		l.node = raft.StartNode(&l.config, confNodes)
	}
	l.leaderHintSet = make(chan bool, COMMITTED_BUFFER)
	l.waitCommitted = make(chan replication.LogEntry, COMMITTED_BUFFER)
	l.stop = make(chan struct{})
	l.stopped = make(chan struct{})
	l.grpcDropClient = make(chan uint64)
	l.grpcClientCache = make(map[uint64]proto.RaftClient)
	go l.run()
	return nil
}
func (l *raftLog) Stop() error {
	l.stopOnce.Do(func() {
		l.grpcServer.Stop()
		if l.stop != nil {
			close(l.stop)
		}
		if l.stopped != nil {
			<-l.stopped
		}
		if l.node != nil {
			l.node.Stop()
		}
	})
	return nil
}
func (l *raftLog) Propose(ctx context.Context, en replication.LogEntry) {
	if en.ConfChange == nil {
		l.node.Propose(ctx, en.Data)
		return
	}
	nodeID := en.ConfChange.NodeID
	var ct raftpb.ConfChangeType
	switch en.ConfChange.Operation {
	case replication.ConfChangeNOP:
		nodeID = raft.None
		ct = raftpb.ConfChangeUpdateNode
	case replication.ConfChangeAddNode:
		ct = raftpb.ConfChangeAddNode
	case replication.ConfChangeRemoveNode:
		ct = raftpb.ConfChangeRemoveNode
	case replication.ConfChangeUpdateNode:
		ct = raftpb.ConfChangeUpdateNode
	default:
		panic("unknown conf change type proposed")
	}
	l.node.ProposeConfChange(ctx, raftpb.ConfChange{
		NodeID:  nodeID,
		Type:    ct,
		Context: en.Data,
	})
}
func (l *raftLog) ApplyConfChange(chn *replication.ConfChange) {
	nodeID := chn.NodeID
	var ct raftpb.ConfChangeType
	switch chn.Operation {
	case replication.ConfChangeNOP:
		nodeID = raft.None
		ct = raftpb.ConfChangeUpdateNode
	case replication.ConfChangeAddNode:
		ct = raftpb.ConfChangeAddNode
	case replication.ConfChangeRemoveNode:
		ct = raftpb.ConfChangeRemoveNode
		l.grpcDropClient <- nodeID
	case replication.ConfChangeUpdateNode:
		ct = raftpb.ConfChangeUpdateNode
		l.grpcDropClient <- nodeID
	default:
		panic("unknown conf change type applied")
	}
	l.node.ApplyConfChange(raftpb.ConfChange{
		Type:   raftpb.ConfChangeType(ct),
		NodeID: nodeID,
	})
}