headrush
/
go-ethereum
forked from mirror/go-ethereum
1
1
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

node: customizable protocol and service stacks

Browse Source
release/1.4
Péter Szilágyi 9 years ago
parent 168d0e9e45
commit 9e1d9bff3b
13 changed files with 1333 additions and 13 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 171
      node/config.go
  2. 120
      node/config_test.go
  3. 31
      node/errors.go
  4. 252
      node/node.go
  5. 87
      node/node_example_test.go
  6. 492
      node/node_test.go
  7. 67
      node/service.go
  8. 60
      node/service_test.go
  9. 33
      node/utils.go
  10. 7
      p2p/discover/table.go
  11. 8
      p2p/discover/table_test.go
  12. 16
      p2p/discover/udp.go
  13. 2
      p2p/discover/udp_test.go

171
node/config.go
Unescape View File

@ -0,0 +1,171 @@
// Copyright 2015 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum 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 go-ethereum 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 go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package node
import (
"crypto/ecdsa"
"encoding/json"
"io/ioutil"
"net"
"os"
"path/filepath"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/nat"
)
var (
datadirPrivateKey = "nodekey" // Path within the datadir to the node's private key
datadirStaticNodes = "static-nodes.json" // Path within the datadir to the static node list
datadirTrustedNodes = "trusted-nodes.json" // Path within the datadir to the trusted node list
datadirNodeDatabase = "nodes" // Path within the datadir to store the node infos
)
// Config represents a small collection of configuration values to fine tune the
// P2P network layer of a protocol stack. These values can be further extended by
// all registered services.
type Config struct {
// DataDir is the file system folder the node should use for any data storage
// requirements. The configured data directory will not be directly shared with
// registered services, instead those can use utility methods to create/access
// databases or flat files. This enables ephemeral nodes which can fully reside
// in memory.
DataDir string
// This field should be a valid secp256k1 private key that will be used for both
// remote peer identification as well as network traffic encryption. If no key
// is configured, the preset one is loaded from the data dir, generating it if
// needed.
PrivateKey *ecdsa.PrivateKey
// Name sets the node name of this server. Use common.MakeName to create a name
// that follows existing conventions.
Name string
// NoDiscovery specifies whether the peer discovery mechanism should be started
// or not. Disabling is usually useful for protocol debugging (manual topology).
NoDiscovery bool
// Bootstrap nodes used to establish connectivity with the rest of the network.
BootstrapNodes []*discover.Node
// Network interface address on which the node should listen for inbound peers.
ListenAddr string
// If set to a non-nil value, the given NAT port mapper is used to make the
// listening port available to the Internet.
NAT nat.Interface
// If Dialer is set to a non-nil value, the given Dialer is used to dial outbound
// peer connections.
Dialer *net.Dialer
// If NoDial is true, the node will not dial any peers.
NoDial bool
// MaxPeers is the maximum number of peers that can be connected. If this is
// set to zero, then only the configured static and trusted peers can connect.
MaxPeers int
// MaxPendingPeers is the maximum number of peers that can be pending in the
// handshake phase, counted separately for inbound and outbound connections.
// Zero defaults to preset values.
MaxPendingPeers int
}
// NodeKey retrieves the currently configured private key of the node, checking
// first any manually set key, falling back to the one found in the configured
// data folder. If no key can be found, a new one is generated.
func (c *Config) NodeKey() *ecdsa.PrivateKey {
// Use any specifically configured key
if c.PrivateKey != nil {
return c.PrivateKey
}
// Generate ephemeral key if no datadir is being used
if c.DataDir == "" {
key, err := crypto.GenerateKey()
if err != nil {
glog.Fatalf("Failed to generate ephemeral node key: %v", err)
}
return key
}
// Fall back to persistent key from the data directory
keyfile := filepath.Join(c.DataDir, datadirPrivateKey)
if key, err := crypto.LoadECDSA(keyfile); err == nil {
return key
}
// No persistent key found, generate and store a new one
key, err := crypto.GenerateKey()
if err != nil {
glog.Fatalf("Failed to generate node key: %v", err)
}
if err := crypto.SaveECDSA(keyfile, key); err != nil {
glog.V(logger.Error).Infof("Failed to persist node key: %v", err)
}
return key
}
// StaticNodes returns a list of node enode URLs configured as static nodes.
func (c *Config) StaticNodes() []*discover.Node {
return c.parsePersistentNodes(datadirStaticNodes)
}
// TrusterNodes returns a list of node enode URLs configured as trusted nodes.
func (c *Config) TrusterNodes() []*discover.Node {
return c.parsePersistentNodes(datadirTrustedNodes)
}
// parsePersistentNodes parses a list of discovery node URLs loaded from a .json
// file from within the data directory.
func (c *Config) parsePersistentNodes(file string) []*discover.Node {
// Short circuit if no node config is present
if c.DataDir == "" {
return nil
}
path := filepath.Join(c.DataDir, file)
if _, err := os.Stat(path); err != nil {
return nil
}
// Load the nodes from the config file
blob, err := ioutil.ReadFile(path)
if err != nil {
glog.V(logger.Error).Infof("Failed to access nodes: %v", err)
return nil
}
nodelist := []string{}
if err := json.Unmarshal(blob, &nodelist); err != nil {
glog.V(logger.Error).Infof("Failed to load nodes: %v", err)
return nil
}
// Interpret the list as a discovery node array
var nodes []*discover.Node
for _, url := range nodelist {
if url == "" {
continue
}
node, err := discover.ParseNode(url)
if err != nil {
glog.V(logger.Error).Infof("Node URL %s: %v\n", url, err)
continue
}
nodes = append(nodes, node)
}
return nodes
}

