aergoio/aergo

package syncer

import (
    "sync"
    "time"

    "github.com/aergoio/aergo/v2/internal/enc/base58"
    "github.com/aergoio/aergo/v2/pkg/component"
    "github.com/aergoio/aergo/v2/types"
    "github.com/aergoio/aergo/v2/types/message"
    "github.com/pkg/errors"
)

type HashFetcher struct {
    compRequester component.IComponentRequester //for communicate with other service

    ctx *types.SyncContext

    responseCh chan *message.GetHashesRsp //HashSet response channel (<- Syncer)
    resultCh   chan *HashSet              //BlockFetcher input channel (-> BlockFetcher)
    //HashFetcher can wait in resultCh
    quitCh chan interface{}

    lastBlockInfo *types.BlockInfo
    reqCount      uint64
    reqTime       time.Time
    isRequesting  bool

    maxHashReq uint64
    name       string

    timeout time.Duration
    debug   bool

    isRunning bool

    waitGroup *sync.WaitGroup
}

type HashSet struct {
    Count   int
    Hashes  []message.BlockHash
    StartNo types.BlockNo
}

type HashRequest struct {
    prevInfo *types.BlockInfo
    count    uint64
}

var (
    dfltTimeout     = time.Second * 180
    DfltHashReqSize = uint64(1000)
)

var (
    ErrQuitHashFetcher    = errors.New("Hashfetcher quit")
    ErrInvalidHashSet     = errors.New("Invalid hash set reply")
    ErrHashFetcherTimeout = errors.New("HashFetcher response timeout")
)

func newHashFetcher(ctx *types.SyncContext, compRequester component.IComponentRequester, bfCh chan *HashSet, cfg *SyncerConfig) *HashFetcher {
    hf := &HashFetcher{ctx: ctx, compRequester: compRequester, name: NameHashFetcher}

    hf.quitCh = make(chan interface{})
    hf.responseCh = make(chan *message.GetHashesRsp)

    hf.resultCh = bfCh

    hf.lastBlockInfo = &types.BlockInfo{Hash: ctx.CommonAncestor.GetHash(), No: ctx.CommonAncestor.BlockNo()}

    hf.maxHashReq = cfg.maxHashReqSize

    hf.timeout = dfltTimeout

    return hf
}

func (hf *HashFetcher) setTimeout(timeout time.Duration) {
    hf.timeout = timeout
}

func (hf *HashFetcher) recover() {

}
func (hf *HashFetcher) Start() {
    hf.waitGroup = &sync.WaitGroup{}
    hf.waitGroup.Add(1)

    hf.isRunning = true

    run := func() {
        defer RecoverSyncer(NameHashFetcher, hf.GetSeq(), hf.compRequester, func() { hf.waitGroup.Done() })

        logger.Debug().Msg("start hash fetcher")

        timer := time.NewTimer(hf.timeout)

        hf.requestHashSet()

        for {
            select {
            case msg, ok := <-hf.responseCh:
                if !ok {
                    logger.Error().Msg("HashFetcher responseCh is closed. Syncer is stopping now")
                    return
                }
                logger.Debug().Msg("process GetHashesRsp")

                timer.Stop()
                res, err := hf.isValidResponse(msg)
                if res {
                    HashSet := &HashSet{Count: len(msg.Hashes), Hashes: msg.Hashes, StartNo: msg.PrevInfo.No + 1}

                    if err := hf.processHashSet(HashSet); err != nil {
                        //TODO send errmsg to syncer & stop sync
                        logger.Error().Err(err).Msg("error! process hash chunk, HashFetcher exited")
                        if err != ErrQuitHashFetcher {
                            stopSyncer(hf.compRequester, hf.GetSeq(), hf.name, err)
                        }
                        return
                    }

                    if hf.isFinished(HashSet) {
                        closeFetcher(hf.compRequester, hf.GetSeq(), hf.name)
                        logger.Info().Msg("HashFetcher finished")
                        return
                    }
                    hf.requestHashSet()
                } else if err != nil {
                    stopSyncer(hf.compRequester, hf.GetSeq(), hf.name, err)
                }

                //timer restart
                timer.Reset(hf.timeout)
            case <-timer.C:
                if hf.requestTimeout() {
                    logger.Error().Msg("HashFetcher response timeout.")
                    stopSyncer(hf.compRequester, hf.GetSeq(), hf.name, ErrHashFetcherTimeout)
                }

            case <-hf.quitCh:
                logger.Info().Msg("HashFetcher exited")
                return
            }
        }
    }

    go run()
}

