path: root/node/node.go

// 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"
    "net"
    "os"
    "path/filepath"
    "reflect"
    "sync"
    "syscall"

    "github.com/ethereum/go-ethereum/event"
    "github.com/ethereum/go-ethereum/internal/debug"
    "github.com/ethereum/go-ethereum/logger"
    "github.com/ethereum/go-ethereum/logger/glog"
    "github.com/ethereum/go-ethereum/p2p"
    "github.com/ethereum/go-ethereum/rpc"
)

var (
    ErrDatadirUsed    = errors.New("datadir already used")
    ErrNodeStopped    = errors.New("node not started")
    ErrNodeRunning    = errors.New("node already running")
    ErrServiceUnknown = errors.New("unknown service")

    datadirInUseErrnos = map[uint]bool{11: true, 32: true, 35: true}
)

// Node represents a P2P node into which arbitrary (uniquely typed) services might
// be registered.
type Node struct {
    datadir  string         // Path to the currently used data directory
    eventmux *event.TypeMux // Event multiplexer used between the services of a stack

    serverConfig p2p.Config
    server       *p2p.Server // Currently running P2P networking layer

    serviceFuncs []ServiceConstructor     // Service constructors (in dependency order)
    services     map[reflect.Type]Service // Currently running services

    rpcAPIs       []rpc.API   // List of APIs currently provided by the node
    inprocHandler *rpc.Server // In-process RPC request handler to process the API requests

    ipcEndpoint string       // IPC endpoint to listen at (empty = IPC disabled)
    ipcListener net.Listener // IPC RPC listener socket to serve API requests
    ipcHandler  *rpc.Server  // IPC RPC request handler to process the API requests

    httpHost      string       // HTTP hostname
    httpPort      int          // HTTP post
    httpEndpoint  string       // HTTP endpoint (interface + port) to listen at (empty = HTTP disabled)
    httpWhitelist []string     // HTTP RPC modules to allow through this endpoint
    httpCors      string       // HTTP RPC Cross-Origin Resource Sharing header
    httpListener  net.Listener // HTTP RPC listener socket to server API requests
    httpHandler   *rpc.Server  // HTTP RPC request handler to process the API requests

    wsHost      string       // Websocket host
    wsPort      int          // Websocket post
    wsEndpoint  string       // Websocket endpoint (interface + port) to listen at (empty = websocket disabled)
    wsWhitelist []string     // Websocket RPC modules to allow through this endpoint
    wsOrigins   string       // Websocket RPC allowed origin domains
    wsListener  net.Listener // Websocket RPC listener socket to server API requests
    wsHandler   *rpc.Server  // Websocket RPC request handler to process the API requests

    stop chan struct{} // Channel to wait for termination notifications
    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,
        serverConfig: p2p.Config{
            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,
        },
        serviceFuncs:  []ServiceConstructor{},
        ipcEndpoint:   conf.IPCEndpoint(),
        httpHost:      conf.HTTPHost,
        httpPort:      conf.HTTPPort,
        httpEndpoint:  conf.HTTPEndpoint(),
        httpWhitelist: conf.HTTPModules,
        httpCors:      conf.HTTPCors,
        wsHost:        conf.WSHost,
        wsPort:        conf.WSPort,
        wsEndpoint:    conf.WSEndpoint(),
        wsWhitelist:   conf.WSModules,
        wsOrigins:     conf.WSOrigins,
        eventmux:      new(event.TypeMux),
    }, nil
}

