path: root/les/benchmark.go

// Copyright 2018 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 (
    "encoding/binary"
    "fmt"
    "math/big"
    "math/rand"
    "time"

    "github.com/ethereum/go-ethereum/common"
    "github.com/ethereum/go-ethereum/common/mclock"
    "github.com/ethereum/go-ethereum/core/rawdb"
    "github.com/ethereum/go-ethereum/core/types"
    "github.com/ethereum/go-ethereum/crypto"
    "github.com/ethereum/go-ethereum/les/flowcontrol"
    "github.com/ethereum/go-ethereum/log"
    "github.com/ethereum/go-ethereum/p2p"
    "github.com/ethereum/go-ethereum/p2p/enode"
    "github.com/ethereum/go-ethereum/params"
    "github.com/ethereum/go-ethereum/rlp"
)

// requestBenchmark is an interface for different randomized request generators
type requestBenchmark interface {
    // init initializes the generator for generating the given number of randomized requests
    init(pm *ProtocolManager, count int) error
    // request initiates sending a single request to the given peer
    request(peer *peer, index int) error
}

// benchmarkBlockHeaders implements requestBenchmark
type benchmarkBlockHeaders struct {
    amount, skip    int
    reverse, byHash bool
    offset, randMax int64
    hashes          []common.Hash
}

func (b *benchmarkBlockHeaders) init(pm *ProtocolManager, count int) error {
    d := int64(b.amount-1) * int64(b.skip+1)
    b.offset = 0
    b.randMax = pm.blockchain.CurrentHeader().Number.Int64() + 1 - d
    if b.randMax < 0 {
        return fmt.Errorf("chain is too short")
    }
    if b.reverse {
        b.offset = d
    }
    if b.byHash {
        b.hashes = make([]common.Hash, count)
        for i := range b.hashes {
            b.hashes[i] = rawdb.ReadCanonicalHash(pm.chainDb, uint64(b.offset+rand.Int63n(b.randMax)))
        }
    }
    return nil
}

func (b *benchmarkBlockHeaders) request(peer *peer, index int) error {
    if b.byHash {
        return peer.RequestHeadersByHash(0, 0, b.hashes[index], b.amount, b.skip, b.reverse)
    } else {
        return peer.RequestHeadersByNumber(0, 0, uint64(b.offset+rand.Int63n(b.randMax)), b.amount, b.skip, b.reverse)
    }
}

// benchmarkBodiesOrReceipts implements requestBenchmark
type benchmarkBodiesOrReceipts struct {
    receipts bool
    hashes   []common.Hash
}

func (b *benchmarkBodiesOrReceipts) init(pm *ProtocolManager, count int) error {
    randMax := pm.blockchain.CurrentHeader().Number.Int64() + 1
    b.hashes = make([]common.Hash, count)
    for i := range b.hashes {
        b.hashes[i] = rawdb.ReadCanonicalHash(pm.chainDb, uint64(rand.Int63n(randMax)))
    }
    return nil
}

func (b *benchmarkBodiesOrReceipts) request(peer *peer, index int) error {
    if b.receipts {
        return peer.RequestReceipts(0, 0, []common.Hash{b.hashes[index]})
    } else {
        return peer.RequestBodies(0, 0, []common.Hash{b.hashes[index]})
    }
}

// benchmarkProofsOrCode implements requestBenchmark
type benchmarkProofsOrCode struct {
    code     bool
    headHash common.Hash
}

func (b *benchmarkProofsOrCode) init(pm *ProtocolManager, count int) error {
    b.headHash = pm.blockchain.CurrentHeader().Hash()
    return nil
}

func (b *benchmarkProofsOrCode) request(peer *peer, index int) error {
    key := make([]byte, 32)
    rand.Read(key)
    if b.code {
        return peer.RequestCode(0, 0, []CodeReq{{BHash: b.headHash, AccKey: key}})
    } else {
        return peer.RequestProofs(0, 0, []ProofReq{{BHash: b.headHash, Key: key}})
    }
}

