path: root/core/total-ordering.go

// Copyright 2018 The dexon-consensus-core Authors
// This file is part of the dexon-consensus-core library.
//
// The dexon-consensus-core library is free software: you can redistribute it
// and/or modify it under the terms of the GNU Lesser General Public License as
// published by the Free Software Foundation, either version 3 of the License,
// or (at your option) any later version.
//
// The dexon-consensus-core library is distributed in the hope that it will be
// useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
// General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the dexon-consensus-core library. If not, see
// <http://www.gnu.org/licenses/>.

package core

import (
    "fmt"
    "math"
    "sort"

    "github.com/dexon-foundation/dexon-consensus-core/common"
    "github.com/dexon-foundation/dexon-consensus-core/core/types"
)

const (
    infinity uint64 = math.MaxUint64
)

// ErrNotValidDAG would be reported when block subbmitted to totalOrdering
// didn't form a DAG.
var ErrNotValidDAG = fmt.Errorf("not a valid dag")

// ackingStatusVector describes the acking status, either globally or just
// for one candidate.
//
// When block A acks block B, all blocks proposed from the same proposer
// as block A with higher height would also acks block B. Therefore,
// we just need to record:
//  - the minimum height of acking block from that proposer
//  - count of acking blocks from that proposer
// to repsent the acking status for block A.
type ackingStatusVector map[types.ValidatorID]*struct{ minHeight, count uint64 }

// addBlock would update ackingStatusVector, it's caller's duty
// to make sure the input block acutally acking the target block.
func (v ackingStatusVector) addBlock(b *types.Block) (err error) {
    rec, exists := v[b.ProposerID]
    if !exists {
        v[b.ProposerID] = &struct {
            minHeight, count uint64
        }{
            minHeight: b.Height,
            count:     1,
        }
    } else {
        if b.Height < rec.minHeight {
            err = ErrNotValidDAG
            return
        }
        rec.count++
    }
    return
}

// getAckingNodeSet would generate the Acking Node Set.
// Only block height larger than
//
//    global minimum height + k
//
// would be taken into consideration, ex.
//
//  For some validator X:
//   - the global minimum acking height = 1,
//   - k = 1
//  then only block height >= 2 would be added to acking node set.
func (v ackingStatusVector) getAckingNodeSet(
    global ackingStatusVector, k uint64) map[types.ValidatorID]struct{} {

    ret := make(map[types.ValidatorID]struct{})
    for vID, gRec := range global {
        rec, exists := v[vID]
        if !exists {
            continue
        }

        // This line would check if these two ranges would overlap:
        //  - (global minimum height + k, infinity)
        //  - (local minimum height, local minimum height + count - 1)
        if rec.minHeight+rec.count-1 >= gRec.minHeight+k {
            ret[vID] = struct{}{}
        }
    }
    return ret
}

// getAckingHeightVector would convert 'ackingStatusVector' to
// Acking Height Vector.
//
// Only block height equals to (global minimum block height + k) would be
// taken into consideration.
func (v ackingStatusVector) getAckingHeightVector(
    global ackingStatusVector, k uint64) map[types.ValidatorID]uint64 {

    ret := make(map[types.ValidatorID]uint64)
    for vID, gRec := range global {
        rec, exists := v[vID]

        if gRec.count <= k {
            continue
        } else if !exists {
            ret[vID] = infinity
        } else if rec.minHeight <= gRec.minHeight+k {
            // This check is sufficient to make sure the block height:
            //
            //   gRec.minHeight + k
            //
            // would be included in this ackingStatusVector.
            ret[vID] = gRec.minHeight + k
        } else {
            ret[vID] = infinity
        }
    }
    return ret
}

// blockVector stores all blocks grouped by their proposers and
// sorted by their block height.
type blockVector map[types.ValidatorID][]*types.Block

func (v blockVector) addBlock(b *types.Block) (err error) {
    blocksFromProposer := v[b.ProposerID]
    if len(blocksFromProposer) > 0 {
        lastBlock := blocksFromProposer[len(blocksFromProposer)-1]
        if b.Height-lastBlock.Height != 1 {
            err = ErrNotValidDAG
            return
        }
    }
    v[b.ProposerID] = append(blocksFromProposer, b)
    return
}

// getAckingStatusVector would convert a blockVector to
// ackingStatusVectorAckingStatus.
func (v blockVector) getAckingStatusVector() ackingStatusVector {
    ret := ackingStatusVector{}
    for vID, vec := range v {
        if len(vec) == 0 {
            continue
        }
        ret[vID] = &struct {
            minHeight, count uint64
        }{
            minHeight: vec[0].Height,
            count:     uint64(len(vec)),
        }
    }
    return ret
}

