func (conn *connector) runStateMachine() bool {
	log := conn.log
	log.Infof("current state: %s", conn.state)
	continueLooping := true
	switch conn.state {
	case cStateConnecting:
		mode.Set(mode.Resynchronise)
		globalData.clientCount = conn.getConnectedClientCount()
		log.Infof("connections: %d", globalData.clientCount)
		if isConnectionEnough(globalData.clientCount) {
			conn.nextState(cStateHighestBlock)
		} else {
			log.Warnf("connections: %d below minimum client count: %d", globalData.clientCount, minimumClients)
			messagebus.Bus.Announce.Send("reconnect")
		}
		continueLooping = false
	case cStateHighestBlock:
		if conn.updateHeightAndClient() {
			log.Infof("highest block number: %d  client: %s", conn.height, conn.theClient.Name())
			if conn.hasBetterChain(blockheader.Height()) {
				log.Infof("new chain from %s, height %d, digest %s", conn.theClient.Name(), conn.height, conn.theClient.CachedRemoteDigestOfLocalHeight().String())
				log.Info("enter fork detect state")
				conn.nextState(cStateForkDetect)
			} else if conn.isSameChain() {
				log.Info("remote same chain")
				conn.nextState(cStateRebuild)
			} else {
				log.Info("remote chain invalid, stop looping for now")
				continueLooping = false
			}
		} else {
			log.Warn("highest block: connection lost")
			conn.nextState(cStateConnecting)
			continueLooping = false
		}
	case cStateForkDetect:
		height := blockheader.Height()
		if !conn.hasBetterChain(height) {
			log.Info("remote without better chain, enter state rebuild")
			conn.nextState(cStateRebuild)
		} else {
			if conn.height > fastSyncPivotBlocks {
				conn.pivotPoint = conn.height - fastSyncPivotBlocks
			} else {
				conn.pivotPoint = 0
			}
			log.Infof("Pivot point for fast sync: %d", conn.pivotPoint)
			conn.startBlockNumber = genesis.BlockNumber + 1
			conn.nextState(cStateFetchBlocks)  
			log.Infof("local block number: %d", height)
			blockheader.ClearCache()
		check_digests:
			for h := height; h >= genesis.BlockNumber; h -= 1 {
				digest, err := blockheader.DigestForBlock(h)
				if nil != err {
					log.Infof("block number: %d  local digest error: %s", h, err)
					conn.nextState(cStateHighestBlock)  
					break check_digests
				}
				d, err := conn.theClient.RemoteDigestOfHeight(h)
				if nil != err {
					log.Infof("block number: %d  fetch digest error: %s", h, err)
					conn.nextState(cStateHighestBlock)  
					break check_digests
				} else if d == digest {
					if height-h >= forkProtection {
						log.Errorf("fork protection at: %d - %d >= %d", height, h, forkProtection)
						conn.nextState(cStateHighestBlock)
						break check_digests
					}
					conn.startBlockNumber = h + 1
					log.Infof("fork from block number: %d", conn.startBlockNumber)
					err := block.DeleteDownToBlock(conn.startBlockNumber)
					if nil != err {
						log.Errorf("delete down to block number: %d  error: %s", conn.startBlockNumber, err)
						conn.nextState(cStateHighestBlock)  
					}
					break check_digests
				}
			}
		}
	case cStateFetchBlocks:
		continueLooping = false
		var packedBlock []byte
		var packedNextBlock []byte
		if conn.fastSyncEnabled && conn.pivotPoint >= conn.startBlockNumber+fastSyncFetchBlocksPerCycle {
			conn.blocksPerCycle = fastSyncFetchBlocksPerCycle
		} else {
			conn.blocksPerCycle = fetchBlocksPerCycle
		}
	fetch_blocks:
		for i := 0; i < conn.blocksPerCycle; i++ {
			if conn.startBlockNumber > conn.height {
				log.Infof("height changed from: %d to: %d", conn.height, conn.startBlockNumber)
				conn.nextState(cStateHighestBlock)
				continueLooping = true
				break fetch_blocks
			}
			if conn.startBlockNumber%100 == 0 {
				log.Warnf("fetch block number: %d", conn.startBlockNumber)
			} else {
				log.Infof("fetch block number: %d", conn.startBlockNumber)
			}
			if packedNextBlock == nil {
				p, err := conn.theClient.GetBlockData(conn.startBlockNumber)
				if nil != err {
					log.Errorf("fetch block number: %d  error: %s", conn.startBlockNumber, err)
					conn.nextState(cStateHighestBlock)  
					break fetch_blocks
				}
				packedBlock = p
			} else {
				packedBlock = packedNextBlock
			}
			if conn.fastSyncEnabled {
				if i > 0 && i%fastSyncSkipPerBlocks == 0 {
					h := conn.startBlockNumber - uint64(rand.Intn(fastSyncSkipPerBlocks))
					log.Debugf("select random block: %d to test for forgery", h)
					digest, err := blockheader.DigestForBlock(h)
					if nil != err {
						log.Infof("block number: %d  local digest error: %s", h, err)
						conn.nextState(cStateHighestBlock)  
						break fetch_blocks
					}
					d, err := conn.theClient.RemoteDigestOfHeight(h)
					if nil != err {
						log.Infof("block number: %d  fetch digest error: %s", h, err)
						conn.nextState(cStateHighestBlock)  
						break fetch_blocks
					}
					if d != digest {
						log.Warnf("potetial block forgery: %d", h)
						startingPoint := conn.startBlockNumber - uint64(i)
						err := block.DeleteDownToBlock(startingPoint)
						if nil != err {
							log.Errorf("delete down to block number: %d  error: %s", startingPoint, err)
						}
						conn.fastSyncEnabled = false
						conn.nextState(cStateHighestBlock)
						conn.startBlockNumber = startingPoint
						break fetch_blocks
					}
				}
				var err error
				packedNextBlock, err = conn.theClient.GetBlockData(conn.startBlockNumber + 1)
				if nil != err {
					log.Debugf("fetch next block number: %d  error: %s", conn.startBlockNumber+1, err)
				}
			} else {
				packedNextBlock = nil
			}
			log.Debugf("store block number: %d", conn.startBlockNumber)
			err := block.StoreIncoming(packedBlock, packedNextBlock, block.NoRescanVerified)
			if nil != err {
				log.Errorf(
					"store block number: %d  error: %s",
					conn.startBlockNumber,
					err,
				)
				conn.nextState(cStateHighestBlock)  
				break fetch_blocks
			}
			conn.startBlockNumber++
		}
	case cStateRebuild:
		conn.nextState(cStateSampling)
		conn.samples = 0  
		mode.Set(mode.Normal)
		continueLooping = false
	case cStateSampling:
		globalData.clientCount = conn.getConnectedClientCount()
		if !isConnectionEnough(globalData.clientCount) {
			log.Warnf("connections: %d below minimum client count: %d", globalData.clientCount, minimumClients)
			continueLooping = true
			conn.nextState(cStateConnecting)
			return continueLooping
		}
		log.Infof("connections: %d", globalData.clientCount)
		continueLooping = false
		if conn.updateHeightAndClient() {
			height := blockheader.Height()
			log.Infof("height remote: %d, local: %d", conn.height, height)
			if conn.hasBetterChain(height) {
				log.Warn("check height: better chain")
				conn.nextState(cStateForkDetect)
				continueLooping = true
			} else {
				conn.samples = 0
			}
		} else {
			conn.samples++
			if conn.samples > samplingLimit {
				log.Warn("check height: time to resync")
				conn.nextState(cStateForkDetect)
				continueLooping = true
			}
		}
	}
	return continueLooping
}
run state machine
return:
true  if want more cycles
false to pase for I/O
package peer
import (
	"bytes"
	"container/list"
	"fmt"
	"math/rand"
	"strings"
	"sync"
	"time"
	"github.com/bitmark-inc/bitmarkd/block"
	"github.com/bitmark-inc/bitmarkd/blockheader"
	"github.com/bitmark-inc/bitmarkd/fault"
	"github.com/bitmark-inc/bitmarkd/genesis"
	"github.com/bitmark-inc/bitmarkd/messagebus"
	"github.com/bitmark-inc/bitmarkd/mode"
	"github.com/bitmark-inc/bitmarkd/peer/upstream"
	"github.com/bitmark-inc/bitmarkd/peer/voting"
	"github.com/bitmark-inc/bitmarkd/util"
	"github.com/bitmark-inc/bitmarkd/zmqutil"
	"github.com/bitmark-inc/logger"
)
const (
	cycleInterval = 15 * time.Second
	connectorTimeout = 60 * time.Second
	samplingLimit = 6
	fetchBlocksPerCycle = 200
	forkProtection = 60
	minimumClients = 5
	maximumDynamicClients = 25
	activeTime = 60 * time.Second
	fastSyncFetchBlocksPerCycle = 2000
	fastSyncSkipPerBlocks       = 100
	fastSyncPivotBlocks         = 1000
)
type connector struct {
	sync.RWMutex
	log        *logger.L
	preferIPv6 bool
	staticClients []upstream.Upstream
	dynamicClients list.List
	state connectorState
	theClient        upstream.Upstream  
	startBlockNumber uint64             
	height           uint64             
	samples          int                
	votes            voting.Voting
	fastSyncEnabled bool    
	blocksPerCycle  int     
	pivotPoint      uint64  
}
func (conn *connector) initialise(
	privateKey []byte,
	publicKey []byte,
	connect []Connection,
	dynamicEnabled bool,
	preferIPv6 bool,
	fastSync bool,
) error {
	log := logger.New("connector")
	conn.log = log
	conn.preferIPv6 = preferIPv6
	conn.fastSyncEnabled = fastSync
	log.Info("initialising…")
	staticCount := len(connect)  
	if 0 == staticCount && !dynamicEnabled {
		log.Error("zero static connections and dynamic is disabled")
		return fault.NoConnectionsAvailable
	}
	conn.staticClients = make([]upstream.Upstream, staticCount)
	wg := sync.WaitGroup{}
	errCh := make(chan error, len(connect))
	conn.log.Debugf("static connection count: %d", len(connect))
	for i, c := range connect {
		wg.Add(1)
		go func(conn *connector, c Connection, i int, wg *sync.WaitGroup, ch chan error) {
			errF := func(wg *sync.WaitGroup, ch chan error, e error) {
				ch <- e
				wg.Done()
			}
			canonicalErrF := func(c Connection, e error) error {
				return fmt.Errorf("client: %q error: %s", c.Address, e)
			}
			address, err := util.NewConnection(c.Address)
			if nil != err {
				log.Errorf("client[%d]=address: %q  error: %s", i, c.Address, err)
				errF(wg, ch, canonicalErrF(c, err))
				return
			}
			serverPublicKey, err := zmqutil.ReadPublicKey(c.PublicKey)
			if nil != err {
				log.Errorf("client[%d]=public: %q  error: %s", i, c.PublicKey, err)
				errF(wg, ch, canonicalErrF(c, err))
				return
			}
			if bytes.Equal(publicKey, serverPublicKey) {
				err := fault.ConnectingToSelfForbidden
				log.Errorf("client[%d]=public: %q  error: %s", i, c.PublicKey, err)
				errF(wg, ch, canonicalErrF(c, err))
				return
			}
			client, err := upstream.New(privateKey, publicKey, connectorTimeout)
			if nil != err {
				log.Errorf("client[%d]=%q  error: %s", i, address, err)
				errF(wg, ch, canonicalErrF(c, err))
				return
			}
			conn.Lock()
			conn.staticClients[i] = client
			globalData.connectorClients = append(globalData.connectorClients, client)
			conn.Unlock()
			err = client.Connect(address, serverPublicKey)
			if nil != err {
				log.Errorf("connect[%d]=%q  error: %s", i, address, err)
				errF(wg, ch, canonicalErrF(c, err))
				return
			}
			log.Infof("public key: %x  at: %q", serverPublicKey, c.Address)
			wg.Done()
		}(conn, c, i, &wg, errCh)
	}
	conn.log.Debug("waiting for all static connections...")
	wg.Wait()
	errs := make([]error, 0)
	for len(errCh) > 0 {
		errs = append(errs, <-errCh)
	}
	err := error(nil)
	if len(errs) == 1 {
		err = errs[0]
		goto fail
	} else if len(errs) > 1 {
		err = compositeError(errs)
		goto fail
	}
	for i := 0; i < maximumDynamicClients; i++ {
		client, e := upstream.New(privateKey, publicKey, connectorTimeout)
		if nil != err {
			log.Errorf("client[%d]  error: %s", i, e)
			err = e
			goto fail
		}
		conn.dynamicClients.PushBack(client)
		globalData.connectorClients = append(globalData.connectorClients, client)
	}
	conn.votes = voting.NewVoting()
	conn.nextState(cStateConnecting)
	return nil
fail:
	conn.destroy()
	return err
}
func compositeError(errors []error) error {
	if nil == errors || 0 == len(errors) {
		return nil
	}
	var ce strings.Builder
	ce.WriteString("composite error: [")
	len := len(errors)
	for i, e := range errors {
		ce.WriteString(e.Error())
		if i < len-1 {
			ce.WriteString(", ")
		}
	}
	ce.WriteString("]")
	return fmt.Errorf(ce.String())
}
func (conn *connector) allClients(
	f func(client upstream.Upstream, e *list.Element),
) {
	for _, client := range conn.staticClients {
		if client != nil {
			f(client, nil)
		}
	}
	for e := conn.dynamicClients.Front(); nil != e; e = e.Next() {
		if client := e.Value.(upstream.Upstream); client != nil {
			f(client, e)
		}
	}
}
func (conn *connector) searchClients(
	f func(client upstream.Upstream, e *list.Element) bool,
) {
	for _, client := range conn.staticClients {
		if f(client, nil) {
			return
		}
	}
	for e := conn.dynamicClients.Front(); nil != e; e = e.Next() {
		if f(e.Value.(upstream.Upstream), e) {
			return
		}
	}
}
func (conn *connector) destroy() {
	conn.allClients(func(client upstream.Upstream, e *list.Element) {
		client.Destroy()
	})
}
func (conn *connector) Run(args interface{}, shutdown <-chan struct{}) {
	log := conn.log
	log.Info("starting…")
	queue := messagebus.Bus.Connector.Chan()
	timer := time.After(cycleInterval)
loop:
	for {
		log.Debug("waiting…")
		select {
		case <-shutdown:
			break loop
		case <-timer:  
			timer = time.After(cycleInterval)
			conn.process()
		case item := <-queue:
			c, _ := util.PackedConnection(item.Parameters[1]).Unpack()
			conn.log.Debugf(
				"received control: %s  public key: %x  connect: %x %q",
				item.Command,
				item.Parameters[0],
				item.Parameters[1],
				c,
			)
			switch item.Command {
			case "@D":  
				conn.releaseServerKey(item.Parameters[0])
				conn.log.Infof(
					"connector receive server public key: %x",
					item.Parameters[0],
				)
			default:
				err := conn.connectUpstream(
					item.Command,
					item.Parameters[0],
					item.Parameters[1],
				)
				if nil != err {
					conn.log.Warnf("connect upstream error: %s", err)
				}
			}
		}
	}
	log.Info("shutting down…")
	conn.destroy()
	log.Info("stopped")
}
func (conn *connector) process() {
	for conn.runStateMachine() {
	}
}