go-ethereum源码分析 PartIII 共识流程

A: js指令转化为新transaction

在web3上操作流程

1. import to web3

2. connect to peers

3. read local key store

4. decrypt local key store

5. create a transaction obj

i. 创建contract： 向0发送，data为主要内容

ii. 执行contract

6. sign and send

在geth里怎么做到的呢？

首先看到cmdgethconsolecmd.go里面的localConsole，这个函数的功能是在启动geth节点的同时attach一个JS的控制台，没错，即使在本地geth处也是js的控制台前后端分开。

SUGAR:

gopkg.in/urfave/cli.v1 是用来辅助go命令行application的包

SUGAR END

// localConsole starts a new geth node, attaching a JavaScript console to it at the
// same time.
func localConsole(ctx *cli.Context) error {
    // Create and start the node based on the CLI flags
    node := makeFullNode(ctx)
    startNode(ctx, node)
    defer node.Close()

    // Attach to the newly started node and start the JavaScript console
    client, err := node.Attach()
    if err != nil {
        utils.Fatalf("Failed to attach to the inproc geth: %v", err)
    }
    config := console.Config{
        DataDir: utils.MakeDataDir(ctx),
        DocRoot: ctx.GlobalString(utils.JSpathFlag.Name),
        Client:  client,
        Preload: utils.MakeConsolePreloads(ctx),
    }

    console, err := console.New(config)
    if err != nil {
        utils.Fatalf("Failed to start the JavaScript console: %v", err)
    }
    defer console.Stop(false)

    // If only a short execution was requested, evaluate and return
    if script := ctx.GlobalString(utils.ExecFlag.Name); script != "" {
        console.Evaluate(script)
        return nil
    }
    // Otherwise print the welcome screen and enter interactive mode
    console.Welcome()
    console.Interactive()

    return nil
}

先只看consoleconsole.go，里面有console的方法，可以看到真的还是调用web3，不过会帮你做好import web3，关联方法，管理钱包，log之类的操作。

可以看到socker和bridge在这里起到了前端web3与后台node之间的连接作用

