From ee815ab292cd0902aa2e7a005151d1c46d298814 Mon Sep 17 00:00:00 2001 From: Vlad Date: Thu, 23 Feb 2017 17:49:46 +0100 Subject: [PATCH] whisper: expiry refactoring --- whisper/whisperv5/api.go | 47 ++++++++------ whisper/whisperv5/filter.go | 13 ++-- whisper/whisperv5/message.go | 16 ++--- whisper/whisperv5/peer.go | 21 +++---- whisper/whisperv5/peer_test.go | 3 +- whisper/whisperv5/whisper.go | 111 ++++++++++++++++++++++----------- 6 files changed, 127 insertions(+), 84 deletions(-) diff --git a/whisper/whisperv5/api.go b/whisper/whisperv5/api.go index d34213e05d..8be6d774f2 100644 --- a/whisper/whisperv5/api.go +++ b/whisper/whisperv5/api.go @@ -25,7 +25,8 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" "github.com/ethereum/go-ethereum/crypto" - "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" ) var whisperOffLineErr = errors.New("whisper is offline") @@ -64,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 { @@ -169,25 +178,25 @@ func (api *PublicWhisperAPI) NewFilter(args WhisperFilterArgs) (string, error) { if len(args.Topics) == 0 && len(args.KeyName) != 0 { info := "NewFilter: at least one topic must be specified" - log.Error(fmt.Sprintf(info)) + glog.V(logger.Error).Infof(info) return "", errors.New(info) } if len(args.KeyName) != 0 && len(filter.KeySym) == 0 { info := "NewFilter: key was not found by name: " + args.KeyName - log.Error(fmt.Sprintf(info)) + glog.V(logger.Error).Infof(info) return "", errors.New(info) } if len(args.To) == 0 && len(filter.KeySym) == 0 { info := "NewFilter: filter must contain either symmetric or asymmetric key" - log.Error(fmt.Sprintf(info)) + glog.V(logger.Error).Infof(info) return "", errors.New(info) } if len(args.To) != 0 && len(filter.KeySym) != 0 { info := "NewFilter: filter must not contain both symmetric and asymmetric key" - log.Error(fmt.Sprintf(info)) + glog.V(logger.Error).Infof(info) return "", errors.New(info) } @@ -195,13 +204,13 @@ func (api *PublicWhisperAPI) NewFilter(args WhisperFilterArgs) (string, error) { dst := crypto.ToECDSAPub(common.FromHex(args.To)) if !ValidatePublicKey(dst) { info := "NewFilter: Invalid 'To' address" - log.Error(fmt.Sprintf(info)) + glog.V(logger.Error).Infof(info) return "", errors.New(info) } filter.KeyAsym = api.whisper.GetIdentity(string(args.To)) if filter.KeyAsym == nil { info := "NewFilter: non-existent identity provided" - log.Error(fmt.Sprintf(info)) + glog.V(logger.Error).Infof(info) return "", errors.New(info) } } @@ -209,7 +218,7 @@ func (api *PublicWhisperAPI) NewFilter(args WhisperFilterArgs) (string, error) { if len(args.From) > 0 { if !ValidatePublicKey(filter.Src) { info := "NewFilter: Invalid 'From' address" - log.Error(fmt.Sprintf(info)) + glog.V(logger.Error).Infof(info) return "", errors.New(info) } } @@ -268,13 +277,13 @@ func (api *PublicWhisperAPI) Post(args PostArgs) error { pub := crypto.ToECDSAPub(common.FromHex(args.From)) if !ValidatePublicKey(pub) { info := "Post: Invalid 'From' address" - log.Error(fmt.Sprintf(info)) + glog.V(logger.Error).Infof(info) return errors.New(info) } params.Src = api.whisper.GetIdentity(string(args.From)) if params.Src == nil { info := "Post: non-existent identity provided" - log.Error(fmt.Sprintf(info)) + glog.V(logger.Error).Infof(info) return errors.New(info) } } @@ -282,7 +291,7 @@ func (api *PublicWhisperAPI) Post(args PostArgs) error { filter := api.whisper.GetFilter(args.FilterID) if filter == nil && len(args.FilterID) > 0 { info := fmt.Sprintf("Post: wrong filter id %s", args.FilterID) - log.Error(fmt.Sprintf(info)) + glog.V(logger.Error).Infof(info) return errors.New(info) } @@ -298,7 +307,7 @@ func (api *PublicWhisperAPI) Post(args PostArgs) error { sz := len(filter.Topics) if sz < 1 { info := fmt.Sprintf("Post: no topics in filter # %s", args.FilterID) - log.Error(fmt.Sprintf(info)) + glog.V(logger.Error).Infof(info) return errors.New(info) } else if sz == 1 { params.Topic = filter.Topics[0] @@ -313,26 +322,26 @@ func (api *PublicWhisperAPI) Post(args PostArgs) error { // validate if len(args.KeyName) != 0 && len(params.KeySym) == 0 { info := "Post: key was not found by name: " + args.KeyName - log.Error(fmt.Sprintf(info)) + glog.V(logger.Error).Infof(info) return errors.New(info) } if len(args.To) == 0 && len(params.KeySym) == 0 { info := "Post: message must be encrypted either symmetrically or asymmetrically" - log.Error(fmt.Sprintf(info)) + glog.V(logger.Error).Infof(info) return errors.New(info) } if len(args.To) != 0 && len(params.KeySym) != 0 { info := "Post: ambigous encryption method requested" - log.Error(fmt.Sprintf(info)) + glog.V(logger.Error).Infof(info) return errors.New(info) } if len(args.To) > 0 { if !ValidatePublicKey(params.Dst) { info := "Post: Invalid 'To' address" - log.Error(fmt.Sprintf(info)) + glog.V(logger.Error).Infof(info) return errors.New(info) } } @@ -341,17 +350,17 @@ func (api *PublicWhisperAPI) Post(args PostArgs) error { message := NewSentMessage(¶ms) envelope, err := message.Wrap(¶ms) if err != nil { - log.Error(fmt.Sprintf(err.Error())) + glog.V(logger.Error).Infof(err.Error()) return err } if len(envelope.Data) > MaxMessageLength { info := "Post: message is too big" - log.Error(fmt.Sprintf(info)) + glog.V(logger.Error).Infof(info) return errors.New(info) } if (envelope.Topic == TopicType{} && envelope.IsSymmetric()) { info := "Post: topic is missing for symmetric encryption" - log.Error(fmt.Sprintf(info)) + glog.V(logger.Error).Infof(info) return errors.New(info) } diff --git a/whisper/whisperv5/filter.go b/whisper/whisperv5/filter.go index 8aa7b2429d..832ebe3f6b 100644 --- a/whisper/whisperv5/filter.go +++ b/whisper/whisperv5/filter.go @@ -18,12 +18,13 @@ package whisperv5 import ( "crypto/ecdsa" - "crypto/rand" + crand "crypto/rand" "fmt" "sync" "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" ) type Filter struct { @@ -55,7 +56,7 @@ func NewFilters(w *Whisper) *Filters { func (fs *Filters) generateRandomID() (id string, err error) { buf := make([]byte, 20) for i := 0; i < 3; i++ { - _, err = rand.Read(buf) + _, err = crand.Read(buf) if err != nil { continue } @@ -106,7 +107,7 @@ func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) { var msg *ReceivedMessage for j, watcher := range fs.watchers { if p2pMessage && !watcher.AcceptP2P { - log.Trace(fmt.Sprintf("msg [%x], filter [%s]: p2p messages are not allowed", env.Hash(), j)) + glog.V(logger.Detail).Infof("msg [%x], filter [%d]: p2p messages are not allowed \n", env.Hash(), j) continue } @@ -118,10 +119,10 @@ func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) { if match { msg = env.Open(watcher) if msg == nil { - log.Trace(fmt.Sprintf("msg [%x], filter [%s]: failed to open", env.Hash(), j)) + glog.V(logger.Detail).Infof("msg [%x], filter [%d]: failed to open \n", env.Hash(), j) } } else { - log.Trace(fmt.Sprintf("msg [%x], filter [%s]: does not match", env.Hash(), j)) + glog.V(logger.Detail).Infof("msg [%x], filter [%d]: does not match \n", env.Hash(), j) } } diff --git a/whisper/whisperv5/message.go b/whisper/whisperv5/message.go index 9677f278e5..a095f7c0cd 100644 --- a/whisper/whisperv5/message.go +++ b/whisper/whisperv5/message.go @@ -30,8 +30,8 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/crypto" - "github.com/ethereum/go-ethereum/crypto/ecies" - "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" "golang.org/x/crypto/pbkdf2" ) @@ -143,7 +143,7 @@ func (msg *SentMessage) appendPadding(params *MessageParams) { func (msg *SentMessage) sign(key *ecdsa.PrivateKey) error { if isMessageSigned(msg.Raw[0]) { // this should not happen, but no reason to panic - log.Error(fmt.Sprintf("Trying to sign a message which was already signed")) + glog.V(logger.Error).Infof("Trying to sign a message which was already signed") return nil } @@ -163,7 +163,7 @@ func (msg *SentMessage) encryptAsymmetric(key *ecdsa.PublicKey) error { if !ValidatePublicKey(key) { return fmt.Errorf("Invalid public key provided for asymmetric encryption") } - encrypted, err := ecies.Encrypt(crand.Reader, ecies.ImportECDSAPublic(key), msg.Raw, nil, nil) + encrypted, err := crypto.Encrypt(key, msg.Raw) if err == nil { msg.Raw = encrypted } @@ -237,7 +237,7 @@ func (msg *SentMessage) Wrap(options *MessageParams) (envelope *Envelope, err er } } if len(msg.Raw) > MaxMessageLength { - log.Error(fmt.Sprintf("Message size must not exceed %d bytes", MaxMessageLength)) + glog.V(logger.Error).Infof("Message size must not exceed %d bytes", MaxMessageLength) return nil, errors.New("Oversized message") } var salt, nonce []byte @@ -280,7 +280,7 @@ func (msg *ReceivedMessage) decryptSymmetric(key []byte, salt []byte, nonce []by } if len(nonce) != aesgcm.NonceSize() { info := fmt.Sprintf("Wrong AES nonce size - want: %d, got: %d", len(nonce), aesgcm.NonceSize()) - log.Error(fmt.Sprintf(info)) + glog.V(logger.Error).Infof(info) return errors.New(info) } decrypted, err := aesgcm.Open(nil, nonce, msg.Raw, nil) @@ -293,7 +293,7 @@ func (msg *ReceivedMessage) decryptSymmetric(key []byte, salt []byte, nonce []by // decryptAsymmetric decrypts an encrypted payload with a private key. func (msg *ReceivedMessage) decryptAsymmetric(key *ecdsa.PrivateKey) error { - decrypted, err := ecies.ImportECDSA(key).Decrypt(crand.Reader, msg.Raw, nil, nil) + decrypted, err := crypto.Decrypt(key, msg.Raw) if err == nil { msg.Raw = decrypted } @@ -351,7 +351,7 @@ func (msg *ReceivedMessage) SigToPubKey() *ecdsa.PublicKey { pub, err := crypto.SigToPub(msg.hash(), msg.Signature) if err != nil { - log.Error(fmt.Sprintf("Could not get public key from signature: %v", err)) + glog.V(logger.Error).Infof("Could not get public key from signature: %v", err) return nil } return pub diff --git a/whisper/whisperv5/peer.go b/whisper/whisperv5/peer.go index e137613f5c..0ae6998d6b 100644 --- a/whisper/whisperv5/peer.go +++ b/whisper/whisperv5/peer.go @@ -21,7 +21,8 @@ import ( "time" "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/rlp" set "gopkg.in/fatih/set.v0" @@ -55,13 +56,13 @@ func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) *Peer { // into the network. func (p *Peer) start() { go p.update() - log.Debug(fmt.Sprintf("%v: whisper started", p.peer)) + glog.V(logger.Debug).Infof("%v: whisper started", p.peer) } // stop terminates the peer updater, stopping message forwarding to it. func (p *Peer) stop() { close(p.quit) - log.Debug(fmt.Sprintf("%v: whisper stopped", p.peer)) + glog.V(logger.Debug).Infof("%v: whisper stopped", p.peer) } // handshake sends the protocol initiation status message to the remote peer and @@ -110,7 +111,7 @@ func (p *Peer) update() { case <-transmit.C: if err := p.broadcast(); err != nil { - log.Info(fmt.Sprintf("%v: broadcast failed: %v", p.peer, err)) + glog.V(logger.Info).Infof("%v: broadcast failed: %v", p.peer, err) return } @@ -133,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) } @@ -171,7 +166,7 @@ func (p *Peer) broadcast() error { if err := p2p.Send(p.ws, messagesCode, transmit); err != nil { return err } - log.Trace(fmt.Sprint(p.peer, "broadcasted", len(transmit), "message(s)")) + glog.V(logger.Detail).Infoln(p.peer, "broadcasted", len(transmit), "message(s)") return nil } diff --git a/whisper/whisperv5/peer_test.go b/whisper/whisperv5/peer_test.go index 3dcf3bc702..cce2c92ba6 100644 --- a/whisper/whisperv5/peer_test.go +++ b/whisper/whisperv5/peer_test.go @@ -107,7 +107,8 @@ func TestSimulation(t *testing.T) { } func initialize(t *testing.T) { - // log.Root().SetHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat()))) + //glog.SetV(6) + //glog.SetToStderr(true) var err error ip := net.IPv4(127, 0, 0, 1) diff --git a/whisper/whisperv5/whisper.go b/whisper/whisperv5/whisper.go index 558e2909f9..771b7b3457 100644 --- a/whisper/whisperv5/whisper.go +++ b/whisper/whisperv5/whisper.go @@ -28,13 +28,20 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/crypto" - "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/rpc" "golang.org/x/crypto/pbkdf2" set "gopkg.in/fatih/set.v0" ) +type Statistics struct { + messagesCleared int + memoryCleared int + totalMemoryUsed int +} + // Whisper represents a dark communication interface through the Ethereum // network, using its very own P2P communication layer. type Whisper struct { @@ -59,6 +66,8 @@ type Whisper struct { p2pMsgQueue chan *Envelope quit chan struct{} + stats Statistics + overflow bool test bool } @@ -287,13 +296,14 @@ 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 // of the Whisper protocol. func (w *Whisper) Start(*p2p.Server) error { - log.Info(fmt.Sprint("Whisper started")) + glog.V(logger.Info).Infoln("Whisper started") go w.update() numCPU := runtime.NumCPU() @@ -308,7 +318,7 @@ func (w *Whisper) Start(*p2p.Server) error { // of the Whisper protocol. func (w *Whisper) Stop() error { close(w.quit) - log.Info(fmt.Sprint("Whisper stopped")) + glog.V(logger.Info).Infoln("Whisper stopped") return nil } @@ -350,21 +360,24 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error { switch packet.Code { case statusCode: // this should not happen, but no need to panic; just ignore this message. - log.Warn(fmt.Sprintf("%v: unxepected status message received", p.peer)) + glog.V(logger.Warn).Infof("%v: unxepected status message received", p.peer) case messagesCode: // decode the contained envelopes var envelopes []*Envelope if err := packet.Decode(&envelopes); err != nil { - log.Warn(fmt.Sprintf("%v: failed to decode envelope: [%v], peer will be disconnected", p.peer, err)) + glog.V(logger.Warn).Infof("%v: failed to decode envelope: [%v], peer will be disconnected", p.peer, err) return fmt.Errorf("garbage received") } // inject all envelopes into the internal pool for _, envelope := range envelopes { - if err := wh.add(envelope); err != nil { - log.Warn(fmt.Sprintf("%v: bad envelope received: [%v], peer will be disconnected", p.peer, err)) + 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") } - p.mark(envelope) + if cached { + p.mark(envelope) + } } case p2pCode: // peer-to-peer message, sent directly to peer bypassing PoW checks, etc. @@ -374,7 +387,7 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error { if p.trusted { var envelope Envelope if err := packet.Decode(&envelope); err != nil { - log.Warn(fmt.Sprintf("%v: failed to decode direct message: [%v], peer will be disconnected", p.peer, err)) + glog.V(logger.Warn).Infof("%v: failed to decode direct message: [%v], peer will be disconnected", p.peer, err) return fmt.Errorf("garbage received (directMessage)") } wh.postEvent(&envelope, true) @@ -384,7 +397,7 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error { if wh.mailServer != nil { var request Envelope if err := packet.Decode(&request); err != nil { - log.Warn(fmt.Sprintf("%v: failed to decode p2p request message: [%v], peer will be disconnected", p.peer, err)) + glog.V(logger.Warn).Infof("%v: failed to decode p2p request message: [%v], peer will be disconnected", p.peer, err) return fmt.Errorf("garbage received (p2p request)") } wh.mailServer.DeliverMail(p, &request) @@ -401,13 +414,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) @@ -416,34 +429,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 { - log.Debug(fmt.Sprintf("expired envelope dropped [%x]", envelope.Hash())) - return nil // drop envelope without error + glog.V(logger.Debug).Infof("expired envelope dropped [%x]", envelope.Hash()) + 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 { - log.Debug(fmt.Sprintf("envelope with low PoW dropped: %f [%x]", envelope.PoW(), envelope.Hash())) - return nil // drop envelope without error + glog.V(logger.Debug).Infof("envelope with low PoW dropped: %f [%x]", envelope.PoW(), envelope.Hash()) + return false, nil // drop envelope without error } hash := envelope.Hash() @@ -462,15 +475,16 @@ func (wh *Whisper) add(envelope *Envelope) error { wh.poolMu.Unlock() if alreadyCached { - log.Trace(fmt.Sprintf("whisper envelope already cached [%x]\n", envelope.Hash())) + glog.V(logger.Detail).Infof("whisper envelope already cached [%x]\n", envelope.Hash()) } else { - log.Trace(fmt.Sprintf("cached whisper envelope [%x]: %v\n", envelope.Hash(), envelope)) + glog.V(logger.Detail).Infof("cached whisper envelope [%x]: %v\n", envelope.Hash(), envelope) + wh.stats.totalMemoryUsed += envelope.size() wh.postEvent(envelope, false) // notify the local node about the new message if wh.mailServer != nil { wh.mailServer.Archive(envelope) } } - return nil + return true, nil } // postEvent queues the message for further processing. @@ -495,7 +509,7 @@ func (w *Whisper) checkOverflow() { if queueSize == messageQueueLimit { if !w.overflow { w.overflow = true - log.Warn(fmt.Sprint("message queue overflow")) + glog.V(logger.Warn).Infoln("message queue overflow") } } else if queueSize <= messageQueueLimit/2 { if w.overflow { @@ -545,22 +559,32 @@ 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.messagesCleared++ + + // Dump all expired messages and remove timestamp + hashSet.Each(func(v interface{}) bool { + sz := w.envelopes[v.(common.Hash)].size() + w.stats.memoryCleared += sz + w.stats.totalMemoryUsed -= sz + delete(w.envelopes, v.(common.Hash)) + delete(w.messages, v.(common.Hash)) + return true + }) + w.expirations[expiry].Clear() + delete(w.expirations, expiry) } - // Dump all expired messages and remove timestamp - hashSet.Each(func(v interface{}) bool { - delete(w.envelopes, v.(common.Hash)) - delete(w.messages, v.(common.Hash)) - return true - }) - w.expirations[then].Clear() } } +func (w *Whisper) Stats() string { + return fmt.Sprintf("Latest expiry cycle cleared %d messages (%d bytes). Memory usage: %d bytes.", + w.stats.messagesCleared, w.stats.memoryCleared, w.stats.totalMemoryUsed) +} + // envelopes retrieves all the messages currently pooled by the node. func (w *Whisper) Envelopes() []*Envelope { w.poolMu.RLock() @@ -589,6 +613,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() @@ -596,6 +628,11 @@ func (w *Whisper) addDecryptedMessage(msg *ReceivedMessage) { w.messages[msg.EnvelopeHash] = msg } +func (s *Statistics) clear() { + s.memoryCleared = 0 + s.messagesCleared = 0 +} + func ValidatePublicKey(k *ecdsa.PublicKey) bool { return k != nil && k.X != nil && k.Y != nil && k.X.Sign() != 0 && k.Y.Sign() != 0 }