// Register injects a new service into the node's stack. The service created by
// the passed constructor must be unique in its type with regard to sibling ones.
func (n *Node) Register(constructor ServiceConstructor) error {
    n.lock.Lock()
    defer n.lock.Unlock()

    if n.server != nil {
        return ErrNodeRunning
    }
    n.serviceFuncs = append(n.serviceFuncs, constructor)
    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.server != nil {
        return ErrNodeRunning
    }
    // Otherwise copy and specialize the P2P configuration
    running := &p2p.Server{Config: n.serverConfig}
    services := make(map[reflect.Type]Service)
    for _, constructor := range n.serviceFuncs {
        // Create a new context for the particular service
        ctx := &ServiceContext{
            datadir:  n.datadir,
            services: make(map[reflect.Type]Service),
            EventMux: n.eventmux,
        }
        for kind, s := range services { // copy needed for threaded access
            ctx.services[kind] = s
        }
        // Construct and save the service
        service, err := constructor(ctx)
        if err != nil {
            return err
        }
        kind := reflect.TypeOf(service)
        if _, exists := services[kind]; exists {
            return &DuplicateServiceError{Kind: kind}
        }
        services[kind] = 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 := []reflect.Type{}
    for kind, service := range services {
        // Start the next service, stopping all previous upon failure
        if err := service.Start(running); err != nil {
            for _, kind := range started {
                services[kind].Stop()
            }
            running.Stop()

            return err
        }
        // Mark the service started for potential cleanup
        started = append(started, kind)
    }
    // Lastly start the configured RPC interfaces
    if err := n.startRPC(services); err != nil {
        for _, service := range services {
            service.Stop()
        }
        running.Stop()
        return err
    }
    // Finish initializing the startup
    n.services = services
    n.server = running
    n.stop = make(chan struct{})

    return nil
}

// startRPC is a helper method to start all the various RPC endpoint during node
// startup. It's not meant to be called at any time afterwards as it makes certain
// assumptions about the state of the node.
func (n *Node) startRPC(services map[reflect.Type]Service) error {
    // Gather all the possible APIs to surface
    apis := n.apis()
    for _, service := range services {
        apis = append(apis, service.APIs()...)
    }
    // Start the various API endpoints, terminating all in case of errors
    if err := n.startInProc(apis); err != nil {
        return err
    }
    if err := n.startIPC(apis); err != nil {
        n.stopInProc()
        return err
    }
    if err := n.startHTTP(n.httpEndpoint, apis, n.httpWhitelist, n.httpCors); err != nil {
        n.stopIPC()
        n.stopInProc()
        return err
    }
    if err := n.startWS(n.wsEndpoint, apis, n.wsWhitelist, n.wsOrigins); err != nil {
        n.stopHTTP()
        n.stopIPC()
        n.stopInProc()
        return err
    }
    // All API endpoints started successfully
    n.rpcAPIs = apis
    return nil
}

// startInProc initializes an in-process RPC endpoint.
func (n *Node) startInProc(apis []rpc.API) error {
    // Register all the APIs exposed by the services
    handler := rpc.NewServer()
    for _, api := range apis {
        if err := handler.RegisterName(api.Namespace, api.Service); err != nil {
            return err
        }
        glog.V(logger.Debug).Infof("InProc registered %T under '%s'", api.Service, api.Namespace)
    }
    n.inprocHandler = handler
    return nil
}

// stopInProc terminates the in-process RPC endpoint.
func (n *Node) stopInProc() {
    if n.inprocHandler != nil {
        n.inprocHandler.Stop()
        n.inprocHandler = nil
    }
}

// startIPC initializes and starts the IPC RPC endpoint.
func (n *Node) startIPC(apis []rpc.API) error {
    // Short circuit if the IPC endpoint isn't being exposed
    if n.ipcEndpoint == "" {
        return nil
    }
    // Register all the APIs exposed by the services
    handler := rpc.NewServer()
    for _, api := range apis {
        if err := handler.RegisterName(api.Namespace, api.Service); err != nil {
            return err
        }
        glog.V(logger.Debug).Infof("IPC registered %T under '%s'", api.Service, api.Namespace)
    }
    // All APIs registered, start the IPC listener
    var (
        listener net.Listener
        err      error
    )
    if listener, err = rpc.CreateIPCListener(n.ipcEndpoint); err != nil {
        return err
    }
    go func() {
        glog.V(logger.Info).Infof("IPC endpoint opened: %s", n.ipcEndpoint)

        for {
            conn, err := listener.Accept()
            if err != nil {
                // Terminate if the listener was closed
                n.lock.RLock()
                closed := n.ipcListener == nil
                n.lock.RUnlock()
                if closed {
                    return
                }
                // Not closed, just some error; report and continue
                glog.V(logger.Error).Infof("IPC accept failed: %v", err)
                continue
            }
            go handler.ServeCodec(rpc.NewJSONCodec(conn), rpc.OptionMethodInvocation|rpc.OptionSubscriptions)
        }
    }()
    // All listeners booted successfully
    n.ipcListener = listener
    n.ipcHandler = handler

    return nil
}

// stopIPC terminates the IPC RPC endpoint.
func (n *Node) stopIPC() {
    if n.ipcListener != nil {
        n.ipcListener.Close()
        n.ipcListener = nil

        glog.V(logger.Info).Infof("IPC endpoint closed: %s", n.ipcEndpoint)
    }
    if n.ipcHandler != nil {
        n.ipcHandler.Stop()
        n.ipcHandler = nil
    }
}

// startHTTP initializes and starts the HTTP RPC endpoint.
func (n *Node) startHTTP(endpoint string, apis []rpc.API, modules []string, cors string) error {
    // Short circuit if the HTTP endpoint isn't being exposed
    if endpoint == "" {
        return nil
    }
    // Generate the whitelist based on the allowed modules
    whitelist := make(map[string]bool)
    for _, module := range modules {
        whitelist[module] = true
    }
    // Register all the APIs exposed by the services
    handler := rpc.NewServer()
    for _, api := range apis {
        if whitelist[api.Namespace] || (len(whitelist) == 0 && api.Public) {
            if err := handler.RegisterName(api.Namespace, api.Service); err != nil {
                return err
            }
            glog.V(logger.Debug).Infof("HTTP registered %T under '%s'", api.Service, api.Namespace)
        }
    }
    // All APIs registered, start the HTTP listener
    var (
        listener net.Listener
        err      error
    )
    if listener, err = net.Listen("tcp", endpoint); err != nil {
        return err
    }
    go rpc.NewHTTPServer(cors, handler).Serve(listener)
    glog.V(logger.Info).Infof("HTTP endpoint opened: http://%s", endpoint)

    // All listeners booted successfully
    n.httpEndpoint = endpoint
    n.httpListener = listener
    n.httpHandler = handler
    n.httpCors = cors

    return nil
}

// stopHTTP terminates the HTTP RPC endpoint.
func (n *Node) stopHTTP() {
    if n.httpListener != nil {
        n.httpListener.Close()
        n.httpListener = nil

        glog.V(logger.Info).Infof("HTTP endpoint closed: http://%s", n.httpEndpoint)
    }
    if n.httpHandler != nil {
        n.httpHandler.Stop()
        n.httpHandler = nil
    }
}

// startWS initializes and starts the websocket RPC endpoint.
func (n *Node) startWS(endpoint string, apis []rpc.API, modules []string, wsOrigins string) error {
    // Short circuit if the WS endpoint isn't being exposed
    if endpoint == "" {
        return nil
    }
    // Generate the whitelist based on the allowed modules
    whitelist := make(map[string]bool)
    for _, module := range modules {
        whitelist[module] = true
    }
    // Register all the APIs exposed by the services
    handler := rpc.NewServer()
    for _, api := range apis {
        if whitelist[api.Namespace] || (len(whitelist) == 0 && api.Public) {
            if err := handler.RegisterName(api.Namespace, api.Service); err != nil {
                return err
            }
            glog.V(logger.Debug).Infof("WebSocket registered %T under '%s'", api.Service, api.Namespace)
        }
    }
    // All APIs registered, start the HTTP listener
    var (
        listener net.Listener
        err      error
    )
    if listener, err = net.Listen("tcp", endpoint); err != nil {
        return err
    }
    go rpc.NewWSServer(wsOrigins, handler).Serve(listener)
    glog.V(logger.Info).Infof("WebSocket endpoint opened: ws://%s", endpoint)

    // All listeners booted successfully
    n.wsEndpoint = endpoint
    n.wsListener = listener
    n.wsHandler = handler
    n.wsOrigins = wsOrigins

    return nil
}

// stopWS terminates the websocket RPC endpoint.
func (n *Node) stopWS() {
    if n.wsListener != nil {
        n.wsListener.Close()
        n.wsListener = nil

        glog.V(logger.Info).Infof("WebSocket endpoint closed: ws://%s", n.wsEndpoint)
    }
    if n.wsHandler != nil {
        n.wsHandler.Stop()
        n.wsHandler = 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.server == nil {
        return ErrNodeStopped
    }
    // Otherwise terminate the API, all services and the P2P server too
    n.stopWS()
    n.stopHTTP()
    n.stopIPC()
    n.rpcAPIs = nil

    failure := &StopError{
        Services: make(map[reflect.Type]error),
    }
    for kind, service := range n.services {
        if err := service.Stop(); err != nil {
            failure.Services[kind] = err
        }
    }
    n.server.Stop()

    n.services = nil
    n.server = nil
    close(n.stop)

    if len(failure.Services) > 0 {
        return failure
    }
    return nil
}

// Wait blocks the thread until the node is stopped. If the node is not running
// at the time of invocation, the method immediately returns.
func (n *Node) Wait() {
    n.lock.RLock()
    if n.server == nil {
        return
    }
    stop := n.stop
    n.lock.RUnlock()

    <-stop
}

// 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
}

// Attach creates an RPC client attached to an in-process API handler.
func (n *Node) Attach() (rpc.Client, error) {
    n.lock.RLock()
    defer n.lock.RUnlock()

    // Short circuit if the node's not running
    if n.server == nil {
        return nil, ErrNodeStopped
    }
    // Otherwise attach to the API and return
    return rpc.NewInProcRPCClient(n.inprocHandler), 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.server
}

// Service retrieves a currently running service registered of a specific type.
func (n *Node) Service(service interface{}) error {
    n.lock.RLock()
    defer n.lock.RUnlock()

    // Short circuit if the node's not running
    if n.server == nil {
        return ErrNodeStopped
    }
    // Otherwise try to find the service to return
    element := reflect.ValueOf(service).Elem()
    if running, ok := n.services[element.Type()]; ok {
        element.Set(reflect.ValueOf(running))
        return nil
    }
    return ErrServiceUnknown
}

// DataDir retrieves the current datadir used by the protocol stack.
func (n *Node) DataDir() string {
    return n.datadir
}

// IPCEndpoint retrieves the current IPC endpoint used by the protocol stack.
func (n *Node) IPCEndpoint() string {
    return n.ipcEndpoint
}

// HTTPEndpoint retrieves the current HTTP endpoint used by the protocol stack.
func (n *Node) HTTPEndpoint() string {
    return n.httpEndpoint
}

// WSEndpoint retrieves the current WS endpoint used by the protocol stack.
func (n *Node) WSEndpoint() string {
    return n.wsEndpoint
}

// EventMux retrieves the event multiplexer used by all the network services in
// the current protocol stack.
func (n *Node) EventMux() *event.TypeMux {
    return n.eventmux
}

// apis returns the collection of RPC descriptors this node offers.
func (n *Node) apis() []rpc.API {
    return []rpc.API{
        {
            Namespace: "admin",
            Version:   "1.0",
            Service:   NewPrivateAdminAPI(n),
        }, {
            Namespace: "admin",
            Version:   "1.0",
            Service:   NewPublicAdminAPI(n),
            Public:    true,
        }, {
            Namespace: "debug",
            Version:   "1.0",
            Service:   debug.Handler,
        }, {
            Namespace: "debug",
            Version:   "1.0",
            Service:   NewPublicDebugAPI(n),
            Public:    true,
        }, {
            Namespace: "web3",
            Version:   "1.0",
            Service:   NewPublicWeb3API(n),
            Public:    true,
        },
    }
}