// benchmarkHelperTrie implements requestBenchmark
type benchmarkHelperTrie struct {
    bloom                 bool
    reqCount              int
    sectionCount, headNum uint64
}

func (b *benchmarkHelperTrie) init(pm *ProtocolManager, count int) error {
    if b.bloom {
        b.sectionCount, b.headNum, _ = pm.server.bloomTrieIndexer.Sections()
    } else {
        b.sectionCount, _, _ = pm.server.chtIndexer.Sections()
        b.headNum = b.sectionCount*params.CHTFrequency - 1
    }
    if b.sectionCount == 0 {
        return fmt.Errorf("no processed sections available")
    }
    return nil
}

func (b *benchmarkHelperTrie) request(peer *peer, index int) error {
    reqs := make([]HelperTrieReq, b.reqCount)

    if b.bloom {
        bitIdx := uint16(rand.Intn(2048))
        for i := range reqs {
            key := make([]byte, 10)
            binary.BigEndian.PutUint16(key[:2], bitIdx)
            binary.BigEndian.PutUint64(key[2:], uint64(rand.Int63n(int64(b.sectionCount))))
            reqs[i] = HelperTrieReq{Type: htBloomBits, TrieIdx: b.sectionCount - 1, Key: key}
        }
    } else {
        for i := range reqs {
            key := make([]byte, 8)
            binary.BigEndian.PutUint64(key[:], uint64(rand.Int63n(int64(b.headNum))))
            reqs[i] = HelperTrieReq{Type: htCanonical, TrieIdx: b.sectionCount - 1, Key: key, AuxReq: auxHeader}
        }
    }

    return peer.RequestHelperTrieProofs(0, 0, reqs)
}

// benchmarkTxSend implements requestBenchmark
type benchmarkTxSend struct {
    txs types.Transactions
}

func (b *benchmarkTxSend) init(pm *ProtocolManager, count int) error {
    key, _ := crypto.GenerateKey()
    addr := crypto.PubkeyToAddress(key.PublicKey)
    signer := types.NewEIP155Signer(big.NewInt(18))
    b.txs = make(types.Transactions, count)

    for i := range b.txs {
        data := make([]byte, txSizeCostLimit)
        rand.Read(data)
        tx, err := types.SignTx(types.NewTransaction(0, addr, new(big.Int), 0, new(big.Int), data), signer, key)
        if err != nil {
            panic(err)
        }
        b.txs[i] = tx
    }
    return nil
}

func (b *benchmarkTxSend) request(peer *peer, index int) error {
    enc, _ := rlp.EncodeToBytes(types.Transactions{b.txs[index]})
    return peer.SendTxs(0, 0, enc)
}

// benchmarkTxStatus implements requestBenchmark
type benchmarkTxStatus struct{}

func (b *benchmarkTxStatus) init(pm *ProtocolManager, count int) error {
    return nil
}

func (b *benchmarkTxStatus) request(peer *peer, index int) error {
    var hash common.Hash
    rand.Read(hash[:])
    return peer.RequestTxStatus(0, 0, []common.Hash{hash})
}

// benchmarkSetup stores measurement data for a single benchmark type
type benchmarkSetup struct {
    req                   requestBenchmark
    totalCount            int
    totalTime, avgTime    time.Duration
    maxInSize, maxOutSize uint32
    err                   error
}