120
node/config_test.go
Unescape View File

@ -0,0 +1,120 @@
// Copyright 2015 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum 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 go-ethereum 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 go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package node
import (
"bytes"
"io/ioutil"
"os"
"path/filepath"
"testing"
"github.com/ethereum/go-ethereum/crypto"
)
// Tests that datadirs can be successfully created, be them manually configured
// ones or automatically generated temporary ones.
func TestDatadirCreation(t *testing.T) {
// Create a temporary data dir and check that it can be used by a node
dir, err := ioutil.TempDir("", "")
if err != nil {
t.Fatalf("failed to create manual data dir: %v", err)
}
defer os.RemoveAll(dir)
if _, err := New(&Config{DataDir: dir}); err != nil {
t.Fatalf("failed to create stack with existing datadir: %v", err)
}
// Generate a long non-existing datadir path and check that it gets created by a node
dir = filepath.Join(dir, "a", "b", "c", "d", "e", "f")
if _, err := New(&Config{DataDir: dir}); err != nil {
t.Fatalf("failed to create stack with creatable datadir: %v", err)
}
if _, err := os.Stat(dir); err != nil {
t.Fatalf("freshly created datadir not accessible: %v", err)
}
// Verify that an impossible datadir fails creation
file, err := ioutil.TempFile("", "")
if err != nil {
t.Fatalf("failed to create temporary file: %v", err)
}
defer os.Remove(file.Name())
dir = filepath.Join(file.Name(), "invalid/path")
if _, err := New(&Config{DataDir: dir}); err == nil {
t.Fatalf("protocol stack created with an invalid datadir")
}
}
// Tests that node keys can be correctly created, persisted, loaded and/or made
// ephemeral.
func TestNodeKeyPersistency(t *testing.T) {
// Create a temporary folder and make sure no key is present
dir, err := ioutil.TempDir("", "")
if err != nil {
t.Fatalf("failed to create temporary data directory: %v", err)
}
defer os.RemoveAll(dir)
if _, err := os.Stat(filepath.Join(dir, datadirPrivateKey)); err == nil {
t.Fatalf("non-created node key already exists")
}
// Configure a node with a preset key and ensure it's not persisted
key, err := crypto.GenerateKey()
if err != nil {
t.Fatalf("failed to generate one-shot node key: %v", err)
}
if _, err := New(&Config{DataDir: dir, PrivateKey: key}); err != nil {
t.Fatalf("failed to create empty stack: %v", err)
}
if _, err := os.Stat(filepath.Join(dir, datadirPrivateKey)); err == nil {
t.Fatalf("one-shot node key persisted to data directory")
}
// Configure a node with no preset key and ensure it is persisted this time
if _, err := New(&Config{DataDir: dir}); err != nil {
t.Fatalf("failed to create newly keyed stack: %v", err)
}
if _, err := os.Stat(filepath.Join(dir, datadirPrivateKey)); err != nil {
t.Fatalf("node key not persisted to data directory: %v", err)
}
key, err = crypto.LoadECDSA(filepath.Join(dir, datadirPrivateKey))
if err != nil {
t.Fatalf("failed to load freshly persisted node key: %v", err)
}
blob1, err := ioutil.ReadFile(filepath.Join(dir, datadirPrivateKey))
if err != nil {
t.Fatalf("failed to read freshly persisted node key: %v", err)
}
// Configure a new node and ensure the previously persisted key is loaded
if _, err := New(&Config{DataDir: dir}); err != nil {
t.Fatalf("failed to create previously keyed stack: %v", err)
}
blob2, err := ioutil.ReadFile(filepath.Join(dir, datadirPrivateKey))
if err != nil {
t.Fatalf("failed to read previously persisted node key: %v", err)
}
if bytes.Compare(blob1, blob2) != 0 {
t.Fatalf("persisted node key mismatch: have %x, want %x", blob2, blob1)
}
// Configure ephemeral node and ensure no key is dumped locally
if _, err := New(&Config{DataDir: ""}); err != nil {
t.Fatalf("failed to create ephemeral stack: %v", err)
}
if _, err := os.Stat(filepath.Join(".", datadirPrivateKey)); err == nil {
t.Fatalf("ephemeral node key persisted to disk")
}
}

31
node/errors.go
Unescape View File

@ -0,0 +1,31 @@
// Copyright 2015 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum 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 go-ethereum 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 go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package node
import "fmt"
// StopError is returned if a node fails to stop either any of its registered
// services or itself.
type StopError struct {
Server error
Services map[string]error
}
// Error generates a textual representation of the stop error.
func (e *StopError) Error() string {
return fmt.Sprintf("server: %v, services: %v", e.Server, e.Services)
}

252
node/node.go
Unescape View File