// totalOrdering represent a process unit to handle total ordering
// for blocks.
type totalOrdering struct {
    // pendings stores blocks awaiting to be ordered.
    pendings map[common.Hash]*types.Block

    // k represents the k in 'k-level total ordering'.
    // In short, only block height equals to (global minimum height + k)
    // would be taken into consideration.
    k uint64

    // phi is a const to control how strong the leading preceding block
    // should be.
    phi uint64

    // validatorCount is the count of validator set.
    validatorCount uint64

    // globalVector group all pending blocks by proposers and
    // sort them by block height. This structure is helpful when:
    //
    //  - build global height vector
    //  - picking candidates next round
    globalVector blockVector

    // candidateAckingStatusVectors caches ackingStatusVector of candidates.
    candidateAckingStatusVectors map[common.Hash]ackingStatusVector

    // acked cache the 'block A acked by block B' relation by
    // keeping a record in acked[A.Hash][B.Hash]
    acked map[common.Hash]map[common.Hash]struct{}
}

func newTotalOrdering(k, phi, validatorCount uint64) *totalOrdering {
    return &totalOrdering{
        candidateAckingStatusVectors: make(map[common.Hash]ackingStatusVector),
        pendings:                     make(map[common.Hash]*types.Block),
        k:                            k,
        phi:                          phi,
        validatorCount:               validatorCount,
        globalVector:                 blockVector{},
        acked:                        make(map[common.Hash]map[common.Hash]struct{}),
    }
}

// buildBlockRelation populates the acked according their acking relationships.
func (to *totalOrdering) buildBlockRelation(b *types.Block) {
    // populateAcked would update all blocks implcitly acked
    // by input block recursively.
    var populateAcked func(bx, target *types.Block)
    populateAcked = func(bx, target *types.Block) {
        for ack := range bx.Acks {
            acked, exists := to.acked[ack]
            if !exists {
                acked = make(map[common.Hash]struct{})
                to.acked[ack] = acked
            }

            // This means we've walked this block already.
            if _, alreadyPopulated := acked[target.Hash]; alreadyPopulated {
                continue
            }
            acked[target.Hash] = struct{}{}

            // See if we need to go forward.
            if nextBlock, exists := to.pendings[ack]; !exists {
                continue
            } else {
                populateAcked(nextBlock, target)
            }
        }
    }
    populateAcked(b, b)
}

// clean would remove a block from working set. This behaviour
// would prevent our memory usage growing infinity.
func (to *totalOrdering) clean(h common.Hash) {
    delete(to.acked, h)
    delete(to.pendings, h)
    delete(to.candidateAckingStatusVectors, h)
}

// updateVectors is a helper function to update all cached vectors.
func (to *totalOrdering) updateVectors(b *types.Block) (err error) {
    // Update global height vector
    err = to.globalVector.addBlock(b)
    if err != nil {
        return
    }

    // Update acking status of candidates.
    for candidate, vector := range to.candidateAckingStatusVectors {
        if _, acked := to.acked[candidate][b.Hash]; !acked {
            continue
        }
        if err = vector.addBlock(b); err != nil {
            return
        }
    }
    return
}

// grade implements the 'grade' potential function described in white paper.
func (to *totalOrdering) grade(
    hvFrom, hvTo map[types.ValidatorID]uint64,
    globalAns map[types.ValidatorID]struct{}) int {

    count := uint64(0)
    for vID, hFrom := range hvFrom {
        hTo, exists := hvTo[vID]
        if !exists {
            continue
        }

        if hFrom != infinity && hTo == infinity {
            count++
        }
    }

    if count >= to.phi {
        return 1
    } else if count < to.phi-to.validatorCount+uint64(len(globalAns)) {
        return 0
    } else {
        return -1
    }
}

// buildAckingStatusVectorForNewCandidate is a helper function to
// build ackingStatusVector for new candidate.
func (to *totalOrdering) buildAckingStatusVectorForNewCandidate(
    candidate *types.Block) (hVec ackingStatusVector) {

    blocks := to.globalVector[candidate.ProposerID]
    hVec = ackingStatusVector{
        candidate.ProposerID: &struct {
            minHeight, count uint64
        }{
            minHeight: candidate.Height,
            count:     uint64(len(blocks)),
        },
    }

    ackedsForCandidate, exists := to.acked[candidate.Hash]
    if !exists {
        // This candidate is acked by nobody.
        return
    }

    for vID, blocks := range to.globalVector {
        if vID == candidate.ProposerID {
            continue
        }

        for i, b := range blocks {
            if _, acked := ackedsForCandidate[b.Hash]; !acked {
                continue
            }

            // If this block acks this candidate, all newer blocks
            // from the same validator also 'indirect' acks it.
            hVec[vID] = &struct {
                minHeight, count uint64
            }{
                minHeight: b.Height,
                count:     uint64(len(blocks) - i),
            }
            break
        }
    }
    return
}

// isAckOnlyPrecedings is a helper function to check if a block
// only contain acks to delivered blocks.
func (to *totalOrdering) isAckOnlyPrecedings(b *types.Block) bool {
    for ack := range b.Acks {
        if _, pending := to.pendings[ack]; pending {
            return false
        }
    }
    return true
}