// init retrieves the available APIs from the remote RPC provider and initializes
// the console's JavaScript namespaces based on the exposed modules.
func (c *Console) init(preload []string) error {
    // Initialize the JavaScript <-> Go RPC bridge
    bridge := newBridge(c.client, c.prompter, c.printer)
    c.jsre.Set("jeth", struct{}{})

    jethObj, _ := c.jsre.Get("jeth")
    jethObj.Object().Set("send", bridge.Send)
    jethObj.Object().Set("sendAsync", bridge.Send)

    consoleObj, _ := c.jsre.Get("console")
    consoleObj.Object().Set("log", c.consoleOutput)
    consoleObj.Object().Set("error", c.consoleOutput)

    // Load all the internal utility JavaScript libraries
    if err := c.jsre.Compile("bignumber.js", jsre.BignumberJs); err != nil {
        return fmt.Errorf("bignumber.js: %v", err)
    }
    if err := c.jsre.Compile("web3.js", jsre.Web3Js); err != nil {
        return fmt.Errorf("web3.js: %v", err)
    }
    if _, err := c.jsre.Run("var Web3 = require('web3');"); err != nil {
        return fmt.Errorf("web3 require: %v", err)
    }
    if _, err := c.jsre.Run("var web3 = new Web3(jeth);"); err != nil {
        return fmt.Errorf("web3 provider: %v", err)
    }
    // Load the supported APIs into the JavaScript runtime environment
    apis, err := c.client.SupportedModules()
    if err != nil {
        return fmt.Errorf("api modules: %v", err)
    }
    flatten := "var eth = web3.eth; var personal = web3.personal; "
    for api := range apis {
        if api == "web3" {
            continue // manually mapped or ignore
        }
        if file, ok := web3ext.Modules[api]; ok {
            // Load our extension for the module.
            if err = c.jsre.Compile(fmt.Sprintf("%s.js", api), file); err != nil {
                return fmt.Errorf("%s.js: %v", api, err)
            }
            flatten += fmt.Sprintf("var %s = web3.%s; ", api, api)
        } else if obj, err := c.jsre.Run("web3." + api); err == nil && obj.IsObject() {
            // Enable web3.js built-in extension if available.
            flatten += fmt.Sprintf("var %s = web3.%s; ", api, api)
        }
    }
    if _, err = c.jsre.Run(flatten); err != nil {
        return fmt.Errorf("namespace flattening: %v", err)
    }
    // Initialize the global name register (disabled for now)
    //c.jsre.Run(`var GlobalRegistrar = eth.contract(` + registrar.GlobalRegistrarAbi + `);   registrar = GlobalRegistrar.at("` + registrar.GlobalRegistrarAddr + `");`)

    // If the console is in interactive mode, instrument password related methods to query the user
    if c.prompter != nil {
        // Retrieve the account management object to instrument
        personal, err := c.jsre.Get("personal")
        if err != nil {
            return err
        }
        // Override the openWallet, unlockAccount, newAccount and sign methods since
        // these require user interaction. Assign these method in the Console the
        // original web3 callbacks. These will be called by the jeth.* methods after
        // they got the password from the user and send the original web3 request to
        // the backend.
        if obj := personal.Object(); obj != nil { // make sure the personal api is enabled over the interface
            if _, err = c.jsre.Run(`jeth.openWallet = personal.openWallet;`); err != nil {
                return fmt.Errorf("personal.openWallet: %v", err)
            }
            if _, err = c.jsre.Run(`jeth.unlockAccount = personal.unlockAccount;`); err != nil {
                return fmt.Errorf("personal.unlockAccount: %v", err)
            }
            if _, err = c.jsre.Run(`jeth.newAccount = personal.newAccount;`); err != nil {
                return fmt.Errorf("personal.newAccount: %v", err)
            }
            if _, err = c.jsre.Run(`jeth.sign = personal.sign;`); err != nil {
                return fmt.Errorf("personal.sign: %v", err)
            }
            obj.Set("openWallet", bridge.OpenWallet)
            obj.Set("unlockAccount", bridge.UnlockAccount)
            obj.Set("newAccount", bridge.NewAccount)
            obj.Set("sign", bridge.Sign)
        }
    }
    // The admin.sleep and admin.sleepBlocks are offered by the console and not by the RPC layer.
    admin, err := c.jsre.Get("admin")
    if err != nil {
        return err
    }
    if obj := admin.Object(); obj != nil { // make sure the admin api is enabled over the interface
        obj.Set("sleepBlocks", bridge.SleepBlocks)
        obj.Set("sleep", bridge.Sleep)
        obj.Set("clearHistory", c.clearHistory)
    }
    // Preload any JavaScript files before starting the console
    for _, path := range preload {
        if err := c.jsre.Exec(path); err != nil {
            failure := err.Error()
            if ottoErr, ok := err.(*otto.Error); ok {
                failure = ottoErr.String()
            }
            return fmt.Errorf("%s: %v", path, failure)
        }
    }
    // Configure the console's input prompter for scrollback and tab completion
    if c.prompter != nil {
        if content, err := ioutil.ReadFile(c.histPath); err != nil {
            c.prompter.SetHistory(nil)
        } else {
            c.history = strings.Split(string(content), "
")
            c.prompter.SetHistory(c.history)
        }
        c.prompter.SetWordCompleter(c.AutoCompleteInput)
    }
    return nil
}

makefullnode设置了一些参数，注册了服务

func makeFullNode(ctx *cli.Context) *node.Node {
    stack, cfg := makeConfigNode(ctx)
    if ctx.GlobalIsSet(utils.ConstantinopleOverrideFlag.Name) {
        cfg.Eth.ConstantinopleOverride = new(big.Int).SetUint64(ctx.GlobalUint64(utils.ConstantinopleOverrideFlag.Name))
    }
    utils.RegisterEthService(stack, &cfg.Eth)

    if ctx.GlobalBool(utils.DashboardEnabledFlag.Name) {
        utils.RegisterDashboardService(stack, &cfg.Dashboard, gitCommit)
    }
    // Whisper must be explicitly enabled by specifying at least 1 whisper flag or in dev mode
    shhEnabled := enableWhisper(ctx)
    shhAutoEnabled := !ctx.GlobalIsSet(utils.WhisperEnabledFlag.Name) && ctx.GlobalIsSet(utils.DeveloperFlag.Name)
    if shhEnabled || shhAutoEnabled {
        if ctx.GlobalIsSet(utils.WhisperMaxMessageSizeFlag.Name) {
            cfg.Shh.MaxMessageSize = uint32(ctx.Int(utils.WhisperMaxMessageSizeFlag.Name))
        }
        if ctx.GlobalIsSet(utils.WhisperMinPOWFlag.Name) {
            cfg.Shh.MinimumAcceptedPOW = ctx.Float64(utils.WhisperMinPOWFlag.Name)
        }
        if ctx.GlobalIsSet(utils.WhisperRestrictConnectionBetweenLightClientsFlag.Name) {
            cfg.Shh.RestrictConnectionBetweenLightClients = true
        }
        utils.RegisterShhService(stack, &cfg.Shh)
    }

    // Configure GraphQL if required
    if ctx.GlobalIsSet(utils.GraphQLEnabledFlag.Name) {
        if err := graphql.RegisterGraphQLService(stack, cfg.Node.GraphQLEndpoint(), cfg.Node.GraphQLCors, cfg.Node.GraphQLVirtualHosts, cfg.Node.HTTPTimeouts); err != nil {
            utils.Fatalf("Failed to register the Ethereum service: %v", err)
        }
    }

    // Add the Ethereum Stats daemon if requested.
    if cfg.Ethstats.URL != "" {
        utils.RegisterEthStatsService(stack, cfg.Ethstats.URL)
    }
    return stack
}

startNode操作则提供了相当多有用的功能

1. 启动accountManager，unlock特定用户, suscribe，允许用户arrived wallet, open wallet和drop wallet

2. 启动downloader同步

3. 启动mining

// startNode boots up the system node and all registered protocols, after which
// it unlocks any requested accounts, and starts the RPC/IPC interfaces and the
// miner.
func startNode(ctx *cli.Context, stack *node.Node) {
    debug.Memsize.Add("node", stack)

    // Start up the node itself
    utils.StartNode(stack)

    // Unlock any account specifically requested
    if keystores := stack.AccountManager().Backends(keystore.KeyStoreType); len(keystores) > 0 {
        ks := keystores[0].(*keystore.KeyStore)
        passwords := utils.MakePasswordList(ctx)
        unlocks := strings.Split(ctx.GlobalString(utils.UnlockedAccountFlag.Name), ",")
        for i, account := range unlocks {
            if trimmed := strings.TrimSpace(account); trimmed != "" {
                unlockAccount(ctx, ks, trimmed, i, passwords)
            }
        }
    }
    // Register wallet event handlers to open and auto-derive wallets
    events := make(chan accounts.WalletEvent, 16)
    stack.AccountManager().Subscribe(events)

    go func() {
        // Create a chain state reader for self-derivation
        rpcClient, err := stack.Attach()
        if err != nil {
            utils.Fatalf("Failed to attach to self: %v", err)
        }
        stateReader := ethclient.NewClient(rpcClient)

        // Open any wallets already attached
        for _, wallet := range stack.AccountManager().Wallets() {
            if err := wallet.Open(""); err != nil {
                log.Warn("Failed to open wallet", "url", wallet.URL(), "err", err)
            }
        }
        // Listen for wallet event till termination
        for event := range events {
            switch event.Kind {
            case accounts.WalletArrived:
                if err := event.Wallet.Open(""); err != nil {
                    log.Warn("New wallet appeared, failed to open", "url", event.Wallet.URL(), "err", err)
                }
            case accounts.WalletOpened:
                status, _ := event.Wallet.Status()
                log.Info("New wallet appeared", "url", event.Wallet.URL(), "status", status)

                derivationPath := accounts.DefaultBaseDerivationPath
                if event.Wallet.URL().Scheme == "ledger" {
                    derivationPath = accounts.DefaultLedgerBaseDerivationPath
                }
                event.Wallet.SelfDerive(derivationPath, stateReader)

            case accounts.WalletDropped:
                log.Info("Old wallet dropped", "url", event.Wallet.URL())
                event.Wallet.Close()
            }
        }
    }()

    // Spawn a standalone goroutine for status synchronization monitoring,
    // close the node when synchronization is complete if user required.
    if ctx.GlobalBool(utils.ExitWhenSyncedFlag.Name) {
        go func() {
            sub := stack.EventMux().Subscribe(downloader.DoneEvent{})
            defer sub.Unsubscribe()
            for {
                event := <-sub.Chan()
                if event == nil {
                    continue
                }
                done, ok := event.Data.(downloader.DoneEvent)
                if !ok {
                    continue
                }
                if timestamp := time.Unix(done.Latest.Time.Int64(), 0); time.Since(timestamp) < 10*time.Minute {
                    log.Info("Synchronisation completed", "latestnum", done.Latest.Number, "latesthash", done.Latest.Hash(),
                        "age", common.PrettyAge(timestamp))
                    stack.Stop()
                }

            }
        }()
    }

    // Start auxiliary services if enabled
    if ctx.GlobalBool(utils.MiningEnabledFlag.Name) || ctx.GlobalBool(utils.DeveloperFlag.Name) {
        // Mining only makes sense if a full Ethereum node is running
        if ctx.GlobalString(utils.SyncModeFlag.Name) == "light" {
            utils.Fatalf("Light clients do not support mining")
        }
        var ethereum *eth.Ethereum
        if err := stack.Service(&ethereum); err != nil {
            utils.Fatalf("Ethereum service not running: %v", err)
        }
        // Set the gas price to the limits from the CLI and start mining
        gasprice := utils.GlobalBig(ctx, utils.MinerLegacyGasPriceFlag.Name)
        if ctx.IsSet(utils.MinerGasPriceFlag.Name) {
            gasprice = utils.GlobalBig(ctx, utils.MinerGasPriceFlag.Name)
        }
        ethereum.TxPool().SetGasPrice(gasprice)

        threads := ctx.GlobalInt(utils.MinerLegacyThreadsFlag.Name)
        if ctx.GlobalIsSet(utils.MinerThreadsFlag.Name) {
            threads = ctx.GlobalInt(utils.MinerThreadsFlag.Name)
        }
        if err := ethereum.StartMining(threads); err != nil {
            utils.Fatalf("Failed to start mining: %v", err)
        }
    }
}

B. 新transaction用私钥来sign

web3.js -> net.listener->RPC.server

i. AccountManager

accountManager存储了远端的wallet信息

// Manager is an overarching account manager that can communicate with various
// backends for signing transactions.
type Manager struct {
    backends map[reflect.Type][]Backend // Index of backends currently registered
    updaters []event.Subscription       // Wallet update subscriptions for all backends
    updates  chan WalletEvent           // Subscription sink for backend wallet changes
    wallets  []Wallet                   // Cache of all wallets from all registered backends

    feed event.Feed // Wallet feed notifying of arrivals/departures

    quit chan chan error
    lock sync.RWMutex
}

C. 在本地验证新transaction可行

D. 广播

E. 其他节点接受并验证

F: Miner节点接受并打包

G: 其他节点接受并承认