func (r *Replica) inSccStack(other *Instance) bool {
	iter := r.sccStack.Front()
	for iter != nil {
		self := iter.Value.(*Instance)
		if self == other {
			return true
		}
		iter = iter.Next()
	}
	return false
}
TODO: this could be optimized in O(1) with marking flag.
package replica
import (
	"bytes"
	"container/list"
	"encoding/gob"
	"errors"
	"fmt"
	"sort"
	"time"
	"github.com/go-distributed/epaxos"
	"github.com/go-distributed/epaxos/message"
	"github.com/go-distributed/epaxos/persistent"
	"github.com/golang/glog"
)
var v1Log = glog.V(0)
var v2Log = glog.V(0)
var (
	errConflictsNotFullyResolved = errors.New("Conflicts not fully resolved")
)
const (
	defaultInstancesLength = 1024 * 64
	conflictNotFound       = 0
	epochStart             = 1
)
const (
	noAction uint8 = iota + 1
	replyAction
	fastQuorumAction
	broadcastAction
)
const (
	defaultClusterSize     = 3
	defaultCheckpointCycle = 1024
	defaultBatchInterval   = time.Millisecond * 50
	defaultTimeoutInterval = time.Millisecond * 50
	defaultExecuteInterval = time.Millisecond * 50
)
const defaultStartPort = 8080
type PackedReplica struct {
	Id             uint8
	Size           uint8
	MaxInstanceNum []uint64
	ExecutedUpTo   []uint64
	ProposeNum     uint64
}
type Replica struct {
	Id              uint8
	Size            uint8
	MaxInstanceNum  []uint64  
	ProposeNum      uint64
	ProposeChan     chan *proposeRequest
	BatchInterval   time.Duration
	TimeoutInterval time.Duration
	CheckpointCycle uint64
	ExecutedUpTo    []uint64
	InstanceMatrix  [][]*Instance
	StateMachine    epaxos.StateMachine
	Epoch           uint32
	MessageChan     chan message.Message
	Addrs           []string
	Transporter     epaxos.Transporter
	sccStack   *list.List
	sccResults [][]*Instance
	sccIndex   int
	executeTicker *time.Ticker
	timeoutTicker *time.Ticker
	proposeTicker *time.Ticker
	executeTrigger chan bool
	enableBatching bool
	stop           chan struct{}
	enablePersistent bool
	store            *persistent.LevelDB
}
type Param struct {
	ReplicaId        uint8
	Size             uint8
	StateMachine     epaxos.StateMachine
	CheckpointCycle  uint64
	BatchInterval    time.Duration
	TimeoutInterval  time.Duration
	ExecuteInterval  time.Duration
	Addrs            []string
	Transporter      epaxos.Transporter
	EnableBatching   bool
	EnablePersistent bool
	Restore          bool
	PersistentPath   string
}
type proposeRequest struct {
	cmds message.Commands
	id   chan uint64
}
func newProposeRequest(command ...message.Command) *proposeRequest {
	return &proposeRequest{
		cmds: message.Commands(command),
		id:   make(chan uint64, 1),  
	}
}
func verifyparam(param *Param) error {
	if param.Size == 0 {
		param.Size = defaultClusterSize
	}
	if param.Size%2 == 0 {
		return fmt.Errorf("Use odd number as quorum size")
	}
	if param.CheckpointCycle == 0 {
		param.CheckpointCycle = defaultCheckpointCycle
	}
	if param.BatchInterval == 0 {
		param.BatchInterval = defaultBatchInterval
	}
	if param.TimeoutInterval == 0 {
		param.TimeoutInterval = defaultTimeoutInterval
	}
	if param.ExecuteInterval == 0 {
		param.ExecuteInterval = defaultExecuteInterval
	}
	if param.Addrs == nil {
		param.Addrs = make([]string, param.Size)
		for i := 0; i < int(param.Size); i++ {
			param.Addrs[i] = fmt.Sprintf("localhost:%d", defaultStartPort+i)
		}
	}
	if param.Transporter == nil {
		return fmt.Errorf("No specified transporter")
	}
	return nil
}
func New(param *Param) (*Replica, error) {
	err := verifyparam(param)
	if err != nil {
		return nil, err
	}
	r := &Replica{
		Id:              param.ReplicaId,
		Size:            param.Size,
		MaxInstanceNum:  make([]uint64, param.Size),
		ProposeNum:      1,  
		ProposeChan:     make(chan *proposeRequest, 1024),
		BatchInterval:   param.BatchInterval,
		TimeoutInterval: param.TimeoutInterval,
		CheckpointCycle: param.CheckpointCycle,
		ExecutedUpTo:    make([]uint64, param.Size),
		InstanceMatrix:  make([][]*Instance, param.Size),
		StateMachine:    param.StateMachine,
		Epoch:           epochStart,
		MessageChan:     make(chan message.Message, 1024),
		sccStack:        list.New(),
		Addrs:           param.Addrs,
		Transporter:     param.Transporter,
		executeTicker: time.NewTicker(param.ExecuteInterval),
		timeoutTicker: time.NewTicker(param.TimeoutInterval),
		executeTrigger:   make(chan bool),
		stop:             make(chan struct{}),
		enableBatching:   param.EnableBatching,
		enablePersistent: param.EnablePersistent,
	}
	var path string
	if param.PersistentPath == "" {
		path = fmt.Sprintf("%s-%d", "/dev/shm/test", r.Id)
	} else {
		path = param.PersistentPath
	}
	r.store, err = persistent.NewLevelDB(path, param.Restore)
	if err != nil {
		glog.Errorln("replica.New: failed to make new storage")
		return nil, err
	}
	for i := uint8(0); i < param.Size; i++ {
		r.InstanceMatrix[i] = make([]*Instance, defaultInstancesLength)
		r.MaxInstanceNum[i] = conflictNotFound
		r.ExecutedUpTo[i] = conflictNotFound
	}
	if param.Restore {
		err := r.RecoverFromPersistent()
		if err != nil {
			glog.Errorln("Recover from persistent failed!")
			return nil, err
		}
	}
	r.Transporter.RegisterChannel(r.MessageChan)
	if r.enableBatching {
		r.proposeTicker = time.NewTicker(param.BatchInterval)
	}
	return r, nil
}
func (r *Replica) Start() error {
	go r.eventLoop()
	go r.executeLoop()
	go r.proposeLoop()
	go r.timeoutLoop()
	return r.Transporter.Start()
}
func (r *Replica) stopTickers() {
	r.executeTicker.Stop()
	r.timeoutTicker.Stop()
	r.proposeTicker.Stop()
}
func (r *Replica) Stop() {
	close(r.stop)
	r.stopTickers()
	r.Transporter.Stop()
	r.store.Close()
}
func (r *Replica) timeoutLoop() {
	for {
		select {
		case <-r.stop:
			return
		case <-r.timeoutTicker.C:
			r.checkTimeout()
		}
	}
}
func (r *Replica) checkTimeout() {
	for i, instance := range r.InstanceMatrix {
		for j := r.ExecutedUpTo[i] + 1; j <= r.MaxInstanceNum[i]; j++ {
			if r.IsCheckpoint(j) {  
				continue
			}
			if instance[j] == nil || instance[j].isTimeout() {
				r.MessageChan <- r.makeTimeout(uint8(i), j)
			}
		}
	}
}
func (r *Replica) makeTimeout(rowId uint8, instanceId uint64) message.Message {
	return &message.Timeout{
		ReplicaId:  rowId,
		InstanceId: instanceId,
		From:       r.Id,
	}
}
func (r *Replica) eventLoop() {
	for {
		select {
		case <-r.stop:
			return
		case msg := <-r.MessageChan:
			r.dispatch(msg)
		}
	}
}
func (r *Replica) executeLoop() {
	for {
		select {
		case <-r.stop:
			return
		case <-r.executeTicker.C:
			r.findAndExecute()
		case <-r.executeTrigger:
			r.findAndExecute()
		}
	}
}
func (r *Replica) proposeLoop() {
	if r.enableBatching {
		r.proposeLoopWithBatching()
	} else {
		r.proposeLoopWithoutBatching()
	}
}
func (r *Replica) proposeLoopWithBatching() {
	bufferedRequests := make([]*proposeRequest, 0)  
	for {
		select {
		case <-r.stop:
			return
		case req := <-r.ProposeChan:
			bufferedRequests = append(bufferedRequests, req)
		case <-r.proposeTicker.C:
			r.batchPropose(&bufferedRequests)
		}
	}
}
func (r *Replica) proposeLoopWithoutBatching() {
	bufferedRequests := make([]*proposeRequest, 0)  
	for {
		select {
		case <-r.stop:
			return
		case req := <-r.ProposeChan:
			bufferedRequests = append(bufferedRequests, req)
			r.batchPropose(&bufferedRequests)
		}
	}
}
func (r *Replica) Propose(cmds ...message.Command) chan uint64 {
	req := newProposeRequest(cmds...)
	r.ProposeChan <- req
	return req.id
}
func (r *Replica) batchPropose(batchedRequests *[]*proposeRequest) {
	defer func() { *batchedRequests = (*batchedRequests)[:0] }()  
	br := *batchedRequests
	if len(br) == 0 {
		return
	}
	cmds := make([]message.Command, 0)
	for i := range br {
		cmds = append(cmds, br[i].cmds...)
	}
	iid := r.ProposeNum
	proposal := message.NewPropose(r.Id, iid, cmds)
	r.ProposeNum++
	if r.IsCheckpoint(r.ProposeNum) {
		r.ProposeNum++
	}
	r.StoreReplica()
	r.MessageChan <- proposal
	<-proposal.Created
	for _, req := range br {
		req.id <- iid
		close(req.id)
	}
}
func (r *Replica) dispatch(msg message.Message) {
	replicaId := msg.Replica()
	instanceId := msg.Instance()
	r.updateMaxInstanceNum(replicaId, instanceId)
	v1Log.Infof("Replica[%v]: recv message[%s], from Replica[%v]\n",
		r.Id, msg.String(), msg.Sender())
	if glog.V(0) {
		printDependencies(msg)
	}
	if instanceId <= conflictNotFound {
		panic("")
	}
	if r.InstanceMatrix[replicaId][instanceId] == nil {
		r.InstanceMatrix[replicaId][instanceId] = NewInstance(r, replicaId, instanceId)
		if p, ok := msg.(*message.Propose); ok {
			close(p.Created)
		}
	}
	i := r.InstanceMatrix[replicaId][instanceId]
	i.touch()  
	v1Log.Infof("Replica[%v]: instance[%v][%v] status before = %v, ballot = [%v]\n",
		r.Id, replicaId, instanceId, i.StatusString(), i.ballot.String())
	v2Log.Infof("dependencies before: %v\n", i.Dependencies())
	var action uint8
	var rep message.Message
	switch i.status {
	case nilStatus:
		action, rep = i.nilStatusProcess(msg)
	case preAccepted:
		action, rep = i.preAcceptedProcess(msg)
	case accepted:
		action, rep = i.acceptedProcess(msg)
	case committed:
		action, rep = i.committedProcess(msg)
	case preparing:
		action, rep = i.preparingProcess(msg)
	default:
		panic("")
	}
	v2Log.Infof("dependencies after: %v\n", i.Dependencies())
	if action == noAction {
		v1Log.Infof("Replica[%v]: instance[%v][%v] status after = %v, ballot = [%v]\n\n\n",
			r.Id, replicaId, instanceId, i.StatusString(), i.ballot.String())
	} else {
		v1Log.Infof("Replica[%v]: instance[%v][%v] status after = %v, ballot = [%v]\n",
			r.Id, replicaId, instanceId, i.StatusString(), i.ballot.String())
	}
	if r.enablePersistent {
		r.StoreSingleInstance(i)
	}
	switch action {
	case noAction:
		return
	case replyAction:
		v1Log.Infof("Replica[%v]: send message[%s], to Replica[%v]\n\n\n",
			r.Id, rep.String(), msg.Sender())
		r.Transporter.Send(msg.Sender(), rep)  
	case fastQuorumAction:
		v1Log.Infof("Replica[%v]: send message[%s], to FastQuorum\n\n\n",
			r.Id, rep.String())
		r.Transporter.MulticastFastquorum(rep)
	case broadcastAction:
		v1Log.Infof("Replica[%v]: send message[%s], to Everyone\n\n\n",
			r.Id, rep.String())
		r.Transporter.Broadcast(rep)
	default:
		panic("")
	}
}
func (r *Replica) fastQuorum() int {
	if r.Size < 2 {
		panic("")
	}
	return int(r.Size - 2)
}
func (r *Replica) quorum() int {
	return int(r.Size / 2)
}
func (r *Replica) F() int {
	return int(r.Size / 2)
}
func (r *Replica) makeInitialBallot() *message.Ballot {
	return message.NewBallot(r.Epoch, 0, r.Id)
}
func (r *Replica) makeInitialDeps() message.Dependencies {
	return make(message.Dependencies, r.Size)
}
func (r *Replica) initInstance(cmds message.Commands, i *Instance) {
	if i.rowId != r.Id {
		panic("")
	}
	deps := make(message.Dependencies, r.Size)
	for curr := range r.InstanceMatrix {
		instances := r.InstanceMatrix[curr]
		start := r.MaxInstanceNum[curr]
		if curr == int(i.rowId) {
			deps[curr] = uint64(i.id - 1)
			continue
		}
		conflict := r.scanConflicts(instances, cmds, start, 0)
		deps[curr] = conflict
	}
	i.cmds, i.deps = cmds.Clone(), deps.Clone()
	i.replica.updateMaxInstanceNum(i.rowId, i.id)
}
func (r *Replica) updateInstance(cmds message.Commands, deps message.Dependencies, from uint8, i *Instance) bool {
	changed := false
	for curr := range r.InstanceMatrix {
		if curr == int(from) {
			continue
		}
		instances := r.InstanceMatrix[curr]
		start, end := r.MaxInstanceNum[curr], deps[curr]
		conflict := r.scanConflicts(instances, cmds, start, end)
		if deps[curr] < conflict {
			changed = true
			deps[curr] = conflict
		}
	}
	i.cmds, i.deps = cmds, deps
	i.replica.updateMaxInstanceNum(i.rowId, i.id)
	return changed
}
func (r *Replica) IsCheckpoint(n uint64) bool {
	return n%r.CheckpointCycle == 0
}
func (r *Replica) scanConflicts(instances []*Instance, cmds message.Commands, start uint64, end uint64) uint64 {
	for i := start; i > end; i-- {
		if r.IsCheckpoint(i) {
			return i
		}
		if instances[i] == nil {
			continue
		}
		if cmds == nil {
			return i
		}
		if r.StateMachine.HaveConflicts(cmds, instances[i].cmds) {
			return i
		}
	}
	return end
}
func (r *Replica) updateMaxInstanceNum(rowId uint8, instanceId uint64) bool {
	if r.enablePersistent {
		defer r.StoreReplica()
	}
	if r.MaxInstanceNum[rowId] < instanceId {
		r.MaxInstanceNum[rowId] = instanceId
		return true
	}
	return false
}
func (r *Replica) findAndExecute() {
	for i := 0; i < int(r.Size); i++ {
		for {
			up := r.ExecutedUpTo[i] + 1
			if r.IsCheckpoint(up) {
				r.ExecutedUpTo[i]++
				continue
			}
			instance := r.InstanceMatrix[i][up]
			if instance == nil {
				break
			}
			if !instance.isAtStatus(committed) {
				break
			}
			if instance.isExecuted() {
				r.ExecutedUpTo[i]++
				continue
			}
			if err := r.execute(instance); err != nil {
				switch err {
				case errConflictsNotFullyResolved:
					break
				case epaxos.ErrStateMachineExecution:
					panic("")
					break
				default:
					panic("unexpected error")
				}
			}
		}
	}
}
func (r *Replica) execute(i *Instance) error {
	r.sccStack.Init()
	r.sccResults = make([][]*Instance, 0)
	r.sccIndex = 1
	v2Log.Infoln("start resolve")
	if ok := r.resolveConflicts(i); !ok {
		v2Log.Infoln("there is incomplete scc")
	}
	if err := r.executeList(); err != nil {
		return err
	}
	return nil
}
func (r *Replica) executeList() error {
	cmdsBuffer := make([]message.Command, 0)
	v2Log.Infoln("execute list")
	for _, sccNodes := range r.sccResults {
		v2Log.Infoln("one scc")
		for _, instance := range sccNodes {
			v2Log.Infof("Instance [%v][%v] executed\n", instance.rowId, instance.id)
		}
		v2Log.Infoln("scc end")
		cmdsBuffer = cmdsBuffer[:0]
		for _, instance := range sccNodes {
			cmdsBuffer = append(cmdsBuffer, instance.cmds...)
		}
		_, err := r.StateMachine.Execute(cmdsBuffer)
		if err != nil {
			return err
		}
		for _, instance := range sccNodes {
			instance.SetExecuted()
		}
		if r.enablePersistent {
			r.StoreInstances(sccNodes...)
		}
	}
	return nil
}
func (r *Replica) resolveConflicts(node *Instance) bool {
	v2Log.Infof("resolve for [%v][%v]\n", node.rowId, node.id)
	if node == nil || !node.isAtStatus(committed) {
		panic("")
	}
	node.sccIndex = r.sccIndex
	node.sccLowlink = r.sccIndex
	r.sccIndex++
	r.pushSccStack(node)
	for iSpace := 0; iSpace < int(r.Size); iSpace++ {
		dep := node.deps[iSpace]
		if r.IsCheckpoint(dep) {
			continue
		}
		neighbor := r.InstanceMatrix[iSpace][dep]
		if neighbor == nil || !neighbor.isAtStatus(committed) {
			r.clearStack()
			return false
		}
		if neighbor.isExecuted() {
			continue
		}
		if neighbor.sccIndex == 0 {
			if ok := r.resolveConflicts(neighbor); !ok {
				r.clearStack()
				return false
			}
			if neighbor.sccLowlink < node.sccLowlink {
				node.sccLowlink = neighbor.sccLowlink
			}
		} else if r.inSccStack(neighbor) {
			if neighbor.sccLowlink < node.sccLowlink {
				node.sccLowlink = neighbor.sccLowlink
			}
		}
	}
	if node.sccLowlink == node.sccIndex {
		singleScc := make(sccNodesQueue, 0)
		for {
			n := r.popSccStack()
			singleScc = append(singleScc, n)
			if node == n {
				break
			}
		}
		sort.Sort(singleScc)
		r.sccResults = append(r.sccResults, singleScc)
	}
	return true
}
func (r *Replica) pushSccStack(i *Instance) {
	v2Log.Infof("push stack [%v][%v]\n", i.rowId, i.id)
	r.sccStack.PushBack(i)
}