123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252 |
- // 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 whisperv6
- import (
- "fmt"
- "math"
- "sync"
- "time"
- "github.com/ethereum/go-ethereum/common"
- "github.com/ethereum/go-ethereum/log"
- "github.com/ethereum/go-ethereum/p2p"
- "github.com/ethereum/go-ethereum/rlp"
- set "gopkg.in/fatih/set.v0"
- )
- // Peer represents a whisper protocol peer connection.
- type Peer struct {
- host *Whisper
- peer *p2p.Peer
- ws p2p.MsgReadWriter
- trusted bool
- powRequirement float64
- bloomMu sync.Mutex
- bloomFilter []byte
- fullNode bool
- known *set.Set // Messages already known by the peer to avoid wasting bandwidth
- quit chan struct{}
- }
- // newPeer creates a new whisper peer object, but does not run the handshake itself.
- func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) *Peer {
- return &Peer{
- host: host,
- peer: remote,
- ws: rw,
- trusted: false,
- powRequirement: 0.0,
- known: set.New(),
- quit: make(chan struct{}),
- bloomFilter: MakeFullNodeBloom(),
- fullNode: true,
- }
- }
- // start initiates the peer updater, periodically broadcasting the whisper packets
- // into the network.
- func (peer *Peer) start() {
- go peer.update()
- log.Trace("start", "peer", peer.ID())
- }
- // stop terminates the peer updater, stopping message forwarding to it.
- func (peer *Peer) stop() {
- close(peer.quit)
- log.Trace("stop", "peer", peer.ID())
- }
- // handshake sends the protocol initiation status message to the remote peer and
- // verifies the remote status too.
- func (peer *Peer) handshake() error {
- // Send the handshake status message asynchronously
- errc := make(chan error, 1)
- go func() {
- pow := peer.host.MinPow()
- powConverted := math.Float64bits(pow)
- bloom := peer.host.BloomFilter()
- errc <- p2p.SendItems(peer.ws, statusCode, ProtocolVersion, powConverted, bloom)
- }()
- // Fetch the remote status packet and verify protocol match
- packet, err := peer.ws.ReadMsg()
- if err != nil {
- return err
- }
- if packet.Code != statusCode {
- return fmt.Errorf("peer [%x] sent packet %x before status packet", peer.ID(), packet.Code)
- }
- s := rlp.NewStream(packet.Payload, uint64(packet.Size))
- _, err = s.List()
- if err != nil {
- return fmt.Errorf("peer [%x] sent bad status message: %v", peer.ID(), err)
- }
- peerVersion, err := s.Uint()
- if err != nil {
- return fmt.Errorf("peer [%x] sent bad status message (unable to decode version): %v", peer.ID(), err)
- }
- if peerVersion != ProtocolVersion {
- return fmt.Errorf("peer [%x]: protocol version mismatch %d != %d", peer.ID(), peerVersion, ProtocolVersion)
- }
- // only version is mandatory, subsequent parameters are optional
- powRaw, err := s.Uint()
- if err == nil {
- pow := math.Float64frombits(powRaw)
- if math.IsInf(pow, 0) || math.IsNaN(pow) || pow < 0.0 {
- return fmt.Errorf("peer [%x] sent bad status message: invalid pow", peer.ID())
- }
- peer.powRequirement = pow
- var bloom []byte
- err = s.Decode(&bloom)
- if err == nil {
- sz := len(bloom)
- if sz != BloomFilterSize && sz != 0 {
- return fmt.Errorf("peer [%x] sent bad status message: wrong bloom filter size %d", peer.ID(), sz)
- }
- peer.setBloomFilter(bloom)
- }
- }
- if err := <-errc; err != nil {
- return fmt.Errorf("peer [%x] failed to send status packet: %v", peer.ID(), err)
- }
- return nil
- }
- // update executes periodic operations on the peer, including message transmission
- // and expiration.
- func (peer *Peer) update() {
- // Start the tickers for the updates
- expire := time.NewTicker(expirationCycle)
- transmit := time.NewTicker(transmissionCycle)
- // Loop and transmit until termination is requested
- for {
- select {
- case <-expire.C:
- peer.expire()
- case <-transmit.C:
- if err := peer.broadcast(); err != nil {
- log.Trace("broadcast failed", "reason", err, "peer", peer.ID())
- return
- }
- case <-peer.quit:
- return
- }
- }
- }
- // mark marks an envelope known to the peer so that it won't be sent back.
- func (peer *Peer) mark(envelope *Envelope) {
- peer.known.Add(envelope.Hash())
- }
- // marked checks if an envelope is already known to the remote peer.
- func (peer *Peer) marked(envelope *Envelope) bool {
- return peer.known.Has(envelope.Hash())
- }
- // expire iterates over all the known envelopes in the host and removes all
- // expired (unknown) ones from the known list.
- func (peer *Peer) expire() {
- unmark := make(map[common.Hash]struct{})
- peer.known.Each(func(v interface{}) bool {
- if !peer.host.isEnvelopeCached(v.(common.Hash)) {
- unmark[v.(common.Hash)] = struct{}{}
- }
- return true
- })
- // Dump all known but no longer cached
- for hash := range unmark {
- peer.known.Remove(hash)
- }
- }
- // broadcast iterates over the collection of envelopes and transmits yet unknown
- // ones over the network.
- func (peer *Peer) broadcast() error {
- envelopes := peer.host.Envelopes()
- bundle := make([]*Envelope, 0, len(envelopes))
- for _, envelope := range envelopes {
- if !peer.marked(envelope) && envelope.PoW() >= peer.powRequirement && peer.bloomMatch(envelope) {
- bundle = append(bundle, envelope)
- }
- }
- if len(bundle) > 0 {
- // transmit the batch of envelopes
- if err := p2p.Send(peer.ws, messagesCode, bundle); err != nil {
- return err
- }
- // mark envelopes only if they were successfully sent
- for _, e := range bundle {
- peer.mark(e)
- }
- log.Trace("broadcast", "num. messages", len(bundle))
- }
- return nil
- }
- // ID returns a peer's id
- func (peer *Peer) ID() []byte {
- id := peer.peer.ID()
- return id[:]
- }
- func (peer *Peer) notifyAboutPowRequirementChange(pow float64) error {
- i := math.Float64bits(pow)
- return p2p.Send(peer.ws, powRequirementCode, i)
- }
- func (peer *Peer) notifyAboutBloomFilterChange(bloom []byte) error {
- return p2p.Send(peer.ws, bloomFilterExCode, bloom)
- }
- func (peer *Peer) bloomMatch(env *Envelope) bool {
- peer.bloomMu.Lock()
- defer peer.bloomMu.Unlock()
- return peer.fullNode || BloomFilterMatch(peer.bloomFilter, env.Bloom())
- }
- func (peer *Peer) setBloomFilter(bloom []byte) {
- peer.bloomMu.Lock()
- defer peer.bloomMu.Unlock()
- peer.bloomFilter = bloom
- peer.fullNode = isFullNode(bloom)
- if peer.fullNode && peer.bloomFilter == nil {
- peer.bloomFilter = MakeFullNodeBloom()
- }
- }
- func MakeFullNodeBloom() []byte {
- bloom := make([]byte, BloomFilterSize)
- for i := 0; i < BloomFilterSize; i++ {
- bloom[i] = 0xFF
- }
- return bloom
- }
|