diff options
| -rw-r--r-- | dex/handler.go | 17 | ||||
| -rw-r--r-- | dex/peer.go | 32 |
2 files changed, 42 insertions, 7 deletions
diff --git a/dex/handler.go b/dex/handler.go index 21322e4e0..3c8d25ea3 100644 --- a/dex/handler.go +++ b/dex/handler.go @@ -1037,10 +1037,25 @@ func (pm *ProtocolManager) BroadcastPullVotes( } func (pm *ProtocolManager) txBroadcastLoop() { + queueSizeMax := common.StorageSize(100 * 1024) // 100 KB + currentSize := common.StorageSize(0) + txs := make(types.Transactions, 0) for { select { + case <-time.After(500 * time.Millisecond): + pm.BroadcastTxs(txs) + txs = txs[:0] + currentSize = 0 case event := <-pm.txsCh: - pm.BroadcastTxs(event.Txs) + txs = append(txs, event.Txs...) + for _, tx := range event.Txs { + currentSize += tx.Size() + } + if currentSize >= queueSizeMax { + pm.BroadcastTxs(txs) + txs = txs[:0] + currentSize = 0 + } // Err() channel will be closed when unsubscribing. case <-pm.txsSub.Err(): diff --git a/dex/peer.go b/dex/peer.go index 5aa1c1b80..263dc5647 100644 --- a/dex/peer.go +++ b/dex/peer.go @@ -1,3 +1,20 @@ +// Copyright 2018 The dexon-consensus Authors +// This file is part of the dexon-consensus library. +// +// The dexon-consensus 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 dexon-consensus 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 dexon-consensus library. If not, see +// <http://www.gnu.org/licenses/>. + // Copyright 2015 The go-ethereum Authors // This file is part of the go-ethereum library. // @@ -188,12 +205,6 @@ func newPeer(version int, p *p2p.Peer, rw p2p.MsgReadWriter) *peer { func (p *peer) broadcast() { for { select { - case txs := <-p.queuedTxs: - if err := p.SendTransactions(txs); err != nil { - return - } - p.Log().Trace("Broadcast transactions", "count", len(txs)) - case metas := <-p.queuedMetas: if err := p.SendNodeMetas(metas); err != nil { return @@ -253,6 +264,15 @@ func (p *peer) broadcast() { p.Log().Trace("Pulling Votes", "position", pos) case <-p.term: return + case <-time.After(100 * time.Millisecond): + } + select { + case txs := <-p.queuedTxs: + if err := p.SendTransactions(txs); err != nil { + return + } + p.Log().Trace("Broadcast transactions", "count", len(txs)) + default: } } } |