@ -0,0 +1,252 @@
// Copyright 2015 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum 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 go-ethereum 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 go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package node represents the Ethereum protocol stack container.
package node
import (
"errors"
"os"
"path/filepath"
"sync"
"syscall"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/p2p"
)
var (
ErrDatadirUsed = errors.New("datadir already used")
ErrNodeStopped = errors.New("node not started")
ErrNodeRunning = errors.New("node already running")
ErrServiceUnknown = errors.New("service not registered")
ErrServiceRegistered = errors.New("service already registered")
datadirInUseErrnos = map[uint]bool{11: true, 32: true, 35: true}
)
// Node represents a P2P node into which arbitrary services might be registered.
type Node struct {
datadir string // Path to the currently used data directory
config *p2p.Server // Configuration of the underlying P2P networking layer
stack map[string]ServiceConstructor // Protocol stack registered into this node
emux *event.TypeMux // Event multiplexer used between the services of a stack
running *p2p.Server // Currently running P2P networking layer
services map[string]Service // Currently running services
lock sync.RWMutex
}
// New creates a new P2P node, ready for protocol registration.
func New(conf *Config) (*Node, error) {
// Ensure the data directory exists, failing if it cannot be created
if conf.DataDir != "" {
if err := os.MkdirAll(conf.DataDir, 0700); err != nil {
return nil, err
}
}
// Assemble the networking layer and the node itself
nodeDbPath := ""
if conf.DataDir != "" {
nodeDbPath = filepath.Join(conf.DataDir, datadirNodeDatabase)
}
return &Node{
datadir: conf.DataDir,
config: &p2p.Server{
PrivateKey: conf.NodeKey(),
Name: conf.Name,
Discovery: !conf.NoDiscovery,
BootstrapNodes: conf.BootstrapNodes,
StaticNodes: conf.StaticNodes(),
TrustedNodes: conf.TrusterNodes(),
NodeDatabase: nodeDbPath,
ListenAddr: conf.ListenAddr,
NAT: conf.NAT,
Dialer: conf.Dialer,
NoDial: conf.NoDial,
MaxPeers: conf.MaxPeers,
MaxPendingPeers: conf.MaxPendingPeers,
},
stack: make(map[string]ServiceConstructor),
emux: new(event.TypeMux),
}, nil
}
// Register injects a new service into the node's stack.
func (n *Node) Register(id string, constructor ServiceConstructor) error {
n.lock.Lock()
defer n.lock.Unlock()
// Short circuit if the node is running or if the id is taken
if n.running != nil {
return ErrNodeRunning
}
if _, ok := n.stack[id]; ok {
return ErrServiceRegistered
}
// Otherwise register the service and return
n.stack[id] = constructor
return nil
}
// Unregister removes a service from a node's stack. If the node is currently
// running, an error will be returned.
func (n *Node) Unregister(id string) error {
n.lock.Lock()
defer n.lock.Unlock()
// Short circuit if the node is running, or if the service is unknown
if n.running != nil {
return ErrNodeRunning
}
if _, ok := n.stack[id]; !ok {
return ErrServiceUnknown
}
// Otherwise drop the service and return
delete(n.stack, id)
return nil
}
// Start create a live P2P node and starts running it.
func (n *Node) Start() error {
n.lock.Lock()
defer n.lock.Unlock()
// Short circuit if the node's already running
if n.running != nil {
return ErrNodeRunning
}
// Otherwise copy and specialize the P2P configuration
running := new(p2p.Server)
*running = *n.config
ctx := &ServiceContext{
dataDir: n.datadir,
EventMux: n.emux,
}
services := make(map[string]Service)
for id, constructor := range n.stack {
service, err := constructor(ctx)
if err != nil {
return err
}
services[id] = service
}
// Gather the protocols and start the freshly assembled P2P server
for _, service := range services {
running.Protocols = append(running.Protocols, service.Protocols()...)
}
if err := running.Start(); err != nil {
if errno, ok := err.(syscall.Errno); ok && datadirInUseErrnos[uint(errno)] {
return ErrDatadirUsed
}
return err
}
// Start each of the services
started := []string{}
for id, service := range services {
// Start the next service, stopping all previous upon failure
if err := service.Start(); err != nil {
for _, id := range started {
services[id].Stop()
}
return err
}
// Mark the service started for potential cleanup
started = append(started, id)
}
// Finish initializing the startup
n.services = services
n.running = running
return nil
}
// Stop terminates a running node along with all it's services. In the node was
// not started, an error is returned.
func (n *Node) Stop() error {
n.lock.Lock()
defer n.lock.Unlock()
// Short circuit if the node's not running
if n.running == nil {
return ErrNodeStopped
}
// Otherwise terminate all the services and the P2P server too
failure := &StopError{
Services: make(map[string]error),
}
for id, service := range n.services {
if err := service.Stop(); err != nil {
failure.Services[id] = err
}
}
n.running.Stop()
n.services = nil
n.running = nil
if len(failure.Services) > 0 {
return failure
}
return nil
}
// Restart terminates a running node and boots up a new one in its place. If the
// node isn't running, an error is returned.
func (n *Node) Restart() error {
if err := n.Stop(); err != nil {
return err
}
if err := n.Start(); err != nil {
return err
}
return nil
}
// 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 {
n.lock.RLock()
defer n.lock.RUnlock()
return n.running
}
// Service retrieves a currently running services registered under a given id.
func (n *Node) Service(id string) Service {
n.lock.RLock()
defer n.lock.RUnlock()
if n.services == nil {
return nil
}
return n.services[id]
}
// DataDir retrieves the current datadir used by the protocol stack.
func (n *Node) DataDir() string {
return n.datadir
}
// EventMux retrieves the event multiplexer used by all the network services in
// the current protocol stack.
func (n *Node) EventMux() *event.TypeMux {
return n.emux
}