// runBenchmark runs a benchmark cycle for all benchmark types in the specified
// number of passes
func (pm *ProtocolManager) runBenchmark(benchmarks []requestBenchmark, passCount int, targetTime time.Duration) []*benchmarkSetup {
    setup := make([]*benchmarkSetup, len(benchmarks))
    for i, b := range benchmarks {
        setup[i] = &benchmarkSetup{req: b}
    }
    for i := 0; i < passCount; i++ {
        log.Info("Running benchmark", "pass", i+1, "total", passCount)
        todo := make([]*benchmarkSetup, len(benchmarks))
        copy(todo, setup)
        for len(todo) > 0 {
            // select a random element
            index := rand.Intn(len(todo))
            next := todo[index]
            todo[index] = todo[len(todo)-1]
            todo = todo[:len(todo)-1]

            if next.err == nil {
                // calculate request count
                count := 50
                if next.totalTime > 0 {
                    count = int(uint64(next.totalCount) * uint64(targetTime) / uint64(next.totalTime))
                }
                if err := pm.measure(next, count); err != nil {
                    next.err = err
                }
            }
        }
    }
    log.Info("Benchmark completed")

    for _, s := range setup {
        if s.err == nil {
            s.avgTime = s.totalTime / time.Duration(s.totalCount)
        }
    }
    return setup
}

// meteredPipe implements p2p.MsgReadWriter and remembers the largest single
// message size sent through the pipe
type meteredPipe struct {
    rw      p2p.MsgReadWriter
    maxSize uint32
}

func (m *meteredPipe) ReadMsg() (p2p.Msg, error) {
    return m.rw.ReadMsg()
}

func (m *meteredPipe) WriteMsg(msg p2p.Msg) error {
    if msg.Size > m.maxSize {
        m.maxSize = msg.Size
    }
    return m.rw.WriteMsg(msg)
}

// measure runs a benchmark for a single type in a single pass, with the given
// number of requests
func (pm *ProtocolManager) measure(setup *benchmarkSetup, count int) error {
    clientPipe, serverPipe := p2p.MsgPipe()
    clientMeteredPipe := &meteredPipe{rw: clientPipe}
    serverMeteredPipe := &meteredPipe{rw: serverPipe}
    var id enode.ID
    rand.Read(id[:])
    clientPeer := pm.newPeer(lpv2, NetworkId, p2p.NewPeer(id, "client", nil), clientMeteredPipe)
    serverPeer := pm.newPeer(lpv2, NetworkId, p2p.NewPeer(id, "server", nil), serverMeteredPipe)
    serverPeer.sendQueue = newExecQueue(count)
    serverPeer.announceType = announceTypeNone
    serverPeer.fcCosts = make(requestCostTable)
    c := &requestCosts{}
    for code := range requests {
        serverPeer.fcCosts[code] = c
    }
    serverPeer.fcParams = flowcontrol.ServerParams{BufLimit: 1, MinRecharge: 1}
    serverPeer.fcClient = flowcontrol.NewClientNode(pm.server.fcManager, serverPeer.fcParams)
    defer serverPeer.fcClient.Disconnect()

    if err := setup.req.init(pm, count); err != nil {
        return err
    }

    errCh := make(chan error, 10)
    start := mclock.Now()

    go func() {
        for i := 0; i < count; i++ {
            if err := setup.req.request(clientPeer, i); err != nil {
                errCh <- err
                return
            }
        }
    }()
    go func() {
        for i := 0; i < count; i++ {
            if err := pm.handleMsg(serverPeer); err != nil {
                errCh <- err
                return
            }
        }
    }()
    go func() {
        for i := 0; i < count; i++ {
            msg, err := clientPipe.ReadMsg()
            if err != nil {
                errCh <- err
                return
            }
            var i interface{}
            msg.Decode(&i)
        }
        // at this point we can be sure that the other two
        // goroutines finished successfully too
        close(errCh)
    }()
    select {
    case err := <-errCh:
        if err != nil {
            return err
        }
    case <-pm.quitSync:
        clientPipe.Close()
        serverPipe.Close()
        return fmt.Errorf("Benchmark cancelled")
    }

    setup.totalTime += time.Duration(mclock.Now() - start)
    setup.totalCount += count
    setup.maxInSize = clientMeteredPipe.maxSize
    setup.maxOutSize = serverMeteredPipe.maxSize
    clientPipe.Close()
    serverPipe.Close()
    return nil
}