wshiper: broadcast received hashes to peers

Signed-off-by: Dmitry Shulyak <yashulyak@gmail.com>

whisper/whisperv5/doc.go

@@ -44,6 +44,7 @@ const (
 	messagesCode         = 1 // normal whisper message
 	p2pCode              = 2 // peer-to-peer message (to be consumed by the peer, but not forwarded any further)
 	p2pRequestCode       = 3 // peer-to-peer message, used by Dapp protocol
+	hashesCode           = 4 // code used to populate known hashes of the peer
 	NumberOfMessageCodes = 64
 
 	paddingMask   = byte(3)

whisper/whisperv5/peer.go

@@ -34,20 +34,23 @@ type Peer struct {
 	ws      p2p.MsgReadWriter
 	trusted bool
 
-	known *set.Set // Messages already known by the peer to avoid wasting bandwidth
+	known      *set.Set // Messages already known by the peer to avoid wasting bandwidth
-
+	advertised map[common.Hash]struct{}
-	quit chan struct{}
+	hashes     chan common.Hash
+	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,
+		host:       host,
-		peer:    remote,
+		peer:       remote,
-		ws:      rw,
+		ws:         rw,
-		trusted: false,
+		trusted:    false,
-		known:   set.New(),
+		known:      set.New(),
-		quit:    make(chan struct{}),
+		advertised: make(map[common.Hash]struct{}),
+		hashes:     make(chan common.Hash, 20),
+		quit:       make(chan struct{}),
 	}
 }
 
@@ -101,7 +104,8 @@ func (p *Peer) update() {
 	// Start the tickers for the updates
 	expire := time.NewTicker(expirationCycle)
 	transmit := time.NewTicker(transmissionCycle)
-
+	hashesTransmit := time.NewTicker(100 * time.Millisecond)
+	hashes := make([]common.Hash, 0, 10)
 	// Loop and transmit until termination is requested
 	for {
 		select {
@@ -113,7 +117,27 @@ func (p *Peer) update() {
 				log.Trace("broadcast failed", "reason", err, "peer", p.ID())
 				return
 			}
-
+		case <-hashesTransmit.C:
+			if err := p.broadcastHashes(hashes); err != nil {
+				log.Trace("broadcast of hashes failed", err, "peer", p.ID)
+				return
+			}
+			hashes = hashes[:0]
+		case hash := <-p.hashes:
+			if _, ok := p.advertised[hash]; ok {
+				continue
+			}
+			if p.known.Has(hash) {
+				continue
+			}
+			hashes = append(hashes, hash)
+			if len(hashes) == cap(hashes) {
+				if err := p.broadcastHashes(hashes); err != nil {
+					log.Trace("broadcast of hashes failed", err, "peer", p.ID)
+					return
+				}
+				hashes = hashes[:0]
+			}
 		case <-p.quit:
 			return
 		}
@@ -146,6 +170,10 @@ func (peer *Peer) expire() {
 	}
 }
 
+func (p *Peer) addHash(hash common.Hash) {
+	p.hashes <- hash
+}
+
 // broadcast iterates over the collection of envelopes and transmits yet unknown
 // ones over the network.
 func (p *Peer) broadcast() error {
@@ -168,6 +196,20 @@ func (p *Peer) broadcast() error {
 	return nil
 }
 
+func (p *Peer) broadcastHashes(hashes []common.Hash) error {
+	if len(hashes) == 0 {
+		return nil
+	}
+	log.Trace("broadcast", "hashes", hashes)
+	if err := p2p.Send(p.ws, hashesCode, hashes); err != nil {
+		return err
+	}
+	for _, hash := range hashes {
+		p.advertised[hash] = struct{}{}
+	}
+	return nil
+}
+
 func (p *Peer) ID() []byte {
 	id := p.peer.ID()
 	return id[:]

whisper/whisperv5/whisper.go

@@ -511,6 +511,15 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
 		case statusCode:
 			// this should not happen, but no need to panic; just ignore this message.
 			log.Warn("unxepected status message received", "peer", p.peer.ID())
+		case hashesCode:
+			var hashes []common.Hash
+			err := packet.Decode(&hashes)
+			if err != nil {
+				return errors.New("invalid haashes")
+			}
+			for _, hash := range hashes {
+				p.known.Add(hash)
+			}
 		case messagesCode:
 			// decode the contained envelopes
 			var envelope Envelope
@@ -617,6 +626,13 @@ func (wh *Whisper) add(envelope *Envelope) (bool, error) {
 		}
 	}
 	wh.poolMu.Unlock()
+	wh.peerMu.RLock()
+	if !alreadyCached {
+		for p := range wh.peers {
+			p.addHash(hash)
+		}
+	}
+	wh.peerMu.RUnlock()
 
 	if alreadyCached {
 		log.Trace("whisper envelope already cached", "hash", envelope.Hash().Hex())