mirror of https://github.com/poanetwork/gecko.git
380 lines
10 KiB
Go
380 lines
10 KiB
Go
// (c) 2019-2020, Ava Labs, Inc. All rights reserved.
|
|
// See the file LICENSE for licensing terms.
|
|
|
|
package snowman
|
|
|
|
import (
|
|
"github.com/ava-labs/gecko/ids"
|
|
"github.com/ava-labs/gecko/snow"
|
|
"github.com/ava-labs/gecko/snow/choices"
|
|
"github.com/ava-labs/gecko/snow/consensus/snowman"
|
|
"github.com/ava-labs/gecko/snow/engine/common"
|
|
"github.com/ava-labs/gecko/snow/events"
|
|
"github.com/ava-labs/gecko/utils/formatting"
|
|
)
|
|
|
|
// Transitive implements the Engine interface by attempting to fetch all
|
|
// transitive dependencies.
|
|
type Transitive struct {
|
|
Config
|
|
bootstrapper
|
|
|
|
polls polls // track people I have asked for their preference
|
|
|
|
blkReqs, pending ids.Set // prevent asking validators for the same block
|
|
|
|
blocked events.Blocker // track operations that are blocked on blocks
|
|
|
|
bootstrapped bool
|
|
}
|
|
|
|
// Initialize implements the Engine interface
|
|
func (t *Transitive) Initialize(config Config) {
|
|
config.Context.Log.Info("Initializing Snowman consensus")
|
|
|
|
t.Config = config
|
|
t.metrics.Initialize(config.Context.Log, config.Params.Namespace, config.Params.Metrics)
|
|
|
|
t.onFinished = t.finishBootstrapping
|
|
t.bootstrapper.Initialize(config.BootstrapConfig)
|
|
|
|
t.polls.log = config.Context.Log
|
|
t.polls.numPolls = t.numPolls
|
|
t.polls.alpha = t.Params.Alpha
|
|
t.polls.m = make(map[uint32]poll)
|
|
}
|
|
|
|
func (t *Transitive) finishBootstrapping() {
|
|
tail := t.Config.VM.LastAccepted()
|
|
t.Config.VM.SetPreference(tail)
|
|
t.Consensus.Initialize(t.Config.Context, t.Params, tail)
|
|
t.bootstrapped = true
|
|
}
|
|
|
|
// Shutdown implements the Engine interface
|
|
func (t *Transitive) Shutdown() {
|
|
t.Config.Context.Log.Info("Shutting down Snowman consensus")
|
|
t.Config.VM.Shutdown()
|
|
}
|
|
|
|
// Context implements the Engine interface
|
|
func (t *Transitive) Context() *snow.Context { return t.Config.Context }
|
|
|
|
// Get implements the Engine interface
|
|
func (t *Transitive) Get(vdr ids.ShortID, requestID uint32, blkID ids.ID) {
|
|
if blk, err := t.Config.VM.GetBlock(blkID); err == nil {
|
|
t.Config.Sender.Put(vdr, requestID, blkID, blk.Bytes())
|
|
}
|
|
}
|
|
|
|
// Put implements the Engine interface
|
|
func (t *Transitive) Put(vdr ids.ShortID, requestID uint32, blkID ids.ID, blkBytes []byte) {
|
|
t.Config.Context.Log.Verbo("Put called for blockID %s", blkID)
|
|
|
|
if !t.bootstrapped {
|
|
t.bootstrapper.Put(vdr, requestID, blkID, blkBytes)
|
|
return
|
|
}
|
|
|
|
blk, err := t.Config.VM.ParseBlock(blkBytes)
|
|
if err != nil {
|
|
t.Config.Context.Log.Warn("ParseBlock failed due to %s for block:\n%s",
|
|
err,
|
|
formatting.DumpBytes{Bytes: blkBytes})
|
|
t.GetFailed(vdr, requestID, blkID)
|
|
return
|
|
}
|
|
|
|
t.insertFrom(vdr, blk)
|
|
}
|
|
|
|
// GetFailed implements the Engine interface
|
|
func (t *Transitive) GetFailed(vdr ids.ShortID, requestID uint32, blkID ids.ID) {
|
|
if !t.bootstrapped {
|
|
t.bootstrapper.GetFailed(vdr, requestID, blkID)
|
|
return
|
|
}
|
|
|
|
t.pending.Remove(blkID)
|
|
t.blocked.Abandon(blkID)
|
|
t.blkReqs.Remove(blkID)
|
|
|
|
// Tracks performance statistics
|
|
t.numBlockedBlk.Set(float64(t.pending.Len()))
|
|
}
|
|
|
|
// PullQuery implements the Engine interface
|
|
func (t *Transitive) PullQuery(vdr ids.ShortID, requestID uint32, blkID ids.ID) {
|
|
if !t.bootstrapped {
|
|
t.Config.Context.Log.Debug("Dropping PullQuery for %s due to bootstrapping", blkID)
|
|
return
|
|
}
|
|
|
|
c := &convincer{
|
|
consensus: t.Consensus,
|
|
sender: t.Config.Sender,
|
|
vdr: vdr,
|
|
requestID: requestID,
|
|
}
|
|
|
|
if !t.reinsertFrom(vdr, blkID) {
|
|
c.deps.Add(blkID)
|
|
}
|
|
|
|
t.blocked.Register(c)
|
|
}
|
|
|
|
// PushQuery implements the Engine interface
|
|
func (t *Transitive) PushQuery(vdr ids.ShortID, requestID uint32, blkID ids.ID, blk []byte) {
|
|
if !t.bootstrapped {
|
|
t.Config.Context.Log.Debug("Dropping PushQuery for %s due to bootstrapping", blkID)
|
|
return
|
|
}
|
|
|
|
t.Put(vdr, requestID, blkID, blk)
|
|
t.PullQuery(vdr, requestID, blkID)
|
|
}
|
|
|
|
// Chits implements the Engine interface
|
|
func (t *Transitive) Chits(vdr ids.ShortID, requestID uint32, votes ids.Set) {
|
|
if !t.bootstrapped {
|
|
t.Config.Context.Log.Warn("Dropping Chits due to bootstrapping")
|
|
return
|
|
}
|
|
|
|
// Since this is snowman, there should only be one ID in the vote set
|
|
if votes.Len() != 1 {
|
|
t.Config.Context.Log.Warn("Chits was called with the wrong number of votes %d. ValidatorID: %s, RequestID: %d", votes.Len(), vdr, requestID)
|
|
t.QueryFailed(vdr, requestID)
|
|
return
|
|
}
|
|
vote := votes.List()[0]
|
|
|
|
t.Config.Context.Log.Verbo("Chit was called. RequestID: %v. Vote: %s", requestID, vote)
|
|
|
|
v := &voter{
|
|
t: t,
|
|
vdr: vdr,
|
|
requestID: requestID,
|
|
response: vote,
|
|
}
|
|
|
|
if !t.reinsertFrom(vdr, vote) {
|
|
v.deps.Add(vote)
|
|
}
|
|
|
|
t.blocked.Register(v)
|
|
}
|
|
|
|
// QueryFailed implements the Engine interface
|
|
func (t *Transitive) QueryFailed(vdr ids.ShortID, requestID uint32) {
|
|
if !t.bootstrapped {
|
|
t.Config.Context.Log.Warn("Dropping QueryFailed due to bootstrapping")
|
|
return
|
|
}
|
|
|
|
t.blocked.Register(&voter{
|
|
t: t,
|
|
vdr: vdr,
|
|
requestID: requestID,
|
|
})
|
|
}
|
|
|
|
// Notify implements the Engine interface
|
|
func (t *Transitive) Notify(msg common.Message) {
|
|
if !t.bootstrapped {
|
|
t.Config.Context.Log.Warn("Dropping Notify due to bootstrapping")
|
|
return
|
|
}
|
|
|
|
t.Config.Context.Log.Verbo("Snowman engine notified of %s from the vm", msg)
|
|
switch msg {
|
|
case common.PendingTxs:
|
|
if blk, err := t.Config.VM.BuildBlock(); err == nil {
|
|
if status := blk.Status(); status != choices.Processing {
|
|
t.Config.Context.Log.Warn("Attempting to issue a block with status: %s, expected Processing", status)
|
|
}
|
|
parentID := blk.Parent().ID()
|
|
if pref := t.Consensus.Preference(); !parentID.Equals(pref) {
|
|
t.Config.Context.Log.Warn("Built block with parent: %s, expected %s", parentID, pref)
|
|
}
|
|
if t.insertAll(blk) {
|
|
t.Config.Context.Log.Verbo("Successfully issued new block from the VM")
|
|
} else {
|
|
t.Config.Context.Log.Warn("VM.BuildBlock returned a block that is pending for ancestors")
|
|
}
|
|
} else {
|
|
t.Config.Context.Log.Verbo("VM.BuildBlock errored with %s", err)
|
|
}
|
|
default:
|
|
t.Config.Context.Log.Warn("Unexpected message from the VM: %s", msg)
|
|
}
|
|
}
|
|
|
|
func (t *Transitive) repoll() {
|
|
prefID := t.Consensus.Preference()
|
|
t.pullSample(prefID)
|
|
}
|
|
|
|
func (t *Transitive) reinsertFrom(vdr ids.ShortID, blkID ids.ID) bool {
|
|
blk, err := t.Config.VM.GetBlock(blkID)
|
|
if err != nil {
|
|
t.sendRequest(vdr, blkID)
|
|
return false
|
|
}
|
|
return t.insertFrom(vdr, blk)
|
|
}
|
|
|
|
func (t *Transitive) insertFrom(vdr ids.ShortID, blk snowman.Block) bool {
|
|
blkID := blk.ID()
|
|
for !t.Consensus.Issued(blk) && !t.pending.Contains(blkID) {
|
|
t.insert(blk)
|
|
|
|
parent := blk.Parent()
|
|
parentID := parent.ID()
|
|
if parentStatus := parent.Status(); !parentStatus.Fetched() {
|
|
t.sendRequest(vdr, parentID)
|
|
return false
|
|
}
|
|
|
|
blk = parent
|
|
blkID = parentID
|
|
}
|
|
return !t.pending.Contains(blkID)
|
|
}
|
|
|
|
func (t *Transitive) insertAll(blk snowman.Block) bool {
|
|
blkID := blk.ID()
|
|
for blk.Status().Fetched() && !t.Consensus.Issued(blk) && !t.pending.Contains(blkID) {
|
|
t.insert(blk)
|
|
blk = blk.Parent()
|
|
}
|
|
return !t.pending.Contains(blkID)
|
|
}
|
|
|
|
func (t *Transitive) insert(blk snowman.Block) {
|
|
blkID := blk.ID()
|
|
|
|
t.pending.Add(blkID)
|
|
t.blkReqs.Remove(blkID)
|
|
|
|
i := &issuer{
|
|
t: t,
|
|
blk: blk,
|
|
}
|
|
|
|
if parent := blk.Parent(); !t.Consensus.Issued(parent) {
|
|
parentID := parent.ID()
|
|
t.Config.Context.Log.Verbo("Block waiting for parent %s", parentID)
|
|
i.deps.Add(parentID)
|
|
}
|
|
|
|
t.blocked.Register(i)
|
|
|
|
// Tracks performance statistics
|
|
t.numBlkRequests.Set(float64(t.blkReqs.Len()))
|
|
t.numBlockedBlk.Set(float64(t.pending.Len()))
|
|
}
|
|
|
|
func (t *Transitive) sendRequest(vdr ids.ShortID, blkID ids.ID) {
|
|
if !t.blkReqs.Contains(blkID) {
|
|
t.blkReqs.Add(blkID)
|
|
|
|
t.numBlkRequests.Set(float64(t.blkReqs.Len())) // Tracks performance statistics
|
|
|
|
t.RequestID++
|
|
t.Config.Context.Log.Verbo("Sending Get message for %s", blkID)
|
|
t.Config.Sender.Get(vdr, t.RequestID, blkID)
|
|
}
|
|
}
|
|
|
|
func (t *Transitive) pullSample(blkID ids.ID) {
|
|
t.Config.Context.Log.Verbo("About to sample from: %s", t.Config.Validators)
|
|
p := t.Consensus.Parameters()
|
|
vdrs := t.Config.Validators.Sample(p.K)
|
|
vdrSet := ids.ShortSet{}
|
|
for _, vdr := range vdrs {
|
|
vdrSet.Add(vdr.ID())
|
|
}
|
|
|
|
t.RequestID++
|
|
if numVdrs := len(vdrs); numVdrs == p.K && t.polls.Add(t.RequestID, vdrSet.Len()) {
|
|
t.Config.Sender.PullQuery(vdrSet, t.RequestID, blkID)
|
|
} else if numVdrs < p.K {
|
|
t.Config.Context.Log.Error("Query for %s was dropped due to an insufficient number of validators", blkID)
|
|
}
|
|
}
|
|
|
|
func (t *Transitive) pushSample(blk snowman.Block) {
|
|
t.Config.Context.Log.Verbo("About to sample from: %s", t.Config.Validators)
|
|
p := t.Consensus.Parameters()
|
|
vdrs := t.Config.Validators.Sample(p.K)
|
|
vdrSet := ids.ShortSet{}
|
|
for _, vdr := range vdrs {
|
|
vdrSet.Add(vdr.ID())
|
|
}
|
|
|
|
t.RequestID++
|
|
if numVdrs := len(vdrs); numVdrs == p.K && t.polls.Add(t.RequestID, vdrSet.Len()) {
|
|
t.Config.Sender.PushQuery(vdrSet, t.RequestID, blk.ID(), blk.Bytes())
|
|
} else if numVdrs < p.K {
|
|
t.Config.Context.Log.Error("Query for %s was dropped due to an insufficient number of validators", blk.ID())
|
|
}
|
|
}
|
|
|
|
func (t *Transitive) deliver(blk snowman.Block) {
|
|
if t.Consensus.Issued(blk) {
|
|
return
|
|
}
|
|
|
|
blkID := blk.ID()
|
|
t.pending.Remove(blkID)
|
|
|
|
if err := blk.Verify(); err != nil {
|
|
t.Config.Context.Log.Debug("Block failed verification due to %s, dropping block", err)
|
|
t.blocked.Abandon(blkID)
|
|
t.numBlockedBlk.Set(float64(t.pending.Len())) // Tracks performance statistics
|
|
return
|
|
}
|
|
|
|
t.Config.Context.Log.Verbo("Adding block to consensus: %s", blkID)
|
|
|
|
t.Consensus.Add(blk)
|
|
t.pushSample(blk)
|
|
|
|
added := []snowman.Block{}
|
|
dropped := []snowman.Block{}
|
|
switch blk := blk.(type) {
|
|
case OracleBlock:
|
|
for _, blk := range blk.Options() {
|
|
if err := blk.Verify(); err != nil {
|
|
t.Config.Context.Log.Debug("Block failed verification due to %s, dropping block", err)
|
|
t.blocked.Abandon(blk.ID())
|
|
dropped = append(dropped, blk)
|
|
} else {
|
|
t.Consensus.Add(blk)
|
|
t.pushSample(blk)
|
|
added = append(added, blk)
|
|
}
|
|
}
|
|
}
|
|
|
|
t.Config.VM.SetPreference(t.Consensus.Preference())
|
|
t.blocked.Fulfill(blkID)
|
|
|
|
for _, blk := range added {
|
|
blkID := blk.ID()
|
|
t.pending.Remove(blkID)
|
|
t.blocked.Fulfill(blkID)
|
|
}
|
|
for _, blk := range dropped {
|
|
blkID := blk.ID()
|
|
t.pending.Remove(blkID)
|
|
t.blocked.Abandon(blkID)
|
|
}
|
|
|
|
// Tracks performance statistics
|
|
t.numBlkRequests.Set(float64(t.blkReqs.Len()))
|
|
t.numBlockedBlk.Set(float64(t.pending.Len()))
|
|
}
|