path: root/les/server.go

// Copyright 2016 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 les

import (
    "crypto/ecdsa"
    "time"

    "github.com/ethereum/go-ethereum/accounts/abi/bind"
    "github.com/ethereum/go-ethereum/common/mclock"
    "github.com/ethereum/go-ethereum/core"
    "github.com/ethereum/go-ethereum/eth"
    "github.com/ethereum/go-ethereum/les/flowcontrol"
    "github.com/ethereum/go-ethereum/light"
    "github.com/ethereum/go-ethereum/log"
    "github.com/ethereum/go-ethereum/p2p"
    "github.com/ethereum/go-ethereum/p2p/discv5"
    "github.com/ethereum/go-ethereum/p2p/enode"
    "github.com/ethereum/go-ethereum/params"
    "github.com/ethereum/go-ethereum/rpc"
)

type LesServer struct {
    lesCommons

    archiveMode bool // Flag whether the ethereum node runs in archive mode.
    handler     *serverHandler
    lesTopics   []discv5.Topic
    privateKey  *ecdsa.PrivateKey

    // Flow control and capacity management
    fcManager    *flowcontrol.ClientManager
    costTracker  *costTracker
    defParams    flowcontrol.ServerParams
    servingQueue *servingQueue
    clientPool   *clientPool

    freeCapacity uint64 // The minimal client capacity used for free client.
    threadsIdle  int    // Request serving threads count when system is idle.
    threadsBusy  int    // Request serving threads count when system is busy(block insertion).
}

func NewLesServer(e *eth.Ethereum, config *eth.Config) (*LesServer, error) {
    // Collect les protocol version information supported by local node.
    lesTopics := make([]discv5.Topic, len(AdvertiseProtocolVersions))
    for i, pv := range AdvertiseProtocolVersions {
        lesTopics[i] = lesTopic(e.BlockChain().Genesis().Hash(), pv)
    }
    // Calculate the number of threads used to service the light client
    // requests based on the user-specified value.
    threads := config.LightServ * 4 / 100
    if threads < 4 {
        threads = 4
    }
    srv := &LesServer{
        lesCommons: lesCommons{
            genesis:          e.BlockChain().Genesis().Hash(),
            config:           config,
            chainConfig:      e.BlockChain().Config(),
            iConfig:          light.DefaultServerIndexerConfig,
            chainDb:          e.ChainDb(),
            peers:            newPeerSet(),
            chainReader:      e.BlockChain(),
            chtIndexer:       light.NewChtIndexer(e.ChainDb(), nil, params.CHTFrequency, params.HelperTrieProcessConfirmations),
            bloomTrieIndexer: light.NewBloomTrieIndexer(e.ChainDb(), nil, params.BloomBitsBlocks, params.BloomTrieFrequency),
            closeCh:          make(chan struct{}),
        },
        archiveMode:  e.ArchiveMode(),
        lesTopics:    lesTopics,
        fcManager:    flowcontrol.NewClientManager(nil, &mclock.System{}),
        servingQueue: newServingQueue(int64(time.Millisecond*10), float64(config.LightServ)/100),
        threadsBusy:  config.LightServ/100 + 1,
        threadsIdle:  threads,
    }
    srv.handler = newServerHandler(srv, e.BlockChain(), e.ChainDb(), e.TxPool(), e.Synced)
    srv.costTracker, srv.freeCapacity = newCostTracker(e.ChainDb(), config)

    // Set up checkpoint oracle.
    oracle := config.CheckpointOracle
    if oracle == nil {
        oracle = params.CheckpointOracles[e.BlockChain().Genesis().Hash()]
    }
    srv.oracle = newCheckpointOracle(oracle, srv.localCheckpoint)

    // Initialize server capacity management fields.
    srv.defParams = flowcontrol.ServerParams{
        BufLimit:    srv.freeCapacity * bufLimitRatio,
        MinRecharge: srv.freeCapacity,
    }
    // LES flow control tries to more or less guarantee the possibility for the
    // clients to send a certain amount of requests at any time and get a quick
    // response. Most of the clients want this guarantee but don't actually need
    // to send requests most of the time. Our goal is to serve as many clients as
    // possible while the actually used server capacity does not exceed the limits
    totalRecharge := srv.costTracker.totalRecharge()
    maxCapacity := srv.freeCapacity * uint64(srv.config.LightPeers)
    if totalRecharge > maxCapacity {
        maxCapacity = totalRecharge
    }
    srv.fcManager.SetCapacityLimits(srv.freeCapacity, maxCapacity, srv.freeCapacity*2)
    srv.clientPool = newClientPool(srv.chainDb, srv.freeCapacity, 10000, mclock.System{}, func(id enode.ID) { go srv.peers.Unregister(peerIdToString(id)) })

    checkpoint := srv.latestLocalCheckpoint()
    if !checkpoint.Empty() {
        log.Info("Loaded latest checkpoint", "section", checkpoint.SectionIndex, "head", checkpoint.SectionHead,
            "chtroot", checkpoint.CHTRoot, "bloomroot", checkpoint.BloomRoot)
    }
    srv.chtIndexer.Start(e.BlockChain())
    return srv, nil
}

