Welcome

1
func makeConfigNode(ctx *cli.Context) (*node.Node, gethConfig) {
2
    cfg := loadBaseConfig(ctx)
3
    stack, err := node.New(&cfg.Node)
4
    if err != nil {
5
        utils.Fatalf("Failed to create the protocol stack: %v", err)
6
    }
7
    if err := setAccountManagerBackends(stack.Config(), stack.AccountManager(), stack.KeyStoreDir()); err != nil {
8
        utils.Fatalf("Failed to set account manager backends: %v", err)
9
    }
10
    utils.SetEthConfig(ctx, stack, &cfg.Eth)
11
    // ...
12
    return stack, cfg
13
}

1
backend, eth := utils.RegisterEthService(stack, &cfg.Eth)

1
func RegisterEthService(stack *node.Node, cfg *ethconfig.Config) (*eth.EthAPIBackend, *eth.Ethereum) {
2
    backend, err := eth.New(stack, cfg)
3
    if err != nil {
4
        Fatalf("Failed to register the Ethereum service: %v", err)
5
    }
6
    stack.RegisterAPIs(tracers.APIs(backend.APIBackend))
7
    return backend.APIBackend, backend
8
}

1
filterSystem := utils.RegisterFilterAPI(stack, backend, &cfg.Eth)
2

3
if ctx.IsSet(utils.GraphQLEnabledFlag.Name) {
4
    utils.RegisterGraphQLService(stack, backend, filterSystem, &cfg.Node)
5
}
6
if cfg.Ethstats.URL != "" {
7
    utils.RegisterEthStatsService(stack, backend, cfg.Ethstats.URL)
8
}

1
if ctx.IsSet(utils.DeveloperFlag.Name) {
2
    simBeacon, err := catalyst.NewSimulatedBeacon(...)
3
    // ...
4
    catalyst.RegisterSimulatedBeaconAPIs(stack, simBeacon)
5
    stack.RegisterLifecycle(simBeacon)
6
} else if ctx.IsSet(utils.BeaconApiFlag.Name) {
7
    blsyncer := blsync.NewClient(...)
8
    stack.RegisterLifecycle(blsyncer)
9
} else {
10
    err := catalyst.Register(stack, eth)
11
    // ...
12
}

1
func (n *Node) RegisterLifecycle(lifecycle Lifecycle) {
2
    n.lock.Lock()
3
    defer n.lock.Unlock()
4

5
    if n.state != initializingState {
6
        panic("can't register lifecycle on running/stopped node")
7
    }
8
    if slices.Contains(n.lifecycles, lifecycle) {
9
        panic(fmt.Sprintf("attempt to register lifecycle %T more than once", lifecycle))
10
    }
11
    n.lifecycles = append(n.lifecycles, lifecycle)
12
}

1
func New(conf *Config) (*Node, error) {
2
    // ... copy config, resolve absolute DataDir, validate Name ...
3

4
    server := rpc.NewServer()
5
    server.SetBatchLimits(conf.BatchRequestLimit, conf.BatchResponseMaxSize)
6
    node := &Node{
7
        config:        conf,
8
        inprocHandler: server,
9
        eventmux:      new(event.TypeMux),
10
        log:           conf.Logger,
11
        stop:          make(chan struct{}),
12
        server:        &p2p.Server{Config: conf.P2P},
13
        databases:     make(map[*closeTrackingDB]struct{}),
14
    }
15

16
    node.rpcAPIs = append(node.rpcAPIs, node.apis()...)
17
    if err := node.openDataDir(); err != nil {
18
        return nil, err
19
    }
20
    // ... set up keyDir, accman, p2p server config, RPC servers ...
21
    return node, nil
22
}

1
func (n *Node) Start() error {
2
    n.startStopLock.Lock()
3
    defer n.startStopLock.Unlock()
4

5
    n.lock.Lock()
6
    switch n.state {
7
    case runningState:
8
        n.lock.Unlock()
9
        return ErrNodeRunning
10
    case closedState:
11
        n.lock.Unlock()
12
        return ErrNodeStopped
13
    }
14
    n.state = runningState
15
    err := n.openEndpoints()
16
    lifecycles := make([]Lifecycle, len(n.lifecycles))
17
    copy(lifecycles, n.lifecycles)
18
    n.lock.Unlock()
19

20
    if err != nil {
21
        n.doClose(nil)
22
        return err
23
    }
24
    var started []Lifecycle
25
    for _, lifecycle := range lifecycles {
26
        if err = lifecycle.Start(); err != nil {
27
            break
28
        }
29
        started = append(started, lifecycle)
30
    }
31
    if err != nil {
32
        n.stopServices(started)
33
        n.doClose(nil)
34
    }
35
    return err
36
}

1
func (n *Node) openEndpoints() error {
2
    n.log.Info("Starting peer-to-peer node", "instance", n.server.Name)
3
    if err := n.server.Start(); err != nil {
4
        return convertFileLockError(err)
5
    }
6
    err := n.startRPC()
7
    if err != nil {
8
        n.stopRPC()
9
        n.server.Stop()
10
    }
11
    return err
12
}

