prysm-pulse/beacon-chain/p2p/handshake.go
terence tsao 798bbbdc82 Cold start for interop (#3437)
* coldstart flags for validator

* WIP beacon node flags

* wip beacon chain, flag fix in validator, arg fix in validator

* checkpoint

* Added interop service

* working on mock chainstart

* save the state lol

* fix tests

* Save genesis validators

* gaz

* fix validator help flags

* WaitForChainStart actually waits for genesis time

* cold start fixes

* cache

* change back

* allow for genesis state too

* remove logs

* increase mmap size

* dont process if head doesn't exist

* add 10ms tolerance

* enable libp2p debug at debug, fix pubsub

* works with checkpt

* initialize justified and finalized in db

* Removed preloadStatePath from blockchain service

* Clean up

* Write to disk for now post state

* revert b466dd536f8eadbdae2264a545a755370223d917

* Builds

* Only RPC test fails now

* use minimal config, no demo config

* clean up branch

* Lint

* resolve lint

* more lint fixes

* lint

* fix viz

* Fixing RPC test

* skip before epoch 2

* RPC time out

* Fixed ordering

* rename

* remove some dbg statements

* ensure index is correct

* fix some panics

* getting closer

* fix tests

* Fix private key

* Fixed RPC test

* Fixed beacon chain build for docker

* Add interop.go to validator go_image

* Fixed docker build

* handle errors

* skip test, skip disconnecting peers

* Fixed docker build

* tolerance for attestation processing

* revert copy

* clearer err message parse

* fix launching from dep contract
2019-09-11 13:38:35 -05:00

73 lines
2.4 KiB
Go

package p2p
import (
"context"
"io"
"sync"
"github.com/libp2p/go-libp2p-core/network"
"github.com/libp2p/go-libp2p-core/peer"
pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
)
var handshakes = make(map[peer.ID]*pb.Hello)
var handshakeLock sync.Mutex
// AddHandshake to the local records for initial sync.
func (s *Service) AddHandshake(pid peer.ID, hello *pb.Hello) {
handshakeLock.Lock()
defer handshakeLock.Unlock()
handshakes[pid] = hello
}
// Handshakes has not been implemented yet and it may be moved to regular sync...
func (s *Service) Handshakes() map[peer.ID]*pb.Hello {
return nil
}
// AddConnectionHandler adds a callback function which handles the connection with a
// newly added peer. It performs a handshake with that peer by sending a hello request
// and validating the response from the peer.
func (s *Service) AddConnectionHandler(reqFunc func(ctx context.Context, id peer.ID) error) {
s.host.Network().Notify(&network.NotifyBundle{
ConnectedF: func(net network.Network, conn network.Conn) {
// Must be handled in a goroutine as this callback cannot be blocking.
go func() {
ctx := context.Background()
log := log.WithField("peer", conn.RemotePeer())
if conn.Stat().Direction == network.DirInbound {
log.Debug("Not sending hello for inbound connection")
return
}
log.Debug("Performing handshake with peer")
if err := reqFunc(ctx, conn.RemotePeer()); err != nil && err != io.EOF {
log.WithError(err).Error("Could not send successful hello rpc request")
log.Error("Not disconnecting for interop testing :)")
//if err := s.Disconnect(conn.RemotePeer()); err != nil {
// log.WithError(err).Errorf("Unable to close peer %s", conn.RemotePeer())
// return
//}
return
}
log.WithField("peer", conn.RemotePeer().Pretty()).Info("New peer connected.")
}()
},
})
}
// addDisconnectionHandler ensures that previously disconnected peers aren't dialed again. Due
// to either their ports being closed, nodes are no longer active,etc.
func (s *Service) addDisconnectionHandler() {
s.host.Network().Notify(&network.NotifyBundle{
DisconnectedF: func(net network.Network, conn network.Conn) {
// Must be handled in a goroutine as this callback cannot be blocking.
go func() {
s.exclusionList.Set(conn.RemotePeer().String(), true, ttl)
log.WithField("peer", conn.RemotePeer()).Debug(
"Peer is added to exclusion list",
)
}()
},
})
}