func (h *Hare) Start(ctx context.Context) error {
	h.WithContext(ctx).With().Info("starting protocol", log.String("protocol", protoName))
	ctxBroker := log.WithNewSessionID(ctx, log.String("protocol", protoName+"_broker"))
	ctxTickLoop := log.WithNewSessionID(ctx, log.String("protocol", protoName+"_tickloop"))
	ctxOutputLoop := log.WithNewSessionID(ctx, log.String("protocol", protoName+"_outputloop"))
	if err := h.broker.Start(ctxBroker); err != nil {
		return fmt.Errorf("start broker: %w", err)
	}
	go h.tickLoop(ctxTickLoop)
	go h.outputCollectionLoop(ctxOutputLoop)
	return nil
}
Start starts listening for layers and outputs.
package hare
import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
	"github.com/libp2p/go-libp2p-core/peer"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/spacemeshos/go-spacemesh/blocks"
	"github.com/spacemeshos/go-spacemesh/common/types"
	"github.com/spacemeshos/go-spacemesh/common/util"
	"github.com/spacemeshos/go-spacemesh/database"
	"github.com/spacemeshos/go-spacemesh/hare/config"
	"github.com/spacemeshos/go-spacemesh/hare/metrics"
	"github.com/spacemeshos/go-spacemesh/log"
	"github.com/spacemeshos/go-spacemesh/p2p/pubsub"
)
const LayerBuffer = 20
type consensusFactory func(cfg config.Config, instanceId types.LayerID, s *Set, oracle Rolacle, signing Signer, p2p pubsub.Publisher, clock RoundClock, terminationReport chan TerminationOutput) Consensus
type Consensus interface {
	ID() types.LayerID
	Close()
	CloseChannel() chan struct{}
	Start(ctx context.Context) error
	SetInbox(chan *Msg)
}
type TerminationOutput interface {
	ID() types.LayerID
	Set() *Set
	Coinflip() bool
	Completed() bool
}
type RoundClock interface {
	AwaitWakeup() <-chan struct{}
	AwaitEndOfRound(round uint32) <-chan struct{}
}
type LayerClock interface {
	LayerToTime(id types.LayerID) time.Time
	AwaitLayer(layerID types.LayerID) chan struct{}
	GetCurrentLayer() types.LayerID
}
type Hare struct {
	util.Closer
	log.Log
	config config.Config
	publisher  pubsub.Publisher
	layerClock LayerClock
	newRoundClock func(LayerID types.LayerID) RoundClock
	broker *Broker
	sign Signer
	mesh    meshProvider
	beacons blocks.BeaconGetter
	rolacle Rolacle
	patrol  layerPatrol
	networkDelta time.Duration
	layerLock sync.RWMutex
	lastLayer types.LayerID
	bufferSize uint32
	outputChan chan TerminationOutput
	mu         sync.RWMutex
	outputs    map[types.LayerID][]types.BlockID
	factory consensusFactory
	nid types.NodeID
	totalCPs int32
}
func New(
	conf config.Config,
	pid peer.ID,
	publisher pubsub.PublishSubsciber,
	sign Signer,
	nid types.NodeID,
	syncState syncStateFunc,
	mesh meshProvider,
	beacons blocks.BeaconGetter,
	rolacle Rolacle,
	patrol layerPatrol,
	layersPerEpoch uint16,
	idProvider identityProvider,
	stateQ StateQuerier,
	layerClock LayerClock,
	logger log.Log,
) *Hare {
	h := new(Hare)
	h.Closer = util.NewCloser()
	h.Log = logger
	h.config = conf
	h.publisher = publisher
	h.layerClock = layerClock
	h.newRoundClock = func(layerID types.LayerID) RoundClock {
		layerTime := layerClock.LayerToTime(layerID)
		wakeupDelta := time.Duration(conf.WakeupDelta) * time.Second
		roundDuration := time.Duration(h.config.RoundDuration) * time.Second
		h.With().Info("creating hare round clock", layerID,
			log.String("layer_time", layerTime.String()),
			log.Duration("wakeup_delta", wakeupDelta),
			log.Duration("round_duration", roundDuration))
		return NewSimpleRoundClock(layerTime, wakeupDelta, roundDuration)
	}
	ev := newEligibilityValidator(rolacle, layersPerEpoch, idProvider, conf.N, conf.ExpectedLeaders, logger)
	h.broker = newBroker(pid, ev, stateQ, syncState, layersPerEpoch, conf.LimitConcurrent, h.Closer, logger)
	publisher.Register(protoName, h.broker.HandleMessage)
	h.sign = sign
	h.mesh = mesh
	h.beacons = beacons
	h.rolacle = rolacle
	h.patrol = patrol
	h.networkDelta = time.Duration(conf.WakeupDelta) * time.Second
	h.bufferSize = LayerBuffer  
	h.outputChan = make(chan TerminationOutput, h.bufferSize)
	h.outputs = make(map[types.LayerID][]types.BlockID, h.bufferSize)  
	h.factory = func(conf config.Config, instanceId types.LayerID, s *Set, oracle Rolacle, signing Signer, p2p pubsub.Publisher, clock RoundClock, terminationReport chan TerminationOutput) Consensus {
		return newConsensusProcess(conf, instanceId, s, oracle, stateQ, layersPerEpoch, signing, nid, p2p, terminationReport, ev, clock, logger)
	}
	h.nid = nid
	return h
}
func (h *Hare) getLastLayer() types.LayerID {
	h.layerLock.RLock()
	defer h.layerLock.RUnlock()
	return h.lastLayer
}
func (h *Hare) outOfBufferRange(id types.LayerID) bool {
	last := h.getLastLayer()
	if !last.After(types.NewLayerID(h.bufferSize)) {
		return false
	}
	if id.Before(last.Sub(h.bufferSize)) {  
		return true
	}
	return false
}
func (h *Hare) oldestResultInBuffer() types.LayerID {
	lyr := h.getLastLayer()
	for k := range h.outputs {
		if k.Before(lyr) {
			lyr = k
		}
	}
	return lyr
}
var ErrTooLate = errors.New("consensus process finished too late")
func (h *Hare) collectOutput(ctx context.Context, output TerminationOutput) error {
	set := output.Set()
	blocks := make([]types.BlockID, len(set.values))
	i := 0
	for v := range set.values {
		blocks[i] = v
		i++
	}
	id := output.ID()
	h.mesh.HandleValidatedLayer(ctx, id, blocks)
	if h.outOfBufferRange(id) {
		return ErrTooLate
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	if uint32(len(h.outputs)) >= h.bufferSize {
		delete(h.outputs, h.oldestResultInBuffer())
	}
	h.outputs[id] = blocks
	return nil
}
func (h *Hare) onTick(ctx context.Context, id types.LayerID) (bool, error) {
	logger := h.WithContext(ctx).WithFields(id)
	if h.IsClosed() {
		logger.Info("hare exiting")
		return false, nil
	}
	h.layerLock.Lock()
	if id.After(h.lastLayer) {
		h.lastLayer = id
	} else {
		logger.With().Error("received out of order layer tick", log.FieldNamed("last_layer", h.lastLayer))
	}
	h.layerLock.Unlock()
	if id.GetEpoch().IsGenesis() {
		logger.Info("not starting hare since we are in genesis epoch")
		return false, nil
	}
	var err error
	defer func() {
		if err != nil {
			h.outputChan <- procReport{id, &Set{}, false, notCompleted}
		}
	}()
	beacon, err := h.beacons.GetBeacon(id.GetEpoch())
	if err != nil {
		logger.Info("not starting hare since beacon is not retrieved")
		return false, nil
	}
	go func() {
		if isActive, err := h.rolacle.IsIdentityActiveOnConsensusView(ctx, h.nid.Key, id); err != nil {
			logger.With().Error("error checking if identity is active",
				log.Bool("isActive", isActive), log.Err(err))
		}
	}()
	logger.With().Debug("hare got tick, sleeping", log.String("delta", fmt.Sprint(h.networkDelta)))
	clock := h.newRoundClock(id)
	select {
	case <-clock.AwaitWakeup():
		break  
	case <-h.CloseChannel():
		return false, errors.New("closed while waiting for hare delta")
	}
	if !h.broker.Synced(ctx, id) {
		logger.Info("not processing hare tick since node is not synced")
		return false, nil
	}
	blocks := h.getGoodBlocks(h.lastLayer, beacon, logger)
	logger.With().Info("starting hare consensus with blocks", log.Int("num_blocks", len(blocks)))
	set := NewSet(blocks)
	instID := id
	c, err := h.broker.Register(ctx, instID)
	if err != nil {
		logger.With().Error("could not register consensus process on broker", log.Err(err))
		return false, fmt.Errorf("broker register: %w", err)
	}
	cp := h.factory(h.config, instID, set, h.rolacle, h.sign, h.publisher, clock, h.outputChan)
	cp.SetInbox(c)
	if err = cp.Start(ctx); err != nil {
		logger.With().Error("could not start consensus process", log.Err(err))
		h.broker.Unregister(ctx, cp.ID())
		return false, fmt.Errorf("start consensus: %w", err)
	}
	h.patrol.SetHareInCharge(instID)
	logger.With().Info("number of consensus processes (after register)",
		log.Int32("count", atomic.AddInt32(&h.totalCPs, 1)))
	metrics.TotalConsensusProcesses.With(prometheus.Labels{"layer": id.String()}).Inc()
	return true, nil
}
func (h *Hare) getGoodBlocks(lyrID types.LayerID, epochBeacon []byte, logger log.Log) []types.BlockID {
	blocks, err := h.mesh.LayerBlocks(lyrID)
	if err != nil {
		if err != database.ErrNotFound {
			logger.With().Error("no blocks found for hare, using empty set", log.Err(err))
		}
		return []types.BlockID{}
	}
	var goodBlocks []types.BlockID
	for _, b := range blocks {
		beacon := b.TortoiseBeacon
		if b.RefBlock != nil {
			refBlock, err := h.mesh.GetBlock(*b.RefBlock)
			if err != nil {
				logger.With().Error("failed to find ref block",
					b.ID(),
					log.String("ref_block_id", b.RefBlock.AsHash32().ShortString()))
				return []types.BlockID{}
			}
			beacon = refBlock.TortoiseBeacon
		}
		if bytes.Equal(beacon, epochBeacon) {
			goodBlocks = append(goodBlocks, b.ID())
		} else {
			logger.With().Warning("block has different beacon value",
				b.ID(),
				log.String("block_beacon", types.BytesToHash(beacon).ShortString()),
				log.String("epoch_beacon", types.BytesToHash(epochBeacon).ShortString()))
		}
	}
	return goodBlocks
}
var (
	errTooOld   = errors.New("layer has already been evacuated from buffer")
	errNoResult = errors.New("no result for the requested layer")
)
func (h *Hare) GetResult(lid types.LayerID) ([]types.BlockID, error) {
	if h.outOfBufferRange(lid) {
		return nil, errTooOld
	}
	h.mu.RLock()
	defer h.mu.RUnlock()
	blks, ok := h.outputs[lid]
	if !ok {
		return nil, errNoResult
	}
	return blks, nil
}
func (h *Hare) outputCollectionLoop(ctx context.Context) {
	h.WithContext(ctx).With().Info("starting collection loop")
	for {
		select {
		case out := <-h.outputChan:
			layerID := out.ID()
			coin := out.Coinflip()
			ctx := log.WithNewSessionID(ctx)
			logger := h.WithContext(ctx).WithFields(layerID)
			logger.With().Info("recording weak coinflip result for layer",
				log.Bool("coinflip", coin))
			h.mesh.RecordCoinflip(ctx, layerID, coin)
			if out.Completed() {  
				logger.With().Info("collecting results for completed hare instance", layerID)
				if err := h.collectOutput(ctx, out); err != nil {
					logger.With().Warning("error collecting output from hare", log.Err(err))
				}
			} else {
				h.WithContext(ctx).With().Info("recording hare instance failure", layerID)
				h.mesh.InvalidateLayer(ctx, layerID)
			}
			h.broker.Unregister(ctx, out.ID())
			logger.With().Info("number of consensus processes (after unregister)",
				log.Int32("count", atomic.AddInt32(&h.totalCPs, -1)))
			metrics.TotalConsensusProcesses.With(prometheus.Labels{"layer": out.ID().String()}).Dec()
		case <-h.CloseChannel():
			return
		}
	}
}
func (h *Hare) tickLoop(ctx context.Context) {
	for layer := h.layerClock.GetCurrentLayer(); ; layer = layer.Add(1) {
		select {
		case <-h.layerClock.AwaitLayer(layer):
			if h.layerClock.LayerToTime(layer).Sub(time.Now()) > (time.Duration(h.config.WakeupDelta) * time.Second) {
				h.With().Warning("missed hare window, skipping layer", layer)
				continue
			}
			go func(l types.LayerID) {
				started, err := h.onTick(ctx, l)
				if err != nil {
					h.With().Error("failed to handle tick", log.Err(err))
				} else if !started {
					h.WithContext(ctx).With().Warning("consensus not started for layer", layer)
				}
			}(layer)
		case <-h.CloseChannel():
			return
		}
	}
}