87
node/node_example_test.go
Unescape View File

@ -0,0 +1,87 @@
// Copyright 2015 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum 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 go-ethereum 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 go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package node_test
import (
"fmt"
"log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
)
// SampleService is a trivial network service that can be attached to a node for
// life cycle management.
//
// The following methods are needed to implement a node.Service:
// - Protocols() []p2p.Protocol - devp2p protocols the service can communicate on
// - Start() error - method invoked when the node is ready to start the service
// - Stop() error - method invoked when the node terminates the service
type SampleService struct{}
func (s *SampleService) Protocols() []p2p.Protocol { return nil }
func (s *SampleService) Start() error { fmt.Println("Sample service starting..."); return nil }
func (s *SampleService) Stop() error { fmt.Println("Sample service stopping..."); return nil }
func ExampleUsage() {
// Create a network node to run protocols with the default values. The below list
// is only used to display each of the configuration options. All of these could
// have been ommited if the default behavior is desired.
nodeConfig := &node.Config{
DataDir: "", // Empty uses ephemeral storage
PrivateKey: nil, // Nil generates a node key on the fly
Name: "", // Any textual node name is allowed
NoDiscovery: false, // Can disable discovering remote nodes
BootstrapNodes: []*discover.Node{}, // List of bootstrap nodes to use
ListenAddr: ":0", // Network interface to listen on
NAT: nil, // UPnP port mapper to use for crossing firewalls
Dialer: nil, // Custom dialer to use for establishing peer connections
NoDial: false, // Can prevent this node from dialing out
MaxPeers: 0, // Number of peers to allow
MaxPendingPeers: 0, // Number of peers allowed to handshake concurrently
}
stack, err := node.New(nodeConfig)
if err != nil {
log.Fatalf("Failed to create network node: %v", err)
}
// Create and register a simple network service. This is done through the definition
// of a node.ServiceConstructor that will instantiate a node.Service. The reason for
// the factory method approach is to support service restarts without relying on the
// individual implementations' support for such operations.
constructor := func(context *node.ServiceContext) (node.Service, error) {
return new(SampleService), nil
}
if err := stack.Register("my sample service", constructor); err != nil {
log.Fatalf("Failed to register service: %v", err)
}
// Boot up the entire protocol stack, do a restart and terminate
if err := stack.Start(); err != nil {
log.Fatalf("Failed to start the protocol stack: %v", err)
}
if err := stack.Restart(); err != nil {
log.Fatalf("Failed to restart the protocol stack: %v", err)
}
if err := stack.Stop(); err != nil {
log.Fatalf("Failed to stop the protocol stack: %v", err)
}
// Output:
// Sample service starting...
// Sample service stopping...
// Sample service starting...
// Sample service stopping...
}

492
node/node_test.go
Unescape View File

