dex: Tx message optimization (#39)

* dex: Add a tx queue in broadcast

* Modify queue parameter

* Priority select all messages except tx

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:
 		}
 	}
 }