func (s *LesServer) APIs() []rpc.API {
    return []rpc.API{
        {
            Namespace: "les",
            Version:   "1.0",
            Service:   NewPrivateLightAPI(&s.lesCommons),
            Public:    false,
        },
    }
}

func (s *LesServer) Protocols() []p2p.Protocol {
    return s.makeProtocols(ServerProtocolVersions, s.handler.runPeer, func(id enode.ID) interface{} {
        if p := s.peers.Peer(peerIdToString(id)); p != nil {
            return p.Info()
        }
        return nil
    })
}

// Start starts the LES server
func (s *LesServer) Start(srvr *p2p.Server) {
    s.privateKey = srvr.PrivateKey
    s.handler.start()

    s.wg.Add(1)
    go s.capacityManagement()

    if srvr.DiscV5 != nil {
        for _, topic := range s.lesTopics {
            topic := topic
            go func() {
                logger := log.New("topic", topic)
                logger.Info("Starting topic registration")
                defer logger.Info("Terminated topic registration")

                srvr.DiscV5.RegisterTopic(topic, s.closeCh)
            }()
        }
    }
}

// Stop stops the LES service
func (s *LesServer) Stop() {
    close(s.closeCh)

    // Disconnect existing sessions.
    // This also closes the gate for any new registrations on the peer set.
    // sessions which are already established but not added to pm.peers yet
    // will exit when they try to register.
    s.peers.Close()

    s.fcManager.Stop()
    s.clientPool.stop()
    s.costTracker.stop()
    s.handler.stop()
    s.servingQueue.stop()

    // Note, bloom trie indexer is closed by parent bloombits indexer.
    s.chtIndexer.Close()
    s.wg.Wait()
    log.Info("Les server stopped")
}

func (s *LesServer) SetBloomBitsIndexer(bloomIndexer *core.ChainIndexer) {
    bloomIndexer.AddChildIndexer(s.bloomTrieIndexer)
}

// SetClient sets the rpc client and starts running checkpoint contract if it is not yet watched.
func (s *LesServer) SetContractBackend(backend bind.ContractBackend) {
    if s.oracle == nil {
        return
    }
    s.oracle.start(backend)
}

// capacityManagement starts an event handler loop that updates the recharge curve of
// the client manager and adjusts the client pool's size according to the total
// capacity updates coming from the client manager
func (s *LesServer) capacityManagement() {
    defer s.wg.Done()

    processCh := make(chan bool, 100)
    sub := s.handler.blockchain.SubscribeBlockProcessingEvent(processCh)
    defer sub.Unsubscribe()

    totalRechargeCh := make(chan uint64, 100)
    totalRecharge := s.costTracker.subscribeTotalRecharge(totalRechargeCh)

    totalCapacityCh := make(chan uint64, 100)
    totalCapacity := s.fcManager.SubscribeTotalCapacity(totalCapacityCh)
    s.clientPool.setLimits(s.config.LightPeers, totalCapacity)

    var (
        busy         bool
        freePeers    uint64
        blockProcess mclock.AbsTime
    )
    updateRecharge := func() {
        if busy {
            s.servingQueue.setThreads(s.threadsBusy)
            s.fcManager.SetRechargeCurve(flowcontrol.PieceWiseLinear{{0, 0}, {totalRecharge, totalRecharge}})
        } else {
            s.servingQueue.setThreads(s.threadsIdle)
            s.fcManager.SetRechargeCurve(flowcontrol.PieceWiseLinear{{0, 0}, {totalRecharge / 10, totalRecharge}, {totalRecharge, totalRecharge}})
        }
    }
    updateRecharge()

    for {
        select {
        case busy = <-processCh:
            if busy {
                blockProcess = mclock.Now()
            } else {
                blockProcessingTimer.Update(time.Duration(mclock.Now() - blockProcess))
            }
            updateRecharge()
        case totalRecharge = <-totalRechargeCh:
            totalRechargeGauge.Update(int64(totalRecharge))
            updateRecharge()
        case totalCapacity = <-totalCapacityCh:
            totalCapacityGauge.Update(int64(totalCapacity))
            newFreePeers := totalCapacity / s.freeCapacity
            if newFreePeers < freePeers && newFreePeers < uint64(s.config.LightPeers) {
                log.Warn("Reduced free peer connections", "from", freePeers, "to", newFreePeers)
            }
            freePeers = newFreePeers
            s.clientPool.setLimits(s.config.LightPeers, totalCapacity)
        case <-s.closeCh:
            return
        }
    }
}