whisper: expiry refactoring

whisper/whisperv5/api.go

@@ -65,6 +65,14 @@ func (api *PublicWhisperAPI) Version() (hexutil.Uint, error) {
 	return hexutil.Uint(api.whisper.Version()), nil
 }
 
+// Stats returns the Whisper statistics for diagnostics.
+func (api *PublicWhisperAPI) Stats() (string, error) {
+	if api.whisper == nil {
+		return "", whisperOffLineErr
+	}
+	return api.whisper.Stats(), nil
+}
+
 // MarkPeerTrusted marks specific peer trusted, which will allow it
 // to send historic (expired) messages.
 func (api *PublicWhisperAPI) MarkPeerTrusted(peerID hexutil.Bytes) error {

whisper/whisperv5/peer.go

@@ -134,20 +134,14 @@ func (peer *Peer) marked(envelope *Envelope) bool {
 // expire iterates over all the known envelopes in the host and removes all
 // expired (unknown) ones from the known list.
 func (peer *Peer) expire() {
-	// Assemble the list of available envelopes
-	available := set.NewNonTS()
-	for _, envelope := range peer.host.Envelopes() {
-		available.Add(envelope.Hash())
-	}
-	// Cross reference availability with known status
 	unmark := make(map[common.Hash]struct{})
 	peer.known.Each(func(v interface{}) bool {
-		if !available.Has(v.(common.Hash)) {
+		if !peer.host.isEnvelopeCached(v.(common.Hash)) {
 			unmark[v.(common.Hash)] = struct{}{}
 		}
 		return true
 	})
-	// Dump all known but unavailable
+	// Dump all known but no longer cached
 	for hash := range unmark {
 		peer.known.Remove(hash)
 	}

whisper/whisperv5/whisper.go

@@ -36,6 +36,11 @@ import (
 	set "gopkg.in/fatih/set.v0"
 )
 
+type Statistics struct {
+	messages int
+	memory   int
+}
+
 // Whisper represents a dark communication interface through the Ethereum
 // network, using its very own P2P communication layer.
 type Whisper struct {
@@ -60,6 +65,8 @@ type Whisper struct {
 	p2pMsgQueue  chan *Envelope
 	quit         chan struct{}
 
+	stats Statistics
+
 	overflow bool
 	test     bool
 }
@@ -288,7 +295,8 @@ func (w *Whisper) Unwatch(id string) {
 // Send injects a message into the whisper send queue, to be distributed in the
 // network in the coming cycles.
 func (w *Whisper) Send(envelope *Envelope) error {
-	return w.add(envelope)
+	_, err := w.add(envelope)
+	return err
 }
 
 // Start implements node.Service, starting the background data propagation thread
@@ -361,12 +369,15 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
 			}
 			// inject all envelopes into the internal pool
 			for _, envelope := range envelopes {
-				if err := wh.add(envelope); err != nil {
+				cached, err := wh.add(envelope)
+				if err != nil {
 					glog.V(logger.Warn).Infof("%v: bad envelope received: [%v], peer will be disconnected", p.peer, err)
 					return fmt.Errorf("invalid envelope")
 				}
+				if cached {
 					p.mark(envelope)
 				}
+			}
 		case p2pCode:
 			// peer-to-peer message, sent directly to peer bypassing PoW checks, etc.
 			// this message is not supposed to be forwarded to other peers, and
@@ -402,13 +413,13 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
 // add inserts a new envelope into the message pool to be distributed within the
 // whisper network. It also inserts the envelope into the expiration pool at the
 // appropriate time-stamp. In case of error, connection should be dropped.
-func (wh *Whisper) add(envelope *Envelope) error {
+func (wh *Whisper) add(envelope *Envelope) (bool, error) {
 	now := uint32(time.Now().Unix())
 	sent := envelope.Expiry - envelope.TTL
 
 	if sent > now {
 		if sent-SynchAllowance > now {
-			return fmt.Errorf("envelope created in the future [%x]", envelope.Hash())
+			return false, fmt.Errorf("envelope created in the future [%x]", envelope.Hash())
 		} else {
 			// recalculate PoW, adjusted for the time difference, plus one second for latency
 			envelope.calculatePoW(sent - now + 1)
@@ -417,34 +428,34 @@ func (wh *Whisper) add(envelope *Envelope) error {
 
 	if envelope.Expiry < now {
 		if envelope.Expiry+SynchAllowance*2 < now {
-			return fmt.Errorf("very old message")
+			return false, fmt.Errorf("very old message")
 		} else {
 			glog.V(logger.Debug).Infof("expired envelope dropped [%x]", envelope.Hash())
-			return nil // drop envelope without error
+			return false, nil // drop envelope without error
 		}
 	}
 
 	if len(envelope.Data) > MaxMessageLength {
-		return fmt.Errorf("huge messages are not allowed [%x]", envelope.Hash())
+		return false, fmt.Errorf("huge messages are not allowed [%x]", envelope.Hash())
 	}
 
 	if len(envelope.Version) > 4 {
-		return fmt.Errorf("oversized version [%x]", envelope.Hash())
+		return false, fmt.Errorf("oversized version [%x]", envelope.Hash())
 	}
 
 	if len(envelope.AESNonce) > AESNonceMaxLength {
 		// the standard AES GSM nonce size is 12,
 		// but const gcmStandardNonceSize cannot be accessed directly
-		return fmt.Errorf("oversized AESNonce [%x]", envelope.Hash())
+		return false, fmt.Errorf("oversized AESNonce [%x]", envelope.Hash())
 	}
 
 	if len(envelope.Salt) > saltLength {
-		return fmt.Errorf("oversized salt [%x]", envelope.Hash())
+		return false, fmt.Errorf("oversized salt [%x]", envelope.Hash())
 	}
 
 	if envelope.PoW() < MinimumPoW && !wh.test {
 		glog.V(logger.Debug).Infof("envelope with low PoW dropped: %f [%x]", envelope.PoW(), envelope.Hash())
-		return nil // drop envelope without error
+		return false, nil // drop envelope without error
 	}
 
 	hash := envelope.Hash()
@@ -471,7 +482,7 @@ func (wh *Whisper) add(envelope *Envelope) error {
 			wh.mailServer.Archive(envelope)
 		}
 	}
-	return nil
+	return true, nil
 }
 
 // postEvent queues the message for further processing.
@@ -546,20 +557,27 @@ func (w *Whisper) expire() {
 	w.poolMu.Lock()
 	defer w.poolMu.Unlock()
 
+	w.stats.clear()
 	now := uint32(time.Now().Unix())
-	for then, hashSet := range w.expirations {
-		// Short circuit if a future time
-		if then > now {
-			continue
-		}
+	for expiry, hashSet := range w.expirations {
+		if expiry < now {
+			w.stats.messages++
+
 			// Dump all expired messages and remove timestamp
 			hashSet.Each(func(v interface{}) bool {
+				w.stats.memory += w.envelopes[v.(common.Hash)].size()
 				delete(w.envelopes, v.(common.Hash))
 				delete(w.messages, v.(common.Hash))
 				return true
 			})
-		w.expirations[then].Clear()
+			w.expirations[expiry].Clear()
+			delete(w.expirations, expiry)
 		}
+	}
+}
+
+func (w *Whisper) Stats() string {
+	return fmt.Sprintf("Latest expiry cycle cleared %d messages (%d bytes)", w.stats.messages, w.stats.memory)
 }
 
 // envelopes retrieves all the messages currently pooled by the node.
@@ -590,6 +608,14 @@ func (w *Whisper) Messages(id string) []*ReceivedMessage {
 	return result
 }
 
+func (w *Whisper) isEnvelopeCached(hash common.Hash) bool {
+	w.poolMu.Lock()
+	defer w.poolMu.Unlock()
+
+	_, exist := w.envelopes[hash]
+	return exist
+}
+
 func (w *Whisper) addDecryptedMessage(msg *ReceivedMessage) {
 	w.poolMu.Lock()
 	defer w.poolMu.Unlock()
@@ -597,6 +623,11 @@ func (w *Whisper) addDecryptedMessage(msg *ReceivedMessage) {
 	w.messages[msg.EnvelopeHash] = msg
 }
 
+func (s *Statistics) clear() {
+	s.memory = 0
+	s.messages = 0
+}
+
 func ValidatePublicKey(k *ecdsa.PublicKey) bool {
 	return k != nil && k.X != nil && k.Y != nil && k.X.Sign() != 0 && k.Y.Sign() != 0
 }