prysm-pulse/beacon-chain/sync/initial-sync/service.go

188 lines
4.6 KiB
Go
Raw Normal View History

package initialsync
import (
"context"
2019-09-02 19:36:14 +00:00
"errors"
"fmt"
"time"
"github.com/libp2p/go-libp2p-core/peer"
"github.com/prysmaticlabs/prysm/beacon-chain/blockchain"
"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
"github.com/prysmaticlabs/prysm/beacon-chain/db"
"github.com/prysmaticlabs/prysm/beacon-chain/p2p"
"github.com/prysmaticlabs/prysm/beacon-chain/sync"
pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
eth "github.com/prysmaticlabs/prysm/proto/eth/v1alpha1"
"github.com/prysmaticlabs/prysm/shared"
"github.com/prysmaticlabs/prysm/shared/params"
"github.com/prysmaticlabs/prysm/shared/roughtime"
)
var _ = shared.Service(&InitialSync{})
type blockchainService interface {
blockchain.BlockReceiver
[Interop] Improve RPC Codebase + Start Beacon Chain With Mock ETH1 Values (#3407) * add main.go * interop readme * proper visibility * standardize and abstract into simpler funcs * formatting * no os pkg * add test * no panics anywhere, properly and nicely handle errors * proper comments * fix broken test * readme * comment * recommend ssz * install * tool now works * README * build * readme * 64 validators * rem print * register the no powchain flag * work on mock eth1 start * common interface * getting closer with the interface defs * only two uses of powchain * remove powchain dependency * remove powchain dependency * common powchain interface * proper comment in case of flag * proper args into rpc services * rename fields * pass in mock flag into RPC * conforms to iface * use client instead of block fetcher iface * broken tests * block fetcher * finalized * resolved broken build * fix build * comment * fix tests * tests pass * resolved confs * took them out * rename into smaller interfaces * resolve some confs * ensure tests pass * properly utilize mock instead of localized mock * res lint * lint * finish test for mock eth1data * run gazelle * include flag again * fix broken build * disable powchain * dont dial eth1 nodes * reenable pow * use smaller interfaces, standardize naming * abstract mock into its own package * faulty mock lint * fix stutter in lint * rpc tests all passing * use mocks for operations * no more mocks in the entire rpc package * no mock * viz * testonly
2019-09-09 21:13:50 +00:00
blockchain.HeadFetcher
blockchain.ChainFeeds
}
const (
minHelloCount = 1 // TODO(3147): Set this to more than 1, maybe configure from flag?
handshakePollingInterval = 5 * time.Second // Polling interval for checking the number of received handshakes.
)
// Config to set up the initial sync service.
type Config struct {
P2P p2p.P2P
DB db.Database
Chain blockchainService
RegSync sync.StatusTracker
}
// InitialSync service.
type InitialSync struct {
statusTracker sync.StatusTracker
chain blockchainService
p2p p2p.P2P
synced bool
chainStarted bool
}
// NewInitialSync configures the initial sync service responsible for bringing the node up to the
// latest head of the blockchain.
func NewInitialSync(cfg *Config) *InitialSync {
return &InitialSync{
statusTracker: cfg.RegSync,
chain: cfg.Chain,
p2p: cfg.P2P,
}
}
// Start the initial sync service.
func (s *InitialSync) Start() {
ch := make(chan time.Time)
sub := s.chain.StateInitializedFeed().Subscribe(ch)
defer sub.Unsubscribe()
// Wait until chain start.
genesis := <-ch
if genesis.After(roughtime.Now()) {
time.Sleep(roughtime.Until(genesis))
}
s.chainStarted = true
currentSlot := slotsSinceGenesis(genesis)
if helpers.SlotToEpoch(currentSlot) == 0 {
log.Info("Chain started within the last epoch. Not syncing.")
2019-09-02 19:36:14 +00:00
s.synced = true
return
}
// Are we already in sync, or close to it?
if helpers.SlotToEpoch(s.chain.HeadSlot()) == helpers.SlotToEpoch(currentSlot) {
log.Info("Already synced to the current epoch.")
2019-09-02 19:36:14 +00:00
s.synced = true
return
}
// Every 5 sec, report handshake count.
for {
helloCount := len(s.statusTracker.PeerStatuses())
log.WithField(
"hellos",
fmt.Sprintf("%d/%d", helloCount, minHelloCount),
).Info("Waiting for enough peer handshakes before syncing.")
if helloCount >= minHelloCount {
break
}
time.Sleep(handshakePollingInterval)
}
pid, best := bestStatus(s.statusTracker.PeerStatuses())
var last *eth.BeaconBlock
for headSlot := s.chain.HeadSlot(); headSlot < slotsSinceGenesis(genesis); {
req := &pb.BeaconBlocksByRangeRequest{
StartSlot: headSlot + 1,
HeadBlockRoot: s.chain.HeadRoot(),
Count: 64,
Step: 1,
}
log.WithField("data", fmt.Sprintf("%+v", req)).Info("Sending msg")
strm, err := s.p2p.Send(context.Background(), req, pid)
if err != nil {
panic(err)
}
// Read status code.
code, errMsg, err := sync.ReadStatusCode(strm, s.p2p.Encoding())
if err != nil {
panic(err)
}
if code != 0 {
log.Errorf("Request failed. Request was %+v", req)
panic(errMsg)
}
2019-09-10 14:24:14 +00:00
resp := make([]*eth.BeaconBlock, 0)
if err := s.p2p.Encoding().DecodeWithLength(strm, &resp); err != nil {
log.Error(err)
continue
}
2019-09-10 14:24:14 +00:00
for _, blk := range resp {
if blk.Slot <= headSlot {
continue
}
if blk.Slot < helpers.StartSlot(best.FinalizedEpoch+1) {
if err := s.chain.ReceiveBlockNoPubsubForkchoice(context.Background(), blk); err != nil {
panic(err)
}
} else {
if err := s.chain.ReceiveBlockNoPubsub(context.Background(), blk); err != nil {
panic(err)
}
}
last = blk
}
headSlot = s.chain.HeadSlot()
}
// Force a fork choice update since fork choice was not run during initial sync.
if err := s.chain.ReceiveBlockNoPubsub(context.Background(), last); err != nil {
panic(err)
}
log.Infof("Synced up to %d", s.chain.HeadSlot())
2019-09-02 19:36:14 +00:00
s.synced = true
}
// Stop initial sync.
func (s *InitialSync) Stop() error {
return nil
}
// Status of initial sync.
func (s *InitialSync) Status() error {
if !s.synced && s.chainStarted {
2019-09-02 19:36:14 +00:00
return errors.New("syncing")
}
return nil
}
func bestStatus(data map[peer.ID]*pb.Status) (peer.ID, *pb.Status) {
for pid, status := range data {
return pid, status
}
return "", nil
}
func slotsSinceGenesis(genesisTime time.Time) uint64 {
return uint64(roughtime.Since(genesisTime).Seconds()) / params.BeaconConfig().SecondsPerSlot
}