// 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 . package les import ( "context" "sync" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/rlp" ) type ltrInfo struct { tx *types.Transaction sentTo map[*peer]struct{} } type lesTxRelay struct { txSent map[common.Hash]*ltrInfo txPending map[common.Hash]struct{} ps *peerSet peerList []*peer peerStartPos int lock sync.RWMutex stop chan struct{} retriever *retrieveManager } func newLesTxRelay(ps *peerSet, retriever *retrieveManager) *lesTxRelay { r := &lesTxRelay{ txSent: make(map[common.Hash]*ltrInfo), txPending: make(map[common.Hash]struct{}), ps: ps, retriever: retriever, stop: make(chan struct{}), } ps.notify(r) return r } func (self *lesTxRelay) Stop() { close(self.stop) } func (self *lesTxRelay) registerPeer(p *peer) { self.lock.Lock() defer self.lock.Unlock() self.peerList = self.ps.AllPeers() } func (self *lesTxRelay) unregisterPeer(p *peer) { self.lock.Lock() defer self.lock.Unlock() self.peerList = self.ps.AllPeers() } // send sends a list of transactions to at most a given number of peers at // once, never resending any particular transaction to the same peer twice func (self *lesTxRelay) send(txs types.Transactions, count int) { sendTo := make(map[*peer]types.Transactions) self.peerStartPos++ // rotate the starting position of the peer list if self.peerStartPos >= len(self.peerList) { self.peerStartPos = 0 } for _, tx := range txs { hash := tx.Hash() ltr, ok := self.txSent[hash] if !ok { ltr = <rInfo{ tx: tx, sentTo: make(map[*peer]struct{}), } self.txSent[hash] = ltr self.txPending[hash] = struct{}{} } if len(self.peerList) > 0 { cnt := count pos := self.peerStartPos for { peer := self.peerList[pos] if _, ok := ltr.sentTo[peer]; !ok { sendTo[peer] = append(sendTo[peer], tx) ltr.sentTo[peer] = struct{}{} cnt-- } if cnt == 0 { break // sent it to the desired number of peers } pos++ if pos == len(self.peerList) { pos = 0 } if pos == self.peerStartPos { break // tried all available peers } } } } for p, list := range sendTo { pp := p ll := list enc, _ := rlp.EncodeToBytes(ll) reqID := genReqID() rq := &distReq{ getCost: func(dp distPeer) uint64 { peer := dp.(*peer) return peer.GetTxRelayCost(len(ll), len(enc)) }, canSend: func(dp distPeer) bool { return !dp.(*peer).isOnlyAnnounce && dp.(*peer) == pp }, request: func(dp distPeer) func() { peer := dp.(*peer) cost := peer.GetTxRelayCost(len(ll), len(enc)) peer.fcServer.QueuedRequest(reqID, cost) return func() { peer.SendTxs(reqID, cost, enc) } }, } go self.retriever.retrieve(context.Background(), reqID, rq, func(p distPeer, msg *Msg) error { return nil }, self.stop) } } func (self *lesTxRelay) Send(txs types.Transactions) { self.lock.Lock() defer self.lock.Unlock() self.send(txs, 3) } func (self *lesTxRelay) NewHead(head common.Hash, mined []common.Hash, rollback []common.Hash) { self.lock.Lock() defer self.lock.Unlock() for _, hash := range mined { delete(self.txPending, hash) } for _, hash := range rollback { self.txPending[hash] = struct{}{} } if len(self.txPending) > 0 { txs := make(types.Transactions, len(self.txPending)) i := 0 for hash := range self.txPending { txs[i] = self.txSent[hash].tx i++ } self.send(txs, 1) } } func (self *lesTxRelay) Discard(hashes []common.Hash) { self.lock.Lock() defer self.lock.Unlock() for _, hash := range hashes { delete(self.txSent, hash) delete(self.txPending, hash) } }