p2p: create an abstraction layer for devp2p to pick its transport

The goal is to be able to add a libp2p layer and
let nodes pick if they want a rlpx or libp2p backend.

The abstraction layer has been tested on rinkeby and it syncs.

dashboard/dashboard.go

@@ -129,7 +129,7 @@ func (db *Dashboard) APIs() []rpc.API { return nil }
 
 // Start starts the data collection thread and the listening server of the dashboard.
 // Implements the node.Service interface.
-func (db *Dashboard) Start(server *p2p.Server) error {
+func (db *Dashboard) Start(server p2p.ServerIf) error {
 	log.Info("Starting dashboard")
 
 	db.wg.Add(2)

eth/backend.go

@@ -482,7 +482,7 @@ func (s *Ethereum) Protocols() []p2p.Protocol {
 
 // Start implements node.Service, starting all internal goroutines needed by the
 // Ethereum protocol implementation.
-func (s *Ethereum) Start(srvr *p2p.Server) error {
+func (s *Ethereum) Start(srvr p2p.ServerIf) error {
 	// Start the bloom bits servicing goroutines
 	s.startBloomHandlers(params.BloomBitsBlocks)
 
@@ -490,9 +490,9 @@ func (s *Ethereum) Start(srvr *p2p.Server) error {
 	s.netRPCService = ethapi.NewPublicNetAPI(srvr, s.NetVersion())
 
 	// Figure out a max peers count based on the server limits
-	maxPeers := srvr.MaxPeers
+	maxPeers := srvr.MaxPeers()
 	if s.config.LightServ > 0 {
-		if s.config.LightPeers >= srvr.MaxPeers {
+		if s.config.LightPeers >= srvr.MaxPeers() {
 			return fmt.Errorf("invalid peer config: light peer count (%d) >= total peer count (%d)", s.config.LightPeers, srvr.MaxPeers)
 		}
 		maxPeers -= s.config.LightPeers
@@ -500,7 +500,7 @@ func (s *Ethereum) Start(srvr *p2p.Server) error {
 	// Start the networking layer and the light server if requested
 	s.protocolManager.Start(maxPeers)
 	if s.lesServer != nil {
-		s.lesServer.Start(srvr)
+		s.lesServer.Start(srvr.(*p2p.Server))
 	}
 	return nil
 }

ethstats/ethstats.go

@@ -69,7 +69,7 @@ type blockChain interface {
 // Service implements an Ethereum netstats reporting daemon that pushes local
 // chain statistics up to a monitoring server.
 type Service struct {
-	server *p2p.Server        // Peer-to-peer server to retrieve networking infos
+	server p2p.ServerIf       // Peer-to-peer server to retrieve networking infos
 	eth    *eth.Ethereum      // Full Ethereum service if monitoring a full node
 	les    *les.LightEthereum // Light Ethereum service if monitoring a light node
 	engine consensus.Engine   // Consensus engine to retrieve variadic block fields
@@ -118,7 +118,7 @@ func (s *Service) Protocols() []p2p.Protocol { return nil }
 func (s *Service) APIs() []rpc.API { return nil }
 
 // Start implements node.Service, starting up the monitoring and reporting daemon.
-func (s *Service) Start(server *p2p.Server) error {
+func (s *Service) Start(server p2p.ServerIf) error {
 	s.server = server
 	go s.loop()
 

internal/ethapi/api.go

@@ -1466,12 +1466,12 @@ func (api *PrivateDebugAPI) SetHead(number hexutil.Uint64) {
 
 // PublicNetAPI offers network related RPC methods
 type PublicNetAPI struct {
-	net            *p2p.Server
+	net            p2p.ServerIf
 	networkVersion uint64
 }
 
 // NewPublicNetAPI creates a new net API instance.
-func NewPublicNetAPI(net *p2p.Server, networkVersion uint64) *PublicNetAPI {
+func NewPublicNetAPI(net p2p.ServerIf, networkVersion uint64) *PublicNetAPI {
 	return &PublicNetAPI{net, networkVersion}
 }
 

les/backend.go

@@ -230,7 +230,7 @@ func (s *LightEthereum) Protocols() []p2p.Protocol {
 
 // Start implements node.Service, starting all internal goroutines needed by the
 // Ethereum protocol implementation.
-func (s *LightEthereum) Start(srvr *p2p.Server) error {
+func (s *LightEthereum) Start(srvr p2p.ServerIf) error {
 	log.Warn("Light client mode is an experimental feature")
 	s.startBloomHandlers(params.BloomBitsBlocksClient)
 	s.netRPCService = ethapi.NewPublicNetAPI(srvr, s.networkId)

les/serverpool.go

@@ -115,7 +115,7 @@ type registerReq struct {
 type serverPool struct {
 	db     ethdb.Database
 	dbKey  []byte
-	server *p2p.Server
+	server p2p.ServerIf
 	quit   chan struct{}
 	wg     *sync.WaitGroup
 	connWg sync.WaitGroup
@@ -162,14 +162,14 @@ func newServerPool(db ethdb.Database, quit chan struct{}, wg *sync.WaitGroup) *s
 	return pool
 }
 
-func (pool *serverPool) start(server *p2p.Server, topic discv5.Topic) {
+func (pool *serverPool) start(server p2p.ServerIf, topic discv5.Topic) {
 	pool.server = server
 	pool.topic = topic
 	pool.dbKey = append([]byte("serverPool/"), []byte(topic)...)
 	pool.wg.Add(1)
 	pool.loadNodes()
 
-	if pool.server.DiscV5 != nil {
+	if pool.server.(*p2p.Server).DiscV5 != nil {
 		pool.discSetPeriod = make(chan time.Duration, 1)
 		pool.discNodes = make(chan *enode.Node, 100)
 		pool.discLookups = make(chan bool, 100)
@@ -183,7 +183,7 @@ func (pool *serverPool) start(server *p2p.Server, topic discv5.Topic) {
 func (pool *serverPool) discoverNodes() {
 	ch := make(chan *discv5.Node)
 	go func() {
-		pool.server.DiscV5.SearchTopic(pool.topic, pool.discSetPeriod, ch, pool.discLookups)
+		pool.server.(*p2p.Server).DiscV5.SearchTopic(pool.topic, pool.discSetPeriod, ch, pool.discLookups)
 		close(ch)
 	}()
 	for n := range ch {

node/api.go

@@ -454,7 +454,7 @@ func NewPublicWeb3API(stack *Node) *PublicWeb3API {
 
 // ClientVersion returns the node name
 func (s *PublicWeb3API) ClientVersion() string {
-	return s.stack.Server().Name
+	return s.stack.Server().Name()
 }
 
 // Sha3 applies the ethereum sha3 implementation on the input.

node/node.go

@@ -46,7 +46,7 @@ type Node struct {
 	instanceDirLock   flock.Releaser // prevents concurrent use of instance directory
 
 	serverConfig p2p.Config
-	server       *p2p.Server // Currently running P2P networking layer
+	server       p2p.ServerIf // Currently running P2P networking layer
 
 	serviceFuncs []ServiceConstructor     // Service constructors (in dependency order)
 	services     map[reflect.Type]Service // Currently running services
@@ -162,7 +162,7 @@ func (n *Node) Start() error {
 	if n.serverConfig.NodeDatabase == "" {
 		n.serverConfig.NodeDatabase = n.config.NodeDB()
 	}
-	running := &p2p.Server{Config: n.serverConfig}
+	var running p2p.ServerIf = &p2p.Server{Config: n.serverConfig}
 	n.log.Info("Starting peer-to-peer node", "instance", n.serverConfig.Name)
 
 	// Otherwise copy and specialize the P2P configuration
@@ -191,7 +191,7 @@ func (n *Node) Start() error {
 	}
 	// Gather the protocols and start the freshly assembled P2P server
 	for _, service := range services {
-		running.Protocols = append(running.Protocols, service.Protocols()...)
+		running.AddProtocols(service.Protocols())
 	}
 	if err := running.Start(); err != nil {
 		return convertFileLockError(err)
@@ -501,7 +501,7 @@ func (n *Node) RPCHandler() (*rpc.Server, error) {
 // Server retrieves the currently running P2P network layer. This method is meant
 // only to inspect fields of the currently running server, life cycle management
 // should be left to this Node entity.
-func (n *Node) Server() *p2p.Server {
+func (n *Node) Server() p2p.ServerIf {
 	n.lock.RLock()
 	defer n.lock.RUnlock()
 

node/service.go

@@ -90,7 +90,7 @@ type Service interface {
 
 	// Start is called after all services have been constructed and the networking
 	// layer was also initialized to spawn any goroutines required by the service.
-	Start(server *p2p.Server) error
+	Start(server p2p.ServerIf) error
 
 	// Stop terminates all goroutines belonging to the service, blocking until they
 	// are all terminated.

p2p/dial.go

@@ -73,7 +73,7 @@ type dialstate struct {
 	netrestrict *netutil.Netlist
 
 	lookupRunning bool
-	dialing       map[enode.ID]connFlag
+	dialing       map[enode.ID]ConnFlag
 	lookupBuf     []*enode.Node // current discovery lookup results
 	randomNodes   []*enode.Node // filled from Table
 	static        map[enode.ID]*dialTask
@@ -107,7 +107,7 @@ type task interface {
 // A dialTask is generated for each node that is dialed. Its
 // fields cannot be accessed while the task is running.
 type dialTask struct {
-	flags        connFlag
+	flags        ConnFlag
 	dest         *enode.Node
 	lastResolved time.Time
 	resolveDelay time.Duration
@@ -132,7 +132,7 @@ func newDialState(static []*enode.Node, bootnodes []*enode.Node, ntab discoverTa
 		ntab:        ntab,
 		netrestrict: netrestrict,
 		static:      make(map[enode.ID]*dialTask),
-		dialing:     make(map[enode.ID]connFlag),
+		dialing:     make(map[enode.ID]ConnFlag),
 		bootnodes:   make([]*enode.Node, len(bootnodes)),
 		randomNodes: make([]*enode.Node, maxdyn/2),
 		hist:        new(dialHistory),
@@ -164,7 +164,7 @@ func (s *dialstate) newTasks(nRunning int, peers map[enode.ID]*Peer, now time.Ti
 	}
 
 	var newtasks []task
-	addDial := func(flag connFlag, n *enode.Node) bool {
+	addDial := func(flag ConnFlag, n *enode.Node) bool {
 		if err := s.checkDial(n, peers); err != nil {
 			log.Trace("Skipping dial candidate", "id", n.ID(), "addr", &net.TCPAddr{IP: n.IP(), Port: n.TCP()}, "err", err)
 			return false

p2p/server.go

@@ -147,6 +147,27 @@ type Config struct {
 	Logger log.Logger `toml:",omitempty"`
 }
 
+type ServerIf interface {
+	AddProtocols([]Protocol)
+	Start() error
+	Stop()
+	// ca cree tout chez devp2p (rlpx, encrypted handhake, etc.
+	// Je pense que nous on va rien faire meme si je dois verifier que secio est actif)
+	SetupConn(net.Conn, ConnFlag, *enode.Node) error
+	NodeInfo() *NodeInfo
+	PeersInfo() []*PeerInfo
+	Self() *enode.Node
+	AddPeer(node *enode.Node)
+	RemovePeer(node *enode.Node)
+	AddTrustedPeer(node *enode.Node)
+	RemoveTrustedPeer(node *enode.Node)
+	SubscribeEvents(chan *PeerEvent) event.Subscription
+	PeerCount() int
+	Peers() []*Peer
+	Name() string
+	MaxPeers() int
+}
+
 // Server manages all peer connections.
 type Server struct {
 	// Config fields may not be modified while the server is running.
@@ -191,10 +212,10 @@ type peerDrop struct {
 	requested bool // true if signaled by the peer
 }
 
-type connFlag int32
+type ConnFlag int32
 
 const (
-	dynDialedConn connFlag = 1 << iota
+	dynDialedConn ConnFlag = 1 << iota
 	staticDialedConn
 	inboundConn
 	trustedConn
@@ -206,7 +227,7 @@ type conn struct {
 	fd net.Conn
 	transport
 	node  *enode.Node
-	flags connFlag
+	flags ConnFlag
 	cont  chan error // The run loop uses cont to signal errors to SetupConn.
 	caps  []Cap      // valid after the protocol handshake
 	name  string     // valid after the protocol handshake
@@ -235,7 +256,7 @@ func (c *conn) String() string {
 	return s
 }
 
-func (f connFlag) String() string {
+func (f ConnFlag) String() string {
 	s := ""
 	if f&trustedConn != 0 {
 		s += "-trusted"
@@ -255,14 +276,14 @@ func (f connFlag) String() string {
 	return s
 }
 
-func (c *conn) is(f connFlag) bool {
+func (c *conn) is(f ConnFlag) bool {
-	flags := connFlag(atomic.LoadInt32((*int32)(&c.flags)))
+	flags := ConnFlag(atomic.LoadInt32((*int32)(&c.flags)))
 	return flags&f != 0
 }
 
-func (c *conn) set(f connFlag, val bool) {
+func (c *conn) set(f ConnFlag, val bool) {
 	for {
-		oldFlags := connFlag(atomic.LoadInt32((*int32)(&c.flags)))
+		oldFlags := ConnFlag(atomic.LoadInt32((*int32)(&c.flags)))
 		flags := oldFlags
 		if val {
 			flags |= f
@@ -275,6 +296,10 @@ func (c *conn) set(f connFlag, val bool) {
 	}
 }
 
+func (srv *Server) Name() string {
+	return srv.Config.Name
+}
+
 // Peers returns all connected peers.
 func (srv *Server) Peers() []*Peer {
 	var ps []*Peer
@@ -293,6 +318,10 @@ func (srv *Server) Peers() []*Peer {
 	return ps
 }
 
+func (srv *Server) AddProtocols(protos []Protocol) {
+	srv.Protocols = append(srv.Protocols, protos...)
+}
+
 // PeerCount returns the number of connected peers.
 func (srv *Server) PeerCount() int {
 	var count int
@@ -540,7 +569,7 @@ func (srv *Server) Start() (err error) {
 
 	// handshake
 	pubkey := crypto.FromECDSAPub(&srv.PrivateKey.PublicKey)
-	srv.ourHandshake = &protoHandshake{Version: baseProtocolVersion, Name: srv.Name, ID: pubkey[1:]}
+	srv.ourHandshake = &protoHandshake{Version: baseProtocolVersion, Name: srv.Name(), ID: pubkey[1:]}
 	for _, p := range srv.Protocols {
 		srv.ourHandshake.Caps = append(srv.ourHandshake.Caps, p.cap())
 	}
@@ -775,7 +804,7 @@ func (srv *Server) protoHandshakeChecks(peers map[enode.ID]*Peer, inboundCount i
 
 func (srv *Server) encHandshakeChecks(peers map[enode.ID]*Peer, inboundCount int, c *conn) error {
 	switch {
-	case !c.is(trustedConn|staticDialedConn) && len(peers) >= srv.MaxPeers:
+	case !c.is(trustedConn|staticDialedConn) && len(peers) >= srv.MaxPeers():
 		return DiscTooManyPeers
 	case !c.is(trustedConn) && c.is(inboundConn) && inboundCount >= srv.maxInboundConns():
 		return DiscTooManyPeers
@@ -789,7 +818,7 @@ func (srv *Server) encHandshakeChecks(peers map[enode.ID]*Peer, inboundCount int
 }
 
 func (srv *Server) maxInboundConns() int {
-	return srv.MaxPeers - srv.maxDialedConns()
+	return srv.MaxPeers() - srv.maxDialedConns()
 }
 func (srv *Server) maxDialedConns() int {
 	if srv.NoDiscovery || srv.NoDial {
@@ -799,7 +828,7 @@ func (srv *Server) maxDialedConns() int {
 	if r == 0 {
 		r = defaultDialRatio
 	}
-	return srv.MaxPeers / r
+	return srv.MaxPeers() / r
 }
 
 type tempError interface {
@@ -863,7 +892,7 @@ func (srv *Server) listenLoop() {
 // SetupConn runs the handshakes and attempts to add the connection
 // as a peer. It returns when the connection has been added as a peer
 // or the handshakes have failed.
-func (srv *Server) SetupConn(fd net.Conn, flags connFlag, dialDest *enode.Node) error {
+func (srv *Server) SetupConn(fd net.Conn, flags ConnFlag, dialDest *enode.Node) error {
 	self := srv.Self()
 	if self == nil {
 		return errors.New("shutdown")
@@ -877,7 +906,7 @@ func (srv *Server) SetupConn(fd net.Conn, flags connFlag, dialDest *enode.Node)
 	return err
 }
 
-func (srv *Server) setupConn(c *conn, flags connFlag, dialDest *enode.Node) error {
+func (srv *Server) setupConn(c *conn, flags ConnFlag, dialDest *enode.Node) error {
 	// Prevent leftover pending conns from entering the handshake.
 	srv.lock.Lock()
 	running := srv.running
@@ -1018,7 +1047,7 @@ func (srv *Server) NodeInfo() *NodeInfo {
 
 	// Gather and assemble the generic node infos
 	info := &NodeInfo{
-		Name:       srv.Name,
+		Name:       srv.Name(),
 		Enode:      node.String(),
 		ID:         node.ID().String(),
 		IP:         node.IP().String(),
@@ -1060,3 +1089,7 @@ func (srv *Server) PeersInfo() []*PeerInfo {
 	}
 	return infos
 }
+
+func (srv *Server) MaxPeers() int {
+	return srv.Config.MaxPeers
+}

vendor/vendor.json

@@ -261,6 +261,12 @@
 			"revisionTime": "2017-08-21T10:38:37Z",
 			"tree": true
 		},
+		{
+			"checksumSHA1": "Zkp39z/brPDQUmRVisP90WpSljM=",
+			"path": "github.com/libp2p/go-libp2p-blankhost",
+			"revision": "d1dca03c3ddffd1bc45cd6259eae9a7ede5da35c",
+			"revisionTime": "2018-09-24T12:10:29Z"
+		},
 		{
 			"checksumSHA1": "7hln62oZPZmyqEmgXaybf9WxQ7A=",
 			"path": "github.com/maruel/panicparse/stack",
@@ -298,6 +304,12 @@
 			"revision": "c48cc78d482608239f6c4c92a4abd87eb8761c90",
 			"revisionTime": "2017-09-29T03:49:55Z"
 		},
+		{
+			"checksumSHA1": "oThw7uD1i97WvthJmCUi8DwRwf8=",
+			"path": "github.com/multiformats/go-multiaddr",
+			"revision": "a80cfc331ffff8bc519e401d8789d324196d1775",
+			"revisionTime": "2018-09-24T09:07:54Z"
+		},
 		{
 			"checksumSHA1": "FYM/8R2CqS6PSNAoKl6X5gNJ20A=",
 			"path": "github.com/naoina/toml",

whisper/whisperv6/whisper.go

@@ -624,7 +624,7 @@ func (whisper *Whisper) Send(envelope *Envelope) error {
 
 // Start implements node.Service, starting the background data propagation thread
 // of the Whisper protocol.
-func (whisper *Whisper) Start(*p2p.Server) error {
+func (whisper *Whisper) Start(p2p.ServerIf) error {
 	log.Info("started whisper v." + ProtocolVersionStr)
 	go whisper.update()