From c63b03681233714316c3b7c508d5c321d119287f Mon Sep 17 00:00:00 2001 From: Jared Wasinger Date: Wed, 4 Apr 2018 02:47:22 +0000 Subject: [PATCH 1/5] pss: remove expired entries from forward cache --- swarm/pss/pss.go | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index 1b88c95db2..fed71a389b 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -756,6 +756,17 @@ func (self *Pss) forward(msg *PssMsg) { // SECTION: Caching ///////////////////////////////////////////////////////////////////// +// remove expired entries from forward cache +func (self *Pss) cleanFwdCache() { + self.fwdCacheMu.Lock() + defer self.fwdCacheMu.Unlock() + for k,v := range self.fwdCache { + if v.expiresAt.Before(time.Now()) { + delete(self.fwdCache[k]) + } + } +} + // add a message to the cache func (self *Pss) addFwdCache(msg *PssMsg) error { var entry pssCacheEntry From 8054d2f3a603a3426f00649320cb98b5694050c1 Mon Sep 17 00:00:00 2001 From: Jared Wasinger Date: Wed, 4 Apr 2018 21:26:14 +0000 Subject: [PATCH 2/5] format. Add separate tick for cache cleaning --- swarm/pss/pss.go | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index fed71a389b..267620ce00 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -177,7 +177,10 @@ func (self *Pss) Start(srv *p2p.Server) error { go func() { for { tickC := time.Tick(defaultCleanInterval) + cacheTickC := time.Tick(cacheTTL) select { + case <-cacheTickC: + self.cleanFwdCache() case <-tickC: self.cleanKeys() case <-self.quitC: @@ -758,13 +761,13 @@ func (self *Pss) forward(msg *PssMsg) { // remove expired entries from forward cache func (self *Pss) cleanFwdCache() { - self.fwdCacheMu.Lock() - defer self.fwdCacheMu.Unlock() - for k,v := range self.fwdCache { - if v.expiresAt.Before(time.Now()) { - delete(self.fwdCache[k]) - } - } + self.fwdCacheMu.Lock() + defer self.fwdCacheMu.Unlock() + for k, v := range self.fwdCache { + if v.expiresAt.Before(time.Now()) { + delete(self.fwdCache[k]) + } + } } // add a message to the cache From 83355516c302b54c371eb41d296117846f316103 Mon Sep 17 00:00:00 2001 From: Jared Wasinger Date: Thu, 5 Apr 2018 00:15:41 +0000 Subject: [PATCH 3/5] fix errors --- swarm/pss/pss.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index 267620ce00..327f2bd46d 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -177,7 +177,7 @@ func (self *Pss) Start(srv *p2p.Server) error { go func() { for { tickC := time.Tick(defaultCleanInterval) - cacheTickC := time.Tick(cacheTTL) + cacheTickC := time.Tick(self.cacheTTL) select { case <-cacheTickC: self.cleanFwdCache() @@ -765,7 +765,7 @@ func (self *Pss) cleanFwdCache() { defer self.fwdCacheMu.Unlock() for k, v := range self.fwdCache { if v.expiresAt.Before(time.Now()) { - delete(self.fwdCache[k]) + delete(self.fwdCache, k) } } } From 04367fc5ea2434f721234744716bf77fc767a52f Mon Sep 17 00:00:00 2001 From: Jared Wasinger Date: Fri, 6 Apr 2018 07:11:21 +0000 Subject: [PATCH 4/5] pss: add testing for forwarding cache clearing. --- swarm/pss/pss_test.go | 30 +++++++++++++++++++++++++++++- 1 file changed, 29 insertions(+), 1 deletion(-) diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index 254fa5c09e..5a6648982d 100644 --- a/swarm/pss/pss_test.go +++ b/swarm/pss/pss_test.go @@ -144,9 +144,11 @@ func TestCache(t *testing.T) { t.Fatal(err) } ps := newTestPss(privkey, nil, nil) + pp := NewPssParams(privkey) data := []byte("foo") datatwo := []byte("bar") + datathree := []byte("baz") wparams := &whisper.MessageParams{ TTL: defaultWhisperTTL, Src: privkey, @@ -169,6 +171,13 @@ func TestCache(t *testing.T) { Payload: envtwo, To: to, } + wparams.Payload = datathree + woutmsg, err = whisper.NewSentMessage(wparams) + envthree, err := woutmsg.Wrap(wparams) + msgthree := &PssMsg{ + Payload: envthree, + To: to, + } digest := ps.digest(msg) if err != nil { @@ -178,6 +187,11 @@ func TestCache(t *testing.T) { if err != nil { t.Fatalf("could not store cache msgtwo: %v", err) } + digestthree := ps.digest(msgthree) + if err != nil { + t.Fatalf("could not store cache msgthree: %v", err) + } + if digest == digesttwo { t.Fatalf("different msgs return same hash: %d", digesttwo) @@ -197,10 +211,23 @@ func TestCache(t *testing.T) { t.Fatalf("message %v should NOT have EXPIRE record in cache but checkCache returned true", msgtwo) } - time.Sleep(pp.CacheTTL) + time.Sleep(pp.CacheTTL + 1*time.Second) + err = ps.addFwdCache(msgthree) + if err != nil { + t.Fatalf("write to pss expire cache failed: %v", err) + } + if ps.checkFwdCache(msg) { t.Fatalf("message %v should have expired from cache but checkCache returned true", msg) } + + if _, ok := ps.fwdCache[digestthree]; !ok { + t.Fatalf("unexpired message should be in the cache: %v", digestthree) + } + + if _, ok := ps.fwdCache[digesttwo]; ok { + t.Fatalf("expired message should have been cleared from the cache: %v", digesttwo) + } } // matching of address hints; whether a message could be or is for the node @@ -1309,6 +1336,7 @@ func newTestPss(privkey *ecdsa.PrivateKey, overlay network.Overlay, ppextra *Pss pp.SymKeyCacheCapacity = ppextra.SymKeyCacheCapacity } ps := NewPss(overlay, pp) + ps.Start(nil) return ps } From ac39a0ccb19337dc2e36669435af8fbb2f2640f5 Mon Sep 17 00:00:00 2001 From: Jared Wasinger Date: Mon, 9 Apr 2018 06:57:37 +0000 Subject: [PATCH 5/5] add documentation. fix formatting --- swarm/pss/pss.go | 2 +- swarm/pss/pss_test.go | 10 ++++------ 2 files changed, 5 insertions(+), 7 deletions(-) diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index 327f2bd46d..3a91b546d5 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -759,7 +759,7 @@ func (self *Pss) forward(msg *PssMsg) { // SECTION: Caching ///////////////////////////////////////////////////////////////////// -// remove expired entries from forward cache +// cleanFwdCache is used to periodically remove expired entries from the forward cache func (self *Pss) cleanFwdCache() { self.fwdCacheMu.Lock() defer self.fwdCacheMu.Unlock() diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index 5a6648982d..9eaba00ccc 100644 --- a/swarm/pss/pss_test.go +++ b/swarm/pss/pss_test.go @@ -144,11 +144,10 @@ func TestCache(t *testing.T) { t.Fatal(err) } ps := newTestPss(privkey, nil, nil) - pp := NewPssParams(privkey) data := []byte("foo") datatwo := []byte("bar") - datathree := []byte("baz") + datathree := []byte("baz") wparams := &whisper.MessageParams{ TTL: defaultWhisperTTL, Src: privkey, @@ -192,7 +191,6 @@ func TestCache(t *testing.T) { t.Fatalf("could not store cache msgthree: %v", err) } - if digest == digesttwo { t.Fatalf("different msgs return same hash: %d", digesttwo) } @@ -212,7 +210,7 @@ func TestCache(t *testing.T) { } time.Sleep(pp.CacheTTL + 1*time.Second) - err = ps.addFwdCache(msgthree) + err = ps.addFwdCache(msgthree) if err != nil { t.Fatalf("write to pss expire cache failed: %v", err) } @@ -221,9 +219,9 @@ func TestCache(t *testing.T) { t.Fatalf("message %v should have expired from cache but checkCache returned true", msg) } - if _, ok := ps.fwdCache[digestthree]; !ok { + if _, ok := ps.fwdCache[digestthree]; !ok { t.Fatalf("unexpired message should be in the cache: %v", digestthree) - } + } if _, ok := ps.fwdCache[digesttwo]; ok { t.Fatalf("expired message should have been cleared from the cache: %v", digesttwo)