From 2d07f52e67401706f33f55fba2efa294b18cd51e Mon Sep 17 00:00:00 2001 From: Dmitry Shulyak Date: Sat, 17 Feb 2018 15:14:24 +0200 Subject: [PATCH] whisper: periodically rebuild bloom filter to remove stale topics In the current version of whisper there is no way to remove topic from bloom filter after it was added. It leads to unnecessary accumulation of such topics over time and undesirable network traffic. In this change I added an option to periodically rebuild bloom filter from available filter topics as there is no way to simply revert an update to bloom filter. But it would be possible if whisper used cuckoo filters. Current interaction between peers allows to reset bloom filter to any desirable state, allowing message from old bloom filter for certain time. At the moment this time is 10s. --- whisper/whisperv6/config.go | 12 +++++--- whisper/whisperv6/filter.go | 9 ++++++ whisper/whisperv6/whisper.go | 46 +++++++++++++++++++++++++------ whisper/whisperv6/whisper_test.go | 37 +++++++++++++++++++++++++ 4 files changed, 91 insertions(+), 13 deletions(-) diff --git a/whisper/whisperv6/config.go b/whisper/whisperv6/config.go index 61419de007..5be6a1847f 100644 --- a/whisper/whisperv6/config.go +++ b/whisper/whisperv6/config.go @@ -16,14 +16,18 @@ package whisperv6 +import "time" + // Config represents the configuration state of a whisper node. type Config struct { - MaxMessageSize uint32 `toml:",omitempty"` - MinimumAcceptedPOW float64 `toml:",omitempty"` + MaxMessageSize uint32 `toml:",omitempty"` + MinimumAcceptedPOW float64 `toml:",omitempty"` + BloomFilterRebuildPeriod time.Duration `toml:",omitempty"` } // DefaultConfig represents (shocker!) the default configuration. var DefaultConfig = Config{ - MaxMessageSize: DefaultMaxMessageSize, - MinimumAcceptedPOW: DefaultMinimumPoW, + MaxMessageSize: DefaultMaxMessageSize, + MinimumAcceptedPOW: DefaultMinimumPoW, + BloomFilterRebuildPeriod: 0, } diff --git a/whisper/whisperv6/filter.go b/whisper/whisperv6/filter.go index eb0c65fa3b..10e7b49c46 100644 --- a/whisper/whisperv6/filter.go +++ b/whisper/whisperv6/filter.go @@ -85,6 +85,15 @@ func (fs *Filters) Install(watcher *Filter) (string, error) { return id, err } +// Each is a thread safe iterator over all existing filters. +func (fs *Filters) Each(f func(filter *Filter)) { + fs.mutex.RLock() + defer fs.mutex.RUnlock() + for _, filter := range fs.watchers { + f(filter) + } +} + // Uninstall will remove a filter whose id has been specified from // the filter collection func (fs *Filters) Uninstall(id string) bool { diff --git a/whisper/whisperv6/whisper.go b/whisper/whisperv6/whisper.go index 600f9cb286..abdfd467ab 100644 --- a/whisper/whisperv6/whisper.go +++ b/whisper/whisperv6/whisper.go @@ -49,12 +49,13 @@ type Statistics struct { } const ( - maxMsgSizeIdx = iota // Maximal message length allowed by the whisper node - overflowIdx // Indicator of message queue overflow - minPowIdx // Minimal PoW required by the whisper node - minPowToleranceIdx // Minimal PoW tolerated by the whisper node for a limited time - bloomFilterIdx // Bloom filter for topics of interest for this node - bloomFilterToleranceIdx // Bloom filter tolerated by the whisper node for a limited time + maxMsgSizeIdx = iota // Maximal message length allowed by the whisper node + overflowIdx // Indicator of message queue overflow + minPowIdx // Minimal PoW required by the whisper node + minPowToleranceIdx // Minimal PoW tolerated by the whisper node for a limited time + bloomFilterIdx // Bloom filter for topics of interest for this node + bloomFilterToleranceIdx // Bloom filter tolerated by the whisper node for a limited time + bloomFilterRebuildPeriodIdx // Used to define period how often bloom filter is rebuild from scratch. ) // Whisper represents a dark communication interface through the Ethereum @@ -111,6 +112,7 @@ func New(cfg *Config) *Whisper { whisper.settings.Store(minPowIdx, cfg.MinimumAcceptedPOW) whisper.settings.Store(maxMsgSizeIdx, cfg.MaxMessageSize) whisper.settings.Store(overflowIdx, false) + whisper.settings.Store(bloomFilterRebuildPeriodIdx, cfg.BloomFilterRebuildPeriod) // p2p whisper sub protocol handler whisper.protocol = p2p.Protocol{ @@ -236,7 +238,6 @@ func (whisper *Whisper) SetBloomFilter(bloom []byte) error { b := make([]byte, bloomFilterSize) copy(b, bloom) - whisper.settings.Store(bloomFilterIdx, b) whisper.notifyPeersAboutBloomFilterChange(b) @@ -553,6 +554,24 @@ func (whisper *Whisper) Subscribe(f *Filter) (string, error) { return s, err } +// rebuildBloomFilter makes a new bloom filter from currently subscribed watchers +// if new bloom filter is different from current bloom filter - peers will be notified +func (whisper *Whisper) rebuildBloomFilter() { + if isFullNode(whisper.BloomFilter()) { + return + } + aggregate := make([]byte, bloomFilterSize) + whisper.filters.Each(func(filter *Filter) { + for _, t := range filter.Topics { + top := BytesToTopic(t) + aggregate = addBloom(aggregate, TopicToBloom(top)) + } + }) + if !bloomFilterMatch(aggregate, whisper.BloomFilter()) { + whisper.SetBloomFilter(aggregate) + } +} + // updateBloomFilter recalculates the new value of bloom filter, // and informs the peers if necessary. func (whisper *Whisper) updateBloomFilter(f *Filter) { @@ -868,13 +887,22 @@ func (whisper *Whisper) processQueue() { func (whisper *Whisper) update() { // Start a ticker to check for expirations expire := time.NewTicker(expirationCycle) - + defer expire.Stop() + // bloomFilterRebuildPeriodIdx is always set in New + val, _ := whisper.settings.Load(bloomFilterRebuildPeriodIdx) + var rebuildTicker <-chan time.Time + if duration := val.(time.Duration); duration > 0 { + bloomRebuild := time.NewTicker(val.(time.Duration)) + defer bloomRebuild.Stop() + rebuildTicker = bloomRebuild.C + } // Repeat updates until termination is requested for { select { case <-expire.C: whisper.expire() - + case <-rebuildTicker: + whisper.rebuildBloomFilter() case <-whisper.quit: return } diff --git a/whisper/whisperv6/whisper_test.go b/whisper/whisperv6/whisper_test.go index 99e5f0bbb4..dd4eaa1af1 100644 --- a/whisper/whisperv6/whisper_test.go +++ b/whisper/whisperv6/whisper_test.go @@ -25,6 +25,7 @@ import ( "time" "github.com/ethereum/go-ethereum/common" + "github.com/stretchr/testify/require" "golang.org/x/crypto/pbkdf2" ) @@ -892,3 +893,39 @@ func TestBloom(t *testing.T) { t.Fatalf("retireved wrong bloom filter") } } + +// TestPeriodicFilterRebuild verifies that bloom filter will be periodically +// rebuild to avoid consuming unnecessary traffic. +func TestPeriodicBloomFilterRebuild(t *testing.T) { + InitSingleTest() + w := New(&Config{ + MaxMessageSize: DefaultMaxMessageSize, + MinimumAcceptedPOW: DefaultMinimumPoW, + BloomFilterRebuildPeriod: 100 * time.Millisecond, + }) + w.Start(nil) + defer w.Stop() + + emptyBloom := make([]byte, bloomFilterSize) + require.NoError(t, w.SetBloomFilter(emptyBloom)) + + f1, err := generateFilter(t, true) + require.NoError(t, err) + f2, err := generateFilter(t, true) + require.NoError(t, err) + sub1, err := w.Subscribe(f1) + require.NoError(t, err) + bloomf1 := w.BloomFilter() + sub2, err := w.Subscribe(f2) + require.NoError(t, err) + + require.NoError(t, w.Unsubscribe(sub2)) + time.Sleep(200 * time.Millisecond) + require.Equal(t, bloomf1, w.BloomFilter()) + // tolerance will be updated only after 10s + require.Nil(t, w.BloomFilterTolerance()) + + require.NoError(t, w.Unsubscribe(sub1)) + time.Sleep(200 * time.Millisecond) + require.Equal(t, emptyBloom, w.BloomFilter()) +}