@ -0,0 +1,492 @@
// Copyright 2015 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum 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 go-ethereum 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 go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package node
import (
"errors"
"io/ioutil"
"os"
"testing"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/p2p"
)
var (
testNodeKey, _ = crypto.GenerateKey()
testNodeConfig = &Config{
PrivateKey: testNodeKey,
Name: "test node",
}
)
// Tests that an empty protocol stack can be started, restarted and stopped.
func TestNodeLifeCycle(t *testing.T) {
stack, err := New(testNodeConfig)
if err != nil {
t.Fatalf("failed to create protocol stack: %v", err)
}
// Ensure that a stopped node can be stopped again
for i := 0; i < 3; i++ {
if err := stack.Stop(); err != ErrNodeStopped {
t.Fatalf("iter %d: stop failure mismatch: have %v, want %v", i, err, ErrNodeStopped)
}
}
// Ensure that a node can be successfully started, but only once
if err := stack.Start(); err != nil {
t.Fatalf("failed to start node: %v", err)
}
if err := stack.Start(); err != ErrNodeRunning {
t.Fatalf("start failure mismatch: have %v, want %v ", err, ErrNodeRunning)
}
// Ensure that a node can be restarted arbitrarily many times
for i := 0; i < 3; i++ {
if err := stack.Restart(); err != nil {
t.Fatalf("iter %d: failed to restart node: %v", i, err)
}
}
// Ensure that a node can be stopped, but only once
if err := stack.Stop(); err != nil {
t.Fatalf("failed to stop node: %v", err)
}
if err := stack.Stop(); err != ErrNodeStopped {
t.Fatalf("stop failure mismatch: have %v, want %v ", err, ErrNodeStopped)
}
}
// Tests that if the data dir is already in use, an appropriate error is returned.
func TestNodeUsedDataDir(t *testing.T) {
// Create a temporary folder to use as the data directory
dir, err := ioutil.TempDir("", "")
if err != nil {
t.Fatalf("failed to create temporary data directory: %v", err)
}
defer os.RemoveAll(dir)
// Create a new node based on the data directory
original, err := New(&Config{DataDir: dir})
if err != nil {
t.Fatalf("failed to create original protocol stack: %v", err)
}
if err := original.Start(); err != nil {
t.Fatalf("failed to start original protocol stack: %v", err)
}
defer original.Stop()
// Create a second node based on the same data directory and ensure failure
duplicate, err := New(&Config{DataDir: dir})
if err != nil {
t.Fatalf("failed to create duplicate protocol stack: %v", err)
}
if err := duplicate.Start(); err != ErrDatadirUsed {
t.Fatalf("duplicate datadir failure mismatch: have %v, want %v", err, ErrDatadirUsed)
}
}
// NoopService is a trivial implementation of the Service interface.
type NoopService struct{}
func (s *NoopService) Protocols() []p2p.Protocol { return nil }
func (s *NoopService) Start() error { return nil }
func (s *NoopService) Stop() error { return nil }
func NewNoopService(*ServiceContext) (Service, error) { return new(NoopService), nil }
// Tests whether services can be registered and unregistered.
func TestServiceRegistry(t *testing.T) {
stack, err := New(testNodeConfig)
if err != nil {
t.Fatalf("failed to create protocol stack: %v", err)
}
// Create a batch of dummy services and ensure they don't exist
ids := []string{"A", "B", "C"}
for i, id := range ids {
if err := stack.Unregister(id); err != ErrServiceUnknown {
t.Fatalf("service %d: pre-unregistration failure mismatch: have %v, want %v", i, err, ErrServiceUnknown)
}
}
// Register the services, checking that the operation succeeds only once
for i, id := range ids {
if err := stack.Register(id, NewNoopService); err != nil {
t.Fatalf("service %d: registration failed: %v", i, err)
}
if err := stack.Register(id, NewNoopService); err != ErrServiceRegistered {
t.Fatalf("service %d: registration failure mismatch: have %v, want %v", i, err, ErrServiceRegistered)
}
}
// Unregister the services, checking that the operation succeeds only once
for i, id := range ids {
if err := stack.Unregister(id); err != nil {
t.Fatalf("service %d: unregistration failed: %v", i, err)
}
if err := stack.Unregister(id); err != ErrServiceUnknown {
t.Fatalf("service %d: unregistration failure mismatch: have %v, want %v", i, err, ErrServiceUnknown)
}
}
}
// InstrumentedService is an implementation of Service for which all interface
// methods can be instrumented both return value as well as event hook wise.
type InstrumentedService struct {
protocols []p2p.Protocol
start error
stop error
protocolsHook func()
startHook func()
stopHook func()
}
func (s *InstrumentedService) Protocols() []p2p.Protocol {
if s.protocolsHook != nil {
s.protocolsHook()
}
return s.protocols
}
func (s *InstrumentedService) Start() error {
if s.startHook != nil {
s.startHook()
}
return s.start
}
func (s *InstrumentedService) Stop() error {
if s.stopHook != nil {
s.stopHook()
}
return s.stop
}
// Tests that registered services get started and stopped correctly.
func TestServiceLifeCycle(t *testing.T) {
stack, err := New(testNodeConfig)
if err != nil {
t.Fatalf("failed to create protocol stack: %v", err)
}
// Register a batch of life-cycle instrumented services
ids := []string{"A", "B", "C"}
started := make(map[string]bool)
stopped := make(map[string]bool)
for i, id := range ids {
id := id // Closure for the constructor
constructor := func(*ServiceContext) (Service, error) {
return &InstrumentedService{
startHook: func() { started[id] = true },
stopHook: func() { stopped[id] = true },
}, nil
}
if err := stack.Register(id, constructor); err != nil {
t.Fatalf("service %d: registration failed: %v", i, err)
}
}
// Start the node and check that all services are running
if err := stack.Start(); err != nil {
t.Fatalf("failed to start protocol stack: %v", err)
}
for i, id := range ids {
if !started[id] {
t.Fatalf("service %d: freshly started service not running", i)
}
if stopped[id] {
t.Fatalf("service %d: freshly started service already stopped", i)
}
if stack.Service(id) == nil {
t.Fatalf("service %d: freshly started service unaccessible", i)
}
}
// Stop the node and check that all services have been stopped
if err := stack.Stop(); err != nil {
t.Fatalf("failed to stop protocol stack: %v", err)
}
for i, id := range ids {
if !stopped[id] {
t.Fatalf("service %d: freshly terminated service still running", i)
}
if service := stack.Service(id); service != nil {
t.Fatalf("service %d: freshly terminated service still accessible: %v", i, service)
}
}
}
// Tests that services are restarted cleanly as new instances.
func TestServiceRestarts(t *testing.T) {
stack, err := New(testNodeConfig)
if err != nil {
t.Fatalf("failed to create protocol stack: %v", err)
}
// Define a service that does not support restarts
var (
running bool
started int
)
constructor := func(*ServiceContext) (Service, error) {
running = false
return &InstrumentedService{
startHook: func() {
if running {
panic("already running")
}
running = true
started++
},
}, nil
}
// Register the service and start the protocol stack
if err := stack.Register("service", constructor); err != nil {
t.Fatalf("failed to register the service: %v", err)
}
if err := stack.Start(); err != nil {
t.Fatalf("failed to start protocol stack: %v", err)
}
defer stack.Stop()
if running != true || started != 1 {
t.Fatalf("running/started mismatch: have %v/%d, want true/1", running, started)
}
// Restart the stack a few times and check successful service restarts
for i := 0; i < 3; i++ {
if err := stack.Restart(); err != nil {
t.Fatalf("iter %d: failed to restart stack: %v", i, err)
}
}
if running != true || started != 4 {
t.Fatalf("running/started mismatch: have %v/%d, want true/4", running, started)
}
}
// Tests that if a service fails to initialize itself, none of the other services
// will be allowed to even start.
func TestServiceConstructionAbortion(t *testing.T) {
stack, err := New(testNodeConfig)
if err != nil {
t.Fatalf("failed to create protocol stack: %v", err)
}
// Define a batch of good services
ids := []string{"A", "B", "C", "D", "E", "F"}
started := make(map[string]bool)
for i, id := range ids {
id := id // Closure for the constructor
constructor := func(*ServiceContext) (Service, error) {
return &InstrumentedService{
startHook: func() { started[id] = true },
}, nil
}
if err := stack.Register(id, constructor); err != nil {
t.Fatalf("service %d: registration failed: %v", i, err)
}
}
// Register a service that fails to construct itself
failure := errors.New("fail")
failer := func(*ServiceContext) (Service, error) {
return nil, failure
}
if err := stack.Register("failer", failer); err != nil {
t.Fatalf("failer registration failed: %v", err)
}
// Start the protocol stack and ensure none of the services get started
for i := 0; i < 100; i++ {
if err := stack.Start(); err != failure {
t.Fatalf("iter %d: stack startup failure mismatch: have %v, want %v", i, err, failure)
}
for i, id := range ids {
if started[id] {
t.Fatalf("service %d: started should not have", i)
}
delete(started, id)
}
}
}
// Tests that if a service fails to start, all others started before it will be
// shut down.
func TestServiceStartupAbortion(t *testing.T) {
stack, err := New(testNodeConfig)
if err != nil {
t.Fatalf("failed to create protocol stack: %v", err)
}
// Register a batch of good services
ids := []string{"A", "B", "C", "D", "E", "F"}
started := make(map[string]bool)
stopped := make(map[string]bool)
for i, id := range ids {
id := id // Closure for the constructor
constructor := func(*ServiceContext) (Service, error) {
return &InstrumentedService{
startHook: func() { started[id] = true },
stopHook: func() { stopped[id] = true },
}, nil
}
if err := stack.Register(id, constructor); err != nil {
t.Fatalf("service %d: registration failed: %v", i, err)
}
}
// Register a service that fails to start
failure := errors.New("fail")
failer := func(*ServiceContext) (Service, error) {
return &InstrumentedService{
start: failure,
}, nil
}
if err := stack.Register("failer", failer); err != nil {
t.Fatalf("failer registration failed: %v", err)
}
// Start the protocol stack and ensure all started services stop
for i := 0; i < 100; i++ {
if err := stack.Start(); err != failure {
t.Fatalf("iter %d: stack startup failure mismatch: have %v, want %v", i, err, failure)
}
for i, id := range ids {
if started[id] && !stopped[id] {
t.Fatalf("service %d: started but not stopped", i)
}
delete(started, id)
delete(stopped, id)
}
}
}
// Tests that even if a registered service fails to shut down cleanly, it does
// not influece the rest of the shutdown invocations.
func TestServiceTerminationGuarantee(t *testing.T) {
stack, err := New(testNodeConfig)
if err != nil {
t.Fatalf("failed to create protocol stack: %v", err)
}
// Register a batch of good services
ids := []string{"A", "B", "C", "D", "E", "F"}
started := make(map[string]bool)
stopped := make(map[string]bool)
for i, id := range ids {
id := id // Closure for the constructor
constructor := func(*ServiceContext) (Service, error) {
return &InstrumentedService{
startHook: func() { started[id] = true },
stopHook: func() { stopped[id] = true },
}, nil
}
if err := stack.Register(id, constructor); err != nil {
t.Fatalf("service %d: registration failed: %v", i, err)
}
}
// Register a service that fails to shot down cleanly
failure := errors.New("fail")
failer := func(*ServiceContext) (Service, error) {
return &InstrumentedService{
stop: failure,
}, nil
}
if err := stack.Register("failer", failer); err != nil {
t.Fatalf("failer registration failed: %v", err)
}
// Start the protocol stack, and ensure that a failing shut down terminates all
for i := 0; i < 100; i++ {
// Start the stack and make sure all is online
if err := stack.Start(); err != nil {
t.Fatalf("iter %d: failed to start protocol stack: %v", i, err)
}
for j, id := range ids {
if !started[id] {
t.Fatalf("iter %d, service %d: service not running", i, j)
}
if stopped[id] {
t.Fatalf("iter %d, service %d: service already stopped", i, j)
}
}
// Stop the stack, verify failure and check all terminations
err := stack.Stop()
if err, ok := err.(*StopError); !ok {
t.Fatalf("iter %d: termination failure mismatch: have %v, want StopError", i, err)
} else {
if err.Services["failer"] != failure {
t.Fatalf("iter %d: failer termination failure mismatch: have %v, want %v", i, err.Services["failer"], failure)
}
if len(err.Services) != 1 {
t.Fatalf("iter %d: failure count mismatch: have %d, want %d", i, len(err.Services), 1)
}
}
for j, id := range ids {
if !stopped[id] {
t.Fatalf("iter %d, service %d: service not terminated", i, j)
}
delete(started, id)
delete(stopped, id)
}
}
}
// Tests that all protocols defined by individual services get launched.
func TestProtocolGather(t *testing.T) {
stack, err := New(testNodeConfig)
if err != nil {
t.Fatalf("failed to create protocol stack: %v", err)
}
// Register a batch of services with some configured number of protocols
services := map[string]int{
"Zero Protocols": 0,
"Single Protocol": 1,
"Many Protocols": 25,
}
for id, count := range services {
protocols := make([]p2p.Protocol, count)
for i := 0; i < len(protocols); i++ {
protocols[i].Name = id
protocols[i].Version = uint(i)
}
constructor := func(*ServiceContext) (Service, error) {
return &InstrumentedService{
protocols: protocols,
}, nil
}
if err := stack.Register(id, constructor); err != nil {
t.Fatalf("service %s: registration failed: %v", id, err)
}
}
// Start the services and ensure all protocols start successfully
if err := stack.Start(); err != nil {
t.Fatalf("failed to start protocol stack: %v", err)
}
defer stack.Stop()
protocols := stack.Server().Protocols
if len(protocols) != 26 {
t.Fatalf("mismatching number of protocols launched: have %d, want %d", len(protocols), 26)
}
for id, count := range services {
for ver := 0; ver < count; ver++ {
launched := false
for i := 0; i < len(protocols); i++ {
if protocols[i].Name == id && protocols[i].Version == uint(ver) {
launched = true
break
}
}
if !launched {
t.Errorf("configured protocol not launched: %s v%d", id, ver)
}
}
}
}

