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()) +}