func (hf *HashFetcher) GetSeq() uint64 {
    return hf.ctx.Seq
}

func (hf *HashFetcher) isFinished(HashSet *HashSet) bool {
    return (hf.lastBlockInfo.No == hf.ctx.TargetNo)
}

func (hf *HashFetcher) requestTimeout() bool {
    return hf.isRequesting && time.Now().Sub(hf.reqTime) > hf.timeout
}

func (hf *HashFetcher) requestHashSet() {
    count := hf.maxHashReq
    if hf.ctx.TargetNo < hf.lastBlockInfo.No+hf.maxHashReq {
        count = hf.ctx.TargetNo - hf.lastBlockInfo.No
    }

    hf.reqCount = count
    hf.reqTime = time.Now()
    hf.isRequesting = true

    logger.Debug().Uint64("prev", hf.lastBlockInfo.No).Str("prevhash", base58.Encode(hf.lastBlockInfo.Hash)).Uint64("count", count).Msg("request hashset to peer")

    hf.compRequester.TellTo(message.P2PSvc, &message.GetHashes{Seq: hf.GetSeq(), ToWhom: hf.ctx.PeerID, PrevInfo: hf.lastBlockInfo, Count: count})
}

func (hf *HashFetcher) processHashSet(hashSet *HashSet) error {
    //get HashSet reply
    lastHash := hashSet.Hashes[len(hashSet.Hashes)-1]
    lastHashNo := hashSet.StartNo + uint64(hashSet.Count) - 1

    if lastHashNo > hf.ctx.TargetNo {
        logger.Error().Uint64("target", hf.ctx.TargetNo).Uint64("last", lastHashNo).Msg("invalid hashset reponse")
        return ErrInvalidHashSet
    }

    hf.lastBlockInfo = &types.BlockInfo{Hash: lastHash, No: lastHashNo}

    //total HashSet in memory can be 3 (network + resultCh + blockFetcher)
    select {
    case hf.resultCh <- hashSet:
    case <-hf.quitCh:
        logger.Info().Msg("hash fetcher quit while pushing result")
        return ErrQuitHashFetcher
    }

    hf.isRequesting = false

    logger.Debug().Uint64("target", hf.ctx.TargetNo).Uint64("start", hashSet.StartNo).Uint64("last", lastHashNo).Int("count", len(hashSet.Hashes)).Msg("push hashset to BlockFetcher")

    return nil
}

func (hf *HashFetcher) stop() {
    if hf == nil {
        return
    }

    if hf.isRunning {
        logger.Info().Msg("HashFetcher stop#1")

        close(hf.quitCh)
        logger.Info().Msg("HashFetcher close quitCh")

        close(hf.responseCh)

        hf.waitGroup.Wait()
        hf.isRunning = false
    }
    logger.Info().Msg("HashFetcher stopped")
}

func (hf *HashFetcher) isValidResponse(msg *message.GetHashesRsp) (bool, error) {
    isValid := true
    var err error

    if msg == nil {
        panic("nil message error")
    }

    if msg.Err != nil {
        logger.Error().Err(msg.Err).Msg("receive GetHashesRsp with error")
        err = msg.Err
        isValid = false
    }

    if !hf.lastBlockInfo.Equal(msg.PrevInfo) || hf.reqCount != msg.Count {
        isValid = false
    }

    if !isValid {
        logger.Error().Str("req prev", base58.Encode(hf.lastBlockInfo.Hash)).
            Str("msg prev", base58.Encode(msg.PrevInfo.Hash)).
            Uint64("req count", hf.reqCount).
            Uint64("msg count", msg.Count).
            Msg("invalid GetHashesRsp")
        return false, err
    }

    return true, nil
}

func (hf *HashFetcher) GetHahsesRsp(msg *message.GetHashesRsp) {
    if hf == nil {
        return
    }

    count := len(msg.Hashes)

    if count == 0 {
        logger.Error().Int("count", count).
            Uint64("prev", msg.PrevInfo.No).Msg("receive empty GetHashesRsp")
        return
    }

    logger.Debug().Int("count", count).
        Uint64("prev", msg.PrevInfo.No).
        Str("start", base58.Encode(msg.Hashes[0])).
        Str("end", base58.Encode(msg.Hashes[count-1])).Msg("receive GetHashesRsp")

    hf.responseCh <- msg
    return
}