67
node/service.go
Unescape View File

@ -0,0 +1,67 @@
// Copyright 2015 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum 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 go-ethereum 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 go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package node
import (
"path/filepath"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/p2p"
)
// ServiceContext is a collection of service independent options inherited from
// the protocol stack, that is passed to all constructors to be optionally used;
// as well as utility methods to operate on the service environment.
type ServiceContext struct {
dataDir string // Data directory for protocol persistence
EventMux *event.TypeMux // Event multiplexer used for decoupled notifications
}
// Database opens an existing database with the given name (or creates one if no
// previous can be found) from within the node's data directory. If the node is
// an ephemeral one, a memory database is returned.
func (ctx *ServiceContext) Database(name string, cache int) (ethdb.Database, error) {
if ctx.dataDir == "" {
return ethdb.NewMemDatabase()
}
return ethdb.NewLDBDatabase(filepath.Join(ctx.dataDir, name), cache)
}
// ServiceConstructor is the function signature of the constructors needed to be
// registered for service instantiation.
type ServiceConstructor func(ctx *ServiceContext) (Service, error)
// Service is an individual protocol that can be registered into a node.
//
// Notes:
// - Service life-cycle management is delegated to the node. The service is
// allowed to initialize itself upon creation, but no goroutines should be
// spun up outside of the Start method.
// - Restart logic is not required as the node will create a fresh instance
// every time a service is started.
type Service interface {
// Protocol retrieves the P2P protocols the service wishes to start.
Protocols() []p2p.Protocol
// Start spawns any goroutines required by the service.
Start() error
// Stop terminates all goroutines belonging to the service, blocking until they
// are all terminated.
Stop() error
}

