以太坊源碼分析(43）node源碼分析

        confCopy := *conf

        conf = &confCopy

        if conf.DataDir != "" { //轉化爲絕對路徑。

            absdatadir, err := filepath.Abs(conf.DataDir)

            if err != nil {

                return nil, err

            }

            conf.DataDir = absdatadir

        }

        // Ensure that the instance name doesn't cause weird conflicts with

        // other files in the data directory.

        if strings.ContainsAny(conf.Name, `/\`) {

            return nil, errors.New(`Config.Name must not contain '/' or '\'`)

        }

        if conf.Name == datadirDefaultKeyStore {

            return nil, errors.New(`Config.Name cannot be "` + datadirDefaultKeyStore + `"`)

        }

        if strings.HasSuffix(conf.Name, ".ipc") {

            return nil, errors.New(`Config.Name cannot end in ".ipc"`)

        }

        // Ensure that the AccountManager method works before the node has started.

        // We rely on this in cmd/geth.

        am, ephemeralKeystore, err := makeAccountManager(conf)

        if err != nil {

            return nil, err

        }

        // Note: any interaction with Config that would create/touch files

        // in the data directory or instance directory is delayed until Start.

        return &Node{

            accman: am,

            ephemeralKeystore: ephemeralKeystore,

            config: conf,

            serviceFuncs: []ServiceConstructor{},

            ipcEndpoint: conf.IPCEndpoint(),

            httpEndpoint: conf.HTTPEndpoint(),

            wsEndpoint: conf.WSEndpoint(),

            eventmux: new(event.TypeMux),

        }, nil

    }

### node 服務和協議的註冊

由於node並無負責具體的業務邏輯。因此具體的業務邏輯是經過註冊的方式來註冊到node裏面來的。

其餘模塊經過Register方法來註冊了一個 服務構造函數。 使用這個服務構造函數能夠生成服務。

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

    }

服務是什麼

    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.

    // 服務的生命週期管理已經代理給node管理。該服務容許在建立時自動初始化，可是在Start方法以外不該該啓動goroutines。

    // 從新啓動邏輯不是必需的，由於節點將在每次啓動服務時建立一個新的實例。

    type Service interface {

        // Protocols retrieves the P2P protocols the service wishes to start.

        // 服務但願提供的p2p協議

        Protocols() []p2p.Protocol

        

        // APIs retrieves the list of RPC descriptors the service provides

        // 服務但願提供的RPC方法的描述

        APIs() []rpc.API

    

        // Start is called after all services have been constructed and the networking

        // layer was also initialized to spawn any goroutines required by the service.

        // 全部服務已經構建完成後，調用開始，而且網絡層也被初始化以產生服務所需的任何goroutine。

        Start(server *p2p.Server) error

    

        // Stop terminates all goroutines belonging to the service, blocking until they

        // are all terminated.

        

        // Stop方法會中止這個服務擁有的全部goroutine。 須要阻塞到全部的goroutine都已經終止

        Stop() error

    }

### node的啓動

node的啓動過程會建立和運行一個p2p的節點。

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

        }

        if err := n.openDataDir(); err != nil {

            return err

        }

    

        // Initialize the p2p server. This creates the node key and

        // discovery databases.

        n.serverConfig = n.config.P2P

        n.serverConfig.PrivateKey = n.config.NodeKey()

        n.serverConfig.Name = n.config.NodeName()

        if n.serverConfig.StaticNodes == nil {

            // 處理配置文件static-nodes.json

            n.serverConfig.StaticNodes = n.config.StaticNodes()

        }

        if n.serverConfig.TrustedNodes == nil {

            // 處理配置文件trusted-nodes.json

            n.serverConfig.TrustedNodes = n.config.TrustedNodes()

        }

        if n.serverConfig.NodeDatabase == "" {

            n.serverConfig.NodeDatabase = n.config.NodeDB()

        }

        //建立了p2p服務器

        running := &p2p.Server{Config: n.serverConfig}

        log.Info("Starting peer-to-peer node", "instance", n.serverConfig.Name)

    

        // Otherwise copy and specialize the P2P configuration

        services := make(map[reflect.Type]Service)

        for _, constructor := range n.serviceFuncs {

            // Create a new context for the particular service

            ctx := &ServiceContext{

                config: n.config,

                services: make(map[reflect.Type]Service),

                EventMux: n.eventmux,

                AccountManager: n.accman,

            }

            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

        // 收集全部的p2p的protocols並插入p2p.Rrotocols

        for _, service := range services {

            running.Protocols = append(running.Protocols, service.Protocols()...)

        }

        // 啓動了p2p服務器

        if err := running.Start(); err != nil {

            return convertFileLockError(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

        // 最後啓動RPC服務

        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,這個方法收集全部的apis。 並依次調用啓動各個RPC服務器， 默認是啓動InProc和IPC。 若是指定也能夠配置是否啓動HTTP和websocket。

    // 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.config.HTTPModules, n.config.HTTPCors); err != nil {

            n.stopIPC()

            n.stopInProc()

            return err

        }

        if err := n.startWS(n.wsEndpoint, apis, n.config.WSModules, n.config.WSOrigins, n.config.WSExposeAll); err != nil {

            n.stopHTTP()

            n.stopIPC()

            n.stopInProc()

            return err

        }

        // All API endpoints started successfully

        n.rpcAPIs = apis

        return nil

    }

startXXX 是具體的RPC的啓動。 流程都是大同小異。 這裏就只看startWS了

    // startWS initializes and starts the websocket RPC endpoint.

    func (n *Node) startWS(endpoint string, apis []rpc.API, modules []string, wsOrigins []string, exposeAll bool) 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 exposeAll || whitelist[api.Namespace] || (len(whitelist) == 0 && api.Public) {

            // 只有這集中狀況下才會把這個api進行註冊。

                if err := handler.RegisterName(api.Namespace, api.Service); err != nil {

                    return err

                }

                log.Debug(fmt.Sprintf("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)

        log.Info(fmt.Sprintf("WebSocket endpoint opened: ws://%s", listener.Addr()))

    

        // All listeners booted successfully

        n.wsEndpoint = endpoint

        n.wsListener = listener

        n.wsHandler = handler

    

        return nil

    }