1
func (s *Ethereum) Stop() error {
2
    // 1. Stop peer-related components
3
    s.discmix.Close()
4
    s.dropper.Stop()
5
    s.handler.Stop()
6

7
    // 2. Stop internal services
8
    ch := make(chan struct{})
9
    s.closeFilterMaps <- ch
10
    <-ch
11
    s.filterMaps.Stop()
12
    s.txPool.Close()
13
    s.blockchain.Stop()
14
    s.engine.Close()
15

16
    // 3. Mark clean shutdown, close database
17
    s.shutdownTracker.Stop()
18
    s.chainDb.Close()
19
    s.eventMux.Stop()
20
    return nil
21
}

1
func (n *Node) Close() error {
2
    n.startStopLock.Lock()
3
    defer n.startStopLock.Unlock()
4

5
    n.lock.Lock()
6
    state := n.state
7
    n.lock.Unlock()
8
    switch state {
9
    case initializingState:
10
        return n.doClose(nil)
11
    case runningState:
12
        var errs []error
13
        if err := n.stopServices(n.lifecycles); err != nil {
14
            errs = append(errs, err)
15
        }
16
        return n.doClose(errs)
17
    case closedState:
18
        return ErrNodeStopped
19
    }
20
    // ...
21
}

1
func (n *Node) stopServices(running []Lifecycle) error {
2
    n.stopRPC()
3

4
    failure := &StopError{Services: make(map[reflect.Type]error)}
5
    for i := len(running) - 1; i >= 0; i-- {
6
        if err := running[i].Stop(); err != nil {
7
            failure.Services[reflect.TypeOf(running[i])] = err
8
        }
9
    }
10
    n.server.Stop()
11
    // ...
12
}

1
func (n *Node) doClose(errs []error) error {
2
    n.lock.Lock()
3
    n.state = closedState
4
    errs = append(errs, n.closeDatabases()...)
5
    n.lock.Unlock()
6

7
    if err := n.accman.Close(); err != nil {
8
        errs = append(errs, err)
9
    }
10
    if n.keyDirTemp {
11
        if err := os.RemoveAll(n.keyDir); err != nil {
12
            errs = append(errs, err)
13
        }
14
    }
15
    n.closeDataDir()
16
    close(n.stop) // unblock Wait()
17
    // ...
18
}

1
func (f *Feed) Subscribe(channel interface{}) Subscription {
2
    chanval := reflect.ValueOf(channel)
3
    chantyp := chanval.Type()
4
    if chantyp.Kind() != reflect.Chan || chantyp.ChanDir()&reflect.SendDir == 0 {
5
        panic(errBadChannel)
6
    }
7
    sub := &feedSub{feed: f, channel: chanval, err: make(chan error, 1)}
8

9
    f.once.Do(func() { f.init(chantyp.Elem()) })
10
    // ...
11
    f.mu.Lock()
12
    defer f.mu.Unlock()
13
    cas := reflect.SelectCase{Dir: reflect.SelectSend, Chan: chanval}
14
    f.inbox = append(f.inbox, cas)
15
    return sub
16
}

1
func (f *Feed) Send(value interface{}) (nsent int) {
2
    rvalue := reflect.ValueOf(value)
3
    // ...
4
    <-f.sendLock
5

6
    // Merge inbox into sendCases
7
    f.mu.Lock()
8
    f.sendCases = append(f.sendCases, f.inbox...)
9
    f.inbox = nil
10
    f.mu.Unlock()
11

12
    // Set value on all channels
13
    for i := firstSubSendCase; i < len(f.sendCases); i++ {
14
        f.sendCases[i].Send = rvalue
15
    }
16

17
    cases := f.sendCases
18
    for {
19
        // Fast path: TrySend (non-blocking)
20
        for i := firstSubSendCase; i < len(cases); i++ {
21
            if cases[i].Chan.TrySend(rvalue) {
22
                nsent++
23
                cases = cases.deactivate(i)
24
                i--
25
            }
26
        }
27
        if len(cases) == firstSubSendCase {
28
            break
29
        }
30
        // Slow path: reflect.Select (blocking)
31
        chosen, recv, _ := reflect.Select(cases)
32
        if chosen == 0 { // removeSub channel
33
            // ... handle unsubscription ...
34
        } else {
35
            cases = cases.deactivate(chosen)
36
            nsent++
37
        }
38
    }
39
    // ...
40
}

1
// Producer side (e.g., in blockchain.go)
2
type BlockChain struct {
3
    chainHeadFeed event.Feed
4
    // ...
5
}
6

7
func (bc *BlockChain) SubscribeChainHeadEvent(ch chan<- ChainHeadEvent) event.Subscription {
8
    return bc.chainHeadFeed.Subscribe(ch)
9
}
10

11
// Inside block insertion:
12
// bc.chainHeadFeed.Send(ChainHeadEvent{Block: block})

1
// Consumer side (e.g., in miner or handler)
2
headCh := make(chan core.ChainHeadEvent, chainHeadChanSize)
3
sub := bc.SubscribeChainHeadEvent(headCh)
4
defer sub.Unsubscribe()
5

6
for {
7
    select {
8
    case head := <-headCh:
9
        // react to new chain head
10
    case err := <-sub.Err():
11
        return
12
    }
13
}