60
node/service_test.go
Unescape View File

@ -0,0 +1,60 @@
// Copyright 2015 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum 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 go-ethereum 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 go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package node
import (
"io/ioutil"
"os"
"path/filepath"
"testing"
)
// Tests that service context methods work properly.
func TestServiceContext(t *testing.T) {
// Create a temporary folder and ensure no database is contained within
dir, err := ioutil.TempDir("", "")
if err != nil {
t.Fatalf("failed to create temporary data directory: %v", err)
}
defer os.RemoveAll(dir)
if _, err := os.Stat(filepath.Join(dir, "database")); err == nil {
t.Fatalf("non-created database already exists")
}
// Request the opening/creation of a database and ensure it persists to disk
ctx := &ServiceContext{dataDir: dir}
db, err := ctx.Database("persistent", 0)
if err != nil {
t.Fatalf("failed to open persistent database: %v", err)
}
db.Close()
if _, err := os.Stat(filepath.Join(dir, "persistent")); err != nil {
t.Fatalf("persistent database doesn't exists: %v", err)
}
// Request th opening/creation of an ephemeral database and ensure it's not persisted
ctx = &ServiceContext{dataDir: ""}
db, err = ctx.Database("ephemeral", 0)
if err != nil {
t.Fatalf("failed to open ephemeral database: %v", err)
}
db.Close()
if _, err := os.Stat(filepath.Join(dir, "ephemeral")); err == nil {
t.Fatalf("ephemeral database exists")
}
}

33
node/utils.go
Unescape View File

@ -0,0 +1,33 @@
// Copyright 2015 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum 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 go-ethereum 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 go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package node
import (
"path/filepath"
"github.com/ethereum/go-ethereum/ethdb"
)
// openDatabase opens an existing database with the given name from within the
// specified data directory, creating one if none exists. If the data directory
// is empty, an ephemeral memory database is returned.
func openDatabase(dataDir string, name string, cache int) (ethdb.Database, error) {
if dataDir == "" {
return ethdb.NewMemDatabase()
}
return ethdb.NewLDBDatabase(filepath.Join(dataDir, name), cache)
}