// output is a helper function to finish the delivery of
// deliverable preceding set.
func (to *totalOrdering) output(precedings map[common.Hash]struct{}) (ret []*types.Block) {
    for p := range precedings {
        // Remove the first element from corresponding blockVector.
        b := to.pendings[p]
        to.globalVector[b.ProposerID] = to.globalVector[b.ProposerID][1:]
        ret = append(ret, b)

        // Remove block relations.
        to.clean(p)
    }
    sort.Sort(types.ByHash(ret))

    // Find new candidates from tip of globalVector of each validator.
    // The complexity here is O(N^2logN).
    for _, blocks := range to.globalVector {
        if len(blocks) == 0 {
            continue
        }

        tip := blocks[0]
        if _, alreadyCandidate :=
            to.candidateAckingStatusVectors[tip.Hash]; alreadyCandidate {
            continue
        }

        if !to.isAckOnlyPrecedings(tip) {
            continue
        }

        // Build ackingStatusVector for new candidate.
        to.candidateAckingStatusVectors[tip.Hash] =
            to.buildAckingStatusVectorForNewCandidate(tip)
    }
    return ret
}

// generateDeliverSet would:
//  - generate preceding set
//  - check if the preceding set deliverable by checking potential function
func (to *totalOrdering) generateDeliverSet() (
    delivered map[common.Hash]struct{}, early bool) {

    globalAckingStatusVector := to.globalVector.getAckingStatusVector()
    ahvs := map[common.Hash]map[types.ValidatorID]uint64{}
    for candidate, v := range to.candidateAckingStatusVectors {
        ahvs[candidate] = v.getAckingHeightVector(globalAckingStatusVector, to.k)
    }

    globalAns := globalAckingStatusVector.getAckingNodeSet(
        globalAckingStatusVector, to.k)
    precedings := make(map[common.Hash]struct{})

CheckNextCandidateLoop:
    for candidate := range to.candidateAckingStatusVectors {
        for otherCandidate := range to.candidateAckingStatusVectors {
            if candidate == otherCandidate {
                continue
            }
            if to.grade(ahvs[otherCandidate], ahvs[candidate], globalAns) != 0 {
                continue CheckNextCandidateLoop
            }
        }
        precedings[candidate] = struct{}{}
    }

    if len(precedings) == 0 {
        return
    }

    // internal is a helper function to verify internal stability.
    internal := func() bool {
        for candidate := range to.candidateAckingStatusVectors {
            if _, isPreceding := precedings[candidate]; isPreceding {
                continue
            }

            beaten := false
            for p := range precedings {
                if beaten =
                    to.grade(ahvs[p], ahvs[candidate], globalAns) == 1; beaten {
                    break
                }
            }
            if !beaten {
                return false
            }
        }
        return true
    }

    // checkAHV is a helper function to verify external stability.
    // It would make sure some preceding block is strong enough
    // to lead the whole preceding set.
    checkAHV := func() bool {
        for p := range precedings {
            count := uint64(0)
            for _, v := range ahvs[p] {
                if v != infinity {
                    count++
                }
            }

            if count > to.phi {
                return true
            }
        }
        return false
    }

    // checkANS is a helper function to verify external stability.
    // It would make sure all preceding blocks are strong enough
    // to be delivered.
    checkANS := func() bool {
        for p := range precedings {
            validatorAns := to.candidateAckingStatusVectors[p].getAckingNodeSet(
                globalAckingStatusVector, to.k)
            if uint64(len(validatorAns)) < to.validatorCount-to.phi {
                return false
            }
        }

        return true
    }

    // Check internal stability first.
    if !internal() {
        return
    }

    // If all validators propose enough blocks, we should force
    // to deliver since the whole picture of the DAG is revealed.
    if uint64(len(globalAns)) != to.validatorCount {
        // The whole picture is not ready, we need to check if
        // exteranl stability is met, and we can deliver earlier.
        if checkAHV() && checkANS() {
            early = true
        } else {
            return
        }
    }

    delivered = precedings
    return
}

// processBlock is the entry point of totalOrdering.
func (to *totalOrdering) processBlock(b *types.Block) (
    delivered []*types.Block, early bool, err error) {
    // NOTE: I assume the block 'b' is already safe for total ordering.
    //       That means, all its acking blocks are during/after
    //       total ordering stage.

    // Incremental part.
    to.pendings[b.Hash] = b
    to.buildBlockRelation(b)
    if err = to.updateVectors(b); err != nil {
        return
    }
    if to.isAckOnlyPrecedings(b) {
        to.candidateAckingStatusVectors[b.Hash] =
            to.buildAckingStatusVectorForNewCandidate(b)
    }

    // Not-Incremental part (yet).
    //  - generate ahv for each candidate
    //  - generate ans for each candidate
    //  - generate global ans
    //  - find preceding set
    hashes, early := to.generateDeliverSet()

    // output precedings
    delivered = to.output(hashes)
    return
}