From 9f1007e554e223b12354d3c91ae7fb040cf8b865 Mon Sep 17 00:00:00 2001
From: gluk256 <gluk256@users.noreply.github.com>
Date: Thu, 21 Dec 2017 11:31:44 +0200
Subject: [PATCH] whisper/whisperv6: message bundling (#15666)
Changed the communication protocol for ordinary message,
according to EIP 627. Messages will be send in bundles, i.e.
array of messages will be sent instead of single message.
---
whisper/whisperv6/peer.go | 25 +++++++++++++++----------
whisper/whisperv6/whisper.go | 28 ++++++++++++++++++----------
2 files changed, 33 insertions(+), 20 deletions(-)
diff --git a/whisper/whisperv6/peer.go b/whisper/whisperv6/peer.go
index ac7b3b12b..ffc39505e 100644
--- a/whisper/whisperv6/peer.go
+++ b/whisper/whisperv6/peer.go
@@ -149,21 +149,26 @@ func (peer *Peer) expire() {
// broadcast iterates over the collection of envelopes and transmits yet unknown
// ones over the network.
func (p *Peer) broadcast() error {
- var cnt int
envelopes := p.host.Envelopes()
+ bundle := make([]*Envelope, 0, len(envelopes))
for _, envelope := range envelopes {
if !p.marked(envelope) {
- err := p2p.Send(p.ws, messagesCode, envelope)
- if err != nil {
- return err
- } else {
- p.mark(envelope)
- cnt++
- }
+ bundle = append(bundle, envelope)
}
}
- if cnt > 0 {
- log.Trace("broadcast", "num. messages", cnt)
+
+ if len(bundle) > 0 {
+ // transmit the batch of envelopes
+ if err := p2p.Send(p.ws, messagesCode, bundle); err != nil {
+ return err
+ }
+
+ // mark envelopes only if they were successfully sent
+ for _, e := range bundle {
+ p.mark(e)
+ }
+
+ log.Trace("broadcast", "num. messages", len(bundle))
}
return nil
}
diff --git a/whisper/whisperv6/whisper.go b/whisper/whisperv6/whisper.go
index d09baab3f..2cc1e64f5 100644
--- a/whisper/whisperv6/whisper.go
+++ b/whisper/whisperv6/whisper.go
@@ -515,18 +515,26 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
log.Warn("unxepected status message received", "peer", p.peer.ID())
case messagesCode:
// decode the contained envelopes
- var envelope Envelope
- if err := packet.Decode(&envelope); err != nil {
- log.Warn("failed to decode envelope, peer will be disconnected", "peer", p.peer.ID(), "err", err)
- return errors.New("invalid envelope")
+ var envelopes []*Envelope
+ if err := packet.Decode(&envelopes); err != nil {
+ log.Warn("failed to decode envelopes, peer will be disconnected", "peer", p.peer.ID(), "err", err)
+ return errors.New("invalid envelopes")
}
- cached, err := wh.add(&envelope)
- if err != nil {
- log.Warn("bad envelope received, peer will be disconnected", "peer", p.peer.ID(), "err", err)
- return errors.New("invalid envelope")
+
+ trouble := false
+ for _, env := range envelopes {
+ cached, err := wh.add(env)
+ if err != nil {
+ trouble = true
+ log.Error("bad envelope received, peer will be disconnected", "peer", p.peer.ID(), "err", err)
+ }
+ if cached {
+ p.mark(env)
+ }
}
- if cached {
- p.mark(&envelope)
+
+ if trouble {
+ return errors.New("invalid envelope")
}
case p2pCode:
// peer-to-peer message, sent directly to peer bypassing PoW checks, etc.
--
GitLab