7
p2p/discover/table.go
Unescape View File

@ -90,12 +90,11 @@ type transport interface {
// that was most recently active is the first element in entries.
type bucket struct{ entries []*Node }
func newTable(t transport, ourID NodeID, ourAddr *net.UDPAddr, nodeDBPath string) *Table {
func newTable(t transport, ourID NodeID, ourAddr *net.UDPAddr, nodeDBPath string) (*Table, error) {
// If no node database was given, use an in-memory one
db, err := newNodeDB(nodeDBPath, Version, ourID)
if err != nil {
glog.V(logger.Warn).Infoln("Failed to open node database:", err)
db, _ = newNodeDB("", Version, ourID)
return nil, err
}
tab := &Table{
net: t,
@ -114,7 +113,7 @@ func newTable(t transport, ourID NodeID, ourAddr *net.UDPAddr, nodeDBPath string
tab.buckets[i] = new(bucket)
}
go tab.refreshLoop()
return tab
return tab, nil
}
// Self returns the local node.

8
p2p/discover/table_test.go
Unescape View File

@ -34,7 +34,7 @@ import (
func TestTable_pingReplace(t *testing.T) {
doit := func(newNodeIsResponding, lastInBucketIsResponding bool) {
transport := newPingRecorder()
tab := newTable(transport, NodeID{}, &net.UDPAddr{}, "")
tab, _ := newTable(transport, NodeID{}, &net.UDPAddr{}, "")
defer tab.Close()
pingSender := newNode(MustHexID("a502af0f59b2aab7746995408c79e9ca312d2793cc997e44fc55eda62f0150bbb8c59a6f9269ba3a081518b62699ee807c7c19c20125ddfccca872608af9e370"), net.IP{}, 99, 99)
@ -177,7 +177,7 @@ func TestTable_closest(t *testing.T) {
test := func(test *closeTest) bool {
// for any node table, Target and N
tab := newTable(nil, test.Self, &net.UDPAddr{}, "")
tab, _ := newTable(nil, test.Self, &net.UDPAddr{}, "")
defer tab.Close()
tab.stuff(test.All)
@ -236,7 +236,7 @@ func TestTable_ReadRandomNodesGetAll(t *testing.T) {
},
}
test := func(buf []*Node) bool {
tab := newTable(nil, NodeID{}, &net.UDPAddr{}, "")
tab, _ := newTable(nil, NodeID{}, &net.UDPAddr{}, "")
defer tab.Close()
for i := 0; i < len(buf); i++ {
ld := cfg.Rand.Intn(len(tab.buckets))
@ -279,7 +279,7 @@ func (*closeTest) Generate(rand *rand.Rand, size int) reflect.Value {
func TestTable_Lookup(t *testing.T) {
self := nodeAtDistance(common.Hash{}, 0)
tab := newTable(lookupTestnet, self.ID, &net.UDPAddr{}, "")
tab, _ := newTable(lookupTestnet, self.ID, &net.UDPAddr{}, "")
defer tab.Close()
// lookup on empty table returns no nodes

16
p2p/discover/udp.go
Unescape View File

@ -200,12 +200,15 @@ func ListenUDP(priv *ecdsa.PrivateKey, laddr string, natm nat.Interface, nodeDBP
if err != nil {
return nil, err
}
tab, _ := newUDP(priv, conn, natm, nodeDBPath)
tab, _, err := newUDP(priv, conn, natm, nodeDBPath)
if err != nil {
return nil, err
}
glog.V(logger.Info).Infoln("Listening,", tab.self)
return tab, nil
}
func newUDP(priv *ecdsa.PrivateKey, c conn, natm nat.Interface, nodeDBPath string) (*Table, *udp) {
func newUDP(priv *ecdsa.PrivateKey, c conn, natm nat.Interface, nodeDBPath string) (*Table, *udp, error) {
udp := &udp{
conn: c,
priv: priv,
@ -225,10 +228,15 @@ func newUDP(priv *ecdsa.PrivateKey, c conn, natm nat.Interface, nodeDBPath strin
}
// TODO: separate TCP port
udp.ourEndpoint = makeEndpoint(realaddr, uint16(realaddr.Port))
udp.Table = newTable(udp, PubkeyID(&priv.PublicKey), realaddr, nodeDBPath)
tab, err := newTable(udp, PubkeyID(&priv.PublicKey), realaddr, nodeDBPath)
if err != nil {
return nil, nil, err
}
udp.Table = tab
go udp.loop()
go udp.readLoop()
return udp.Table, udp
return udp.Table, udp, nil
}
func (t *udp) close() {

2
p2p/discover/udp_test.go
Unescape View File

@ -69,7 +69,7 @@ func newUDPTest(t *testing.T) *udpTest {
remotekey: newkey(),
remoteaddr: &net.UDPAddr{IP: net.IP{1, 2, 3, 4}, Port: 30303},
}
test.table, test.udp = newUDP(test.localkey, test.pipe, nil, "")
test.table, test.udp, _ = newUDP(test.localkey, test.pipe, nil, "")
return test
}

Write Preview
Loading…
Cancel
Save